From fa4ccbc0a956876728cb542da1802a2288a5566a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:03:32 +0100 Subject: [PATCH] CAT-N10: the bus module's one whole-module integration test, against the real bus server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/nats-tools/keeper_integration_test.go | 359 ++++++++++++++++++ 1 file changed, 359 insertions(+) create mode 100644 modules/nats/cmd/nats-tools/keeper_integration_test.go diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go new file mode 100644 index 00000000..cacbb570 --- /dev/null +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -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 +}