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

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 Dockerfile's `# upstream:` line names the same one).
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)
	}
}