diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 3a35cd3..592cc60 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -6,9 +6,10 @@ iteration: EXPECTED_SLOT_FIELDS (code contract) vs daily_slot actual columns """ from __future__ import annotations +from datetime import UTC, datetime + import psycopg import pytest -from datetime import datetime, timezone from trex_core.adapters.persistence.migrate import apply_migrations, migration_files from trex_core.adapters.persistence.postgresql import ( @@ -33,24 +34,22 @@ _EXPECTED_TABLES = { def _columns(dsn: str, table: str) -> set[str]: - with psycopg.connect(dsn) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT column_name FROM information_schema.columns " - "WHERE table_schema = 'public' AND table_name = %s", - (table,), - ) - return {r[0] for r in cur.fetchall()} + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = %s", + (table,), + ) + return {r[0] for r in cur.fetchall()} def test_migrations_create_expected_tables(pg_clean): assert migration_files(), "no migration files found" - with psycopg.connect(pg_clean) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT table_name FROM information_schema.tables WHERE table_schema='public'" - ) - tables = {r[0] for r in cur.fetchall()} + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT table_name FROM information_schema.tables WHERE table_schema='public'" + ) + tables = {r[0] for r in cur.fetchall()} assert _EXPECTED_TABLES <= tables @@ -61,20 +60,19 @@ def test_daily_slot_columns_match_expected_contract(pg_clean): def test_strategy_state_schema_matches_instance_map_contract(pg_clean): - with psycopg.connect(pg_clean) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT column_name, data_type, udt_name " - "FROM information_schema.columns " - "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'" - ) - columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()} - cur.execute( - "SELECT a.attname FROM pg_index i " - "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " - "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary" - ) - primary_key = {row[0] for row in cur.fetchall()} + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT column_name, data_type, udt_name " + "FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'" + ) + columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()} + cur.execute( + "SELECT a.attname FROM pg_index i " + "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " + "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary" + ) + primary_key = {row[0] for row in cur.fetchall()} assert columns == {"slot_id": ("uuid", "uuid"), "data": ("jsonb", "jsonb")} assert primary_key == {"slot_id"} @@ -111,13 +109,12 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean): apply_migrations(pg_clean) assert repo.fetch_slot(slot.slot_id).venue == "SOR" - with psycopg.connect(pg_clean) as conn: - with conn.cursor() as cur: - with pytest.raises(psycopg.errors.CheckViolation): - cur.execute( - "UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s", - (slot.slot_id,), - ) + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + with pytest.raises(psycopg.errors.CheckViolation): + cur.execute( + "UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s", + (slot.slot_id,), + ) def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): @@ -129,7 +126,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): side="buy", qty=1, price=1000, - executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc), + executed_at=datetime(2026, 8, 19, tzinfo=UTC), venue="NXT", order_venue="KRX", ) @@ -140,7 +137,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): side="buy", qty=1, price=1000, - executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc), + executed_at=datetime(2026, 8, 19, tzinfo=UTC), venue="KRX", order_venue="SOR", ) @@ -172,10 +169,39 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): assert legacy_row.trade.venue is None assert legacy_row.trade.order_venue == "SOR" + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + with pytest.raises(psycopg.errors.CheckViolation): + cur.execute( + "UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s", + (normal.correlation_id,), + ) + + +def test_migration_028_fail_closes_unowned_legacy_outbox_rows(pg_clean): + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") + trade = LedgerTrade( + asset_id="PF-LEGACY", + order_id="O-LEGACY", + execution_id="E-LEGACY", + side="buy", + qty=1, + price=1000, + executed_at=datetime(2026, 9, 12, tzinfo=UTC), + ) + outbox.enqueue(trade) + with psycopg.connect(pg_clean) as conn: with conn.cursor() as cur: - with pytest.raises(psycopg.errors.CheckViolation): - cur.execute( - "UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s", - (normal.correlation_id,), - ) + cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN instance_key DROP NOT NULL") + cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN instance_key DROP DEFAULT") + cur.execute( + "UPDATE ledger_outbox SET instance_key=NULL WHERE correlation_id=%s", + (trade.correlation_id,), + ) + conn.commit() + + apply_migrations(pg_clean) + stored = outbox.get(trade.correlation_id) + assert stored is not None + assert stored.instance_key == "__unowned__" + assert outbox.list_pending() == [] diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py index ef65a3c..6f51388 100644 --- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py +++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py @@ -4,7 +4,7 @@ Skipped without TREX_POSTGRESQL_TEST_DSN (same gate as other PG suites). """ from __future__ import annotations -from datetime import datetime, timezone +from datetime import UTC, datetime from trex_core.adapters.ledger.in_memory import InMemoryLedger from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox @@ -13,34 +13,40 @@ from trex_core.application.ledger_outbox import ( OutboxOutcome, WorkItemStatus, ) -from trex_core.application.ledger_port import LedgerTrade +from trex_core.application.ledger_port import ( + LedgerRetryableError, + LedgerTrade, + LedgerWriteOutcome, + LedgerWriteResult, +) def _trade(execution_id="paper:1:10") -> LedgerTrade: return LedgerTrade( asset_id="PF-1", order_id="ORD-1", execution_id=execution_id, side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0, - executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc), + executed_at=datetime(2026, 6, 4, tzinfo=UTC), venue="NXT", order_venue="SOR", ) def test_enqueue_persists_and_is_idempotent(pg_clean): - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") a = outbox.enqueue(_trade()) b = outbox.enqueue(_trade()) assert a.correlation_id == b.correlation_id pending = outbox.list_pending() assert len(pending) == 1 assert pending[0].status is WorkItemStatus.PENDING + assert pending[0].instance_key == "inst-a" assert pending[0].trade.venue == "NXT" assert pending[0].trade.order_venue == "SOR" def test_writer_completes_persisted_item(pg_clean): - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") ledger = InMemoryLedger() - writer = LedgerWriter(outbox, ledger) + writer = LedgerWriter(outbox, ledger, instance_key="inst-a") outbox.enqueue(_trade(execution_id="pg-1")) stats = writer.run_once() assert stats.completed == 1 and ledger.count() == 1 @@ -50,9 +56,9 @@ def test_writer_completes_persisted_item(pg_clean): def test_writer_marks_conflict_blocked_on_pg(pg_clean): - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") ledger = InMemoryLedger(conflict_executions={"pg-x"}) - writer = LedgerWriter(outbox, ledger) + writer = LedgerWriter(outbox, ledger, instance_key="inst-a") outbox.enqueue(_trade(execution_id="pg-x")) writer.run_once() row = outbox.get("PF-1:ORD-1:pg-x") @@ -61,10 +67,10 @@ def test_writer_marks_conflict_blocked_on_pg(pg_clean): def test_writer_retries_on_pg(pg_clean): - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") ledger = InMemoryLedger(fail_n_then_succeed=1) # dcy-015: base 0 => next_retry_at == now, so the row is immediately due again. - writer = LedgerWriter(outbox, ledger, retry_base_sec=0.0) + writer = LedgerWriter(outbox, ledger, retry_base_sec=0.0, instance_key="inst-a") outbox.enqueue(_trade(execution_id="pg-r")) writer.run_once() row1 = outbox.get("PF-1:ORD-1:pg-r") @@ -79,9 +85,9 @@ def test_writer_retries_on_pg(pg_clean): def test_writer_marks_delivery_failed_permanent_on_pg(pg_clean): """dcy-015 / REQ-019: 404/permanent → terminal delivery_failed persisted (migration 007 status enum + last_outcome). No retry.""" - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") ledger = InMemoryLedger(permanent_executions={"pg-perm"}) - writer = LedgerWriter(outbox, ledger) + writer = LedgerWriter(outbox, ledger, instance_key="inst-a") outbox.enqueue(_trade(execution_id="pg-perm")) s = writer.run_once() assert s.delivery_failed == 1 @@ -94,22 +100,112 @@ def test_writer_marks_delivery_failed_permanent_on_pg(pg_clean): def test_pending_cash_delta_on_pg(pg_clean): """dcy-018 / REQ-029: estimated cash delta of pending rows summed by asset (SQL).""" - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") outbox.enqueue(LedgerTrade(asset_id="PF-1", order_id="O-1", execution_id="cd-buy", side="buy", qty=10, price=1000.0, fee=1.0, tax=0.0, - executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc))) + executed_at=datetime(2026, 6, 4, tzinfo=UTC))) deltas = outbox.pending_cash_delta() assert deltas["PF-1"] == -(10 * 1000.0 + 1.0) # buy => cash down by notional+fee def test_backoff_row_skipped_by_pending_scan_on_pg(pg_clean): """dcy-015 / REQ-019: a future next_retry_at excludes the row from list_pending.""" - outbox = PostgreSQLLedgerOutbox(pg_clean) + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") ledger = InMemoryLedger(fail_n_then_succeed=99) # large base => next_retry_at well in the future - writer = LedgerWriter(outbox, ledger, retry_base_sec=600.0, max_attempts=99) + writer = LedgerWriter( + outbox, ledger, retry_base_sec=600.0, max_attempts=99, instance_key="inst-a" + ) outbox.enqueue(_trade(execution_id="pg-bo")) writer.run_once() # -> retryable, next_retry_at ~ now + 600s row = outbox.get("PF-1:ORD-1:pg-bo") assert row.status is WorkItemStatus.PENDING and row.next_retry_at is not None assert outbox.list_pending() == [] # still backing off -> skipped + + +def test_pending_claims_are_isolated_per_instance(pg_clean): + ledger = InMemoryLedger() + outbox_a = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") + outbox_b = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-b") + writer_a = LedgerWriter(outbox_a, ledger, instance_key="inst-a") + writer_b = LedgerWriter(outbox_b, ledger, instance_key="inst-b") + row_a = outbox_a.enqueue(_trade(execution_id="pg-a")) + row_b = outbox_b.enqueue(_trade(execution_id="pg-b")) + + stats_a = writer_a.run_once() + assert stats_a.completed == 1 and ledger.count() == 1 + assert outbox_a.get(row_a.correlation_id).status is WorkItemStatus.COMPLETED + assert outbox_b.get(row_b.correlation_id).status is WorkItemStatus.PENDING + + stats_b = writer_b.run_once() + assert stats_b.completed == 1 and ledger.count() == 2 + assert outbox_b.get(row_b.correlation_id).status is WorkItemStatus.COMPLETED + + +def test_claim_expiry_allows_safe_recovery_after_worker_crash(pg_clean): + now = {"t": datetime(2026, 6, 4, tzinfo=UTC)} + outbox_a = PostgreSQLLedgerOutbox( + pg_clean, + instance_key="inst-a", + claim_ttl_sec=5.0, + worker_id="worker-a", + clock=lambda: now["t"], + ) + outbox_b = PostgreSQLLedgerOutbox( + pg_clean, + instance_key="inst-a", + claim_ttl_sec=5.0, + worker_id="worker-b", + clock=lambda: now["t"], + ) + row = outbox_a.enqueue(_trade(execution_id="pg-crash")) + + claimed_a = outbox_a.claim_pending() + assert [item.correlation_id for item in claimed_a] == [row.correlation_id] + assert outbox_b.claim_pending() == [] + + now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC) + claimed_b = outbox_b.claim_pending() + assert [item.correlation_id for item in claimed_b] == [row.correlation_id] + + +def test_writer_records_retryable_403_without_cross_instance_delivery(pg_clean): + class _ForbiddenLedger: + def __init__(self): + self.calls: list[str] = [] + + def record_trade(self, trade): + self.calls.append(trade.execution_id) + raise LedgerRetryableError("portfolio-hub 403 Forbidden") + + outbox_a = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") + outbox_b = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-b") + writer_b = LedgerWriter( + outbox_b, _ForbiddenLedger(), retry_base_sec=0.0, instance_key="inst-b" + ) + row_a = outbox_a.enqueue(_trade(execution_id="pg-403-a")) + row_b = outbox_b.enqueue(_trade(execution_id="pg-403-b")) + + stats_b = writer_b.run_once() + + assert stats_b.retryable == 1 + assert outbox_b.get(row_b.correlation_id).last_outcome is OutboxOutcome.RETRYABLE_FAILURE + assert outbox_a.get(row_a.correlation_id).status is WorkItemStatus.PENDING + + +def test_writer_marks_duplicate_completed_from_adapter_response_on_pg(pg_clean): + class _DuplicateLedger: + def record_trade(self, _trade): + return LedgerWriteOutcome(LedgerWriteResult.DUPLICATE_OK) + + outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a") + writer = LedgerWriter(outbox, _DuplicateLedger(), instance_key="inst-a") + row = outbox.enqueue(_trade(execution_id="pg-dup")) + + outcome = writer.process_correlation_id(row.correlation_id) + + assert outcome is OutboxOutcome.DUPLICATE_COMPLETED + stored = outbox.get(row.correlation_id) + assert stored is not None + assert stored.status is WorkItemStatus.COMPLETED + assert stored.last_outcome is OutboxOutcome.DUPLICATE_COMPLETED diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py index 5ce7f71..e085f30 100644 --- a/tests/application/test_ledger_outbox.py +++ b/tests/application/test_ledger_outbox.py @@ -83,3 +83,30 @@ def test_process_correlation_id_only_drains_requested_row(): assert ledger.count() == 1 assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING + + +def test_process_correlation_id_skips_foreign_instance_row(): + outbox = InMemoryLedgerOutbox(instance_key="inst-a") + ledger = InMemoryLedger() + writer = LedgerWriter(outbox, ledger, instance_key="inst-b") + item = outbox.enqueue(_trade(execution_id="e-foreign")) + + outcome = writer.process_correlation_id(item.correlation_id) + + assert outcome is None + assert ledger.count() == 0 + assert writer.counters["claim:foreign_instance"] == 1 + assert outbox.get(item.correlation_id).status is WorkItemStatus.PENDING + + +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 diff --git a/tests/runtime/test_cli_ledger_backfill.py b/tests/runtime/test_cli_ledger_backfill.py index e09282f..17455ff 100644 --- a/tests/runtime/test_cli_ledger_backfill.py +++ b/tests/runtime/test_cli_ledger_backfill.py @@ -187,7 +187,11 @@ def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_p import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod - monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox()) + monkeypatch.setattr( + pg_mod, + "PostgreSQLLedgerOutbox", + lambda dsn, *, instance_key: _Outbox(), + ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( @@ -246,7 +250,11 @@ def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod - monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: outbox) + monkeypatch.setattr( + pg_mod, + "PostgreSQLLedgerOutbox", + lambda dsn, *, instance_key: outbox, + ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( @@ -310,7 +318,11 @@ def test_ledger_backfill_resolves_matching_skip(monkeypatch, tmp_path, capsys): import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod - monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox()) + monkeypatch.setattr( + pg_mod, + "PostgreSQLLedgerOutbox", + lambda dsn, *, instance_key: _Outbox(), + ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 6a52b3d..64cabc7 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -13,9 +13,10 @@ domain layer (NFR-1). from __future__ import annotations import json +from collections.abc import Iterator from contextlib import contextmanager from datetime import datetime, timedelta, timezone -from typing import Any, Iterator, Optional +from typing import Any, Optional from uuid import UUID, uuid4 from trex_core.application.ledger_outbox import ( @@ -26,9 +27,9 @@ from trex_core.application.ledger_outbox import ( from trex_core.application.ledger_port import LedgerTrade from trex_core.application.ports import IntentRecord from trex_core.application.strategy_state import ( + _STATE_CAP_BYTES, SetResult, StrategyStateEntry, - _STATE_CAP_BYTES, ) from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord @@ -565,16 +566,32 @@ 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, next_retry_at, venue, order_venue" + "created_at, updated_at, next_retry_at, venue, order_venue, instance_key" ) class PostgreSQLLedgerOutbox: """ledger_outbox-backed LedgerOutbox. correlation_id PK is the dedup gate.""" - def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: + def __init__( + self, + dsn: str, + *, + instance_key: str = "default", + claim_ttl_sec: float = 60.0, + worker_id: str | None = None, + clock=None, + psycopg_module: Any = None, + ) -> None: self._psycopg = _require_psycopg(psycopg_module) self._dsn = dsn + self._instance_key = instance_key + self._claim_ttl_sec = claim_ttl_sec + self._worker_id = worker_id or f"ledger-writer-{uuid4().hex}" + 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._psycopg.connect(self._dsn, autocommit=False) as conn: @@ -582,8 +599,8 @@ class PostgreSQLLedgerOutbox: cur.execute( "INSERT INTO ledger_outbox " "(correlation_id, asset_id, order_id, execution_id, side, " - " qty, price, fee, tax, executed_at, venue, order_venue) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) " + " qty, price, fee, tax, executed_at, venue, order_venue, instance_key) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) " "ON CONFLICT (correlation_id) DO NOTHING", ( trade.correlation_id, @@ -598,6 +615,7 @@ class PostgreSQLLedgerOutbox: trade.executed_at, trade.venue, trade.order_venue, + self._instance_key, ), ) conn.commit() @@ -611,12 +629,53 @@ class PostgreSQLLedgerOutbox: cur.execute( f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox " "WHERE status = 'pending' " + "AND instance_key = %s " "AND (next_retry_at IS NULL OR next_retry_at <= now()) " - "ORDER BY created_at LIMIT %s", - (limit,), + "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) " + "ORDER BY created_at, correlation_id LIMIT %s", + (self._instance_key, limit), ) return [self._row_to_item(r) for r in cur.fetchall()] + 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) + correlation_sql = "" + params: list[Any] = [self._instance_key] + if correlation_id is not None: + correlation_sql = "AND correlation_id = %s " + params.append(correlation_id) + params.extend([limit, self._worker_id, lease_until, now]) + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "WITH candidates AS (" + "SELECT correlation_id FROM ledger_outbox " + "WHERE status = 'pending' " + "AND instance_key = %s " + "AND (next_retry_at IS NULL OR next_retry_at <= now()) " + "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) " + f"{correlation_sql}" + "ORDER BY created_at, correlation_id " + "LIMIT %s " + "FOR UPDATE SKIP LOCKED" + ") " + "UPDATE ledger_outbox o " + "SET claim_owner = %s, claim_expires_at = %s, updated_at = %s " + "FROM candidates " + "WHERE o.correlation_id = candidates.correlation_id " + f"RETURNING {_OUTBOX_COLUMNS}", + params, + ) + rows = [self._row_to_item(r) for r in cur.fetchall()] + conn.commit() + return rows + def pending_cash_delta(self) -> dict[str, float]: # dcy-018 / REQ-029 (OQ-4): estimated cash delta of all pending rows by # asset_id. SUM in SQL: sell → +(qty*price - fee - tax), buy → negative. @@ -627,7 +686,10 @@ class PostgreSQLLedgerOutbox: "SUM(CASE WHEN side='sell' " " THEN (qty*price - fee - tax) " " ELSE -(qty*price + fee + tax) END) " - "FROM ledger_outbox WHERE status='pending' GROUP BY asset_id" + "FROM ledger_outbox " + "WHERE status='pending' AND instance_key=%s " + "GROUP BY asset_id", + (self._instance_key,), ) return {r[0]: float(r[1]) for r in cur.fetchall() if r[1] is not None} @@ -671,16 +733,20 @@ class PostgreSQLLedgerOutbox: 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, next_retry_at=%s, updated_at=%s " - "WHERE correlation_id=%s", + "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, " + "claim_owner=NULL, claim_expires_at=NULL " + "WHERE correlation_id=%s AND instance_key=%s " + "AND (claim_owner=%s OR claim_owner IS NULL)", ( status_value, attempts_delta, outcome.value, error, nra, - datetime.now(tz=timezone.utc), + self._now(), correlation_id, + self._instance_key, + self._worker_id, ), ) conn.commit() @@ -714,6 +780,7 @@ class PostgreSQLLedgerOutbox: created_at=row[14], updated_at=row[15], next_retry_at=row[16], + instance_key=row[19], ) diff --git a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql new file mode 100644 index 0000000..186c809 --- /dev/null +++ b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql @@ -0,0 +1,22 @@ +-- GitHub #84: persist outbox owner instance and add claim/lease fields so +-- shared-DB daemon writers only deliver their own rows and can safely recover +-- abandoned work after lease expiry. Existing rows remain fail-closed when the +-- owner instance cannot be inferred. +ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS instance_key TEXT; +ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS claim_owner TEXT; +ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS claim_expires_at TIMESTAMPTZ; + +UPDATE ledger_outbox +SET instance_key='__unowned__' +WHERE instance_key IS NULL; + +ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET DEFAULT '__unowned__'; +ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_pending + ON ledger_outbox (instance_key, status, created_at) + WHERE status = 'pending'; + +CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_claim_expires + ON ledger_outbox (instance_key, claim_expires_at) + WHERE status = 'pending'; diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index 574fe87..79e7b32 100644 --- a/trex_core/application/ledger_outbox.py +++ b/trex_core/application/ledger_outbox.py @@ -104,6 +104,7 @@ class LedgerWorkItem: next_retry_at: Optional[datetime] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None + instance_key: Optional[str] = None class LedgerOutbox(Protocol): @@ -118,6 +119,15 @@ class LedgerOutbox(Protocol): """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). @@ -191,13 +201,27 @@ class LedgerWriter: self._alarm_emitter = alarm_emitter self._ledger_skip_store = ledger_skip_store self._instance_key = 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() - for item in self._outbox.list_pending(limit): + 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: outcome = self._process(item) if outcome is OutboxOutcome.COMPLETED: stats.completed += 1 @@ -216,6 +240,32 @@ class LedgerWriter: item = self._outbox.get(correlation_id) if item is None: return None + if ( + self._instance_key is not None + and item.instance_key is not None + and item.instance_key != self._instance_key + ): + self._count("claim:foreign_instance") + _log.info( + "ledger outbox claim: instance_key=%s claim_result=foreign_instance " + "correlation_id=%s owner_instance_key=%s", + self._instance_key, + correlation_id, + item.instance_key, + ) + 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] return self._process(item) def _backoff_delay(self, attempts: int) -> float: @@ -229,6 +279,7 @@ class LedgerWriter: 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) + self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED) return OutboxOutcome.DUPLICATE_COMPLETED try: outcome = self._ledger.record_trade(item.trade) @@ -236,6 +287,7 @@ class LedgerWriter: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc) ) + self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)) # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed. if self._alarm_emitter is not None: self._alarm_emitter.publish_alarm( @@ -252,8 +304,10 @@ class LedgerWriter: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.PERMANENT_FAILURE, error=str(exc) ) + self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc)) _log.warning( - "ledger delivery_failed (permanent): correlation_id=%s reason=%s", + "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s reason=%s", + item.instance_key or self._instance_key, item.correlation_id, exc, ) # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry. @@ -273,10 +327,14 @@ class LedgerWriter: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc) ) + self._log_outcome(item, 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, + "instance_key=%s correlation_id=%s reason=%s", + new_attempts, + item.instance_key or self._instance_key, + item.correlation_id, + exc, ) # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries. if self._alarm_emitter is not None: @@ -296,6 +354,7 @@ class LedgerWriter: item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc), next_retry_at=next_retry_at, ) + self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc)) 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 @@ -308,11 +367,32 @@ class LedgerWriter: if outcome.result is LedgerWriteResult.DUPLICATE_OK: self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED) 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) self._resolve_ledger_skip(item.trade) + self._log_outcome(item, OutboxOutcome.COMPLETED) return OutboxOutcome.COMPLETED + def _log_outcome( + self, + item: LedgerWorkItem, + outcome: OutboxOutcome, + *, + error: str | None = None, + ) -> 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, + ) + if error is not None: + msg += " error=%s" + args = (*args, error) + _log.info(msg, *args) + def _resolve_ledger_skip(self, trade: LedgerTrade) -> None: if ( self._ledger_skip_store is None @@ -358,7 +438,12 @@ class LedgerWriter: class InMemoryLedgerOutbox: """Process-local outbox. Thread-safe; idempotent on correlation_id.""" - def __init__(self, *, clock: Optional[Callable[[], datetime]] = None) -> None: + def __init__( + self, + *, + clock: Optional[Callable[[], datetime]] = None, + instance_key: str = "default", + ) -> None: import threading self._rows: dict[str, LedgerWorkItem] = {} @@ -367,6 +452,7 @@ class InMemoryLedgerOutbox: # 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)) + self._instance_key = instance_key def _now(self) -> datetime: return self._clock() @@ -383,6 +469,7 @@ class InMemoryLedgerOutbox: status=WorkItemStatus.PENDING, created_at=self._now(), updated_at=self._now(), + instance_key=self._instance_key, ) self._rows[cid] = row self._order.append(cid) @@ -404,6 +491,17 @@ class InMemoryLedgerOutbox: break return out + def claim_pending( + self, + limit: int = 32, + *, + correlation_id: str | None = None, + ) -> list[LedgerWorkItem]: + pending = self.list_pending(limit if correlation_id is None else max(limit, 1)) + if correlation_id is None: + return pending[:limit] + return [row for row in pending if row.correlation_id == correlation_id][:1] + def pending_cash_delta(self) -> dict[str, float]: with self._lock: out: dict[str, float] = {} diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 128ef21..182eeea 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -1333,7 +1333,7 @@ def _cmd_ledger_backfill(args, cfg, confirm) -> int: _emit_pretty_json(report) return 1 - outbox = PostgreSQLLedgerOutbox(cfg.dsn) + outbox = PostgreSQLLedgerOutbox(cfg.dsn, instance_key=instance) if not cfg.portfolio_hub or not cfg.portfolio_hub.base_url: report["result"] = {"status": "rejected", "reason": "portfolio-hub ledger adapter is not configured"} report["audit_path"] = _write_ledger_backfill_audit(args, report) diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index d232973..95bcd66 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -190,7 +190,7 @@ class Daemon: instance_key=config.instance_key, ) - self.outbox = PostgreSQLLedgerOutbox(config.dsn) + self.outbox = PostgreSQLLedgerOutbox(config.dsn, instance_key=config.instance_key) self.ledger_skip_store = PostgreSQLLedgerSkipStore(config.dsn) # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler. # (Created before the ledger writer so option-D write-back can target it.)