652 lines
26 KiB
Python
652 lines
26 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; 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 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 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 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}")
|
|
_log.info(
|
|
"ledger outbox claim: instance_key=%s claim_result=%s count=%d limit=%d",
|
|
self._instance_key,
|
|
claim_result,
|
|
len(items),
|
|
limit,
|
|
)
|
|
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]
|