trex-core/tests/application/test_ledger_outbox_backoff.py

"""Ledger outbox permanent-error + backoff + dead-letter tests (dcy-015 / issue #13).

A programmable LedgerPort raises chosen errors so we can drive: 404→permanent
terminal (no retry), retryable→backoff (next_retry_at skip), max_attempts→
dead-letter terminal, and head-of-line protection.
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

from trex_core.application.ledger_outbox import (
    InMemoryLedgerOutbox,
    LedgerWriter,
    OutboxOutcome,
    WorkItemStatus,
)
from trex_core.application.ledger_port import (
    LedgerPermanentError,
    LedgerRetryableError,
    LedgerTrade,
    LedgerWriteOutcome,
    LedgerWriteResult,
)

UTC = timezone.utc


def _trade(cid_suffix="1", asset="PF-1") -> LedgerTrade:
    return LedgerTrade(
        asset_id=asset, order_id=f"O-{cid_suffix}", execution_id=f"e-{cid_suffix}",
        side="buy", qty=10, price=1000.0,
        executed_at=datetime(2026, 6, 4, tzinfo=UTC),
    )


class _ProgrammableLedger:
    """LedgerPort double: `behavior(trade)` returns a result or raises."""

    def __init__(self, behavior):
        self.behavior = behavior
        self.calls = 0

    def record_trade(self, trade) -> LedgerWriteOutcome:
        self.calls += 1
        return self.behavior(trade)


def _wire(behavior, *, clock, **writer_kw):
    outbox = InMemoryLedgerOutbox(clock=clock)
    ledger = _ProgrammableLedger(behavior)
    writer = LedgerWriter(outbox, ledger, clock=clock, rng=lambda: 0.0, **writer_kw)
    return outbox, ledger, writer


def test_permanent_error_terminates_immediately_no_retry():
    """S-1: 404/permanent → delivery_failed terminal, adapter called exactly once."""
    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}

    def behavior(trade):
        raise LedgerPermanentError("404 portfolio not registered")

    outbox, ledger, writer = _wire(behavior, clock=lambda: now["t"])
    outbox.enqueue(_trade())
    s = writer.run_once()
    assert s.delivery_failed == 1 and s.retryable == 0
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.DELIVERY_FAILED
    assert item.last_outcome is OutboxOutcome.PERMANENT_FAILURE
    # no further attempts — terminal, not pending
    now["t"] += timedelta(seconds=3600)
    assert writer.run_once().delivery_failed == 0
    assert ledger.calls == 1  # never retried


def test_retryable_backoff_skips_until_next_retry_at():
    """S-2/S-4: a retryable row backs off and is skipped by list_pending until due."""
    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}

    def behavior(trade):
        raise LedgerRetryableError("503 upstream")

    outbox, ledger, writer = _wire(
        behavior, clock=lambda: now["t"], retry_base_sec=10.0, max_attempts=99,
    )
    outbox.enqueue(_trade())
    assert writer.run_once().retryable == 1
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.PENDING and item.next_retry_at is not None
    # within backoff window -> skipped (no adapter call)
    calls_before = ledger.calls
    assert writer.run_once().retryable == 0
    assert ledger.calls == calls_before
    # advance past backoff -> retried
    now["t"] += timedelta(seconds=60)
    assert writer.run_once().retryable == 1
    assert ledger.calls == calls_before + 1


def test_retryable_dead_letters_after_max_attempts():
    """S-2: retryable exhausts max_attempts → delivery_failed (dead-letter)."""
    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}

    def behavior(trade):
        raise LedgerRetryableError("503 upstream")

    outbox, ledger, writer = _wire(
        behavior, clock=lambda: now["t"], retry_base_sec=1.0, retry_cap_sec=2.0,
        max_attempts=3,
    )
    outbox.enqueue(_trade())
    # attempt 1, 2 -> retryable; attempt 3 -> dead-letter
    assert writer.run_once().retryable == 1
    now["t"] += timedelta(seconds=10)
    assert writer.run_once().retryable == 1
    now["t"] += timedelta(seconds=10)
    s = writer.run_once()
    assert s.delivery_failed == 1
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.DELIVERY_FAILED
    assert item.last_outcome is OutboxOutcome.DEAD_LETTERED
    assert item.attempts == 3


def test_permanent_failure_emits_warning(caplog):
    """S-3: terminal delivery_failed is visible (WARN), not silent."""
    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}

    def behavior(trade):
        raise LedgerPermanentError("404 not found")

    outbox, ledger, writer = _wire(behavior, clock=lambda: now["t"])
    outbox.enqueue(_trade())
    with caplog.at_level(logging.WARNING):
        writer.run_once()
    assert any("delivery_failed" in r.message for r in caplog.records)


def test_backoff_row_does_not_block_new_rows():
    """S-4: a backing-off row does not occupy the FIFO batch; new rows proceed."""
    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}

    def behavior(trade):
        if trade.asset_id == "PF-bad":
            raise LedgerRetryableError("503")
        return LedgerWriteOutcome(LedgerWriteResult.NEW)

    outbox, ledger, writer = _wire(
        behavior, clock=lambda: now["t"], retry_base_sec=100.0, max_attempts=99,
    )
    outbox.enqueue(_trade(cid_suffix="bad", asset="PF-bad"))
    writer.run_once()  # PF-bad -> backoff
    outbox.enqueue(_trade(cid_suffix="good", asset="PF-good"))
    s = writer.run_once()  # PF-bad skipped (backoff), PF-good completes
    assert s.completed == 1 and s.retryable == 0
    good = outbox.get("PF-good:O-good:e-good")
    assert good.status is WorkItemStatus.COMPLETED