waved: answer inbound RPCs to the envelope sender

serverconn/e2e_test.go

@@ -205,9 +205,14 @@ func (s *testServer) handleRequest(ctx context.Context,
 		ProtocolVersion:    1,
 		ArkProtocolVersion: 1,
 		Sender:             s.serverMailboxID,
-		Recipient:          env.Rpc.ReplyTo,
-		Headers:            headers,
-		Body:               body,
+
+		// Mirror the production responder (waved's handleInboundRPC),
+		// which answers the envelope sender rather than the request's
+		// ReplyTo. The two are equal here, so this changes nothing
+		// today -- it keeps the double honest if they ever diverge.
+		Recipient: env.Sender,
+		Headers:   headers,
+		Body:      body,
 		Rpc: &mailboxpb.RpcMeta{
 			Kind:          mailboxpb.RpcMeta_KIND_RESPONSE,
 			CorrelationId: env.Rpc.CorrelationId,

waved/server.go

@@ -3605,8 +3605,27 @@ func (s *Server) handleInboundRPC(ctx context.Context,
 	}
 
 	responseEnv := &mailboxpb.Envelope{
-		Sender:    s.localMailboxID,
-		Recipient: env.Rpc.ReplyTo,
+		Sender: s.localMailboxID,
+
+		// A response belongs to whoever sent the request, so address
+		// it to the sender rather than to the request's ReplyTo.
+		//
+		// The two agree for every producer in this repo: each sets
+		// Sender and Rpc.ReplyTo to the same LocalMailboxID (see
+		// serverconn/unary_facade.go, serverconn/actor.go,
+		// serverconn/heartbeat.go and the operator's clientconn
+		// equivalents). So this changes nothing for a well-formed
+		// peer, and it stops a request choosing a destination that
+		// disagrees with where it came from.
+		//
+		// It also fixes an absent ReplyTo. That used to produce an
+		// empty recipient, which the mailbox store rejects outright,
+		// so the caller lost its answer entirely.
+		//
+		// Rpc.ReplyTo is therefore advisory from the responder's
+		// point of view: producers still set it, and nothing here
+		// reads it.
+		Recipient: env.Sender,
 		Headers:   headers,
 		Body:      body,
 		Rpc: &mailboxpb.RpcMeta{

waved/server_reply_to_test.go

@@ -0,0 +1,127 @@
+package waved
+
+import (
+	"context"
+	"testing"
+
+	mailboxpb "github.com/lightninglabs/wavelength/mailbox/pb"
+	mailboxrpc "github.com/lightninglabs/wavelength/mailbox/rpc"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+// recordingReplyEdge captures the envelopes handleInboundRPC sends so a test
+// can assert where a response was addressed.
+type recordingReplyEdge struct {
+	sent []*mailboxpb.Envelope
+}
+
+// Send records the outbound envelope and reports success.
+func (r *recordingReplyEdge) Send(_ context.Context, in *mailboxpb.SendRequest,
+	_ ...grpc.CallOption) (*mailboxpb.SendResponse, error) {
+
+	r.sent = append(r.sent, in.Envelope)
+
+	return &mailboxpb.SendResponse{
+		Status: &mailboxpb.Status{
+			Ok: true,
+		},
+	}, nil
+}
+
+// Pull is unused by these tests.
+func (r *recordingReplyEdge) Pull(_ context.Context, _ *mailboxpb.PullRequest,
+	_ ...grpc.CallOption) (*mailboxpb.PullResponse, error) {
+
+	return &mailboxpb.PullResponse{
+		Status: &mailboxpb.Status{
+			Ok: true,
+		},
+	}, nil
+}
+
+// AckUpTo is unused by these tests.
+func (r *recordingReplyEdge) AckUpTo(_ context.Context,
+	_ *mailboxpb.AckUpToRequest, _ ...grpc.CallOption) (
+	*mailboxpb.AckUpToResponse, error) {
+
+	return &mailboxpb.AckUpToResponse{
+		Status: &mailboxpb.Status{
+			Ok: true,
+		},
+	}, nil
+}
+
+// TestHandleInboundRPCAnswersTheSender asserts an inbound request is answered
+// to the mailbox it came from, whatever its Rpc.ReplyTo says.
+//
+// The method dispatched here is deliberately unregistered, so ServeRPC fails
+// and the handler takes its error path. That is the interesting path to pin:
+// it still sends a response envelope, so it still has to address one.
+func TestHandleInboundRPCAnswersTheSender(t *testing.T) {
+	t.Parallel()
+
+	const (
+		operatorMailboxID = "operator-1"
+		otherMailboxID    = "somebody-else"
+		daemonMailboxID   = "this-daemon"
+	)
+
+	tests := []struct {
+		name    string
+		replyTo string
+	}{{
+		name:    "matching reply-to",
+		replyTo: operatorMailboxID,
+	}, {
+		// Previously produced an empty Recipient, which the mailbox
+		// store rejects, so the caller never got its answer.
+		name:    "absent reply-to",
+		replyTo: "",
+	}, {
+		name:    "reply-to naming another mailbox",
+		replyTo: otherMailboxID,
+	}}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			edge := &recordingReplyEdge{}
+			s := newCompatTestServer(t, edge)
+			s.mailboxMux = mailboxrpc.NewServeMux()
+			s.localMailboxID = daemonMailboxID
+
+			env := &mailboxpb.Envelope{
+				Sender: operatorMailboxID,
+				Body:   &anypb.Any{},
+				Rpc: &mailboxpb.RpcMeta{
+					Kind: mailboxpb.
+						RpcMeta_KIND_REQUEST,
+					Service:       "svc.Unregistered",
+					Method:        "Method",
+					CorrelationId: "corr-1",
+					ReplyTo:       tc.replyTo,
+				},
+			}
+
+			err := s.handleInboundRPC(t.Context(), edge, env)
+			require.NoError(t, err)
+
+			require.Len(t, edge.sent, 1)
+			require.Equal(
+				t, operatorMailboxID, edge.sent[0].Recipient,
+			)
+			require.NotEqual(
+				t, otherMailboxID, edge.sent[0].Recipient,
+			)
+
+			// The other half of the envelope: the response is from
+			// us, whoever it is addressed to.
+			require.Equal(
+				t, daemonMailboxID, edge.sent[0].Sender,
+			)
+		})
+	}
+}