From d7214bb576f05499c65ca51757dca037fac4f69c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:58:44 +0100 Subject: [PATCH] CAT-N7: the keeper says each of mesh-broker's events on the seat's subject, and keeps it on its own stream for history The lead's ruling of 13:0x (registry milestone): mesh-broker's events are mesh.seat.mesh-broker.event. on the shared events stream like every seat's, and the bus module's own 30-day stream keeps each for its history tool. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- modules/nats/cmd/nats-tools/keep.go | 42 +++++++++++++------ .../cmd/nats-tools/keeper_integration_test.go | 17 ++++++-- modules/nats/cmd/nats-tools/main.go | 3 +- modules/nats/internal/keeper/keeper.go | 2 +- 4 files changed, 46 insertions(+), 18 deletions(-) diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 3ecaf53c..9288bb08 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -20,9 +20,11 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "nats-tools/internal/bussnap" "nats-tools/internal/keeper" @@ -74,28 +76,44 @@ func (s processStore) Delete(state, key string) error { return s.c.State(state). func (s processStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() } -// streamEmitter says an event on the bus module's own stream, with the SDK's headers. +// streamEmitter says an event of the seat mesh-broker, with the SDK's headers, twice under one event id (the lead's ruling, +// registry milestone, 13:0x; design 25 § 12.3): on the seat's own subject `mesh.seat.mesh-broker.event.`, which the +// shared events stream keeps 7 days and every consumer of `mesh-broker.` hears, like any seat's event; then on the +// bus module's own stream (its old `streams` section, 30 days, 256 MiB), which its tool `history` reads. The seat's subject +// first: a reader hears it before the history holds it, never the other way round. func streamEmitter(js jetstream.JetStream, a process.Account) func(string, keeper.Event) error { return func(event string, body keeper.Event) error { data, err := json.Marshal(body) if err != nil { return err } - msg := nats.NewMsg(a.StreamSubject(event)) id := eventID() - msg.Header.Set("x-event-id", id) - msg.Header.Set("x-source", a.Module) - msg.Header.Set("x-node", a.Node) - msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) - msg.Header.Set("content-type", "application/json") - msg.Data = data - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) - return err + at := time.Now().UTC().Format(time.RFC3339Nano) + for _, subject := range EventSubjects(a, event) { + msg := nats.NewMsg(subject) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", a.Module) + msg.Header.Set("x-node", a.Node) + msg.Header.Set("x-time", at) + msg.Header.Set("content-type", "application/json") + msg.Data = data + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) + cancel() + if err != nil { + return fmt.Errorf("saying %s on %s: %w", event, subject, err) + } + } + return nil } } +// EventSubjects are where one event of the seat mesh-broker is said, in order: the seat's subject, then the bus module's +// own stream. +func EventSubjects(a process.Account, event string) []string { + return []string{seatdef.EventSubject(catalogue.BrokerSeat, event), a.StreamSubject(event)} +} + func eventID() string { var b [16]byte _, _ = rand.Read(b[:]) diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index 08a92477..da61f2e5 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "os" + "slices" "strings" "sync" "testing" @@ -114,7 +115,7 @@ func TestTheBusModuleRunsWhole(t *testing.T) { // What the grants part creates for the bus module from its manifest's old sections: EVENTS and the keeper's // consumer on it, its own stream, and its two states. for _, c := range []jetstream.StreamConfig{ - {Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}, + {Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}, {Name: BrokerEventsStream, Subjects: []string{"mesh.stream.nats.event.>"}, MaxAge: 30 * 24 * time.Hour, MaxBytes: 256 << 20}, {Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}}, } { @@ -177,8 +178,10 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Fatal(err) } } - said := func() []string { - cons, err := js.OrderedConsumer(ctx, BrokerEventsStream, jetstream.OrderedConsumerConfig{}) + // saidOn reads what one stream holds of the bus module's events, by its subjects' prefix: its own stream (its + // history), or EVENTS on the seat mesh-broker's subjects (what every consumer of mesh-broker. hears). + saidOn := func(stream, prefix string) []string { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{prefix + ">"}}) if err != nil { t.Fatal(err) } @@ -191,10 +194,11 @@ func TestTheBusModuleRunsWhole(t *testing.T) { for m := range batch.Messages() { var e keeper.Event _ = json.Unmarshal(m.Data(), &e) - out = append(out, strings.TrimPrefix(m.Subject(), "mesh.stream.nats.event.")+" "+e.Module+" "+e.Local) + out = append(out, strings.TrimPrefix(m.Subject(), prefix)+" "+e.Module+" "+e.Local) } return out } + said := func() []string { return saidOn(BrokerEventsStream, "mesh.stream.nats.event.") } saidHas := func(want string) func() bool { return func() bool { for _, l := range said() { @@ -221,6 +225,11 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Run("a verdict, then created, with its binding", func(t *testing.T) { accepted("m", 2, state, log) eventually(t, "created m drafts and m log", func() bool { return saidHas("created m drafts")() && saidHas("created m log")() }) + // Said on the seat's subject as well, on the shared events stream (the lead's ruling, 13:0x). + if seat := saidOn("EVENTS", "mesh.seat.mesh-broker.event."); !slices.Contains(seat, "created m drafts") || + !slices.Contains(seat, "created m log") { + t.Errorf("EVENTS holds %v of mesh-broker's events: created m drafts and m log, as on the bus module's own stream", seat) + } if _, ok := held("KV_m_drafts"); !ok { t.Error("the bus holds no KV_m_drafts") } diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 561e3536..2c757f07 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -168,7 +168,8 @@ func main() { // them. var ReadingVerbs = []string{"list", "show", "check", "orphans"} -// Events are the bus module's own events (ADR 0595 §6, ADR 0620 §2), on its own stream broker-events (BROKER_EVENTS). +// Events are the seat mesh-broker's events (ADR 0595 §6, ADR 0620 §2): said on the seat's subjects on the shared events stream +// and kept 30 days on the bus module's own stream broker-events (BROKER_EVENTS), which `history` reads (keep.go). var Events = []string{"created", "refused", "drifted", "drift-cleared", "full", "full-cleared", "orphaned", "recreated", "dropped"} // allTools is every tool and seat verb this bundle serves. diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index ac3fc53e..5a69c76a 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -28,7 +28,7 @@ const ( KeyStatus = "_keeper" ) -// The bus module's events (ADR 0595 §6), on its own stream (ADR 0620 §2). +// The seat mesh-broker's events (ADR 0595 §6): on the seat's subjects, and kept on the bus module's own stream (ADR 0620 §2). const ( EventCreated = "created" EventRefused = "refused"