serverconn: ack stale unary responses

darepod/server.go

@@ -2456,6 +2456,30 @@ func (s *Server) registerOOREventRoutes(router *serverconn.EventRouter) { //noli
 					"response envelope must be provided")
 			}
 
+			// This route is shared with live, in-memory unary
+			// callers of ListVTXOsByScripts. By the time a response
+			// reaches durable dispatch without the OOR metadata
+			// correlation prefix it is either a stale response from
+			// a prior process or a malformed metadata ID; in both
+			// cases we consume and ack it so ingress can advance
+			// rather than wedging on a response we cannot adapt.
+			if !oor.IsIncomingMetadataCorrelationID(
+				env.Rpc.CorrelationId,
+			) {
+
+				s.log.DebugS(context.Background(),
+					"Acking response without OOR "+
+						"correlation prefix",
+					slog.String(
+						"correlation_id",
+						env.Rpc.CorrelationId,
+					),
+					slog.String("service", env.Rpc.Service),
+					slog.String("method", env.Rpc.Method))
+
+				return nil, serverconn.ErrEnvelopeHandled
+			}
+
 			sessionID, err := oor.ParseIncomingMetadataCorrelationID( //nolint:ll
 				env.Rpc.CorrelationId,
 			)
@@ -2520,6 +2544,29 @@ func (s *Server) registerOOREventRoutes(router *serverconn.EventRouter) { //noli
 					"response envelope must be provided")
 			}
 
