mesh-host/internal/link/asking_nats_test.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
)

// The enrolment round trip against a real server.
//
// **This is the test that keeps a reason from becoming folklore.** The reply address travels in the
// request's payload because a JetStream consumer's delivery has had the transport's reply field
// claimed for its own ack address — which is a fact about a server, not a rule anybody can check by
// reading. Both halves are asserted here: that the field really is eaten, and that the answer
// reaches the node anyway.
//
//	docker run -d --rm --name t -p 14223:4222 nats:2.10-alpine -js
//	MESH_TEST_NATS=nats://127.0.0.1:14223 go test ./internal/link/ -run TestNatsAnEnrolment

// asking is a conversation on a bus with no TLS. Built directly rather than through Present because
// the pin is what Present adds and PinnedConfig's own tests cover it; what is under test here is the
// address the answer comes back on.
func asking(t *testing.T, conn *nats.Conn, js nats.JetStreamContext, node string) *natsAsking {
	t.Helper()
	inbox, err := enrolInbox(node)
	if err != nil {
		t.Fatal(err)
	}
	answers, err := conn.SubscribeSync(inbox)
	if err != nil {
		t.Fatal(err)
	}
	if err := conn.Flush(); err != nil {
		t.Fatal(err)
	}
	a := &natsAsking{conn: conn, js: js, inbox: inbox, answers: answers, lost: make(chan error, 1)}
	// **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
}

// theMeshAnswers stands in for the controller: it consumes the enrolment off the stream, reads the
// reply address out of the payload — never from the transport field — and answers there. It reports
// what the transport field actually held, which is the claim design 25 §2 rests on.
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:
		default:
		}
		var addressed struct {
			ReplyTo string `json:"reply_to"`
		}
		if err := json.Unmarshal(msg.Data, &addressed); err != nil || addressed.ReplyTo == "" {
			_ = msg.Ack()
			return
		}
		body, _ := json.Marshal(reply)
		// Published explicitly to the address the payload named, never msg.Respond — which would
		// send it to whatever the transport's reply field holds, and that is the point.
		_ = conn.Publish(addressed.ReplyTo, body)
		_ = msg.Ack()
	}, nats.Durable(durable), nats.ManualAck())
	if err != nil {
		t.Fatal(err)
	}
	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) {
	conn, js := aBus(t)
	const node = "joining"

	sawReplyField := theMeshAnswers(t, conn, js, EnrolReply{Accepted: true, Node: node, Password: "p"})
	a := asking(t, conn, js, node)

	request, _ := json.Marshal(EnrolRequest{Node: node, Secret: "t"})
	answer, err := a.Ask(context.Background(), request, 8*time.Second)
	if err != nil {
		t.Fatalf("no answer reached the node: %v", err)
	}
	var reply EnrolReply
	if err := json.Unmarshal(answer, &reply); err != nil {
		t.Fatal(err)
	}
	if !reply.Accepted || reply.Node != node {
		t.Fatalf("the answer was not the mesh's: %+v", reply)
	}

	// And the field the answer would have gone to, had it used the transport's: the consumer's own
	// ack address. If a future server stopped doing this, the payload-borne address would still
	// work and this line is what would say the reason had changed.
	select {
	case field := <-sawReplyField:
		if field == a.inbox {
			t.Fatalf("the transport's reply field held this node's inbox (%s), so the payload "+
				"address is no longer load-bearing — check design 25 §2 before relying on it", field)
		}
		if field == "" {
			t.Fatal("the transport's reply field was empty rather than claimed, which is a third " +
				"behaviour from the two design 25 §2 describes")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the stand-in never saw the request")
	}
}

// A request that names no reply address is not answered, and the node says so as a mesh that is not
// running rather than hanging. The controller has nowhere to send an answer, which is the failure
// the payload field exists to make impossible — asserted so that a request built without it fails
// loudly here rather than quietly on a machine.
func TestNatsAnEnrolmentWithNoReplyAddressIsNotAnswered(t *testing.T) {
	conn, js := aBus(t)
	const node = "silent"

	theMeshAnswers(t, conn, js, EnrolReply{Accepted: true, Node: node})
	a := asking(t, conn, js, node)

	// Published without going through Ask, so the reply address is genuinely absent.
	request, _ := json.Marshal(EnrolRequest{Node: node, Secret: "t"})
	if _, err := js.Publish(EnrolSubject, request); err != nil {
		t.Fatal(err)
	}
	if _, err := a.Ask(context.Background(), []byte(`{"node":"`+node+`"}`), 0); err == nil {
		t.Fatal("a node with no answer coming was told it had one")
	}
}