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".
This commit is contained in:
2026-10-20 01:30:20 +00:00
parent 2474bf2833
commit a9789b5125
+49 -2
View File
@@ -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) {