fix(ledger): dcy-015 outbox 영속 오류 분류 + 백오프/dead-letter (BAT-019)

portfolio-hub 404(미등록 포트폴리오)가 retryable 로 분류되어 백오프·상한 없이
초당 영구 재시도되던 버그(issue #13, 실측 930회) 정정.

거버넌스: 전부 change-req (신규 REQ 없음, dcy-015 OQ-3; 백오프 env 는 REQ-019 흡수).
- REQ-017: 404/400/422 → LedgerPermanentError 분류 (409 conflict / 408·429·5xx·
  network retryable 불변). LedgerPort 에 permanent 타입 추가
- REQ-019: 영속 → 터미널 delivery_failed 즉시(재시도 0). retryable 지수 백오프
  +jitter+next_retry_at, max_attempts 초과 시 dead-letter delivery_failed.
  list_pending 이 백오프 행 스킵(head-of-line 보호). 마이그레이션 007(status enum
  +next_retry_at). 백오프 env(base/cap/max-attempts)
- REQ-018: 터미널/dead-letter WARN + WriterStats.delivery_failed 카운터 가시화
- REQ-020: runtime-flows/data-model/interface-contract 정합

단일 터미널 delivery_failed(permanent/dead-letter, last_outcome 로 사유 구분) —
conflict_blocked 패턴과 동형. dead-letter 는 운영자 재처리 가능 상태.

검증: 단위 311 passed/34 skipped, 인프라 게이트(PG+RabbitMQ) 345 passed, ruff clean.
신규 test_ledger_outbox_backoff + adapter/PG 테스트, 기존 retryable 백오프 정합.
BAT-019 release-candidate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-08 16:12:17 +09:00
co-authored by Claude Opus 4.8
parent 0b7db66463
commit b851b927ee
29 changed files with 1069 additions and 40 deletions
+9
View File
@@ -5,6 +5,7 @@ import threading
from trex_core.application.ledger_port import (
LedgerConflictError,
LedgerPermanentError,
LedgerRetryableError,
LedgerTrade,
LedgerWriteResult,
@@ -21,6 +22,8 @@ class InMemoryLedger:
then accept.
conflict_executions -- execution_ids that raise LedgerConflictError
(simulates portfolio-hub 409).
permanent_executions -- execution_ids that raise LedgerPermanentError
(simulates portfolio-hub 404/400/422, dcy-015).
"""
def __init__(
@@ -28,12 +31,14 @@ class InMemoryLedger:
*,
fail_n_then_succeed: int = 0,
conflict_executions: set[str] | None = None,
permanent_executions: set[str] | None = None,
) -> None:
self._trades: dict[str, LedgerTrade] = {}
self._order: list[LedgerTrade] = []
self._lock = threading.Lock()
self._fail_remaining = fail_n_then_succeed
self._conflicts = set(conflict_executions or set())
self._permanent = set(permanent_executions or set())
def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
with self._lock:
@@ -41,6 +46,10 @@ class InMemoryLedger:
raise LedgerConflictError(
f"ledger conflict for execution_id={trade.execution_id}"
)
if trade.execution_id in self._permanent:
raise LedgerPermanentError(
f"ledger permanent error for execution_id={trade.execution_id}"
)
if self._fail_remaining > 0:
self._fail_remaining -= 1
raise LedgerRetryableError("simulated transient failure")
+15 -1
View File
@@ -6,7 +6,9 @@ Live portfolio-hub contract (verified against openapi, REQ-017 CHG-04):
execution_id, order_id
- 2xx (201) => recorded (or duplicate accepted idempotently)
- 409 => LedgerConflictError (operator resolution)
- 4xx (other)/5xx => LedgerRetryableError (outbox retries)
- 404/400/422 => LedgerPermanentError (dcy-015/REQ-017 — retry can't fix:
portfolio not registered / malformed trade)
- 4xx (other)/5xx => LedgerRetryableError (outbox backoff retries)
- httpx error => LedgerRetryableError
`asset_id` on LedgerTrade carries the portfolio_id value (REQ-018:
@@ -21,11 +23,15 @@ import httpx
from trex_core.application.ledger_port import (
LedgerConflictError,
LedgerPermanentError,
LedgerRetryableError,
LedgerTrade,
LedgerWriteResult,
)
# dcy-015 / REQ-017: client errors that retrying cannot resolve.
_PERMANENT_STATUS = frozenset({400, 404, 422})
class PortfolioHubLedger:
"""portfolio-hub ledger adapter (single endpoint, single responsibility)."""
@@ -91,6 +97,14 @@ class PortfolioHubLedger:
f"portfolio-hub 409 conflict for execution_id={trade.execution_id}: "
f"{resp.text[:200]}"
)
if resp.status_code in _PERMANENT_STATUS:
# dcy-015 / REQ-017 (issue #13): 404/400/422 won't be fixed by
# retrying — surface as permanent so the outbox terminates the row.
raise LedgerPermanentError(
f"portfolio-hub {resp.status_code} permanent error for "
f"asset_id={trade.asset_id} execution_id={trade.execution_id}: "
f"{resp.text[:200]}"
)
raise LedgerRetryableError(
f"portfolio-hub {resp.status_code} response: {resp.text[:200]}"
)
+14 -4
View File
@@ -325,7 +325,7 @@ class _PgIntentRecord:
_OUTBOX_COLUMNS = (
"correlation_id, asset_id, order_id, execution_id, side, qty, price, "
"fee, tax, executed_at, status, attempts, last_outcome, last_error, "
"created_at, updated_at"
"created_at, updated_at, next_retry_at"
)
@@ -362,7 +362,9 @@ class PostgreSQLLedgerOutbox:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox "
"WHERE status = 'pending' ORDER BY created_at LIMIT %s",
"WHERE status = 'pending' "
"AND (next_retry_at IS NULL OR next_retry_at <= now()) "
"ORDER BY created_at LIMIT %s",
(limit,),
)
return [self._row_to_item(r) for r in cur.fetchall()]
@@ -383,6 +385,7 @@ class PostgreSQLLedgerOutbox:
outcome: OutboxOutcome,
*,
error: Optional[str] = None,
next_retry_at: Optional[datetime] = None,
) -> LedgerWorkItem:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
@@ -392,18 +395,24 @@ class PostgreSQLLedgerOutbox:
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
status_value = WorkItemStatus.CONFLICT_BLOCKED.value
attempts_delta = 1
elif outcome in (OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED):
# dcy-015 / REQ-019: terminal delivery_failed (reason in last_outcome).
status_value = WorkItemStatus.DELIVERY_FAILED.value
attempts_delta = 1
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
status_value = WorkItemStatus.COMPLETED.value
attempts_delta = 0
else: # retryable
status_value = WorkItemStatus.PENDING.value
attempts_delta = 1
# next_retry_at only persists for a retryable backoff; cleared otherwise.
nra = next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
cur.execute(
"UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
"last_outcome=%s, last_error=%s, updated_at=%s "
"last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s "
"WHERE correlation_id=%s",
(
status_value, attempts_delta, outcome.value, error,
status_value, attempts_delta, outcome.value, error, nra,
datetime.now(tz=timezone.utc), correlation_id,
),
)
@@ -429,4 +438,5 @@ class PostgreSQLLedgerOutbox:
last_error=row[13],
created_at=row[14],
updated_at=row[15],
next_retry_at=row[16],
)
@@ -0,0 +1,25 @@
-- BAT-019 / dcy-015 / REQ-019 (issue #13): ledger_outbox terminal failure +
-- backoff scheduling. Adds:
-- - next_retry_at: earliest re-attempt time for a retryable row (backoff).
-- - status 'delivery_failed': single terminal for permanent (404/400/422)
-- and dead-lettered (max_attempts exhausted) rows; reason in last_outcome.
-- - last_outcome 'permanent_failure'/'dead_lettered'.
-- Idempotent: ADD COLUMN IF NOT EXISTS + DROP/ADD CONSTRAINT (re-runnable since
-- migrations are re-applied by glob).
ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ;
ALTER TABLE ledger_outbox DROP CONSTRAINT IF EXISTS ledger_outbox_status_check;
ALTER TABLE ledger_outbox ADD CONSTRAINT ledger_outbox_status_check
CHECK (status IN ('pending','completed','conflict_blocked','delivery_failed'));
ALTER TABLE ledger_outbox DROP CONSTRAINT IF EXISTS ledger_outbox_last_outcome_check;
ALTER TABLE ledger_outbox ADD CONSTRAINT ledger_outbox_last_outcome_check
CHECK (last_outcome IS NULL OR last_outcome IN (
'completed','duplicate_completed','retryable_failure','conflict_blocked',
'permanent_failure','dead_lettered'
));
-- Partial index so the writer's pending scan skips rows still in backoff cheaply.
CREATE INDEX IF NOT EXISTS idx_ledger_outbox_pending_retry
ON ledger_outbox (next_retry_at)
WHERE status = 'pending';
+108 -12
View File
@@ -9,39 +9,64 @@ 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.
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, timezone
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Optional, Protocol
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)
@@ -58,6 +83,9 @@ class LedgerWorkItem:
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
@@ -83,8 +111,12 @@ class LedgerOutbox(Protocol):
outcome: OutboxOutcome,
*,
error: Optional[str] = None,
next_retry_at: Optional[datetime] = None,
) -> LedgerWorkItem:
"""Record the outcome of a writer attempt and advance status."""
"""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)."""
...
@@ -94,6 +126,7 @@ class WriterStats:
duplicates: int = 0
retryable: int = 0
conflicts: int = 0
delivery_failed: int = 0 # dcy-015: permanent + dead-lettered terminals
class LedgerWriter:
@@ -104,9 +137,25 @@ class LedgerWriter:
the adapter (REQ-019 AC-4).
"""
def __init__(self, outbox: LedgerOutbox, ledger: LedgerPort) -> None:
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)
@@ -123,8 +172,15 @@ class LedgerWriter:
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.
@@ -139,9 +195,36 @@ class LedgerWriter:
item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
)
return OutboxOutcome.CONFLICT_BLOCKED
except LedgerRetryableError as exc:
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.RETRYABLE_FAILURE, error=str(exc)
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
@@ -159,16 +242,18 @@ class LedgerWriter:
class InMemoryLedgerOutbox:
"""Process-local outbox. Thread-safe; idempotent on correlation_id."""
def __init__(self) -> None:
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))
@staticmethod
def _now() -> datetime:
return datetime.now(tz=timezone.utc)
def _now(self) -> datetime:
return self._clock()
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
with self._lock:
@@ -188,11 +273,16 @@ class InMemoryLedgerOutbox:
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]
if row.status is WorkItemStatus.PENDING:
# 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
@@ -208,6 +298,7 @@ class InMemoryLedgerOutbox:
outcome: OutboxOutcome,
*,
error: Optional[str] = None,
next_retry_at: Optional[datetime] = None,
) -> LedgerWorkItem:
with self._lock:
row = self._rows[correlation_id]
@@ -215,6 +306,8 @@ class InMemoryLedgerOutbox:
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
@@ -226,6 +319,9 @@ class InMemoryLedgerOutbox:
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
+10
View File
@@ -63,6 +63,16 @@ class LedgerRetryableError(LedgerError):
loop should re-attempt this work item on the next tick (REQ-019 AC-2)."""
class LedgerPermanentError(LedgerError):
"""Non-retryable client error (dcy-015 / REQ-017, issue #13).
portfolio-hub returned a status that retrying cannot resolve — e.g. 404
(portfolio not registered), 400/422 (malformed/invalid trade). The outbox
must move the row to a terminal `delivery_failed` state immediately instead
of retrying forever. Distinct from LedgerConflictError (409) which is its
own terminal state."""
class LedgerPort(Protocol):
"""Ledger adapter boundary (REQ-017 AC-1/2).
+12
View File
@@ -155,6 +155,11 @@ class DaemonConfig:
# per-instance TOML (REQ-048); these are the env-only defaults.
trading_timezone: str = "Asia/Seoul"
daily_preallocation_time: str = "00:00"
# dcy-015 / REQ-019 (BAT-019): ledger outbox retry backoff. base*2^(n-1)
# capped at cap (+jitter); after max_attempts a retryable row dead-letters.
ledger_retry_base_sec: float = 1.0
ledger_retry_cap_sec: float = 300.0
ledger_max_attempts: int = 8
@staticmethod
def from_env() -> "DaemonConfig":
@@ -237,6 +242,13 @@ class DaemonConfig:
pending_stale_threshold_sec=int(
os.environ.get("TREX_PENDING_STALE_THRESHOLD", "0")
),
ledger_retry_base_sec=float(
os.environ.get("TREX_LEDGER_RETRY_BASE_SEC", "1.0")
),
ledger_retry_cap_sec=float(
os.environ.get("TREX_LEDGER_RETRY_CAP_SEC", "300.0")
),
ledger_max_attempts=int(os.environ.get("TREX_LEDGER_MAX_ATTEMPTS", "8")),
)
@staticmethod
+6 -1
View File
@@ -147,7 +147,12 @@ class Daemon:
)
self.broker = _build_broker(config)
self.outbox = PostgreSQLLedgerOutbox(config.dsn)
self.ledger_writer = LedgerWriter(self.outbox, _build_ledger(config))
self.ledger_writer = LedgerWriter(
self.outbox, _build_ledger(config),
retry_base_sec=config.ledger_retry_base_sec,
retry_cap_sec=config.ledger_retry_cap_sec,
max_attempts=config.ledger_max_attempts,
)
# BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
self.snapshot_store = SnapshotStore()