NATS1: the bus module plans each state's <state>-objects store (16 MiB, objects-max-mib) and says its binding

Builds: design 47 G §58, §91, §513, §591, §607, §549

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 04:55:47 +01:00
parent 500c8c8cec
commit 0699741fab
4 changed files with 54 additions and 17 deletions
+3 -2
View File
@@ -37,6 +37,7 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) {
}}}
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":"OBJ_m_drafts-objects","config":{"name":"OBJ_m_drafts-objects","retention":"limits","discard":"new","max_bytes":16777216},"state":{"messages":0,"bytes":0}},
{"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 },
@@ -54,9 +55,9 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) {
}{
// 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}`,
`"KV_m_drafts":{"max-bytes":67108864`, `"fill":{"bytes":1048576,"cap":83886080,"count":3}`, `"options":{"max-mib":16}`,
// The binding of an entry the bus holds is the one its created carries (design 25 bindings-first-view §2).
`"binding":{"bucket":"m_drafts","local":"drafts","storage":"key-value","template":"state"}`}, ""},
`"binding":{"bucket":"m_drafts","local":"drafts","objects":"m_drafts-objects","storage":"key-value","template":"state"}`}, ""},
{"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": "later"}, []string{`"local-name":"later"`}, ""},
+5 -1
View File
@@ -488,6 +488,9 @@ func (k *Keeper) ensure(held *Asked, m *Made) {
defer cancel()
asked, holds := map[string]Settings{}, map[string]Settings{}
var drifts []string
// madeLater is an object newly made for an entry that was made whole before (a state's `<state>-objects`, design 47 G
// §58): its binding gained a name, so `created` is said again.
madeLater := false
for _, o := range m.Plan.Objects {
asked[o.Name] = o.Settings
s, _, found, err := k.Bus.Held(ctx, o)
@@ -509,6 +512,7 @@ func (k *Keeper) ensure(held *Asked, m *Made) {
return
}
holds[o.Name] = o.Settings
madeLater = madeLater || m.Created
continue
}
holds[o.Name] = s
@@ -530,7 +534,7 @@ func (k *Keeper) ensure(held *Asked, m *Made) {
if wasRefused {
k.clear(conditionRefused, held.Module, m.Plan.Local)
}
if !m.Created {
if !m.Created || madeLater {
// Marked created only once `created` is said whole: the grants part binds on it, so one the bus did not take is
// said again on the backoff (Waiting), never marked made and forgotten.
if err := k.sayNow(EventCreated, m.Plan, held.Revision, "", asked, holds); err != nil {
+29 -12
View File
@@ -328,8 +328,13 @@ func TestEachTemplateIsPlannedUnderTheModulesNames(t *testing.T) {
refused string
}{
{name: "state", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"history": "5"}),
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 5, new, size 262144}",
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 5, new, size 262144}; object-store mesh-knowledge_drafts-objects {16 MiB, age 0, history 0, new, size 0}",
binding: "bucket mesh-knowledge_drafts"},
{name: "state naming objects-max-mib (§591)", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"objects-max-mib": "512"}),
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 1, new, size 262144}; object-store mesh-knowledge_drafts-objects {512 MiB, age 0, history 0, new, size 0}",
binding: "bucket mesh-knowledge_drafts"},
{name: "state naming objects-max-mib 0", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"objects-max-mib": "0"}),
refused: "not a size above 0"},
{name: "events, default age and cap", entry: entry("events", catalogue.StorageStream, catalogue.TemplateEvents, nil),
objects: "stream mesh-knowledge_events [mesh.bus.mesh-knowledge.events.>] {1024 MiB, age 604800, history 0, old, size 0}",
binding: "stream mesh-knowledge_events on mesh.bus.mesh-knowledge.events"},
@@ -434,7 +439,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
}
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
if want := "[KV_m_drafts m_log]"; fmt.Sprint(bus.created) != want {
if want := "[KV_m_drafts OBJ_m_drafts-objects m_log]"; fmt.Sprint(bus.created) != want {
t.Errorf("created %v, want %s", bus.created, want)
}
if want := "[created m drafts created m log]"; fmt.Sprint(out.events) != want {
@@ -455,7 +460,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
k.Sends("m")
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts}})
k.Sends("m")
if want := "[OBJ_m_sent-objects KV_m_drafts]"; fmt.Sprint(bus.created) != want {
if want := "[OBJ_m_sent-objects KV_m_drafts OBJ_m_drafts-objects]"; fmt.Sprint(bus.created) != want {
t.Fatalf("created %v, want %s", bus.created, want)
}
if s := bus.held["OBJ_m_sent-objects"]; s.MaxBytes != 256*MiB || s.MaxAgeSeconds != 7*24*3600 || s.Discard != DiscardNew {
@@ -543,10 +548,22 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
t.Errorf("said %v %v: cleared once asked as held", out.events, out.conds)
}
}},
{"a bus that holds an entry the keeper never made is taken as made", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
{"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
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
out.reset()
bus.created = nil
delete(bus.held, "OBJ_m_drafts-objects")
k.CheckAll()
if fmt.Sprint(bus.created) != "[OBJ_m_drafts-objects]" || fmt.Sprint(out.events) != "[created m drafts]" || out.bodies[0].Binding.Objects != "m_drafts-objects" {
t.Errorf("created %v, said %v: the missing store made, its binding said again", bus.created, out.events)
}
}},
{"a bus that holds an entry the keeper never made is taken as made", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
p, _ := PlanOf("m", drafts)
bus.held["KV_m_drafts"], bus.held["OBJ_m_drafts-objects"] = p.Objects[0].Settings, p.Objects[1].Settings
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
if len(bus.created) != 0 || fmt.Sprint(out.events) != "[created m drafts]" {
t.Errorf("created %v, said %v: its binding said, nothing made twice", bus.created, out.events)
}
@@ -696,7 +713,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
bus.snaps = nil
out.reset()
got := put(Request{ID: "r2", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo", Why: "unused"})
if got.Error != "" || fmt.Sprint(bus.snaps) != "[[KV_m_drafts]]" || fmt.Sprint(bus.deleted) != "[KV_m_drafts]" ||
if got.Error != "" || fmt.Sprint(bus.snaps) != "[[KV_m_drafts OBJ_m_drafts-objects]]" || fmt.Sprint(bus.deleted) != "[KV_m_drafts OBJ_m_drafts-objects]" ||
fmt.Sprint(out.events) != "[dropped m drafts]" || len(acts) != 1 || !strings.Contains(acts[0], "nats_drop: dropped drafts of m") ||
!strings.Contains(acts[0], "person.jo") {
t.Errorf("ended %+v, snapshots %v, deleted %v, said %v, recorded %v", got, bus.snaps, bus.deleted, out.events, acts)
@@ -733,7 +750,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
out.reset()
k.ActOnRequests()
got := get("r10")
if got.Path != "/snapshots/KV_m_drafts.tar" || got.Size != 100 || !strings.HasPrefix(got.Digest, "sha256:") || !got.Protected || got.Error != "" {
if got.Path != "/snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar" || got.Size != 100 || !strings.HasPrefix(got.Digest, "sha256:") || !got.Protected || got.Error != "" {
t.Fatalf("the drop's record: %+v", got)
}
findings := k.Finding.(*findingLog)
@@ -753,7 +770,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
k.Now = func() time.Time { return base.AddDate(0, 0, 91) }
findings.calls = nil
k.ExpireSnapshots()
if _, _, found, _ := store.Get(StateRequests, "r10"); found || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar /snapshots/KV_m_drafts.tar]" ||
if _, _, found, _ := store.Get(StateRequests, "r10"); found || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar /snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar]" ||
len(journal) != 1 || !strings.Contains(journal[0], "past its 90 days") || fmt.Sprint(findings.calls) != "[clear module.nats.snapshots.snapshots-filling]" {
t.Errorf("after 91 days: removed %v, journal %v, finding calls %v", files.removed, journal, findings.calls)
}
@@ -767,7 +784,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
{"a snapshot is restored in place, verified, refused while its names are live, cancelled, and its failure removes the partial stream", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store, files := k.Store.(*memStore), k.Files.(*memFiles)
out.sig = make(chan struct{}, 8)
path := "/snapshots/KV_m_drafts.tar"
path := "/snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar"
bus.snapStreams = map[string][]string{path: {"KV_m_drafts"}}
files.missing = map[string]bool{"/snapshots/gone.tar": true}
digest := "sha256:" + hex.EncodeToString([]byte(path))
@@ -942,13 +959,13 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
_ = json.Unmarshal(raw, &got)
return got
}
if got := put("r5"); got.Error != "" || len(bus.deleted) != 1 {
if got := put("r5"); got.Error != "" || len(bus.deleted) != 2 {
t.Fatalf("the first drop: %+v, deleted %v", got, bus.deleted)
}
// The entry is asked for and orphaned again, so a second drop would remove it: the spent warrant refuses it.
k.Heard(Accepted{Module: "m", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 4, BusState: []catalogue.BusStateEntry{log}})
if got := put("r6"); !strings.Contains(got.Error, "already acted (drop of m's drafts, request r5") || len(bus.deleted) != 1 {
if got := put("r6"); !strings.Contains(got.Error, "already acted (drop of m's drafts, request r5") || len(bus.deleted) != 2 {
t.Errorf("the second request on the same warrant: %+v, deleted %v", got, bus.deleted)
}
if raw, _, found, _ := store.Get(StateSpent, "bus-5"); !found || !strings.Contains(string(raw), `"request":"r5"`) {
@@ -960,7 +977,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
for _, key := range keys {
_ = store.Delete(StateRequests, key)
}
if got := put("r6b"); !strings.Contains(got.Error, "already acted") || len(bus.deleted) != 1 {
if got := put("r6b"); !strings.Contains(got.Error, "already acted") || len(bus.deleted) != 2 {
t.Errorf("with requests wiped, the spent warrant: %+v, deleted %v", got, bus.deleted)
}
}},
@@ -1080,7 +1097,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
raw, _, _, _ = store.Get(StateRequests, "r9")
var got Request
_ = json.Unmarshal(raw, &got)
if got.Outcome != "the first view answered at revision 8" || fmt.Sprint(bus.created) != "[KV_m_drafts]" ||
if got.Outcome != "the first view answered at revision 8" || fmt.Sprint(bus.created) != "[KV_m_drafts OBJ_m_drafts-objects]" ||
len(acts) != 1 || !strings.Contains(acts[0], "nats_first_view_again: asked mesh-delivery-indexer.indexer_bus_state again by hand") {
t.Errorf("ended %+v, created %v, recorded %v", got, bus.created, acts)
}
+17 -2
View File
@@ -49,6 +49,10 @@ const (
SentAgeDays = 7
)
// StateObjectsCapMiB is the cap of a state's object store `<state>-objects` unless the state names `objects-max-mib`
// (design 47 G §591, §547).
const StateObjectsCapMiB = 16
// Sends says whether a module sends (design 47 G §344): it claims a seat whose definition has a verb or an event, or it
// declares an event of its own (emitsOwn). claims are the seats its catalog graph entry claims, defs the seat list's
// definitions by seat; a claimed seat the list does not hold counts for nothing. A definition that includes parts takes
@@ -156,8 +160,19 @@ func PlanOf(module string, e catalogue.BusStateEntry) (Plan, error) {
if err != nil {
return Plan{}, err
}
p.Binding.Bucket = name
p.Objects = []Object{{Kind: KindBucket, Name: name, Settings: s}}
// Beside each state its object store `<state>-objects`, which the grants part grants to the state's owner alone (design
// 47 G §58, §591): 16 MiB unless the state's requires names `objects-max-mib`.
objCap, err := opts.int(catalogue.OptionObjectsCapMiB, StateObjectsCapMiB)
if err != nil {
return Plan{}, err
}
if objCap <= 0 {
return Plan{}, fmt.Errorf("the option %s of %q is %d, not a size above 0", catalogue.OptionObjectsCapMiB, e.Name, objCap)
}
objects := BusName(module, catalogue.StateObjectsName(e.Name))
p.Binding.Bucket, p.Binding.Objects = name, objects
p.Objects = []Object{{Kind: KindBucket, Name: name, Settings: s},
{Kind: KindObjects, Name: objects, Settings: Settings{MaxBytes: objCap * MiB, Discard: DiscardNew}}}
case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateEvents:
age, err := opts.int(catalogue.OptionAgeDays, EventsDays)
if err != nil {