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 28aaa6ce..6d77bef6 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 @@ -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) {