jschoubben
abce68fdf0
Design 25 §2 builds enrolment around carrying the reply subject in the payload, because a JetStream consumer claims the transport Reply field for its own ack address. The whole handshake rests on it, so it is checked: the caller asked for _INBOX.LCr3M83q... and the consumer saw $JS.ACK.PROBE.probe_consumer... The design was right, and the workaround is necessary rather than defensive. Worth having as a test rather than a note: if a future server version stopped doing this, enrolment would keep working and the reason for the payload field would quietly become folklore.
82 lines
2.9 KiB
Go
82 lines
2.9 KiB
Go
package link
|
|
|
|
import (
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// **Verified 2026-09-27**: the caller asked for a reply to `_INBOX.LCr3M83q…` and the consumer
|
|
// saw `$JS.ACK.PROBE.probe_consumer.1.1.1…`. The design was right, and enrolment's payload-borne
|
|
// reply subject is necessary rather than defensive.
|
|
//
|
|
// Design 25 §2 asserts that a reply address is **eaten** when a message crosses a stream: core
|
|
// request/reply puts the requester's inbox in the message's Reply field, but a JetStream consumer
|
|
// has already claimed that field for its own ack address by the time a handler sees it. The whole
|
|
// enrolment design rests on it — the reply subject travels in the payload instead — so it is
|
|
// checked rather than believed.
|
|
//
|
|
// docker run -d --rm --name t -p 14222:4222 nats:2.10-alpine -js
|
|
// MESH_TEST_NATS=nats://127.0.0.1:14222 go test ./internal/link/ -run TestAReplyAddress
|
|
func TestAReplyAddressDoesNotSurviveAStream(t *testing.T) {
|
|
url := os.Getenv("MESH_TEST_NATS")
|
|
if url == "" {
|
|
t.Skip("MESH_TEST_NATS unset")
|
|
}
|
|
conn, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
js, err := conn.JetStream()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := js.AddStream(&nats.StreamConfig{
|
|
Name: "PROBE", Subjects: []string{"probe.>"}, Retention: nats.WorkQueuePolicy,
|
|
}); err != nil && err != nats.ErrStreamNameAlreadyInUse {
|
|
t.Fatal(err)
|
|
}
|
|
defer js.DeleteStream("PROBE")
|
|
|
|
seen := make(chan *nats.Msg, 1)
|
|
sub, err := js.Subscribe("probe.enrol", func(m *nats.Msg) { seen <- m },
|
|
nats.Durable("probe_consumer"), nats.ManualAck())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer sub.Unsubscribe()
|
|
|
|
// A caller doing what core request/reply does: publish with its own inbox as the reply.
|
|
inbox := nats.NewInbox()
|
|
if err := conn.PublishMsg(&nats.Msg{Subject: "probe.enrol", Reply: inbox, Data: []byte("{}")}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
select {
|
|
case m := <-seen:
|
|
t.Logf("the caller asked for a reply to %s", inbox)
|
|
t.Logf("the consumer sees a Reply field of %s", m.Reply)
|
|
if m.Reply == inbox {
|
|
t.Fatalf("the reply address SURVIVED the stream. Design 25 §2 says it does not, and "+
|
|
"builds enrolment around carrying the reply subject in the payload to work "+
|
|
"around it. If this holds generally, that work is unnecessary and the design "+
|
|
"should say so.")
|
|
}
|
|
if m.Reply == "" {
|
|
t.Fatal("the Reply field is empty rather than claimed; the design says it carries " +
|
|
"the consumer's ack address, which is a different fact")
|
|
}
|
|
// Answering it would send the enrolling node's credentials to an ack subject.
|
|
if len(m.Reply) < 7 || m.Reply[:7] != "$JS.ACK" {
|
|
t.Errorf("the Reply field is %q, which is neither the caller's inbox nor an ack "+
|
|
"address — the design's reasoning assumes one of the two", m.Reply)
|
|
}
|
|
m.Ack()
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("nothing was delivered")
|
|
}
|
|
}
|