"""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; backoff + retry (next_retry_at). conflict_blocked -- portfolio-hub 409; operator must resolve, no retry. permanent_failure -- 404/400/422 client error; terminal immediately (dcy-015). dead_lettered -- retryable that exhausted max_attempts; terminal (dcy-015). `permanent_failure` and `dead_lettered` both land in the single terminal `delivery_failed` status (distinguished by last_outcome/last_error), mirroring the `conflict_blocked` single-status pattern (dcy-015 OQ-2). This separation (orbit `LedgerWriter` pattern) keeps slot transitions on the critical path while making ledger writes eventually-consistent and idempotent. """ from __future__ import annotations import json import logging import random from collections.abc import Callable from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from enum import Enum from typing import Any, Optional, Protocol from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.ledger_port import ( LedgerConflictError, LedgerPermanentError, LedgerPort, LedgerRetryableError, LedgerTrade, LedgerWriteResult, ) from trex_core.application.ledger_skip import publish_ledger_skip_alarm _log = logging.getLogger(__name__) # dcy-015 / REQ-019: exponential backoff defaults (overridable via config/env). _DEFAULT_RETRY_BASE_SEC = 1.0 _DEFAULT_RETRY_CAP_SEC = 300.0 _DEFAULT_MAX_ATTEMPTS = 8 class OutboxOutcome(str, Enum): COMPLETED = "completed" DUPLICATE_COMPLETED = "duplicate_completed" RETRYABLE_FAILURE = "retryable_failure" CONFLICT_BLOCKED = "conflict_blocked" PERMANENT_FAILURE = "permanent_failure" # dcy-015: 404/400/422 — terminal now DEAD_LETTERED = "dead_lettered" # dcy-015: retryable exhausted max_attempts 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) 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} ) def trade_cash_delta(trade: LedgerTrade) -> float: """Estimated cash impact of a trade (dcy-018 / REQ-029 OQ-4 estimate). buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax). The fee is the core's approximation, so this is an estimate that converges to the portfolio-hub authoritative value on option-D write-back.""" notional = float(trade.qty) * float(trade.price) fees = float(trade.fee) + float(trade.tax) if trade.side == "sell": return notional - fees return -(notional + fees) @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 # dcy-015 / REQ-019: earliest time a retryable row may be re-attempted. # list_pending skips rows whose next_retry_at is still in the future. next_retry_at: Optional[datetime] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None instance_key: Optional[str] = None claim_expires_at: Optional[datetime] = None 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).""" ... def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]: """Return pending work items in FIFO order, up to `limit`.""" ... def claim_pending( self, limit: int = 32, *, correlation_id: str | None = None, ) -> list[LedgerWorkItem]: """Atomically claim pending items for delivery.""" ... def pending_cash_delta(self) -> dict[str, float]: """Sum the estimated cash delta of ALL not-yet-settled (pending) rows, grouped by asset_id (dcy-018 / REQ-029, OQ-4 estimate overlay). buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax). Used to show an estimated cash before option-D write-back lands.""" ... def get(self, correlation_id: str) -> Optional[LedgerWorkItem]: ... def mark_outcome( self, correlation_id: str, outcome: OutboxOutcome, *, 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. For a retryable outcome `next_retry_at` schedules the backoff; terminal outcomes ignore it (dcy-015 / REQ-019).""" ... @dataclass class WriterStats: completed: int = 0 duplicates: int = 0 retryable: int = 0 conflicts: int = 0 delivery_failed: int = 0 # dcy-015: permanent + dead-lettered terminals claim_lost: 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, *, snapshot_store=None, retry_base_sec: float = _DEFAULT_RETRY_BASE_SEC, retry_cap_sec: float = _DEFAULT_RETRY_CAP_SEC, max_attempts: int = _DEFAULT_MAX_ATTEMPTS, clock: Optional[Callable[[], datetime]] = None, rng: Optional[Callable[[], float]] = None, alarm_emitter: Optional[AlarmEmitter] = None, ledger_skip_store=None, instance_key: str | None = None, ) -> None: self._outbox = outbox self._ledger = ledger # dcy-018 / REQ-019/030 (option D): SnapshotStore for cash/quantity # write-back. None => write-back disabled (e.g. slice-1 / tests). self._snapshot_store = snapshot_store # dcy-015 / REQ-019: exponential backoff + jitter + max-attempts cap. self._retry_base = retry_base_sec self._retry_cap = retry_cap_sec self._max_attempts = max_attempts self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) self._rng = rng or random.random # jitter source; inject for determinism # dcy-033 / REQ-100: best-effort alarm emitter. self._alarm_emitter = alarm_emitter self._ledger_skip_store = ledger_skip_store 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: self.counters[key] = self.counters.get(key, 0) + 1 def metrics_snapshot(self) -> dict[str, Any]: """Return a detached process-lifetime observability snapshot.""" return { "available": True, "scope": "process_lifetime", "instance_key": self._instance_key, "claims": { "empty_polls_total": self.counters.get("claim:empty", 0), "claimed_batches_total": self.counters.get("claim:claimed", 0), "unavailable_total": self.counters.get("claim:unavailable", 0), "lost_total": self.counters.get("claim:lost", 0), "foreign_instance_total": self.counters.get( "claim:foreign_instance", 0 ), }, "outcomes": { "completed_total": self.counters.get("outcome:completed", 0), "duplicate_completed_total": self.counters.get( "outcome:duplicate_completed", 0 ), "retryable_failure_total": self.counters.get( "outcome:retryable_failure", 0 ), "conflict_blocked_total": self.counters.get( "outcome:conflict_blocked", 0 ), "permanent_failure_total": self.counters.get( "outcome:permanent_failure", 0 ), "dead_lettered_total": self.counters.get( "outcome:dead_lettered", 0 ), }, } def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem: return self._outbox.enqueue(trade) def run_once(self, limit: int = 32) -> WriterStats: stats = WriterStats() items = self._outbox.claim_pending(limit) claim_result = "claimed" if items else "empty" self._count(f"claim:{claim_result}") if items: correlation_ids = json.dumps( [item.correlation_id for item in items], ensure_ascii=True, separators=(",", ":"), ) _log.info( "ledger outbox claim: instance_key=%s claim_result=claimed " "count=%d limit=%d correlation_ids=%s", self._instance_key, len(items), limit, correlation_ids, ) for item in items: 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: stats.duplicates += 1 elif outcome is OutboxOutcome.RETRYABLE_FAILURE: stats.retryable += 1 elif outcome is OutboxOutcome.CONFLICT_BLOCKED: stats.conflicts += 1 elif outcome in _DELIVERY_FAILED_OUTCOMES: stats.delivery_failed += 1 return stats def process_correlation_id(self, correlation_id: str) -> OutboxOutcome | None: """Process one specific outbox row immediately.""" item = self._outbox.get(correlation_id) if item is None: return None if not self._validate_instance(item): return None if item.status is WorkItemStatus.PENDING: claimed = self._outbox.claim_pending(1, correlation_id=correlation_id) if not claimed: self._count("claim:unavailable") _log.info( "ledger outbox claim: instance_key=%s claim_result=unavailable " "correlation_id=%s", self._instance_key, correlation_id, ) 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, ) 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%.""" base_delay = min(self._retry_base * (2 ** (attempts - 1)), self._retry_cap) return base_delay * (1.0 + 0.25 * self._rng()) 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._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED) return OutboxOutcome.DUPLICATE_COMPLETED try: outcome = self._ledger.record_trade(item.trade) except LedgerConflictError as exc: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc), expected_claim_expires_at=item.claim_expires_at, ) 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( "ledger_delivery_failed", f"원장 409 충돌 (conflict_blocked): {item.correlation_id}", fields={"correlation_id": item.correlation_id, "asset_id": item.trade.asset_id, "reason": "conflict_blocked"}, ) return OutboxOutcome.CONFLICT_BLOCKED except LedgerPermanentError as exc: # 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), expected_claim_expires_at=item.claim_expires_at, ) self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE) _log.warning( "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s", item.instance_key or self._instance_key, item.correlation_id, ) # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry. if self._alarm_emitter is not None: self._alarm_emitter.publish_alarm( "ledger_delivery_failed", f"원장 영구 실패 (permanent): {item.correlation_id}", fields={"correlation_id": item.correlation_id, "asset_id": item.trade.asset_id, "reason": "permanent_failure"}, ) return OutboxOutcome.PERMANENT_FAILURE except LedgerRetryableError as exc: # dcy-015 / REQ-019: backoff, or dead-letter once max_attempts hit. new_attempts = item.attempts + 1 if new_attempts >= self._max_attempts: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc), expected_claim_expires_at=item.claim_expires_at, ) self._log_outcome(item, OutboxOutcome.DEAD_LETTERED) _log.warning( "ledger delivery_failed (dead-letter, attempts=%d): " "instance_key=%s correlation_id=%s", new_attempts, item.instance_key or self._instance_key, item.correlation_id, ) # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries. if self._alarm_emitter is not None: self._alarm_emitter.publish_alarm( "ledger_delivery_failed", f"원장 데드레터 (attempts={new_attempts}): {item.correlation_id}", fields={"correlation_id": item.correlation_id, "asset_id": item.trade.asset_id, "attempts": new_attempts, "reason": "dead_lettered"}, ) return OutboxOutcome.DEAD_LETTERED next_retry_at = self._clock() + timedelta( seconds=self._backoff_delay(new_attempts) ) self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc), next_retry_at=next_retry_at, expected_claim_expires_at=item.claim_expires_at, ) 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 # is absolute, so applying it on duplicate is idempotent. best-effort — # never let a write-back failure block the ledger outcome (NFR-1). self._writeback(item.trade, outcome) # 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 outcome.result is LedgerWriteResult.DUPLICATE_OK: 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 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 def _log_outcome( self, item: LedgerWorkItem, outcome: OutboxOutcome, ) -> None: self._count(f"outcome:{outcome.value}") msg = "ledger outbox outcome: instance_key=%s correlation_id=%s outcome=%s" args: tuple[object, ...] = ( item.instance_key or self._instance_key, item.correlation_id, outcome.value, ) _log.info(msg, *args) def _resolve_ledger_skip(self, trade: LedgerTrade) -> None: if ( self._ledger_skip_store is None or self._instance_key is None or not trade.asset_id or not trade.order_id or not trade.execution_id ): return records = self._ledger_skip_store.resolve_for_delivery( instance_key=self._instance_key, portfolio_id=trade.asset_id, side=trade.side, order_id=trade.order_id, execution_id=trade.execution_id, resolution_reason="ledger_recorded", ) for record in records: publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=True) def _writeback(self, trade: LedgerTrade, outcome) -> None: """Option D cash/quantity write-back (dcy-018). best-effort; skips when no SnapshotStore is wired or the response omits authoritative values.""" if self._snapshot_store is None: return if outcome.cash_balance_after is None or outcome.asset_quantity_after is None: return # older hub / missing fields → leave snapshot as-is try: self._snapshot_store.update_cash_position( trade.asset_id, outcome.cash_balance_after, outcome.asset_quantity_after, ) 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, ) # --- in-memory outbox (tests + slice-1 runtime) ------------------------------ class InMemoryLedgerOutbox: """Process-local outbox. Thread-safe; idempotent on correlation_id.""" def __init__( self, *, clock: Optional[Callable[[], datetime]] = None, instance_key: str, claim_ttl_sec: float = 60.0, ) -> None: import threading self._rows: dict[str, LedgerWorkItem] = {} self._order: list[str] = [] self._lock = threading.Lock() # 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() 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(), instance_key=self._instance_key, ) self._rows[cid] = row self._order.append(cid) return row def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]: now = self._now() with self._lock: out: list[LedgerWorkItem] = [] for cid in self._order: 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.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: break return out def claim_pending( self, limit: int = 32, *, correlation_id: str | None = None, ) -> list[LedgerWorkItem]: now = self._now() lease_until = now + timedelta(seconds=self._claim_ttl_sec) with self._lock: 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: out: dict[str, float] = {} for row in self._rows.values(): if row.status is WorkItemStatus.PENDING: aid = row.trade.asset_id out[aid] = out.get(aid, 0.0) + trade_cash_delta(row.trade) 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, next_retry_at: Optional[datetime] = None, expected_claim_expires_at: Optional[datetime] = None, ) -> 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: new_status = WorkItemStatus.CONFLICT_BLOCKED elif outcome in _DELIVERY_FAILED_OUTCOMES: new_status = WorkItemStatus.DELIVERY_FAILED # dcy-015 terminal 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, next_retry_at=( 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 # --- introspection --- def all_items(self) -> list[LedgerWorkItem]: with self._lock: return [self._rows[cid] for cid in self._order]