serverconn: fold ingress dispatch and checkpoint into one transaction

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()

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))
+}

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

p-models/durableactor/infra.pproj

@@ -2,7 +2,9 @@
 <ProjectName>MailboxInfraModels</ProjectName>
 <InputFiles>
 <PFile>src/mailbox_fifo.p</PFile>
+<PFile>src/ingress_fold.p</PFile>
 <PFile>test/mailbox_fifo_test.p</PFile>
+<PFile>test/ingress_fold_test.p</PFile>
 </InputFiles>
 <OutputDir>PGenerated</OutputDir>
 </Project>

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;
+            }
+        }
+    }
+}

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 };

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.