trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: fence ledger outbox claims by instance

Browse Source
This commit is contained in:
Hermes Agent
2026-09-13 09:41:51 +09:00
parent f801c53e0c
commit 762a953c95
15 changed files with 316 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/adapters/persistence/postgresql.py
+23 -8
View File
@@ -21,6 +21,7 @@ from uuid import UUID, uuid4
from trex_core.application.ledger_outbox import (
LedgerWorkItem,
OutboxClaimLostError,
OutboxOutcome,
WorkItemStatus,
)
@@ -569,6 +570,9 @@ _OUTBOX_COLUMNS = (
"created_at, updated_at, next_retry_at, venue, order_venue, instance_key, "
"claim_expires_at"
)
_OUTBOX_RETURN_COLUMNS = ", ".join(
f"o.{column.strip()}" for column in _OUTBOX_COLUMNS.split(",")
)
class PostgreSQLLedgerOutbox:
@@ -578,12 +582,16 @@ class PostgreSQLLedgerOutbox:
self,
dsn: str,
*,
instance_key: str = "default",
instance_key: str,
claim_ttl_sec: float = 60.0,
worker_id: str | None = None,
clock=None,
psycopg_module: Any = None,
) -> None:
if not isinstance(instance_key, str) or not instance_key.strip():
raise ValueError("instance_key must be a non-empty string")
if claim_ttl_sec <= 0:
raise ValueError("claim_ttl_sec must be positive")
self._psycopg = _require_psycopg(psycopg_module)
self._dsn = dsn
self._instance_key = instance_key
@@ -594,6 +602,10 @@ class PostgreSQLLedgerOutbox:
def _now(self) -> datetime:
return self._clock()
@property
def instance_key(self) -> str:
return self._instance_key
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
@@ -625,16 +637,17 @@ class PostgreSQLLedgerOutbox:
return existing
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
now = self._now()
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox "
"WHERE status = 'pending' "
"AND instance_key = %s "
"AND (next_retry_at IS NULL OR next_retry_at <= now()) "
"AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
"AND (next_retry_at IS NULL OR next_retry_at <= %s) "
"AND (claim_expires_at IS NULL OR claim_expires_at <= %s) "
"ORDER BY created_at, correlation_id LIMIT %s",
(self._instance_key, limit),
(self._instance_key, now, now, limit),
)
return [self._row_to_item(r) for r in cur.fetchall()]
@@ -647,7 +660,7 @@ class PostgreSQLLedgerOutbox:
now = self._now()
lease_until = now + timedelta(seconds=self._claim_ttl_sec)
correlation_sql = ""
params: list[Any] = [self._instance_key]
params: list[Any] = [self._instance_key, now, now]
if correlation_id is not None:
correlation_sql = "AND correlation_id = %s "
params.append(correlation_id)
@@ -659,8 +672,8 @@ class PostgreSQLLedgerOutbox:
"SELECT correlation_id FROM ledger_outbox "
"WHERE status = 'pending' "
"AND instance_key = %s "
"AND (next_retry_at IS NULL OR next_retry_at <= now()) "
"AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
"AND (next_retry_at IS NULL OR next_retry_at <= %s) "
"AND (claim_expires_at IS NULL OR claim_expires_at <= %s) "
f"{correlation_sql}"
"ORDER BY created_at, correlation_id "
"LIMIT %s "
@@ -670,7 +683,7 @@ class PostgreSQLLedgerOutbox:
"SET claim_owner = %s, claim_expires_at = %s, updated_at = %s "
"FROM candidates "
"WHERE o.correlation_id = candidates.correlation_id "
f"RETURNING {_OUTBOX_COLUMNS}",
f"RETURNING {_OUTBOX_RETURN_COLUMNS}",
params,
)
rows = [self._row_to_item(r) for r in cur.fetchall()]
@@ -759,6 +772,8 @@ class PostgreSQLLedgerOutbox:
f"{where}",
params,
)
if cur.rowcount != 1:
raise OutboxClaimLostError(correlation_id)
conn.commit()
row = self.get(correlation_id)
assert row is not None
trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql → trex_core/adapters/persistence/sql/029_ledger_outbox_instance_claims.sql
View File
trex_core/application/ledger_outbox.py
+106 -43
View File
@@ -66,6 +66,10 @@ class WorkItemStatus(str, Enum):
DELIVERY_FAILED = "delivery_failed" # dcy-015 — permanent or dead-lettered (terminal)
class OutboxClaimLostError(RuntimeError):
"""The caller no longer owns the lease required for an outcome update."""
# Outcomes that move a row to the terminal delivery_failed state.
_DELIVERY_FAILED_OUTCOMES = frozenset(
{OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}
@@ -111,6 +115,11 @@ class LedgerWorkItem:
class LedgerOutbox(Protocol):
"""Durable work-item store backing the ledger writer."""
@property
def instance_key(self) -> str:
"""Immutable owner partition for rows handled by this outbox."""
...
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
"""Enqueue a trade. Idempotent on correlation_id (returns the
existing row unchanged if it is already present)."""
@@ -163,6 +172,7 @@ class WriterStats:
retryable: int = 0
conflicts: int = 0
delivery_failed: int = 0 # dcy-015: permanent + dead-lettered terminals
claim_lost: int = 0
class LedgerWriter:
@@ -202,7 +212,12 @@ class LedgerWriter:
# dcy-033 / REQ-100: best-effort alarm emitter.
self._alarm_emitter = alarm_emitter
self._ledger_skip_store = ledger_skip_store
self._instance_key = instance_key
resolved_instance_key = outbox.instance_key if instance_key is None else instance_key
if not isinstance(resolved_instance_key, str) or not resolved_instance_key.strip():
raise ValueError("instance_key must be a non-empty string")
if resolved_instance_key != outbox.instance_key:
raise ValueError("writer instance_key must match outbox instance_key")
self._instance_key = resolved_instance_key
self.counters: dict[str, int] = {}
def _count(self, key: str) -> None:
@@ -224,7 +239,19 @@ class LedgerWriter:
limit,
)
for item in items:
outcome = self._process(item)
if not self._validate_instance(item):
continue
try:
outcome = self._process(item)
except OutboxClaimLostError:
self._count("claim:lost")
stats.claim_lost += 1
_log.warning(
"ledger outbox claim lost: instance_key=%s correlation_id=%s",
self._instance_key,
item.correlation_id,
)
continue
if outcome is OutboxOutcome.COMPLETED:
stats.completed += 1
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
@@ -242,19 +269,7 @@ class LedgerWriter:
item = self._outbox.get(correlation_id)
if item is None:
return None
if (
self._instance_key is not None
and item.instance_key is not None
and item.instance_key != self._instance_key
):
self._count("claim:foreign_instance")
_log.info(
"ledger outbox claim: instance_key=%s claim_result=foreign_instance "
"correlation_id=%s owner_instance_key=%s",
self._instance_key,
correlation_id,
item.instance_key,
)
if not self._validate_instance(item):
return None
if item.status is WorkItemStatus.PENDING:
claimed = self._outbox.claim_pending(1, correlation_id=correlation_id)
@@ -276,7 +291,28 @@ class LedgerWriter:
len(claimed),
correlation_id,
)
return self._process(item)
try:
return self._process(item)
except OutboxClaimLostError:
self._count("claim:lost")
_log.warning(
"ledger outbox claim lost: instance_key=%s correlation_id=%s",
self._instance_key,
correlation_id,
)
return None
def _validate_instance(self, item: LedgerWorkItem) -> bool:
if item.instance_key == self._instance_key:
return True
self._count("claim:foreign_instance")
_log.warning(
"ledger outbox claim rejected: instance_key=%s "
"claim_result=foreign_or_unowned correlation_id=%s",
self._instance_key,
item.correlation_id,
)
return False
def _backoff_delay(self, attempts: int) -> float:
"""Exponential backoff with jitter: min(base*2^(n-1), cap) + up to 25%."""
@@ -288,7 +324,6 @@ class LedgerWriter:
# outbox row. Skip the adapter call.
existing = self._outbox.get(item.correlation_id)
if existing is not None and existing.status is WorkItemStatus.COMPLETED:
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
return OutboxOutcome.DUPLICATE_COMPLETED
try:
@@ -300,7 +335,7 @@ class LedgerWriter:
error=str(exc),
expected_claim_expires_at=item.claim_expires_at,
)
self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc))
self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED)
# dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
if self._alarm_emitter is not None:
self._alarm_emitter.publish_alarm(
@@ -320,11 +355,11 @@ class LedgerWriter:
error=str(exc),
expected_claim_expires_at=item.claim_expires_at,
)
self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc))
self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE)
_log.warning(
"ledger delivery_failed (permanent): instance_key=%s correlation_id=%s reason=%s",
"ledger delivery_failed (permanent): instance_key=%s correlation_id=%s",
item.instance_key or self._instance_key,
item.correlation_id, exc,
item.correlation_id,
)
# dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry.
if self._alarm_emitter is not None:
@@ -346,14 +381,13 @@ class LedgerWriter:
error=str(exc),
expected_claim_expires_at=item.claim_expires_at,
)
self._log_outcome(item, OutboxOutcome.DEAD_LETTERED, error=str(exc))
self._log_outcome(item, OutboxOutcome.DEAD_LETTERED)
_log.warning(
"ledger delivery_failed (dead-letter, attempts=%d): "
"instance_key=%s correlation_id=%s reason=%s",
"instance_key=%s correlation_id=%s",
new_attempts,
item.instance_key or self._instance_key,
item.correlation_id,
exc,
)
# dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries.
if self._alarm_emitter is not None:
@@ -375,7 +409,7 @@ class LedgerWriter:
next_retry_at=next_retry_at,
expected_claim_expires_at=item.claim_expires_at,
)
self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc))
self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE)
return OutboxOutcome.RETRYABLE_FAILURE
# dcy-018 / REQ-019/030 (option D): write back portfolio-hub authoritative
# cash/quantity to the snapshot store on new+duplicate (OQ-2 b). *_after
@@ -386,7 +420,11 @@ class LedgerWriter:
# duplicate_completed; the ledger has exactly one entry for this
# execution_id either way (REQ-019 AC-1, AC-4).
if outcome.result is LedgerWriteResult.DUPLICATE_OK:
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
self._outbox.mark_outcome(
item.correlation_id,
OutboxOutcome.DUPLICATE_COMPLETED,
expected_claim_expires_at=item.claim_expires_at,
)
self._resolve_ledger_skip(item.trade)
self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
return OutboxOutcome.DUPLICATE_COMPLETED
@@ -403,8 +441,6 @@ class LedgerWriter:
self,
item: LedgerWorkItem,
outcome: OutboxOutcome,
*,
error: str | None = None,
) -> None:
self._count(f"outcome:{outcome.value}")
msg = "ledger outbox outcome: instance_key=%s correlation_id=%s outcome=%s"
@@ -413,9 +449,6 @@ class LedgerWriter:
item.correlation_id,
outcome.value,
)
if error is not None:
msg += " error=%s"
args = (*args, error)
_log.info(msg, *args)
def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
@@ -454,7 +487,7 @@ class LedgerWriter:
except Exception: # noqa: BLE001 - write-back must not block the ledger path
_log.warning(
"snapshot cash/position write-back failed: asset_id=%s",
trade.asset_id, exc_info=True,
trade.asset_id,
)
@@ -467,7 +500,8 @@ class InMemoryLedgerOutbox:
self,
*,
clock: Optional[Callable[[], datetime]] = None,
instance_key: str = "default",
instance_key: str,
claim_ttl_sec: float = 60.0,
) -> None:
import threading
@@ -477,7 +511,16 @@ class InMemoryLedgerOutbox:
# dcy-015: shared clock lets tests align backoff (next_retry_at) with
# list_pending's "now" deterministically.
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
if not isinstance(instance_key, str) or not instance_key.strip():
raise ValueError("instance_key must be a non-empty string")
if claim_ttl_sec <= 0:
raise ValueError("claim_ttl_sec must be positive")
self._instance_key = instance_key
self._claim_ttl_sec = claim_ttl_sec
@property
def instance_key(self) -> str:
return self._instance_key
def _now(self) -> datetime:
return self._clock()
@@ -508,8 +551,11 @@ class InMemoryLedgerOutbox:
row = self._rows[cid]
# dcy-015 / REQ-019: skip rows still in backoff so they don't
# occupy the FIFO batch (head-of-line protection, NFR-3).
if row.status is WorkItemStatus.PENDING and (
row.next_retry_at is None or row.next_retry_at <= now
if (
row.status is WorkItemStatus.PENDING
and row.instance_key == self._instance_key
and (row.next_retry_at is None or row.next_retry_at <= now)
and (row.claim_expires_at is None or row.claim_expires_at <= now)
):
out.append(row)
if len(out) >= limit:
@@ -522,16 +568,27 @@ class InMemoryLedgerOutbox:
*,
correlation_id: str | None = None,
) -> list[LedgerWorkItem]:
if correlation_id is None:
return self.list_pending(limit)
now = self._now()
lease_until = now + timedelta(seconds=self._claim_ttl_sec)
with self._lock:
row = self._rows.get(correlation_id)
if row is None or row.status is not WorkItemStatus.PENDING:
return []
if row.next_retry_at is not None and row.next_retry_at > now:
return []
return [row]
claimed: list[LedgerWorkItem] = []
candidates = [correlation_id] if correlation_id is not None else self._order
for cid in candidates:
row = self._rows.get(cid)
if row is None or row.status is not WorkItemStatus.PENDING:
continue
if row.instance_key != self._instance_key:
continue
if row.next_retry_at is not None and row.next_retry_at > now:
continue
if row.claim_expires_at is not None and row.claim_expires_at > now:
continue
updated = replace(row, claim_expires_at=lease_until, updated_at=now)
self._rows[cid] = updated
claimed.append(updated)
if len(claimed) >= limit:
break
return claimed
def pending_cash_delta(self) -> dict[str, float]:
with self._lock:
@@ -557,6 +614,11 @@ class InMemoryLedgerOutbox:
) -> LedgerWorkItem:
with self._lock:
row = self._rows[correlation_id]
if expected_claim_expires_at is None:
if row.claim_expires_at is not None:
raise OutboxClaimLostError(correlation_id)
elif row.claim_expires_at != expected_claim_expires_at:
raise OutboxClaimLostError(correlation_id)
if outcome is OutboxOutcome.COMPLETED:
new_status = WorkItemStatus.COMPLETED
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
@@ -578,6 +640,7 @@ class InMemoryLedgerOutbox:
next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
),
updated_at=self._now(),
claim_expires_at=None,
)
self._rows[correlation_id] = updated
return updated