286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""LedgerOutbox + LedgerWriter direct unit tests (REQ-019)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import replace
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from trex_core.adapters.ledger.in_memory import InMemoryLedger
|
|
from trex_core.application.ledger_outbox import (
|
|
InMemoryLedgerOutbox,
|
|
LedgerWriter,
|
|
OutboxClaimLostError,
|
|
OutboxOutcome,
|
|
WorkItemStatus,
|
|
)
|
|
from trex_core.application.ledger_port import LedgerPermanentError, LedgerTrade
|
|
|
|
|
|
def _trade(execution_id="paper:PAPER-1:10") -> LedgerTrade:
|
|
return LedgerTrade(
|
|
asset_id="PF-1", order_id="PAPER-1", execution_id=execution_id,
|
|
side="buy", qty=10, price=1000.0,
|
|
executed_at=datetime(2026, 6, 4, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
def test_enqueue_is_idempotent_on_correlation_id():
|
|
outbox = InMemoryLedgerOutbox(instance_key="test-instance")
|
|
a = outbox.enqueue(_trade())
|
|
b = outbox.enqueue(_trade())
|
|
assert a.correlation_id == b.correlation_id
|
|
assert len(outbox.all_items()) == 1
|
|
|
|
|
|
def test_writer_run_once_drains_pending_and_marks_completed():
|
|
outbox = InMemoryLedgerOutbox(instance_key="test-instance")
|
|
ledger = InMemoryLedger()
|
|
writer = LedgerWriter(outbox, ledger)
|
|
outbox.enqueue(_trade(execution_id="e-1"))
|
|
outbox.enqueue(_trade(execution_id="e-2"))
|
|
stats = writer.run_once()
|
|
assert stats.completed == 2 and ledger.count() == 2
|
|
assert all(it.status is WorkItemStatus.COMPLETED for it in outbox.all_items())
|
|
assert all(it.last_outcome is OutboxOutcome.COMPLETED for it in outbox.all_items())
|
|
|
|
|
|
def test_writer_marks_conflict_blocked_and_does_not_retry():
|
|
outbox = InMemoryLedgerOutbox(instance_key="test-instance")
|
|
ledger = InMemoryLedger(conflict_executions={"e-x"})
|
|
writer = LedgerWriter(outbox, ledger)
|
|
outbox.enqueue(_trade(execution_id="e-x"))
|
|
writer.run_once()
|
|
[item] = outbox.all_items()
|
|
assert item.status is WorkItemStatus.CONFLICT_BLOCKED
|
|
assert item.last_outcome is OutboxOutcome.CONFLICT_BLOCKED
|
|
# No pending items remain, so a subsequent run is a no-op.
|
|
assert outbox.list_pending() == []
|
|
|
|
|
|
def test_writer_retries_until_success():
|
|
# dcy-015: retryable rows now back off; share an injected clock between the
|
|
# outbox (list_pending "now") and the writer (next_retry_at) to advance past it.
|
|
now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
|
|
outbox = InMemoryLedgerOutbox(
|
|
clock=lambda: now["t"], instance_key="test-instance"
|
|
)
|
|
ledger = InMemoryLedger(fail_n_then_succeed=1)
|
|
writer = LedgerWriter(outbox, ledger, clock=lambda: now["t"])
|
|
outbox.enqueue(_trade(execution_id="e-r"))
|
|
s1 = writer.run_once()
|
|
assert s1.retryable == 1 and ledger.count() == 0
|
|
assert writer.run_once().completed == 0 # still in backoff -> skipped
|
|
now["t"] += timedelta(seconds=600) # past the backoff window
|
|
s2 = writer.run_once()
|
|
assert s2.completed == 1 and ledger.count() == 1
|
|
|
|
|
|
def test_process_correlation_id_only_drains_requested_row():
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
ledger = InMemoryLedger()
|
|
writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
|
|
first = outbox.enqueue(_trade(execution_id="e-1"))
|
|
second = outbox.enqueue(_trade(execution_id="e-2"))
|
|
|
|
outcome = writer.process_correlation_id(second.correlation_id)
|
|
|
|
assert outcome is OutboxOutcome.COMPLETED
|
|
assert ledger.count() == 1
|
|
assert writer.counters["claim:claimed"] == 1
|
|
assert writer.counters["outcome:completed"] == 1
|
|
assert outbox.get(first.correlation_id).status is WorkItemStatus.PENDING
|
|
assert outbox.get(second.correlation_id).status is WorkItemStatus.COMPLETED
|
|
|
|
|
|
def test_process_correlation_id_skips_foreign_instance_row():
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
ledger = InMemoryLedger()
|
|
item = outbox.enqueue(_trade(execution_id="e-foreign"))
|
|
|
|
with pytest.raises(ValueError, match="must match"):
|
|
LedgerWriter(outbox, ledger, instance_key="inst-b")
|
|
assert ledger.count() == 0
|
|
assert outbox.get(item.correlation_id).status is WorkItemStatus.PENDING
|
|
|
|
|
|
def test_process_correlation_id_rejects_legacy_row_without_instance_owner():
|
|
class _LegacyOutbox(InMemoryLedgerOutbox):
|
|
def get(self, correlation_id):
|
|
item = super().get(correlation_id)
|
|
return replace(item, instance_key=None) if item is not None else None
|
|
|
|
outbox = _LegacyOutbox(instance_key="inst-a")
|
|
ledger = InMemoryLedger()
|
|
writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
|
|
item = outbox.enqueue(_trade())
|
|
|
|
assert writer.process_correlation_id(item.correlation_id) is None
|
|
assert ledger.count() == 0
|
|
|
|
|
|
def test_writer_exposes_claim_and_outcome_counters():
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
ledger = InMemoryLedger()
|
|
writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
|
|
outbox.enqueue(_trade(execution_id="e-count"))
|
|
|
|
stats = writer.run_once()
|
|
|
|
assert stats.completed == 1
|
|
assert writer.counters["claim:claimed"] == 1
|
|
assert writer.counters["outcome:completed"] == 1
|
|
|
|
|
|
def test_idle_day_is_silent_and_counted_per_instance(caplog):
|
|
writers = [
|
|
LedgerWriter(
|
|
InMemoryLedgerOutbox(instance_key=instance_key),
|
|
InMemoryLedger(),
|
|
instance_key=instance_key,
|
|
)
|
|
for instance_key in ("instance-a", "instance-b")
|
|
]
|
|
|
|
with caplog.at_level(logging.INFO, logger="trex_core.application.ledger_outbox"):
|
|
for writer in writers:
|
|
for _ in range(86_400):
|
|
writer.run_once()
|
|
|
|
assert [
|
|
writer.metrics_snapshot()["claims"]["empty_polls_total"]
|
|
for writer in writers
|
|
] == [86_400, 86_400]
|
|
assert "claim_result=empty" not in caplog.text
|
|
assert "correlation_ids" not in caplog.text
|
|
|
|
|
|
def test_nonempty_claim_log_has_ordered_json_correlation_ids(caplog):
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a")
|
|
first = outbox.enqueue(_trade(execution_id="e-1"))
|
|
second = outbox.enqueue(_trade(execution_id="e-2"))
|
|
|
|
with caplog.at_level(logging.INFO, logger="trex_core.application.ledger_outbox"):
|
|
writer.run_once(limit=2)
|
|
|
|
claim_logs = [
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if "ledger outbox claim:" in record.getMessage()
|
|
]
|
|
assert claim_logs == [
|
|
"ledger outbox claim: instance_key=inst-a claim_result=claimed "
|
|
f"count=2 limit=2 correlation_ids=[\"{first.correlation_id}\","
|
|
f"\"{second.correlation_id}\"]"
|
|
]
|
|
|
|
|
|
def test_metrics_snapshot_is_complete_and_detached():
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a")
|
|
|
|
snapshot = writer.metrics_snapshot()
|
|
|
|
assert snapshot == {
|
|
"available": True,
|
|
"scope": "process_lifetime",
|
|
"instance_key": "inst-a",
|
|
"claims": {
|
|
"empty_polls_total": 0,
|
|
"claimed_batches_total": 0,
|
|
"unavailable_total": 0,
|
|
"lost_total": 0,
|
|
"foreign_instance_total": 0,
|
|
},
|
|
"outcomes": {
|
|
"completed_total": 0,
|
|
"duplicate_completed_total": 0,
|
|
"retryable_failure_total": 0,
|
|
"conflict_blocked_total": 0,
|
|
"permanent_failure_total": 0,
|
|
"dead_lettered_total": 0,
|
|
},
|
|
}
|
|
snapshot["claims"]["empty_polls_total"] = 999
|
|
assert writer.metrics_snapshot()["claims"]["empty_polls_total"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("instance_key", ["", " "])
|
|
def test_writer_rejects_explicit_empty_instance_key(instance_key):
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
LedgerWriter(outbox, InMemoryLedger(), instance_key=instance_key)
|
|
|
|
|
|
@pytest.mark.parametrize("instance_key", ["", " "])
|
|
def test_outbox_rejects_empty_instance_key(instance_key):
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
InMemoryLedgerOutbox(instance_key=instance_key)
|
|
|
|
|
|
def test_in_memory_claim_is_exclusive_and_stale_lease_is_fenced():
|
|
now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
|
|
outbox = InMemoryLedgerOutbox(
|
|
instance_key="inst-a",
|
|
claim_ttl_sec=5.0,
|
|
clock=lambda: now["t"],
|
|
)
|
|
item = outbox.enqueue(_trade(execution_id="e-lease"))
|
|
|
|
[claim_a] = outbox.claim_pending()
|
|
assert outbox.claim_pending() == []
|
|
assert outbox.list_pending() == []
|
|
|
|
now["t"] += timedelta(seconds=6)
|
|
[claim_b] = outbox.claim_pending()
|
|
assert claim_b.claim_expires_at != claim_a.claim_expires_at
|
|
with pytest.raises(OutboxClaimLostError):
|
|
outbox.mark_outcome(
|
|
item.correlation_id,
|
|
OutboxOutcome.COMPLETED,
|
|
expected_claim_expires_at=claim_a.claim_expires_at,
|
|
)
|
|
|
|
stored = outbox.mark_outcome(
|
|
item.correlation_id,
|
|
OutboxOutcome.COMPLETED,
|
|
expected_claim_expires_at=claim_b.claim_expires_at,
|
|
)
|
|
assert stored.status is WorkItemStatus.COMPLETED
|
|
assert stored.claim_expires_at is None
|
|
|
|
|
|
def test_outcome_logs_do_not_render_exception_text(caplog):
|
|
sentinel = "TOKEN_SENTINEL_ACCOUNT_DSN_PAYLOAD"
|
|
|
|
class _FailingLedger:
|
|
def record_trade(self, _trade):
|
|
raise LedgerPermanentError(sentinel)
|
|
|
|
outbox = InMemoryLedgerOutbox(instance_key="inst-a")
|
|
writer = LedgerWriter(outbox, _FailingLedger(), instance_key="inst-a")
|
|
outbox.enqueue(_trade(execution_id="e-sensitive"))
|
|
|
|
with caplog.at_level("INFO"):
|
|
stats = writer.run_once()
|
|
|
|
assert stats.delivery_failed == 1
|
|
assert sentinel not in caplog.text
|
|
|
|
|
|
def test_lost_claim_is_counted_without_false_completion():
|
|
class _ClaimLosingOutbox(InMemoryLedgerOutbox):
|
|
def mark_outcome(self, correlation_id, outcome, **kwargs):
|
|
raise OutboxClaimLostError(correlation_id)
|
|
|
|
outbox = _ClaimLosingOutbox(instance_key="inst-a")
|
|
outbox.enqueue(_trade())
|
|
writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a")
|
|
|
|
stats = writer.run_once()
|
|
|
|
assert stats.claim_lost == 1
|
|
assert stats.completed == 0
|