From 62770674657a5618eaa9747121e6f6d59d274a76 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:00:34 +0100 Subject: [PATCH] NATS6: the bus module keeps its plans and plans again only when a module's entries differ; nats_plans reads them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §538 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/nats/cmd/nats-tools/broker.go | 54 ++++++++++++++++++++ modules/nats/cmd/nats-tools/broker_test.go | 16 ++++++ modules/nats/cmd/nats-tools/main.go | 1 + modules/nats/cmd/nats-tools/manifest_test.go | 8 +-- modules/nats/internal/keeper/keeper.go | 52 +++++++++++++++++++ modules/nats/internal/keeper/keeper_test.go | 26 ++++++++++ modules/nats/module.json | 9 +++- 7 files changed, 161 insertions(+), 5 deletions(-) diff --git a/modules/nats/cmd/nats-tools/broker.go b/modules/nats/cmd/nats-tools/broker.go index 9647be55b..a78b45d97 100644 --- a/modules/nats/cmd/nats-tools/broker.go +++ b/modules/nats/cmd/nats-tools/broker.go @@ -331,3 +331,57 @@ func sortedNames[V any](m map[string]V) []string { sort.Strings(out) return out } + +// PlansInState reads the keeper's state plans (design 47 G §538): each module's entries as last planned. +func PlansInState() (map[string]keeper.Planned, error) { + s := stdio.State(keeper.StatePlans) + keys, err := s.Keys() + if err != nil { + return nil, fmt.Errorf("the keeper's own %s cannot be read: %w", keeper.StatePlans, err) + } + out := map[string]keeper.Planned{} + for _, key := range keys { + e, err := s.Get(key) + if err != nil || e == nil { + continue + } + var p keeper.Planned + if json.Unmarshal(e.Value, &p) == nil { + out[key] = p + } + } + return out, nil +} + +// plansTool is nats_plans: what the bus module planned for each module, from its own state plans (design 47 G §538). +func plansTool(read func() (map[string]keeper.Planned, error)) stdio.Tool { + return stdio.Tool{Name: "nats_plans", + Description: "What the bus module planned for each module: the bus-state entries it last planned, their digest (a verdict whose entries digest the same plans nothing), the revision of the verdict and when. Narrowed by a module.", + Input: map[string]any{"module": str("one module")}, + Run: func(a map[string]any) (any, error) { + plans, err := read() + if err != nil { + return nil, err + } + type line struct { + Module string `json:"module"` + Digest string `json:"digest"` + Revision uint64 `json:"revision"` + Entries []string `json:"entries"` + At string `json:"at"` + } + out := []line{} + for _, module := range sortedNames(plans) { + if m := arg(a, "module"); m != "" && m != module { + continue + } + p := plans[module] + names := []string{} + for _, e := range p.Entries { + names = append(names, e.Name) + } + out = append(out, line{Module: module, Digest: p.Digest, Revision: p.Revision, Entries: names, At: p.At.UTC().Format(time.RFC3339)}) + } + return map[string]any{"plans": out}, nil + }} +} diff --git a/modules/nats/cmd/nats-tools/broker_test.go b/modules/nats/cmd/nats-tools/broker_test.go index b3fb30360..a30b5407b 100644 --- a/modules/nats/cmd/nats-tools/broker_test.go +++ b/modules/nats/cmd/nats-tools/broker_test.go @@ -131,3 +131,19 @@ func TestTheServersStorageLimitIsReadFromItsMonitor(t *testing.T) { } } } + +// nats_plans answers each module's entries as last planned, narrowed by a module (design 47 G §538). +func TestNatsPlansReadsTheKeepersPlans(t *testing.T) { + read := func() (map[string]keeper.Planned, error) { + return map[string]keeper.Planned{ + "a": {Module: "a", Digest: "sha256:1", Revision: 3, Entries: []catalogue.BusStateEntry{{Name: "drafts"}}}, + "b": {Module: "b", Digest: "sha256:2", Revision: 4}, + }, nil + } + tool := plansTool(read) + out, err := tool.Run(map[string]any{"module": "a"}) + raw, _ := json.Marshal(out) + if err != nil || !strings.Contains(string(raw), `"entries":["drafts"]`) || !strings.Contains(string(raw), `"digest":"sha256:1"`) || strings.Contains(string(raw), `"module":"b"`) { + t.Errorf("nats_plans answered %s (%v)", raw, err) + } +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index c915c3f55..1f5e82f4b 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -197,5 +197,6 @@ func allTools() []stdio.Tool { out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) rs := RequestsInState() out = append(out, snapshotListTool(rs, int(number(os.Getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)))...) + out = append(out, plansTool(PlansInState)) return append(out, requestTools(rs)...) } diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index 998cb2a9f..3e3214030 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -228,7 +228,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { } // spent, the warrants the keeper spent, is written by the keeper alone and only read by the tools (review round 2, // security F2: the tool runner's account, which writes requests, may never delete a spend record). - if !slices.Equal(states, []string{"asked", "requests", "spent", "findings"}) || !slices.Equal(readOnly, []string{"spent"}) { + if !slices.Equal(states, []string{"asked", "requests", "spent", "plans", "findings"}) || !slices.Equal(readOnly, []string{"spent", "plans"}) { t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), and spent, read-only", states, readOnly) } if _, ok := nm.OwnSecrets["broker-keeper"]; !ok { @@ -238,7 +238,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { bus, _ := keeper["bus"].(map[string]any) health, _ := keeper["health"].(map[string]any) if keeper["type"] != "process" || keeper["artifact"] != "tools" || fmt.Sprint(keeper["run"]) != "[./nats-tools keep]" || - fmt.Sprint(bus["state"]) != "[asked requests spent findings]" || + fmt.Sprint(bus["state"]) != "[asked requests spent plans findings]" || health["kind"] != "tool" || health["tool"] != "nats_keeper_health" { t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests and spent, judged by nats_keeper_health", keeper) } @@ -248,8 +248,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { t.Fatal(err) } for _, st := range whole.State { - if (st.Name == "spent") != st.ReadOnly { - t.Errorf("the SDK reads %s read-only %v: spent alone is read-only", st.Name, st.ReadOnly) + if (st.Name == "spent" || st.Name == "plans") != st.ReadOnly { + t.Errorf("the SDK reads %s read-only %v: spent and plans alone are read-only", st.Name, st.ReadOnly) } } if b := whole.ProcessBuses()["keeper"]; !b.Keeps { diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index a25ba9e08..a6b69975b 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -3,6 +3,7 @@ package keeper import ( "context" "crypto/rand" + "crypto/sha256" "encoding/hex" "encoding/json" "errors" @@ -26,6 +27,9 @@ const Module = "nats" const ( StateAsked = "asked" StateRequests = "requests" + // StatePlans holds, per module, the bus-state entries as last planned and their digest (design 47 G §538): what a + // verdict's entries are compared with, never the manifest's digest, which changes with any field. + StatePlans = "plans" // KeyStatus is the keeper's own status in StateAsked, read by its health tool: a key no module can have, since a // module's name holds no underscore. KeyStatus = "_keeper" @@ -92,6 +96,24 @@ type FirstView struct { Modules []Accepted `json:"modules"` } +// Planned is the value of a module's key in StatePlans: the entries as planned, their digest and the revision of the verdict. +type Planned struct { + V int `json:"v"` + Module string `json:"module"` + Digest string `json:"digest"` + Revision uint64 `json:"revision"` + Entries []catalogue.BusStateEntry `json:"entries"` + At time.Time `json:"at"` +} + +// EntriesDigest is the digest of bus-state entries alone (design 47 G §538): equal entries have an equal digest whatever +// else a manifest changed. +func EntriesDigest(entries []catalogue.BusStateEntry) string { + raw, _ := json.Marshal(entries) + sum := sha256.Sum256(raw) + return "sha256:" + hex.EncodeToString(sum[:]) +} + // Asked is what the keeper holds of one module, the value of its key in StateAsked: what its newest accepted manifest // asks for, and every entry the keeper made for it and the bus holds, by local name. type Asked struct { @@ -455,10 +477,22 @@ func (k *Keeper) Heard(a Accepted) { if held.Made == nil { held.Made = map[string]Made{} } + // It plans again only when its bus-state entries differ from those it last planned, never on the manifest's digest + // (design 47 G §538). + digest := EntriesDigest(a.BusState) + if p, ok := k.planned(a.Module); ok && found && p.Digest == digest { + held.Digest, held.At = a.Digest, k.now() + if a.Revision > held.Revision { + held.Revision = a.Revision + } + k.write(held) + return + } held.Digest, held.Entries, held.At = a.Digest, a.BusState, k.now() if a.Revision > held.Revision { held.Revision = a.Revision } + defer k.writePlan(a.Module, digest, held.Revision, a.BusState) asked := map[string]bool{} for _, e := range a.BusState { asked[e.Name] = true @@ -829,6 +863,23 @@ func EventID() string { return hex.EncodeToString(b[:]) } +func (k *Keeper) planned(module string) (Planned, bool) { + raw, _, found, err := k.Store.Get(StatePlans, module) + if err != nil || !found { + return Planned{}, false + } + var p Planned + return p, json.Unmarshal(raw, &p) == nil +} + +// writePlan keeps the entries a module was last planned for and their digest (§538). +func (k *Keeper) writePlan(module, digest string, revision uint64, entries []catalogue.BusStateEntry) { + raw, _ := json.Marshal(Planned{V: 1, Module: module, Digest: digest, Revision: revision, Entries: entries, At: k.now()}) + if _, err := k.Store.Put(StatePlans, module, raw); err != nil { + k.logf("the plan of %s cannot be written to its own %s: %v", module, StatePlans, err) + } +} + func (k *Keeper) asked(module string) (Asked, uint64, bool) { raw, rev, found, err := k.Store.Get(StateAsked, module) if err != nil || !found { @@ -848,6 +899,7 @@ func (k *Keeper) write(a Asked) { if err := k.Store.Delete(StateAsked, a.Module); err != nil { k.logf("the key of %s cannot be removed from its own %s: %v", a.Module, StateAsked, err) } + _ = k.Store.Delete(StatePlans, a.Module) return } raw, _ := json.Marshal(a) diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 30c07cb47..d1eab8b7a 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -40,6 +40,8 @@ type fakeBus struct { deletedStreams []string // limits are the bus's storage limits the fake answers. limits Limits + // reads counts the reads of an object's stream. + reads int } func (b *fakeBus) SnapshotStreams(path string) ([]string, error) { @@ -88,6 +90,7 @@ func newFakeBus() *fakeBus { func (b *fakeBus) Held(_ context.Context, o Object) (Settings, Fill, bool, error) { b.mu.Lock() defer b.mu.Unlock() + b.reads++ s, ok := b.held[o.Stream()] return s, Fill{Bytes: b.fill[o.Stream()]}, ok, nil } @@ -575,6 +578,29 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { t.Errorf("with no account limit nothing is refused: %+v", k.Held()["m"].Made["log"]) } }}, + {"a verdict plans again only when its entries differ, never on the manifest's digest (§538)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}}) + reads := bus.reads + k.Heard(Accepted{Module: "m", Digest: "d2", Revision: 2, BusState: []catalogue.BusStateEntry{drafts}}) + if bus.reads != reads { + t.Errorf("the bus was read %d times more for a manifest that changed outside its entries", bus.reads-reads) + } + raw, _, found, _ := k.Store.Get(StatePlans, "m") + var p Planned + _ = json.Unmarshal(raw, &p) + if !found || p.Digest != EntriesDigest([]catalogue.BusStateEntry{drafts}) || len(p.Entries) != 1 || k.Held()["m"].Digest != "d2" || k.Held()["m"].Revision != 2 { + t.Errorf("the plan kept %+v (%v), the digest and revision follow the verdict: %+v", p, found, k.Held()["m"]) + } + bigger := entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"history": "7"}) + k.Heard(Accepted{Module: "m", Digest: "d3", Revision: 3, BusState: []catalogue.BusStateEntry{bigger}}) + if bus.reads == reads || bus.updated == nil { + t.Errorf("a changed option plans the difference: reads %d (was %d), updated %v", bus.reads, reads, bus.updated) + } + k.Heard(Accepted{Module: "m", Digest: "d4", Revision: 4, BusState: nil}) + if _, _, found, _ := k.Store.Get(StatePlans, "m"); !found { + t.Errorf("an entry made and now unasked is an orphan the keeper still holds, with its plan") + } + }}, {"a state made before its objects store existed gets the store and its binding said again (§58)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { p, _ := PlanOf("m", drafts) bus.held["KV_m_drafts"] = p.Objects[0].Settings diff --git a/modules/nats/module.json b/modules/nats/module.json index 7acc5ebdd..ba571cf81 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -115,6 +115,11 @@ "ttl-seconds": 86400, "read-only": true }, + { + "name": "plans", + "history": 1, + "read-only": true + }, { "name": "findings", "history": 1 @@ -154,7 +159,8 @@ "nats_keeper_health", "nats_first_view_again", "nats_requests", - "nats_snapshots" + "nats_snapshots", + "nats_plans" ], "guards": [ 8222 @@ -287,6 +293,7 @@ "asked", "requests", "spent", + "plans", "findings" ], "asks": true,