novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

NATS6: the bus module keeps its plans and plans again only when a module's entries differ; nats_plans reads them

Browse Source 
Builds: design 47 G §538

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:00:34 +01:00
parent 78b12be752
commit 6277067465
7 changed files with 161 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/broker.go
+54
View File
@@ -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
}}
}
modules/nats/cmd/nats-tools/broker_test.go
+16
View File
@@ -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)
}
}
modules/nats/cmd/nats-tools/main.go
+1
View File
@@ -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)...)
}
modules/nats/cmd/nats-tools/manifest_test.go
+4 -4
View File
@@ -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 {
modules/nats/internal/keeper/keeper.go
+52
View File
@@ -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)
modules/nats/internal/keeper/keeper_test.go
+26
View File
@@ -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
modules/nats/module.json
+8 -1
View File
@@ -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,