From dd9e2509ae89714afb847e6fbb7ba2a1cb9e7ac8 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 19 Feb 2026 19:56:15 -0800 Subject: [PATCH 1/5] mailbox/rpc: add gRPC status error header encoding for envelope transport Add EncodeErrorHeaders and DecodeErrorHeaders for round-tripping gRPC status errors through mailbox envelope headers using base64-encoded google.rpc.Status protobufs. The canonical header key is mailboxrpc.grpc_status_b64. This enables the unary facade to propagate typed gRPC errors across the mailbox transport without requiring a dedicated error envelope kind. The encoder preserves existing gRPC status codes and falls back to codes.Unknown for plain errors. Includes comprehensive test coverage for nil input, round-trip for various gRPC codes, empty/absent headers, and malformed payloads. NEWMSG "$@" --- go.mod | 2 +- mailbox/rpc/grpc_status.go | 79 ++++++++++++++++++ mailbox/rpc/grpc_status_test.go | 139 ++++++++++++++++++++++++++++++++ 3 files changed, 219 insertions(+), 1 deletion(-) create mode 100644 mailbox/rpc/grpc_status.go create mode 100644 mailbox/rpc/grpc_status_test.go diff --git a/go.mod b/go.mod index 7c982502d..beed05e73 100644 --- a/go.mod +++ b/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 diff --git a/mailbox/rpc/grpc_status.go b/mailbox/rpc/grpc_status.go new file mode 100644 index 000000000..807730cab --- /dev/null +++ b/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() +} diff --git a/mailbox/rpc/grpc_status_test.go b/mailbox/rpc/grpc_status_test.go new file mode 100644 index 000000000..b57772916 --- /dev/null +++ b/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) +} From 5f8983c58fc40a59b82d2b2b968d9521c0b51562 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 19 Feb 2026 19:58:01 -0800 Subject: [PATCH 2/5] serverconn: add EventRouter and InboundServerMessage interface This commit introduces the actor-aware event routing layer for inbound server push events. EventRouter maps (service, method) envelope keys to typed durable actor mailboxes via ServiceKey, complementing the existing EventMux in mailbox/rpc which uses plain function callbacks. InboundServerMessage is added to actor.go as the symmetric counterpart to ServerMessage.ToProto, completing the bidirectional proto<->actor message conversion pair. Types implementing InboundServerMessage can use the NewEventRoute convenience helper to avoid boilerplate Adapt closures. Key additions: - InboundServerMessage interface with FromProto(proto.Message) error - InboundActorMessage constraint combining actor.Message + InboundServerMessage - EventRouteConfig[M, R] for fully customizable route registration - AddRoute[M, R] generic function (package-level, Go disallows method type params) - NewEventRoute[M InboundActorMessage, R any] convenience wrapper - DispatcherMap type alias for ConnectorConfig.Dispatchers compatibility --- serverconn/actor.go | 15 +++ serverconn/event_router.go | 233 +++++++++++++++++++++++++++++++++++++ 2 files changed, 248 insertions(+) create mode 100644 serverconn/event_router.go diff --git a/serverconn/actor.go b/serverconn/actor.go index 0a8b204fb..88ed37172 100644 --- a/serverconn/actor.go +++ b/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. diff --git a/serverconn/event_router.go b/serverconn/event_router.go new file mode 100644 index 000000000..5ced4522a --- /dev/null +++ b/serverconn/event_router.go @@ -0,0 +1,233 @@ +package serverconn + +import ( + "context" + "fmt" + "sync" + + "github.com/lightninglabs/darepo-client/baselib/actor" + mailboxpb "github.com/lightninglabs/darepo-client/mailbox/pb" + mailboxrpc "github.com/lightninglabs/darepo-client/mailbox/rpc" + "google.golang.org/protobuf/proto" +) + +// InboundActorMessage is the type-constraint for actor messages that arrive +// from the server. It combines actor.Message (for dispatch via the actor +// system's Receptionist) with InboundServerMessage (for proto +// deserialization). Types that implement this constraint can be used with +// the NewEventRoute helper. +type InboundActorMessage interface { + actor.Message + InboundServerMessage +} + +// EventRouteConfig holds parameters for registering a single typed event +// route with an EventRouter. +// +// M is the actor message type that the target actor accepts. R is the +// response type (typically unused for fire-and-forget events but required by +// the actor framework's ServiceKey generic constraint). +type EventRouteConfig[M actor.Message, R any] struct { + // Service is the fully-qualified protobuf service name that appears + // in the inbound envelope's RPC metadata + // (e.g., "hellotest.v1.RoundService"). + Service string + + // Method is the protobuf method name (e.g., "RoundStarted"). + Method string + + // NewEvent must return a fresh zero-value proto.Message for the + // expected event type. It is called once per delivered envelope to + // provide the unmarshal target. + NewEvent func() proto.Message + + // Key is the ServiceKey for the target durable actor. The router + // calls key.Ref(system).Tell(ctx, msg) for each dispatched event, + // which persists the message to the actor's durable mailbox before + // returning nil. + Key actor.ServiceKey[M, R] + + // Adapt converts the deserialized proto.Message to the actor message + // type M. Return an error to reject envelopes whose body cannot be + // converted. + Adapt func(proto.Message) (M, error) +} + +// EventRouter maps inbound KIND_REQUEST and KIND_EVENT envelope routes to +// typed durable actor mailboxes via ServiceKey. +// +// EventRouter resolves target actors through the actor system's Receptionist, +// guaranteeing durable delivery before returning from each dispatch call. +// +// At wiring time, callers call AddRoute for each (service, method) pair they +// want to handle, then pass AsDispatcherMap() to ConnectorConfig.Dispatchers. +type EventRouter struct { + mu sync.RWMutex + system actor.SystemContext + routes map[mailboxrpc.ServiceMethod]EnvelopeDispatcher +} + +// NewEventRouter creates an empty EventRouter backed by the given actor system. +// The system is used to resolve ServiceKeys to actor references at dispatch +// time. +func NewEventRouter(system actor.SystemContext) *EventRouter { + return &EventRouter{ + system: system, + routes: make(map[mailboxrpc.ServiceMethod]EnvelopeDispatcher), + } +} + +// AddRoute registers a typed event route with the router. The generic +// parameters [M, R] must match the ServiceKey's type parameters. +// +// AddRoute is a package-level generic function rather than a method because Go +// does not allow methods with type parameters on non-generic types. +// +// Registration is idempotent — re-registering the same (service, method) pair +// replaces the previous route. +func AddRoute[M actor.Message, R any](r *EventRouter, + cfg EventRouteConfig[M, R]) { + + if cfg.Service == "" { + panic("serverconn: empty service name in EventRouteConfig") + } + if cfg.Method == "" { + panic("serverconn: empty method name in EventRouteConfig") + } + if cfg.NewEvent == nil { + panic("serverconn: nil NewEvent in EventRouteConfig") + } + if cfg.Adapt == nil { + panic("serverconn: nil Adapt in EventRouteConfig") + } + + // Capture config fields and system in a closure to produce the + // type-erased EnvelopeDispatcher. The closure owns the full dispatch + // chain: deserialize → adapt → Tell (persist to durable mailbox). + system := r.system + actorKey := cfg.Key + + dispatcher := func(ctx context.Context, + env *mailboxpb.Envelope) error { + + if env == nil || env.Body == nil { + return fmt.Errorf("nil envelope or body for %s/%s", + cfg.Service, cfg.Method) + } + + // Deserialize the envelope body. The body is an anypb.Any; + // its Value field carries the raw proto bytes of the inner + // event message. We unmarshal those bytes directly into the + // registered event type for forward-compatible decoding. + event := cfg.NewEvent() + if event == nil { + return fmt.Errorf("nil event prototype for %s/%s", + cfg.Service, cfg.Method) + } + + if err := (proto.UnmarshalOptions{ + DiscardUnknown: true, + }).Unmarshal(env.Body.Value, event); err != nil { + return fmt.Errorf("unmarshal %s/%s event: %w", + cfg.Service, cfg.Method, err) + } + + // Convert the proto event to the actor's message type. + actorMsg, err := cfg.Adapt(event) + if err != nil { + return fmt.Errorf("adapt %s/%s event: %w", + cfg.Service, cfg.Method, err) + } + + // Dispatch to the target actor via the service key. Ref + // returns a virtual router that load-balances across all + // registered actors for this key. Tell persists the message + // to the actor's durable mailbox before returning, satisfying + // the EnvelopeDispatcher contract of committed delivery. + return actorKey.Ref(system).Tell(ctx, actorMsg) + } + + serviceMethod := mailboxrpc.ServiceMethod{ + Service: cfg.Service, + Method: cfg.Method, + } + + r.mu.Lock() + r.routes[serviceMethod] = dispatcher + r.mu.Unlock() +} + +// InboundEventRouteConfig holds parameters for registering an event route +// where the actor message type implements InboundActorMessage. NewEventRoute +// auto-generates the Adapt closure from M.FromProto, so callers don't need +// to write one manually. +type InboundEventRouteConfig[M InboundActorMessage, R any] struct { + // Service is the fully-qualified protobuf service name + // (e.g., "hellotest.v1.HelloService"). + Service string + + // Method is the protobuf method name (e.g., "HelloStarted"). + Method string + + // Key is the ServiceKey for the target durable actor. + Key actor.ServiceKey[M, R] + + // NewEvent must return a fresh zero-value proto.Message for the + // expected event type. + NewEvent func() proto.Message + + // NewMsg must return a non-nil zero-value M. FromProto is called + // on the returned value to populate it from the deserialized event. + NewMsg func() M +} + +// NewEventRoute registers a typed event route for actor message types that +// implement InboundActorMessage. It auto-generates the Adapt closure from +// M.FromProto, eliminating boilerplate for the common case where the actor +// message knows how to deserialize itself from proto. +func NewEventRoute[M InboundActorMessage, R any](r *EventRouter, + cfg InboundEventRouteConfig[M, R]) { + + if cfg.NewMsg == nil { + panic("serverconn: nil NewMsg in InboundEventRouteConfig") + } + + newMsg := cfg.NewMsg + + AddRoute(r, EventRouteConfig[M, R]{ + Service: cfg.Service, + Method: cfg.Method, + NewEvent: cfg.NewEvent, + Key: cfg.Key, + Adapt: func(p proto.Message) (M, error) { + m := newMsg() + + return m, m.FromProto(p) + }, + }) +} + +// DispatcherMap maps envelope routing keys to their dispatch closures. It +// is the type returned by AsDispatcherMap and consumed by +// ConnectorConfig.Dispatchers. +type DispatcherMap = map[mailboxrpc.ServiceMethod]EnvelopeDispatcher + +// AsDispatcherMap returns a shallow copy of the registered routes as a +// DispatcherMap suitable for use as ConnectorConfig.Dispatchers. +// +// The returned map is safe to read concurrently. Callers should call +// AsDispatcherMap after all routes have been registered, before +// constructing the ConnectorConfig. +func (r *EventRouter) AsDispatcherMap() DispatcherMap { + r.mu.RLock() + defer r.mu.RUnlock() + + m := make( + map[mailboxrpc.ServiceMethod]EnvelopeDispatcher, len(r.routes), + ) + for k, v := range r.routes { + m[k] = v + } + + return m +} From 7691885ae7dd1409fe3b0e76b4cfed884a52c221 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 19 Feb 2026 19:58:15 -0800 Subject: [PATCH 3/5] serverconn: add hello.proto test fixture with generated stubs Add a purpose-built protobuf definition for the e2e integration tests in serverconn/testdata/hello.proto. The HelloService defines two unary RPCs (SayHello, SayGoodbye) for testing the UnaryFacade round-trip, plus three event messages for exercising server push and client fire-and-forget dispatch: - JoinGreetingRequest: client-to-server KIND_EVENT via DurableActor - HelloStartedEvent: server-to-client push via EventRouter - HelloFinalizedEvent: server-to-client push with multi-field payload Generated Go stubs live in serverconn/hellotestpb/ and include both the standard protoc output and the protoc-gen-mailboxrpc typed client and server wrappers (HelloServiceMailboxClient, RegisterHelloServiceMailboxServer). --- serverconn/hellotestpb/hello.pb.go | 445 ++++++++++++++++++ serverconn/hellotestpb/hello_mailboxrpc.pb.go | 101 ++++ serverconn/testdata/hello.proto | 64 +++ 3 files changed, 610 insertions(+) create mode 100644 serverconn/hellotestpb/hello.pb.go create mode 100644 serverconn/hellotestpb/hello_mailboxrpc.pb.go create mode 100644 serverconn/testdata/hello.proto diff --git a/serverconn/hellotestpb/hello.pb.go b/serverconn/hellotestpb/hello.pb.go new file mode 100644 index 000000000..15bc4c8f1 --- /dev/null +++ b/serverconn/hellotestpb/hello.pb.go @@ -0,0 +1,445 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.6 +// protoc v5.28.0 +// source: hello.proto + +package hellotestpb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// HelloRequest is the request message for HelloService.SayHello. +type HelloRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // name is the caller's name, echoed back in the response greeting. + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HelloRequest) Reset() { + *x = HelloRequest{} + mi := &file_hello_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HelloRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HelloRequest) ProtoMessage() {} + +func (x *HelloRequest) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HelloRequest.ProtoReflect.Descriptor instead. +func (*HelloRequest) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{0} +} + +func (x *HelloRequest) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +// HelloResponse is the response message for HelloService.SayHello. +type HelloResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // greeting is the server's reply (e.g., "Hello, Alice!"). + Greeting string `protobuf:"bytes,1,opt,name=greeting,proto3" json:"greeting,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HelloResponse) Reset() { + *x = HelloResponse{} + mi := &file_hello_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HelloResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HelloResponse) ProtoMessage() {} + +func (x *HelloResponse) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HelloResponse.ProtoReflect.Descriptor instead. +func (*HelloResponse) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{1} +} + +func (x *HelloResponse) GetGreeting() string { + if x != nil { + return x.Greeting + } + return "" +} + +// GoodbyeRequest is the request message for HelloService.SayGoodbye. +type GoodbyeRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // name is the caller's name, echoed back in the farewell. + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GoodbyeRequest) Reset() { + *x = GoodbyeRequest{} + mi := &file_hello_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GoodbyeRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GoodbyeRequest) ProtoMessage() {} + +func (x *GoodbyeRequest) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GoodbyeRequest.ProtoReflect.Descriptor instead. +func (*GoodbyeRequest) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{2} +} + +func (x *GoodbyeRequest) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +// GoodbyeResponse is the response message for HelloService.SayGoodbye. +type GoodbyeResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // farewell is the server's parting message. + Farewell string `protobuf:"bytes,1,opt,name=farewell,proto3" json:"farewell,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GoodbyeResponse) Reset() { + *x = GoodbyeResponse{} + mi := &file_hello_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GoodbyeResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GoodbyeResponse) ProtoMessage() {} + +func (x *GoodbyeResponse) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GoodbyeResponse.ProtoReflect.Descriptor instead. +func (*GoodbyeResponse) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{3} +} + +func (x *GoodbyeResponse) GetFarewell() string { + if x != nil { + return x.Farewell + } + return "" +} + +// JoinGreetingRequest is a client-to-server fire-and-forget event sent as a +// KIND_EVENT envelope. The client dispatches it via SendClientEventRequest +// through the durable actor mailbox; no response is expected. +type JoinGreetingRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // session_id identifies the greeting session the client wishes to join. + SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *JoinGreetingRequest) Reset() { + *x = JoinGreetingRequest{} + mi := &file_hello_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *JoinGreetingRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*JoinGreetingRequest) ProtoMessage() {} + +func (x *JoinGreetingRequest) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use JoinGreetingRequest.ProtoReflect.Descriptor instead. +func (*JoinGreetingRequest) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{4} +} + +func (x *JoinGreetingRequest) GetSessionId() string { + if x != nil { + return x.SessionId + } + return "" +} + +// HelloStartedEvent is a server-to-client push notification sent as a +// KIND_EVENT envelope when the server opens a new greeting session. Clients +// receive this via the EventRouter dispatcher registered on +// "hellotest.v1.HelloService"/"HelloStarted". +type HelloStartedEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // session_id uniquely identifies the started greeting session. + SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HelloStartedEvent) Reset() { + *x = HelloStartedEvent{} + mi := &file_hello_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HelloStartedEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HelloStartedEvent) ProtoMessage() {} + +func (x *HelloStartedEvent) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HelloStartedEvent.ProtoReflect.Descriptor instead. +func (*HelloStartedEvent) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{5} +} + +func (x *HelloStartedEvent) GetSessionId() string { + if x != nil { + return x.SessionId + } + return "" +} + +// HelloFinalizedEvent is a server-to-client push notification sent as a +// KIND_EVENT envelope when a greeting session ends with a farewell message. +type HelloFinalizedEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // session_id identifies the finalized greeting session. + SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"` + // farewell_message is the closing message from the server. + FarewellMessage string `protobuf:"bytes,2,opt,name=farewell_message,json=farewellMessage,proto3" json:"farewell_message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HelloFinalizedEvent) Reset() { + *x = HelloFinalizedEvent{} + mi := &file_hello_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HelloFinalizedEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HelloFinalizedEvent) ProtoMessage() {} + +func (x *HelloFinalizedEvent) ProtoReflect() protoreflect.Message { + mi := &file_hello_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HelloFinalizedEvent.ProtoReflect.Descriptor instead. +func (*HelloFinalizedEvent) Descriptor() ([]byte, []int) { + return file_hello_proto_rawDescGZIP(), []int{6} +} + +func (x *HelloFinalizedEvent) GetSessionId() string { + if x != nil { + return x.SessionId + } + return "" +} + +func (x *HelloFinalizedEvent) GetFarewellMessage() string { + if x != nil { + return x.FarewellMessage + } + return "" +} + +var File_hello_proto protoreflect.FileDescriptor + +const file_hello_proto_rawDesc = "" + + "\n" + + "\vhello.proto\x12\fhellotest.v1\"\"\n" + + "\fHelloRequest\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\"+\n" + + "\rHelloResponse\x12\x1a\n" + + "\bgreeting\x18\x01 \x01(\tR\bgreeting\"$\n" + + "\x0eGoodbyeRequest\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\"-\n" + + "\x0fGoodbyeResponse\x12\x1a\n" + + "\bfarewell\x18\x01 \x01(\tR\bfarewell\"4\n" + + "\x13JoinGreetingRequest\x12\x1d\n" + + "\n" + + "session_id\x18\x01 \x01(\tR\tsessionId\"2\n" + + "\x11HelloStartedEvent\x12\x1d\n" + + "\n" + + "session_id\x18\x01 \x01(\tR\tsessionId\"_\n" + + "\x13HelloFinalizedEvent\x12\x1d\n" + + "\n" + + "session_id\x18\x01 \x01(\tR\tsessionId\x12)\n" + + "\x10farewell_message\x18\x02 \x01(\tR\x0ffarewellMessage2\x9e\x01\n" + + "\fHelloService\x12C\n" + + "\bSayHello\x12\x1a.hellotest.v1.HelloRequest\x1a\x1b.hellotest.v1.HelloResponse\x12I\n" + + "\n" + + "SayGoodbye\x12\x1c.hellotest.v1.GoodbyeRequest\x1a\x1d.hellotest.v1.GoodbyeResponseBKZIgithub.com/lightninglabs/darepo-client/serverconn/hellotestpb;hellotestpbb\x06proto3" + +var ( + file_hello_proto_rawDescOnce sync.Once + file_hello_proto_rawDescData []byte +) + +func file_hello_proto_rawDescGZIP() []byte { + file_hello_proto_rawDescOnce.Do(func() { + file_hello_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_hello_proto_rawDesc), len(file_hello_proto_rawDesc))) + }) + return file_hello_proto_rawDescData +} + +var file_hello_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_hello_proto_goTypes = []any{ + (*HelloRequest)(nil), // 0: hellotest.v1.HelloRequest + (*HelloResponse)(nil), // 1: hellotest.v1.HelloResponse + (*GoodbyeRequest)(nil), // 2: hellotest.v1.GoodbyeRequest + (*GoodbyeResponse)(nil), // 3: hellotest.v1.GoodbyeResponse + (*JoinGreetingRequest)(nil), // 4: hellotest.v1.JoinGreetingRequest + (*HelloStartedEvent)(nil), // 5: hellotest.v1.HelloStartedEvent + (*HelloFinalizedEvent)(nil), // 6: hellotest.v1.HelloFinalizedEvent +} +var file_hello_proto_depIdxs = []int32{ + 0, // 0: hellotest.v1.HelloService.SayHello:input_type -> hellotest.v1.HelloRequest + 2, // 1: hellotest.v1.HelloService.SayGoodbye:input_type -> hellotest.v1.GoodbyeRequest + 1, // 2: hellotest.v1.HelloService.SayHello:output_type -> hellotest.v1.HelloResponse + 3, // 3: hellotest.v1.HelloService.SayGoodbye:output_type -> hellotest.v1.GoodbyeResponse + 2, // [2:4] is the sub-list for method output_type + 0, // [0:2] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_hello_proto_init() } +func file_hello_proto_init() { + if File_hello_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_hello_proto_rawDesc), len(file_hello_proto_rawDesc)), + NumEnums: 0, + NumMessages: 7, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_hello_proto_goTypes, + DependencyIndexes: file_hello_proto_depIdxs, + MessageInfos: file_hello_proto_msgTypes, + }.Build() + File_hello_proto = out.File + file_hello_proto_goTypes = nil + file_hello_proto_depIdxs = nil +} diff --git a/serverconn/hellotestpb/hello_mailboxrpc.pb.go b/serverconn/hellotestpb/hello_mailboxrpc.pb.go new file mode 100644 index 000000000..b82735c35 --- /dev/null +++ b/serverconn/hellotestpb/hello_mailboxrpc.pb.go @@ -0,0 +1,101 @@ +// Code generated by protoc-gen-mailboxrpc. DO NOT EDIT. + +package hellotestpb + +import ( + context "context" + fmt "fmt" + rpc "github.com/lightninglabs/darepo-client/mailbox/rpc" + proto "google.golang.org/protobuf/proto" +) + +// HelloServiceMailboxClient is a typed mailbox RPC client for HelloService. +type HelloServiceMailboxClient struct { + // C is the underlying RPC-over-mailbox runtime client. + C rpc.RPCClient +} + +// NewHelloServiceMailboxClient creates a typed mailbox client. +func NewHelloServiceMailboxClient(c rpc.RPCClient) *HelloServiceMailboxClient { + return &HelloServiceMailboxClient{ + C: c, + } +} + +// HelloServiceMailboxServer is the mailbox server interface for HelloService. +type HelloServiceMailboxServer interface { + // SayHello handles SayHello. + SayHello(ctx context.Context, req *HelloRequest) (*HelloResponse, error) + // SayGoodbye handles SayGoodbye. + SayGoodbye(ctx context.Context, req *GoodbyeRequest) (*GoodbyeResponse, error) +} + +// RegisterHelloServiceMailboxServer registers handlers for HelloService. +func RegisterHelloServiceMailboxServer(r rpc.Router, impl HelloServiceMailboxServer) { + r.Handle("hellotest.v1.HelloService", "SayHello", func() proto.Message { + return &HelloRequest{} + }, func(ctx context.Context, msg proto.Message) (proto.Message, error) { + req, ok := msg.(*HelloRequest) + if !ok { + return nil, fmt.Errorf("unexpected request type: %T", msg) + } + + return impl.SayHello(ctx, req) + }) + r.Handle("hellotest.v1.HelloService", "SayGoodbye", func() proto.Message { + return &GoodbyeRequest{} + }, func(ctx context.Context, msg proto.Message) (proto.Message, error) { + req, ok := msg.(*GoodbyeRequest) + if !ok { + return nil, fmt.Errorf("unexpected request type: %T", msg) + } + + return impl.SayGoodbye(ctx, req) + }) +} + +// SayHello calls the SayHello RPC. +func (c *HelloServiceMailboxClient) SayHello(ctx context.Context, req *HelloRequest, opts ...rpc.RPCOptions) (*HelloResponse, error) { + var opt rpc.RPCOptions + if len(opts) > 0 { + opt = opts[0] + } + + result, err := c.C.SendRPC(ctx, rpc.ServiceMethod{ + Service: "hellotest.v1.HelloService", + Method: "SayHello", + }, req, opt) + if err != nil { + return nil, err + } + + resp := new(HelloResponse) + if err := c.C.AwaitRPC(ctx, result.CorrelationID, resp); err != nil { + return nil, err + } + + return resp, nil +} + +// SayGoodbye calls the SayGoodbye RPC. +func (c *HelloServiceMailboxClient) SayGoodbye(ctx context.Context, req *GoodbyeRequest, opts ...rpc.RPCOptions) (*GoodbyeResponse, error) { + var opt rpc.RPCOptions + if len(opts) > 0 { + opt = opts[0] + } + + result, err := c.C.SendRPC(ctx, rpc.ServiceMethod{ + Service: "hellotest.v1.HelloService", + Method: "SayGoodbye", + }, req, opt) + if err != nil { + return nil, err + } + + resp := new(GoodbyeResponse) + if err := c.C.AwaitRPC(ctx, result.CorrelationID, resp); err != nil { + return nil, err + } + + return resp, nil +} diff --git a/serverconn/testdata/hello.proto b/serverconn/testdata/hello.proto new file mode 100644 index 000000000..2cabbe163 --- /dev/null +++ b/serverconn/testdata/hello.proto @@ -0,0 +1,64 @@ +syntax = "proto3"; + +package hellotest.v1; + +option go_package = "github.com/lightninglabs/darepo-client/serverconn/hellotestpb;hellotestpb"; + +// HelloService is a simple unary echo service used to exercise the mailbox +// unary facade in e2e tests. SayHello and SayGoodbye are pure request/response +// RPCs sent as KIND_REQUEST/KIND_RESPONSE envelope pairs via the unary facade. +service HelloService { + rpc SayHello (HelloRequest) returns (HelloResponse); + rpc SayGoodbye (GoodbyeRequest) returns (GoodbyeResponse); +} + +// HelloRequest is the request message for HelloService.SayHello. +message HelloRequest { + // name is the caller's name, echoed back in the response greeting. + string name = 1; +} + +// HelloResponse is the response message for HelloService.SayHello. +message HelloResponse { + // greeting is the server's reply (e.g., "Hello, Alice!"). + string greeting = 1; +} + +// GoodbyeRequest is the request message for HelloService.SayGoodbye. +message GoodbyeRequest { + // name is the caller's name, echoed back in the farewell. + string name = 1; +} + +// GoodbyeResponse is the response message for HelloService.SayGoodbye. +message GoodbyeResponse { + // farewell is the server's parting message. + string farewell = 1; +} + +// JoinGreetingRequest is a client-to-server fire-and-forget event sent as a +// KIND_EVENT envelope. The client dispatches it via SendClientEventRequest +// through the durable actor mailbox; no response is expected. +message JoinGreetingRequest { + // session_id identifies the greeting session the client wishes to join. + string session_id = 1; +} + +// HelloStartedEvent is a server-to-client push notification sent as a +// KIND_EVENT envelope when the server opens a new greeting session. Clients +// receive this via the EventRouter dispatcher registered on +// "hellotest.v1.HelloService"/"HelloStarted". +message HelloStartedEvent { + // session_id uniquely identifies the started greeting session. + string session_id = 1; +} + +// HelloFinalizedEvent is a server-to-client push notification sent as a +// KIND_EVENT envelope when a greeting session ends with a farewell message. +message HelloFinalizedEvent { + // session_id identifies the finalized greeting session. + string session_id = 1; + + // farewell_message is the closing message from the server. + string farewell_message = 2; +} From 0675844254613fa79c190faa8b75a6d9a5d709f0 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 19 Feb 2026 19:58:31 -0800 Subject: [PATCH 4/5] serverconn: add e2e tests for unary RPC, push events, and durable egress Add four end-to-end integration tests that exercise the full mailbox transport stack through a real DurableActor with in-memory durability. Each test stands up a testServer that simulates server-side behavior using the same inMemoryMailbox transport, and wires up EventRouter for inbound push events alongside the UnaryFacade for request/response. Test coverage: - TestE2E_UnaryRPC: round-trips SayHello and SayGoodbye through the UnaryFacade, verifying KIND_REQUEST/KIND_RESPONSE envelope pairs flow correctly through Send/Pull/AckUpTo. - TestE2E_ServerPushEvent: registers a HelloStartedEvent route via EventRouter backed by a greetingBehavior actor, then pushes an event from the server and verifies the actor receives the correctly deserialized message through its durable mailbox. - TestE2E_ClientFireAndForget: sends a JoinGreetingRequest as a KIND_EVENT through the DurableActor's egress path (TLV type 2000), verifying the server receives the envelope with the correct service and method metadata. - TestE2E_UnaryAndPush: combines unary RPC and server push in a single session, verifying both dispatch paths coexist without interference on a shared transport. --- serverconn/connector_test.go | 9 +- serverconn/e2e_test.go | 605 ++++++++++++++++++++++++++++++ serverconn/restart_replay_test.go | 6 +- serverconn/runtime_test.go | 15 +- serverconn/testutil_test.go | 21 ++ 5 files changed, 630 insertions(+), 26 deletions(-) create mode 100644 serverconn/e2e_test.go diff --git a/serverconn/connector_test.go b/serverconn/connector_test.go index 6e8dd6260..8875583fc 100644 --- a/serverconn/connector_test.go +++ b/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 diff --git a/serverconn/e2e_test.go b/serverconn/e2e_test.go new file mode 100644 index 000000000..0844e4708 --- /dev/null +++ b/serverconn/e2e_test.go @@ -0,0 +1,605 @@ +package serverconn + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/lightninglabs/darepo-client/baselib/actor" + mailboxpb "github.com/lightninglabs/darepo-client/mailbox/pb" + mailboxrpc "github.com/lightninglabs/darepo-client/mailbox/rpc" + "github.com/lightninglabs/darepo-client/serverconn/hellotestpb" + fn "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" +) + +// helloStartedMsg is the actor-local representation of a server-pushed +// HelloStartedEvent. It implements InboundActorMessage so it can be +// registered with NewEventRoute for automatic FromProto dispatch. +type helloStartedMsg struct { + actor.BaseMessage + + // SessionID identifies the greeting session. + SessionID string +} + +// MessageType returns a human-readable type name for logging. +func (m *helloStartedMsg) MessageType() string { return "HelloStartedMsg" } + +// FromProto populates the message from a deserialized HelloStartedEvent. +func (m *helloStartedMsg) FromProto(p proto.Message) error { + ev, ok := p.(*hellotestpb.HelloStartedEvent) + if !ok { + return fmt.Errorf("unexpected proto type: %T", p) + } + + m.SessionID = ev.SessionId + + return nil +} + +// joinGreetingServerMsg wraps a JoinGreetingRequest for outbound durable +// delivery. It implements ServerMessage so it can be sent via +// SendClientEventRequest through the DurableActor. +type joinGreetingServerMsg struct { + actor.BaseMessage + + // SessionID identifies the greeting session to join. + SessionID string +} + +// ToProto converts to a proto message for mailbox envelope transport. +func (m *joinGreetingServerMsg) ToProto() proto.Message { + return &hellotestpb.JoinGreetingRequest{ + SessionId: m.SessionID, + } +} + +// Compile-time interface checks. +var ( + _ InboundServerMessage = (*helloStartedMsg)(nil) + _ ServerMessage = (*joinGreetingServerMsg)(nil) +) + +// greetingBehavior is a trivial actor behavior that records received +// helloStartedMsg messages on a channel for test assertions. +type greetingBehavior struct { + received chan *helloStartedMsg +} + +// Receive processes a single helloStartedMsg by forwarding it to the +// test's observation channel. +func (b *greetingBehavior) Receive( + ctx context.Context, msg *helloStartedMsg, +) fn.Result[struct{}] { + + select { + case b.received <- msg: + case <-ctx.Done(): + return fn.Err[struct{}](ctx.Err()) + } + + return fn.Ok(struct{}{}) +} + +// testServer simulates the remote server side of a mailbox connection. It +// polls the server-side mailbox for inbound client envelopes and dispatches +// them: KIND_REQUEST envelopes go through a ServeMux, KIND_EVENT envelopes +// are recorded on the received channel. +type testServer struct { + mb *inMemoryMailbox + mux *mailboxrpc.ServeMux + serverMailboxID string + + // received tracks fire-and-forget events delivered to the server. + received chan *mailboxpb.Envelope +} + +// newTestServer creates a server simulator backed by the given mailbox. +func newTestServer( + mb *inMemoryMailbox, serverMailboxID string, +) *testServer { + + return &testServer{ + mb: mb, + mux: mailboxrpc.NewServeMux(), + serverMailboxID: serverMailboxID, + received: make(chan *mailboxpb.Envelope, 20), + } +} + +// run polls the server mailbox and dispatches incoming envelopes until +// ctx is cancelled. +func (s *testServer) run(ctx context.Context) { + var cursor uint64 + + for { + select { + case <-ctx.Done(): + return + + default: + } + + envs, next, status := s.mb.pull( + ctx, s.serverMailboxID, cursor, 10, + 50*time.Millisecond, + ) + if !status.Ok || len(envs) == 0 { + continue + } + + cursor = next + + for _, env := range envs { + if env.Rpc == nil { + continue + } + + switch env.Rpc.Kind { + case mailboxpb.RpcMeta_KIND_REQUEST: + s.handleRequest(ctx, env) + + case mailboxpb.RpcMeta_KIND_EVENT: + select { + case s.received <- env: + default: + } + } + } + } +} + +// handleRequest dispatches a KIND_REQUEST envelope through the ServeMux +// and sends the response envelope back to the client's ReplyTo mailbox. +func (s *testServer) handleRequest( + ctx context.Context, env *mailboxpb.Envelope, +) { + + if env.Body == nil { + return + } + + respMsg, err := s.mux.ServeRPC( + ctx, env.Rpc.Service, env.Rpc.Method, env.Body.Value, + ) + + var ( + body *anypb.Any + headers map[string]string + ) + + if err != nil { + // Transport the error via grpc_status headers so the client + // can surface it as a gRPC status error. + headers = mailboxrpc.EncodeErrorHeaders(err) + body = &anypb.Any{} + } else if body, err = anypb.New(respMsg); err != nil { + // If wrapping the response fails (e.g., unregistered type + // URL), surface it as a server-side Internal error so the + // client sees a clear gRPC failure rather than garbled + // bytes. + headers = mailboxrpc.EncodeErrorHeaders(fmt.Errorf( + "wrap response in Any: %w", err, + )) + body = &anypb.Any{} + } + + responseEnv := &mailboxpb.Envelope{ + ProtocolVersion: 1, + Sender: s.serverMailboxID, + Recipient: env.Rpc.ReplyTo, + Headers: headers, + Body: body, + Rpc: &mailboxpb.RpcMeta{ + Kind: mailboxpb.RpcMeta_KIND_RESPONSE, + CorrelationId: env.Rpc.CorrelationId, + Service: env.Rpc.Service, + Method: env.Rpc.Method, + }, + } + + s.mb.send(responseEnv) +} + +// pushEvent injects a KIND_EVENT envelope into the client's mailbox. +func (s *testServer) pushEvent( + t *testing.T, recipientID, service, method string, + event proto.Message, +) { + + t.Helper() + + body, err := anypb.New(event) + require.NoError(t, err) + + env := &mailboxpb.Envelope{ + ProtocolVersion: 1, + Sender: s.serverMailboxID, + Recipient: recipientID, + Body: body, + Rpc: &mailboxpb.RpcMeta{ + Kind: mailboxpb.RpcMeta_KIND_EVENT, + Service: service, + Method: method, + ReplyTo: s.serverMailboxID, + }, + } + + status := s.mb.send(env) + require.True(t, status.Ok, "push event failed: %s", status.Message) +} + +// helloServer implements the generated HelloServiceMailboxServer interface. +type helloServer struct{} + +// SayHello echoes a greeting containing the caller's name. +func (s *helloServer) SayHello( + _ context.Context, req *hellotestpb.HelloRequest, +) (*hellotestpb.HelloResponse, error) { + + return &hellotestpb.HelloResponse{ + Greeting: fmt.Sprintf("Hello, %s!", req.Name), + }, nil +} + +// SayGoodbye echoes a farewell containing the caller's name. +func (s *helloServer) SayGoodbye( + _ context.Context, req *hellotestpb.GoodbyeRequest, +) (*hellotestpb.GoodbyeResponse, error) { + + return &hellotestpb.GoodbyeResponse{ + Farewell: fmt.Sprintf("Goodbye, %s!", req.Name), + }, nil +} + +// TestE2EUnaryRPC verifies a full unary request/response round-trip through +// the mailbox transport. The client sends SayHello via the generated +// HelloServiceMailboxClient, the server simulator dispatches it through +// ServeMux, and the client receives the typed response. +func TestE2EUnaryRPC(t *testing.T) { + t.Parallel() + + mb := newInMemoryMailbox() + store := newMemCheckpointStore() + + // Server side: register the HelloService handler. + server := newTestServer(mb, "server-1") + hellotestpb.RegisterHelloServiceMailboxServer( + server.mux, &helloServer{}, + ) + + cfg := newTestConnectorConfig(mb, store) + + runtime, err := NewRuntime(cfg) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + go server.run(ctx) + require.NoError(t, runtime.Start(ctx)) + defer runtime.Stop() + + // Issue a unary SayHello RPC via the generated mailbox client. + client := hellotestpb.NewHelloServiceMailboxClient(runtime.Unary()) + + resp, err := client.SayHello(ctx, &hellotestpb.HelloRequest{ + Name: "Alice", + }) + require.NoError(t, err) + require.Equal(t, "Hello, Alice!", resp.Greeting) + + // Also exercise SayGoodbye to confirm independent method routing. + goodbye, err := client.SayGoodbye(ctx, &hellotestpb.GoodbyeRequest{ + Name: "Bob", + }) + require.NoError(t, err) + require.Equal(t, "Goodbye, Bob!", goodbye.Farewell) +} + +// TestE2EServerPushEvent verifies that a server-pushed KIND_EVENT envelope +// is routed through the EventRouter to a registered greeting actor. The +// test creates a full actor system, registers a greeting actor under a +// ServiceKey, and verifies that the actor receives the deserialized +// helloStartedMsg via the FromProto interface. +func TestE2EServerPushEvent(t *testing.T) { + t.Parallel() + + mb := newInMemoryMailbox() + store := newMemCheckpointStore() + system := actor.NewActorSystem() + + // Register a greeting actor to receive server push events. + greetingKey := actor.NewServiceKey[*helloStartedMsg, struct{}]( + "greeting-actor", + ) + behavior := &greetingBehavior{ + received: make(chan *helloStartedMsg, 10), + } + actor.RegisterWithSystem( + system, "greeting-1", greetingKey, behavior, + ) + + // Wire up the EventRouter with the InboundServerMessage-based + // helper. NewEventRoute auto-generates the Adapt function from + // helloStartedMsg.FromProto. + router := NewEventRouter(system) + NewEventRoute( + router, InboundEventRouteConfig[*helloStartedMsg, struct{}]{ + Service: "hellotest.v1.HelloService", + Method: "HelloStarted", + Key: greetingKey, + NewEvent: func() proto.Message { + return &hellotestpb.HelloStartedEvent{} + }, + NewMsg: func() *helloStartedMsg { + return &helloStartedMsg{} + }, + }) + + server := newTestServer(mb, "server-1") + + cfg := newTestConnectorConfig(mb, store) + + cfg.Dispatchers = router.AsDispatcherMap() + + runtime, err := NewRuntime(cfg) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + go server.run(ctx) + require.NoError(t, runtime.Start(ctx)) + defer runtime.Stop() + + // Server pushes a HelloStartedEvent to the client. + server.pushEvent( + t, "client-1", + "hellotest.v1.HelloService", "HelloStarted", + &hellotestpb.HelloStartedEvent{SessionId: "session-42"}, + ) + + // Wait for the greeting actor to receive the dispatched message. + select { + case msg := <-behavior.received: + require.Equal(t, "session-42", msg.SessionID) + + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for greeting actor to receive event") + } +} + +// TestE2EClientFireAndForget verifies that the client can send a +// fire-and-forget event to the server via the DurableActor egress path. +// The test sends a JoinGreetingRequest through SendClientEventRequest, +// which is durably persisted in the actor's mailbox, serialized to proto +// via ToProto, and sent as a KIND_EVENT envelope to the server. +func TestE2EClientFireAndForget(t *testing.T) { + t.Parallel() + + mb := newInMemoryMailbox() + store := newMemCheckpointStore() + server := newTestServer(mb, "server-1") + + cfg := newTestConnectorConfig(mb, store) + + runtime, err := NewRuntime(cfg) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + go server.run(ctx) + require.NoError(t, runtime.Start(ctx)) + defer runtime.Stop() + + // Client sends a fire-and-forget event via the DurableActor. The + // message is persisted in the actor mailbox before the DurableActor + // processes it and sends it to the server via Edge.Send. + err = runtime.TellRef().Tell(ctx, &SendClientEventRequest{ + Message: &joinGreetingServerMsg{SessionID: "greeting-99"}, + }) + require.NoError(t, err) + + // Wait for the server to receive the event envelope. + select { + case env := <-server.received: + require.NotNil(t, env.Rpc) + require.Equal(t, + mailboxpb.RpcMeta_KIND_EVENT, env.Rpc.Kind, + ) + + // Unmarshal the body to verify the JoinGreetingRequest + // arrived intact. + var joinReq hellotestpb.JoinGreetingRequest + err := proto.Unmarshal(env.Body.Value, &joinReq) + require.NoError(t, err) + require.Equal(t, "greeting-99", joinReq.SessionId) + + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for server to receive event") + } +} + +// TestE2EUnaryAndPush verifies a combined scenario where the client +// issues a unary RPC and also receives server-push events in the same +// session. This exercises concurrent ingress (response delivery + +// event dispatch) with the EventRouter and UnaryFacade both active. +func TestE2EUnaryAndPush(t *testing.T) { + t.Parallel() + + mb := newInMemoryMailbox() + store := newMemCheckpointStore() + system := actor.NewActorSystem() + + // Register greeting actor. + greetingKey := actor.NewServiceKey[*helloStartedMsg, struct{}]( + "greeting-actor", + ) + behavior := &greetingBehavior{ + received: make(chan *helloStartedMsg, 10), + } + actor.RegisterWithSystem( + system, "greeting-1", greetingKey, behavior, + ) + + // Wire EventRouter. + router := NewEventRouter(system) + NewEventRoute( + router, InboundEventRouteConfig[*helloStartedMsg, struct{}]{ + Service: "hellotest.v1.HelloService", + Method: "HelloStarted", + Key: greetingKey, + NewEvent: func() proto.Message { + return &hellotestpb.HelloStartedEvent{} + }, + NewMsg: func() *helloStartedMsg { + return &helloStartedMsg{} + }, + }) + + // Server with HelloService handler. + server := newTestServer(mb, "server-1") + hellotestpb.RegisterHelloServiceMailboxServer( + server.mux, &helloServer{}, + ) + + cfg := newTestConnectorConfig(mb, store) + + cfg.Dispatchers = router.AsDispatcherMap() + + runtime, err := NewRuntime(cfg) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + go server.run(ctx) + require.NoError(t, runtime.Start(ctx)) + defer runtime.Stop() + + // Phase 1: Server pushes an event before the client sends an RPC. + server.pushEvent( + t, "client-1", + "hellotest.v1.HelloService", "HelloStarted", + &hellotestpb.HelloStartedEvent{SessionId: "pre-rpc"}, + ) + + select { + case msg := <-behavior.received: + require.Equal(t, "pre-rpc", msg.SessionID) + + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for pre-RPC push event") + } + + // Phase 2: Client issues a unary SayHello RPC. + client := hellotestpb.NewHelloServiceMailboxClient(runtime.Unary()) + + resp, err := client.SayHello(ctx, &hellotestpb.HelloRequest{ + Name: "Charlie", + }) + require.NoError(t, err) + require.Equal(t, "Hello, Charlie!", resp.Greeting) + + // Phase 3: Server pushes another event after the RPC. + server.pushEvent( + t, "client-1", + "hellotest.v1.HelloService", "HelloStarted", + &hellotestpb.HelloStartedEvent{SessionId: "post-rpc"}, + ) + + select { + case msg := <-behavior.received: + require.Equal(t, "post-rpc", msg.SessionID) + + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for post-RPC push event") + } +} + +// errHelloServer implements HelloServiceMailboxServer with handlers that +// return gRPC status errors so the error-header transport path can be +// exercised end-to-end. +type errHelloServer struct{} + +// SayHello returns a gRPC NotFound error for any request. +func (s *errHelloServer) SayHello( + _ context.Context, _ *hellotestpb.HelloRequest, +) (*hellotestpb.HelloResponse, error) { + + return nil, status.Errorf( + codes.NotFound, "user not found", + ) +} + +// SayGoodbye returns a gRPC InvalidArgument error for any request. +func (s *errHelloServer) SayGoodbye( + _ context.Context, _ *hellotestpb.GoodbyeRequest, +) (*hellotestpb.GoodbyeResponse, error) { + + return nil, status.Errorf( + codes.InvalidArgument, "name is required", + ) +} + +// TestE2EUnaryRPCError verifies that a server-side gRPC error is +// transported through the mailbox envelope headers and surfaced to the +// client as a proper gRPC status error via AwaitRPC. This exercises the +// EncodeErrorHeaders → HeaderGRPCStatusB64 → DecodeErrorHeaders path +// end-to-end. +func TestE2EUnaryRPCError(t *testing.T) { + t.Parallel() + + mb := newInMemoryMailbox() + store := newMemCheckpointStore() + + // Server side: register a handler that always returns errors. + server := newTestServer(mb, "server-1") + hellotestpb.RegisterHelloServiceMailboxServer( + server.mux, &errHelloServer{}, + ) + + cfg := newTestConnectorConfig(mb, store) + + runtime, err := NewRuntime(cfg) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + go server.run(ctx) + require.NoError(t, runtime.Start(ctx)) + defer runtime.Stop() + + client := hellotestpb.NewHelloServiceMailboxClient(runtime.Unary()) + + // SayHello should surface a NotFound gRPC error. + _, helloErr := client.SayHello(ctx, &hellotestpb.HelloRequest{ + Name: "Alice", + }) + require.Error(t, helloErr) + + st, ok := status.FromError(helloErr) + require.True(t, ok, "expected gRPC status error, got: %v", helloErr) + require.Equal(t, codes.NotFound, st.Code()) + require.Contains(t, st.Message(), "user not found") + + // SayGoodbye should surface an InvalidArgument gRPC error. + _, goodbyeErr := client.SayGoodbye( + ctx, &hellotestpb.GoodbyeRequest{Name: "Bob"}, + ) + require.Error(t, goodbyeErr) + + st, ok = status.FromError(goodbyeErr) + require.True(t, ok, "expected gRPC status error, got: %v", goodbyeErr) + require.Equal(t, codes.InvalidArgument, st.Code()) + require.Contains(t, st.Message(), "name is required") +} diff --git a/serverconn/restart_replay_test.go b/serverconn/restart_replay_test.go index 2a44bd1f5..1f1c7aa00 100644 --- a/serverconn/restart_replay_test.go +++ b/serverconn/restart_replay_test.go @@ -108,13 +108,9 @@ func TestEgress_RestartReplayPreservesStableIDs(t *testing.T) { edge := newFailFirstSendEdge(mb) store := newMemCheckpointStore() - cfg := DefaultConnectorConfig() + cfg := newTestConnectorConfig(mb, store) cfg.Edge = edge - cfg.Store = store cfg.Codec = NewServerConnCodec() - cfg.LocalMailboxID = "client-1" - cfg.RemoteMailboxID = "server-1" - cfg.ProtocolVersion = 1 // First actor instance fails once and nacks the message for retry. durable1 := newDurableConnectorForTest(cfg, 500*time.Millisecond) diff --git a/serverconn/runtime_test.go b/serverconn/runtime_test.go index 255fbe88f..9303a2a58 100644 --- a/serverconn/runtime_test.go +++ b/serverconn/runtime_test.go @@ -55,12 +55,7 @@ func TestNewRuntime_DefaultCodec(t *testing.T) { t.Parallel() mb := newInMemoryMailbox() - cfg := DefaultConnectorConfig() - cfg.Edge = &fakeMailboxServiceClient{mb: mb} - cfg.Store = newMemCheckpointStore() - cfg.LocalMailboxID = "client-1" - cfg.RemoteMailboxID = "server-1" - cfg.ProtocolVersion = 1 + cfg := newTestConnectorConfig(mb, newMemCheckpointStore()) cfg.Codec = nil runtime, err := NewRuntime(cfg) @@ -81,13 +76,7 @@ func TestRuntime_StartStop(t *testing.T) { t.Parallel() mb := newInMemoryMailbox() - - cfg := DefaultConnectorConfig() - cfg.Edge = &fakeMailboxServiceClient{mb: mb} - cfg.Store = newMemCheckpointStore() - cfg.LocalMailboxID = "client-1" - cfg.RemoteMailboxID = "server-1" - cfg.ProtocolVersion = 1 + cfg := newTestConnectorConfig(mb, newMemCheckpointStore()) cfg.PullWaitTimeout = 25 * time.Millisecond runtime, err := NewRuntime(cfg) diff --git a/serverconn/testutil_test.go b/serverconn/testutil_test.go index 0a3aec712..12a4bbd36 100644 --- a/serverconn/testutil_test.go +++ b/serverconn/testutil_test.go @@ -863,3 +863,24 @@ func (s *memCheckpointStore) CleanupExpired( // Compile-time check. var _ actor.DeliveryStore = (*memCheckpointStore)(nil) + +// newTestConnectorConfig returns a ConnectorConfig pre-populated with test +// defaults: an in-memory mailbox edge, the given store, fixed mailbox IDs +// ("client-1" / "server-1"), protocol version 1, and a short pull wait +// timeout suitable for test speed. Callers can override individual fields +// on the returned config before passing it to NewRuntime or +// NewServerConnectionActor. +func newTestConnectorConfig( + mb *inMemoryMailbox, store *memCheckpointStore, +) ConnectorConfig { + + cfg := DefaultConnectorConfig() + cfg.Edge = &fakeMailboxServiceClient{mb: mb} + cfg.Store = store + cfg.LocalMailboxID = "client-1" + cfg.RemoteMailboxID = "server-1" + cfg.ProtocolVersion = 1 + cfg.PullWaitTimeout = 50 * time.Millisecond + + return cfg +} From 8015a345be48d67d6fd29456db1745ee669c1261 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 20 Feb 2026 19:21:41 -0800 Subject: [PATCH 5/5] serverconn: check error headers in AwaitRPC before unmarshal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AwaitRPC was not calling DecodeErrorHeaders on the response envelope before attempting to unmarshal the body. When the server returned a gRPC error, the body contained an empty anypb.Any rather than a populated response. The proto unmarshal of empty bytes succeeded silently, yielding a zero-value response and nil error — swallowing the server-side failure completely. Insert a DecodeErrorHeaders check immediately after the nil body guard so server-side gRPC errors are surfaced to callers as proper status errors. --- serverconn/unary_facade.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/serverconn/unary_facade.go b/serverconn/unary_facade.go index 9195f942d..6d347dafe 100644 --- a/serverconn/unary_facade.go +++ b/serverconn/unary_facade.go @@ -179,6 +179,15 @@ func (f *UnaryFacade) AwaitRPC(ctx context.Context, ) } + // Check for a server-side gRPC status error before inspecting + // the body. This covers servers that set error headers with or + // without a populated body field. + if rpcErr := mailboxrpc.DecodeErrorHeaders( + env.Headers, + ); rpcErr != nil { + return rpcErr + } + if env.Body == nil { return fmt.Errorf("response envelope has nil body") }