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 cb12c55a..eb8d2f19 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -552,13 +552,10 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro } } -// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable -// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of -// indexerViews, and starts once both are registered. -func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) { - account := process.Account{Node: thisNode, Module: assignment.Module} - w := notHeardWriter(findings, logf) - handle := func(m jetstream.Msg) { +// readsChangedHandler is what the assigner does with each event manifest-reads-changed: acknowledges it, keeps what it says +// of the node (a statement older than it does not stand) and asks the node again. +func readsChangedHandler(reads *enginereads.Keeper, logf func(string, ...any)) func(jetstream.Msg) { + return func(m jetstream.Msg) { if err := m.Ack(); err != nil { logf("acknowledging %s: %v", m.Subject(), err) } @@ -576,6 +573,15 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process. reads.AskNode(node, enginereads.Event) } } +} + +// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable +// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of +// indexerViews, and starts once both are registered. +func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) { + account := process.Account{Node: thisNode, Module: assignment.Module} + w := notHeardWriter(findings, logf) + handle := readsChangedHandler(reads, logf) err := hub.Handle(enginereads.SubjectPrefix+"*", handle) // At each bind every node the assigner holds a statement of is asked again: an event said before its consumer existed // is not delivered to it (issue 1966 note 155: ace's bootstrapped statement stood; ADR 0668 ยง4). diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index e2a019c4..570aca3e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -319,17 +319,15 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { } // Issue 1975: ace's engine answered its bootstrapped seats, then stated the full list and said so; an answer older // than that event does not stand: the rules read what the event says, and the node is asked until it is as new. - stale := true + // The engine's verb keeps answering the bootstrapped statement (its stated-at is older than the event); the event is + // handed to the assigner's own handler (readsChangedHandler, what hearReadsChanged registers) as the bus delivers it. bootstrapped := func(node, _ string, _ any) (json.RawMessage, error) { - if stale { - stale = false - return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"}) - } - return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a", "seat:b"}, StatedAt: "2026-11-12T01:14:39.001Z"}) + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"}) } rk := &enginereads.Keeper{Call: bootstrapped, Sync: true} rk.AskNode("ace", "the assigner started") - rk.HeardChanged(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}}) + body, _ := json.Marshal(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}}) + readsChangedHandler(rk, t.Logf)(readsMsg{subject: enginereads.SubjectPrefix + "ace", data: body}) rp := &passer{w: w, reads: rk, online: map[string]bool{"ace": true}} rn := w.nodes(nil, nil) rp.engineReads(rn) @@ -1510,3 +1508,14 @@ func readAskID(t *testing.T, p *passer, key string) string { } return d.Ask.ID } + +// readsMsg is a message of the events stream as the assigner's handler reads it. +type readsMsg struct { + jetstream.Msg + subject string + data []byte +} + +func (m readsMsg) Ack() error { return nil } +func (m readsMsg) Subject() string { return m.subject } +func (m readsMsg) Data() []byte { return m.data }