diff --git a/mailbox/conn/response_registry.go b/mailbox/conn/response_registry.go
index 286078b0a..64055c8a5 100644
--- a/mailbox/conn/response_registry.go
+++ b/mailbox/conn/response_registry.go
@@ -147,6 +147,24 @@ func (r *ResponseRegistry) RemoveWaiter(id CorrelationID) {
}
}
+// HasWaiter reports whether an active in-memory waiter is currently registered
+// for correlation ID id. It lets the ingress loop classify a KIND_RESPONSE at
+// split time: a response with a live waiter can be delivered on the fast
+// pre-transaction path, while a response without one must fold into the durable
+// dispatch transaction so its enqueue commits atomically with the cursor. Stale
+// waiters are pruned before the check so an expired entry never masquerades as
+// a live one.
+func (r *ResponseRegistry) HasWaiter(id CorrelationID) bool {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ r.pruneStaleLocked(time.Now())
+
+ _, ok := r.waiters[id]
+
+ return ok
+}
+
// RemovePending drops any buffered early response for the correlation ID.
func (r *ResponseRegistry) RemovePending(id CorrelationID) {
r.mu.Lock()
diff --git a/mailbox/conn/response_registry_test.go b/mailbox/conn/response_registry_test.go
index 3a98b0fa5..f01d0dbc1 100644
--- a/mailbox/conn/response_registry_test.go
+++ b/mailbox/conn/response_registry_test.go
@@ -124,3 +124,48 @@ func TestResponseRegistry_DeliverNilReturnsFalse(t *testing.T) {
require.Equal(t, DeliveryDropped, registry.DeliverResponse("any",
nil))
}
+
+// TestResponseRegistryHasWaiterTracksRegistration verifies HasWaiter reflects
+// the live-waiter state the ingress split path relies on: false before
+// registration, true while a waiter is registered, and false again once the
+// waiter is removed or a buffered (waiterless) response is delivered.
+func TestResponseRegistryHasWaiterTracksRegistration(t *testing.T) {
+ t.Parallel()
+
+ registry := NewResponseRegistry(time.Minute)
+ id := CorrelationID("corr-has-waiter")
+
+ // No waiter registered yet.
+ require.False(t, registry.HasWaiter(id))
+
+ // A buffered early response does not count as a live waiter, so the
+ // ingress loop folds it into the durable transaction.
+ registry.DeliverResponse(id, &mailboxpb.Envelope{EventSeq: 1})
+ require.False(t, registry.HasWaiter(id))
+
+ // Registering a waiter flips it true; this drains the buffered
+ // response into the promise.
+ registry.RegisterWaiter(id)
+ require.True(t, registry.HasWaiter(id))
+
+ // Removing the waiter flips it back to false.
+ registry.RemoveWaiter(id)
+ require.False(t, registry.HasWaiter(id))
+}
+
+// TestResponseRegistryHasWaiterPrunesStale verifies HasWaiter prunes an expired
+// waiter before answering, so a TTL-lapsed entry never masquerades as live and
+// misroutes a response onto the fast path.
+func TestResponseRegistryHasWaiterPrunesStale(t *testing.T) {
+ t.Parallel()
+
+ registry := NewResponseRegistry(5 * time.Millisecond)
+ id := CorrelationID("corr-stale")
+
+ registry.RegisterWaiter(id)
+ require.True(t, registry.HasWaiter(id))
+
+ time.Sleep(10 * time.Millisecond)
+
+ require.False(t, registry.HasWaiter(id))
+}
diff --git a/p-models/durableactor/CLAUDE.md b/p-models/durableactor/CLAUDE.md
index 66a587b63..eb090b9c9 100644
--- a/p-models/durableactor/CLAUDE.md
+++ b/p-models/durableactor/CLAUDE.md
@@ -10,8 +10,15 @@ exactly-once effect application under lease-expiry-during-IO).
- `infra.pproj` — P project for durable actor infrastructure checks.
- `src/mailbox_fifo.p` — ideal mailbox spec plus claim-ordering profiles.
+- `src/ingress_fold.p` — connection-actor ingress cursor spec: the persisted
+ PullCursor must never cover an envelope whose local enqueue did not commit
+ (the transactional dispatch fold makes batch enqueues + cursor one atomic
+ commit).
- `test/mailbox_fifo_test.p` — green conformance tests and separate
counterexample tests.
+- `test/ingress_fold_test.p` — green atomic-fold drain test plus the two
+ cursor-loss counterexamples (eager in-memory cursor after rollback;
+ checkpoint commit ordered before the enqueue commits).
- `traces/*.json` — concrete scenarios replayed by the Go bridge.
- `bridge/` — Go conformance harness against the real `db/actordelivery`
SQLite store and claim SQL.
@@ -29,6 +36,9 @@ exactly-once effect application under lease-expiry-during-IO).
| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcMailboxStageCommitExactlyOnce` | Run the green Stage-then-Commit replay-safety test |
| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcMailboxStagedDoubleBroadcastCounterexample` | Demonstrate the unstable-broadcast double-broadcast bug |
| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcMailboxStaleStageRegressesCounterexample` | Demonstrate the unfenced-stage checkpoint regression bug |
+| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcIngressFoldNoLoss` | Run the green transactional ingress-fold no-loss test |
+| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcIngressEagerCursorCounterexample` | Demonstrate the eager-cursor-after-rollback message loss |
+| `p check PGenerated/PChecker/net8.0/MailboxInfraModels.dll --testcase tcIngressCheckpointFirstCounterexample` | Demonstrate the checkpoint-before-enqueue message loss |
| `go test ./p-models/durableactor/bridge` | Replay traces against Go |
## Modeling Guidance
diff --git a/p-models/durableactor/infra.pproj b/p-models/durableactor/infra.pproj
index 0e7d59f0e..b2b4a7296 100644
--- a/p-models/durableactor/infra.pproj
+++ b/p-models/durableactor/infra.pproj
@@ -2,7 +2,9 @@
MailboxInfraModels
src/mailbox_fifo.p
+src/ingress_fold.p
test/mailbox_fifo_test.p
+test/ingress_fold_test.p
PGenerated
diff --git a/p-models/durableactor/src/ingress_fold.p b/p-models/durableactor/src/ingress_fold.p
new file mode 100644
index 000000000..fdc0d844d
--- /dev/null
+++ b/p-models/durableactor/src/ingress_fold.p
@@ -0,0 +1,78 @@
+// ingress_fold.p - Connection-actor ingress cursor specification.
+//
+// Models the pull-dispatch-checkpoint loop run by the client's
+// ServerConnectionActor and the operator's ClientConnectionActor against a
+// remote mailbox. The loop pulls a batch of envelopes (identified by
+// monotonically increasing event sequence numbers), enqueues each into a
+// local durable mailbox, and persists the advanced PullCursor (exclusive
+// next-pull position) in an AckState checkpoint.
+//
+// Distributed-systems contract:
+//
+// * The persisted cursor must never cover an envelope whose local enqueue
+// did not durably commit: a cursor that runs ahead of the enqueues
+// silently skips those envelopes forever (message loss).
+// * The transactional fold makes the batch enqueues and the cursor
+// advance ONE atomic commit, so a dispatch failure or crash rolls back
+// both and the batch is re-pulled intact.
+// * Redelivery after rollback or crash is safe (at-least-once): the
+// rollback erased the partial enqueues, so the retry starts clean.
+//
+// The counterexample drivers in the test file demonstrate the two ways an
+// implementation can break the contract: advancing the in-memory cursor
+// past a rolled-back commit, and checkpointing the cursor in a separate
+// commit issued before the enqueues land.
+//
+// Only WAITER-BACKED response envelopes are out of scope: a KIND_RESPONSE
+// with a live in-memory unary waiter delivers to that waiter at most once,
+// before and outside the dispatch transaction (it cannot roll back, and
+// gating it on the writer lock starves RPC callers), so the cursor may
+// cover it without a durable enqueue. The implementation classifies these
+// at split time via the waiter registry.
+//
+// A response with NO live waiter is NOT out of scope: it falls back to
+// durable route dispatch and folds into the SAME transaction as requests
+// and events, so its enqueue commits atomically with the cursor exactly
+// like any other durable envelope. This model therefore drives all such
+// durable dispatches uniformly — a waiterless response is indistinguishable
+// from a request/event here, and the no-loss property below guards it the
+// same way. (Were the implementation to instead enqueue a waiterless
+// response OUTSIDE the fold, ahead of the cursor commit, that would be the
+// eager-cursor counterexample shape; the split keeps it inside the fold.)
+
+// eIngressEnvelopeCommitted announces that the local enqueue of the given
+// envelope sequence number durably committed, namespaced by the driver
+// machine so concurrent test machines do not interfere.
+event eIngressEnvelopeCommitted: (machine, int);
+
+// eIngressCursorPersisted announces that an AckState checkpoint carrying
+// the given exclusive PullCursor durably committed.
+event eIngressCursorPersisted: (machine, int);
+
+// IngressCursorCoversOnlyCommittedEnvelopes is the no-message-loss safety
+// property: whenever a checkpoint persists cursor c, every envelope with
+// sequence number below c must have a committed local enqueue. A violation
+// means the loop would resume past envelopes that were never delivered.
+spec IngressCursorCoversOnlyCommittedEnvelopes observes
+ eIngressEnvelopeCommitted, eIngressCursorPersisted {
+
+ var committed: map[(machine, int), bool];
+
+ start state Monitoring {
+ on eIngressEnvelopeCommitted do (enq: (machine, int)) {
+ committed[(enq.0, enq.1)] = true;
+ }
+
+ on eIngressCursorPersisted do (cp: (machine, int)) {
+ var s: int;
+
+ s = 1;
+ while (s < cp.1) {
+ assert (cp.0, s) in committed,
+ "persisted cursor covers an envelope whose local "+
+ "enqueue never committed (message loss)";
+ s = s + 1;
+ }
+ }
+ }
+}
diff --git a/p-models/durableactor/test/ingress_fold_test.p b/p-models/durableactor/test/ingress_fold_test.p
new file mode 100644
index 000000000..ad31435ff
--- /dev/null
+++ b/p-models/durableactor/test/ingress_fold_test.p
@@ -0,0 +1,142 @@
+// ingress_fold_test.p - Ingress cursor specification tests.
+
+// TestIngressFold_AtomicBatchDispatchNoLoss drives the transactional
+// ingress design through nondeterministic batch sizes, injected commit
+// failures (rollbacks), and crash-restarts. The fold's contract is that the
+// batch enqueues and the cursor advance are announced as one atomic commit,
+// a failed commit announces nothing and resets the in-memory cursor to the
+// durable one, and a crash reloads the durable cursor. The run must drain
+// every envelope with no loss.
+machine TestIngressFold_AtomicBatchDispatchNoLoss {
+ start state Init {
+ entry {
+ var total: int;
+ var durableCursor: int;
+ var memCursor: int;
+ var batchEnd: int;
+ var faultBudget: int;
+ var delivered: map[int, bool];
+ var s: int;
+
+ total = 3;
+ durableCursor = 1;
+ memCursor = 1;
+ faultBudget = 2;
+
+ while (durableCursor <= total) {
+ // Pull a batch of one or two envelopes starting at the
+ // in-memory cursor (which always matches the durable one
+ // at the top of a healthy iteration).
+ memCursor = durableCursor;
+ batchEnd = memCursor;
+ if ($ && batchEnd < total) {
+ batchEnd = batchEnd + 1;
+ }
+
+ if (faultBudget > 0 && $) {
+ // Commit failure: the transaction rolls back, so
+ // neither the enqueues nor the cursor advance are
+ // announced, and the in-memory cursor resets to the
+ // durable position for the re-pull.
+ faultBudget = faultBudget - 1;
+ memCursor = durableCursor;
+ } else if (faultBudget > 0 && $) {
+ // Crash before the commit: restart reloads the
+ // durable cursor; nothing was announced.
+ faultBudget = faultBudget - 1;
+ memCursor = durableCursor;
+ } else {
+ // Atomic commit: every enqueue in the batch and the
+ // advanced cursor become durable together.
+ s = memCursor;
+ while (s <= batchEnd) {
+ announce eIngressEnvelopeCommitted, (this, s);
+ delivered[s] = true;
+ s = s + 1;
+ }
+
+ durableCursor = batchEnd + 1;
+ announce eIngressCursorPersisted,
+ (this, durableCursor);
+ }
+ }
+
+ // Liveness-by-construction: the loop only exits once the
+ // durable cursor passed every envelope, and the monitor has
+ // checked that the cursor never covered an undelivered one.
+ s = 1;
+ while (s <= total) {
+ assert s in delivered,
+ "drained ingress loop left an envelope undelivered";
+ s = s + 1;
+ }
+
+ goto Done;
+ }
+ }
+
+ state Done {}
+}
+
+// TestIngressFold_EagerCursorLossCounterexample reproduces the bug the fold
+// must avoid: the loop mutates its cursor state inside the transaction
+// closure and keeps the advanced value after the commit ROLLS BACK. The
+// next successful batch then persists a cursor that covers the rolled-back
+// envelopes, which the monitor flags as message loss.
+machine TestIngressFold_EagerCursorLossCounterexample {
+ start state Init {
+ entry {
+ // Batch [1,2] dispatches, but the commit fails and rolls the
+ // enqueues back, so neither enqueue is announced. The buggy
+ // loop advanced its in-memory cursor to 3 anyway.
+ //
+ // Batch [3] then commits fine, persisting cursor 4 — which
+ // claims envelopes 1 and 2 were delivered. They never were.
+ announce eIngressEnvelopeCommitted, (this, 3);
+ announce eIngressCursorPersisted, (this, 4);
+
+ goto Done;
+ }
+ }
+
+ state Done {}
+}
+
+// TestIngressFold_CheckpointBeforeEnqueueCounterexample reproduces the
+// other ordering bug: persisting the cursor checkpoint in its own commit
+// BEFORE the batch enqueues land. A crash between the two commits resumes
+// past envelopes that never reached the local mailbox.
+machine TestIngressFold_CheckpointBeforeEnqueueCounterexample {
+ start state Init {
+ entry {
+ // Cursor checkpoint for batch [1,2] commits first ...
+ announce eIngressCursorPersisted, (this, 3);
+
+ // ... and the process crashes before the enqueue commit; the
+ // envelopes are never announced. The monitor flags the
+ // persisted cursor at announce time.
+ goto Done;
+ }
+ }
+
+ state Done {}
+}
+
+test tcIngressFoldNoLoss [main=TestIngressFold_AtomicBatchDispatchNoLoss]:
+ assert IngressCursorCoversOnlyCommittedEnvelopes in
+ { TestIngressFold_AtomicBatchDispatchNoLoss };
+
+// tcIngressEagerCursorCounterexample runs the rolled-back-commit scenario
+// with the buggy eager in-memory cursor. It is expected to find a bug.
+test tcIngressEagerCursorCounterexample
+ [main=TestIngressFold_EagerCursorLossCounterexample]:
+ assert IngressCursorCoversOnlyCommittedEnvelopes in
+ { TestIngressFold_EagerCursorLossCounterexample };
+
+// tcIngressCheckpointFirstCounterexample runs the split-commit scenario
+// where the checkpoint lands before the enqueues. It is expected to find a
+// bug.
+test tcIngressCheckpointFirstCounterexample
+ [main=TestIngressFold_CheckpointBeforeEnqueueCounterexample]:
+ assert IngressCursorCoversOnlyCommittedEnvelopes in
+ { TestIngressFold_CheckpointBeforeEnqueueCounterexample };
diff --git a/serverconn/actor.go b/serverconn/actor.go
index 0b5a18f35..9af72673d 100644
--- a/serverconn/actor.go
+++ b/serverconn/actor.go
@@ -1128,6 +1128,15 @@ func (a *ServerConnectionActor) removePendingResponse(id CorrelationID) {
a.responseRegistry.RemovePending(id)
}
+// hasResponseWaiter reports whether an active in-memory waiter is registered
+// for the correlation ID. The ingress loop uses this to classify a
+// KIND_RESPONSE at split time: only responses with a live waiter take the fast
+// pre-transaction delivery path; everything else folds into the durable
+// dispatch transaction.
+func (a *ServerConnectionActor) hasResponseWaiter(id CorrelationID) bool {
+ return a.responseRegistry.HasWaiter(id)
+}
+
// deliverResponse looks up a waiter by correlation ID and delivers the
// envelope. If no waiter exists yet, the response is buffered so a later
// AwaitRPC call can still observe it.
diff --git a/serverconn/connector_test.go b/serverconn/connector_test.go
index b2b0bf363..1962ce40e 100644
--- a/serverconn/connector_test.go
+++ b/serverconn/connector_test.go
@@ -1,6 +1,7 @@
package serverconn
import (
+ "bytes"
"context"
"fmt"
"sync"
@@ -757,12 +758,21 @@ func TestEventRoutingMetadata(t *testing.T) {
require.Equal(t, "", method)
}
-// TestIngress_PartialDispatch_NoDuplicateRedelivery verifies that when
-// a batch dispatch fails mid-way, the already-dispatched envelopes are
-// not re-dispatched on the next loop iteration. This is a regression test
-// for the off-by-one where the inclusive event_seq returned on the error
-// path was used directly as PullCursor, causing the last committed
-// envelope to be re-pulled and re-dispatched.
+// TestIngress_PartialDispatch_NoDuplicateRedelivery verifies the cursor
+// half of the transactional dispatch contract for a batch that fails
+// mid-way: the PullCursor does not advance on the failed commit, so the
+// whole batch is re-dispatched on retry, and once a batch commits none of
+// its envelopes are ever dispatched again. The post-commit stability check
+// is the regression guard for the original off-by-one where the inclusive
+// event_seq returned on the error path was used directly as PullCursor,
+// re-pulling the last committed envelope.
+//
+// Scope note: the in-memory store's ExecTx has no real savepoint (it runs
+// the closure against itself), and these dispatchers are counters that
+// never call EnqueueMessage, so this test asserts cursor non-advancement
+// only — NOT that a rolled-back batch's enqueues are physically erased.
+// The enqueue/cursor atomicity itself is proven by the ingress_fold.p
+// P-model (tcIngressFoldNoLoss and the two counterexample cases).
func TestIngress_PartialDispatch_NoDuplicateRedelivery(t *testing.T) {
t.Parallel()
@@ -789,10 +799,10 @@ func TestIngress_PartialDispatch_NoDuplicateRedelivery(t *testing.T) {
dispatchCountsMu.Unlock()
// Fail on the second envelope in the first batch.
- // The first envelope (count==1) succeeds, and the
- // second (count==2) fails. On retry, we expect the
- // second to be dispatched (count==3) but NOT the
- // first again.
+ // The first envelope (count==1) succeeds and the
+ // second (count==2) fails, rolling back the whole
+ // batch. The retry re-dispatches both envelopes
+ // (counts 3 and 4) and commits.
if count == 2 {
return &statusError{
Op: "dispatch",
@@ -829,24 +839,91 @@ func TestIngress_PartialDispatch_NoDuplicateRedelivery(t *testing.T) {
return dispatchCounts[2] >= 2
}, 5*time.Second, 10*time.Millisecond)
+ // Give the loop room to misbehave: a cursor bug would re-pull and
+ // re-dispatch committed envelopes on subsequent iterations.
+ time.Sleep(200 * time.Millisecond)
+
dispatchCountsMu.Lock()
defer dispatchCountsMu.Unlock()
- // The first envelope (event_seq=1) must have been dispatched
- // exactly once — not re-dispatched after the partial failure.
+ // Both envelopes dispatch exactly twice: once in the first batch
+ // whose commit failed (so the cursor did not advance) and once in
+ // the committed retry. Anything beyond two means the loop re-pulled
+ // past a committed checkpoint.
require.Equal(
- t, 1, dispatchCounts[1],
- "first envelope should be dispatched exactly once",
+ t, 2, dispatchCounts[1], "first envelope should be "+
+ "dispatched exactly twice (rolled-back batch + retry)",
)
-
- // The second envelope (event_seq=2) should be dispatched
- // exactly twice: once failed, once succeeded on retry.
require.Equal(
t, 2, dispatchCounts[2], "second envelope should be "+
"dispatched exactly twice (1 fail + 1 retry)",
)
}
+// TestIngress_IdleFlushPersistsAckWatermark exercises the ackDirty
+// idle-flush convergence path on the transactional store. After an event
+// dispatches and the loop acks the remote, the advanced ack watermark is
+// NOT checkpointed inline (it rides the next dispatch checkpoint); on a
+// connection that then goes quiet the empty long-poll branch must flush it.
+// This guards against a regression that dropped the idle flush or never
+// cleared ackDirty, which would leave the loop re-acking from a stale
+// AckCommittedTo on every restart.
+func TestIngress_IdleFlushPersistsAckWatermark(t *testing.T) {
+ t.Parallel()
+
+ dispatchers := map[mailboxrpc.ServiceMethod]EnvelopeDispatcher{
+ {
+ Service: "test.Svc",
+ Method: "DoThing",
+ }: func(
+ ctx context.Context,
+ env *mailboxpb.Envelope,
+ ) error {
+
+ return nil
+ },
+ }
+
+ actor, mb, store := newTestConnector(t, dispatchers)
+
+ // Inject a single event. Once it dispatches and the loop acks the
+ // remote, the advanced watermark stays in memory (ackDirty) until the
+ // next empty long-poll flushes it durably.
+ sendEventToMailbox(t, mb, "client-1", "test.Svc", "DoThing")
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ require.NoError(t, actor.StartIngress(ctx))
+ defer actor.StopIngress()
+
+ // Wait for the remote ack, which advances AckCommittedTo in memory and
+ // sets ackDirty without an inline checkpoint on the transactional path.
+ require.Eventually(t, func() bool {
+ return mb.getAckedUpTo("client-1") > 0
+ }, 5*time.Second, 10*time.Millisecond)
+
+ // The idle flush runs on a subsequent empty poll. Wait until the
+ // persisted checkpoint decodes to the acked watermark, proving the
+ // flush converged the durable state.
+ actorID := DurableActorID("client-1")
+ require.Eventually(t, func() bool {
+ cp, err := store.LoadCheckpoint(t.Context(), actorID)
+ if err != nil || cp == nil {
+ return false
+ }
+
+ var persisted AckState
+ if err := persisted.Decode(
+ bytes.NewReader(cp.StateData),
+ ); err != nil {
+ return false
+ }
+
+ return persisted.AckCommittedTo >= 1
+ }, 5*time.Second, 10*time.Millisecond)
+}
+
// Backoff formula tests now live in serverconn/mailboxpull/pull_test.go
// (TestRetryDelayClampsToMax, TestRetryDelayUsesDefaults) -- the formula
// itself moved to that subpackage so the SDK pull loop and this actor's
diff --git a/serverconn/ingress.go b/serverconn/ingress.go
index d20d79af7..64b92c31d 100644
--- a/serverconn/ingress.go
+++ b/serverconn/ingress.go
@@ -36,6 +36,14 @@ func (a *ServerConnectionActor) ingressLoop(ctx context.Context,
var failCount int
+ // When the delivery store supports transactions, each pulled batch is
+ // dispatched and checkpointed in ONE write transaction below. The
+ // ack watermark then rides along with the next dispatch checkpoint
+ // instead of paying its own commit; ackDirty tracks the in-memory
+ // advance until some checkpoint persists it.
+ txStore, txOK := a.cfg.Store.(actor.TxAwareDeliveryStore)
+ var ackDirty bool
+
for {
select {
case <-ctx.Done():
@@ -73,7 +81,18 @@ func (a *ServerConnectionActor) ingressLoop(ctx context.Context,
state.AdvanceAck()
- if err := a.saveCheckpoint(ctx, state); err != nil {
+ // On the transactional path the advanced watermark is
+ // persisted by the next dispatch checkpoint (or the
+ // idle flush below); losing it to a crash only costs
+ // one redundant idempotent AckUpTo on restart. The
+ // legacy path keeps the immediate checkpoint.
+ if txOK {
+ ackDirty = true
+ failCount = 0
+ } else if err := a.saveCheckpoint(
+ ctx, state,
+ ); err != nil {
+
a.log.WarnS(
ctx,
"Failed to save checkpoint after ack",
@@ -87,9 +106,9 @@ func (a *ServerConnectionActor) ingressLoop(ctx context.Context,
a.sleepBackoff(ctx, &failCount)
continue
+ } else {
+ failCount = 0
}
-
- failCount = 0
}
// Step 2: Pull a batch of envelopes from the remote mailbox.
@@ -114,9 +133,34 @@ func (a *ServerConnectionActor) ingressLoop(ctx context.Context,
}
if len(envelopes) == 0 {
- // Long-poll returned empty. Reset fail count and loop
- // again immediately — the long-poll timeout already
- // provides the delay.
+ // Long-poll returned empty. Flush a dirty ack
+ // watermark while the connection is idle so a
+ // restart does not re-ack forever.
+ if ackDirty {
+ if err := a.saveCheckpoint(
+ ctx, state,
+ ); err != nil {
+
+ // Back off on a failing checkpoint
+ // store rather than retrying at the
+ // bare long-poll cadence, mirroring the
+ // ack-path policy above. ackDirty stays
+ // set so the next attempt re-flushes.
+ a.log.WarnS(ctx,
+ "Failed to flush ack "+
+ "checkpoint while idle",
+ err)
+
+ a.sleepBackoff(ctx, &failCount)
+
+ continue
+ }
+
+ ackDirty = false
+ }
+
+ // Reset fail count and loop again immediately — the
+ // long-poll timeout already provides the delay.
failCount = 0
continue
@@ -129,9 +173,38 @@ func (a *ServerConnectionActor) ingressLoop(ctx context.Context,
slog.Uint64("next_cursor", nextCursor),
)
- // Step 3: Dispatch the batch. On partial failure, the
- // committed cursor reflects only the successfully dispatched
- // portion.
+ // Step 3 (transactional path): deliver in-memory responses
+ // outside the transaction, then fold the durable dispatches
+ // and the advanced watermark into one commit.
+ if txOK {
+ newState, foldErr := a.runFoldedDispatch(
+ ctx, txStore, envelopes, nextCursor, state,
+ )
+ if foldErr != nil {
+ a.log.WarnS(ctx,
+ "Transactional dispatch failed",
+ foldErr,
+ slog.Uint64(
+ "cursor", state.PullCursor,
+ ))
+
+ a.sleepBackoff(ctx, &failCount)
+
+ continue
+ }
+
+ // The commit covered the dispatch watermark and any
+ // pending ack advance together.
+ state = newState
+ ackDirty = false
+ failCount = 0
+
+ continue
+ }
+
+ // Step 3 (legacy path): dispatch the batch. On partial
+ // failure, the committed cursor reflects only the
+ // successfully dispatched portion.
committedCursor, dispatchErr := a.dispatchBatch(
ctx, envelopes, nextCursor,
)
@@ -447,11 +520,187 @@ func (a *ServerConnectionActor) loadCheckpoint(ctx context.Context) (AckState,
return state, nil
}
+// runFoldedDispatch delivers a pulled batch's waiter-backed response
+// envelopes BEFORE the write transaction, then folds the durable dispatches
+// and the advanced AckState checkpoint into ONE commit. Waiter delivery is
+// in-memory and at-most-once, cannot be rolled back, and must never wait in
+// the single-writer queue: unary callers sit blocked on these with RPC
+// deadlines, so gating them on the writer lock turns write contention into
+// payment-wide timeout collapse. Every durable dispatcher Tell joins the
+// ambient transaction via the context (DurableMailbox.Send flows it into
+// EnqueueMessage), so a batch of k durable envelopes costs one commit
+// instead of k+1 and the cursor can never run ahead of the enqueues: any
+// failure rolls back both, leaves the returned state untouched, and the
+// batch is re-pulled intact.
+//
+// The split-time waiter peek is only a hint: a waiter can vanish (RPC
+// deadline cancel or TTL prune) between the peek and the actual delivery
+// below. The pre-transaction step therefore delivers to LIVE waiters only
+// and folds any straggler whose waiter disappeared back into the durable
+// transaction, so a durable response enqueue never commits outside the
+// cursor fold even if the peek was stale.
+func (a *ServerConnectionActor) runFoldedDispatch(ctx context.Context,
+ txStore actor.TxAwareDeliveryStore, envelopes []*mailboxpb.Envelope,
+ nextCursor uint64, state AckState) (AckState, error) {
+
+ responses, durables := splitIngressEnvelopes(
+ envelopes, a.hasResponseWaiter,
+ )
+
+ // Deliver the waiter-backed responses to their live waiters outside
+ // the transaction. Any whose waiter vanished since the split peek come
+ // back as stragglers and fold into the durable batch in event_seq
+ // order, so their enqueue commits inside the cursor fold, never ahead
+ // of it.
+ if stragglers := a.deliverWaiterResponses(
+ responses,
+ ); len(stragglers) > 0 {
+
+ durables = mergeEnvelopesByEventSeq(durables, stragglers)
+ }
+
+ newState := state
+ err := txStore.ExecTx(ctx, false, func(txCtx context.Context,
+ store actor.DeliveryStore) error {
+
+ if len(durables) > 0 {
+ _, dispatchErr := a.dispatchBatch(
+ txCtx, durables, nextCursor,
+ )
+ if dispatchErr != nil {
+ return dispatchErr
+ }
+ }
+
+ newState.AdvanceDispatch(nextCursor)
+ newState.PullCursor = nextCursor
+
+ return a.saveCheckpointTo(txCtx, store, newState)
+ })
+ if err != nil {
+ return state, err
+ }
+
+ return newState, nil
+}
+
+// splitIngressEnvelopes partitions a pulled batch into in-memory response
+// envelopes and durable dispatch envelopes. A KIND_RESPONSE only takes the
+// pre-transaction path when an active in-memory waiter is registered for its
+// correlation ID, as reported by hasWaiter: those callers sit blocked on an
+// RPC deadline and must never queue behind the database writer lock. A
+// KIND_RESPONSE with no live waiter would otherwise fall through to the durable
+// dispatch table; folding it into the transaction alongside requests and events
+// keeps event_seq order on the target actor lane and ties its enqueue to the
+// cursor commit, so a rolled-back batch never re-delivers it. Everything else
+// (requests, events, and malformed or correlation-less envelopes, which the
+// dispatch loop skip-warns) folds into the transaction too.
+func splitIngressEnvelopes(envelopes []*mailboxpb.Envelope,
+ hasWaiter func(CorrelationID) bool) ([]*mailboxpb.Envelope,
+ []*mailboxpb.Envelope) {
+
+ var responses, durables []*mailboxpb.Envelope
+ for _, env := range envelopes {
+ isResponse := env.Rpc != nil &&
+ env.Rpc.Kind == mailboxpb.RpcMeta_KIND_RESPONSE
+
+ // Route a response to the fast pre-transaction path only when a
+ // live waiter is registered for its correlation ID; otherwise
+ // it folds into the durable transaction with the rest of the
+ // batch.
+ corrID := CorrelationID("")
+ if isResponse {
+ corrID = CorrelationID(env.Rpc.CorrelationId)
+ }
+ if isResponse && corrID != "" && hasWaiter(corrID) {
+ responses = append(responses, env)
+ } else {
+ durables = append(durables, env)
+ }
+ }
+
+ return responses, durables
+}
+
+// deliverWaiterResponses delivers each split-time waiter-backed response to
+// its live in-memory waiter, outside the dispatch transaction. It returns the
+// stragglers: responses whose waiter vanished (RPC deadline cancel or TTL
+// prune) between the split peek and this delivery. Those must NOT be durably
+// dispatched here — that would commit a durable enqueue ahead of the cursor
+// fold — so the caller folds them into the transactional durable batch
+// instead. A miss may have buffered an early response copy, which is dropped
+// so the durable fold remains the single source of truth.
+func (a *ServerConnectionActor) deliverWaiterResponses(
+ responses []*mailboxpb.Envelope) []*mailboxpb.Envelope {
+
+ var stragglers []*mailboxpb.Envelope
+ for _, env := range responses {
+ corrID := CorrelationID(env.Rpc.CorrelationId)
+
+ // A correlation-less response cannot match a waiter; defer it
+ // to the durable fold like any other non-waiter envelope.
+ if corrID == "" {
+ stragglers = append(stragglers, env)
+
+ continue
+ }
+
+ delivery := a.deliverResponse(corrID, env)
+ if delivery == mailboxconn.DeliveryWaiter {
+ continue
+ }
+
+ // The waiter disappeared after the split peek. Drop any
+ // buffered copy and defer the envelope into the durable
+ // transaction.
+ a.removePendingResponse(corrID)
+ stragglers = append(stragglers, env)
+ }
+
+ return stragglers
+}
+
+// mergeEnvelopesByEventSeq merges two event_seq-ascending envelope slices
+// into one ascending slice. Both the durable partition and the straggler set
+// derive from a single ordered pass over the pulled batch, so each input is
+// already sorted; the merge preserves per-lane FIFO order when stragglers
+// fold back into the durable batch.
+func mergeEnvelopesByEventSeq(
+ a, b []*mailboxpb.Envelope) []*mailboxpb.Envelope {
+
+ merged := make([]*mailboxpb.Envelope, 0, len(a)+len(b))
+
+ i, j := 0, 0
+ for i < len(a) && j < len(b) {
+ if a[i].EventSeq <= b[j].EventSeq {
+ merged = append(merged, a[i])
+ i++
+ } else {
+ merged = append(merged, b[j])
+ j++
+ }
+ }
+
+ merged = append(merged, a[i:]...)
+ merged = append(merged, b[j:]...)
+
+ return merged
+}
+
// saveCheckpoint persists the AckState to the checkpoint store.
func (a *ServerConnectionActor) saveCheckpoint(
ctx context.Context, state AckState,
) error {
+ return a.saveCheckpointTo(ctx, a.cfg.Store, state)
+}
+
+// saveCheckpointTo persists the AckState through the given store, which may
+// be a transaction-scoped store so the checkpoint joins an ambient dispatch
+// transaction instead of paying its own commit.
+func (a *ServerConnectionActor) saveCheckpointTo(ctx context.Context,
+ store actor.DeliveryStore, state AckState) error {
+
var buf bytes.Buffer
if err := state.Encode(&buf); err != nil {
return err
@@ -459,7 +708,7 @@ func (a *ServerConnectionActor) saveCheckpoint(
actorID := DurableActorID(a.cfg.LocalMailboxID)
- return a.cfg.Store.SaveCheckpoint(ctx, actor.CheckpointParams{
+ return store.SaveCheckpoint(ctx, actor.CheckpointParams{
ActorID: actorID,
StateType: ackStateType,
StateData: buf.Bytes(),
diff --git a/serverconn/ingress_split_test.go b/serverconn/ingress_split_test.go
new file mode 100644
index 000000000..6a9707b31
--- /dev/null
+++ b/serverconn/ingress_split_test.go
@@ -0,0 +1,151 @@
+package serverconn
+
+import (
+ "testing"
+
+ mailboxpb "github.com/lightninglabs/darepo-client/mailbox/pb"
+ "github.com/stretchr/testify/require"
+)
+
+// responseEnvelope builds a KIND_RESPONSE envelope with the given correlation
+// ID and event_seq for split-partition tests.
+func responseEnvelope(corrID string, seq uint64) *mailboxpb.Envelope {
+ return &mailboxpb.Envelope{
+ EventSeq: seq,
+ Rpc: &mailboxpb.RpcMeta{
+ Kind: mailboxpb.RpcMeta_KIND_RESPONSE,
+ CorrelationId: corrID,
+ },
+ }
+}
+
+// eventEnvelope builds a KIND_EVENT envelope with the given event_seq for
+// split-partition tests.
+func eventEnvelope(seq uint64) *mailboxpb.Envelope {
+ return &mailboxpb.Envelope{
+ EventSeq: seq,
+ Rpc: &mailboxpb.RpcMeta{
+ Kind: mailboxpb.RpcMeta_KIND_EVENT,
+ },
+ }
+}
+
+// TestSplitIngressEnvelopesWaiterRouting verifies that splitIngressEnvelopes
+// keeps the fast pre-transaction path only for responses with a live waiter,
+// while a waiterless response folds into the durable bucket alongside events.
+// This is the no-reorder/single-commit contract: a durable-fallback response
+// must commit in the same transaction as the cursor, not ahead of it.
+func TestSplitIngressEnvelopesWaiterRouting(t *testing.T) {
+ t.Parallel()
+
+ // Only "corr-live" has a registered in-memory waiter.
+ hasWaiter := func(id CorrelationID) bool {
+ return id == CorrelationID("corr-live")
+ }
+
+ envelopes := []*mailboxpb.Envelope{
+ responseEnvelope("corr-live", 1),
+ responseEnvelope("corr-gone", 2),
+ eventEnvelope(3),
+ responseEnvelope("", 4),
+ }
+
+ responses, durables := splitIngressEnvelopes(envelopes, hasWaiter)
+
+ // Only the waiter-backed response takes the pre-transaction path.
+ require.Len(t, responses, 1)
+ require.Equal(t, uint64(1), responses[0].EventSeq)
+
+ // The waiterless response, the event, and the correlation-less
+ // response all fold into the durable transaction in event_seq order.
+ require.Len(t, durables, 3)
+ require.Equal(t, uint64(2), durables[0].EventSeq)
+ require.Equal(t, uint64(3), durables[1].EventSeq)
+ require.Equal(t, uint64(4), durables[2].EventSeq)
+}
+
+// TestSplitIngressEnvelopesNoWaiters verifies that with no live waiters every
+// envelope folds into the durable transaction, so nothing dispatches ahead of
+// the cursor commit on a crash-replay batch where all waiters are gone.
+func TestSplitIngressEnvelopesNoWaiters(t *testing.T) {
+ t.Parallel()
+
+ hasWaiter := func(CorrelationID) bool { return false }
+
+ envelopes := []*mailboxpb.Envelope{
+ responseEnvelope("corr-a", 1),
+ responseEnvelope("corr-b", 2),
+ eventEnvelope(3),
+ }
+
+ responses, durables := splitIngressEnvelopes(envelopes, hasWaiter)
+
+ require.Empty(t, responses)
+ require.Len(t, durables, 3)
+}
+
+// TestDeliverWaiterResponsesDefersVanishedWaiters covers the TOCTOU guard in
+// the pre-transaction response path. splitIngressEnvelopes routes a response
+// to the fast bucket on a split-time waiter peek, but the waiter can vanish
+// (RPC deadline cancel or TTL prune) before delivery runs. deliverWaiter
+// Responses must deliver only to a still-live waiter and return every other
+// response as a straggler — to fold into the durable transaction — rather than
+// dispatch it durably outside the cursor fold.
+func TestDeliverWaiterResponsesDefersVanishedWaiters(t *testing.T) {
+ t.Parallel()
+
+ actor, _, _ := newTestConnector(t, nil)
+
+ // corr-live keeps a live waiter; corr-gone models a waiter that
+ // vanished after the split peek (never registered here), and the empty
+ // correlation ID can never match a waiter.
+ actor.RegisterWaiter(CorrelationID("corr-live"))
+
+ responses := []*mailboxpb.Envelope{
+ responseEnvelope("corr-live", 1),
+ responseEnvelope("corr-gone", 2),
+ responseEnvelope("", 3),
+ }
+
+ stragglers := actor.deliverWaiterResponses(responses)
+
+ // The live-waiter response is delivered in memory and excluded; only
+ // the vanished-waiter and correlation-less responses fold back into the
+ // durable batch, preserved in event_seq order.
+ require.Len(t, stragglers, 2)
+ require.Equal(t, uint64(2), stragglers[0].EventSeq)
+ require.Equal(t, uint64(3), stragglers[1].EventSeq)
+}
+
+// TestMergeEnvelopesByEventSeq verifies the straggler fold preserves global
+// event_seq order when deferred responses merge back into the durable
+// partition, and that empty inputs are handled.
+func TestMergeEnvelopesByEventSeq(t *testing.T) {
+ t.Parallel()
+
+ durables := []*mailboxpb.Envelope{
+ eventEnvelope(1),
+ eventEnvelope(3),
+ eventEnvelope(5),
+ }
+ stragglers := []*mailboxpb.Envelope{
+ responseEnvelope("a", 2),
+ responseEnvelope("b", 4),
+ }
+
+ merged := mergeEnvelopesByEventSeq(durables, stragglers)
+ require.Len(t, merged, 5)
+ for i, env := range merged {
+ require.Equal(t, uint64(i+1), env.EventSeq)
+ }
+
+ // Empty straggler set returns the durable partition unchanged.
+ require.Equal(
+ t, durables, mergeEnvelopesByEventSeq(durables, nil),
+ )
+
+ // Empty durable partition returns the stragglers unchanged.
+ require.Equal(
+ t, stragglers, mergeEnvelopesByEventSeq(nil, stragglers),
+ )
+}