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
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.<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.<event>, the seat's own subject, which
|
||||
// the shared events stream keeps and every consumer of `mesh-module-assigner.<event>` 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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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.<state>` 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"}) {
|
||||
|
||||
@@ -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"
|
||||
},
|
||||
|
||||
@@ -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"},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user