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

@@ -3,6 +3,8 @@ package link
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -37,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
 }
 
@@ -48,6 +56,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 +86,50 @@ 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 {
+		_, 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)
+		}
+		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) {