mesh-host/internal/link/asking_nats_test.go

package link

import (
	"context"
	"encoding/json"
	"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)}
	t.Cleanup(a.Close)
	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)
	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("controller-standin"), nats.ManualAck())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	return sawReplyField
}

// 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")
	}
}