diff --git a/modules/nats/cmd/mesh-nats-snapshot/bussnap.go b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go index e4d8a8f59..0c205ed83 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/bussnap.go +++ b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go @@ -1,6 +1,6 @@ package main -import "nats-tools/internal/bussnap" +import "git.novox.be/novox/mesh-sdk/go/bussnap" // The snapshot itself is bussnap's, which the keeper takes one entry's with too (design 25 § 12.3). type ( diff --git a/modules/nats/cmd/mesh-nats-snapshot/image_test.go b/modules/nats/cmd/mesh-nats-snapshot/image_test.go index d53fcee36..ddf3b86ca 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/image_test.go +++ b/modules/nats/cmd/mesh-nats-snapshot/image_test.go @@ -19,8 +19,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/busarchive" "github.com/nats-io/nats.go" - "nats-tools/internal/busarchive" ) // manifestOf is the nats module's own definition, beside this program. diff --git a/modules/nats/cmd/mesh-nats-snapshot/live_test.go b/modules/nats/cmd/mesh-nats-snapshot/live_test.go index cd64c0fc7..cbea74248 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/live_test.go +++ b/modules/nats/cmd/mesh-nats-snapshot/live_test.go @@ -34,8 +34,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/busarchive" "github.com/nats-io/nats.go" - "nats-tools/internal/busarchive" ) // image is the bus's release (the manifest's server artifact names the same release in its tag). diff --git a/modules/nats/cmd/mesh-nats-snapshot/main.go b/modules/nats/cmd/mesh-nats-snapshot/main.go index 983b33393..4ff65226b 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/main.go +++ b/modules/nats/cmd/mesh-nats-snapshot/main.go @@ -30,10 +30,11 @@ import ( "flag" "fmt" "io" - "nats-tools/internal/busarchive" - "nats-tools/internal/bussnap" "os" "time" + + "git.novox.be/novox/mesh-sdk/go/busarchive" + "git.novox.be/novox/mesh-sdk/go/bussnap" ) func main() { diff --git a/modules/nats/cmd/mesh-nats-snapshot/restore.go b/modules/nats/cmd/mesh-nats-snapshot/restore.go index 5d5866094..f56c80d05 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/restore.go +++ b/modules/nats/cmd/mesh-nats-snapshot/restore.go @@ -16,8 +16,8 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-sdk/go/busarchive" "github.com/nats-io/nats.go" - "nats-tools/internal/busarchive" ) // restoreChunk is what is sent at a time; the server answers each before the next goes. diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 37c18e506..2547bf5b1 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -29,7 +29,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/views" - "nats-tools/internal/bussnap" + "git.novox.be/novox/mesh-sdk/go/bussnap" "nats-tools/internal/keeper" ) diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index bb06a2a05..63e38779f 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -28,7 +28,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "nats-tools/internal/bussnap" + "git.novox.be/novox/mesh-sdk/go/bussnap" "nats-tools/internal/keeper" ) diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 535eeb75b..b20eefcba 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -11,7 +11,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" - "nats-tools/internal/busarchive" + "git.novox.be/novox/mesh-sdk/go/busarchive" "nats-tools/internal/keeper" ) diff --git a/modules/nats/internal/busarchive/archive.go b/modules/nats/internal/busarchive/archive.go deleted file mode 100644 index 0d14f545a..000000000 --- a/modules/nats/internal/busarchive/archive.go +++ /dev/null @@ -1,162 +0,0 @@ -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/modules/nats/internal/busarchive/types.go b/modules/nats/internal/busarchive/types.go deleted file mode 100644 index 12ca5eba3..000000000 --- a/modules/nats/internal/busarchive/types.go +++ /dev/null @@ -1,86 +0,0 @@ -// 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/modules/nats/internal/busarchive/verifyfile.go b/modules/nats/internal/busarchive/verifyfile.go deleted file mode 100644 index fdb4a2fcf..000000000 --- a/modules/nats/internal/busarchive/verifyfile.go +++ /dev/null @@ -1,109 +0,0 @@ -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/modules/nats/internal/busarchive/verifyfile_test.go b/modules/nats/internal/busarchive/verifyfile_test.go deleted file mode 100644 index be62737fb..000000000 --- a/modules/nats/internal/busarchive/verifyfile_test.go +++ /dev/null @@ -1,173 +0,0 @@ -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/modules/nats/internal/bussnap/restore.go b/modules/nats/internal/bussnap/restore.go deleted file mode 100644 index 030f8d7b1..000000000 --- a/modules/nats/internal/bussnap/restore.go +++ /dev/null @@ -1,158 +0,0 @@ -package bussnap - -import ( - "archive/tar" - "context" - "encoding/json" - "errors" - "fmt" - "io" - "os" - "time" - - "nats-tools/internal/busarchive" - - "github.com/nats-io/nats.go" -) - -// The keeper's restore in place of a snapshot it took (design 47 G §681): each stream of the file put back under its own -// name, only where no stream of that name is live. - -// APIStreamDelete is the subject the keeper removes its own partial stream with. -const APIStreamDelete = "$JS.API.STREAM.DELETE." - -const restoreChunk = 64 * 1024 - -// ManifestOf reads the manifest of a snapshot file: the streams it holds, without unpacking them. -func ManifestOf(path string) (*busarchive.Manifest, error) { - f, err := os.Open(path) - if err != nil { - return nil, err - } - defer f.Close() - tr := tar.NewReader(f) - hdr, err := tr.Next() - if err != nil { - return nil, fmt.Errorf("the snapshot does not read: %w", err) - } - if hdr.Name != busarchive.ManifestName { - return nil, fmt.Errorf("the snapshot starts with %q, not its manifest", hdr.Name) - } - raw, err := io.ReadAll(io.LimitReader(tr, 64<<20)) - if err != nil { - return nil, err - } - var m busarchive.Manifest - if err := json.Unmarshal(raw, &m); err != nil { - return nil, fmt.Errorf("its manifest: %w", err) - } - return &m, nil -} - -// Live says whether a stream of that name exists. -func Live(ctx context.Context, nc *nats.Conn, name string) (bool, error) { - err := Request(ctx, nc, APIStreamInfo+name, nil, nil) - var api *APIError - switch { - case err == nil: - return true, nil - case errors.As(err, &api) && api.Code == 404: - return false, nil - } - return false, err -} - -// DeleteStream removes a stream; one that is not there is no error. -func DeleteStream(ctx context.Context, nc *nats.Conn, name string) error { - err := Request(ctx, nc, APIStreamDelete+name, nil, nil) - var api *APIError - if errors.As(err, &api) && api.Code == 404 { - return nil - } - return err -} - -// RestoreFile puts every stream of the snapshot file back under its own name, each sent in chunks the server answers; it -// ends with the context's bound. A stream already there is the caller's to have refused beforehand. -func RestoreFile(ctx context.Context, nc *nats.Conn, path, scratch string, log func(string, ...any)) error { - f, err := os.Open(path) - if err != nil { - return err - } - defer f.Close() - a, err := busarchive.ReadArchive(f, scratch) - if err != nil { - return err - } - defer a.Close() - if err := a.Verify(); err != nil { - return err - } - for _, s := range a.Manifest.Streams { - started := time.Now() - if err := restoreStream(ctx, nc, a, s); err != nil { - return fmt.Errorf("stream %s: %w", s.Name, err) - } - log("%s: restored in %.2fs", s.Name, time.Since(started).Seconds()) - } - return nil -} - -func restoreStream(ctx context.Context, nc *nats.Conn, a *busarchive.Archive, s busarchive.StreamEntry) error { - meta, archive, err := a.Open(s) - if err != nil { - return err - } - defer archive.Close() - var req map[string]json.RawMessage - if err := json.Unmarshal(meta, &req); err != nil || req["config"] == nil { - return errors.New("its meta is not a stream's configuration and state") - } - var resp struct { - DeliverSubject string `json:"deliver_subject"` - } - if err := Request(ctx, nc, APIStreamRestore+s.Name, req, &resp); err != nil { - return err - } - if resp.DeliverSubject == "" { - return errors.New("the server gave nowhere to send the archive") - } - buf := make([]byte, restoreChunk) - for { - n, err := archive.Read(buf) - if n > 0 { - reply, rerr := nc.RequestWithContext(ctx, resp.DeliverSubject, buf[:n]) - if rerr != nil { - return fmt.Errorf("sending the archive: %w", rerr) - } - if len(reply.Data) > 0 { - return fmt.Errorf("the server refused the archive: %s", Truncate(reply.Data)) - } - } - if err == io.EOF { - break - } - if err != nil { - return err - } - } - final, err := nc.RequestWithContext(ctx, resp.DeliverSubject, nil) - if err != nil { - return fmt.Errorf("finishing the restore: %w", err) - } - var created struct { - Error *APIError `json:"error"` - State StreamState `json:"state"` - } - if err := json.Unmarshal(final.Data, &created); err != nil { - return fmt.Errorf("the server's last answer is not JSON: %q", Truncate(final.Data)) - } - if created.Error != nil { - return created.Error - } - if created.State.LastSeq < s.LastSeq || (s.Messages > 0 && created.State.Messages == 0) { - return fmt.Errorf("restored with %d message(s) up to sequence %d; the snapshot held %d up to %d", - created.State.Messages, created.State.LastSeq, s.Messages, s.LastSeq) - } - return nil -} diff --git a/modules/nats/internal/bussnap/snapshot.go b/modules/nats/internal/bussnap/snapshot.go deleted file mode 100644 index dd4517d40..000000000 --- a/modules/nats/internal/bussnap/snapshot.go +++ /dev/null @@ -1,433 +0,0 @@ -// 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" - - "nats-tools/internal/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("20060102T150405.000000Z")+".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/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index 5ff2d51c0..c3e20079f 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/modules/nats/internal/keeper/jetstream.go @@ -10,7 +10,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "nats-tools/internal/bussnap" + "git.novox.be/novox/mesh-sdk/go/bussnap" ) // JetStream is the bus as the keeper's own account reaches it (Bus): each object a stream, a key-value bucket or an