Run the bus from its pinned upstream image with the snapshot program and the entrypoint mounted, and write and rehearse the way there and back (issues 1640, 1641, 1642, ADR 0480 clauses 1-3, 5, 6, 11)

2026-10-22 09:55:03 +02:00
parent 0ec446ab75
commit b4e90e5fce
22 changed files with 682 additions and 266 deletions
@@ -0,0 +1,162 @@
+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
+}
@@ -0,0 +1,105 @@
+package main
+
+import (
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Credential is what the mesh seals to a module so it can reach the bus (mesh-controller
+// cmd/mesh-controller issueWith): where, the server's certificate by fingerprint, who, and the
+// password. The address is not used here — this runs beside the server, and the server is reached on
+// loopback — but the fingerprint is: loopback or not, the certificate is the mesh's own.
+type Credential struct {
+	URL         string `json:"url"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+}
+
+func readCredential(path string) (Credential, error) {
+	var raw []byte
+	var err error
+	if path == "-" {
+		raw, err = io.ReadAll(io.LimitReader(os.Stdin, 1<<20))
+	} else {
+		raw, err = os.ReadFile(path)
+	}
+	if err != nil {
+		return Credential{}, fmt.Errorf("cannot read the credential: %w", err)
+	}
+	var c Credential
+	if err := json.Unmarshal(raw, &c); err != nil {
+		return Credential{}, errors.New("the credential is not the JSON the mesh seals to a module (url, fingerprint, user, password)")
+	}
+	if c.User == "" || c.Password == "" {
+		return Credential{}, errors.New("the credential names no user or password, so the bus would refuse it")
+	}
+	return c, nil
+}
+
+// connect dials the bus as the credential's user. A `tls://` server is held to the credential's
+// fingerprint and nothing else — the mesh's bus presents a certificate of its own, in no trust store,
+// so a name check could only fail or be skipped. A plain `nats://` server is accepted only on
+// loopback, and is what a restore's own private server is.
+func connect(server string, c Credential) (*nats.Conn, error) {
+	opts := []nats.Option{
+		nats.UserInfo(c.User, c.Password),
+		nats.Name("mesh-nats-snapshot"),
+		// Answers come to the user's own inbox and no other: the bus grants every user its own
+		// `_INBOX.<user>.>` and nothing wider (novox/hq design 25 §4).
+		nats.CustomInboxPrefix("_INBOX." + c.User),
+		nats.Timeout(10 * time.Second),
+		// One attempt: a snapshot that cannot reach the bus fails the night, loudly, rather than
+		// waiting for it.
+		nats.NoReconnect(),
+	}
+	switch {
+	case strings.HasPrefix(server, "tls://"):
+		if c.Fingerprint == "" {
+			return nil, errors.New("the credential carries no fingerprint, so the bus's certificate cannot be checked")
+		}
+		opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
+	case strings.HasPrefix(server, "nats://127.0.0.1:"), strings.HasPrefix(server, "nats://localhost:"):
+		if c.Fingerprint != "" {
+			// The mesh's bus requires TLS; a plain connection to it would be refused anyway.
+			opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
+		}
+	default:
+		return nil, fmt.Errorf("%s: the bus is reached over tls:// (or plain nats:// on loopback, for a restore's own server)", server)
+	}
+	nc, err := nats.Connect(server, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("cannot reach the bus at %s as %s: %w", server, c.User, err)
+	}
+	return nc, nil
+}
+
+// pinnedTo accepts exactly the certificate with this SHA-256 — the same pin every machine holds
+// (mesh-controller internal/broker PinnedToFingerprint).
+func pinnedTo(want string) *tls.Config {
+	return &tls.Config{
+		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
+		MinVersion:         tls.VersionTLS12,
+		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
+			if len(rawCerts) == 0 {
+				return errors.New("the bus presented no certificate")
+			}
+			sum := sha256.Sum256(rawCerts[0])
+			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
+				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s)", got)
+			}
+			return nil
+		},
+	}
+}
@@ -0,0 +1,309 @@
+package main
+
+// The declared dump and the mounted entrypoint, exactly as the manifest declares them, against the upstream image
+// the manifest pins: the mesh's snapshot program and entrypoint mounted at /opt (novox/hq ADR 0480 clauses 1, 2, 5).
+//
+//	MESH_TEST_DOCKER=1 go test -run 'TestTheDeclaredDump|TestTheEntrypoint' .
+//
+// The image is pulled by the manifest's reference, so the server these run is the one the bus runs.
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// manifestOf is the nats module's own definition, beside this program.
+func manifestOf(t *testing.T) map[string]any {
+	t.Helper()
+	raw, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m map[string]any
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	return m
+}
+
+func declaredDump(t *testing.T, m map[string]any) string {
+	t.Helper()
+	for _, item := range m["data"].(map[string]any)["own"].([]any) {
+		it := item.(map[string]any)
+		if it["id"] != "jetstream" {
+			continue
+		}
+		backup, _ := it["backup"].(map[string]any)
+		if backup == nil || backup["into"] != "snapshots" {
+			t.Fatalf("the bus's streams are not protected by a dump into its snapshots: %v", it["backup"])
+		}
+		return backup["dump"].(string)
+	}
+	t.Fatal("the nats module declares no jetstream data item")
+	return ""
+}
+
+// serverConf is the module's own server settings, from the template its resource names.
+func serverConf(t *testing.T, m map[string]any) string {
+	t.Helper()
+	for _, r := range m["resources"].([]any) {
+		res := r.(map[string]any)
+		if res["id"] == "server-conf" {
+			return templateText(t, res)
+		}
+	}
+	t.Fatal("the nats module declares no server configuration")
+	return ""
+}
+
+// busImage is the upstream image the manifest pins, as a reference docker pulls (name:tag@digest).
+func busImage(t *testing.T, m map[string]any) string {
+	t.Helper()
+	for _, a := range m["build"].(map[string]any)["artifacts"].([]any) {
+		art := a.(map[string]any)
+		if art["name"] == "server" {
+			if art["kind"] != "upstream" {
+				t.Fatalf("the server artifact is a %v, not an upstream image", art["kind"])
+			}
+			return art["from"].(string)
+		}
+	}
+	t.Fatal("the nats module declares no server artifact")
+	return ""
+}
+
+// programs builds the snapshot program for alpine (static, as the mesh's build does) and lays out the two
+// directories the manifest mounts: the program at /opt/mesh and the entrypoint at /opt/mesh-entrypoint.
+func programs(t *testing.T) (program, entry string) {
+	t.Helper()
+	program, entry = t.TempDir(), t.TempDir()
+	build := exec.Command("go", "build", "-trimpath", "-o", filepath.Join(program, "mesh-nats-snapshot"), ".")
+	build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOOS=linux")
+	if out, err := build.CombinedOutput(); err != nil {
+		t.Fatalf("building the program: %v\n%s", err, out)
+	}
+	script, err := os.ReadFile(filepath.Join(moduleDir, "program", "entrypoint.sh"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Mode 0644: the manifest runs the script through sh, so no execute bit is relied on.
+	if err := os.WriteFile(filepath.Join(entry, "entrypoint.sh"), script, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	return program, entry
+}
+
+// runBus starts the upstream image the way the manifest does: its volumes, its args, the module's configuration
+// and the given user list. The poll is a second so a test need not wait out the production five.
+func runBus(t *testing.T, m map[string]any, name, accounts string) (port, pin, conf string) {
+	t.Helper()
+	if out, _ := exec.Command("docker", "ps", "-a", "-q", "--filter", "name=^"+name+"$").Output(); len(strings.TrimSpace(string(out))) > 0 {
+		t.Skipf("a container called %s already runs here; this test will not touch it", name)
+	}
+	conf, tlsDir, data := t.TempDir(), t.TempDir(), t.TempDir()
+	pin = selfSigned(t, tlsDir)
+	_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(serverConf(t, m)), 0o644)
+	_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(accounts), 0o644)
+	program, entry := programs(t)
+	docker(t, "run", "-d", "--rm", "--name", name, "--user", owner(), "-p", "127.0.0.1::4222",
+		"-e", "MESH_NATS_CONF_POLL_SECONDS=1",
+		"-v", data+":/data", "-v", conf+":/etc/nats:ro", "-v", tlsDir+":/tls:ro",
+		"-v", program+":/opt/mesh:ro", "-v", entry+":/opt/mesh-entrypoint:ro",
+		busImage(t, m), "sh", "/opt/mesh-entrypoint/entrypoint.sh")
+	t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
+	_, port, _ = strings.Cut(docker(t, "port", name, "4222/tcp"), ":")
+	return port, pin, conf
+}
+
+// The dump is the one the manifest declares, run by `sh -c` as the holder runs it, with the module's
+// directories filled in; the server is the module's image with the module's own configuration and a
+// user list holding the bus's own module with the controller's grants for it.
+func TestTheDeclaredDumpRunsInTheBussImage(t *testing.T) {
+	needDocker(t)
+	const name = "mesh-broker-nats" // the container the dump names
+	m := manifestOf(t)
+	dump := declaredDump(t, m)
+
+	state, snapshots := t.TempDir(), t.TempDir()
+	accounts := liveBusConf()
+	accounts = accounts[strings.Index(accounts, "accounts {"):]
+	port, pin, _ := runBus(t, m, name, accounts)
+
+	admin := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+	js, _ := admin.JetStream()
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_hand-acts"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"a", "b", "c"} {
+		if _, err := kv.Put(k, []byte("act "+k)); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	run := func(cred Credential) (string, error) {
+		raw, _ := json.Marshal(cred)
+		if err := os.WriteFile(filepath.Join(state, "broker"), raw, 0o600); err != nil {
+			t.Fatal(err)
+		}
+		command := strings.NewReplacer("${dir:mesh-state}", state, "${dir:snapshots}", snapshots).Replace(dump)
+		out, err := exec.Command("sh", "-c", command).CombinedOutput()
+		return string(out), err
+	}
+
+	began := time.Now()
+	out, err := run(Credential{URL: "nats://bus.example:4222", Fingerprint: pin, User: snapshotUser, Password: "snap"})
+	if err != nil {
+		t.Fatalf("the declared dump failed: %v\n%s", err, out)
+	}
+	t.Logf("the declared dump took %s:\n%s", time.Since(began).Round(time.Millisecond), out)
+	tarPath := filepath.Join(snapshots, "bus.tar")
+	f, err := os.Open(tarPath)
+	if err != nil {
+		t.Fatalf("the dump left no bus.tar: %v", err)
+	}
+	a, err := ReadArchive(f, "")
+	f.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer a.Close()
+	if err := a.Verify(); err != nil {
+		t.Fatal(err)
+	}
+	if len(a.Manifest.Streams) != 1 || a.Manifest.Streams[0].Name != "KV_mesh-controller_hand-acts" || a.Manifest.Streams[0].Messages != 3 {
+		t.Fatalf("the snapshot holds %+v", a.Manifest.Streams)
+	}
+	before, _ := os.ReadFile(tarPath)
+
+	// A night that cannot snapshot fails, says why, and leaves last night's snapshot as it was.
+	out, err = run(Credential{Fingerprint: pin, User: snapshotUser, Password: "wrong"})
+	if err == nil {
+		t.Fatalf("a dump the bus refused succeeded:\n%s", out)
+	}
+	if !strings.Contains(out, "Authorization Violation") && !strings.Contains(strings.ToLower(out), "authorization") {
+		t.Errorf("the failure does not say the bus refused the credential: %s", out)
+	}
+	after, _ := os.ReadFile(tarPath)
+	if string(after) != string(before) {
+		t.Error("a failed night replaced the last good snapshot")
+	}
+	if _, err := os.Stat(tarPath + ".partial"); !os.IsNotExist(err) {
+		t.Error("a failed night left its partial file behind")
+	}
+}
+
+// **A change of the user list reloads the server and keeps its connections** (ADR 0480 clause 5, ADR 0236 §37).
+// The mounted entrypoint watches accounts.conf and signals the server; a client connected meanwhile is not
+// dropped, and a user added by the change can then connect.
+func TestTheEntrypointReloadsTheServerWithoutDroppingAConnection(t *testing.T) {
+	needDocker(t)
+	m := manifestOf(t)
+	accounts := plainConf()
+	accounts = accounts[strings.Index(accounts, "accounts {"):]
+	port, pin, conf := runBus(t, m, fmt.Sprintf("mesh-reload-test-%d", os.Getpid()), accounts)
+
+	admin := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+	if err := admin.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	changed := strings.Replace(accounts, `{ user: "admin", password: "admin" }`,
+		`{ user: "admin", password: "admin" }, { user: "late", password: "late" }`, 1)
+	if changed == accounts {
+		t.Fatal("the test's user list did not change")
+	}
+	// Written as the mesh writes it: replaced, not edited in place, so the watcher sees whole content.
+	tmp := filepath.Join(conf, "accounts.conf.new")
+	if err := os.WriteFile(tmp, []byte(changed), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Rename(tmp, filepath.Join(conf, "accounts.conf")); err != nil {
+		t.Fatal(err)
+	}
+
+	// dial retries until the server takes the user: the reload happens within the poll interval.
+	late := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "late", Password: "late"}, nil)
+	_ = late
+	if !admin.IsConnected() {
+		t.Fatal("the connection open before the reload was dropped by it")
+	}
+	if err := admin.Flush(); err != nil {
+		t.Fatalf("the connection open before the reload no longer works: %v", err)
+	}
+	if n := admin.Stats().Reconnects; n != 0 {
+		t.Fatalf("the connection open before the reload reconnected %d times", n)
+	}
+}
+
+// **The bus can go back** (ADR 0480 clause 6): the new declaration and then the old one run on the same store, both
+// well, which is what makes `reversible` true for the bus upgrade. The server binary (the upstream image's digest),
+// the data directory and the configuration are the same in both; only where the entrypoint and the program come
+// from differs. The old declaration is run as the image it replaced ran it: the same script as the image's own
+// entrypoint, the program on the image's path. Then the new one again, so a way there, back and there again is
+// shown, each time with the stream's messages as the one before left them.
+func TestTheStoreSurvivesTheNewDeclarationTheOldAndTheNewAgain(t *testing.T) {
+	needDocker(t)
+	m := manifestOf(t)
+	accounts := liveBusConf()
+	accounts = accounts[strings.Index(accounts, "accounts {"):]
+	conf, tlsDir, data := t.TempDir(), t.TempDir(), t.TempDir()
+	pin := selfSigned(t, tlsDir)
+	_ = os.WriteFile(filepath.Join(conf, "nats.conf"), []byte(serverConf(t, m)), 0o644)
+	_ = os.WriteFile(filepath.Join(conf, "accounts.conf"), []byte(accounts), 0o644)
+	program, entry := programs(t)
+	script := filepath.Join(entry, "entrypoint.sh")
+	if err := os.Chmod(script, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	common := []string{"-d", "--rm", "--user", owner(), "-p", "127.0.0.1::4222", "-e", "MESH_NATS_CONF_POLL_SECONDS=1",
+		"-v", data + ":/data", "-v", conf + ":/etc/nats:ro", "-v", tlsDir + ":/tls:ro"}
+	declarations := map[string][]string{
+		"new": {"-v", program + ":/opt/mesh:ro", "-v", entry + ":/opt/mesh-entrypoint:ro", busImage(t, m), "sh", "/opt/mesh-entrypoint/entrypoint.sh"},
+		"old": {"-v", script + ":/usr/local/bin/mesh-nats-entrypoint:ro",
+			"-v", filepath.Join(program, "mesh-nats-snapshot") + ":/usr/local/bin/mesh-nats-snapshot:ro",
+			"--entrypoint", "/usr/local/bin/mesh-nats-entrypoint", busImage(t, m)},
+	}
+	var kv nats.KeyValue
+	puts := 0
+	step := func(which string) {
+		name := fmt.Sprintf("mesh-rehearsal-%s-%d", which, os.Getpid())
+		docker(t, append(append([]string{"run", "--name", name}, common...), declarations[which]...)...)
+		t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
+		_, port, _ := strings.Cut(docker(t, "port", name, "4222/tcp"), ":")
+		c := dial(t, "tls://127.0.0.1:"+port, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+		js, _ := c.JetStream()
+		var err error
+		if puts == 0 {
+			kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-rehearsal"})
+		} else {
+			kv, err = js.KeyValue("mesh-rehearsal")
+		}
+		if err != nil {
+			t.Fatalf("%s declaration: the store's bucket: %v", which, err)
+		}
+		for i := 1; i <= puts; i++ {
+			e, err := kv.Get(fmt.Sprintf("k%d", i))
+			if err != nil || string(e.Value()) != fmt.Sprintf("v%d", i) {
+				t.Fatalf("%s declaration: key k%d reads %v, %v", which, i, e, err)
+			}
+		}
+		puts++
+		if _, err := kv.Put(fmt.Sprintf("k%d", puts), []byte(fmt.Sprintf("v%d", puts))); err != nil {
+			t.Fatal(err)
+		}
+		c.Close()
+		docker(t, "stop", "-t", "10", name)
+	}
+	step("new")
+	step("old")
+	step("new")
+}
@@ -0,0 +1,587 @@
+package main
+
+// The snapshot and the way back, against real servers: the bus's own release in throwaway containers.
+//
+//	MESH_TEST_DOCKER=1 go test -race ./...
+//
+// Every container and file it makes is removed when it ends. MESH_TEST_SNAPSHOT_MESSAGES sets how
+// many events the source holds (default 6000), to time a snapshot at the live bus's size.
+
+import (
+	"bytes"
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/hex"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"math/big"
+	mrand "math/rand"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// image is the bus's release (the manifest's server artifact names the same release in its tag).
+const image = "nats:2.11-alpine"
+
+// snapshotUser is the bus's own module on a machine, as the controller names it.
+const snapshotUser = "anchor.nats"
+
+// snapshotGrants are what mesh-controller composes for the module holding mesh-broker
+// (internal/broker BusSnapshotGrants), written out here because this repository does not import the
+// controller. Its own test proves the same list against a real server; this one proves the program
+// needs no more than it.
+var snapshotGrants = struct{ publish, subscribe []string }{
+	publish: []string{
+		"$JS.API.STREAM.NAMES",
+		"$JS.API.STREAM.INFO.*",
+		"$JS.API.STREAM.SNAPSHOT.*",
+		"$JS.SNAPSHOT.ACK.>",
+	},
+	subscribe: []string{"_INBOX." + snapshotUser + ".>"},
+}
+
+func needDocker(t *testing.T) {
+	t.Helper()
+	if os.Getenv("MESH_TEST_DOCKER") != "1" {
+		t.Skip("MESH_TEST_DOCKER is not 1: these tests start throwaway nats containers")
+	}
+}
+
+func docker(t *testing.T, args ...string) string {
+	t.Helper()
+	out, err := exec.Command("docker", args...).CombinedOutput()
+	if err != nil {
+		t.Fatalf("docker %s: %v\n%s", strings.Join(args, " "), err, out)
+	}
+	return strings.TrimSpace(string(out))
+}
+
+func owner() string { return fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()) }
+
+// server is one throwaway container.
+type server struct {
+	name string
+	port string
+}
+
+// startServer runs the bus's release with a configuration, as this user so what it writes into a
+// mounted directory is removable by the test.
+func startServer(t *testing.T, conf string, mounts ...string) server {
+	t.Helper()
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, "nats.conf"), []byte(conf), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	name := fmt.Sprintf("mesh-snapshot-test-%d-%d", os.Getpid(), time.Now().UnixNano())
+	args := []string{"run", "-d", "--rm", "--name", name, "--user", owner(), "-p", "127.0.0.1::4222",
+		"-v", dir + ":/etc/nats:ro"}
+	data := false
+	for _, m := range mounts {
+		args = append(args, "-v", m)
+		data = data || strings.HasSuffix(m, ":/data")
+	}
+	if !data {
+		// The store on a directory of the test's, so the server writes it as this user and the
+		// test removes it.
+		args = append(args, "-v", t.TempDir()+":/data")
+	}
+	args = append(args, image, "-c", "/etc/nats/nats.conf")
+	docker(t, args...)
+	t.Cleanup(func() { _ = exec.Command("docker", "rm", "-f", name).Run() })
+	mapped := docker(t, "port", name, "4222/tcp")
+	_, port, _ := strings.Cut(strings.Split(mapped, "\n")[0], ":")
+	return server{name: name, port: port}
+}
+
+// selfSigned is a certificate with no names, the ordinary case for a mesh's bus, and its pin.
+func selfSigned(t *testing.T, dir string) string {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tmpl := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "bus"},
+		NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour)}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	keyDER, _ := x509.MarshalECPrivateKey(key)
+	_ = os.WriteFile(filepath.Join(dir, "tls.crt"), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644)
+	_ = os.WriteFile(filepath.Join(dir, "tls.key"), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o644)
+	sum := sha256.Sum256(der)
+	return "sha256:" + hex.EncodeToString(sum[:])
+}
+
+func quotedList(subjects []string) string {
+	q := make([]string, len(subjects))
+	for i, s := range subjects {
+		q[i] = strconv.Quote(s)
+	}
+	return strings.Join(q, ", ")
+}
+
+// liveBusConf is a server shaped like the mesh's: TLS, the one account, an administrator to fill it,
+// and the bus's own module with exactly the grants the controller composes for it.
+func liveBusConf() string {
+	return fmt.Sprintf(`listen: "0.0.0.0:4222"
+tls { cert_file: "/tls/tls.crt", key_file: "/tls/tls.key" }
+jetstream { store_dir: "/data" }
+accounts { MESH { jetstream: enabled, users: [
+  { user: "admin", password: "admin" },
+  { user: %q, password: "snap", permissions: {
+      publish: { allow: [%s] },
+      subscribe: { allow: [%s] } } }
+] } }
+`, snapshotUser, quotedList(snapshotGrants.publish), quotedList(snapshotGrants.subscribe))
+}
+
+// plainConf is a fresh server with the one account and an administrator, nothing in it.
+func plainConf() string {
+	return `listen: "0.0.0.0:4222"
+jetstream { store_dir: "/data" }
+accounts { MESH { jetstream: enabled, users: [ { user: "admin", password: "admin" } ] } }
+`
+}
+
+func dial(t *testing.T, url string, c Credential, errs chan<- error) *nats.Conn {
+	t.Helper()
+	var nc *nats.Conn
+	var err error
+	for deadline := time.Now().Add(20 * time.Second); ; {
+		nc, err = connect(url, c)
+		if err == nil {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal(err)
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	if errs != nil {
+		nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+			select {
+			case errs <- err:
+			default:
+			}
+		})
+	}
+	t.Cleanup(nc.Close)
+	return nc
+}
+
+func messageCount() int {
+	if n, err := strconv.Atoi(os.Getenv("MESH_TEST_SNAPSHOT_MESSAGES")); err == nil && n > 0 {
+		return n
+	}
+	return 6000
+}
+
+// fill gives the source what the mesh's bus holds: events with headers and a gap deleted from the
+// middle, a work queue partly worked, key-value buckets with history, deletes and a purge, an
+// object in chunks, durable consumers part way through, and a memory stream.
+func fill(t *testing.T, js nats.JetStreamContext) {
+	t.Helper()
+	must := func(err error) {
+		t.Helper()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	_, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"},
+		MaxMsgsPerSubject: 10000, MaxAge: 7 * 24 * time.Hour})
+	must(err)
+	r := mrand.New(mrand.NewSource(1))
+	payload := make([]byte, 2048)
+	total := messageCount()
+	var acks []nats.PubAckFuture
+	for i := 0; i < total; i++ {
+		r.Read(payload)
+		m := nats.NewMsg(fmt.Sprintf("mesh.mod.m%d.event.happened", i%30))
+		m.Header.Set("Mesh-Seq", strconv.Itoa(i))
+		m.Data = append([]byte(nil), payload[:200+r.Intn(1800)]...)
+		f, err := js.PublishMsgAsync(m)
+		must(err)
+		acks = append(acks, f)
+		if len(acks) == 2000 {
+			<-js.PublishAsyncComplete()
+			acks = acks[:0]
+		}
+	}
+	<-js.PublishAsyncComplete()
+	for seq := uint64(100); seq < 120; seq++ {
+		must(js.DeleteMsg("EVENTS", seq))
+	}
+	_, err = js.AddConsumer("EVENTS", &nats.ConsumerConfig{Durable: "audit", AckPolicy: nats.AckExplicitPolicy})
+	must(err)
+	sub, err := js.PullSubscribe("mesh.mod.*.event.>", "audit", nats.Bind("EVENTS", "audit"))
+	must(err)
+	msgs, err := sub.Fetch(500, nats.MaxWait(5*time.Second))
+	must(err)
+	for _, m := range msgs {
+		must(m.Ack())
+	}
+
+	_, err = js.AddStream(&nats.StreamConfig{Name: "CONTROL", Subjects: []string{"mesh.control.*.report"},
+		Retention: nats.WorkQueuePolicy})
+	must(err)
+	for i := 0; i < 50; i++ {
+		_, err := js.Publish(fmt.Sprintf("mesh.control.n%d.report", i%4), []byte(fmt.Sprintf("report %d", i)))
+		must(err)
+	}
+	_, err = js.AddConsumer("CONTROL", &nats.ConsumerConfig{Durable: "controller", AckPolicy: nats.AckExplicitPolicy})
+	must(err)
+	wsub, err := js.PullSubscribe("mesh.control.*.report", "controller", nats.Bind("CONTROL", "controller"))
+	must(err)
+	worked, err := wsub.Fetch(20, nats.MaxWait(5*time.Second))
+	must(err)
+	for _, m := range worked {
+		must(m.AckSync())
+	}
+
+	conditions, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_conditions"})
+	must(err)
+	history, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_condition-history", History: 5})
+	must(err)
+	for i := 0; i < 200; i++ {
+		key := fmt.Sprintf("machine.n%d.silent", i%40)
+		_, err := conditions.Put(key, []byte(fmt.Sprintf(`{"raised":%d}`, i)))
+		must(err)
+		_, err = history.Put(key, []byte(fmt.Sprintf(`{"at":%d}`, i)))
+		must(err)
+	}
+	for i := 0; i < 5; i++ {
+		must(conditions.Delete(fmt.Sprintf("machine.n%d.silent", i)))
+	}
+	must(history.Purge("machine.n7.silent"))
+	_, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_lease", TTL: time.Hour})
+	must(err)
+
+	objects, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "artifacts"})
+	must(err)
+	big := make([]byte, 3<<20)
+	r.Read(big)
+	_, err = objects.PutBytes("facts/latest", big)
+	must(err)
+
+	_, err = js.AddStream(&nats.StreamConfig{Name: "SCRATCH", Subjects: []string{"scratch.>"}, Storage: nats.MemoryStorage})
+	must(err)
+	_, err = js.Publish("scratch.x", []byte("gone at a restart"))
+	must(err)
+}
+
+// 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) {
+	t.Helper()
+	for _, s := range m.Streams {
+		info, err := dst.StreamInfo(s.Name)
+		if err != nil {
+			t.Errorf("%s was not restored: %v", s.Name, err)
+			continue
+		}
+		if err := heldTo(streamState{Messages: info.State.Msgs, LastSeq: info.State.LastSeq}, s); err != nil {
+			t.Errorf("%s: %v", s.Name, err)
+		}
+		// **Up to the snapshot's last sequence, exactly the source**: every message by sequence —
+		// subject, headers and body — and a message deleted before the snapshot still deleted. Past it,
+		// what was written while the blocks were read out: a message there may or may not have made it
+		// (the server's state says the stream reached it before the block holding it was read), and
+		// one that did is the source's. The source only grows after the snapshot.
+		for seq := info.State.FirstSeq; seq <= info.State.LastSeq && seq > 0; seq++ {
+			want, werr := src.GetMsg(s.Name, seq)
+			got, gerr := dst.GetMsg(s.Name, seq)
+			switch {
+			case errors.Is(werr, nats.ErrMsgNotFound):
+				if gerr == nil {
+					t.Errorf("%s %d was deleted and came back", s.Name, seq)
+				}
+				continue
+			case errors.Is(gerr, nats.ErrMsgNotFound):
+				if seq <= s.LastSeq {
+					t.Errorf("%s %d was in the stream at its snapshot and is missing from the restore", s.Name, seq)
+				}
+				continue
+			case werr != nil || gerr != nil:
+				t.Fatalf("%s %d: source %v, restored %v", s.Name, seq, werr, gerr)
+			}
+			if got.Subject != want.Subject || !bytes.Equal(got.Data, want.Data) || fmt.Sprint(got.Header) != fmt.Sprint(want.Header) {
+				t.Fatalf("%s %d differs: %s %q vs %s %q", s.Name, seq, got.Subject, truncate(got.Data), want.Subject, truncate(want.Data))
+			}
+		}
+		if info.State.LastSeq > s.LastSeq {
+			t.Logf("%s: restored to sequence %d, written while the snapshot read it; the manifest said %d", s.Name, info.State.LastSeq, s.LastSeq)
+		}
+		for _, consumer := range []string{"audit", "controller"} {
+			wantC, err := src.ConsumerInfo(s.Name, consumer)
+			if err != nil {
+				continue
+			}
+			gotC, err := dst.ConsumerInfo(s.Name, consumer)
+			if err != nil {
+				t.Errorf("%s's consumer %s was not restored: %v", s.Name, consumer, err)
+				continue
+			}
+			if gotC.AckFloor.Stream != wantC.AckFloor.Stream {
+				t.Errorf("%s's consumer %s acknowledged up to %d, the source up to %d", s.Name, consumer,
+					gotC.AckFloor.Stream, wantC.AckFloor.Stream)
+			}
+		}
+	}
+	for _, s := range m.Skipped {
+		if _, err := dst.StreamInfo(s.Name); err == nil {
+			t.Errorf("%s was skipped and yet exists", s.Name)
+		}
+	}
+	// And the buckets work as buckets, not only as streams.
+	kv, err := dst.KeyValue("mesh-controller_condition-history")
+	if err != nil {
+		t.Fatalf("the condition history is not a bucket after the restore: %v", err)
+	}
+	h, err := kv.History("machine.n3.silent")
+	if err != nil || len(h) != 5 {
+		t.Errorf("the history of a key came back as %d entries (%v), not 5", len(h), err)
+	}
+	if _, err := kv.Get("machine.n7.silent"); !errors.Is(err, nats.ErrKeyNotFound) {
+		t.Errorf("a purged key came back: %v", err)
+	}
+	cond, _ := dst.KeyValue("mesh-controller_conditions")
+	if _, err := cond.Get("machine.n1.silent"); !errors.Is(err, nats.ErrKeyNotFound) {
+		t.Errorf("a deleted key came back: %v", err)
+	}
+	if e, err := cond.Get("machine.n10.silent"); err != nil || string(e.Value()) != `{"raised":170}` {
+		t.Errorf("a condition came back wrong: %v", err)
+	}
+	srcObj, _ := src.ObjectStore("artifacts")
+	dstObj, err := dst.ObjectStore("artifacts")
+	if err != nil {
+		t.Fatalf("the object store was not restored: %v", err)
+	}
+	want, _ := srcObj.GetBytes("facts/latest")
+	got, err := dstObj.GetBytes("facts/latest")
+	if err != nil || !bytes.Equal(got, want) {
+		t.Errorf("the object came back different (%d bytes, %v)", len(got), err)
+	}
+}
+
+func TestASnapshotOfALiveBusRestoresToTheSameContent(t *testing.T) {
+	needDocker(t)
+	tlsDir := t.TempDir()
+	pin := selfSigned(t, tlsDir)
+	source := startServer(t, liveBusConf(), tlsDir+":/tls:ro")
+	url := "tls://127.0.0.1:" + source.port
+
+	admin := dial(t, url, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+	srcJS, _ := admin.JetStream()
+	fill(t, srcJS)
+
+	// The bus goes on taking writes while it is snapshotted, and every one is accepted.
+	writer := dial(t, url, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+	writerJS, _ := writer.JetStream()
+	stop := make(chan struct{})
+	var wrote, failed atomic.Int64
+	var slowest atomic.Int64
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			began := time.Now()
+			if _, err := writerJS.Publish("mesh.mod.writer.event.tick", []byte("during")); err != nil {
+				failed.Add(1)
+			} else {
+				wrote.Add(1)
+			}
+			if d := time.Since(began).Milliseconds(); d > slowest.Load() {
+				slowest.Store(d)
+			}
+			time.Sleep(2 * time.Millisecond)
+		}
+	}()
+
+	snap := dial(t, url, Credential{Fingerprint: pin, User: snapshotUser, Password: "snap"}, nil)
+	var archive bytes.Buffer
+	var said []string
+	var sayMu sync.Mutex
+	m, err := Snapshot(context.Background(), snap, &archive, SnapshotOptions{StreamTimeout: time.Minute, Pause: 50 * time.Millisecond,
+		Log: func(f string, a ...any) { sayMu.Lock(); said = append(said, fmt.Sprintf(f, a...)); sayMu.Unlock() }})
+	time.Sleep(100 * time.Millisecond)
+	close(stop)
+	wg.Wait()
+	if err != nil {
+		t.Fatalf("the snapshot failed: %v\n%s", err, strings.Join(said, "\n"))
+	}
+	t.Logf("snapshot: %d streams, %d messages, %s of streams → %s archived in %.2fs; %d writes during it, %d refused, slowest %dms",
+		len(m.Streams), m.Messages, human(m.StreamBytes), human(m.ArchiveBytes), m.Seconds, wrote.Load(), failed.Load(), slowest.Load())
+	for _, line := range said {
+		t.Log(line)
+	}
+	if failed.Load() > 0 {
+		t.Errorf("%d writes to the bus failed while it was snapshotted", failed.Load())
+	}
+	if len(m.Skipped) != 1 || m.Skipped[0].Name != "SCRATCH" {
+		t.Errorf("the memory stream was not said as skipped: %+v", m.Skipped)
+	}
+	var names []string
+	for _, s := range m.Streams {
+		names = append(names, s.Name)
+	}
+	wantNames := "CONTROL EVENTS KV_mesh-controller_condition-history KV_mesh-controller_conditions KV_mesh-controller_lease OBJ_artifacts"
+	if strings.Join(names, " ") != wantNames {
+		t.Errorf("snapshotted %v, want %s", names, wantNames)
+	}
+
+	tarPath := filepath.Join(t.TempDir(), "bus.tar")
+	if err := os.WriteFile(tarPath, archive.Bytes(), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	a, err := ReadArchive(bytes.NewReader(archive.Bytes()), "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer a.Close()
+	if err := a.Verify(); err != nil {
+		t.Fatalf("a fresh snapshot does not verify: %v", err)
+	}
+
+	t.Run("nothing is restored over a live stream", func(t *testing.T) {
+		err := RestoreAll(context.Background(), admin, a, t.Logf)
+		if err == nil || !strings.Contains(err.Error(), "already holds") {
+			t.Fatalf("restoring over the live streams was not refused: %v", err)
+		}
+	})
+
+	t.Run("into a fresh server", func(t *testing.T) {
+		fresh := startServer(t, plainConf())
+		nc := dial(t, "nats://127.0.0.1:"+fresh.port, Credential{User: "admin", Password: "admin"}, nil)
+		if err := RestoreAll(context.Background(), nc, a, t.Logf); err != nil {
+			t.Fatal(err)
+		}
+		dstJS, _ := nc.JetStream()
+		sameContent(t, m, srcJS, dstJS)
+	})
+
+	t.Run("into a new store, swapped in and served", func(t *testing.T) {
+		binary := filepath.Join(t.TempDir(), "mesh-nats-snapshot")
+		build := exec.Command("go", "build", "-o", binary, ".")
+		build.Env = append(os.Environ(), "CGO_ENABLED=0")
+		if out, err := build.CombinedOutput(); err != nil {
+			t.Fatalf("building the program: %v\n%s", err, out)
+		}
+		out := t.TempDir()
+		// The program in the bus's own image, as a person runs it: the archive in, a new store out.
+		docker(t, "run", "--rm", "--user", owner(),
+			"-v", binary+":/usr/local/bin/mesh-nats-snapshot:ro",
+			"-v", tarPath+":/in/bus.tar:ro", "-v", out+":/out",
+			"--entrypoint", "/usr/local/bin/mesh-nats-snapshot", image,
+			"restore", "--into", "/out/store", "--from", "/in/bus.tar")
+		if _, err := os.Stat(filepath.Join(out, "store", "jetstream", "MESH", "streams", "EVENTS")); err != nil {
+			t.Fatalf("the new store is not laid out as the bus's: %v", err)
+		}
+		served := startServer(t, plainConf(), filepath.Join(out, "store")+":/data")
+		nc := dial(t, "nats://127.0.0.1:"+served.port, Credential{User: "admin", Password: "admin"}, nil)
+		dstJS, _ := nc.JetStream()
+		sameContent(t, m, srcJS, dstJS)
+	})
+
+	t.Run("a damaged archive is refused", func(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), "")
+		if err == nil {
+			defer d.Close()
+			err = d.Verify()
+		}
+		if err == nil {
+			t.Fatal("a damaged archive verified")
+		}
+	})
+}
+
+// The bus's own module may snapshot and do nothing else: every API that changes a stream, every
+// publish into one, and every subscription beyond its own inbox is refused by the server.
+func TestTheSnapshotUserCannotChangeTheBus(t *testing.T) {
+	needDocker(t)
+	tlsDir := t.TempDir()
+	pin := selfSigned(t, tlsDir)
+	source := startServer(t, liveBusConf(), tlsDir+":/tls:ro")
+	url := "tls://127.0.0.1:" + source.port
+	admin := dial(t, url, Credential{Fingerprint: pin, User: "admin", Password: "admin"}, nil)
+	js, _ := admin.JetStream()
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Publish("mesh.mod.a.event.b", []byte("x")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_conditions"}); err != nil {
+		t.Fatal(err)
+	}
+
+	errs := make(chan error, 64)
+	snap := dial(t, url, Credential{Fingerprint: pin, User: snapshotUser, Password: "snap"}, errs)
+	for _, subject := range []string{
+		"$JS.API.STREAM.CREATE.NEW", "$JS.API.STREAM.UPDATE.EVENTS", "$JS.API.STREAM.DELETE.EVENTS",
+		"$JS.API.STREAM.PURGE.EVENTS", "$JS.API.STREAM.MSG.DELETE.EVENTS", "$JS.API.STREAM.RESTORE.NEW",
+		"$JS.API.CONSUMER.CREATE.EVENTS", "$JS.API.STREAM.MSG.GET.EVENTS",
+		"mesh.mod.a.event.b", "$KV.mesh-controller_conditions.k",
+	} {
+		_, err := snap.Request(subject, []byte(`{}`), 500*time.Millisecond)
+		if err == nil {
+			t.Errorf("%s was answered for the snapshot user", subject)
+		}
+		select {
+		case e := <-errs:
+			if !strings.Contains(strings.ToLower(e.Error()), "permissions violation") {
+				t.Errorf("%s: %v", subject, e)
+			}
+		case <-time.After(2 * time.Second):
+			t.Errorf("the server did not refuse %s", subject)
+		}
+	}
+	for _, subject := range []string{"mesh.>", "_INBOX.other.>", "$JS.API.>"} {
+		if _, err := snap.SubscribeSync(subject); err != nil {
+			t.Fatal(err)
+		}
+		_ = snap.Flush()
+		select {
+		case e := <-errs:
+			if !strings.Contains(strings.ToLower(e.Error()), "permissions violation") {
+				t.Errorf("subscribing %s: %v", subject, e)
+			}
+		case <-time.After(2 * time.Second):
+			t.Errorf("the server let the snapshot user subscribe %s", subject)
+		}
+	}
+	info, err := js.StreamInfo("EVENTS")
+	if err != nil || info.State.Msgs != 1 {
+		t.Fatalf("the stream changed: %v %+v", err, info)
+	}
+	// And what it may do, it can.
+	names, err := StreamNames(context.Background(), snap)
+	if err != nil || len(names) != 2 {
+		t.Fatalf("the snapshot user cannot list the streams: %v %v", names, err)
+	}
+}
@@ -0,0 +1,209 @@
+// mesh-nats-snapshot: a consistent copy of the mesh bus's streams, and the way back from one
+// (novox/hq ADR 0235, to-be 43).
+//
+// The bus keeps everything it holds in JetStream: every stream, and every key-value bucket, which is
+// a stream too (conditions and their history, calls, hand-acts, the controller's lease, assignments,
+// events, every module's state). Copying the store's files while the server writes them is not a
+// copy that can be trusted to restore — a block half-written, an index from before the block it
+// indexes. The server has its own way to hand a stream out whole: the snapshot API, a chunked
+// transfer with flow control that the server answers from a consistent view of the stream while
+// it goes on taking messages. This program asks for it, one stream at a time, and nothing else.
+//
+// It is mounted into the bus's container (at /opt/mesh) because that is where it runs: the machine's
+// backup holder runs the nats module's declared dump — `docker exec` into the server's container — and
+// the archive comes out on stdout, so the credential arrives on stdin.
+//
+//	mesh-nats-snapshot snapshot  [--server URL] [--credential FILE|-]   > bus.tar
+//	mesh-nats-snapshot verify    [--from FILE]
+//	mesh-nats-snapshot restore   --into DIR [--account MESH] [--from FILE]
+//	mesh-nats-snapshot restore   --server URL --credential FILE [--from FILE]
+//
+// **Read-only on the live bus.** A snapshot neither pauses nor reconfigures a stream; the server
+// keeps accepting writes while it streams the copy out. The credential it uses is the bus's own
+// module's, granted the snapshot API and nothing that writes (mesh-controller internal/broker).
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"time"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		usage()
+		os.Exit(2)
+	}
+	var err error
+	switch os.Args[1] {
+	case "snapshot":
+		err = runSnapshot(os.Args[2:])
+	case "verify":
+		err = runVerify(os.Args[2:])
+	case "restore":
+		err = runRestore(os.Args[2:])
+	case "-h", "--help", "help":
+		usage()
+		return
+	default:
+		usage()
+		os.Exit(2)
+	}
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "mesh-nats-snapshot:", err)
+		os.Exit(1)
+	}
+}
+
+func usage() {
+	fmt.Fprint(os.Stderr, `mesh-nats-snapshot — a consistent copy of the bus's streams, and the way back
+
+  snapshot  [--server tls://127.0.0.1:4222] [--credential -]       every stream, one at a time, as a tar on stdout
+  verify    [--from bus.tar]                                       every archive against the manifest
+  restore   --into DIR [--account MESH] [--from bus.tar]           into a new store directory, by a server of its own
+  restore   --server URL --credential FILE [--from bus.tar]        into a running server that holds none of them
+
+The credential is the JSON the mesh seals to the bus's own module: url, fingerprint, user, password.
+`)
+}
+
+func runSnapshot(args []string) error {
+	fs := flag.NewFlagSet("snapshot", flag.ContinueOnError)
+	server := fs.String("server", "tls://127.0.0.1:4222", "the bus, as seen from where this runs")
+	credential := fs.String("credential", "-", "the bus's own module's credential, a file or - for stdin")
+	streamTimeout := fs.Duration("stream-timeout", 10*time.Minute, "the longest one stream may take")
+	timeout := fs.Duration("timeout", 30*time.Minute, "the longest the whole snapshot may take")
+	pause := fs.Duration("pause", 250*time.Millisecond, "a rest between two streams")
+	tmp := fs.String("tmp", "", "where each stream's archive waits before it is written out (default: the system's)")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	cred, err := readCredential(*credential)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+	nc, err := connect(*server, cred)
+	if err != nil {
+		return err
+	}
+	defer nc.Close()
+	opts := SnapshotOptions{StreamTimeout: *streamTimeout, Pause: *pause, TempDir: *tmp, Log: logf}
+	m, err := Snapshot(ctx, nc, os.Stdout, opts)
+	if err != nil {
+		return err
+	}
+	logf("snapshot: %d stream(s), %d message(s), %s of streams in %s of archives, %.1fs",
+		len(m.Streams), m.Messages, human(m.StreamBytes), human(m.ArchiveBytes), m.Seconds)
+	return nil
+}
+
+func runVerify(args []string) error {
+	fs := flag.NewFlagSet("verify", flag.ContinueOnError)
+	from := fs.String("from", "-", "the archive, a file or - for stdin")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	in, closer, err := openFrom(*from)
+	if err != nil {
+		return err
+	}
+	defer closer()
+	a, err := ReadArchive(in, "")
+	if err != nil {
+		return err
+	}
+	defer a.Close()
+	if err := a.Verify(); err != nil {
+		return err
+	}
+	logf("verified: %d stream(s), taken %s, every archive matches the manifest and reads to its end",
+		len(a.Manifest.Streams), a.Manifest.Finished.Format(time.RFC3339))
+	return nil
+}
+
+func runRestore(args []string) error {
+	fs := flag.NewFlagSet("restore", flag.ContinueOnError)
+	from := fs.String("from", "-", "the archive, a file or - for stdin")
+	into := fs.String("into", "", "a new store directory, filled by a server of this program's own on loopback")
+	account := fs.String("account", "MESH", "the account the streams belong to (the bus's one account)")
+	server := fs.String("server", "", "a running server to restore into instead; it must hold none of the streams")
+	credential := fs.String("credential", "", "the credential for --server, a file")
+	timeout := fs.Duration("timeout", time.Hour, "the longest the whole restore may take")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if (*into == "") == (*server == "") {
+		return errors.New("restore needs exactly one of --into (a new store directory) or --server (a running server)")
+	}
+	in, closer, err := openFrom(*from)
+	if err != nil {
+		return err
+	}
+	defer closer()
+	a, err := ReadArchive(in, "")
+	if err != nil {
+		return err
+	}
+	defer a.Close()
+	if err := a.Verify(); err != nil {
+		return fmt.Errorf("the archive does not match its own manifest, so nothing is restored: %w", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+	if *into != "" {
+		where, err := RestoreOffline(ctx, a, *into, *account, logf)
+		if err != nil {
+			return err
+		}
+		logf("restored %d stream(s) into %s. With the bus stopped, put this directory in place of the "+
+			"jetstream directory in the bus's store, then start it (novox/hq to-be 43).", len(a.Manifest.Streams), where)
+		return nil
+	}
+	if *credential == "" {
+		return errors.New("--server needs --credential")
+	}
+	cred, err := readCredential(*credential)
+	if err != nil {
+		return err
+	}
+	nc, err := connect(*server, cred)
+	if err != nil {
+		return err
+	}
+	defer nc.Close()
+	return RestoreAll(ctx, nc, a, logf)
+}
+
+func openFrom(path string) (io.Reader, func(), error) {
+	if path == "-" || path == "" {
+		return os.Stdin, func() {}, nil
+	}
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, nil, err
+	}
+	return f, func() { f.Close() }, nil
+}
+
+func logf(format string, args ...any) {
+	fmt.Fprintf(os.Stderr, format+"\n", args...)
+}
+
+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])
+}
@@ -0,0 +1,261 @@
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// restoreChunk is what is sent at a time; the server answers each before the next goes.
+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 {
+	var present []string
+	for _, s := range a.Manifest.Streams {
+		err := request(ctx, nc, apiStreamInfo+s.Name, nil, nil)
+		var api *apiError
+		switch {
+		case err == nil:
+			present = append(present, s.Name)
+		case errors.As(err, &api) && api.Code == 404:
+		default:
+			return err
+		}
+	}
+	if len(present) > 0 {
+		return fmt.Errorf("the server already holds %s; nothing is restored over a live stream — restore into a new "+
+			"store (--into) and swap it in with the bus stopped", strings.Join(present, ", "))
+	}
+	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, at least %d message(s) to sequence %d, in %.2fs", s.Name, s.Messages, s.LastSeq, time.Since(started).Seconds())
+	}
+	return nil
+}
+
+func restoreStream(ctx context.Context, nc *nats.Conn, a *Archive, s 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 {
+			cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+			reply, rerr := nc.RequestWithContext(cctx, resp.DeliverSubject, buf[:n])
+			cancel()
+			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
+		}
+	}
+	// The end: an empty message, answered once the server has rebuilt the stream from what it was
+	// sent — which for a large stream takes a while.
+	fctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
+	defer cancel()
+	final, err := nc.RequestWithContext(fctx, 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
+	}
+	return heldTo(created.State, s)
+}
+
+// heldTo is whether a restored stream holds at least what its snapshot said it held.
+//
+// **At least, not exactly — found by the restore test, not assumed.** The state the server answers a
+// snapshot request with is taken when the snapshot starts; the stream goes on taking messages while
+// its blocks are read out, and a block read a moment later carries some of what was appended in that
+// moment. So a stream written during its snapshot restores with a later last sequence than the
+// manifest says (90075 where it said 90024, with a writer publishing every 2ms through it), and of
+// 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 {
+	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)
+	}
+	return nil
+}
+
+// RestoreOffline fills a new store directory with the archive's streams, by a server of this
+// program's own: nats-server, the same binary the bus runs, on loopback, with the bus's one account
+// and a user that exists only for this run. It is stopped when the streams are in. What it leaves
+// is a store directory laid out exactly as the bus's — `jetstream/<account>/streams/<name>` — which
+// a person swaps in for the bus's own with the bus stopped. It answers that jetstream directory.
+//
+// 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) {
+		return "", fmt.Errorf("%q is not an account name", account)
+	}
+	if entries, err := os.ReadDir(into); err == nil && len(entries) > 0 {
+		return "", fmt.Errorf("%s is not empty; a restore goes into a new directory, never over one", into)
+	}
+	if err := os.MkdirAll(into, 0o700); err != nil {
+		return "", err
+	}
+	store, err := filepath.Abs(into)
+	if err != nil {
+		return "", err
+	}
+	binary, err := exec.LookPath("nats-server")
+	if err != nil {
+		return "", errors.New("no nats-server here to restore with; run this in the bus's own image")
+	}
+	work, err := os.MkdirTemp("", "mesh-nats-restore-server-*")
+	if err != nil {
+		return "", err
+	}
+	defer os.RemoveAll(work)
+	port, err := freePort()
+	if err != nil {
+		return "", err
+	}
+	password := randomHex(24)
+	conf := filepath.Join(work, "restore.conf")
+	config := fmt.Sprintf(`listen: "127.0.0.1:%d"
+jetstream { store_dir: %q }
+accounts { %s { jetstream: enabled, users: [ { user: "restore", password: %q } ] } }
+`, port, store, account, password)
+	if err := os.WriteFile(conf, []byte(config), 0o600); err != nil {
+		return "", err
+	}
+	cmd := exec.Command(binary, "-c", conf)
+	logFile, err := os.Create(filepath.Join(work, "server.log"))
+	if err != nil {
+		return "", err
+	}
+	defer logFile.Close()
+	cmd.Stdout, cmd.Stderr = logFile, logFile
+	if err := cmd.Start(); err != nil {
+		return "", err
+	}
+	exited := make(chan error, 1)
+	go func() { exited <- cmd.Wait() }()
+	stop := func() error {
+		_ = cmd.Process.Signal(syscall.SIGTERM)
+		select {
+		case err := <-exited:
+			return err
+		case <-time.After(30 * time.Second):
+			_ = cmd.Process.Kill()
+			<-exited
+			return errors.New("the restore's server did not stop within 30s and was killed")
+		}
+	}
+	serverLog := func() string {
+		raw, _ := os.ReadFile(filepath.Join(work, "server.log"))
+		return strings.TrimSpace(string(raw))
+	}
+
+	url := fmt.Sprintf("nats://127.0.0.1:%d", port)
+	var nc *nats.Conn
+	for deadline := time.Now().Add(20 * time.Second); ; {
+		nc, err = connect(url, Credential{User: "restore", Password: password})
+		if err == nil {
+			break
+		}
+		if time.Now().After(deadline) {
+			_ = stop()
+			return "", fmt.Errorf("the restore's server did not come up: %v\n%s", err, serverLog())
+		}
+		select {
+		case err := <-exited:
+			return "", fmt.Errorf("the restore's server stopped: %v\n%s", err, serverLog())
+		case <-time.After(200 * time.Millisecond):
+		}
+	}
+	err = RestoreAll(ctx, nc, a, log)
+	nc.Close()
+	if stopErr := stop(); err == nil && stopErr != nil && !isTerminated(stopErr) {
+		err = stopErr
+	}
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(store, "jetstream"), nil
+}
+
+// isTerminated is a server that stopped because it was asked to.
+func isTerminated(err error) bool {
+	var exit *exec.ExitError
+	if errors.As(err, &exit) {
+		if status, ok := exit.Sys().(syscall.WaitStatus); ok && status.Signaled() && status.Signal() == syscall.SIGTERM {
+			return true
+		}
+	}
+	return false
+}
+
+func freePort() (int, error) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return 0, err
+	}
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port, nil
+}
+
+func randomHex(n int) string {
+	b := make([]byte, n)
+	if _, err := rand.Read(b); err != nil {
+		panic(err)
+	}
+	return hex.EncodeToString(b)
+}
@@ -0,0 +1,446 @@
+package main
+
+import (
+	"archive/tar"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/klauspost/compress/s2"
+	"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)
+}
+
+// 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"`
+	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/<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) {
+	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()}
+	names, err := StreamNames(ctx, nc)
+	if err != nil {
+		return nil, err
+	}
+	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   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, 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, 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) (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 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
+	}
+	if err := nc.FlushWithContext(ctx); err != nil {
+		return 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 StreamEntry{}, nil, err
+	}
+	select {
+	case <-done:
+	case <-ctx.Done():
+		return 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 StreamEntry{}, nil, err
+	}
+	if written == 0 {
+		return StreamEntry{}, nil, errors.New("the server sent an empty archive")
+	}
+	if err := f.Sync(); err != nil {
+		return 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)
+	}
+	if err := json.Unmarshal(resp.State, &state); err != nil {
+		return 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
+	}
+	if err := readsThrough(f); err != nil {
+		return 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 StreamEntry{}, nil, err
+	}
+	metaSum := sha256.Sum256(meta)
+	return 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
+}
+
+// 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 *>")
+}
@@ -0,0 +1,64 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// The module's directory, where its manifest and its files/ are.
+const moduleDir = "../.."
+
+// templateText is the text a file resource places on the machine: its template's, in the module's files/
+// (novox/hq ADR 0349 §2). A resource that names no template, or a template that is not there, fails the test.
+func templateText(t *testing.T, r map[string]any) string {
+	t.Helper()
+	name, _ := r["template"].(string)
+	if name == "" {
+		t.Fatalf("resource %v names no template", r["id"])
+	}
+	raw, err := os.ReadFile(filepath.Join(moduleDir, filepath.FromSlash(name)))
+	if err != nil {
+		t.Fatalf("resource %v names %s: %v", r["id"], name, err)
+	}
+	return string(raw)
+}
+
+// No file resource carries its text inline: each names its template in files/, and the build packs files/
+// into the one archive the controller reads the templates back from (novox/hq ADR 0349 §2).
+func TestEveryFileTakesItsTextFromATemplate(t *testing.T) {
+	raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m struct {
+		Resources []map[string]any `json:"resources"`
+		Build     struct {
+			Artifacts []map[string]any `json:"artifacts"`
+		} `json:"build"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	for _, r := range m.Resources {
+		if _, inline := r["content"]; inline {
+			t.Errorf("resource %v keeps its text inline: keep it in files/ and name it as the template", r["id"])
+		}
+		if r["type"] == "file" {
+			templateText(t, r)
+		}
+	}
+	packs := 0
+	for _, a := range m.Build.Artifacts {
+		if a["kind"] == "archive" && a["from"] == "files" {
+			packs++
+			if a["name"] != "files" {
+				t.Errorf("the templates archive is named %v, not files", a["name"])
+			}
+		}
+	}
+	if packs != 1 {
+		t.Errorf("the build packs files/ into %d archives, not one", packs)
+	}
+}
@@ -0,0 +1,110 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+)
+
+// The shape ADR 0480 clauses 1, 2, 3 and 5 give the bus's manifest: the upstream image by digest, the mesh's two
+// programs mounted read-only under /opt, unpacked by archive resources the container names under restart-on
+// (the node-engine does not look inside a mounted directory), and the container the same one it was.
+func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
+	raw, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m struct {
+		Data struct {
+			Own []struct {
+				ID     string `json:"id"`
+				Backup struct {
+					Dump string `json:"dump"`
+				} `json:"backup"`
+			} `json:"own"`
+		} `json:"data"`
+		Resources []map[string]any `json:"resources"`
+		Build     struct {
+			On        any `json:"on"`
+			Artifacts []struct {
+				Name, Kind, From, Language, System, Binary string
+			} `json:"artifacts"`
+		} `json:"build"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	if m.Build.On != nil {
+		t.Error("build.on is declared: nothing is built on a base image any more")
+	}
+	artifacts := map[string]struct{ Name, Kind, From, Language, System, Binary string }{}
+	for _, a := range m.Build.Artifacts {
+		artifacts[a.Name] = a
+		if a.Kind == "image" {
+			t.Errorf("artifact %s is an image built from a Dockerfile (ADR 0452)", a.Name)
+		}
+	}
+	if a := artifacts["server"]; a.Kind != "upstream" || !strings.Contains(a.From, "@sha256:") {
+		t.Errorf("server is %+v: an upstream image pinned by digest", a)
+	}
+	if a := artifacts["snapshot"]; a.Kind != "bundle" || a.Language != "go" || a.System != "alpine" ||
+		a.From != "cmd/mesh-nats-snapshot" || a.Binary != "mesh-nats-snapshot" {
+		t.Errorf("snapshot is %+v: a Go bundle of the module's own package, stamped for alpine", a)
+	}
+	if a := artifacts["entrypoint"]; a.Kind != "archive" || a.From != "program" {
+		t.Errorf("entrypoint is %+v: an archive of program/", a)
+	}
+	if _, err := os.Stat("../../program/entrypoint.sh"); err != nil {
+		t.Error(err)
+	}
+	if _, err := os.Stat("../../Dockerfile"); err == nil {
+		t.Error("the Dockerfile is still there")
+	}
+
+	byID := map[string]map[string]any{}
+	for _, r := range m.Resources {
+		byID[r["id"].(string)] = r
+	}
+	server := byID["server"]
+	if server["name"] != "mesh-broker-nats" || server["artifact"] != "server" {
+		t.Errorf("the container is %v on artifact %v: it keeps its name, which the dump and the jails use", server["name"], server["artifact"])
+	}
+	volumes, _ := server["volumes"].([]any)
+	restart, _ := server["restart-on"].([]any)
+	for _, p := range []struct{ resource, artifact, mount string }{
+		{"program-snapshot", "snapshot", ":/opt/mesh:ro"},
+		{"program-entrypoint", "entrypoint", ":/opt/mesh-entrypoint:ro"},
+	} {
+		r := byID[p.resource]
+		if r == nil || r["type"] != "archive" || r["artifact"] != p.artifact {
+			t.Errorf("%s is %v: an archive resource unpacking %s", p.resource, r, p.artifact)
+			continue
+		}
+		mounted := false
+		for _, v := range volumes {
+			if v == r["path"].(string)+p.mount {
+				mounted = true
+			}
+		}
+		if !mounted {
+			t.Errorf("%s is not mounted read-only as %s", p.resource, p.mount)
+		}
+		named := false
+		for _, v := range restart {
+			named = named || v == p.resource
+		}
+		if !named {
+			t.Errorf("%s is not in the container's restart-on: a new program would leave the old one running", p.resource)
+		}
+	}
+	args, _ := server["args"].([]any)
+	if len(args) != 2 || args[0] != "sh" || args[1] != "/opt/mesh-entrypoint/entrypoint.sh" {
+		t.Errorf("args are %v: the entrypoint run through sh, so no execute bit is relied on", args)
+	}
+	for _, d := range m.Data.Own {
+		if d.ID == "jetstream" && !strings.Contains(d.Backup.Dump, "docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot snapshot") {
+			t.Errorf("the dump is %q: it calls the mounted program", d.Backup.Dump)
+		}
+	}
+}
@@ -1,6 +1,7 @@
 package main

 import (
+	"encoding/json"
 	"fmt"
 	"io"
 	"math/rand"
@@ -15,21 +16,43 @@ import (
 	"github.com/nats-io/nats.go"
 )

-// The server these tests run is the server the image runs. The image is pinned by a digest, which
-// names no release, so the Dockerfile says which release it is, and this keeps the two equal: a
-// test of the server below is only worth something about the server the mesh actually runs.
+// The server these tests run is the server the mesh runs. The manifest's `server` artifact is the upstream
+// image, written `name:tag@sha256:<digest>` (novox/hq ADR 0480 clause 3): the digest names the image and the
+// tag says the release it is. This keeps the tag and the server compiled in here equal: a test of the server
+// below is only worth something about the server the mesh actually runs.
 func TestTheImageIsTheServerTestedHere(t *testing.T) {
-	raw, err := os.ReadFile("../../Dockerfile")
+	raw, err := os.ReadFile("../../module.json")
 	if err != nil {
 		t.Fatal(err)
 	}
-	said := regexp.MustCompile(`(?m)^# upstream: nats (\d+\.\d+\.\d+)-alpine$`).FindSubmatch(raw)
+	var m struct {
+		Build struct {
+			Artifacts []struct {
+				Name string `json:"name"`
+				Kind string `json:"kind"`
+				From string `json:"from"`
+			} `json:"artifacts"`
+		} `json:"build"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	from := ""
+	for _, a := range m.Build.Artifacts {
+		if a.Name == "server" {
+			if a.Kind != "upstream" {
+				t.Fatalf("the server artifact is a %q, not an upstream image", a.Kind)
+			}
+			from = a.From
+		}
+	}
+	said := regexp.MustCompile(`^nats:(\d+\.\d+\.\d+)-alpine@sha256:[0-9a-f]{64}$`).FindStringSubmatch(from)
 	if said == nil {
-		t.Fatal("the Dockerfile does not say which nats release its digest is (`# upstream: nats X.Y.Z-alpine`)")
+		t.Fatalf("the server artifact is %q: it must read nats:X.Y.Z-alpine@sha256:<digest>, the tag saying the release", from)
 	}
 	// The server's own word for its release, compiled into it: a test binary's build information may list
 	// no dependencies (the build seat's did not, so this read "these tests run no nats-server").
-	if server.VERSION != string(said[1]) {
+	if server.VERSION != said[1] {
 		t.Fatalf("the image runs nats %s and these tests run %s: move them together", said[1], server.VERSION)
 	}
 }