From b9a224dc99348203c944d50af8cd042505ebbbef Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 12:05:11 +0100 Subject: [PATCH] NATS2: the bus module plans each sender's sent store, 256 MiB, 7 days, refusing when full MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The store is made by Keeper.Sends, kept when a verdict asks nothing of it, and a full one says `full`. Which module sends is NATS3's input (the catalog half); this makes the store. Builds: design 47 G §59, §137, §344, §477 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/internal/keeper/jetstream.go | 2 +- modules/nats/internal/keeper/keeper.go | 21 +++++++++++++++- modules/nats/internal/keeper/keeper_test.go | 27 +++++++++++++++++++++ modules/nats/internal/keeper/plan.go | 23 ++++++++++++++++++ 4 files changed, 71 insertions(+), 2 deletions(-) diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index 53e960db9..31e4ab8d8 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/modules/nats/internal/keeper/jetstream.go @@ -74,7 +74,7 @@ func (b JetStream) Create(ctx context.Context, o Object) error { LimitMarkerTTL: limitMarker(o.Settings.KeyTTL)}) case KindObjects: _, err = b.JS.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: o.Name, MaxBytes: o.Settings.MaxBytes, - Storage: jetstream.FileStorage, Replicas: 1}) + TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, Storage: jetstream.FileStorage, Replicas: 1}) default: _, err = b.JS.CreateStream(ctx, streamConfig(o)) } diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 962ae0ea4..073671dd4 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -392,7 +392,8 @@ func (k *Keeper) Heard(a Accepted) { } for _, local := range sortedKeys(held.Made) { m := held.Made[local] - if asked[local] || m.Orphaned { + if asked[local] || m.Orphaned || local == SentLocal { + // The sent store is no manifest's ask: it is made by Sends and goes only by drop (§59, §344). continue } if !m.Created { @@ -411,6 +412,24 @@ func (k *Keeper) Heard(a Accepted) { k.write(held) } +// Sends makes module's sent store where the bus lacks it (§59, §344): called when the catalog half and the seat list +// say the module serves a verb or emits an event, and again on the change that gives a holder its first verb or event. +// A module that no longer sends keeps its store: an object in it expires by age, and the store goes only by drop. +func (k *Keeper) Sends(module string) { + held, _, found := k.asked(module) + if !found { + held = Asked{V: 1, Module: module, At: k.now()} + } + if held.Made == nil { + held.Made = map[string]Made{} + } + m := held.Made[SentLocal] + m.Plan, m.Orphaned = SentPlanOf(module), false + k.ensure(&held, &m) + held.Made[SentLocal] = m + k.write(held) +} + // makeEntry makes one asked entry where the bus lacks it, compares it where it holds it, and says what it did. func (k *Keeper) makeEntry(held *Asked, e catalogue.BusStateEntry) { module := held.Module diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 7c7f3735a..735ed9a4d 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -254,6 +254,8 @@ func TestEachTemplateIsPlannedUnderTheModulesNames(t *testing.T) { binding: "stream mesh-knowledge_reviews on mesh.bus.mesh-knowledge.reviews"}, {name: "a local name holding the separator", entry: entry("my_drafts", catalogue.StorageKeyValue, catalogue.TemplateState, nil), refused: `the local name "my_drafts" is empty or holds "_"`}, + {name: "a local name that is the sent store's", entry: entry("sent-objects", catalogue.StorageObjectStore, catalogue.TemplateObjects, map[string]string{"max-mib": "1"}), + refused: "the bus module's own for the sent store mesh-knowledge_sent-objects"}, {name: "objects with no cap", entry: entry("files", catalogue.StorageObjectStore, catalogue.TemplateObjects, nil), refused: "gives no max-mib"}, {name: "a bare stream with no when-full", entry: entry("raw", catalogue.StorageStream, "", nil), @@ -347,6 +349,31 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { } } }}, + {"a sender gets its sent store: 256 MiB, 7 days, refusing when full, and it is kept when a verdict asks nothing of it", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + 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 { + 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 { + t.Errorf("the sent store is %+v, want 256 MiB, 7 days, refusing when full", s) + } + if want := "[created m sent-objects created m drafts]"; fmt.Sprint(out.events) != want { + t.Errorf("said %v, want %s", out.events, want) + } + if b := out.bodies[0].Binding; b.ObjectStore != "m_sent-objects" { + t.Errorf("binding %+v names no sent store", b) + } + bus.fill["OBJ_m_sent-objects"] = 200 * uint64(MiB) + k.FillPass() + if got := out.events[len(out.events)-1]; got != "full m sent-objects" { + t.Errorf("a full sent store said %q, want full", got) + } + if held := k.Held()["m"]; held.Made[SentLocal].Orphaned || len(bus.deleted) != 0 { + t.Errorf("the sent store was orphaned or removed: %+v %v", held.Made[SentLocal], bus.deleted) + } + }}, {"a replay not newer is not applied", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { k.Heard(Accepted{Module: "m", Digest: "d2", Revision: 9, BusState: []catalogue.BusStateEntry{drafts}}) k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 4, BusState: nil}) diff --git a/modules/nats/internal/keeper/plan.go b/modules/nats/internal/keeper/plan.go index 419331328..080d61755 100644 --- a/modules/nats/internal/keeper/plan.go +++ b/modules/nats/internal/keeper/plan.go @@ -40,6 +40,14 @@ const ( EventsDays = 7 ) +// The sent store of a module that sends (design 47 G §59, §477): the local name its bus name `_sent-objects` is made +// of, its cap and the age after which the bus expires an object in it. +const ( + SentLocal = "sent-objects" + SentCapMiB = 256 + SentAgeDays = 7 +) + // Settings are one bus object's settings, as asked and as held: what check compares. type Settings struct { // MaxBytes is its cap in bytes; MaxAgeSeconds its age, 0 for none. @@ -109,6 +117,9 @@ func StreamSubject(module, local string) string { return "mesh.bus." + module + // type and template that do not go together, an option of the wrong shape. The judge refuses each of these first // (ADR 0595 §8); this is the keeper's own guard, so a manifest judged by an older judge is refused, never half-made. func PlanOf(module string, e catalogue.BusStateEntry) (Plan, error) { + if e.Name == SentLocal { + return Plan{}, fmt.Errorf("the local name %q is the bus module's own for the sent store %s (design 47 G §59)", e.Name, BusName(module, SentLocal)) + } if e.Name == "" || strings.Contains(e.Name, Separator) || strings.ContainsAny(e.Name, ".*> ") { return Plan{}, fmt.Errorf("the local name %q is empty or holds %q, a dot, a wildcard or a space: the mesh names it %s on the bus, and two modules' names must never meet", e.Name, Separator, BusName(module, "")) } @@ -242,6 +253,18 @@ func PlanOf(module string, e catalogue.BusStateEntry) (Plan, error) { return p, nil } +// SentPlanOf is the plan of module's sent store (§59): an object store `_sent-objects` of 256 MiB (§477) that +// refuses when full and keeps an object 7 days, after which the bus expires it. It is transport and not kept data, so +// its expiry is no removal that needs a person's word. It is made with no manifest field (§344). +func SentPlanOf(module string) Plan { + name := BusName(module, SentLocal) + e := catalogue.BusStateEntry{Name: SentLocal, Storage: catalogue.StorageObjectStore, Template: catalogue.TemplateObjects, + Options: map[string]json.RawMessage{catalogue.OptionCapMiB: json.RawMessage(strconv.Itoa(SentCapMiB))}} + return Plan{Module: module, Local: SentLocal, Storage: e.Storage, Template: e.Template, Entry: e, + Binding: membership.BusBinding{Local: SentLocal, Storage: e.Storage, Template: e.Template, ObjectStore: name}, + Objects: []Object{{Kind: KindObjects, Name: name, Settings: Settings{MaxBytes: SentCapMiB * MiB, MaxAgeSeconds: days(SentAgeDays), Discard: DiscardNew}}}} +} + func stateSettings(o options) (Settings, error) { history, err := o.int(catalogue.OptionPastValues, 1) if err != nil {