From d6f8b00bf822989477fbc6268cebbdc43b4d1b4f Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 01:34:27 +0000 Subject: [PATCH] Do not let the asking's Close shut the stand-in's connection before its cleanup (issue 604) The asking helper registered a.Close, which closes the shared test connection; the stand-in's cleanup ran after it on a closed connection, so its consumer was never deleted and blocked the next test on the workqueue stream. --- internal/link/asking_nats_test.go | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/internal/link/asking_nats_test.go b/internal/link/asking_nats_test.go index 60f6dd2..919b9d9 100644 --- a/internal/link/asking_nats_test.go +++ b/internal/link/asking_nats_test.go @@ -39,7 +39,13 @@ func asking(t *testing.T, conn *nats.Conn, js nats.JetStreamContext, node string t.Fatal(err) } a := &natsAsking{conn: conn, js: js, inbox: inbox, answers: answers, lost: make(chan error, 1)} - t.Cleanup(a.Close) + // **Not a.Close.** Close closes the connection, and this connection is the test's own, shared with + // the stand-in. Cleanups run last-in first-out and the stand-in was made first, so its cleanup + // would run on a closed connection: Unsubscribe returns ErrConnectionClosed without deleting the + // consumer it made, and every request after that fails. The consumer then outlives the test on a + // workqueue stream, which allows one consumer per filter, and the next test cannot subscribe + // (issue 604). The connection is closed by aBus, last of all. + t.Cleanup(func() { _ = answers.Unsubscribe() }) return a } @@ -109,9 +115,13 @@ func standinGone(t *testing.T, js nats.JetStreamContext, stream, durable string) t.Helper() deadline := time.Now().Add(5 * time.Second) for { - if _, err := js.ConsumerInfo(stream, durable); errors.Is(err, nats.ErrConsumerNotFound) { + _, err := js.ConsumerInfo(stream, durable) + if errors.Is(err, nats.ErrConsumerNotFound) { return } + if errors.Is(err, nats.ErrConnectionClosed) { + t.Fatalf("the connection closed before consumer %s was deleted: %v", durable, err) + } _ = js.DeleteConsumer(stream, durable) if time.Now().After(deadline) { t.Fatalf("consumer %s on %s still exists 5s after it was deleted", durable, stream)