trex-core/trex_core/application/ledger_outbox.py

"""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 dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Callable, Optional, Protocol

from trex_core.application.ledger_port import (
    LedgerConflictError,
    LedgerPermanentError,
    LedgerPort,
    LedgerRetryableError,
    LedgerTrade,
    LedgerWriteResult,
)

_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)


# Outcomes that move a row to the terminal delivery_failed state.
_DELIVERY_FAILED_OUTCOMES = frozenset(
    {OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}
)


@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


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,
        next_retry_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


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,
        *,
        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,
    ) -> None:
        self._outbox = outbox
        self._ledger = ledger
        # 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

    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
            elif outcome in _DELIVERY_FAILED_OUTCOMES:
                stats.delivery_failed += 1
        return stats

    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._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 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)
            )
            _log.warning(
                "ledger delivery_failed (permanent): correlation_id=%s reason=%s",
                item.correlation_id, exc,
            )
            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)
                )
                _log.warning(
                    "ledger delivery_failed (dead-letter, attempts=%d): "
                    "correlation_id=%s reason=%s",
                    new_attempts, item.correlation_id, exc,
                )
                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,
            )
            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, *, clock: Optional[Callable[[], datetime]] = None) -> 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))

    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(),
            )
            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.next_retry_at is None or row.next_retry_at <= now
                ):
                    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,
        next_retry_at: Optional[datetime] = 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 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(),
            )
            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]