From 6be2e91da8a8420b8e10b7f8bf58467c5f3aa487 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 11:54:55 +0100 Subject: [PATCH] CAT-N7: the seat mesh-broker's reading verbs list, show, check and orphans, and nats_history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Served by the bus module's bundle from the keeper's own state asked and the bus's monitoring endpoint (design 25 § 12.3): each entry's template, bus names, settings asked and held and fill against its cap; check names each object missing or differing; orphans what no accepted manifest asks for. nats_history reads one entry's events, in order, from the bus module's own stream BROKER_EVENTS on the tools' credential. The manifest test now holds every tool and reading verb the manifest names to what the bundle serves. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- modules/nats/cmd/nats-tools/broker.go | 281 +++++++++++++++++++ modules/nats/cmd/nats-tools/broker_test.go | 84 ++++++ modules/nats/cmd/nats-tools/main.go | 15 +- modules/nats/cmd/nats-tools/manifest_test.go | 14 + modules/nats/internal/keeper/jetstream.go | 6 +- modules/nats/module.json | 3 +- 6 files changed, 398 insertions(+), 5 deletions(-) create mode 100644 modules/nats/cmd/nats-tools/broker.go create mode 100644 modules/nats/cmd/nats-tools/broker_test.go diff --git a/modules/nats/cmd/nats-tools/broker.go b/modules/nats/cmd/nats-tools/broker.go new file mode 100644 index 00000000..188c9ffa --- /dev/null +++ b/modules/nats/cmd/nats-tools/broker.go @@ -0,0 +1,281 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "github.com/nats-io/nats.go/jetstream" + + "nats-tools/internal/keeper" +) + +// The seat mesh-broker's reading verbs (design 25 § 12.3), served by this bundle: every entry the keeper made, one +// entry, what the bus holds against what was asked, and the orphans; each read from the keeper's own state asked and +// the bus's monitoring endpoint, never from another part's state. And nats_history: every event the bus module said of +// one entry, in order. + +// heldEntries reads the keeper's own state asked: every module it holds. +type heldEntries func() (map[string]keeper.Asked, error) + +// HeldInState reads asked through the runtime that launched this bundle. +func HeldInState() (map[string]keeper.Asked, error) { + s := stdio.State(keeper.StateAsked) + keys, err := s.Keys() + if err != nil { + return nil, fmt.Errorf("the keeper's own %s cannot be read: %w", keeper.StateAsked, err) + } + out := map[string]keeper.Asked{} + for _, key := range keys { + if key == keeper.KeyStatus { + continue + } + e, err := s.Get(key) + if err != nil || e == nil { + continue + } + var a keeper.Asked + if json.Unmarshal(e.Value, &a) == nil { + out[key] = a + } + } + return out, nil +} + +// busStream is one stream as the monitoring endpoint answers it. +type busStream struct { + Config jetstream.StreamConfig `json:"config"` + State struct { + Messages uint64 `json:"messages"` + Bytes uint64 `json:"bytes"` + } `json:"state"` +} + +// streamsHeld reads every stream the bus holds from the monitoring endpoint's jsz, by name. +func streamsHeld(jsz []byte) (map[string]busStream, error) { + var doc struct { + Accounts []struct { + Streams []struct { + Name string `json:"name"` + busStream + } `json:"stream_detail"` + } `json:"account_details"` + } + if err := json.Unmarshal(jsz, &doc); err != nil { + return nil, fmt.Errorf("the bus's streams cannot be read: %w", err) + } + out := map[string]busStream{} + for _, a := range doc.Accounts { + for _, s := range a.Streams { + out[s.Name] = s.busStream + } + } + return out, nil +} + +// EntryView is one entry as list and show answer it: its template, its binding, each object's settings asked and held, +// its fill against its cap, and what stands of it. +type EntryView struct { + Module string `json:"module"` + Local string `json:"local"` + Storage string `json:"storage"` + Template string `json:"template,omitempty"` + Binding any `json:"binding"` + Objects []ObjectView `json:"objects"` + State keeper.CheckLine `json:"state"` +} + +// ObjectView is one bus object of an entry. +type ObjectView struct { + Stream string `json:"stream"` + Asked keeper.Settings `json:"asked"` + Held *keeper.Settings `json:"held,omitempty"` + Bytes uint64 `json:"bytes"` + Count uint64 `json:"count"` + MaxBytes int64 `json:"max_bytes"` + Differs string `json:"differs,omitempty"` + Missing bool `json:"missing,omitempty"` +} + +func viewOf(module, local string, m keeper.Made, held map[string]busStream) EntryView { + v := EntryView{Module: module, Local: local, Storage: m.Plan.Storage, Template: m.Plan.Template, Binding: m.Plan.Binding, + State: keeper.CheckLine{Module: module, Local: local, Bus: keeper.BusName(module, local), Created: m.Created, + Refused: m.Refused, Waiting: m.Waiting, Drift: m.Drift, Orphaned: m.Orphaned, Full: m.Full, Fill: m.Fill}} + for _, o := range m.Plan.Objects { + ov := ObjectView{Stream: o.Stream(), Asked: o.Settings, MaxBytes: o.Settings.MaxBytes} + if s, ok := held[o.Stream()]; ok { + h := keeper.SettingsOf(o.Kind, s.Config) + ov.Held, ov.Bytes, ov.Count, ov.MaxBytes = &h, s.State.Bytes, s.State.Messages, s.Config.MaxBytes + ov.Differs = keeper.Differs(o.Settings, h) + } else { + ov.Missing = true + } + v.Objects = append(v.Objects, ov) + } + return v +} + +// brokerTools are the four reading verbs and nats_history. +func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(module, local string, limit int) ([]keeper.Event, error)) []stdio.Tool { + both := func() (map[string]keeper.Asked, map[string]busStream, error) { + entries, err := read() + if err != nil { + return nil, nil, err + } + doc, err := jsz() + if err != nil { + return nil, nil, fmt.Errorf("the bus's monitoring endpoint cannot be read: %w", err) + } + held, err := streamsHeld(doc) + return entries, held, err + } + each := func(entries map[string]keeper.Asked, held map[string]busStream, keep func(keeper.Made) bool) []EntryView { + var out []EntryView + for _, module := range sortedNames(entries) { + for _, local := range sortedNames(entries[module].Made) { + if m := entries[module].Made[local]; keep(m) { + out = append(out, viewOf(module, local, m, held)) + } + } + } + return out + } + entryArgs := map[string]any{"module": str("the module, e.g. mesh-knowledge"), "local": str("its local name, e.g. records")} + return []stdio.Tool{ + {Name: "mesh-broker.list", + Description: "Every module's kept data on the bus that the bus module created (design 25 § 12.3): each entry's template, its bus names, its settings asked and held, and its fill against its cap. Narrowed by a module.", + Input: map[string]any{"module": str("one module")}, + Run: func(a map[string]any) (any, error) { + entries, held, err := both() + if err != nil { + return nil, err + } + if m := arg(a, "module"); m != "" { + entries = map[string]keeper.Asked{m: entries[m]} + } + out := each(entries, held, func(keeper.Made) bool { return true }) + return map[string]any{"entries": out, "count": len(out)}, nil + }}, + {Name: "mesh-broker.show", + Description: "One entry of a module's kept data on the bus: its template, its bus names, its settings asked and held, its fill against its cap, and whether it is refused, waiting, drifted, orphaned or full.", + Input: entryArgs, + Run: func(a map[string]any) (any, error) { + entries, held, err := both() + if err != nil { + return nil, err + } + module, local := arg(a, "module"), arg(a, "local") + m, ok := entries[module].Made[local] + if !ok { + return nil, fmt.Errorf("the bus module made no entry %q of %q", local, module) + } + return viewOf(module, local, m, held), nil + }}, + {Name: "mesh-broker.check", + Description: "What the bus holds against what was asked, entry by entry: each object missing, or differing from its manifest's settings and how. A difference that only grows is applied by the bus module itself; any other is a drift, raised to the operator.", + Input: map[string]any{"module": str("one module")}, + Run: func(a map[string]any) (any, error) { + entries, held, err := both() + if err != nil { + return nil, err + } + if m := arg(a, "module"); m != "" { + entries = map[string]keeper.Asked{m: entries[m]} + } + var differing []EntryView + all := each(entries, held, func(m keeper.Made) bool { return m.Created && !m.Orphaned }) + for _, v := range all { + for _, o := range v.Objects { + if o.Missing || o.Differs != "" { + differing = append(differing, v) + break + } + } + } + return map[string]any{"checked": len(all), "differing": differing}, nil + }}, + {Name: "mesh-broker.orphans", + Description: "Kept data on the bus that no accepted manifest asks for any more: each goes only by the bus module's drop, on the operator's warrant (nats_drop). Never an entry of a refused or set-aside manifest, nor a foundation part's old section.", + Run: func(map[string]any) (any, error) { + entries, held, err := both() + if err != nil { + return nil, err + } + out := each(entries, held, func(m keeper.Made) bool { return m.Orphaned }) + return map[string]any{"orphans": out, "count": len(out)}, nil + }}, + {Name: "nats_history", + Description: "Every event the bus module said of one entry of a module's kept data, in order (created, refused, drifted, drift-cleared, full, full-cleared, orphaned, recreated, dropped), from its own stream, kept 30 days.", + Input: map[string]any{"module": str("the module"), "local": str("its local name"), "limit": num("at most this many, the newest (default 100)")}, + Run: func(a map[string]any) (any, error) { + module, local := arg(a, "module"), arg(a, "local") + if module == "" || local == "" { + return nil, fmt.Errorf("history names one entry: a module and its local name") + } + events, err := history(module, local, limitOf(a, 100)) + if err != nil { + return nil, err + } + return map[string]any{"module": module, "local": local, "events": events}, nil + }}, + } +} + +// StreamHistory reads the bus module's events of one entry from its own stream, on the tools' own credential. +func StreamHistory(bus Bus) func(module, local string, limit int) ([]keeper.Event, error) { + return func(module, local string, limit int) ([]keeper.Event, error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + js, done, err := bus(ctx) + if err != nil { + return nil, err + } + defer done() + cons, err := js.OrderedConsumer(ctx, BrokerEventsStream, jetstream.OrderedConsumerConfig{}) + if err != nil { + return nil, fmt.Errorf("the bus module's stream %s cannot be read: %w", BrokerEventsStream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return nil, err + } + var out []keeper.Event + for left := info.NumPending; left > 0; { + batch, err := cons.Fetch(int(min(left, 256)), jetstream.FetchMaxWait(5*time.Second)) + if err != nil { + return nil, err + } + n := uint64(0) + for m := range batch.Messages() { + n++ + var e keeper.Event + if json.Unmarshal(m.Data(), &e) == nil && e.Module == module && e.Local == local { + out = append(out, e) + } + } + if n == 0 { + break + } + left -= min(n, left) + } + if len(out) > limit { + out = out[len(out)-limit:] + } + return out, nil + } +} + +// BrokerEventsStream is the bus module's own stream on the bus (catalogue.StreamBusName("broker-events")). +const BrokerEventsStream = "BROKER_EVENTS" + +func sortedNames[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/modules/nats/cmd/nats-tools/broker_test.go b/modules/nats/cmd/nats-tools/broker_test.go new file mode 100644 index 00000000..88c20826 --- /dev/null +++ b/modules/nats/cmd/nats-tools/broker_test.go @@ -0,0 +1,84 @@ +package main + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "nats-tools/internal/keeper" +) + +// The seat's reading verbs and nats_history (design 25 § 12.3; CAT-N7): list and show answer every entry the keeper +// made with its settings asked and held and its fill against its cap; check names what is missing or differs; orphans +// what no manifest asks for; history one entry's events in order. Each from the keeper's own state and the bus's +// monitoring endpoint. +func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) { + plan := func(local, storage, template string, options string) keeper.Plan { + e := catalogue.BusStateEntry{Name: local, Storage: storage, Template: template} + if options != "" { + _ = json.Unmarshal([]byte(options), &e.Options) + } + p, err := keeper.PlanOf("m", e) + if err != nil { + t.Fatal(err) + } + return p + } + held := map[string]keeper.Asked{"m": {Module: "m", Made: map[string]keeper.Made{ + "drafts": {Plan: plan("drafts", "key-value", "state", ""), Created: true}, + "log": {Plan: plan("log", "stream", "log", `{"max-mib":16}`), Created: true, Drift: "m_log: max-bytes asked 16777216, held 33554432"}, + "old": {Plan: plan("old", "stream", "events", ""), Created: true, Orphaned: true}, + }}} + jsz := `{"account_details":[{"stream_detail":[ + {"name":"KV_m_drafts","config":{"name":"KV_m_drafts","retention":"limits","discard":"new","max_bytes":67108864,"max_msgs_per_subject":1,"max_msg_size":262144},"state":{"messages":3,"bytes":1048576}}, + {"name":"m_log","config":{"name":"m_log","retention":"limits","discard":"new","max_bytes":33554432,"max_msg_size":262144},"state":{"messages":9,"bytes":9000}}]}]}` + tools := map[string]stdio.Tool{} + for _, tl := range brokerTools(func() (map[string]keeper.Asked, error) { return held, nil }, + func() ([]byte, error) { return []byte(jsz), nil }, + func(module, local string, limit int) ([]keeper.Event, error) { + return []keeper.Event{{Module: module, Local: local, Why: "first"}, {Module: module, Local: local, Why: "second"}}[:limit], nil + }) { + tools[tl.Name] = tl + } + for _, row := range []struct { + tool string + args map[string]any + want []string + err string + }{ + {"mesh-broker.list", nil, []string{`"count":3`, `"stream":"KV_m_drafts"`, `"bytes":1048576`, `"max_bytes":67108864`, `"bucket":"m_drafts"`}, ""}, + {"mesh-broker.show", map[string]any{"module": "m", "local": "log"}, []string{`"stream":"m_log"`, `"differs":"max-bytes asked 16777216, held 33554432"`, `"drift":"m_log`}, ""}, + {"mesh-broker.show", map[string]any{"module": "m", "local": "none"}, nil, `made no entry "none" of "m"`}, + {"mesh-broker.check", nil, []string{`"checked":2`, `"local":"log"`}, ""}, + {"mesh-broker.orphans", nil, []string{`"count":1`, `"local":"old"`, `"orphaned":true`}, ""}, + {"nats_history", map[string]any{"module": "m", "local": "log", "limit": float64(1)}, []string{`"why":"first"`}, ""}, + {"nats_history", map[string]any{"module": "m"}, nil, "names one entry"}, + } { + t.Run(fmt.Sprintf("%s %v", row.tool, row.args), func(t *testing.T) { + args := row.args + if args == nil { + args = map[string]any{} + } + out, err := tools[row.tool].Run(args) + if row.err != "" { + if err == nil || !strings.Contains(err.Error(), row.err) { + t.Fatalf("answered %v (%v): refused, naming %q", out, err, row.err) + } + return + } + raw, _ := json.Marshal(out) + for _, w := range row.want { + if !strings.Contains(string(raw), w) { + t.Errorf("%s lacks %s", raw, w) + } + } + if row.tool == "mesh-broker.check" && strings.Contains(string(raw), `"local":"drafts"`) { + t.Errorf("check names drafts, which the bus holds as asked: %s", raw) + } + }) + } +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 7ac873a1..7945dfcc 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -156,7 +156,7 @@ func main() { keepMain() return } - if err := stdio.Serve("", append(append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...), expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...)); err != nil { + if err := stdio.Serve("", allTools()); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } @@ -170,3 +170,16 @@ var ReadingVerbs = []string{"list", "show", "check", "orphans"} // Events are the bus module's own events (ADR 0595 §6, ADR 0620 §2), on its own stream broker-events (BROKER_EVENTS). var Events = []string{"created", "refused", "drifted", "drift-cleared", "full", "full-cleared", "orphaned", "recreated", "dropped"} + +// allTools is every tool and seat verb this bundle serves. +func allTools() []stdio.Tool { + monitor := LiveMonitor() + jsz := func() ([]byte, error) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + return monitor(ctx, "/jsz?accounts=1&streams=1&config=1") + } + out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...) + out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) + return append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) +} diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index 30753464..08067236 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -223,6 +223,20 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { if h, _ := server["health"].(map[string]any); h["kind"] != "tcp" || h["endpoint"] != "bus" { t.Errorf("the server's health is %v: tcp on bus, the check the keeper's tool check stands beside", server["health"]) } + served := map[string]bool{} + for _, tl := range allTools() { + served[tl.Name] = true + } + for _, name := range m.Tools { + if !served[name] { + t.Errorf("the manifest lists the tool %s, which the bundle does not serve", name) + } + } + for _, verb := range ReadingVerbs { + if !served["mesh-broker."+verb] { + t.Errorf("the bundle does not serve the seat verb mesh-broker.%s", verb) + } + } args, _ := server["args"].([]any) if len(args) != 2 || args[0] != "sh" || args[1] != "/opt/mesh-entrypoint/entrypoint.sh" { t.Errorf("args are %v: the entrypoint run through sh, so no execute bit is relied on", args) diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index 9f13e6d7..53e960db 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/modules/nats/internal/keeper/jetstream.go @@ -31,11 +31,11 @@ func (b JetStream) Held(ctx context.Context, o Object) (Settings, Fill, bool, er if err != nil { return Settings{}, Fill{}, false, retryable(err) } - return settingsOf(o.Kind, info.Config), Fill{Bytes: info.State.Bytes, Messages: info.State.Msgs}, true, nil + return SettingsOf(o.Kind, info.Config), Fill{Bytes: info.State.Bytes, Messages: info.State.Msgs}, true, nil } -// settingsOf reads a stream's configuration as the settings an object of kind is asked with. -func settingsOf(kind string, c jetstream.StreamConfig) Settings { +// SettingsOf reads a stream's configuration as the settings an object of kind is asked with. +func SettingsOf(kind string, c jetstream.StreamConfig) Settings { s := Settings{MaxBytes: c.MaxBytes, MaxAgeSeconds: int64(c.MaxAge / time.Second), Discard: DiscardOld, WorkQueue: c.Retention == jetstream.WorkQueuePolicy, DuplicatesSeconds: int64(c.Duplicates / time.Second)} if c.Discard == jetstream.DiscardNew { diff --git a/modules/nats/module.json b/modules/nats/module.json index 413fe33c..a12bb1c6 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -114,7 +114,8 @@ "nats_consumer_remove", "nats_bucket_remove", "nats_stream_remove", - "nats_bucket_expiry" + "nats_bucket_expiry", + "nats_history" ], "guards": [ 8222