diff --git a/internal/link/asking_nats_test.go b/internal/link/asking_nats_test.go index 5d853b2..919b9d9 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" @@ -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) {