diff --git a/go/busarchive/archive.go b/go/busarchive/archive.go new file mode 100644 index 00000000..0d14f545 --- /dev/null +++ b/go/busarchive/archive.go @@ -0,0 +1,162 @@ +package busarchive + +import ( + "archive/tar" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" +) + +// Archive is a snapshot read back: its manifest, and each file beside it unpacked into a scratch +// directory of its own, removed by Close. +type Archive struct { + Manifest Manifest + dir string + files map[string]string +} + +// ReadArchive unpacks a snapshot. Only the files the manifest names are kept, and only under names +// that are one stream's directory: an archive is data, and a path in it is never trusted to say where +// on the machine a file goes. +func ReadArchive(r io.Reader, scratch string) (*Archive, error) { + dir, err := os.MkdirTemp(scratch, "mesh-nats-restore-*") + if err != nil { + return nil, err + } + a := &Archive{dir: dir, files: map[string]string{}} + tr := tar.NewReader(r) + first := true + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + a.Close() + return nil, fmt.Errorf("the archive does not read: %w", err) + } + if first { + if hdr.Name != ManifestName { + a.Close() + return nil, fmt.Errorf("the archive starts with %q, not its manifest: it is not a snapshot of the bus", hdr.Name) + } + raw, err := io.ReadAll(io.LimitReader(tr, 64<<20)) + if err != nil { + a.Close() + return nil, err + } + if err := json.Unmarshal(raw, &a.Manifest); err != nil { + a.Close() + return nil, fmt.Errorf("its manifest: %w", err) + } + if a.Manifest.Format != 1 { + a.Close() + return nil, fmt.Errorf("its manifest is format %d; this program reads format 1", a.Manifest.Format) + } + first = false + continue + } + if !a.named(hdr.Name) { + a.Close() + return nil, fmt.Errorf("the archive holds %q, which its manifest does not name", hdr.Name) + } + local := filepath.Join(dir, fmt.Sprintf("%04d", len(a.files))) + f, err := os.OpenFile(local, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) + if err != nil { + a.Close() + return nil, err + } + _, err = io.Copy(f, tr) + f.Close() + if err != nil { + a.Close() + return nil, err + } + a.files[hdr.Name] = local + } + if first { + a.Close() + return nil, errors.New("the archive is empty") + } + for _, s := range a.Manifest.Streams { + if !ValidName(s.Name) || s.Meta != "streams/"+s.Name+"/backup.json" || s.Archive != "streams/"+s.Name+"/stream.tar.s2" { + a.Close() + return nil, fmt.Errorf("the manifest names a stream %q at paths that are not its own", s.Name) + } + } + return a, nil +} + +func (a *Archive) named(name string) bool { + for _, s := range a.Manifest.Streams { + if name == s.Meta || name == s.Archive { + return true + } + } + return false +} + +// Close removes the unpacked files. +func (a *Archive) Close() { os.RemoveAll(a.dir) } + +// Verify holds every file to the manifest's size and checksum, and reads every stream's archive to +// its end. +func (a *Archive) Verify() error { + for _, s := range a.Manifest.Streams { + for _, f := range []struct{ name, sum string }{{s.Meta, s.MetaSHA256}, {s.Archive, s.ArchiveSHA256}} { + local, ok := a.files[f.name] + if !ok { + return fmt.Errorf("%s is named in the manifest and missing from the archive", f.name) + } + got, size, err := sha256Of(local) + if err != nil { + return err + } + if got != f.sum { + return fmt.Errorf("%s does not match the checksum its manifest gives", f.name) + } + if f.name == s.Archive && size != s.ArchiveBytes { + return fmt.Errorf("%s is %d bytes; its manifest says %d", f.name, size, s.ArchiveBytes) + } + } + af, err := os.Open(a.files[s.Archive]) + if err != nil { + return err + } + err = ReadsThrough(af) + af.Close() + if err != nil { + return fmt.Errorf("%s does not read: %w", s.Archive, err) + } + } + return nil +} + +func sha256Of(path string) (string, int64, error) { + f, err := os.Open(path) + if err != nil { + return "", 0, err + } + defer f.Close() + h := sha256.New() + n, err := io.Copy(h, f) + if err != nil { + return "", 0, err + } + return hex.EncodeToString(h.Sum(nil)), n, nil +} + +// Open is one stream's meta and archive. +func (a *Archive) Open(s StreamEntry) (meta []byte, archive *os.File, err error) { + meta, err = os.ReadFile(a.files[s.Meta]) + if err != nil { + return nil, nil, err + } + archive, err = os.Open(a.files[s.Archive]) + return meta, archive, err +} diff --git a/go/busarchive/types.go b/go/busarchive/types.go new file mode 100644 index 00000000..12ca5eba --- /dev/null +++ b/go/busarchive/types.go @@ -0,0 +1,86 @@ +// Package busarchive is the bus's snapshot archive: its manifest, reading one back and holding it to +// its checksums. The snapshot program (cmd/mesh-nats-snapshot) writes and restores with it and the bus's +// tool nats_snapshot_verify (cmd/nats-tools) checks with it, so the two never differ on what a good +// snapshot is. +package busarchive + +import ( + "archive/tar" + "io" + "strings" + "time" + + "github.com/klauspost/compress/s2" +) + +// Manifest is the first entry of an archive: what was taken, when, how long it took, and the checksum +// of every file beside it. It is what `verify` and `restore` hold the archive to. +type Manifest struct { + Format int `json:"format"` + Server string `json:"server_version"` + Started time.Time `json:"started"` + Finished time.Time `json:"finished"` + Seconds float64 `json:"seconds"` + // Messages and StreamBytes are the streams' own totals at their snapshots; ArchiveBytes what the + // archives take, compressed. + Messages uint64 `json:"messages"` + StreamBytes int64 `json:"stream_bytes"` + ArchiveBytes int64 `json:"archive_bytes"` + Streams []StreamEntry `json:"streams"` + // Skipped are streams that hold nothing across a restart (memory storage), said rather than + // silently left out. + Skipped []Skipped `json:"skipped,omitempty"` +} + +// StreamEntry is one stream as it was snapshotted. +type StreamEntry struct { + Name string `json:"name"` + Subjects []string `json:"subjects,omitempty"` + Retention string `json:"retention,omitempty"` + Messages uint64 `json:"messages"` + Bytes uint64 `json:"bytes"` + FirstSeq uint64 `json:"first_seq"` + LastSeq uint64 `json:"last_seq"` + Deleted int `json:"deleted,omitempty"` + Consumers int `json:"consumers"` + // Meta is the server's own description of the stream (its configuration and state), in the + // shape the restore API and the nats CLI take back: backup.json. + Meta string `json:"meta"` + MetaSHA256 string `json:"meta_sha256"` + // Archive is the server's snapshot of the stream, as the server sent it: stream.tar.s2. + Archive string `json:"archive"` + ArchiveBytes int64 `json:"archive_bytes"` + ArchiveSHA256 string `json:"archive_sha256"` + Seconds float64 `json:"seconds"` +} + +// Skipped is a stream not snapshotted, and why. +type Skipped struct { + Name string `json:"name"` + Why string `json:"why"` +} + +// ManifestName is the archive's first entry. +const ManifestName = "manifest.json" + +// ReadsThrough is whether a stream's archive decompresses and lists to its end. +func ReadsThrough(r io.Reader) error { + tr := tar.NewReader(s2.NewReader(r)) + for { + _, err := tr.Next() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if _, err := io.Copy(io.Discard, tr); err != nil { + return err + } + } +} + +// ValidName is a stream name that is one safe path component. +func ValidName(name string) bool { + return name != "" && name != "." && name != ".." && !strings.ContainsAny(name, "/\\\x00 *>") +} diff --git a/go/busarchive/verifyfile.go b/go/busarchive/verifyfile.go new file mode 100644 index 00000000..fdb4a2fc --- /dev/null +++ b/go/busarchive/verifyfile.go @@ -0,0 +1,109 @@ +package busarchive + +import ( + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strings" + "time" +) + +// StreamReport is one stream of a verified snapshot. +type StreamReport struct { + Name string `json:"name"` + LastSeq uint64 `json:"last_seq"` +} + +// Report is the answer of VerifyFile: which file was held to its manifest, and whether it passed. +type Report struct { + File string `json:"file"` + Modified time.Time `json:"modified"` + Size int64 `json:"size"` + Taken *time.Time `json:"taken,omitempty"` + Streams []StreamReport `json:"streams"` + Skipped []Skipped `json:"skipped,omitempty"` + Passed bool `json:"passed"` + // Fault is the first thing wrong, empty when the snapshot passed. + Fault string `json:"fault,omitempty"` +} + +// VerifyFile verifies one snapshot inside dir: the file called name, or the newest *.tar when name is +// empty. Nothing outside dir is read and nothing is written there; a snapshot that fails is an answer +// (Passed false, the first Fault), and only a directory without a snapshot is an error. +func VerifyFile(dir, name string) (*Report, error) { + path, info, err := pick(dir, name) + if err != nil { + return nil, err + } + r := &Report{File: filepath.Base(path), Modified: info.ModTime().UTC(), Size: info.Size(), Streams: []StreamReport{}} + f, err := os.Open(path) + if err != nil { + r.Fault = err.Error() + return r, nil + } + defer f.Close() + a, err := ReadArchive(f, "") + if err != nil { + r.Fault = err.Error() + return r, nil + } + defer a.Close() + taken := a.Manifest.Finished + r.Taken = &taken + r.Skipped = a.Manifest.Skipped + for _, s := range a.Manifest.Streams { + r.Streams = append(r.Streams, StreamReport{Name: s.Name, LastSeq: s.LastSeq}) + } + if err := a.Verify(); err != nil { + r.Fault = err.Error() + return r, nil + } + r.Passed = true + return r, nil +} + +// pick names the regular file to verify, and nothing outside dir: a name is one path component, and a +// symbolic link is not followed out of the directory. +func pick(dir, name string) (string, fs.FileInfo, error) { + if dir == "" { + return "", nil, errors.New("the snapshots directory is not set") + } + if name != "" { + if name != filepath.Base(name) || name == "." || name == ".." || strings.ContainsRune(name, 0) { + return "", nil, fmt.Errorf("%q is not a file name inside the snapshots directory", name) + } + p := filepath.Join(dir, name) + info, err := os.Lstat(p) + if err != nil { + return "", nil, fmt.Errorf("no snapshot %s in the snapshots directory: %w", name, err) + } + if !info.Mode().IsRegular() { + return "", nil, fmt.Errorf("%s is not a regular file", name) + } + return p, info, nil + } + entries, err := os.ReadDir(dir) + if err != nil { + return "", nil, fmt.Errorf("the snapshots directory does not read: %w", err) + } + var best string + var bestInfo fs.FileInfo + for _, e := range entries { + if !strings.HasSuffix(e.Name(), ".tar") { + continue + } + info, err := e.Info() + if err != nil || !info.Mode().IsRegular() { + continue + } + if bestInfo == nil || info.ModTime().After(bestInfo.ModTime()) { + best, bestInfo = e.Name(), info + } + } + if bestInfo == nil { + return "", nil, errors.New("the snapshots directory holds no .tar snapshot") + } + return filepath.Join(dir, best), bestInfo, nil +} diff --git a/go/busarchive/verifyfile_test.go b/go/busarchive/verifyfile_test.go new file mode 100644 index 00000000..be62737f --- /dev/null +++ b/go/busarchive/verifyfile_test.go @@ -0,0 +1,173 @@ +package busarchive + +import ( + "archive/tar" + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/klauspost/compress/s2" +) + +func sum(b []byte) string { h := sha256.Sum256(b); return hex.EncodeToString(h[:]) } + +// streamArchive is a stream's archive as the server sends it: a tar of one file, s2-compressed. +func streamArchive(t *testing.T) []byte { + t.Helper() + var raw bytes.Buffer + tw := tar.NewWriter(&raw) + body := []byte(strings.Repeat("a message of the stream\n", 50)) + if err := tw.WriteHeader(&tar.Header{Name: "msgs.blk", Mode: 0o600, Size: int64(len(body))}); err != nil { + t.Fatal(err) + } + tw.Write(body) + tw.Close() + var out bytes.Buffer + w := s2.NewWriter(&out) + w.Write(raw.Bytes()) + w.Close() + return out.Bytes() +} + +// snapshotTar builds a snapshot of two streams; the mutations change what the manifest says or what is +// stored, so each fault is made in the one place it is found. +func snapshotTar(t *testing.T, mutate func(m *Manifest, files map[string][]byte)) []byte { + t.Helper() + files := map[string][]byte{} + m := Manifest{Format: 1, Server: "2.11.17", Finished: time.Date(2026, 10, 22, 3, 0, 0, 0, time.UTC)} + for i, name := range []string{"EVENTS", "KV_conditions"} { + meta := []byte(`{"config":{"name":"` + name + `"}}`) + arc := streamArchive(t) + e := StreamEntry{Name: name, LastSeq: uint64(10 * (i + 1)), + Meta: "streams/" + name + "/backup.json", MetaSHA256: sum(meta), + Archive: "streams/" + name + "/stream.tar.s2", ArchiveBytes: int64(len(arc)), ArchiveSHA256: sum(arc)} + m.Streams = append(m.Streams, e) + files[e.Meta], files[e.Archive] = meta, arc + } + if mutate != nil { + mutate(&m, files) + } + var out bytes.Buffer + tw := tar.NewWriter(&out) + put := func(name string, b []byte) { + if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(b))}); err != nil { + t.Fatal(err) + } + tw.Write(b) + } + raw, _ := json.Marshal(m) + put(ManifestName, raw) + for _, s := range m.Streams { + put(s.Meta, files[s.Meta]) + put(s.Archive, files[s.Archive]) + } + tw.Close() + return out.Bytes() +} + +func put(t *testing.T, dir, name string, b []byte, at time.Time) { + t.Helper() + p := filepath.Join(dir, name) + if err := os.WriteFile(p, b, 0o600); err != nil { + t.Fatal(err) + } + os.Chtimes(p, at, at) +} + +func TestAGoodSnapshotPassesAndSaysItsStreams(t *testing.T) { + dir := t.TempDir() + good := snapshotTar(t, nil) + put(t, dir, "bus.tar", good, time.Now()) + r, err := VerifyFile(dir, "") + if err != nil { + t.Fatal(err) + } + if !r.Passed || r.Fault != "" || r.File != "bus.tar" || r.Size != int64(len(good)) { + t.Fatalf("%+v", r) + } + if r.Taken == nil || !r.Taken.Equal(time.Date(2026, 10, 22, 3, 0, 0, 0, time.UTC)) { + t.Errorf("taken %v", r.Taken) + } + if len(r.Streams) != 2 || r.Streams[0] != (StreamReport{"EVENTS", 10}) || r.Streams[1] != (StreamReport{"KV_conditions", 20}) { + t.Errorf("streams %+v", r.Streams) + } +} + +func TestACorruptedChecksumFailsNamingTheFile(t *testing.T) { + dir := t.TempDir() + put(t, dir, "bus.tar", snapshotTar(t, func(m *Manifest, files map[string][]byte) { + m.Streams[1].ArchiveSHA256 = sum([]byte("something else")) + }), time.Now()) + r, err := VerifyFile(dir, "bus.tar") + if err != nil { + t.Fatal(err) + } + if r.Passed || !strings.Contains(r.Fault, "streams/KV_conditions/stream.tar.s2") || !strings.Contains(r.Fault, "checksum") { + t.Fatalf("%+v", r) + } + if len(r.Streams) != 2 { + t.Errorf("a failed snapshot still says its streams: %+v", r.Streams) + } +} + +func TestATruncatedSnapshotFails(t *testing.T) { + dir := t.TempDir() + good := snapshotTar(t, nil) + put(t, dir, "bus.tar", good[:len(good)*2/3], time.Now()) + r, err := VerifyFile(dir, "") + if err != nil { + t.Fatal(err) + } + if r.Passed || r.Fault == "" { + t.Fatalf("a truncated snapshot passed: %+v", r) + } +} + +func TestAnArchiveThatDoesNotReadToItsEndFailsEvenWithMatchingChecksums(t *testing.T) { + dir := t.TempDir() + put(t, dir, "bus.tar", snapshotTar(t, func(m *Manifest, files map[string][]byte) { + bad := []byte("not an s2 stream at all") + s := &m.Streams[0] + files[s.Archive] = bad + s.ArchiveSHA256, s.ArchiveBytes = sum(bad), int64(len(bad)) + }), time.Now()) + r, _ := VerifyFile(dir, "") + if r.Passed || !strings.Contains(r.Fault, "does not read") { + t.Fatalf("%+v", r) + } +} + +func TestTheNewestTarIsVerifiedAndOnlyInsideTheDirectory(t *testing.T) { + dir := t.TempDir() + now := time.Now() + put(t, dir, "old.tar", []byte("junk"), now.Add(-48*time.Hour)) + put(t, dir, "bus.tar", snapshotTar(t, nil), now) + put(t, dir, "bus.tar.partial", []byte("half"), now.Add(time.Hour)) + r, err := VerifyFile(dir, "") + if err != nil || r.File != "bus.tar" || !r.Passed { + t.Fatalf("%+v %v", r, err) + } + if r, _ := VerifyFile(dir, "old.tar"); r.Passed || r.File != "old.tar" { + t.Errorf("a named file is the one checked: %+v", r) + } + outside := t.TempDir() + put(t, outside, "x.tar", snapshotTar(t, nil), now) + for _, name := range []string{"../" + filepath.Base(outside) + "/x.tar", outside + "/x.tar", ".."} { + if _, err := VerifyFile(dir, name); err == nil { + t.Errorf("%q was read", name) + } + } + os.Symlink(filepath.Join(outside, "x.tar"), filepath.Join(dir, "link.tar")) + if _, err := VerifyFile(dir, "link.tar"); err == nil { + t.Error("a symbolic link out of the directory was followed") + } + if _, err := VerifyFile(t.TempDir(), ""); err == nil { + t.Error("an empty directory is not a snapshot") + } +} diff --git a/go/bussnap/snapshot.go b/go/bussnap/snapshot.go new file mode 100644 index 00000000..c2b285ea --- /dev/null +++ b/go/bussnap/snapshot.go @@ -0,0 +1,433 @@ +// Package bussnap takes a snapshot of the bus's streams with the server's own snapshot API (novox/hq ADR 0235): the +// nightly dump of every stream, and the bus module's keeper's snapshot of one entry before it is rebuilt or removed +// (design 25 § 12.3). +package bussnap + +import ( + "archive/tar" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "slices" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/busarchive" + + "github.com/nats-io/nats.go" +) + +// The JetStream API's subjects this program speaks. Each is granted to the bus's own module and +// nothing else is (mesh-controller internal/broker, BusSnapshotGrants); a test holds the two lists +// equal against a real server. +const ( + APIStreamNames = "$JS.API.STREAM.NAMES" + APIStreamInfo = "$JS.API.STREAM.INFO." + APIStreamSnapshot = "$JS.API.STREAM.SNAPSHOT." + APIStreamRestore = "$JS.API.STREAM.RESTORE." +) + +// ChunkSize is what the server is asked to send at a time: its own default. The server sends up to +// an 8 MiB window ahead of the acknowledgements and waits for them past it, so one stream costs the +// server at most that much in flight, whatever its size. +const ChunkSize = 128 * 1024 + +// SnapshotOptions bounds a snapshot. +type SnapshotOptions struct { + // StreamTimeout is the longest one stream may take; past it the snapshot fails, naming it. + StreamTimeout time.Duration + // Pause is a rest between two streams, so a night's snapshot is never one long burst. + Pause time.Duration + // TempDir is where one stream's archive waits until it is written out; the system's when empty. + TempDir string + Log func(format string, args ...any) + // Streams, when set, are the only streams taken (the keeper's snapshot of one entry); every stream when empty. + Streams []string +} + +// APIError is the error every JetStream API answer may carry. +type APIError struct { + Code int `json:"code"` + ErrCode int `json:"err_code"` + Description string `json:"description"` +} + +func (e *APIError) Error() string { return fmt.Sprintf("%s (%d)", e.Description, e.Code) } + +// StreamState is what this program reads of a stream's state; the whole of it is kept as the server +// gave it, in the stream's meta. +type StreamState struct { + Messages uint64 `json:"messages"` + Bytes uint64 `json:"bytes"` + FirstSeq uint64 `json:"first_seq"` + LastSeq uint64 `json:"last_seq"` + NumDelete int `json:"num_deleted"` + Consumers int `json:"consumer_count"` +} + +type streamConfig struct { + Name string `json:"name"` + Subjects []string `json:"subjects"` + Retention string `json:"retention"` + Storage string `json:"storage"` +} + +// Request asks the JetStream API and decodes its answer, refusing an answer that carries an error. +func Request(ctx context.Context, nc *nats.Conn, subject string, body any, into any) error { + var payload []byte + if body != nil { + var err error + if payload, err = json.Marshal(body); err != nil { + return err + } + } + rctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + msg, err := nc.RequestWithContext(rctx, subject, payload) + if err != nil { + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, nats.ErrNoResponders) { + return fmt.Errorf("%s was not answered (%v): the bus refuses a subject this user is not granted by not answering it", subject, err) + } + return fmt.Errorf("%s: %w", subject, err) + } + var wrapped struct { + Error *APIError `json:"error"` + } + if err := json.Unmarshal(msg.Data, &wrapped); err != nil { + return fmt.Errorf("%s answered something that is not JSON: %q", subject, Truncate(msg.Data)) + } + if wrapped.Error != nil { + return fmt.Errorf("%s: %w", subject, wrapped.Error) + } + if into != nil { + return json.Unmarshal(msg.Data, into) + } + return nil +} + +func Truncate(b []byte) string { + if len(b) > 200 { + return string(b[:200]) + "…" + } + return string(b) +} + +// StreamNames is every stream in the account, in name order, all pages of them. +func StreamNames(ctx context.Context, nc *nats.Conn) ([]string, error) { + var names []string + for { + var page struct { + Total int `json:"total"` + Offset int `json:"offset"` + Limit int `json:"limit"` + Streams []string `json:"streams"` + } + if err := Request(ctx, nc, APIStreamNames, map[string]int{"offset": len(names)}, &page); err != nil { + return nil, err + } + names = append(names, page.Streams...) + if len(page.Streams) == 0 || len(names) >= page.Total { + break + } + } + sort.Strings(names) + return names, nil +} + +// Snapshot writes every stream, one at a time, as a tar: the manifest first, then each stream's +// meta and archive under streams//. Any stream failing fails the whole: a partial copy of the +// bus written where a whole one is expected is the silent failure this exists to prevent. +func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptions) (*busarchive.Manifest, error) { + if o.StreamTimeout <= 0 { + o.StreamTimeout = 10 * time.Minute + } + if o.Log == nil { + o.Log = func(string, ...any) {} + } + m := &busarchive.Manifest{Format: 1, Server: nc.ConnectedServerVersion(), Started: time.Now().UTC()} + names, err := StreamNames(ctx, nc) + if err != nil { + return nil, err + } + if len(o.Streams) > 0 { + var only []string + for _, n := range names { + if slices.Contains(o.Streams, n) { + only = append(only, n) + } + } + if len(only) != len(o.Streams) { + return nil, fmt.Errorf("the bus holds %v of the streams %v asked for: a snapshot missing one would pass for a whole one", only, o.Streams) + } + names = only + } + if len(names) == 0 { + // The mesh's own streams always exist once the controller has run; a bus with none is not + // one to call backed up. + return nil, errors.New("the bus holds no streams; a snapshot of nothing would pass for a backup") + } + + type taken struct { + entry busarchive.StreamEntry + meta []byte + archive *os.File + } + var files []taken + defer func() { + for _, t := range files { + t.archive.Close() + os.Remove(t.archive.Name()) + } + }() + + for i, name := range names { + if i > 0 && o.Pause > 0 { + select { + case <-time.After(o.Pause): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + var info struct { + Config streamConfig `json:"config"` + } + if err := Request(ctx, nc, APIStreamInfo+name, nil, &info); err != nil { + return nil, err + } + if info.Config.Storage == "memory" { + m.Skipped = append(m.Skipped, busarchive.Skipped{Name: name, Why: "memory storage: it holds nothing across a restart, and the server cannot snapshot it"}) + o.Log("%s: skipped, memory storage", name) + continue + } + f, err := os.CreateTemp(o.TempDir, "mesh-nats-snapshot-*") + if err != nil { + return nil, err + } + files = append(files, taken{archive: f}) + sctx, cancel := context.WithTimeout(ctx, o.StreamTimeout) + entry, meta, err := snapshotStream(sctx, nc, name, f) + cancel() + if err != nil { + return nil, fmt.Errorf("stream %s: %w", name, err) + } + files[len(files)-1].entry, files[len(files)-1].meta = entry, meta + m.Messages += entry.Messages + m.StreamBytes += int64(entry.Bytes) + m.ArchiveBytes += entry.ArchiveBytes + o.Log("%s: %d message(s), %s, last sequence %d, %d consumer(s) → %s in %.2fs", + name, entry.Messages, Human(int64(entry.Bytes)), entry.LastSeq, entry.Consumers, Human(entry.ArchiveBytes), entry.Seconds) + } + m.Finished = time.Now().UTC() + m.Seconds = m.Finished.Sub(m.Started).Seconds() + for _, t := range files { + m.Streams = append(m.Streams, t.entry) + } + + tw := tar.NewWriter(out) + manifest, err := json.MarshalIndent(m, "", " ") + if err != nil { + return nil, err + } + if err := writeEntry(tw, busarchive.ManifestName, append(manifest, '\n'), m.Finished); err != nil { + return nil, err + } + for _, t := range files { + if err := writeEntry(tw, t.entry.Meta, t.meta, m.Finished); err != nil { + return nil, err + } + if _, err := t.archive.Seek(0, io.SeekStart); err != nil { + return nil, err + } + hdr := &tar.Header{Name: t.entry.Archive, Mode: 0o600, Size: t.entry.ArchiveBytes, ModTime: m.Finished, Format: tar.FormatPAX} + if err := tw.WriteHeader(hdr); err != nil { + return nil, err + } + if _, err := io.Copy(tw, t.archive); err != nil { + return nil, err + } + } + if err := tw.Close(); err != nil { + return nil, err + } + return m, nil +} + +func writeEntry(tw *tar.Writer, name string, body []byte, at time.Time) error { + if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(body)), ModTime: at, Format: tar.FormatPAX}); err != nil { + return err + } + _, err := tw.Write(body) + return err +} + +// snapshotStream takes one stream's snapshot into f: the server's chunks, each acknowledged as it +// arrives so the server's window keeps moving, until the server's last, empty message. It answers +// the stream's entry (the archive's size and checksum included) and its meta. +// +// The protocol is the server's (nats-server jetstream_api.go, streamSnapshot): the request names a +// subject to deliver to; the server answers with the stream's configuration and its state at the +// snapshot, then sends the archive in chunks whose reply subject is the acknowledgement it waits for +// past its window — two seconds without one and it gives up with "408 No Flow Response". The end is +// an empty message whose status header says 204, or the error that ended it. +func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) (busarchive.StreamEntry, []byte, error) { + started := time.Now() + deliver := nc.NewRespInbox() + + hash := sha256.New() + var written int64 + var mu sync.Mutex + var failed error + done := make(chan struct{}) + var once sync.Once + finish := func(err error) { + mu.Lock() + if failed == nil { + failed = err + } + mu.Unlock() + once.Do(func() { close(done) }) + } + sub, err := nc.Subscribe(deliver, func(msg *nats.Msg) { + if len(msg.Data) == 0 { + if status := msg.Header.Get("Status"); status != "" && status != "204" { + finish(fmt.Errorf("the server ended the snapshot: %s %s", status, msg.Header.Get("Description"))) + return + } + finish(nil) + return + } + // Acknowledged before it is written: the chunk is already in memory, and a slow disk must not + // read to the server as a reader that went away. + if msg.Reply != "" { + if err := nc.Publish(msg.Reply, nil); err != nil { + finish(fmt.Errorf("acknowledging a chunk: %w", err)) + return + } + } + if _, err := f.Write(msg.Data); err != nil { + finish(fmt.Errorf("writing the archive: %w", err)) + return + } + hash.Write(msg.Data) + written += int64(len(msg.Data)) + }) + if err != nil { + return busarchive.StreamEntry{}, nil, err + } + defer sub.Unsubscribe() + // Every chunk is held until the callback has it; the window bounds how many there can be. + if err := sub.SetPendingLimits(-1, -1); err != nil { + return busarchive.StreamEntry{}, nil, err + } + if err := nc.FlushWithContext(ctx); err != nil { + return busarchive.StreamEntry{}, nil, err + } + + var resp struct { + Config json.RawMessage `json:"config"` + State json.RawMessage `json:"state"` + } + req := map[string]any{"deliver_subject": deliver, "chunk_size": ChunkSize} + if err := Request(ctx, nc, APIStreamSnapshot+name, req, &resp); err != nil { + return busarchive.StreamEntry{}, nil, err + } + select { + case <-done: + case <-ctx.Done(): + return busarchive.StreamEntry{}, nil, fmt.Errorf("not finished within its bound: %w", ctx.Err()) + } + // The subscription is drained before the archive is read back, so no callback is still writing. + _ = sub.Unsubscribe() + mu.Lock() + err = failed + mu.Unlock() + if err != nil { + return busarchive.StreamEntry{}, nil, err + } + if written == 0 { + return busarchive.StreamEntry{}, nil, errors.New("the server sent an empty archive") + } + if err := f.Sync(); err != nil { + return busarchive.StreamEntry{}, nil, err + } + + var cfg streamConfig + var state StreamState + if err := json.Unmarshal(resp.Config, &cfg); err != nil { + return busarchive.StreamEntry{}, nil, fmt.Errorf("its configuration: %w", err) + } + if err := json.Unmarshal(resp.State, &state); err != nil { + return busarchive.StreamEntry{}, nil, fmt.Errorf("its state: %w", err) + } + // The archive reads to its end — s2 frames, then a tar — before it is called a snapshot. + if _, err := f.Seek(0, io.SeekStart); err != nil { + return busarchive.StreamEntry{}, nil, err + } + if err := busarchive.ReadsThrough(f); err != nil { + return busarchive.StreamEntry{}, nil, fmt.Errorf("the archive the server sent does not read: %w", err) + } + + // The meta, in the shape the restore API takes and the nats CLI writes (backup.json): the + // server's own configuration and state, kept as raw JSON so a field this program does not know + // is never dropped on the way back. + meta, err := json.MarshalIndent(map[string]json.RawMessage{"config": resp.Config, "state": resp.State}, "", " ") + if err != nil { + return busarchive.StreamEntry{}, nil, err + } + metaSum := sha256.Sum256(meta) + return busarchive.StreamEntry{ + Name: name, Subjects: cfg.Subjects, Retention: cfg.Retention, + Messages: state.Messages, Bytes: state.Bytes, FirstSeq: state.FirstSeq, LastSeq: state.LastSeq, + Deleted: state.NumDelete, Consumers: state.Consumers, + Meta: "streams/" + name + "/backup.json", MetaSHA256: hex.EncodeToString(metaSum[:]), + Archive: "streams/" + name + "/stream.tar.s2", ArchiveBytes: written, + ArchiveSHA256: hex.EncodeToString(hash.Sum(nil)), Seconds: time.Since(started).Seconds(), + }, meta, nil +} + +// Human is a size in bytes as a person reads it: 1.5 MiB. +func Human(n int64) string { + const unit = 1024 + if n < unit { + return fmt.Sprintf("%d B", n) + } + div, exp := int64(unit), 0 + for v := n / unit; v >= unit; v /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) +} + +// Into is the keeper's snapshot of some streams before it rebuilds or removes them (design 25 § 12.3): one tar under +// dir, named for the streams and the time, written whole before its name is given, so a partial one is never taken +// for it. It answers the file's path. +func Into(nc *nats.Conn, dir string) func(ctx context.Context, streams []string) (string, error) { + return func(ctx context.Context, streams []string) (string, error) { + if dir == "" { + return "", errors.New("no snapshots directory is set (MESH_NATS_SNAPSHOTS_DIR), so no snapshot is taken and nothing is removed") + } + name := filepath.Join(dir, "entry-"+strings.Join(streams, "+")+"-"+time.Now().UTC().Format("20060102T150405Z")+".tar") + f, err := os.CreateTemp(dir, ".entry-*.partial") + if err != nil { + return "", err + } + defer os.Remove(f.Name()) + if _, err := Snapshot(ctx, nc, f, SnapshotOptions{Streams: streams, TempDir: dir}); err != nil { + f.Close() + return "", err + } + if err := f.Close(); err != nil { + return "", err + } + return name, os.Rename(f.Name(), name) + } +} diff --git a/go/go.mod b/go/go.mod index 830ee762..2b853831 100644 --- a/go/go.mod +++ b/go/go.mod @@ -3,6 +3,7 @@ module git.novox.be/novox/mesh-sdk/go go 1.26.0 require ( + github.com/klauspost/compress v1.20.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) @@ -10,7 +11,6 @@ require ( require ( github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect - github.com/klauspost/compress v1.20.0 // indirect github.com/minio/highwayhash v1.0.4 // indirect github.com/nats-io/jwt/v2 v2.8.1 // indirect github.com/nats-io/nkeys v0.4.16 // indirect diff --git a/go/packages_list_test.go b/go/packages_list_test.go index c4246809..5366d57e 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -23,7 +23,7 @@ import ( // Logic only some modules need is its owner module's package. A package added here is a decision to say in the pull // request: does every module share it, or is it a shape more than one part reads? var meshSDKPackages = []string{ - ".", "announce", "announcetest", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", + ".", "announce", "announcetest", "busarchive", "buscalls", "buscompose", "busstate", "bussnap", "busretry", "calladdress", "catalogue", "declaration", "forgekey", "health", "internal/beside", "keys", "kvlimit", "membership", "motion", "objectlink", "outward", "pernode", "process", "sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "stateobjects", "testbus", "views", "views/viewstest",