Verify the bus's snapshot with a tool, sharing the snapshot program's archive code (issue 1842)

The upgrade and restore steps told a person to docker exec the snapshot program; no mesh tool
did it. nats_snapshot_verify reads the newest bus.tar of the snapshots directory only.

modules/nats/cmd/mesh-nats-snapshot/archive.go

@@ -1,162 +0,0 @@
-package main
-
-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
-}

modules/nats/cmd/mesh-nats-snapshot/image_test.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"nats-tools/internal/busarchive"
 )
 
 // manifestOf is the nats module's own definition, beside this program.
@@ -172,7 +173,7 @@ func TestTheDeclaredDumpRunsInTheBussImage(t *testing.T) {
 	if err != nil {
 		t.Fatalf("the dump left no bus.tar: %v", err)
 	}
-	a, err := ReadArchive(f, "")
+	a, err := busarchive.ReadArchive(f, "")
 	f.Close()
 	if err != nil {
 		t.Fatal(err)

modules/nats/cmd/mesh-nats-snapshot/live_test.go

@@ -35,6 +35,7 @@ import (
 	"time"
 
 	"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).
@@ -296,7 +297,7 @@ func fill(t *testing.T, js nats.JetStreamContext) {
 // sameContent holds every stream the manifest names, in the restored server, to the source: the same
 // state at the snapshot, every message by sequence (subject, headers, body, and a deleted one still
 // deleted), and the durable consumers where they were.
-func sameContent(t *testing.T, m *Manifest, src, dst nats.JetStreamContext) {
+func sameContent(t *testing.T, m *busarchive.Manifest, src, dst nats.JetStreamContext) {
 	t.Helper()
 	for _, s := range m.Streams {
 		info, err := dst.StreamInfo(s.Name)
@@ -464,7 +465,7 @@ func TestASnapshotOfALiveBusRestoresToTheSameContent(t *testing.T) {
 	if err := os.WriteFile(tarPath, archive.Bytes(), 0o644); err != nil {
 		t.Fatal(err)
 	}
-	a, err := ReadArchive(bytes.NewReader(archive.Bytes()), "")
+	a, err := busarchive.ReadArchive(bytes.NewReader(archive.Bytes()), "")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -517,7 +518,7 @@ func TestASnapshotOfALiveBusRestoresToTheSameContent(t *testing.T) {
 		damaged := append([]byte(nil), archive.Bytes()...)
 		// A byte in the last stream's archive, well past every header.
 		damaged[len(damaged)-2048] ^= 0xff
-		d, err := ReadArchive(bytes.NewReader(damaged), "")
+		d, err := busarchive.ReadArchive(bytes.NewReader(damaged), "")
 		if err == nil {
 			defer d.Close()
 			err = d.Verify()

modules/nats/cmd/mesh-nats-snapshot/main.go

@@ -29,6 +29,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"nats-tools/internal/busarchive"
 	"os"
 	"time"
 )
@@ -114,7 +115,7 @@ func runVerify(args []string) error {
 		return err
 	}
 	defer closer()
-	a, err := ReadArchive(in, "")
+	a, err := busarchive.ReadArchive(in, "")
 	if err != nil {
 		return err
 	}
@@ -146,7 +147,7 @@ func runRestore(args []string) error {
 		return err
 	}
 	defer closer()
-	a, err := ReadArchive(in, "")
+	a, err := busarchive.ReadArchive(in, "")
 	if err != nil {
 		return err
 	}

modules/nats/cmd/mesh-nats-snapshot/restore.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"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.
@@ -25,7 +26,7 @@ const restoreChunk = 64 * 1024
 // RestoreAll puts every stream of the archive into the server nc reaches, one at a time, and holds
 // each restored stream's state to the manifest. **Nothing is restored over anything**: a server that
 // already holds any of the streams is refused before the first is sent.
-func RestoreAll(ctx context.Context, nc *nats.Conn, a *Archive, log func(string, ...any)) error {
+func RestoreAll(ctx context.Context, nc *nats.Conn, a *busarchive.Archive, log func(string, ...any)) error {
 	var present []string
 	for _, s := range a.Manifest.Streams {
 		err := request(ctx, nc, apiStreamInfo+s.Name, nil, nil)
@@ -52,8 +53,8 @@ func RestoreAll(ctx context.Context, nc *nats.Conn, a *Archive, log func(string,
 	return nil
 }
 
-func restoreStream(ctx context.Context, nc *nats.Conn, a *Archive, s StreamEntry) error {
-	meta, archive, err := a.open(s)
+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
 	}
@@ -123,7 +124,7 @@ func restoreStream(ctx context.Context, nc *nats.Conn, a *Archive, s StreamEntry
 // the messages in that tail some may be there and some not. Every message up to the manifest's last
 // sequence is there, exactly; that is the snapshot's promise, and what the test holds it to message
 // by message. What must never happen is a restored stream that ends before the manifest said it did.
-func heldTo(restored streamState, s StreamEntry) error {
+func heldTo(restored streamState, s busarchive.StreamEntry) error {
 	if restored.LastSeq < s.LastSeq || (s.Messages > 0 && restored.Messages == 0) {
 		return fmt.Errorf("restored with %d message(s) up to sequence %d; the snapshot held %d up to %d",
 			restored.Messages, restored.LastSeq, s.Messages, s.LastSeq)
@@ -140,8 +141,8 @@ func heldTo(restored streamState, s StreamEntry) error {
 // Why a server of its own rather than the live bus: a stream is restored only where it does not
 // exist, and on the live bus every stream exists — the controller asserts them on every start. So a
 // restore beside the live data, swapped in by a person, is the only one that never touches it.
-func RestoreOffline(ctx context.Context, a *Archive, into, account string, log func(string, ...any)) (string, error) {
-	if !validName(account) {
+func RestoreOffline(ctx context.Context, a *busarchive.Archive, into, account string, log func(string, ...any)) (string, error) {
+	if !busarchive.ValidName(account) {
 		return "", fmt.Errorf("%q is not an account name", account)
 	}
 	if entries, err := os.ReadDir(into); err == nil && len(entries) > 0 {

modules/nats/cmd/mesh-nats-snapshot/snapshot.go

@@ -11,12 +11,11 @@ import (
 	"io"
 	"os"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
-	"github.com/klauspost/compress/s2"
 	"github.com/nats-io/nats.go"
+	"nats-tools/internal/busarchive"
 )
 
 // The JetStream API's subjects this program speaks. Each is granted to the bus's own module and
@@ -45,56 +44,6 @@ type SnapshotOptions struct {
 	Log     func(format string, args ...any)
 }
 
-// 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"
-
 // apiError is the error every JetStream API answer may carry.
 type apiError struct {
 	Code        int    `json:"code"`
@@ -187,14 +136,14 @@ func StreamNames(ctx context.Context, nc *nats.Conn) ([]string, error) {
 // Snapshot writes every stream, one at a time, as a tar: the manifest first, then each stream's
 // meta and archive under streams/<name>/. 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) (*Manifest, error) {
+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 := &Manifest{Format: 1, Server: nc.ConnectedServerVersion(), Started: time.Now().UTC()}
+	m := &busarchive.Manifest{Format: 1, Server: nc.ConnectedServerVersion(), Started: time.Now().UTC()}
 	names, err := StreamNames(ctx, nc)
 	if err != nil {
 		return nil, err
@@ -206,7 +155,7 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio
 	}
 
 	type taken struct {
-		entry   StreamEntry
+		entry   busarchive.StreamEntry
 		meta    []byte
 		archive *os.File
 	}
@@ -233,7 +182,7 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio
 			return nil, err
 		}
 		if info.Config.Storage == "memory" {
-			m.Skipped = append(m.Skipped, Skipped{Name: name, Why: "memory storage: it holds nothing across a restart, and the server cannot snapshot it"})
+			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
 		}
@@ -266,7 +215,7 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio
 	if err != nil {
 		return nil, err
 	}
-	if err := writeEntry(tw, ManifestName, append(manifest, '\n'), m.Finished); err != nil {
+	if err := writeEntry(tw, busarchive.ManifestName, append(manifest, '\n'), m.Finished); err != nil {
 		return nil, err
 	}
 	for _, t := range files {
@@ -307,7 +256,7 @@ func writeEntry(tw *tar.Writer, name string, body []byte, at time.Time) error {
 // 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) (StreamEntry, []byte, error) {
+func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) (busarchive.StreamEntry, []byte, error) {
 	started := time.Now()
 	deliver := nc.NewRespInbox()
 
@@ -350,15 +299,15 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File)
 		written += int64(len(msg.Data))
 	})
 	if err != nil {
-		return StreamEntry{}, nil, err
+		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 StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 	if err := nc.FlushWithContext(ctx); err != nil {
-		return StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 
 	var resp struct {
@@ -367,12 +316,12 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File)
 	}
 	req := map[string]any{"deliver_subject": deliver, "chunk_size": ChunkSize}
 	if err := request(ctx, nc, apiStreamSnapshot+name, req, &resp); err != nil {
-		return StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 	select {
 	case <-done:
 	case <-ctx.Done():
-		return StreamEntry{}, nil, fmt.Errorf("not finished within its bound: %w", ctx.Err())
+		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()
@@ -380,29 +329,29 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File)
 	err = failed
 	mu.Unlock()
 	if err != nil {
-		return StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 	if written == 0 {
-		return StreamEntry{}, nil, errors.New("the server sent an empty archive")
+		return busarchive.StreamEntry{}, nil, errors.New("the server sent an empty archive")
 	}
 	if err := f.Sync(); err != nil {
-		return StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 
 	var cfg streamConfig
 	var state streamState
 	if err := json.Unmarshal(resp.Config, &cfg); err != nil {
-		return StreamEntry{}, nil, fmt.Errorf("its configuration: %w", err)
+		return busarchive.StreamEntry{}, nil, fmt.Errorf("its configuration: %w", err)
 	}
 	if err := json.Unmarshal(resp.State, &state); err != nil {
-		return StreamEntry{}, nil, fmt.Errorf("its state: %w", err)
+		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 StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
-	if err := readsThrough(f); err != nil {
-		return StreamEntry{}, nil, fmt.Errorf("the archive the server sent does not read: %w", 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
@@ -410,10 +359,10 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File)
 	// 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 StreamEntry{}, nil, err
+		return busarchive.StreamEntry{}, nil, err
 	}
 	metaSum := sha256.Sum256(meta)
-	return StreamEntry{
+	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,
@@ -422,25 +371,3 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File)
 		ArchiveSHA256: hex.EncodeToString(hash.Sum(nil)), Seconds: time.Since(started).Seconds(),
 	}, meta, nil
 }
-
-// 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 *>")
-}

modules/nats/cmd/nats-tools/main.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"nats-tools/internal/busarchive"
 )
 
 func str(description string) map[string]any {
@@ -140,6 +141,12 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 				varz, verr := get("/varz")
 				return InForce(answer, at, werr, varz, verr), nil
 			}},
+		{Name: "nats_snapshot_verify",
+			Description: "Verifies the bus's snapshot (novox/hq ADR 0235) before an upgrade or a restore: the newest bus.tar in the module's snapshots directory, or the one named, with every file held to the manifest's checksum and every archive read to its end. Answers the file, its time and size, each stream's name and last sequence, `passed`, and the first fault when it did not. Reads only; nothing outside the snapshots directory.",
+			Input:       map[string]any{"file": str("a snapshot file inside the snapshots directory, e.g. bus.tar (default: the newest .tar there)")},
+			Run: func(a map[string]any) (any, error) {
+				return busarchive.VerifyFile(env("MESH_NATS_SNAPSHOTS_DIR", ""), arg(a, "file"))
+			}},
 	}
 }