mesh-catalog/modules/nats/cmd/nats-tools/keeper_integration_test.go

//go:build integration

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"os"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"git.novox.be/novox/mesh-sdk/go/seatlist"
	"git.novox.be/novox/mesh-sdk/go/views"

	"git.novox.be/novox/mesh-sdk/go/bussnap"
	"nats-tools/internal/keeper"
)

// kvStore is the keeper's own state on the real bus: one bucket per state, as the grants part creates its old sections.
type kvStore struct {
	js jetstream.JetStream
}

func (s kvStore) kv(state string) (jetstream.KeyValue, error) {
	return s.js.KeyValue(context.Background(), "nats_"+state)
}

func (s kvStore) Get(state, key string) ([]byte, uint64, bool, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, 0, false, err
	}
	e, err := kv.Get(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, 0, false, nil
	}
	if err != nil {
		return nil, 0, false, err
	}
	return e.Value(), e.Revision(), true, nil
}

func (s kvStore) Put(state, key string, value []byte) (uint64, error) {
	kv, err := s.kv(state)
	if err != nil {
		return 0, err
	}
	return kv.Put(context.Background(), key, value)
}

func (s kvStore) Delete(state, key string) error {
	kv, err := s.kv(state)
	if err != nil {
		return err
	}
	return kv.Delete(context.Background(), key)
}

func (s kvStore) Create(state, key string, value []byte) (uint64, error) {
	kv, err := s.kv(state)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Create(context.Background(), key, value)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return 0, keeper.ErrExists
	}
	return rev, err
}

