test: cover stale ledger outbox lease updates

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

tests/adapters/persistence/test_postgresql_ledger_outbox.py

@@ -169,6 +169,45 @@ def test_claim_expiry_allows_safe_recovery_after_worker_crash(pg_clean):
     assert [item.correlation_id for item in claimed_b] == [row.correlation_id]
 
 
+def test_stale_claim_cannot_overwrite_new_owner_outcome(pg_clean):
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox_a = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-a",
+        clock=lambda: now["t"],
+    )
+    outbox_b = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-b",
+        clock=lambda: now["t"],
+    )
+    row = outbox_a.enqueue(_trade(execution_id="pg-stale"))
+
+    [claim_a] = outbox_a.claim_pending()
+    now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC)
+    [claim_b] = outbox_b.claim_pending()
+    outbox_b.mark_outcome(
+        row.correlation_id,
+        OutboxOutcome.COMPLETED,
+        expected_claim_expires_at=claim_b.claim_expires_at,
+    )
+
+    stored = outbox_a.mark_outcome(
+        row.correlation_id,
+        OutboxOutcome.RETRYABLE_FAILURE,
+        error="stale owner",
+        next_retry_at=now["t"],
+        expected_claim_expires_at=claim_a.claim_expires_at,
+    )
+
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.last_outcome is OutboxOutcome.COMPLETED
+
+
 def test_writer_records_retryable_403_without_cross_instance_delivery(pg_clean):
     class _ForbiddenLedger:
         def __init__(self):

tests/application/test_ledger_outbox.py

@@ -71,9 +71,9 @@ def test_writer_retries_until_success():
 
 
 def test_process_correlation_id_only_drains_requested_row():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
     ledger = InMemoryLedger()
-    writer = LedgerWriter(outbox, ledger)
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
     first = outbox.enqueue(_trade(execution_id="e-1"))
     second = outbox.enqueue(_trade(execution_id="e-2"))
 
@@ -81,6 +81,8 @@ def test_process_correlation_id_only_drains_requested_row():
 
     assert outcome is OutboxOutcome.COMPLETED
     assert ledger.count() == 1
+    assert writer.counters["claim:claimed"] == 1
+    assert writer.counters["outcome:completed"] == 1
     assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
     assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
 

trex_core/adapters/persistence/postgresql.py

@@ -566,7 +566,8 @@ class _PgIntentRecord:
 _OUTBOX_COLUMNS = (
     "correlation_id, asset_id, order_id, execution_id, side, qty, price, "
     "fee, tax, executed_at, status, attempts, last_outcome, last_error, "
-    "created_at, updated_at, next_retry_at, venue, order_venue, instance_key"
+    "created_at, updated_at, next_retry_at, venue, order_venue, instance_key, "
+    "claim_expires_at"
 )
 
 
@@ -710,6 +711,7 @@ class PostgreSQLLedgerOutbox:
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: datetime | None = None,
     ) -> LedgerWorkItem:
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
             with conn.cursor() as cur:
@@ -731,13 +733,7 @@ class PostgreSQLLedgerOutbox:
                     attempts_delta = 1
                 # next_retry_at only persists for a retryable backoff; cleared otherwise.
                 nra = next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
-                cur.execute(
+                params: list[Any] = [
-                    "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
-                    "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
-                    "claim_owner=NULL, claim_expires_at=NULL "
-                    "WHERE correlation_id=%s AND instance_key=%s "
-                    "AND (claim_owner=%s OR claim_owner IS NULL)",
-                    (
                     status_value,
                     attempts_delta,
                     outcome.value,
@@ -746,8 +742,20 @@ class PostgreSQLLedgerOutbox:
                     self._now(),
                     correlation_id,
                     self._instance_key,
-                        self._worker_id,
+                ]
-                    ),
+                where = "WHERE correlation_id=%s AND instance_key=%s "
+                if expected_claim_expires_at is None:
+                    where += "AND status=%s AND claim_owner IS NULL AND claim_expires_at IS NULL"
+                    params.append(WorkItemStatus.COMPLETED.value)
+                else:
+                    where += "AND claim_owner=%s AND claim_expires_at=%s"
+                    params.extend([self._worker_id, expected_claim_expires_at])
+                cur.execute(
+                    "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
+                    "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
+                    "claim_owner=NULL, claim_expires_at=NULL "
+                    f"{where}",
+                    params,
                 )
                 conn.commit()
         row = self.get(correlation_id)
@@ -781,6 +789,7 @@ class PostgreSQLLedgerOutbox:
             updated_at=row[15],
             next_retry_at=row[16],
             instance_key=row[19],
+            claim_expires_at=row[20],
         )
 
 

trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql

@@ -13,8 +13,9 @@ WHERE instance_key IS NULL;
 ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET DEFAULT '__unowned__';
 ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET NOT NULL;
 
+DROP INDEX IF EXISTS idx_ledger_outbox_instance_pending;
 CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_pending
-    ON ledger_outbox (instance_key, status, created_at)
+    ON ledger_outbox (instance_key, status, created_at, correlation_id)
     WHERE status = 'pending';
 
 CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_claim_expires

trex_core/application/ledger_outbox.py

@@ -105,6 +105,7 @@ class LedgerWorkItem:
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
     instance_key: Optional[str] = None
+    claim_expires_at: Optional[datetime] = None
 
 
 class LedgerOutbox(Protocol):
@@ -146,6 +147,7 @@ class LedgerOutbox(Protocol):
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: Optional[datetime] = None,
     ) -> LedgerWorkItem:
         """Record the outcome of a writer attempt and advance status.
 
@@ -266,6 +268,14 @@ class LedgerWriter:
                 )
                 return None
             item = claimed[0]
+            self._count("claim:claimed")
+            _log.info(
+                "ledger outbox claim: instance_key=%s claim_result=claimed "
+                "count=%d correlation_id=%s",
+                self._instance_key,
+                len(claimed),
+                correlation_id,
+            )
         return self._process(item)
 
     def _backoff_delay(self, attempts: int) -> float:
@@ -285,7 +295,10 @@ class LedgerWriter:
             outcome = self._ledger.record_trade(item.trade)
         except LedgerConflictError as exc:
             self._outbox.mark_outcome(
-                item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
+                item.correlation_id,
+                OutboxOutcome.CONFLICT_BLOCKED,
+                error=str(exc),
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc))
             # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
@@ -302,7 +315,10 @@ class LedgerWriter:
             # dcy-015 / REQ-017/018 (issue #13): non-retryable client error →
             # terminal delivery_failed immediately (no retry spam). Surface for ops.
             self._outbox.mark_outcome(
-                item.correlation_id, OutboxOutcome.PERMANENT_FAILURE, error=str(exc)
+                item.correlation_id,
+                OutboxOutcome.PERMANENT_FAILURE,
+                error=str(exc),
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc))
             _log.warning(
@@ -325,7 +341,10 @@ class LedgerWriter:
             new_attempts = item.attempts + 1
             if new_attempts >= self._max_attempts:
                 self._outbox.mark_outcome(
-                    item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc)
+                    item.correlation_id,
+                    OutboxOutcome.DEAD_LETTERED,
+                    error=str(exc),
+                    expected_claim_expires_at=item.claim_expires_at,
                 )
                 self._log_outcome(item, OutboxOutcome.DEAD_LETTERED, error=str(exc))
                 _log.warning(
@@ -352,7 +371,9 @@ class LedgerWriter:
             )
             self._outbox.mark_outcome(
                 item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE,
-                error=str(exc), next_retry_at=next_retry_at,
+                error=str(exc),
+                next_retry_at=next_retry_at,
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc))
             return OutboxOutcome.RETRYABLE_FAILURE
@@ -369,7 +390,11 @@ class LedgerWriter:
             self._resolve_ledger_skip(item.trade)
             self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
-        self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
+        self._outbox.mark_outcome(
+            item.correlation_id,
+            OutboxOutcome.COMPLETED,
+            expected_claim_expires_at=item.claim_expires_at,
+        )
         self._resolve_ledger_skip(item.trade)
         self._log_outcome(item, OutboxOutcome.COMPLETED)
         return OutboxOutcome.COMPLETED
@@ -522,6 +547,7 @@ class InMemoryLedgerOutbox:
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: Optional[datetime] = None,
     ) -> LedgerWorkItem:
         with self._lock:
             row = self._rows[correlation_id]