+			// As with the ListVTXOsByScripts route above, this
+			// route is shared with live, in-memory unary callers.
+			// A response that reaches durable dispatch without the
+			// OOR resolve correlation prefix is a stale or
+			// malformed response we cannot adapt; consume and ack
+			// it so ingress advances instead of wedging.
+			if !oor.IsIncomingResolveCorrelationID(
+				env.Rpc.CorrelationId,
+			) {
+
+				s.log.DebugS(context.Background(),
+					"Acking response without OOR "+
+						"correlation prefix",
+					slog.String(
+						"correlation_id",
+						env.Rpc.CorrelationId,
+					),
+					slog.String("service", env.Rpc.Service),
+					slog.String("method", env.Rpc.Method))
+
+				return nil, serverconn.ErrEnvelopeHandled
+			}
+
 			sessionID, recipientEventID, err :=
 				oor.ParseIncomingResolveCorrelationID(
 					env.Rpc.CorrelationId,

oor/incoming_adapter.go

@@ -48,15 +48,21 @@ func IncomingResolveCorrelationID(sessionID SessionID,
 		strconv.FormatUint(recipientEventID, 10)
 }
 
+// IsIncomingResolveCorrelationID returns true when correlationID belongs to a
+// durable incoming-transfer resolution query.
+func IsIncomingResolveCorrelationID(correlationID string) bool {
+	return len(correlationID) > len(incomingResolveCorrelationPrefix) &&
+		correlationID[:len(incomingResolveCorrelationPrefix)] ==
+			incomingResolveCorrelationPrefix
+}
+
 // ParseIncomingResolveCorrelationID decodes a durable incoming-transfer
 // resolution query correlation ID back into the OOR session ID and recipient
 // event ID.
 func ParseIncomingResolveCorrelationID(correlationID string) (SessionID, uint64,
 	error) {
 
-	if len(correlationID) <= len(incomingResolveCorrelationPrefix) ||
-		correlationID[:len(incomingResolveCorrelationPrefix)] !=
-			incomingResolveCorrelationPrefix {
+	if !IsIncomingResolveCorrelationID(correlationID) {
 		return SessionID{}, 0, fmt.Errorf("unexpected incoming "+
 			"resolve correlation id: %q", correlationID)
 	}

oor/incoming_adapter_test.go

@@ -0,0 +1,32 @@
+package oor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestIsIncomingResolveCorrelationID verifies only durable incoming-transfer
+// resolution query correlation IDs match the OOR resolve route prefix.
+func TestIsIncomingResolveCorrelationID(t *testing.T) {
+	t.Parallel()
+
+	var sessionID SessionID
+	sessionID[0] = 1
+
+	require.True(
+		t,
+		IsIncomingResolveCorrelationID(
+			IncomingResolveCorrelationID(sessionID, 7),
+		),
+	)
+	require.False(t, IsIncomingResolveCorrelationID(""))
+	require.False(
+		t, IsIncomingResolveCorrelationID("00aa8bfb11f09881bbd2"),
+	)
+	require.False(
+		t, IsIncomingResolveCorrelationID(
+			incomingResolveCorrelationPrefix,
+		),
+	)
+}

oor/incoming_metadata_query.go

@@ -40,14 +40,20 @@ func IncomingMetadataCorrelationID(sessionID SessionID) string {
 		chainhash.Hash(sessionID).String()
 }
 
+// IsIncomingMetadataCorrelationID returns true when correlationID belongs to a
+// durable incoming metadata query.
+func IsIncomingMetadataCorrelationID(correlationID string) bool {
+	return len(correlationID) > len(incomingMetadataCorrelationPrefix) &&
+		correlationID[:len(incomingMetadataCorrelationPrefix)] ==
+			incomingMetadataCorrelationPrefix
+}
+
 // ParseIncomingMetadataCorrelationID decodes a durable incoming metadata query
 // correlation ID back into the OOR session ID.
 func ParseIncomingMetadataCorrelationID(correlationID string) (SessionID,
 	error) {
 
-	if len(correlationID) <= len(incomingMetadataCorrelationPrefix) ||
-		correlationID[:len(incomingMetadataCorrelationPrefix)] !=
-			incomingMetadataCorrelationPrefix {
+	if !IsIncomingMetadataCorrelationID(correlationID) {
 		return SessionID{}, fmt.Errorf("unexpected incoming metadata "+
 			"correlation id: %q", correlationID)
 	}

oor/incoming_metadata_query_test.go

@@ -9,6 +9,31 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// TestIsIncomingMetadataCorrelationID verifies only durable incoming metadata
+// query correlation IDs match the OOR metadata route prefix.
+func TestIsIncomingMetadataCorrelationID(t *testing.T) {
+	t.Parallel()
+
+	var sessionID SessionID
+	sessionID[0] = 1
+
+	require.True(
+		t,
+		IsIncomingMetadataCorrelationID(
+			IncomingMetadataCorrelationID(sessionID),
+		),
+	)
+	require.False(t, IsIncomingMetadataCorrelationID(""))
+	require.False(
+		t, IsIncomingMetadataCorrelationID("00aa8bfb11f09881bbd2"),
+	)
+	require.False(
+		t, IsIncomingMetadataCorrelationID(
+			incomingMetadataCorrelationPrefix,
+		),
+	)
+}
+
 // TestIncomingMetadataFromRPCOperatorKey verifies incoming metadata parsing
 // preserves the per-VTXO operator key returned by the indexer.
 func TestIncomingMetadataFromRPCOperatorKey(t *testing.T) {

serverconn/connector_test.go

@@ -342,9 +342,6 @@ func TestIngress_ResponseDelivery(t *testing.T) {
 	)
 }
 
-// TestIngress_ResponseDispatchWithoutWaiter verifies that a KIND_RESPONSE
-// envelope without an in-memory unary waiter falls back to the durable
-// dispatcher map keyed by service and method.
 // TestIngress_ResponseDispatchHeaderOnlyError verifies that routed response
 // dispatch still reaches EventRouter handlers when the server encodes a gRPC
 // error in headers and intentionally omits the response body.
@@ -414,6 +411,74 @@ func TestIngress_ResponseDispatchHeaderOnlyError(t *testing.T) {
 	}
 }
 
+// TestIngressAckHandledResponseWithoutActorDelivery verifies a routed response
+// can be consumed by the route and acked without enqueueing an actor message.
+func TestIngressAckHandledResponseWithoutActorDelivery(t *testing.T) {
+	t.Parallel()
+
+	mb := newInMemoryMailbox()
+	store := newMemCheckpointStore()
+	system := actor.NewActorSystem()
+
+	routeKey := actor.NewServiceKey[*helloStartedMsg, struct{}](
+		"greeting-actor",
+	)
+	router := NewEventRouter(system)
+	AddEnvelopeRoute(
+		router, EnvelopeRouteConfig[*helloStartedMsg, struct{}]{
+			Service: "test.Svc",
+			Method:  "Unary",
+			NewEvent: func() proto.Message {
+				return &wrapperspb.StringValue{}
+			},
+			Key: routeKey,
+			Adapt: func(_ *mailboxpb.Envelope, _ proto.Message) (
+				*helloStartedMsg, error) {
+
+				return nil, ErrEnvelopeHandled
+			},
+		},
+	)
+
+	cfg := newTestConnectorConfig(mb, store)
+	cfg.Dispatchers = router.AsDispatcherMap()
+	cfg.RetryBaseDelay = 10 * time.Millisecond
+	cfg.RetryMaxDelay = 50 * time.Millisecond
+
+	connector := NewServerConnectionActor(cfg)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	require.NoError(t, connector.StartIngress(ctx))
+	defer connector.StopIngress()
+
+	body, err := anypb.New(wrapperspb.String("stale"))
+	require.NoError(t, err)
+
+	status := mb.send(&mailboxpb.Envelope{
+		ProtocolVersion: 1,
+		Sender:          "server-1",
+		Recipient:       "client-1",
+		Body:            body,
+		Rpc: &mailboxpb.RpcMeta{
+			Kind:          mailboxpb.RpcMeta_KIND_RESPONSE,
+			CorrelationId: "stale-corr",
+			Service:       "test.Svc",
+			Method:        "Unary",
+			ReplyTo:       "server-1",
+		},
+	})
+	require.True(t, status.Ok, "send response failed: %s", status.Message)
+
+	require.Eventually(t, func() bool {
+		return mb.getAckedUpTo("client-1") > 0
+	}, 5*time.Second, 10*time.Millisecond)
+}
+
+// TestIngress_ResponseDispatchWithoutWaiter verifies that a KIND_RESPONSE
+// envelope without an in-memory unary waiter falls back to the durable
+// dispatcher map keyed by service and method.
 func TestIngress_ResponseDispatchWithoutWaiter(t *testing.T) {
 	t.Parallel()
 

serverconn/event_router.go

@@ -2,6 +2,7 @@ package serverconn
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -11,6 +12,11 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// ErrEnvelopeHandled lets an envelope route acknowledge an envelope without
+// delivering an actor message. This is useful for shared RPC methods where a
+// stale response can be identified as unrelated to the durable route.
+var ErrEnvelopeHandled = errors.New("serverconn: envelope handled")
+
 // 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
@@ -145,7 +151,8 @@ type EnvelopeRouteConfig[M actor.Message, R any] struct {
 	Key actor.ServiceKey[M, R]
 
 	// Adapt converts the deserialized proto and envelope metadata into the
-	// actor message type M.
+	// actor message type M. Return ErrEnvelopeHandled when the envelope was
+	// intentionally consumed without actor delivery.
 	Adapt func(*mailboxpb.Envelope, proto.Message) (M, error)
 }
 
@@ -207,6 +214,10 @@ func AddEnvelopeRoute[M actor.Message, R any](r *EventRouter,
 
 		actorMsg, err := cfg.Adapt(env, event)
 		if err != nil {
+			if errors.Is(err, ErrEnvelopeHandled) {
+				return nil
+			}
+
 			return fmt.Errorf("adapt %s/%s event: %w", cfg.Service,
 				cfg.Method, err)
 		}

serverconn/event_router_test.go

@@ -8,13 +8,14 @@ import (
 	mailboxrpc "github.com/lightninglabs/darepo-client/mailbox/rpc"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
-// TestAddEnvelopeRoute_RejectsNilBodyWithoutEncodedError verifies that
+// TestAddEnvelopeRouteRejectsNilBodyWithoutEncodedError verifies that
 // AddEnvelopeRoute still fails closed on malformed routed responses that
 // omit both the proto body and the encoded gRPC status headers.
-func TestAddEnvelopeRoute_RejectsNilBodyWithoutEncodedError(t *testing.T) {
+func TestAddEnvelopeRouteRejectsNilBodyWithoutEncodedError(t *testing.T) {
 	t.Parallel()
 
 	router := NewEventRouter(actor.NewActorSystem())
@@ -60,3 +61,54 @@ func TestAddEnvelopeRoute_RejectsNilBodyWithoutEncodedError(t *testing.T) {
 	)
 	require.False(t, adaptCalled)
 }
+
+// TestAddEnvelopeRouteCanMarkEnvelopeHandled verifies that routes can consume
+// an envelope without forwarding a message to an actor mailbox.
+func TestAddEnvelopeRouteCanMarkEnvelopeHandled(t *testing.T) {
+	t.Parallel()
+
+	router := NewEventRouter(actor.NewActorSystem())
+	routeKey := actor.NewServiceKey[*helloStartedMsg, struct{}](
+		"test-route",
+	)
+
+	adaptCalled := false
+
+	AddEnvelopeRoute(
+		router, EnvelopeRouteConfig[*helloStartedMsg, struct{}]{
+			Service: "test.Svc",
+			Method:  "Unary",
+			NewEvent: func() proto.Message {
+				return &wrapperspb.StringValue{}
+			},
+			Key: routeKey,
+			Adapt: func(_ *mailboxpb.Envelope, _ proto.Message) (
+				*helloStartedMsg, error) {
+
+				adaptCalled = true
+
+				return nil, ErrEnvelopeHandled
+			},
+		},
+	)
+
+	dispatcher := router.AsDispatcherMap()[mailboxrpc.ServiceMethod{
+		Service: "test.Svc",
+		Method:  "Unary",
+	}]
+
+	body, err := anypb.New(wrapperspb.String("stale"))
+	require.NoError(t, err)
+
+	err = dispatcher(t.Context(), &mailboxpb.Envelope{
+		Body: body,
+		Rpc: &mailboxpb.RpcMeta{
+			Kind:          mailboxpb.RpcMeta_KIND_RESPONSE,
+			CorrelationId: "corr-1",
+			Service:       "test.Svc",
+			Method:        "Unary",
+		},
+	})
+	require.NoError(t, err)
+	require.True(t, adaptCalled)
+}