fix: isolate ledger outbox delivery by instance

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

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,8 +34,7 @@ _EXPECTED_TABLES = {
 
 
 def _columns(dsn: str, table: str) -> set[str]:
-    with psycopg.connect(dsn) as conn:
-        with conn.cursor() as cur:
+    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",
@@ -45,8 +45,7 @@ def _columns(dsn: str, table: str) -> set[str]:
 
 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:
+    with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
         cur.execute(
             "SELECT table_name FROM information_schema.tables WHERE table_schema='public'"
         )
@@ -61,8 +60,7 @@ 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:
+    with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
         cur.execute(
             "SELECT column_name, data_type, udt_name "
             "FROM information_schema.columns "
@@ -111,8 +109,7 @@ 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 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",
@@ -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:
-        with conn.cursor() as cur:
+    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:
+            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() == []

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

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

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(

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],
         )
 
 

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';

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] = {}

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)

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