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

CAT-N7: the seat mesh-broker's reading verbs answer the seat's contract

Browse Source 
list, show and orphans answer meshbroker.ListAnswer, Held and OrphansAnswer; check answers CheckAnswer with each
differing entry's settings asked and held by stream and whether the difference only grows; show takes the contract's
local-name. Built against mesh-contracts' seats/meshbroker module in the workspace; the group pin adds its require.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 13:10:43 +01:00
parent d7214bb576
commit 45688f196e
2 changed files with 89 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/broker.go
+78 -58
View File
@@ -10,6 +10,8 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
"nats-tools/internal/keeper"
)
@@ -76,46 +78,64 @@ func streamsHeld(jsz []byte) (map[string]busStream, error) {
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}}
// heldOf is one entry as the seat's list, show and orphans answer it (mesh-contracts meshbroker.Held): its bus name, its
// template, its options or configuration as the manifest asked them, the settings the bus holds of each of its objects
// by stream, and its fill summed over them against their caps.
func heldOf(module, local string, m keeper.Made, held map[string]busStream) meshbroker.Held {
h := meshbroker.Held{BusName: keeper.BusName(module, local), Module: module, LocalName: local, StorageType: m.Plan.Storage,
Template: m.Plan.Template, Options: anyOf(m.Plan.Entry.Options), Configuration: anyOf(m.Plan.Entry.Configuration)}
settings := map[string]any{}
var fill meshbroker.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
s, ok := held[o.Stream()]
if !ok {
continue
}
v.Objects = append(v.Objects, ov)
settings[o.Stream()] = keeper.SettingsOf(o.Kind, s.Config)
fill.Bytes += int64(s.State.Bytes)
fill.Count += int64(s.State.Messages)
fill.Cap += s.Config.MaxBytes
}
return v
if len(settings) > 0 {
h.Settings, h.Fill = settings, &fill
}
return h
}
// differencesOf are an entry's objects the bus holds otherwise than asked, or not at all, as the seat's check answers
// them: each object's settings asked and held by stream, and whether the difference only grows (apply may act on it).
func differencesOf(module, local string, m keeper.Made, held map[string]busStream) (meshbroker.CheckAnswerDifferencesItem, bool) {
d := meshbroker.CheckAnswerDifferencesItem{Module: module, LocalName: local, Asked: map[string]any{}, Held: map[string]any{}, Grows: true}
differs := false
for _, o := range m.Plan.Objects {
s, ok := held[o.Stream()]
if !ok {
d.Asked[o.Stream()], d.Grows, differs = o.Settings, false, true
continue
}
h := keeper.SettingsOf(o.Kind, s.Config)
if keeper.Differs(o.Settings, h) == "" {
continue
}
d.Asked[o.Stream()], d.Held[o.Stream()], differs = o.Settings, h, true
d.Grows = d.Grows && keeper.Grows(o.Settings, h)
}
return d, differs
}
// anyOf is a manifest's options or configuration as the contract carries them.
func anyOf(m map[string]json.RawMessage) map[string]any {
if len(m) == 0 {
return nil
}
out := map[string]any{}
for k, v := range m {
var x any
if json.Unmarshal(v, &x) == nil {
out[k] = x
}
}
return out
}
// brokerTools are the four reading verbs and nats_history.
@@ -132,21 +152,21 @@ func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(modu
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
each := func(entries map[string]keeper.Asked, held map[string]busStream, keep func(keeper.Made) bool) []meshbroker.Held {
out := []meshbroker.Held{}
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))
out = append(out, heldOf(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")}
entryArgs := map[string]any{"module": str("the module, e.g. mesh-knowledge"), "local-name": 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.",
Description: "Every module's kept data on the bus that the bus module created (design 25 § 12.3): each entry's bus name, template and options as asked, the settings the bus holds of each of its objects, 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()
@@ -156,26 +176,25 @@ func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(modu
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
return meshbroker.ListAnswer{Entries: each(entries, held, func(keeper.Made) bool { return true })}, 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.",
Description: "One entry of a module's kept data on the bus: its bus name, template and options as asked, the settings the bus holds of each of its objects, and its fill against its cap. Whether it is refused, waiting, drifted, full or orphaned is what nats_history says of it.",
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")
module, local := arg(a, "module"), arg(a, "local-name")
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
return heldOf(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.",
Description: "What the bus holds against what was asked, entry by entry: each object missing, or differing from its manifest's settings, with both. 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()
@@ -185,17 +204,19 @@ func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(modu
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
out := meshbroker.CheckAnswer{Differences: []meshbroker.CheckAnswerDifferencesItem{}}
for _, module := range sortedNames(entries) {
for _, local := range sortedNames(entries[module].Made) {
m := entries[module].Made[local]
if !m.Created || m.Orphaned {
continue
}
if d, differs := differencesOf(module, local, m, held); differs {
out.Differences = append(out.Differences, d)
}
}
}
return map[string]any{"checked": len(all), "differing": differing}, nil
return out, 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.",
@@ -204,8 +225,7 @@ func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(modu
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
return meshbroker.OrphansAnswer{Entries: each(entries, held, func(m keeper.Made) bool { return m.Orphaned })}, 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.",
modules/nats/cmd/nats-tools/broker_test.go
+11 -6
View File
@@ -50,11 +50,16 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) {
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`}, ""},
// Each answers the seat's contract (mesh-contracts meshbroker ListAnswer, Held, CheckAnswer, OrphansAnswer).
{"mesh-broker.list", nil, []string{`{"entries":[`, `"bus-name":"m_drafts"`, `"local-name":"drafts"`, `"storage-type":"key-value"`,
`"KV_m_drafts":{"max-bytes":67108864`, `"fill":{"bytes":1048576,"cap":67108864,"count":3}`, `"options":{"max-mib":16}`}, ""},
{"mesh-broker.show", map[string]any{"module": "m", "local-name": "log"}, []string{`"bus-name":"m_log"`, `"template":"log"`,
`"m_log":{"max-bytes":33554432`}, ""},
{"mesh-broker.show", map[string]any{"module": "m", "local-name": "none"}, nil, `made no entry "none" of "m"`},
{"mesh-broker.check", nil, []string{`{"differences":[{`, `"local-name":"log"`, `"asked":{"m_log":{"max-bytes":16777216`,
`"held":{"m_log":{"max-bytes":33554432`}, ""},
{"mesh-broker.check", map[string]any{"module": "x"}, []string{`{"differences":[]}`}, ""},
{"mesh-broker.orphans", nil, []string{`{"entries":[{`, `"local-name":"old"`}, ""},
{"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"},
} {
@@ -76,7 +81,7 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) {
t.Errorf("%s lacks %s", raw, w)
}
}
if row.tool == "mesh-broker.check" && strings.Contains(string(raw), `"local":"drafts"`) {
if row.tool == "mesh-broker.check" && strings.Contains(string(raw), `"local-name":"drafts"`) {
t.Errorf("check names drafts, which the bus holds as asked: %s", raw)
}
})