238 lines
8.4 KiB
Python
238 lines
8.4 KiB
Python
"""Ledger outbox + writer (REQ-019, dcy-004 FR-3/FR-4).
|
|
|
|
The execution fill path does NOT call portfolio-hub directly. Instead it
|
|
enqueues a durable `LedgerWorkItem` keyed by `correlation_id`
|
|
(`{asset_id}:{order_id}:{execution_id}`). A separate `LedgerWriter` drains
|
|
the outbox into the `LedgerPort` adapter and classifies each attempt as one
|
|
of:
|
|
|
|
completed -- adapter accepted (new ledger entry).
|
|
duplicate_completed -- the correlation_id was already done in this outbox
|
|
(local dedup, no adapter call).
|
|
retryable_failure -- transient adapter failure; left pending for next tick.
|
|
conflict_blocked -- portfolio-hub 409; operator must resolve, no retry.
|
|
|
|
This separation (orbit `LedgerWriter` pattern) keeps slot transitions on the
|
|
critical path while making ledger writes eventually-consistent and idempotent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from datetime import datetime, timezone
|
|
from enum import Enum
|
|
from typing import Optional, Protocol
|
|
|
|
from trex_core.application.ledger_port import (
|
|
LedgerConflictError,
|
|
LedgerPort,
|
|
LedgerRetryableError,
|
|
LedgerTrade,
|
|
LedgerWriteResult,
|
|
)
|
|
|
|
|
|
class OutboxOutcome(str, Enum):
|
|
COMPLETED = "completed"
|
|
DUPLICATE_COMPLETED = "duplicate_completed"
|
|
RETRYABLE_FAILURE = "retryable_failure"
|
|
CONFLICT_BLOCKED = "conflict_blocked"
|
|
|
|
|
|
class WorkItemStatus(str, Enum):
|
|
PENDING = "pending" # awaiting next writer tick
|
|
COMPLETED = "completed" # adapter accepted (terminal success)
|
|
CONFLICT_BLOCKED = "conflict_blocked" # 409 — operator resolves (terminal failure)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LedgerWorkItem:
|
|
"""One outbox row.
|
|
|
|
Identity is `correlation_id`; the outbox MUST upsert/dedupe on it so the
|
|
same fill enqueued twice produces a single row.
|
|
"""
|
|
|
|
correlation_id: str
|
|
trade: LedgerTrade
|
|
status: WorkItemStatus = WorkItemStatus.PENDING
|
|
attempts: int = 0
|
|
last_outcome: Optional[OutboxOutcome] = None
|
|
last_error: Optional[str] = None
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
|
|
class LedgerOutbox(Protocol):
|
|
"""Durable work-item store backing the ledger writer."""
|
|
|
|
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
|
|
"""Enqueue a trade. Idempotent on correlation_id (returns the
|
|
existing row unchanged if it is already present)."""
|
|
...
|
|
|
|
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
|
|
"""Return pending work items in FIFO order, up to `limit`."""
|
|
...
|
|
|
|
def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
|
|
...
|
|
|
|
def mark_outcome(
|
|
self,
|
|
correlation_id: str,
|
|
outcome: OutboxOutcome,
|
|
*,
|
|
error: Optional[str] = None,
|
|
) -> LedgerWorkItem:
|
|
"""Record the outcome of a writer attempt and advance status."""
|
|
...
|
|
|
|
|
|
@dataclass
|
|
class WriterStats:
|
|
completed: int = 0
|
|
duplicates: int = 0
|
|
retryable: int = 0
|
|
conflicts: int = 0
|
|
|
|
|
|
class LedgerWriter:
|
|
"""Drains the outbox into a `LedgerPort`. One process-wide writer.
|
|
|
|
Local dedup is by correlation_id: if a work item is already marked
|
|
`completed`, the writer reports `duplicate_completed` without touching
|
|
the adapter (REQ-019 AC-4).
|
|
"""
|
|
|
|
def __init__(self, outbox: LedgerOutbox, ledger: LedgerPort) -> None:
|
|
self._outbox = outbox
|
|
self._ledger = ledger
|
|
|
|
def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem:
|
|
return self._outbox.enqueue(trade)
|
|
|
|
def run_once(self, limit: int = 32) -> WriterStats:
|
|
stats = WriterStats()
|
|
for item in self._outbox.list_pending(limit):
|
|
outcome = self._process(item)
|
|
if outcome is OutboxOutcome.COMPLETED:
|
|
stats.completed += 1
|
|
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
|
|
stats.duplicates += 1
|
|
elif outcome is OutboxOutcome.RETRYABLE_FAILURE:
|
|
stats.retryable += 1
|
|
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
|
|
stats.conflicts += 1
|
|
return stats
|
|
|
|
def _process(self, item: LedgerWorkItem) -> OutboxOutcome:
|
|
# Local dedup: another writer pass (or restart) already completed this
|
|
# 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)
|
|
return OutboxOutcome.DUPLICATE_COMPLETED
|
|
try:
|
|
result = self._ledger.record_trade(item.trade)
|
|
except LedgerConflictError as exc:
|
|
self._outbox.mark_outcome(
|
|
item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
|
|
)
|
|
return OutboxOutcome.CONFLICT_BLOCKED
|
|
except LedgerRetryableError as exc:
|
|
self._outbox.mark_outcome(
|
|
item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc)
|
|
)
|
|
return OutboxOutcome.RETRYABLE_FAILURE
|
|
# Adapter-level dedup (server says "already recorded"): outcome is
|
|
# duplicate_completed; the ledger has exactly one entry for this
|
|
# execution_id either way (REQ-019 AC-1, AC-4).
|
|
if result is LedgerWriteResult.DUPLICATE_OK:
|
|
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
|
|
return OutboxOutcome.DUPLICATE_COMPLETED
|
|
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
|
|
return OutboxOutcome.COMPLETED
|
|
|
|
|
|
# --- in-memory outbox (tests + slice-1 runtime) ------------------------------
|
|
|
|
class InMemoryLedgerOutbox:
|
|
"""Process-local outbox. Thread-safe; idempotent on correlation_id."""
|
|
|
|
def __init__(self) -> None:
|
|
import threading
|
|
|
|
self._rows: dict[str, LedgerWorkItem] = {}
|
|
self._order: list[str] = []
|
|
self._lock = threading.Lock()
|
|
|
|
@staticmethod
|
|
def _now() -> datetime:
|
|
return datetime.now(tz=timezone.utc)
|
|
|
|
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
|
|
with self._lock:
|
|
cid = trade.correlation_id
|
|
existing = self._rows.get(cid)
|
|
if existing is not None:
|
|
return existing
|
|
row = LedgerWorkItem(
|
|
correlation_id=cid,
|
|
trade=trade,
|
|
status=WorkItemStatus.PENDING,
|
|
created_at=self._now(),
|
|
updated_at=self._now(),
|
|
)
|
|
self._rows[cid] = row
|
|
self._order.append(cid)
|
|
return row
|
|
|
|
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
|
|
with self._lock:
|
|
out: list[LedgerWorkItem] = []
|
|
for cid in self._order:
|
|
row = self._rows[cid]
|
|
if row.status is WorkItemStatus.PENDING:
|
|
out.append(row)
|
|
if len(out) >= limit:
|
|
break
|
|
return out
|
|
|
|
def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
|
|
with self._lock:
|
|
return self._rows.get(correlation_id)
|
|
|
|
def mark_outcome(
|
|
self,
|
|
correlation_id: str,
|
|
outcome: OutboxOutcome,
|
|
*,
|
|
error: Optional[str] = None,
|
|
) -> LedgerWorkItem:
|
|
with self._lock:
|
|
row = self._rows[correlation_id]
|
|
if outcome is OutboxOutcome.COMPLETED:
|
|
new_status = WorkItemStatus.COMPLETED
|
|
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
|
|
new_status = WorkItemStatus.CONFLICT_BLOCKED
|
|
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
|
|
# already terminal; keep COMPLETED status
|
|
new_status = WorkItemStatus.COMPLETED
|
|
else: # retryable
|
|
new_status = WorkItemStatus.PENDING
|
|
updated = replace(
|
|
row,
|
|
status=new_status,
|
|
attempts=row.attempts + (0 if outcome is OutboxOutcome.DUPLICATE_COMPLETED else 1),
|
|
last_outcome=outcome,
|
|
last_error=error,
|
|
updated_at=self._now(),
|
|
)
|
|
self._rows[correlation_id] = updated
|
|
return updated
|
|
|
|
# --- introspection ---
|
|
def all_items(self) -> list[LedgerWorkItem]:
|
|
with self._lock:
|
|
return [self._rows[cid] for cid in self._order]
|