From a9789b512553db8f401d4fe48231c270d263886c Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 01:30:20 +0000 Subject: [PATCH] Give each enrolment stand-in its own durable and wait for it to be gone (issue 604) The stand-in shared one durable name across tests and let go with Unsubscribe alone, so a test starting before the server dropped the binding was refused with "consumer is already bound to a subscription". --- internal/link/asking_nats_test.go | 51 +++++++++++++++++++++++++++++-- 1 file changed, 49 insertions(+), 2 deletions(-) diff --git a/internal/link/asking_nats_test.go b/internal/link/asking_nats_test.go index 5d853b2..60f6dd2 100644 --- a/internal/link/asking_nats_test.go +++ b/internal/link/asking_nats_test.go @@ -3,6 +3,8 @@ package link import ( "context" "encoding/json" + "errors" + "strings" "testing" "time" @@ -48,6 +50,19 @@ func theMeshAnswers(t *testing.T, conn *nats.Conn, js nats.JetStreamContext, reply EnrolReply) <-chan string { t.Helper() sawReplyField := make(chan string, 1) + // **A consumer of this test's own, gone before it starts and gone when it ends.** The stand-in + // used to be one durable name shared by every test, released by Unsubscribe alone. Unsubscribe + // only asks the server to let go: it deletes the consumer only when this subscription made it, + // and the server drops the binding on its own time. A test that started in that window (or one + // that found the consumer left by an earlier run on a long-lived bus) was refused with "consumer + // is already bound to a subscription". Per-test names remove the sharing; the delete-and-wait + // before and after removes the dependence on the server's timing. + durable := standinName(t) + stream, err := js.StreamNameBySubject(EnrolSubject) + if err != nil { + t.Fatal(err) + } + standinGone(t, js, stream, durable) sub, err := js.Subscribe(EnrolSubject, func(msg *nats.Msg) { select { case sawReplyField <- msg.Reply: @@ -65,14 +80,46 @@ func theMeshAnswers(t *testing.T, conn *nats.Conn, js nats.JetStreamContext, // send it to whatever the transport's reply field holds, and that is the point. _ = conn.Publish(addressed.ReplyTo, body) _ = msg.Ack() - }, nats.Durable("controller-standin"), nats.ManualAck()) + }, nats.Durable(durable), nats.ManualAck()) if err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = sub.Unsubscribe() }) + t.Cleanup(func() { + _ = sub.Unsubscribe() + standinGone(t, js, stream, durable) + }) return sawReplyField } +// standinName is the stand-in's durable name for one test: the test's name with every character a +// consumer name may not hold replaced. +func standinName(t *testing.T) string { + t.Helper() + return "controller-standin-" + strings.Map(func(r rune) rune { + if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' { + return r + } + return '_' + }, t.Name()) +} + +// standinGone deletes the durable consumer if it exists and waits until the server no longer reports +// it, so that nothing after this call can meet its binding. +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) { + return + } + _ = js.DeleteConsumer(stream, durable) + if time.Now().After(deadline) { + t.Fatalf("consumer %s on %s still exists 5s after it was deleted", durable, stream) + } + time.Sleep(20 * time.Millisecond) + } +} + // An enrolment is answered on the address the request carried, and the transport's own reply field // held something else entirely. func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {