MailboxRPC 7/7: end-to-end behavior tests

go.mod

@@ -30,6 +30,7 @@ require (
 	github.com/ory/dockertest/v3 v3.12.0
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250324211829-b45e905df463
 	google.golang.org/grpc v1.73.0
 	google.golang.org/protobuf v1.36.6
 	gopkg.in/macaroon.v2 v2.1.0
@@ -185,7 +186,6 @@ require (
 	golang.org/x/tools v0.39.0 // indirect
 	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250324211829-b45e905df463 // indirect
 	gopkg.in/errgo.v1 v1.0.1 // indirect
 	gopkg.in/macaroon-bakery.v2 v2.1.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect

mailbox/rpc/grpc_status.go

@@ -0,0 +1,79 @@
+package mailboxrpc
+
+import (
+	"encoding/base64"
+	"fmt"
+
+	spb "google.golang.org/genproto/googleapis/rpc/status"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+const (
+	// HeaderGRPCStatusB64 is the envelope header key that carries a
+	// base64-encoded google.rpc.Status protobuf when an RPC fails on the
+	// server side.
+	//
+	// When present on a KIND_RESPONSE envelope, the receiver should decode
+	// the status and surface it as a gRPC error rather than unmarshaling
+	// the body.
+	HeaderGRPCStatusB64 = "mailboxrpc.grpc_status_b64"
+)
+
+// EncodeErrorHeaders serializes err as a gRPC status and returns envelope
+// headers that carry it in base64-encoded protobuf form.
+//
+// Returns nil when err is nil so callers can safely pass the result to
+// envelope construction without a nil check.
+func EncodeErrorHeaders(err error) map[string]string {
+	if err == nil {
+		return nil
+	}
+
+	// Preserve any existing gRPC status code. For plain errors that
+	// don't carry a gRPC status, FromError returns Unknown with the
+	// original error message.
+	st, _ := status.FromError(err)
+
+	blob, marshalErr := proto.Marshal(st.Proto())
+	if marshalErr != nil {
+		// Fall back to a minimal Internal status when marshaling itself
+		// fails, so callers always get a usable error header.
+		fallback := status.New(codes.Internal,
+			"failed to marshal error status")
+		blob, _ = proto.Marshal(fallback.Proto())
+	}
+
+	return map[string]string{
+		HeaderGRPCStatusB64: base64.StdEncoding.EncodeToString(blob),
+	}
+}
+
+// DecodeErrorHeaders returns the gRPC error encoded in headers, or nil if no
+// error header is present.
+//
+// Callers should check the returned error before attempting to unmarshal the
+// response body — a non-nil error means the RPC failed on the server.
+func DecodeErrorHeaders(headers map[string]string) error {
+	if len(headers) == 0 {
+		return nil
+	}
+
+	b64, ok := headers[HeaderGRPCStatusB64]
+	if !ok || b64 == "" {
+		return nil
+	}
+
+	blob, err := base64.StdEncoding.DecodeString(b64)
+	if err != nil {
+		return fmt.Errorf("decode grpc status header: %w", err)
+	}
+
+	var pb spb.Status
+	if err := proto.Unmarshal(blob, &pb); err != nil {
+		return fmt.Errorf("unmarshal grpc status proto: %w", err)
+	}
+
+	return status.FromProto(&pb).Err()
+}

mailbox/rpc/grpc_status_test.go

@@ -0,0 +1,139 @@
+package mailboxrpc
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// TestEncodeErrorHeaders_NilError verifies that EncodeErrorHeaders returns nil
+// when given a nil error, so callers can safely pass the result without a nil
+// check.
+func TestEncodeErrorHeaders_NilError(t *testing.T) {
+	t.Parallel()
+
+	require.Nil(t, EncodeErrorHeaders(nil))
+}
+
+// TestEncodeDecodeRoundTrip verifies that a gRPC status error survives the
+// encode → header → decode round-trip with code and message preserved.
+func TestEncodeDecodeRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+		code codes.Code
+		msg  string
+	}{
+		{
+			name: "plain error becomes Unknown",
+			err:  errors.New("something broke"),
+			code: codes.Unknown,
+			msg:  "something broke",
+		},
+		{
+			name: "NotFound status preserved",
+			err:  status.Error(codes.NotFound, "no such item"),
+			code: codes.NotFound,
+			msg:  "no such item",
+		},
+		{
+			name: "PermissionDenied status preserved",
+			err: status.Error(
+				codes.PermissionDenied, "access denied",
+			),
+			code: codes.PermissionDenied,
+			msg:  "access denied",
+		},
+		{
+			name: "Internal status preserved",
+			err:  status.Error(codes.Internal, "server fault"),
+			code: codes.Internal,
+			msg:  "server fault",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			headers := EncodeErrorHeaders(tc.err)
+			require.NotNil(t, headers)
+			require.Contains(t, headers, HeaderGRPCStatusB64)
+
+			decoded := DecodeErrorHeaders(headers)
+			require.Error(t, decoded)
+
+			st, ok := status.FromError(decoded)
+			require.True(t, ok)
+			require.Equal(t, tc.code, st.Code())
+			require.Equal(t, tc.msg, st.Message())
+		})
+	}
+}
+
+// TestDecodeErrorHeaders_NilAndEmpty verifies that DecodeErrorHeaders returns
+// nil for nil maps, empty maps, and maps without the status header key.
+func TestDecodeErrorHeaders_NilAndEmpty(t *testing.T) {
+	t.Parallel()
+
+	require.NoError(t, DecodeErrorHeaders(nil))
+	require.NoError(t, DecodeErrorHeaders(map[string]string{}))
+	require.NoError(t, DecodeErrorHeaders(map[string]string{
+		"other-key": "value",
+	}))
+}
+
+// TestDecodeErrorHeaders_EmptyValue verifies that an empty string value for
+// the status header key is treated as absent.
+func TestDecodeErrorHeaders_EmptyValue(t *testing.T) {
+	t.Parallel()
+
+	err := DecodeErrorHeaders(map[string]string{
+		HeaderGRPCStatusB64: "",
+	})
+	require.NoError(t, err)
+}
+
+// TestDecodeErrorHeaders_InvalidBase64 verifies that malformed base64 in the
+// header produces a descriptive error rather than a panic.
+func TestDecodeErrorHeaders_InvalidBase64(t *testing.T) {
+	t.Parallel()
+
+	err := DecodeErrorHeaders(map[string]string{
+		HeaderGRPCStatusB64: "not-valid-base64!@#$",
+	})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "decode grpc status header")
+}
+
+// TestDecodeErrorHeaders_InvalidProto verifies that valid base64 containing
+// garbage proto bytes produces a descriptive unmarshal error.
+func TestDecodeErrorHeaders_InvalidProto(t *testing.T) {
+	t.Parallel()
+
+	// Valid base64 but not a valid protobuf Status message. Use a byte
+	// sequence that base64-encodes cleanly but contains invalid proto
+	// field tags.
+	err := DecodeErrorHeaders(map[string]string{
+		HeaderGRPCStatusB64: "////",
+	})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unmarshal grpc status proto")
+}
+
+// TestEncodeErrorHeaders_HeaderKeyPresent verifies the header map always uses
+// the canonical HeaderGRPCStatusB64 key.
+func TestEncodeErrorHeaders_HeaderKeyPresent(t *testing.T) {
+	t.Parallel()
+
+	headers := EncodeErrorHeaders(errors.New("test"))
+	require.Len(t, headers, 1)
+
+	_, ok := headers[HeaderGRPCStatusB64]
+	require.True(t, ok)
+}

serverconn/actor.go

@@ -47,6 +47,21 @@ type ServerMessage interface {
 	ToProto() proto.Message
 }
 
+// InboundServerMessage is implemented by actor messages that arrive from the
+// server via the mailbox ingress loop. FromProto mirrors the ToProto method
+// on ServerMessage, completing the bidirectional proto<->actor message
+// conversion pair.
+//
+// Callers that implement this interface on their actor message types can use
+// the NewEventRoute helper to avoid writing explicit Adapt functions.
+type InboundServerMessage interface {
+	// FromProto populates the receiver from a server-pushed proto message.
+	// It is called by the EventRouter dispatch closure after the envelope
+	// body has been unmarshaled into the expected proto type. Return an
+	// error to reject events whose proto fields cannot be converted.
+	FromProto(proto.Message) error
+}
+
 // rawServerMessage wraps a protobuf Any for reconstructing a ServerMessage
 // after TLV deserialization. The original concrete type is recovered using
 // the global protobuf type registry via anypb.UnmarshalNew.

serverconn/connector_test.go

@@ -35,17 +35,10 @@ func newTestConnector(
 	t.Helper()
 
 	mb := newInMemoryMailbox()
-	edge := &fakeMailboxServiceClient{mb: mb}
 	store := newMemCheckpointStore()
 
-	cfg := DefaultConnectorConfig()
-	cfg.Edge = edge
-	cfg.LocalMailboxID = "client-1"
-	cfg.RemoteMailboxID = "server-1"
-	cfg.ProtocolVersion = 1
+	cfg := newTestConnectorConfig(mb, store)
 	cfg.Dispatchers = dispatchers
-	cfg.Store = store
-	cfg.PullWaitTimeout = 50 * time.Millisecond
 	cfg.RetryBaseDelay = 10 * time.Millisecond
 	cfg.RetryMaxDelay = 50 * time.Millisecond