From 31d6e5c02b50799b01ed5ff950d53c8b3c70354b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:27:27 +0000 Subject: [PATCH] test: cover stale ledger outbox lease updates Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_postgresql_ledger_outbox.py | 39 +++++++++++++++++++ tests/application/test_ledger_outbox.py | 6 ++- trex_core/adapters/persistence/postgresql.py | 37 +++++++++++------- .../sql/028_ledger_outbox_instance_claims.sql | 3 +- trex_core/application/ledger_outbox.py | 36 ++++++++++++++--- 5 files changed, 99 insertions(+), 22 deletions(-) diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py index 6f51388..d553846 100644 --- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py +++ b/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): diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py index e085f30..25e747d 100644 --- a/tests/application/test_ledger_outbox.py +++ b/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 diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 64cabc7..aa5b0f0 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/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,23 +733,29 @@ 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 + params: list[Any] = [ + status_value, + attempts_delta, + outcome.value, + error, + nra, + self._now(), + correlation_id, + self._instance_key, + ] + 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 " - "WHERE correlation_id=%s AND instance_key=%s " - "AND (claim_owner=%s OR claim_owner IS NULL)", - ( - status_value, - attempts_delta, - outcome.value, - error, - nra, - self._now(), - correlation_id, - self._instance_key, - self._worker_id, - ), + 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], ) diff --git a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql index 186c809..1c037c7 100644 --- a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql +++ b/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 diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index 79e7b32..e0bdd5c 100644 --- a/trex_core/application/ledger_outbox.py +++ b/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]