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

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

Browse Source 
The lead's ruling of 13:0x (registry milestone): mesh-broker's events are mesh.seat.mesh-broker.event.<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
This commit is contained in:
jschoubben
2026-11-04 12:58:44 +01:00
parent ee54bae09f
commit d7214bb576
4 changed files with 46 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/keep.go
+30 -12
View File
@@ -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.<event>`, which the
// shared events stream keeps 7 days and every consumer of `mesh-broker.<event>` 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[:])
modules/nats/cmd/nats-tools/keeper_integration_test.go
+13 -4
View File
@@ -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.<event> 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")
}
modules/nats/cmd/nats-tools/main.go
+2 -1
View File
@@ -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.
modules/nats/internal/keeper/keeper.go
+1 -1
View File
@@ -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"