The assigner's integration test puts its responders' interest on the bus before it asks them (issue 1921, milestone 47/G, the group check of 616c42434ac2)

A manifest-reads ask the assigner made on manifest-reads-changed before the
test's SUB reached the server was answered by nobody and, by design, never
asked again; a flush is the barrier after each of the test's subscriptions.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-07 04:27:37 +01:00
parent 932e7ccf02
commit 4a07e7a0ce
@@ -610,6 +610,10 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
}); err != nil {
t.Fatal(err)
}
// The responder's interest is on the bus before the event that asks it is said: the assigner asks a node once per
// manifest-reads-changed and never on a timer, so an ask that lands before this subscription reached the server is
// answered by nobody and is never asked again (the group check of 616c42434ac2 on a loaded build node).
subscribed(t, nc)
say := func(why string) {
body, _ := json.Marshal(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", ManifestReads: stated, At: time.Now().Format(time.RFC3339Nano), Why: why})
if _, err := b.js.Publish(b.ctx, enginereads.SubjectPrefix+"ace", body); err != nil {
@@ -677,6 +681,16 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
}
}
// subscribed returns once the server holds every subscription nc made: nats.go sends a SUB from its flusher, after
// Subscribe has returned, so a responder or listener the test makes is not yet heard by a call or event another
// connection sends at once. A flush's round trip is the barrier; no sleep stands in for it.
func subscribed(t *testing.T, nc *nats.Conn) {
t.Helper()
if err := nc.Flush(); err != nil {
t.Fatalf("the test's subscriptions did not reach the bus: %v", err)
}
}
// bedState is one of the bed's buckets as the engine reads' tool reads it.
type bedState struct {
b *bed
@@ -1166,6 +1180,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
}); err != nil {
t.Fatal(err)
}
subscribed(t, nc)
for _, node := range []string{"anchor", "laptop"} {
rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: node}})
b.putJSON(t, buckets.Have, node+".node", rec)
@@ -1514,6 +1529,7 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) {
if err != nil {
t.Fatal(err)
}
subscribed(t, nc)
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
}
@@ -1612,6 +1628,7 @@ func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]
if err != nil {
t.Fatal(err)
}
subscribed(t, nc)
t.Cleanup(func() { _ = sub.Unsubscribe() })
ask := func(user, tool string) (views.Answer, error) {
return views.AskVia(func(key string, body any) (json.RawMessage, error) {