CAT-N10: the bus module's one whole-module integration test, against the real bus server

Design 25 § 12.7's rows, as subtests of one test on one server (CAT-N10): a verdict, then created on its own stream BROKER_EVENTS, the objects on the bus; a restart catching up a verdict given while it was down; a replay older than the first view not applied and a module missing from the first view keeping its key; a drift said, a growth applied and the drift cleared; a refused create with no created; a log at a small cap refusing, full raised and cleared; an orphan dropped only on the operator's warrant, after a snapshot written whole, recorded as a hand act; nats_first_view_again's hand act. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:34:59 +01:00
parent 76aab05289
commit fa4ccbc0a9
1 changed files with 359 additions and 0 deletions
@@ -0,0 +1,359 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"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-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"nats-tools/internal/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) 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, hearing manifest-accepted on its durable consumer, 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.>"}},
+		{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)
+		}
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
+		FilterSubject: AcceptedSubject, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
+		t.Fatal(err)
+	}
+	for _, st := range []string{keeper.StateAsked, keeper.StateRequests} {
+		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
+	view := keeper.FirstView{Revision: 1}
+	var acts []string
+	var ask *asks.Ask
+	var warrant *asks.Warrant
+	newKeeper := func() *keeper.Keeper {
+		return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots)}, Store: store,
+			Emit: streamEmitter(js, account), Say: sayer(js),
+			View: func() (keeper.FirstView, error) { mu.Lock(); defer mu.Unlock(); return view, nil },
+			Act: func(what, why, cause string) error {
+				mu.Lock()
+				defer mu.Unlock()
+				acts = append(acts, cause+": "+what)
+				return nil
+			},
+			Warrant: func(string) (*asks.Ask, *asks.Warrant, error) { mu.Lock(); defer mu.Unlock(); return ask, warrant, nil },
+			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, t.Logf), watch, t.Logf); close(done) }()
+		return func() { stop(); <-done }
+	}
+	accepted := func(module string, revision uint64, entries ...catalogue.BusStateEntry) {
+		body, _ := json.Marshal(map[string]any{"v": 1, "module": module, "digest": fmt.Sprintf("sha256:%d", revision),
+			"judge": "j", "revision": revision, "bus-state": entries})
+		if _, err := js.Publish(ctx, AcceptedSubject, body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	said := func() []string {
+		cons, err := js.OrderedConsumer(ctx, BrokerEventsStream, jetstream.OrderedConsumerConfig{})
+		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(), "mesh.stream.nats.event.")+" "+e.Module+" "+e.Local)
+		}
+		return out
+	}
+	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"})
+
+	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")() })
+		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("a replay older than the first view is not applied; a set-aside module keeps its key", func(t *testing.T) {
+		stop()
+		mu.Lock()
+		view = keeper.FirstView{Revision: 9, Modules: []keeper.Accepted{{Module: "p", Revision: 9, BusState: []catalogue.BusStateEntry{state}}}}
+		mu.Unlock()
+		accepted("p", 4)
+		k = newKeeper()
+		stop = run(k)
+		eventually(t, "created p drafts", saidHas("created p drafts"))
+		eventually(t, "the replay heard", func() bool { info, _ := js.Consumer(ctx, "EVENTS", account.QueueConsumerName()); i, _ := info.Info(ctx); return i.NumPending == 0 && i.NumAckPending == 0 })
+		if saidHas("orphaned p drafts")() {
+			t.Error("the replay at revision 4 orphaned what the first view at 9 asked")
+		}
+		if _, _, found, _ := store.Get(keeper.StateAsked, "m"); !found {
+			t.Error("m, missing from the first view, lost its key")
+		}
+	})
+	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"}
+		a, _ := askFor(r, time.Now())
+		r.Ask = a.ID
+		mu.Lock()
+		ask = &a
+		mu.Unlock()
+		put(r)
+		k.ActOnRequests()
+		if _, ok := held("KV_m_drafts"); !ok || !get("r2").Done.IsZero() {
+			t.Fatal("dropped before the operator chose")
+		}
+		mu.Lock()
+		warrant = &asks.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: asks.OutcomeChosen, By: &asks.Person{Who: asks.Operator},
+			Option: "go", Label: "Remove", Level: asks.Destroy, At: time.Now(), AskDigest: a.Digest()}
+		mu.Unlock()
+		k.ActOnRequests()
+		eventually(t, "dropped m drafts", saidHas("dropped m drafts"))
+		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)
+		}
+	})
+	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
+}