func (s kvStore) Keys(state string) ([]string, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(context.Background())
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatalf("not within 10 s: %s", what)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// The bus module runs whole (design 25 § 12.7, "The bus module runs whole" and "The bus module's first view"; CAT-N10):
// its keeper against the real bus server, planning from the catalog half of every watched branch and the seat list
// (design 47 G §344, §532, §533, §537, §538) at each move of them, creating on the bus,
// saying its events on its own stream BROKER_EVENTS, keeping its state in its own buckets, and acting on a person's
// requests. One server for every case, in order, as the mesh runs them.
func TestTheBusModuleRunsWhole(t *testing.T) {
	dir := t.TempDir()
	s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: dir, NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(5 * time.Second) {
		t.Fatal("the server did not start")
	}
	defer s.Shutdown()
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, _ := jetstream.New(nc)
	ctx := context.Background()
	account := process.Account{Node: "anchor", Module: keeper.Module, User: "anchor.nats.keeper"}
	// What the grants part creates for the bus module from its manifest's old sections: EVENTS and the keeper's
	// consumer on it, its own stream, and its two states.
	for _, c := range []jetstream.StreamConfig{
		{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}},
		{Name: BrokerEventsStream, Subjects: []string{"mesh.stream.nats.event.>"}, MaxAge: 30 * 24 * time.Hour, MaxBytes: 256 << 20},
		{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}},
	} {
		if _, err := js.CreateStream(ctx, c); err != nil {
			t.Fatal(err)
		}
	}
	// The keeper's consumer hears the indexer's verdicts and the operator channel's decided for the bus module's own
	// asks, as the grants part composes it from its consumes and the seats it uses.
	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
		FilterSubjects: []string{DecidedSubject}, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
		t.Fatal(err)
	}
	// The router's own record of the asks, which the keeper reads under its own process account (process.Conn.Ask), never
	// a stand-in: an ask's record is `<asker>.<ask id>`, as the router keeps it.
	asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: asksBucket, History: 5})
	if err != nil {
		t.Fatal(err)
	}
	pconn, err := process.Connect(process.Account{URL: s.ClientURL(), Node: account.Node, Module: account.Module, User: account.User, Password: "p"})
	if err != nil {
		t.Fatal(err)
	}
	defer pconn.Close()
	for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent, keeper.StatePlans, "findings"} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil {
			t.Fatal(err)
		}
	}
	store := kvStore{js}
	snapshots := t.TempDir()
	var mu sync.Mutex
	// The catalog half as meshcatalogclient's branch follower holds it, branch by branch in the registrations' order, and
	// the seat list's definitions; each change moves the keeper, as the follower's and the seat list's changes do.
	branches := []Branch{{Name: "main", Revision: 1, Entries: map[string]catalog.CatalogEntry{}}}
	order := []string{"main"}
	defs := map[string]seatdef.Definition{}
	moved := make(chan struct{}, 1)
	move := func() {
		select {
		case moved <- struct{}{}:
		default:
		}
	}
	var acts []string
	// The bus module's findings go through a held writer over its state findings, started as the keeper's process starts it.
	writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{
		Asked: meshconditions.SubjectFindingsAsked, Module: keeper.Module, Node: account.Node, State: pconn.State("findings"),
		Kinds: []string{keeper.FindingSnapshotsFilling}, JS: js, Conn: nc, Logf: t.Logf})
	if err != nil {
		t.Fatal(err)
	}
	if err := writer.Start(ctx); err != nil {
		t.Fatal(err)
	}
	defer writer.Close()
	newKeeper := func() *keeper.Keeper {
		return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots), NC: nc, Dir: snapshots, Log: t.Logf}, Store: store,
			Finding: writer, Emit: streamEmitter(js, account), Say: sayer(js),
			Catalog: func() (keeper.CatalogView, error) {
				mu.Lock()
				defer mu.Unlock()
				return ViewOf(branches, func(string) []string { return order })
			},
			Seats: func() map[string]seatdef.Definition { mu.Lock(); defer mu.Unlock(); return maps.Clone(defs) },
			Again: func() error { return nil },
			Act: func(what, why, cause string) error {
				mu.Lock()
				defer mu.Unlock()
				acts = append(acts, cause+": "+what)
				return nil
			},
			Warrant: WarrantFrom(pconn.Ask),
			Logf:    t.Logf}
	}
	run := func(k *keeper.Keeper) func() {
		kctx, stop := context.WithCancel(ctx)
		done := make(chan struct{})
		watch := func(c context.Context, act func()) error {
			kv, _ := js.KeyValue(ctx, "nats_"+keeper.StateRequests)
			w, err := kv.WatchAll(c)
			if err != nil {
				return err
			}
			go func() {
				for range w.Updates() {
					act()
				}
			}()
			return nil
		}
		go func() { _ = Keep(kctx, k, hearer(js, account, nil, nil, t.Logf), watch, moved, t.Logf); close(done) }()
		return func() { stop(); <-done }
	}
	// onBranch puts a module's entry in a branch's newest revision (none: the module leaves it), and moves the keeper.
	onBranch := func(branch, module string, revision int64, manifest map[string]any) {
		mu.Lock()
		for i := range branches {
			if branches[i].Name == branch {
				branches[i].Revision = revision
				if manifest == nil {
					delete(branches[i].Entries, module)
				} else {
					manifest["module"] = module
					branches[i].Entries[module] = catalog.CatalogEntry{Module: module, Verdict: "accepted", ManifestDigest: fmt.Sprintf("sha256:%d", revision), Manifest: manifest}
				}
			}
		}
		mu.Unlock()
		move()
	}
	accepted := func(module string, revision uint64, entries ...catalogue.BusStateEntry) {
		onBranch("main", module, int64(revision), map[string]any{"bus-state": entries})
	}
	// saidOn reads what one stream holds of the bus module's events, by its subjects' prefix: its own stream (its
	// history), or EVENTS on the seat mesh-broker's subjects (what every consumer of mesh-broker.<event> hears).
	saidOn := func(stream, prefix string) []string {
		cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{prefix + ">"}})
		if err != nil {
			t.Fatal(err)
		}
		info, _ := cons.Info(ctx)
		var out []string
		if info.NumPending == 0 {
			return nil
		}
		batch, _ := cons.Fetch(int(info.NumPending), jetstream.FetchMaxWait(2*time.Second))
		for m := range batch.Messages() {
			var e keeper.Event
			_ = json.Unmarshal(m.Data(), &e)
			out = append(out, strings.TrimPrefix(m.Subject(), prefix)+" "+e.Module+" "+e.Local)
		}
		return out
	}
	said := func() []string { return saidOn(BrokerEventsStream, "mesh.stream.nats.event.") }
	saidHas := func(want string) func() bool {
		return func() bool {
			for _, l := range said() {
				if l == want {
					return true
				}
			}
			return false
		}
	}
	held := func(stream string) (*jetstream.StreamInfo, bool) {
		st, err := js.Stream(ctx, stream)
		if err != nil {
			return nil, false
		}
		info, err := st.Info(ctx)
		return info, err == nil
	}
	state := entry("drafts", "key-value", "state", nil)
	log := entry("log", "stream", "log", map[string]string{"max-mib": "1"})

	// Review round 1 (running F1): the fill pass runs from the start, while the first view is still asked, so the store's
	// one guard never waits on the indexer.
	t.Run("the fill pass runs while the first view is asked", func(t *testing.T) {
		down := newKeeper()
		down.Catalog = func() (keeper.CatalogView, error) {
			return keeper.CatalogView{}, errors.New("the catalog half has no first view of main yet")
		}
		stopDown := run(down)
		eventually(t, "a fill pass with no first view", func() bool { return !down.Status().Pass.IsZero() && down.Status().FirstView.Answered.IsZero() })
		stopDown()
		// And while its consumer is bound again on its backoff (review round 2, running F1).
		was := FillEvery
		FillEvery = 20 * time.Millisecond
		defer func() { FillEvery = was }()
		deaf := newKeeper()
		kctx, stopDeaf := context.WithCancel(ctx)
		done := make(chan struct{})
		go func() {
			_ = Keep(kctx, deaf, func(context.Context, func(string, []byte) error) error { return errors.New("no consumer yet") }, nil, nil, t.Logf)
			close(done)
		}()
		eventually(t, "the first view answered", func() bool { return !deaf.Status().FirstView.Answered.IsZero() })
		first := deaf.Status().Pass
		eventually(t, "a fill pass while the consumer is bound again", func() bool { return deaf.Status().Pass.After(first) })
		stopDeaf()
		<-done
	})
	k := newKeeper()
	stop := run(k)
	t.Run("a verdict, then created, with its binding", func(t *testing.T) {
		accepted("m", 2, state, log)
		eventually(t, "created m drafts and m log", func() bool { return saidHas("created m drafts")() && saidHas("created m log")() })
		// Said on the seat's subject as well, on the shared events stream (the lead's ruling, 13:0x).
		if seat := saidOn("EVENTS", "mesh.seat.mesh-broker.event."); !slices.Contains(seat, "created m drafts") ||
			!slices.Contains(seat, "created m log") {
			t.Errorf("EVENTS holds %v of mesh-broker's events: created m drafts and m log, as on the bus module's own stream", seat)
		}
		if _, ok := held("KV_m_drafts"); !ok {
			t.Error("the bus holds no KV_m_drafts")
		}
		if info, ok := held("m_log"); !ok || info.Config.Discard != jetstream.DiscardNew || info.Config.Subjects[0] != "mesh.bus.m.log.>" {
			t.Errorf("m_log is %+v: a log refuses when full, on the module's own subjects", info)
		}
	})
	t.Run("a restart catches up a verdict given while it was down", func(t *testing.T) {
		stop()
		accepted("n", 3, entry("jobs", "stream", "work-queue", nil))
		k = newKeeper()
		stop = run(k)
		eventually(t, "created n jobs", saidHas("created n jobs"))
	})
	t.Run("drift is said, a growth applied and the drift cleared", func(t *testing.T) {
		accepted("m", 5, state, entry("log", "stream", "log", map[string]string{"max-mib": "1", "max-age-days": "2"}))
		eventually(t, "drifted m log", saidHas("drifted m log"))
		accepted("m", 6, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "drift-cleared m log", saidHas("drift-cleared m log"))
		if info, _ := held("m_log"); info.Config.MaxBytes != 4<<20 {
			t.Errorf("m_log's cap is %d: the larger cap applied", info.Config.MaxBytes)
		}
	})
	t.Run("a refused create hands no binding", func(t *testing.T) {
		bare := entry("raw", "stream", "", nil)
		accepted("m", 7, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}), bare)
		eventually(t, "refused m raw", saidHas("refused m raw"))
		if saidHas("created m raw")() {
			t.Error("created m raw")
		}
	})
	t.Run("each template at a small cap refuses, full raised at the raise mark and cleared below the clear mark", func(t *testing.T) {
		for i := 0; ; i++ {
			_, err := js.Publish(ctx, fmt.Sprintf("mesh.bus.m.log.k%d", i), make([]byte, 200<<10))
			if err != nil {
				break
			}
			if i > 40 {
				t.Fatal("a 4 MiB log took more than 8 MiB")
			}
		}
		k.FillPass()
		eventually(t, "full m log", saidHas("full m log"))
		st, _ := js.Stream(ctx, "m_log")
		_ = st.Purge(ctx)
		k.FillPass()
		eventually(t, "full-cleared m log", saidHas("full-cleared m log"))
	})
	t.Run("an orphan is listed, and goes only by drop on the operator's warrant, after a snapshot", func(t *testing.T) {
		accepted("m", 8, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "orphaned m drafts", saidHas("orphaned m drafts"))
		put := func(r keeper.Request) {
			raw, _ := json.Marshal(r)
			if _, err := store.Put(keeper.StateRequests, r.ID, raw); err != nil {
				t.Fatal(err)
			}
		}
		get := func(id string) keeper.Request {
			raw, _, _, _ := store.Get(keeper.StateRequests, id)
			var r keeper.Request
			_ = json.Unmarshal(raw, &r)
			return r
		}
		put(keeper.Request{ID: "r1", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused"})
		eventually(t, "the drop with no ask refused", func() bool { return strings.Contains(get("r1").Error, "operator's warrant") })
		r := keeper.Request{ID: "r2", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused", At: time.Now()}
		a, _ := askFor(r, time.Now())
		r.Ask = a.ID
		record := func(w *oc.Warrant) {
			raw, _ := json.Marshal(map[string]any{"asker": keeper.Module, "ask": a, "state": "open", "warrant": w})
			if _, err := asksKV.Put(ctx, keeper.Module+"."+a.ID, raw); err != nil {
				t.Fatal(err)
			}
		}
		record(nil)
		put(r)
		eventually(t, "the request read, waiting for the operator", func() bool { return get("r2").Waiting != "" })
		if _, ok := held("KV_m_drafts"); !ok || !get("r2").Done.IsZero() {
			t.Fatal("dropped before the operator chose")
		}
		// The operator chooses: the router records the warrant and says decided; the keeper acts on decided, on its
		// durable consumer, with no timer between and no call of its pass by the test (seams F1).
		w := &oc.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
			Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(time.Now()), AskDigest: occ.Digest(a)}
		record(w)
		decided := time.Now()
		body, _ := json.Marshal(w)
		if _, err := js.Publish(ctx, DecidedSubject, body); err != nil {
			t.Fatal(err)
		}
		eventually(t, "dropped m drafts", saidHas("dropped m drafts"))
		if since := time.Since(decided); since > 5*time.Second {
			t.Errorf("dropped %s after decided: within seconds", since)
		}
		if _, ok := held("KV_m_drafts"); ok {
			t.Error("KV_m_drafts is still on the bus")
		}
		if p := get("r2").Snapshot; p == "" {
			t.Error("no snapshot taken before the drop")
		} else if st, err := os.Stat(p); err != nil || st.Size() == 0 {
			t.Errorf("the snapshot %s is not there whole: %v", p, err)
		}
		mu.Lock()
		recorded := strings.Join(acts, "\n")
		mu.Unlock()
		if !strings.Contains(recorded, "nats_drop: dropped drafts of m") {
			t.Errorf("recorded %s: the drop as a hand act", recorded)
		}
		// The same warrant handed in a second request is refused and deletes nothing (security F2): drafts asked for and
		// orphaned again, then a request naming the spent ask.
		accepted("m", 9, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "m drafts made again", func() bool { _, ok := held("KV_m_drafts"); return ok })
		accepted("m", 10, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "m drafts orphaned again", func() bool { return k.Held()["m"].Made["drafts"].Orphaned })
		again := keeper.Request{ID: "r2b", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "again", Ask: a.ID, At: time.Now()}
		put(again)
		eventually(t, "the second request on the spent warrant refused", func() bool { return strings.Contains(get("r2b").Error, "already acted") })
		if _, ok := held("KV_m_drafts"); !ok {
			t.Error("the spent warrant deleted KV_m_drafts a second time")
		}
	})
	// Design 47 G §681, A-107: the snapshot the keeper wrote is sent by the keeper's own process (snapshot-get answered through
	// AnswerSeat, the chunks published by the process and acknowledged to its own subscription), restored in place by the keeper
	// (refused while its name is live, ended by snapshot-restored) and removed by snapshot-remove, no call handed through the requests.
	t.Run("a snapshot is sent, restored in place and removed by the keeper's process, which answers the seat itself", func(t *testing.T) {
		get := func(id string) keeper.Request {
			raw, _, _, _ := store.Get(keeper.StateRequests, id)
			var r keeper.Request
			_ = json.Unmarshal(raw, &r)
			return r
		}
		rec := get("r2")
		if rec.Path == "" || rec.Digest == "" || rec.Protected {
			t.Fatalf("the drop's record holds no snapshot: %+v", rec)
		}
		world := keeperSnapshots(nc, t.Logf)
		// The seat's definition as the seat list holds it, the bootstrapped one (A-186).
		broker := func() (seatdef.Definition, bool) {
			for _, d := range seatlist.Bootstrapped() {
				if d.Seat == meshbroker.Seat {
					return d, true
				}
			}
			return seatdef.Definition{}, false
		}
		stopAnswering, err := pconn.AnswerSeat(meshbroker.Seat, broker, snapshotAnswers(k, heldRequest(store), world))
		if err != nil {
			t.Fatal(err)
		}
		defer stopAnswering()
		// A caller of the seat: its call carries its bus name as the last token, and it acknowledges each chunk it hears.
		verb := func(name string, args map[string]any, into any) {
			t.Helper()
			raw, _ := json.Marshal(args)
			msg, err := nc.Request("mesh.seat.mesh-broker.call."+name+".person~jo", raw, 10*time.Second)
			if err != nil {
				t.Fatal(err)
			}
			var reply struct {
				Result json.RawMessage `json:"result"`
				Error  string          `json:"error"`
			}
			if err := json.Unmarshal(msg.Data, &reply); err != nil || reply.Error != "" {
				t.Fatalf("%s answered %s (%v)", name, msg.Data, err)
			}
			if err := json.Unmarshal(reply.Result, into); err != nil {
				t.Fatal(err)
			}
		}
		call := "0123456789abcdef0123456789abcdef"
		var chunkMu sync.Mutex
		var chunks []Chunk
		hear, err := nc.Subscribe("mesh.seat.mesh-broker.chunk.snapshot-get."+call+".*", func(m *nats.Msg) {
			var c Chunk
			_ = json.Unmarshal(m.Data, &c)
			chunkMu.Lock()
			chunks = append(chunks, c)
			chunkMu.Unlock()
			go func() {
				_, _ = nc.Request(strings.Replace(m.Subject, ".chunk.", ".ack.", 1), []byte("{}"), 5*time.Second)
			}()
		})
		if err != nil {
			t.Fatal(err)
		}
		defer hear.Unsubscribe()
		var got meshbroker.SnapshotGetAnswer
		verb(meshbroker.VerbSnapshotGet, map[string]any{"request": "r2", "call": call}, &got)
		if got.Error != "" || got.Digest != rec.Digest || got.Chunks < 1 || got.Stream != "m_drafts" {
			t.Fatalf("snapshot-get answered %+v for the keeper's record %+v", got, rec)
		}
		var whole []byte
		eventually(t, "the last chunk", func() bool {
			chunkMu.Lock()
			defer chunkMu.Unlock()
			return len(chunks) > 0 && int64(len(chunks)) == chunks[0].Of
		})
		chunkMu.Lock()
		for _, c := range chunks {
			whole = append(whole, c.Data...)
		}
		chunkMu.Unlock()
		if sum := sha256.Sum256(whole); "sha256:"+hex.EncodeToString(sum[:]) != rec.Digest || int64(len(whole)) != rec.Size {
			t.Errorf("the chunks hold %d bytes, not the snapshot the keeper wrote (%d)", len(whole), rec.Size)
		}
		restore := func(why string) meshbroker.SnapshotRestoreAnswer {
			var o meshbroker.SnapshotRestoreAnswer
			verb(meshbroker.VerbSnapshotRestore, map[string]any{"request": "r2", "why": why}, &o)
			return o
		}
		// KV_m_drafts is live (made again after the drop): nothing is restored over it.
		if a := restore("live"); !strings.Contains(a.Error, "names live") {
			t.Errorf("a restore over a live stream answered %+v", a)
		}
		// The state's snapshot holds its objects store beside it (design 47 G §58): a restore goes over neither.
		for _, name := range []string{"KV_m_drafts", "OBJ_m_drafts-objects"} {
			if err := js.DeleteStream(ctx, name); err != nil {
				t.Fatal(err)
			}
		}
		a := restore("it was removed by mistake")
		if a.Accepted == "" {
			t.Fatalf("restore answered %+v", a)
		}
		eventually(t, "snapshot-restored on the seat's subject", func() bool {
			for _, l := range saidOn("EVENTS", "mesh.seat.mesh-broker.event.snapshot-restored.") {
				if strings.HasPrefix(l, "r2."+a.Accepted+" ") {
					return true
				}
			}
			return false
		})
		if _, ok := held("KV_m_drafts"); !ok || get(a.Accepted).Result != "done" {
			t.Errorf("the stream is not restored in place: held %v, the act %+v", ok, get(a.Accepted))
		}
		var o meshbroker.SnapshotRemoveAnswer
		verb(meshbroker.VerbSnapshotRemove, map[string]any{"request": "r2", "why": "no longer needed"}, &o)
		if o.Accepted == "" {
			t.Fatalf("snapshot-remove answered %+v", o)
		}
		if _, err := os.Stat(rec.Path); !os.IsNotExist(err) {
			t.Errorf("the snapshot file %s is still there: %v", rec.Path, err)
		}
		if _, _, found, _ := store.Get(keeper.StateRequests, "r2"); found {
			t.Error("the snapshot's record outlived its file")
		}
	})
	// §532: a branch registered while the keeper runs is followed with no restart; a feature branch asking a local name with
	// other settings than the default branch's is said drifted, naming it, while the default branch's are planned, and its
	// new local name is planned from it.
	t.Run("a branch registered while it runs is planned from, a local name with other settings drifted", func(t *testing.T) {
		mu.Lock()
		branches = append(branches, Branch{Name: "feature", Revision: 4, Entries: map[string]catalog.CatalogEntry{"m": {Module: "m", Verdict: "accepted",
			Manifest: map[string]any{"module": "m", "bus-state": []any{state, entry("log", "stream", "log", map[string]string{"max-mib": "2"}), entry("extra", "key-value", "state", nil)}}}}})
		order = []string{"main", "feature"}
		mu.Unlock()
		move()
		eventually(t, "created m extra", saidHas("created m extra"))
		eventually(t, "drifted m log", saidHas("drifted m log"))
		if info, _ := held("m_log"); info.Config.MaxBytes != 4<<20 {
			t.Errorf("m_log's cap is %d: the default branch's 4 MiB planned", info.Config.MaxBytes)
		}
		mu.Lock()
		branches, order = branches[:1], []string{"main"}
		mu.Unlock()
		move()
		eventually(t, "orphaned m extra", saidHas("orphaned m extra"))
	})
	// §344, §533, §537: a module of the new form claiming a seat with a verb gets a sent store, one with neither none, a seat
	// whose verbs come only from an included part one, a seat including only bodies none, a foundation part none; a
	// definition gaining a verb gives its holder one. §538: a module in no watched branch's newest revision loses its plans key.
	t.Run("the sent stores follow the catalog half and the seat list", func(t *testing.T) {
		mu.Lock()
		defs["verbed"] = seatdef.Definition{Seat: "verbed", Verbs: []seatdef.Verb{{Name: "go"}}}
		defs["bare"] = seatdef.Definition{Seat: "bare"}
		defs["resolved"] = seatdef.Definition{Seat: "resolved", Includes: []string{"part"}, Verbs: []seatdef.Verb{{Name: "from-the-part"}}}
		defs["bodies"] = seatdef.Definition{Seat: "bodies", Includes: []string{"shared-bodies"}}
		mu.Unlock()
		claims := func(seat string) map[string]any { return map[string]any{"claims": []any{map[string]any{"name": seat}}} }
		onBranch("main", "sa", 5, claims("verbed"))
		onBranch("main", "sb", 5, claims("bare"))
		onBranch("main", "sc", 5, claims("resolved"))
		onBranch("main", "sd", 5, claims("bodies"))
		f := claims("verbed")
		f["objects"] = []any{map[string]any{"name": "sent-objects", "max-mib": 256}}
		onBranch("main", "se", 5, f)
		eventually(t, "sa and sc's sent stores", func() bool {
			_, a := held("OBJ_sa_sent-objects")
			_, c := held("OBJ_sc_sent-objects")
			return a && c
		})
		for _, none := range []string{"sb", "sd", "se"} {
			if _, ok := held("OBJ_" + none + "_sent-objects"); ok {
				t.Errorf("%s has a sent store", none)
			}
		}
		mu.Lock()
		defs["bodies"] = seatdef.Definition{Seat: "bodies", Includes: []string{"shared-bodies"}, Verbs: []seatdef.Verb{{Name: "new"}}}
		mu.Unlock()
		move()
		eventually(t, "sd's sent store once its seat gains a verb", func() bool { _, ok := held("OBJ_sd_sent-objects"); return ok })
		onBranch("main", "sb", 6, nil)
		eventually(t, "sb's plan deleted", func() bool { _, _, found, _ := store.Get(keeper.StatePlans, "sb"); return !found })
	})
	t.Run("nats_first_view_again asks at once and records the hand act", func(t *testing.T) {
		raw, _ := json.Marshal(keeper.Request{ID: "r3", Verb: keeper.VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"})
		_, _ = store.Put(keeper.StateRequests, "r3", raw)
		eventually(t, "the first view asked again", func() bool {
			mu.Lock()
			defer mu.Unlock()
			return strings.Contains(strings.Join(acts, "\n"), "nats_first_view_again")
		})
	})
	stop()
}

func entry(name, storage, template string, options map[string]string) catalogue.BusStateEntry {
	e := catalogue.BusStateEntry{Name: name, Storage: storage, Template: template}
	if len(options) > 0 {
		e.Options = map[string]json.RawMessage{}
		for k, v := range options {
			e.Options[k] = json.RawMessage(v)
		}
	}
	return e
}