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

NATS2: the bus module plans each sender's sent store, 256 MiB, 7 days, refusing when full

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-11 12:05:11 +01:00
parent e6aab55008
commit b9a224dc99
4 changed files with 71 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/internal/keeper/jetstream.go
+1 -1
View File
@@ -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))
}
modules/nats/internal/keeper/keeper.go
+20 -1
View File
@@ -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
modules/nats/internal/keeper/keeper_test.go
+27
View File
@@ -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})
modules/nats/internal/keeper/plan.go
+23
View File
@@ -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 `<module>_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, "<local>"))
}
@@ -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 `<module>_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 {