From 3e7281cb14366941a739bae750539b18fa860458 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 17:32:54 +0100 Subject: [PATCH] assigner: the seat mesh-module-assigner's four events are said on the seat's subject, not on a stream of the module's own; its placement stream and the judge's exception for it are gone (V1; ruling 18:4x, issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-assigner/integration_test.go | 3 ++- .../cmd/mesh-delivery-assigner/main.go | 2 +- .../cmd/mesh-delivery-assigner/seat.go | 10 ++++++---- .../cmd/mesh-delivery-assigner/seat_tools.go | 2 +- .../cmd/mesh-delivery-assigner/seat_tools_test.go | 4 ++-- .../internal/placement/state.go | 7 +++---- modules/mesh-delivery-assigner/manifest_test.go | 14 ++++++++------ modules/mesh-delivery-assigner/module.json | 12 ------------ modules/mesh-delivery-indexer/judge/rules.go | 1 - 9 files changed, 23 insertions(+), 32 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 283a83df..8e2446e9 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -43,6 +43,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "mesh-delivery-assigner/internal/judge" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" @@ -1136,7 +1137,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { t.Fatal(err) } t.Cleanup(nc.Close) - if _, err := nc.Subscribe(process.ModuleStreamSubject(mmac.Module, mma.EventPlacementChanged), func(m *nats.Msg) { + if _, err := nc.Subscribe(seatdef.EventSubject(mma.Seat, mma.EventPlacementChanged), func(m *nats.Msg) { var e mma.PlacementEvent if json.Unmarshal(m.Data, &e) == nil { heard <- e diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 67257b1e..775b8969 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -467,7 +467,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, p.seat = newSeatActs(bus, wake, logger.Printf) } if js != nil { - p.publish = streamPublisher(ctx, js, thisNode) + p.publish = seatPublisher(ctx, js, thisNode) } wake() for { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index add29071..7d2a6e3a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -26,7 +26,7 @@ import ( occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" mmac "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient" mma "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner" - "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "mesh-delivery-assigner/internal/assign" "mesh-delivery-assigner/internal/judge" @@ -45,14 +45,16 @@ import ( // publisher says one of the seat's events on the assigner's stream, under a message id the bus keeps one of. type publisher func(event string, body any, id string) error -// streamPublisher publishes on mesh.stream.mesh-delivery-assigner.event., with the headers an emit carries. -func streamPublisher(ctx context.Context, js jetstream.JetStream, node string) publisher { +// seatPublisher publishes the seat's events on mesh.seat.mesh-module-assigner.event., the seat's own subject, which +// the shared events stream keeps and every consumer of `mesh-module-assigner.` hears (the seat's family, 7 days), +// with the headers an emit carries. +func seatPublisher(ctx context.Context, js jetstream.JetStream, node string) publisher { return func(event string, body any, id string) error { data, err := json.Marshal(body) if err != nil { return err } - msg := nats.NewMsg(process.ModuleStreamSubject(mmac.Module, event)) + msg := nats.NewMsg(seatdef.EventSubject(mma.Seat, event)) msg.Header.Set("x-event-id", id) msg.Header.Set("x-source", mmac.Module) msg.Header.Set("x-node", node) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go index c6535858..701aeb1a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -401,7 +401,7 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) { } } out.Older = fmt.Sprintf("the state keeps %d revisions of each key; an older change is in its placement-changed event on "+ - "the assigner's stream %s, kept %d days", plc.PlacementHistory, plc.Stream, plc.StreamDays) + "the seat's subject on the shared events stream, kept %d days", plc.PlacementHistory, plc.EventDays) return out, nil } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index ed4dac8a..f4cd84cf 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -112,10 +112,10 @@ func TestTheSeatsVerbs(t *testing.T) { // placement and proposals read every state the seat keeps (§10); history names the stream for what is older. reads := map[string]bool{} pa, err := s.doPlacement(map[string]any{"module": "web", "history": "true"}) - if a := pa.(placementAnswer); err != nil || !slices.Equal(a.Nodes, []string{"anchor"}) || len(a.Changes) != 1 || !strings.Contains(a.Older, plc.Stream) { + if a := pa.(placementAnswer); err != nil || !slices.Equal(a.Nodes, []string{"anchor"}) || len(a.Changes) != 1 || !strings.Contains(a.Older, "shared events stream") { t.Errorf("placement of web = %+v, %v", pa, err) } else { - reads[mma.StatePlacement], reads["stream "+plc.Stream] = true, true + reads[mma.StatePlacement], reads["events"] = true, true } if _, err := s.doPlacement(map[string]any{"node": "laptop"}); err == nil { t.Error("a node with no key was answered") diff --git a/modules/mesh-delivery-assigner/internal/placement/state.go b/modules/mesh-delivery-assigner/internal/placement/state.go index bc776ee6..b7578ffa 100644 --- a/modules/mesh-delivery-assigner/internal/placement/state.go +++ b/modules/mesh-delivery-assigner/internal/placement/state.go @@ -18,8 +18,7 @@ const ( const ( // PlacementHistory is the revisions the placement bucket keeps per key: the bus's most (§6). PlacementHistory = 64 - // Stream is the holder's stream of the seat's events, kept StreamDays (§6), and declared as the holder's data so the - // mesh's backups keep it. - Stream = "placement" - StreamDays = 30 + // EventDays is how long the seat's events are kept: they ride the shared events stream on the seat's own subject + // (the seat's family, 7 days). + EventDays = 7 ) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index f543a11f..c3c30ccb 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -14,7 +14,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/buckets" - plc "mesh-delivery-assigner/internal/placement" ) func load(t *testing.T) catalogue.Manifest { @@ -111,12 +110,15 @@ func TestManifest(t *testing.T) { } } // ADR 0635 §5: its views family is its ordinary events, `views.` among its emits, on no stream of its own. - if len(m.Streams) != 1 { - t.Errorf("streams = %+v, want placement alone", m.Streams) + // The seat's four events are said on the seat's own subject (the seat's family), so the module keeps no stream for them + // and lists none among its emits. + if len(m.Streams) != 0 { + t.Errorf("streams = %+v, want none: the seat's events ride the shared events stream", m.Streams) } - if m.Streams[0].Name != plc.Stream || m.Streams[0].MaxAgeDays != plc.StreamDays || - !slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(meshmoduleassigner.Events()))) { - t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, plc.StreamDays) + for _, e := range meshmoduleassigner.Events() { + if slices.Contains(m.Emits, e) { + t.Errorf("emits lists %s, an event the seat says", e) + } } // ADR 0618 §4: and it hears each node's engine say its reads changed. if !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "node-engine.manifest-reads-changed"}) { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 8a2ecdb2..7cf94b94 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -97,18 +97,6 @@ { "name": "first-views", "history": 1 }, { "name": "engine-reads", "history": 1 } ], - "streams": [ - { - "name": "placement", - "events": [ - "placement-proposed", - "placement-judged", - "placement-changed", - "placement-withdrawn" - ], - "max-age-days": 30 - } - ], "own-secrets": { "broker": "${dir:state}/broker" }, diff --git a/modules/mesh-delivery-indexer/judge/rules.go b/modules/mesh-delivery-indexer/judge/rules.go index bd97b48b..dfd42f6d 100644 --- a/modules/mesh-delivery-indexer/judge/rules.go +++ b/modules/mesh-delivery-indexer/judge/rules.go @@ -5559,7 +5559,6 @@ var streamsKeptUntilMoved = map[string][]string{ "mesh-build-node": {"toolchains-node"}, "mesh-mail": {"mail", "mail-sends"}, "only-office": {"convert-asks"}, - "mesh-delivery-assigner": {"placement"}, "mesh-delivery-checker": {"checks"}, "mesh-delivery-forge-bridge": {"forge"}, }