diff --git a/modules/nats/cmd/nats-tools/broker_test.go b/modules/nats/cmd/nats-tools/broker_test.go index 33a6cd298..6042b3af3 100644 --- a/modules/nats/cmd/nats-tools/broker_test.go +++ b/modules/nats/cmd/nats-tools/broker_test.go @@ -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"`}, ""}, diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 4139151fd..a3e6c4a20 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -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 `-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 { diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 360c6f5d8..298f57322 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -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) } diff --git a/modules/nats/internal/keeper/plan.go b/modules/nats/internal/keeper/plan.go index ad89ae4e9..baa6a3e88 100644 --- a/modules/nats/internal/keeper/plan.go +++ b/modules/nats/internal/keeper/plan.go @@ -49,6 +49,10 @@ const ( SentAgeDays = 7 ) +// StateObjectsCapMiB is the cap of a state's object store `-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 `-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 {