Build 060.28: implement Persistent Checkpoint and Startup Recovery

app/tests/integration/market_data/acquisition/runtime/loopback_trade_exchange.py

@@ -214,6 +214,26 @@ class LoopbackTradeWebSocketServer:
             ),
         )
 
+    async def send_ack(
+        self,
+        connection_index: int,
+        *,
+        correlation_id: str,
+        status: str = "OK",
+        destination: str = "trades.subscribe",
+    ) -> None:
+        """Отправить управляемый ответ на Trade subscription."""
+        await self.send_raw(
+            connection_index,
+            json.dumps(
+                {
+                    "correlationId": correlation_id,
+                    "destination": destination,
+                    "status": status,
+                }
+            ),
+        )
+
     async def abort_connection(
         self,
         connection_index: int,

app/tests/integration/market_data/acquisition/runtime/test_trade_stream_loopback_integration.py

@@ -3,10 +3,29 @@ from __future__ import annotations
 import asyncio
 import threading
 import time
+from datetime import datetime, timezone
+from decimal import Decimal
 from typing import Any
 
 import pytest
 
+from src.core.config import Settings
+from src.integrations.exchange.rest_client import ExchangeRestClient
+from src.market_data.acquisition.adapters.dzengi.rest import (
+    DzengiTradesDocumentSource,
+)
+from src.market_data.acquisition.adapters.dzengi.websocket import (
+    DzengiUnifiedWebSocketAdapter,
+)
+from src.market_data.acquisition.adapters.dzengi.websocket_control_message_handler import (
+    DzengiWebSocketControlMessageHandler,
+)
+from src.market_data.acquisition.adapters.dzengi.websocket_inbound_message_classifier import (
+    DzengiWebSocketInboundMessageClassifier,
+)
+from src.market_data.acquisition.adapters.dzengi.websocket_transport import (
+    DzengiWebSocketTransport,
+)
 from tests.integration.market_data.acquisition.runtime import (
     loopback_trade_exchange,
 )
@@ -22,6 +41,13 @@ from src.market_data.acquisition.exceptions import (
     TradeTransportError,
     WebSocketMessageDecodeError,
 )
+from src.market_data.acquisition.models.trade import (
+    Trade,
+    TradeAggressorSide,
+)
+from src.market_data.acquisition.runtime.acquisition_runtime_event_publisher import (
+    AcquisitionRuntimeEventPublisher,
+)
 from src.market_data.acquisition.runtime.runtime_reconnect_recovery_coordinator import (
     RuntimeReconnectRecoveryCoordinator,
 )
@@ -29,10 +55,22 @@ from src.market_data.acquisition.runtime.trade_stream_production_runtime import
     TradeStreamProductionRuntime,
     TradeStreamProductionRuntimeState,
 )
+from src.market_data.acquisition.runtime.websocket_session import (
+    WebSocketSession,
+)
+from src.market_data.acquisition.runtime.websocket_subscription_manager import (
+    WebSocketSubscriptionManager,
+)
+from src.market_data.acquisition.trade_stream_runtime_composition import (
+    TradeStreamRuntimeComposition,
+    build_trade_stream_runtime_composition,
+)
+from src.market_data.storage.contracts import PersistentTradeCheckpoint
 from tests.support.trade_stream_runtime import (
     SYMBOL,
     assert_no_owned_tasks,
     build_runtime,
+    make_settings,
     reconnect_coordinator_from,
     run_scenario,
     start_runtime,
@@ -44,6 +82,187 @@ from tests.support.trade_stream_runtime import (
 pytestmark = pytest.mark.integration
 
 
+class StaticCheckpointStorage:
+    """Локальное checkpoint-хранилище для startup boundary."""
+
+    def __init__(
+        self,
+        *,
+        checkpoint: PersistentTradeCheckpoint,
+        tail: tuple[Trade, ...],
+        events: list[str],
+    ) -> None:
+        self._checkpoint = checkpoint
+        self._tail = tail
+        self._events = events
+
+    def load_checkpoint(
+        self,
+        *,
+        venue: str,
+        symbol: str,
+    ) -> PersistentTradeCheckpoint | None:
+        assert venue == "dzengi"
+        assert symbol == SYMBOL
+        self._events.append("checkpoint.load")
+        return self._checkpoint
+
+    def load_checkpoint_tail(
+        self,
+        *,
+        venue: str,
+        checkpoint: PersistentTradeCheckpoint,
+        limit: int,
+    ) -> tuple[Trade, ...]:
+        assert venue == "dzengi"
+        assert checkpoint is self._checkpoint
+        assert limit > 0
+        self._events.append("checkpoint.tail")
+        return self._tail
+
+    def load_latest_trade_tail(
+        self,
+        *,
+        venue: str,
+        symbol: str,
+        limit: int,
+    ) -> tuple[Trade, ...]:
+        raise AssertionError("Checkpoint уже существует.")
+
+    def adopt_existing_trade_as_checkpoint(
+        self,
+        *,
+        venue: str,
+        trade: Trade,
+    ) -> PersistentTradeCheckpoint:
+        raise AssertionError("Checkpoint уже существует.")
+
+    def store_trade_and_advance_checkpoint(
+        self,
+        *,
+        venue: str,
+        expected_trade: Trade | None,
+        trade: Trade,
+        observed_at: datetime,
+    ) -> PersistentTradeCheckpoint:
+        raise AssertionError("Startup Hydrator не выполняет запись.")
+
+
+class RecordingTradeObservationSink:
+    """Фиксирует точный порядок durable acceptance в тесте."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._accepted_trade_ids: list[int] = []
+
+    @property
+    def accepted_trade_ids(self) -> tuple[int, ...]:
+        with self._lock:
+            return tuple(self._accepted_trade_ids)
+
+    def persist_accepted(
+        self,
+        trade: Trade,
+        *,
+        expected_trade: Trade | None,
+    ) -> None:
+        del expected_trade
+
+        with self._lock:
+            self._accepted_trade_ids.append(trade.trade_id)
+
+    def persist_duplicate(self, trade: Trade) -> None:
+        del trade
+
+
+def make_checkpoint_trade(*, timestamp_ms: int) -> Trade:
+    return Trade(
+        symbol=SYMBOL,
+        trade_id=100,
+        price=Decimal("64555.55"),
+        quantity=Decimal("0.002"),
+        executed_at=datetime.fromtimestamp(
+            timestamp_ms / 1_000,
+            tz=timezone.utc,
+        ),
+        aggressor_side=TradeAggressorSide.BUY,
+        source="postgres_trade_history",
+    )
+
+
+def build_startup_recovery_runtime(
+    *,
+    settings: Settings,
+    checkpoint_storage: StaticCheckpointStorage,
+    observation_sink: RecordingTradeObservationSink,
+    recovery_end_time_ms: int,
+) -> tuple[TradeStreamProductionRuntime, TradeStreamRuntimeComposition]:
+    """Собрать persistent startup-граф без использования Bootstrap."""
+    transport = DzengiWebSocketTransport(
+        url=settings.trade_stream.websocket_url,
+        headers={
+            "Origin": settings.exchange_base_url.rstrip("/"),
+            "Content-Type": "application/json",
+        },
+        open_timeout=settings.trade_stream.open_timeout_seconds,
+        ping_interval=None,
+        ping_timeout=None,
+        probe_timeout=settings.trade_stream.probe_timeout_seconds,
+        close_timeout=settings.trade_stream.close_timeout_seconds,
+    )
+    session = WebSocketSession(transport)
+    subscription_manager = WebSocketSubscriptionManager(
+        transport,
+        supports_unsubscribe=False,
+    )
+    event_publisher = AcquisitionRuntimeEventPublisher()
+    composition = build_trade_stream_runtime_composition(
+        session=session,
+        transport=transport,
+        subscription_manager=subscription_manager,
+        event_publisher=event_publisher,
+        message_adapter=DzengiUnifiedWebSocketAdapter(),
+        recovery_document_source=DzengiTradesDocumentSource(
+            ExchangeRestClient(settings=settings),
+        ),
+        symbols=(SYMBOL,),
+        heartbeat_timeout_seconds=60.0,
+        scheduler_interval_seconds=60.0,
+        trade_observation_sink=observation_sink,
+        max_recovery_window_ms=3_599_999,
+        recovery_end_time_clock=lambda: recovery_end_time_ms,
+        checkpoint_storage=checkpoint_storage,
+        checkpoint_venue="dzengi",
+    )
+    startup_recovery = (
+        composition.runtime_startup_recovery_coordinator
+    )
+    assert startup_recovery is not None
+
+    runtime = TradeStreamProductionRuntime(
+        session=session,
+        transport=transport,
+        subscription_manager=subscription_manager,
+        event_publisher=event_publisher,
+        trade_stream_service=(
+            composition.trade_stream_acquisition_service
+        ),
+        message_classifier=DzengiWebSocketInboundMessageClassifier(),
+        control_message_handler=DzengiWebSocketControlMessageHandler(),
+        live_processing_gate=composition.live_processing_gate,
+        reconnect_recovery_coordinator=(
+            composition.runtime_reconnect_recovery_coordinator
+        ),
+        runtime_supervisor=composition.runtime_supervisor,
+        runtime_scheduler=composition.runtime_scheduler,
+        symbols=(SYMBOL,),
+        startup_recovery_coordinator=startup_recovery,
+        subscription_ack_timeout_seconds=1.0,
+        startup_market_buffer_capacity=10,
+    )
+    return runtime, composition
+
+
 def make_recovered_trade(
     *,
     trade_id: int,
@@ -277,6 +496,218 @@ def test_real_loopback_websocket_updates_shared_checkpoint() -> None:
     run_scenario(scenario())
 
 
+def test_startup_recovery_precedes_buffered_live_market() -> None:
+    async def scenario() -> None:
+        events: list[str] = []
+        recovery_release = threading.Event()
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        checkpoint_trade = make_checkpoint_trade(
+            timestamp_ms=base_time_ms,
+        )
+        checkpoint = PersistentTradeCheckpoint(
+            venue="dzengi",
+            trade=checkpoint_trade,
+            revision=7,
+            updated_at=datetime.now(timezone.utc),
+        )
+        checkpoint_storage = StaticCheckpointStorage(
+            checkpoint=checkpoint,
+            tail=(checkpoint_trade,),
+            events=events,
+        )
+        observation_sink = RecordingTradeObservationSink()
+        websocket = LoopbackTradeWebSocketServer(
+            events=events,
+            auto_ack=False,
+        )
+        rest = LoopbackTradeRestServer(
+            responses=(
+                LoopbackHttpResponse(
+                    body=[
+                        make_recovered_trade(
+                            trade_id=101,
+                            timestamp_ms=base_time_ms + 100,
+                        ),
+                    ],
+                    release=recovery_release,
+                ),
+            ),
+            events=events,
+        )
+
+        async with LoopbackTradeEnvironment(
+            websocket=websocket,
+            rest=rest,
+        ) as environment:
+            settings = make_settings(
+                websocket_url=environment.websocket_url,
+                rest_base_url=rest.base_url,
+            )
+            runtime, composition = build_startup_recovery_runtime(
+                settings=settings,
+                checkpoint_storage=checkpoint_storage,
+                observation_sink=observation_sink,
+                recovery_end_time_ms=base_time_ms + 150,
+            )
+            runtime_task = asyncio.create_task(
+                runtime.run(),
+                name="trade-stream-runtime",
+            )
+
+            try:
+                await websocket.wait_for_subscriptions(1)
+
+                assert runtime.state is (
+                    TradeStreamProductionRuntimeState.STARTING
+                )
+                assert composition.state_store.get(
+                    SYMBOL,
+                ).last_trade_id == 100
+                assert events.index("checkpoint.tail") < events.index(
+                    "ws.connect:0"
+                )
+
+                await websocket.send_trade(
+                    0,
+                    symbol=SYMBOL,
+                    trade_id=102,
+                    timestamp_ms=base_time_ms + 200,
+                )
+                subscription = websocket.subscriptions[0]
+                await websocket.send_ack(
+                    0,
+                    correlation_id=subscription.correlation_id,
+                )
+                await rest.wait_for_requests(1)
+
+                assert runtime.state is (
+                    TradeStreamProductionRuntimeState.STARTING
+                )
+                assert composition.state_store.get(
+                    SYMBOL,
+                ).last_trade_id == 100
+                assert observation_sink.accepted_trade_ids == ()
+
+                recovery_release.set()
+                await wait_until(
+                    lambda: runtime.state
+                    is TradeStreamProductionRuntimeState.RUNNING,
+                )
+                await wait_until(
+                    lambda: composition.state_store.get(
+                        SYMBOL,
+                    ).last_trade_id
+                    == 102,
+                )
+
+                assert observation_sink.accepted_trade_ids == (
+                    101,
+                    102,
+                )
+                assert events.index("ws.subscribe:0") < events.index(
+                    "rest.request"
+                )
+                assert runtime_task.done() is False
+            finally:
+                recovery_release.set()
+                await stop_runtime(runtime, runtime_task)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(scenario())
+
+
+def test_startup_stop_waits_for_blocked_recovery_worker() -> None:
+    async def scenario() -> None:
+        events: list[str] = []
+        recovery_release = threading.Event()
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        checkpoint_trade = make_checkpoint_trade(
+            timestamp_ms=base_time_ms,
+        )
+        checkpoint = PersistentTradeCheckpoint(
+            venue="dzengi",
+            trade=checkpoint_trade,
+            revision=3,
+            updated_at=datetime.now(timezone.utc),
+        )
+        checkpoint_storage = StaticCheckpointStorage(
+            checkpoint=checkpoint,
+            tail=(checkpoint_trade,),
+            events=events,
+        )
+        observation_sink = RecordingTradeObservationSink()
+        websocket = LoopbackTradeWebSocketServer(events=events)
+        rest = LoopbackTradeRestServer(
+            responses=(
+                LoopbackHttpResponse(
+                    body=[],
+                    release=recovery_release,
+                ),
+            ),
+            events=events,
+        )
+
+        async with LoopbackTradeEnvironment(
+            websocket=websocket,
+            rest=rest,
+        ) as environment:
+            settings = make_settings(
+                websocket_url=environment.websocket_url,
+                rest_base_url=rest.base_url,
+            )
+            runtime, composition = build_startup_recovery_runtime(
+                settings=settings,
+                checkpoint_storage=checkpoint_storage,
+                observation_sink=observation_sink,
+                recovery_end_time_ms=base_time_ms + 150,
+            )
+            runtime_task = asyncio.create_task(
+                runtime.run(),
+                name="trade-stream-runtime",
+            )
+            stop_task: asyncio.Task[None] | None = None
+
+            try:
+                await websocket.wait_for_subscriptions(1)
+                await rest.wait_for_requests(1)
+
+                stop_task = asyncio.create_task(runtime.stop())
+                await asyncio.sleep(0)
+                await asyncio.sleep(0)
+
+                assert stop_task.done() is False
+                assert runtime_task.done() is False
+                assert composition.live_processing_gate.locked is True
+
+                recovery_release.set()
+                await asyncio.wait_for(stop_task, timeout=3.0)
+                await asyncio.wait_for(runtime_task, timeout=3.0)
+
+                assert runtime.state is (
+                    TradeStreamProductionRuntimeState.STOPPED
+                )
+                assert composition.live_processing_gate.locked is False
+                assert composition.live_processing_gate.failed is False
+                assert observation_sink.accepted_trade_ids == ()
+            finally:
+                recovery_release.set()
+
+                if stop_task is not None and not stop_task.done():
+                    await stop_task
+
+                if not runtime_task.done():
+                    await stop_runtime(runtime, runtime_task)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(scenario())
+
+
 def test_reconnect_restores_then_recovers_before_buffered_live() -> None:
     async def scenario() -> None:
         events: list[str] = []

app/tests/integration/market_data/storage/test_postgres_partition_retention_integration.py

@@ -14,9 +14,11 @@ from src.market_data.acquisition.models.trade import (
 )
 from src.market_data.storage import (
     MARKET_DATA_PARTITION_ADVISORY_LOCK_ID,
+    MarketDataPartitionResult,
     MarketDataPartitionType,
     MarketDataRetentionPolicy,
     MarketDataStorageOperationError,
+    PersistentTradeCheckpoint,
     PostgresMarketDataPartitionManager,
     PostgresMarketDataRetentionService,
     PostgresQuoteRepository,
@@ -27,6 +29,7 @@ from tests.support.postgres_market_data import (
     PostgresTestSettings,
     connect_postgres_test_database,
     wait_for_postgres_advisory_lock_waiters,
+    wait_for_postgres_relation_lock_waiters,
 )
 
 
@@ -34,6 +37,7 @@ pytestmark = pytest.mark.integration
 
 VENUE = "dzengi"
 SYMBOL = "BTC/USD_LEVERAGE"
+PARTITION_WRITER_BARRIER_LOCK_ID = 0x445A5041525457
 
 
 def _trade(*, trade_id: int, executed_at: datetime) -> Trade:
@@ -60,6 +64,91 @@ def _quote(*, received_at: datetime) -> Quote:
     )
 
 
+def _insert_checkpoint(
+    pool: PostgresConnectionPool,
+    *,
+    trade: Trade,
+    revision: int = 1,
+) -> None:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO market_data.trade_stream_checkpoints (
+                    venue,
+                    symbol,
+                    trade_id,
+                    executed_at,
+                    revision
+                )
+                VALUES (%s, %s, %s, %s, %s)
+                """,
+                (
+                    VENUE,
+                    trade.symbol,
+                    trade.trade_id,
+                    trade.executed_at,
+                    revision,
+                ),
+            )
+
+
+def _advance_checkpoint(
+    pool: PostgresConnectionPool,
+    *,
+    trade: Trade,
+    revision: int,
+) -> None:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE market_data.trade_stream_checkpoints
+                SET trade_id = %s,
+                    executed_at = %s,
+                    revision = %s,
+                    updated_at = NOW()
+                WHERE venue = %s
+                  AND symbol = %s
+                """,
+                (
+                    trade.trade_id,
+                    trade.executed_at,
+                    revision,
+                    VENUE,
+                    trade.symbol,
+                ),
+            )
+            assert cursor.rowcount == 1
+
+
+def _assert_checkpoint_foreign_key_is_deferred_no_action(
+    pool: PostgresConnectionPool,
+) -> None:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT constraint_row.confupdtype::text,
+                       constraint_row.confdeltype::text,
+                       constraint_row.condeferrable,
+                       constraint_row.condeferred
+                FROM pg_catalog.pg_constraint AS constraint_row
+                JOIN pg_catalog.pg_class AS table_row
+                  ON table_row.oid = constraint_row.conrelid
+                JOIN pg_catalog.pg_namespace AS namespace
+                  ON namespace.oid = table_row.relnamespace
+                WHERE namespace.nspname = 'market_data'
+                  AND table_row.relname = 'trade_stream_checkpoints'
+                  AND constraint_row.conname =
+                      'trade_stream_checkpoints_trade_fk'
+                """
+            )
+            row = cursor.fetchone()
+
+    assert row == ("a", "a", True, True)
+
+
 def test_real_partition_creation_moves_default_row_and_is_idempotent(
     migrated_postgres_pool: PostgresConnectionPool,
 ) -> None:
@@ -99,6 +188,56 @@ def test_real_partition_creation_moves_default_row_and_is_idempotent(
     assert relation == ("market_data.trades_2026_07",)
 
 
+def test_checkpoint_trade_moves_from_default_to_month_partition(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    event_time = datetime(2026, 7, 15, tzinfo=timezone.utc)
+    trade = _trade(trade_id=1, executed_at=event_time)
+    repository = PostgresTradeRepository(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager = PostgresMarketDataPartitionManager(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=trade,
+        observed_at=event_time,
+    )
+    _insert_checkpoint(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    result = manager.ensure_month_partition(
+        data_type=MarketDataPartitionType.TRADES,
+        month=event_time,
+    )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT trade_row.tableoid::regclass::text,
+                       checkpoint.revision
+                FROM market_data.trade_stream_checkpoints AS checkpoint
+                JOIN market_data.trades AS trade_row
+                  ON trade_row.venue = checkpoint.venue
+                 AND trade_row.symbol = checkpoint.symbol
+                 AND trade_row.trade_id = checkpoint.trade_id
+                 AND trade_row.executed_at = checkpoint.executed_at
+                """
+            )
+            restored = cursor.fetchone()
+
+    assert result.created is True
+    assert result.moved_row_count == 1
+    assert restored == ("market_data.trades_2026_07", 1)
+    _assert_checkpoint_foreign_key_is_deferred_no_action(
+        migrated_postgres_pool
+    )
+
+
 def test_two_real_partition_callers_create_one_partition(
     migrated_postgres_pool: PostgresConnectionPool,
     postgres_test_settings: PostgresTestSettings,
@@ -150,6 +289,164 @@ def test_two_real_partition_callers_create_one_partition(
     assert sorted(results) == [(False, 0), (True, 0)]
 
 
+def test_partition_creation_and_checkpoint_writer_use_one_lock_order(
+    migrated_postgres_pool: PostgresConnectionPool,
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    existing_month = datetime(2026, 7, 1, tzinfo=timezone.utc)
+    new_month = datetime(2026, 8, 1, tzinfo=timezone.utc)
+    previous_trade = _trade(
+        trade_id=1,
+        executed_at=datetime(2026, 7, 14, tzinfo=timezone.utc),
+    )
+    candidate_trade = _trade(
+        trade_id=2,
+        executed_at=datetime(2026, 7, 15, tzinfo=timezone.utc),
+    )
+    repository = PostgresTradeRepository(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager = PostgresMarketDataPartitionManager(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager.ensure_month_partition(
+        data_type=MarketDataPartitionType.TRADES,
+        month=existing_month,
+    )
+    repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=previous_trade,
+        observed_at=previous_trade.executed_at,
+    )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                CREATE FUNCTION market_data.block_partition_writer()
+                RETURNS trigger
+                LANGUAGE plpgsql
+                AS $$
+                BEGIN
+                    PERFORM pg_advisory_xact_lock(
+                        {PARTITION_WRITER_BARRIER_LOCK_ID}
+                    );
+                    RETURN NEW;
+                END;
+                $$
+                """
+            )
+            cursor.execute(
+                """
+                CREATE TRIGGER block_partition_writer
+                BEFORE INSERT ON market_data.trades_2026_07
+                FOR EACH ROW
+                EXECUTE FUNCTION market_data.block_partition_writer()
+                """
+            )
+
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def store_and_advance_checkpoint() -> PersistentTradeCheckpoint:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        return repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=previous_trade,
+            trade=candidate_trade,
+            observed_at=candidate_trade.executed_at,
+        )
+
+    def create_partition() -> MarketDataPartitionResult:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        return manager.ensure_month_partition(
+            data_type=MarketDataPartitionType.TRADES,
+            month=new_month,
+        )
+
+    with connect_postgres_test_database(
+        postgres_test_settings,
+    ) as control:
+        with control.cursor() as cursor:
+            cursor.execute(
+                "SELECT pg_advisory_lock(%s)",
+                (PARTITION_WRITER_BARRIER_LOCK_ID,),
+            )
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            try:
+                writer_future = executor.submit(
+                    store_and_advance_checkpoint
+                )
+                wait_for_postgres_advisory_lock_waiters(
+                    control,
+                    lock_id=PARTITION_WRITER_BARRIER_LOCK_ID,
+                    expected_count=1,
+                )
+                manager_future = executor.submit(create_partition)
+                wait_for_postgres_relation_lock_waiters(
+                    control,
+                    relation_name="market_data.trades",
+                    expected_count=1,
+                )
+
+            finally:
+                with control.cursor() as cursor:
+                    cursor.execute(
+                        "SELECT pg_advisory_unlock(%s)",
+                        (PARTITION_WRITER_BARRIER_LOCK_ID,),
+                    )
+                    barrier_released = cursor.fetchone()
+
+            if barrier_released != (True,):
+                raise AssertionError("Writer barrier was not released.")
+
+            try:
+                checkpoint = writer_future.result(timeout=10.0)
+                partition_result = manager_future.result(timeout=10.0)
+            finally:
+                writer_future.cancel()
+                manager_future.cancel()
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT trade_row.tableoid::regclass::text,
+                       checkpoint.trade_id,
+                       checkpoint.revision
+                FROM market_data.trades AS trade_row
+                JOIN market_data.trade_stream_checkpoints AS checkpoint
+                  ON checkpoint.venue = trade_row.venue
+                 AND checkpoint.symbol = trade_row.symbol
+                 AND checkpoint.trade_id = trade_row.trade_id
+                 AND checkpoint.executed_at = trade_row.executed_at
+                WHERE trade_row.venue = %s
+                  AND trade_row.symbol = %s
+                  AND trade_row.trade_id = %s
+                """,
+                (
+                    VENUE,
+                    SYMBOL,
+                    candidate_trade.trade_id,
+                ),
+            )
+            persisted = cursor.fetchone()
+
+    assert len(caller_ids) == 2
+    assert checkpoint.trade == candidate_trade
+    assert checkpoint.revision == 2
+    assert partition_result.created is True
+    assert partition_result.moved_row_count == 0
+    assert persisted == ("market_data.trades_2026_07", 2, 2)
+    _assert_checkpoint_foreign_key_is_deferred_no_action(
+        migrated_postgres_pool
+    )
+
+
 def test_real_retention_uses_exact_cutoff(
     migrated_postgres_pool: PostgresConnectionPool,
 ) -> None:
@@ -189,6 +486,184 @@ def test_real_retention_uses_exact_cutoff(
     assert remaining == (2, 3)
 
 
+def test_retention_rolls_back_for_active_checkpoint_in_default(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    now = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc)
+    old_time = now - timedelta(days=30)
+    trade = _trade(trade_id=1, executed_at=old_time)
+    repository = PostgresTradeRepository(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    service = PostgresMarketDataRetentionService(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=trade,
+        observed_at=now,
+    )
+    _insert_checkpoint(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    with pytest.raises(MarketDataStorageOperationError):
+        service.apply(
+            policy=MarketDataRetentionPolicy(
+                enabled=True,
+                trade_days=10,
+            ),
+            now=now,
+        )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT COUNT(*) FROM market_data.trades")
+            trade_count = cursor.fetchone()
+            cursor.execute(
+                "SELECT COUNT(*) "
+                "FROM market_data.trade_stream_checkpoints"
+            )
+            checkpoint_count = cursor.fetchone()
+
+    assert trade_count == (1,)
+    assert checkpoint_count == (1,)
+    _assert_checkpoint_foreign_key_is_deferred_no_action(
+        migrated_postgres_pool
+    )
+
+
+def test_retention_rolls_back_for_active_checkpoint_in_partition(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    now = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc)
+    old_time = datetime(2026, 6, 15, tzinfo=timezone.utc)
+    trade = _trade(trade_id=1, executed_at=old_time)
+    repository = PostgresTradeRepository(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager = PostgresMarketDataPartitionManager(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    service = PostgresMarketDataRetentionService(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager.ensure_month_partition(
+        data_type=MarketDataPartitionType.TRADES,
+        month=old_time,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=trade,
+        observed_at=now,
+    )
+    _insert_checkpoint(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    with pytest.raises(MarketDataStorageOperationError):
+        service.apply(
+            policy=MarketDataRetentionPolicy(
+                enabled=True,
+                trade_days=10,
+            ),
+            now=now,
+        )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT COUNT(*) FROM market_data.trades")
+            trade_count = cursor.fetchone()
+            cursor.execute(
+                """
+                SELECT COUNT(*)
+                FROM market_data.partition_registry
+                WHERE partition_name = 'trades_2026_06'
+                """
+            )
+            registry_count = cursor.fetchone()
+
+    assert trade_count == (1,)
+    assert registry_count == (1,)
+
+
+def test_retention_drops_old_partition_after_checkpoint_advances(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    now = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc)
+    old_time = datetime(2026, 1, 15, tzinfo=timezone.utc)
+    current_time = datetime(2026, 7, 15, tzinfo=timezone.utc)
+    old_trade = _trade(trade_id=1, executed_at=old_time)
+    current_trade = _trade(trade_id=2, executed_at=current_time)
+    repository = PostgresTradeRepository(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager = PostgresMarketDataPartitionManager(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    service = PostgresMarketDataRetentionService(
+        connection_provider=migrated_postgres_pool.connection,
+    )
+    manager.ensure_month_partition(
+        data_type=MarketDataPartitionType.TRADES,
+        month=old_time,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=old_trade,
+        observed_at=now,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=current_trade,
+        observed_at=now,
+    )
+    _insert_checkpoint(
+        migrated_postgres_pool,
+        trade=old_trade,
+    )
+    _advance_checkpoint(
+        migrated_postgres_pool,
+        trade=current_trade,
+        revision=2,
+    )
+
+    result = service.apply(
+        policy=MarketDataRetentionPolicy(
+            enabled=True,
+            trade_days=90,
+        ),
+        now=now,
+    )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT trade_id FROM market_data.trades ORDER BY trade_id"
+            )
+            remaining_trade_ids = tuple(
+                row[0] for row in cursor.fetchall()
+            )
+            cursor.execute(
+                """
+                SELECT trade_id, revision
+                FROM market_data.trade_stream_checkpoints
+                """
+            )
+            checkpoint_row = cursor.fetchone()
+
+    trade_result = result.entries[0]
+    assert trade_result.dropped_partitions == ("trades_2026_01",)
+    assert trade_result.dropped_row_count == 1
+    assert remaining_trade_ids == (2,)
+    assert checkpoint_row == (2, 2)
+    _assert_checkpoint_foreign_key_is_deferred_no_action(
+        migrated_postgres_pool
+    )
+
+
 def test_real_retention_failure_rolls_back_all_data_types(
     migrated_postgres_pool: PostgresConnectionPool,
 ) -> None:

app/tests/integration/market_data/storage/test_postgres_schema_integration.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
 import threading
 
 import pytest
+from psycopg.errors import ForeignKeyViolation
 
 from src.storage.exceptions import StorageMigrationError
 from src.storage.migrations import (
@@ -104,6 +106,51 @@ def test_real_migrations_create_expected_schema_and_are_idempotent(
                 partitioned_tables = tuple(
                     row[0] for row in cursor.fetchall()
                 )
+                cursor.execute(
+                    """
+                    SELECT column_name, data_type, is_nullable
+                    FROM information_schema.columns
+                    WHERE table_schema = 'market_data'
+                      AND table_name = 'trade_stream_checkpoints'
+                    ORDER BY ordinal_position
+                    """
+                )
+                checkpoint_columns = tuple(cursor.fetchall())
+                cursor.execute(
+                    """
+                    SELECT pg_get_constraintdef(constraint_row.oid)
+                    FROM pg_catalog.pg_constraint AS constraint_row
+                    JOIN pg_catalog.pg_class AS table_row
+                      ON table_row.oid = constraint_row.conrelid
+                    JOIN pg_catalog.pg_namespace AS namespace
+                      ON namespace.oid = table_row.relnamespace
+                    WHERE namespace.nspname = 'market_data'
+                      AND table_row.relname = 'trade_stream_checkpoints'
+                    ORDER BY constraint_row.contype,
+                             constraint_row.conname
+                    """
+                )
+                checkpoint_constraints = tuple(
+                    row[0] for row in cursor.fetchall()
+                )
+                cursor.execute(
+                    """
+                    SELECT constraint_row.confupdtype::text,
+                           constraint_row.confdeltype::text,
+                           constraint_row.condeferrable,
+                           constraint_row.condeferred
+                    FROM pg_catalog.pg_constraint AS constraint_row
+                    JOIN pg_catalog.pg_class AS table_row
+                      ON table_row.oid = constraint_row.conrelid
+                    JOIN pg_catalog.pg_namespace AS namespace
+                      ON namespace.oid = table_row.relnamespace
+                    WHERE namespace.nspname = 'market_data'
+                      AND table_row.relname = 'trade_stream_checkpoints'
+                      AND constraint_row.conname =
+                          'trade_stream_checkpoints_trade_fk'
+                    """
+                )
+                checkpoint_foreign_key = cursor.fetchone()
     finally:
         pool.close()
 
@@ -121,11 +168,140 @@ def test_real_migrations_create_expected_schema_and_are_idempotent(
         "quotes",
         "trades",
     )
+    assert checkpoint_columns == (
+        ("venue", "text", "NO"),
+        ("symbol", "text", "NO"),
+        ("trade_id", "integer", "NO"),
+        ("executed_at", "timestamp with time zone", "NO"),
+        ("revision", "bigint", "NO"),
+        ("updated_at", "timestamp with time zone", "NO"),
+        ("checkpoint_schema_version", "integer", "NO"),
+    )
+    assert any(
+        definition
+        == "PRIMARY KEY (venue, symbol)"
+        for definition in checkpoint_constraints
+    )
+    assert any(
+        definition.startswith(
+            "FOREIGN KEY (venue, symbol, trade_id, executed_at) "
+            "REFERENCES market_data.trades"
+        )
+        and "DEFERRABLE INITIALLY DEFERRED" in definition
+        for definition in checkpoint_constraints
+    )
+    assert checkpoint_foreign_key == ("a", "a", True, True)
 
     with connect_postgres_test_database(postgres_test_settings) as control:
         assert count_other_test_connections(control) == 0
 
 
+def test_checkpoint_foreign_key_rejects_orphan_and_trade_deletion(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    executed_at = datetime(2026, 8, 1, 10, 0, tzinfo=timezone.utc)
+
+    with pytest.raises(ForeignKeyViolation):
+        with migrated_postgres_pool.connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO market_data.trade_stream_checkpoints (
+                        venue,
+                        symbol,
+                        trade_id,
+                        executed_at,
+                        revision
+                    )
+                    VALUES (%s, %s, %s, %s, %s)
+                    """,
+                    (
+                        "DZENGI",
+                        "BTC/USD_LEVERAGE",
+                        123,
+                        executed_at,
+                        1,
+                    ),
+                )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO market_data.trades (
+                    venue,
+                    symbol,
+                    trade_id,
+                    executed_at,
+                    price,
+                    quantity,
+                    aggressor_side,
+                    source,
+                    first_observed_at,
+                    last_observed_at,
+                    observation_sources,
+                    canonical_schema_version
+                )
+                VALUES (
+                    %s, %s, %s, %s, %s, %s,
+                    %s, %s, %s, %s, %s, %s
+                )
+                """,
+                (
+                    "DZENGI",
+                    "BTC/USD_LEVERAGE",
+                    123,
+                    executed_at,
+                    "65000.25",
+                    "0.001",
+                    "buy",
+                    "dzengi_websocket_trade",
+                    executed_at,
+                    executed_at,
+                    ["dzengi_websocket_trade"],
+                    1,
+                ),
+            )
+            cursor.execute(
+                """
+                INSERT INTO market_data.trade_stream_checkpoints (
+                    venue,
+                    symbol,
+                    trade_id,
+                    executed_at,
+                    revision
+                )
+                VALUES (%s, %s, %s, %s, %s)
+                """,
+                (
+                    "DZENGI",
+                    "BTC/USD_LEVERAGE",
+                    123,
+                    executed_at,
+                    1,
+                ),
+            )
+
+    with pytest.raises(ForeignKeyViolation):
+        with migrated_postgres_pool.connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    DELETE FROM market_data.trades
+                    WHERE venue = %s
+                      AND symbol = %s
+                      AND trade_id = %s
+                      AND executed_at = %s
+                    """,
+                    (
+                        "DZENGI",
+                        "BTC/USD_LEVERAGE",
+                        123,
+                        executed_at,
+                    ),
+                )
+
+
 def test_two_real_migration_runners_apply_each_version_once(
     postgres_test_settings: PostgresTestSettings,
 ) -> None:

app/tests/integration/market_data/storage/test_postgres_trade_checkpoint_repository_integration.py

@@ -0,0 +1,707 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+import threading
+
+import pytest
+
+from src.market_data.acquisition.models.trade import (
+    Trade,
+    TradeAggressorSide,
+)
+from src.market_data.acquisition.trade_id_sequence import (
+    SIGNED_TRADE_ID_MAX,
+    SIGNED_TRADE_ID_MIN,
+)
+from src.market_data.storage import (
+    MarketDataCheckpointConflictError,
+    MarketDataCheckpointIntegrityError,
+    MarketDataStorageOperationError,
+    PostgresTradeRepository,
+)
+from src.storage.postgres_pool import PostgresConnectionPool
+
+
+pytestmark = pytest.mark.integration
+
+VENUE = "dzengi"
+SYMBOL = "BTC/USD_LEVERAGE"
+EVENT_TIME = datetime(2026, 8, 1, 10, 0, tzinfo=timezone.utc)
+OBSERVED_AT = EVENT_TIME + timedelta(seconds=1)
+
+
+def _trade(
+    *,
+    trade_id: int,
+    offset_ms: int = 0,
+    source: str = "dzengi_websocket_trade",
+) -> Trade:
+    return Trade(
+        symbol=SYMBOL,
+        trade_id=trade_id,
+        price=Decimal("65000.25"),
+        quantity=Decimal("0.001"),
+        executed_at=EVENT_TIME + timedelta(milliseconds=offset_ms),
+        aggressor_side=TradeAggressorSide.BUY,
+        source=source,
+    )
+
+
+def _repository(
+    pool: PostgresConnectionPool,
+) -> PostgresTradeRepository:
+    return PostgresTradeRepository(
+        connection_provider=pool.connection,
+    )
+
+
+def _stored_trade_metadata(
+    pool: PostgresConnectionPool,
+    *,
+    trade: Trade,
+) -> tuple[object, ...] | None:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT
+                    source,
+                    first_observed_at,
+                    last_observed_at,
+                    observation_sources
+                FROM market_data.trades
+                WHERE venue = %s
+                  AND symbol = %s
+                  AND trade_id = %s
+                  AND executed_at = %s
+                """,
+                (
+                    VENUE,
+                    trade.symbol,
+                    trade.trade_id,
+                    trade.executed_at,
+                ),
+            )
+            return cursor.fetchone()
+
+
+def test_real_checkpoint_commit_load_and_rollover_tail(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=SIGNED_TRADE_ID_MAX)
+    second = _trade(
+        trade_id=SIGNED_TRADE_ID_MIN,
+        offset_ms=1,
+    )
+    third = _trade(
+        trade_id=SIGNED_TRADE_ID_MIN + 1,
+        offset_ms=2,
+    )
+
+    first_checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    second_checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=first,
+        trade=second,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+    third_checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=second,
+        trade=third,
+        observed_at=OBSERVED_AT + timedelta(seconds=2),
+    )
+
+    loaded = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    tail = repository.load_checkpoint_tail(
+        venue=VENUE,
+        checkpoint=third_checkpoint,
+        limit=3,
+    )
+
+    assert first_checkpoint.revision == 1
+    assert second_checkpoint.revision == 2
+    assert third_checkpoint.revision == 3
+    assert loaded == third_checkpoint
+    assert tail == (first, second, third)
+
+
+def test_real_latest_tail_orders_minus_one_to_zero_without_checkpoint(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=-1)
+    second = _trade(trade_id=0, offset_ms=1)
+    repository.store_trade(
+        venue=VENUE,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=second,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+
+    tail = repository.load_latest_trade_tail(
+        venue=VENUE,
+        symbol=SYMBOL,
+        limit=2,
+    )
+
+    assert tail == (first, second)
+
+
+def test_real_latest_tail_uses_acceptance_order_not_exchange_time(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(
+        trade_id=SIGNED_TRADE_ID_MAX,
+        offset_ms=2,
+    )
+    second = _trade(
+        trade_id=SIGNED_TRADE_ID_MIN,
+        offset_ms=1,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=second,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+
+    tail = repository.load_latest_trade_tail(
+        venue=VENUE,
+        symbol=SYMBOL,
+        limit=2,
+    )
+
+    assert first.executed_at > second.executed_at
+    assert tail == (first, second)
+
+
+def test_real_checkpoint_tail_uses_only_latest_raw_id_cycle(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    previous_cycle = _trade(trade_id=100)
+    current_predecessor = _trade(trade_id=99, offset_ms=1)
+    current = _trade(trade_id=100, offset_ms=2)
+    repository.store_trade(
+        venue=VENUE,
+        trade=previous_cycle,
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=current_predecessor,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+    checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=current,
+        observed_at=OBSERVED_AT + timedelta(seconds=2),
+    )
+
+    tail = repository.load_checkpoint_tail(
+        venue=VENUE,
+        checkpoint=checkpoint,
+        limit=2,
+    )
+
+    assert previous_cycle.trade_id == current.trade_id
+    assert previous_cycle.executed_at != current.executed_at
+    assert tail == (current_predecessor, current)
+
+
+def test_real_adoption_preserves_trade_metadata_and_is_idempotent(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    trade = _trade(trade_id=100)
+    repository.store_trade(
+        venue=VENUE,
+        trade=trade,
+        observed_at=OBSERVED_AT,
+    )
+    before = _stored_trade_metadata(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    adopted = repository.adopt_existing_trade_as_checkpoint(
+        venue=VENUE,
+        trade=trade,
+    )
+    repeated = repository.adopt_existing_trade_as_checkpoint(
+        venue=VENUE,
+        trade=trade,
+    )
+    after = _stored_trade_metadata(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    assert adopted.revision == 1
+    assert repeated == adopted
+    assert before == after
+
+
+def test_real_adoption_rejects_missing_and_conflicting_durable_trade(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    missing = _trade(trade_id=100)
+
+    with pytest.raises(MarketDataCheckpointIntegrityError):
+        repository.adopt_existing_trade_as_checkpoint(
+            venue=VENUE,
+            trade=missing,
+        )
+
+    durable = _trade(trade_id=101, offset_ms=1)
+    repository.store_trade(
+        venue=VENUE,
+        trade=durable,
+        observed_at=OBSERVED_AT,
+    )
+
+    with pytest.raises(MarketDataCheckpointConflictError):
+        repository.adopt_existing_trade_as_checkpoint(
+            venue=VENUE,
+            trade=replace(durable, price=Decimal("65000.26")),
+        )
+
+    assert repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    ) is None
+
+
+def test_two_real_identical_adoption_callers_converge(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    trade = _trade(trade_id=100)
+    repository.store_trade(
+        venue=VENUE,
+        trade=trade,
+        observed_at=OBSERVED_AT,
+    )
+    before = _stored_trade_metadata(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+    start_barrier = threading.Barrier(2)
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def adopt() -> int:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        start_barrier.wait(timeout=5.0)
+        checkpoint = repository.adopt_existing_trade_as_checkpoint(
+            venue=VENUE,
+            trade=trade,
+        )
+        return checkpoint.revision
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        revisions = tuple(executor.map(lambda _: adopt(), range(2)))
+
+    after = _stored_trade_metadata(
+        migrated_postgres_pool,
+        trade=trade,
+    )
+
+    assert len(caller_ids) == 2
+    assert revisions == (1, 1)
+    assert before == after
+
+
+def test_two_real_different_adoption_callers_allow_one_winner(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    candidates = (
+        _trade(trade_id=100),
+        _trade(trade_id=101, offset_ms=1),
+    )
+
+    for index, candidate in enumerate(candidates):
+        repository.store_trade(
+            venue=VENUE,
+            trade=candidate,
+            observed_at=OBSERVED_AT + timedelta(seconds=index),
+        )
+
+    before = tuple(
+        _stored_trade_metadata(
+            migrated_postgres_pool,
+            trade=candidate,
+        )
+        for candidate in candidates
+    )
+    start_barrier = threading.Barrier(2)
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def adopt(candidate: Trade) -> tuple[str, int]:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        start_barrier.wait(timeout=5.0)
+
+        try:
+            checkpoint = repository.adopt_existing_trade_as_checkpoint(
+                venue=VENUE,
+                trade=candidate,
+            )
+        except MarketDataCheckpointConflictError:
+            return ("conflict", candidate.trade_id)
+
+        return ("committed", checkpoint.trade.trade_id)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = tuple(executor.map(adopt, candidates))
+
+    loaded = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    after = tuple(
+        _stored_trade_metadata(
+            migrated_postgres_pool,
+            trade=candidate,
+        )
+        for candidate in candidates
+    )
+    committed_ids = tuple(
+        trade_id
+        for status, trade_id in results
+        if status == "committed"
+    )
+
+    assert len(caller_ids) == 2
+    assert tuple(status for status, _ in results).count("committed") == 1
+    assert tuple(status for status, _ in results).count("conflict") == 1
+    assert loaded is not None
+    assert loaded.trade.trade_id == committed_ids[0]
+    assert loaded.revision == 1
+    assert before == after
+
+
+def test_real_stale_writer_rolls_back_candidate_trade(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=100)
+    winner = _trade(trade_id=101, offset_ms=1)
+    stale_candidate = _trade(trade_id=102, offset_ms=2)
+    repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    winner_checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=first,
+        trade=winner,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+
+    with pytest.raises(MarketDataCheckpointConflictError):
+        repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=first,
+            trade=stale_candidate,
+            observed_at=OBSERVED_AT + timedelta(seconds=2),
+        )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT trade_id FROM market_data.trades ORDER BY trade_id"
+            )
+            trade_ids = tuple(row[0] for row in cursor.fetchall())
+
+    loaded = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    assert trade_ids == (100, 101)
+    assert loaded == winner_checkpoint
+
+
+def test_real_retry_after_commit_preserves_revision_and_updates_provenance(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=100)
+    current = _trade(trade_id=101, offset_ms=1)
+    repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    committed = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=first,
+        trade=current,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+
+    repeated = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=first,
+        trade=replace(current, source="dzengi"),
+        observed_at=OBSERVED_AT + timedelta(seconds=2),
+    )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT observation_sources
+                FROM market_data.trades
+                WHERE trade_id = 101
+                """
+            )
+            sources = cursor.fetchone()
+
+    assert repeated == committed
+    assert repeated.revision == 2
+    assert sources is not None
+    assert set(sources[0]) == {"dzengi_websocket_trade", "dzengi"}
+
+
+def test_two_real_first_writers_leave_only_winner_trade(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    candidates = (
+        _trade(trade_id=100),
+        _trade(trade_id=101, offset_ms=1),
+    )
+    start_barrier = threading.Barrier(2)
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def commit(trade: Trade) -> tuple[str, int]:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        start_barrier.wait(timeout=5.0)
+
+        try:
+            checkpoint = repository.store_trade_and_advance_checkpoint(
+                venue=VENUE,
+                expected_trade=None,
+                trade=trade,
+                observed_at=OBSERVED_AT,
+            )
+        except MarketDataCheckpointConflictError:
+            return ("conflict", trade.trade_id)
+
+        return ("committed", checkpoint.trade.trade_id)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = tuple(executor.map(commit, candidates))
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT trade_id FROM market_data.trades")
+            stored_trade_ids = tuple(row[0] for row in cursor.fetchall())
+
+    loaded = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    committed_ids = tuple(
+        trade_id
+        for status, trade_id in results
+        if status == "committed"
+    )
+    assert len(caller_ids) == 2
+    assert tuple(status for status, _ in results).count("committed") == 1
+    assert tuple(status for status, _ in results).count("conflict") == 1
+    assert stored_trade_ids == committed_ids
+    assert loaded is not None
+    assert loaded.trade.trade_id == committed_ids[0]
+
+
+def test_two_real_identical_first_writers_converge_idempotently(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    candidates = (
+        _trade(trade_id=100),
+        _trade(
+            trade_id=100,
+            source="dzengi",
+        ),
+    )
+    start_barrier = threading.Barrier(2)
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def commit(trade: Trade) -> int:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        start_barrier.wait(timeout=5.0)
+        checkpoint = repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=None,
+            trade=trade,
+            observed_at=OBSERVED_AT,
+        )
+        return checkpoint.revision
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        revisions = tuple(executor.map(commit, candidates))
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT observation_sources FROM market_data.trades"
+            )
+            rows = tuple(cursor.fetchall())
+
+    assert len(caller_ids) == 2
+    assert revisions == (1, 1)
+    assert len(rows) == 1
+    assert set(rows[0][0]) == {"dzengi_websocket_trade", "dzengi"}
+
+
+def test_two_real_checkpoint_updates_allow_only_one_winner(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    previous = _trade(trade_id=100)
+    candidates = (
+        _trade(trade_id=101, offset_ms=1),
+        _trade(trade_id=102, offset_ms=2),
+    )
+    repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=None,
+        trade=previous,
+        observed_at=OBSERVED_AT,
+    )
+    start_barrier = threading.Barrier(2)
+    caller_ids: set[int] = set()
+    caller_ids_lock = threading.Lock()
+
+    def commit(trade: Trade) -> tuple[str, int]:
+        with caller_ids_lock:
+            caller_ids.add(threading.get_ident())
+        start_barrier.wait(timeout=5.0)
+
+        try:
+            checkpoint = repository.store_trade_and_advance_checkpoint(
+                venue=VENUE,
+                expected_trade=previous,
+                trade=trade,
+                observed_at=OBSERVED_AT + timedelta(seconds=1),
+            )
+        except MarketDataCheckpointConflictError:
+            return ("conflict", trade.trade_id)
+
+        return ("committed", checkpoint.trade.trade_id)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = tuple(executor.map(commit, candidates))
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT trade_id FROM market_data.trades ORDER BY trade_id"
+            )
+            stored_trade_ids = tuple(row[0] for row in cursor.fetchall())
+
+    loaded = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    committed_ids = tuple(
+        trade_id
+        for status, trade_id in results
+        if status == "committed"
+    )
+    assert len(caller_ids) == 2
+    assert tuple(status for status, _ in results).count("committed") == 1
+    assert tuple(status for status, _ in results).count("conflict") == 1
+    assert stored_trade_ids == (100, committed_ids[0])
+    assert loaded is not None
+    assert loaded.trade.trade_id == committed_ids[0]
+    assert loaded.revision == 2
+
+
+def test_real_checkpoint_failure_rolls_back_inserted_trade(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    trade = _trade(trade_id=100)
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                CREATE FUNCTION market_data.reject_checkpoint_write()
+                RETURNS trigger
+                LANGUAGE plpgsql
+                AS $$
+                BEGIN
+                    RAISE EXCEPTION 'checkpoint write rejected';
+                END;
+                $$
+                """
+            )
+            cursor.execute(
+                """
+                CREATE TRIGGER reject_checkpoint_write
+                BEFORE INSERT OR UPDATE
+                ON market_data.trade_stream_checkpoints
+                FOR EACH ROW
+                EXECUTE FUNCTION market_data.reject_checkpoint_write()
+                """
+            )
+
+    with pytest.raises(MarketDataStorageOperationError):
+        repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=None,
+            trade=trade,
+            observed_at=OBSERVED_AT,
+        )
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT COUNT(*) FROM market_data.trades")
+            trade_count = cursor.fetchone()
+            cursor.execute(
+                "SELECT COUNT(*) "
+                "FROM market_data.trade_stream_checkpoints"
+            )
+            checkpoint_count = cursor.fetchone()
+
+    assert trade_count == (0,)
+    assert checkpoint_count == (0,)

app/tests/integration/market_data/storage/test_trade_checkpoint_consistency_integration.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+
+import pytest
+
+from src.market_data.acquisition.consistency.trade_stream_consistency_controller import (
+    TradeStreamConsistencyController,
+)
+from src.market_data.acquisition.consistency.trade_stream_state_store import (
+    TradeStreamStateStore,
+)
+from src.market_data.acquisition.models.trade import (
+    Trade,
+    TradeAggressorSide,
+)
+from src.market_data.storage import (
+    MarketDataCheckpointConflictError,
+    PostgresTradeRepository,
+    TradeStorageObservationSink,
+)
+from src.storage.postgres_pool import PostgresConnectionPool
+
+
+pytestmark = pytest.mark.integration
+
+VENUE = "dzengi"
+SYMBOL = "BTC/USD_LEVERAGE"
+EXECUTED_AT = datetime(2026, 8, 1, 12, 0, tzinfo=timezone.utc)
+OBSERVED_AT = EXECUTED_AT + timedelta(seconds=1)
+
+
+def _trade(
+    *,
+    trade_id: int,
+    offset_ms: int = 0,
+    source: str = "dzengi_websocket_trade",
+) -> Trade:
+    return Trade(
+        symbol=SYMBOL,
+        trade_id=trade_id,
+        price=Decimal("65000.25"),
+        quantity=Decimal("0.001"),
+        executed_at=EXECUTED_AT + timedelta(milliseconds=offset_ms),
+        aggressor_side=TradeAggressorSide.BUY,
+        source=source,
+    )
+
+
+def _repository(
+    pool: PostgresConnectionPool,
+) -> PostgresTradeRepository:
+    return PostgresTradeRepository(
+        connection_provider=pool.connection,
+    )
+
+
+def _controller(
+    pool: PostgresConnectionPool,
+) -> tuple[
+    TradeStreamConsistencyController,
+    TradeStreamStateStore,
+]:
+    repository = _repository(pool)
+    state_store = TradeStreamStateStore()
+    sink = TradeStorageObservationSink(
+        trade_storage=repository,
+        venue=VENUE,
+        clock=lambda: OBSERVED_AT,
+    )
+    return (
+        TradeStreamConsistencyController(
+            state_store=state_store,
+            trade_observation_sink=sink,
+        ),
+        state_store,
+    )
+
+
+def _stored_trade_ids(
+    pool: PostgresConnectionPool,
+) -> tuple[int, ...]:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT trade_id
+                FROM market_data.trades
+                WHERE venue = %s AND symbol = %s
+                ORDER BY trade_id
+                """,
+                (VENUE, SYMBOL),
+            )
+            return tuple(row[0] for row in cursor.fetchall())
+
+
+def test_real_consistency_atomically_advances_durable_checkpoint(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    controller, state_store = _controller(migrated_postgres_pool)
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=100)
+    second = _trade(trade_id=101, offset_ms=1)
+
+    controller.accept(first)
+    result = controller.accept(second)
+
+    checkpoint = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    state = state_store.get(SYMBOL)
+
+    assert result is second
+    assert checkpoint is not None
+    assert checkpoint.trade == second
+    assert checkpoint.revision == 2
+    assert state.last_trade is second
+    assert state.last_trade_id == second.trade_id
+    assert _stored_trade_ids(migrated_postgres_pool) == (100, 101)
+
+
+def test_real_duplicate_updates_provenance_without_checkpoint_advance(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    controller, state_store = _controller(migrated_postgres_pool)
+    repository = _repository(migrated_postgres_pool)
+    original = _trade(trade_id=200)
+    duplicate = _trade(
+        trade_id=200,
+        source="dzengi",
+    )
+    controller.accept(original)
+    before_duplicate = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+
+    result = controller.accept(duplicate)
+
+    after_duplicate = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    state = state_store.get(SYMBOL)
+
+    with migrated_postgres_pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT observation_sources
+                FROM market_data.trades
+                WHERE venue = %s
+                  AND symbol = %s
+                  AND trade_id = %s
+                  AND executed_at = %s
+                """,
+                (
+                    VENUE,
+                    SYMBOL,
+                    original.trade_id,
+                    original.executed_at,
+                ),
+            )
+            provenance = cursor.fetchone()
+
+    assert result is None
+    assert before_duplicate is not None
+    assert after_duplicate is not None
+    assert after_duplicate == before_duplicate
+    assert after_duplicate.revision == 1
+    assert provenance == (
+        ["dzengi_websocket_trade", "dzengi"],
+    )
+    assert state.last_trade is original
+    assert state.last_trade_id == original.trade_id
+
+
+def test_real_checkpoint_conflict_rolls_back_trade_and_in_memory_state(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    controller, state_store = _controller(migrated_postgres_pool)
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=300)
+    stale_candidate = _trade(trade_id=301, offset_ms=1)
+    winner = _trade(trade_id=302, offset_ms=2)
+    controller.accept(first)
+    winner_checkpoint = repository.store_trade_and_advance_checkpoint(
+        venue=VENUE,
+        expected_trade=first,
+        trade=winner,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+
+    with pytest.raises(MarketDataCheckpointConflictError):
+        controller.accept(stale_candidate)
+
+    checkpoint = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    state = state_store.get(SYMBOL)
+
+    assert checkpoint == winner_checkpoint
+    assert _stored_trade_ids(migrated_postgres_pool) == (300, 302)
+    assert state.last_trade is first
+    assert state.last_trade_id == first.trade_id
+    assert stale_candidate.trade_id not in state._trades

app/tests/integration/market_data/storage/test_trade_runtime_persistence_integration.py

@@ -391,6 +391,7 @@ def test_application_owns_real_storage_before_and_after_runtime(
             runtime = build_trade_stream_production_runtime(
                 settings,
                 trade_observation_sink=storage.trade_observation_sink,
+                checkpoint_storage=storage.trade_repository,
             )
             assert runtime is not None
             dispatcher = ControlledDispatcher()

app/tests/integration/market_data/storage/test_trade_stream_persistent_cancellation_integration.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import pytest
+
+from src.bootstrap.market_data_storage import build_market_data_storage
+from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
+    TradeStreamProductionRuntimeState,
+)
+from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
+    LoopbackTradeEnvironment,
+    LoopbackTradeRestServer,
+    LoopbackTradeWebSocketServer,
+)
+from tests.integration.market_data.storage.test_trade_stream_persistent_restart_integration import (
+    APPLICATION_TIMEOUT_SECONDS,
+    VENUE,
+    PersistentApplication,
+    _assert_no_pool_connections,
+    _build_application,
+    _canonical_trade,
+    _cleanup_application,
+    _settings_for_database,
+)
+from tests.support.postgres_market_data import (
+    PostgresTestSettings,
+    connect_postgres_test_database,
+    wait_for_postgres_relation_lock_waiters,
+)
+from tests.support.trade_stream_runtime import (
+    assert_no_owned_tasks,
+    run_scenario,
+)
+
+
+pytestmark = pytest.mark.integration
+
+CHECKPOINT_RELATION_NAME = "market_data.trade_stream_checkpoints"
+
+
+def test_cancellation_waits_for_blocked_hydration_before_pool_close(
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    async def scenario() -> None:
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        websocket = LoopbackTradeWebSocketServer(auto_ack=False)
+        rest = LoopbackTradeRestServer()
+        application: PersistentApplication | None = None
+        seed_storage = None
+        blocker = None
+
+        try:
+            async with LoopbackTradeEnvironment(
+                websocket=websocket,
+                rest=rest,
+            ) as environment:
+                settings = _settings_for_database(
+                    postgres=postgres_test_settings,
+                    websocket_url=environment.websocket_url,
+                    rest_base_url=rest.base_url,
+                )
+                seed_storage = build_market_data_storage(settings)
+
+                assert seed_storage is not None
+
+                await asyncio.to_thread(seed_storage.lifecycle.start)
+                await asyncio.to_thread(
+                    seed_storage.trade_repository
+                    .store_trade_and_advance_checkpoint,
+                    venue=VENUE,
+                    expected_trade=None,
+                    trade=_canonical_trade(
+                        trade_id=500,
+                        timestamp_ms=base_time_ms,
+                    ),
+                    observed_at=datetime.now(timezone.utc),
+                )
+                await asyncio.to_thread(seed_storage.lifecycle.stop)
+
+                assert seed_storage.connection_pool.is_open is False
+                await asyncio.to_thread(
+                    _assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+
+                blocker = connect_postgres_test_database(
+                    postgres_test_settings,
+                    autocommit=False,
+                )
+
+                with blocker.cursor() as cursor:
+                    cursor.execute(
+                        "LOCK TABLE "
+                        "market_data.trade_stream_checkpoints "
+                        "IN ACCESS EXCLUSIVE MODE"
+                    )
+
+                application = _build_application(
+                    settings,
+                    task_name="persistent-application-cancellation",
+                )
+                await application.dispatcher.started.wait()
+                await asyncio.to_thread(
+                    wait_for_postgres_relation_lock_waiters,
+                    blocker,
+                    relation_name=CHECKPOINT_RELATION_NAME,
+                    expected_count=1,
+                )
+
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.STARTING
+                )
+                assert application.storage.connection_pool.is_open is True
+                assert websocket.connection_count == 0
+                assert rest.request_count == 0
+
+                application.task.cancel()
+                await asyncio.wait_for(
+                    application.dispatcher.cancelled.wait(),
+                    timeout=APPLICATION_TIMEOUT_SECONDS,
+                )
+
+                assert application.task.done() is False
+                assert application.task.cancelling() == 1
+                assert application.storage.connection_pool.is_open is True
+
+                application.task.cancel()
+                await asyncio.sleep(0)
+
+                assert application.task.done() is False
+                assert application.task.cancelling() == 2
+                assert application.storage.connection_pool.is_open is True
+
+                await asyncio.to_thread(blocker.rollback)
+
+                with pytest.raises(asyncio.CancelledError):
+                    await asyncio.wait_for(
+                        asyncio.shield(application.task),
+                        timeout=APPLICATION_TIMEOUT_SECONDS,
+                    )
+
+                runtime_graph: Any = application.runtime
+
+                assert application.task.cancelled() is True
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.STOPPED
+                )
+                assert runtime_graph._live_processing_gate.failed is False
+                assert runtime_graph._live_processing_gate.locked is False
+                assert application.dispatcher.cancelled.is_set()
+                assert application.storage.connection_pool.is_open is False
+                assert application.storage.lifecycle.started is False
+                assert application.bot_session.close_calls == 1
+        finally:
+            if blocker is not None:
+                try:
+                    if not blocker.closed:
+                        await asyncio.to_thread(blocker.rollback)
+                finally:
+                    await asyncio.to_thread(blocker.close)
+
+            await _cleanup_application(application)
+
+            if (
+                seed_storage is not None
+                and seed_storage.connection_pool.is_open
+            ):
+                await asyncio.to_thread(seed_storage.lifecycle.stop)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await asyncio.to_thread(
+            _assert_no_pool_connections,
+            postgres_test_settings,
+        )
+        await assert_no_owned_tasks()
+
+    run_scenario(
+        scenario(),
+        timeout_seconds=60.0,
+    )

app/tests/integration/market_data/storage/test_trade_stream_persistent_failure_integration.py

@@ -0,0 +1,357 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import pytest
+
+from src.bootstrap.market_data_storage import (
+    MarketDataStorageBootstrapComposition,
+    build_market_data_storage,
+)
+from src.core.config import Settings
+from src.integrations.exchange.exceptions import ExchangeResponseError
+from src.market_data.acquisition.exceptions import TradeTransportError
+from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
+    TradeStreamProductionRuntimeState,
+)
+from src.market_data.storage import MarketDataCheckpointIntegrityError
+from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
+    LoopbackHttpResponse,
+    LoopbackTradeEnvironment,
+    LoopbackTradeRestServer,
+    LoopbackTradeWebSocketServer,
+)
+from tests.integration.market_data.storage import (
+    test_trade_stream_persistent_restart_integration as restart_harness,
+)
+from tests.support.postgres_market_data import (
+    PostgresTestSettings,
+    connect_postgres_test_database,
+)
+from tests.support.trade_stream_runtime import (
+    SYMBOL,
+    assert_no_owned_tasks,
+    run_scenario,
+    state_store_from,
+)
+
+
+pytestmark = pytest.mark.integration
+
+VENUE = "dzengi"
+CHECKPOINT_FOREIGN_KEY = "trade_stream_checkpoints_trade_fk"
+
+
+def _prepare_storage(
+    settings: Settings,
+) -> MarketDataStorageBootstrapComposition:
+    storage = build_market_data_storage(settings)
+
+    assert storage is not None
+    storage.lifecycle.start()
+    return storage
+
+
+def _stop_storage(
+    storage: MarketDataStorageBootstrapComposition | None,
+) -> None:
+    if storage is not None and storage.connection_pool.is_open:
+        storage.lifecycle.stop()
+
+
+def _insert_orphan_checkpoint(
+    postgres: PostgresTestSettings,
+    *,
+    trade_id: int,
+    executed_at: datetime,
+) -> None:
+    """Создать повреждённую точку только в одноразовой тестовой БД."""
+    with connect_postgres_test_database(postgres) as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT COUNT(*)
+                FROM pg_catalog.pg_constraint
+                WHERE conname = %s
+                  AND conrelid = (
+                      'market_data.trade_stream_checkpoints'::regclass
+                  )
+                """,
+                (CHECKPOINT_FOREIGN_KEY,),
+            )
+            assert cursor.fetchone() == (1,)
+            cursor.execute(
+                """
+                ALTER TABLE market_data.trade_stream_checkpoints
+                DROP CONSTRAINT trade_stream_checkpoints_trade_fk
+                """
+            )
+            cursor.execute(
+                """
+                INSERT INTO market_data.trade_stream_checkpoints (
+                    venue,
+                    symbol,
+                    trade_id,
+                    executed_at,
+                    revision,
+                    checkpoint_schema_version
+                )
+                VALUES (%s, %s, %s, %s, 1, 1)
+                """,
+                (
+                    VENUE,
+                    SYMBOL,
+                    trade_id,
+                    executed_at,
+                ),
+            )
+
+
+def _seed_checkpoint(
+    settings: Settings,
+    *,
+    trade_id: int,
+    timestamp_ms: int,
+) -> None:
+    storage: MarketDataStorageBootstrapComposition | None = None
+
+    try:
+        storage = _prepare_storage(settings)
+        trade = restart_harness._canonical_trade(
+            trade_id=trade_id,
+            timestamp_ms=timestamp_ms,
+        )
+        storage.trade_repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=None,
+            trade=trade,
+            observed_at=datetime.now(timezone.utc),
+        )
+    finally:
+        _stop_storage(storage)
+
+
+def test_orphan_checkpoint_fails_before_network_io(
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    async def scenario() -> None:
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        websocket = LoopbackTradeWebSocketServer(auto_ack=False)
+        rest = LoopbackTradeRestServer()
+        application: restart_harness.PersistentApplication | None = None
+        setup_storage: MarketDataStorageBootstrapComposition | None = None
+
+        try:
+            async with LoopbackTradeEnvironment(
+                websocket=websocket,
+                rest=rest,
+            ) as environment:
+                settings = restart_harness._settings_for_database(
+                    postgres=postgres_test_settings,
+                    websocket_url=environment.websocket_url,
+                    rest_base_url=rest.base_url,
+                )
+                setup_storage = await asyncio.to_thread(
+                    _prepare_storage,
+                    settings,
+                )
+                await asyncio.to_thread(
+                    setup_storage.lifecycle.stop,
+                )
+                await asyncio.to_thread(
+                    _insert_orphan_checkpoint,
+                    postgres_test_settings,
+                    trade_id=300,
+                    executed_at=datetime.fromtimestamp(
+                        base_time_ms / 1_000,
+                        tz=timezone.utc,
+                    ),
+                )
+                application = restart_harness._build_application(
+                    settings,
+                    task_name="persistent-application-orphan-checkpoint",
+                )
+
+                with pytest.raises(
+                    MarketDataCheckpointIntegrityError,
+                ) as captured:
+                    await asyncio.wait_for(
+                        asyncio.shield(application.task),
+                        timeout=restart_harness.APPLICATION_TIMEOUT_SECONDS,
+                    )
+
+                assert type(captured.value) is (
+                    MarketDataCheckpointIntegrityError
+                )
+                runtime_graph: Any = application.runtime
+                gate = runtime_graph._live_processing_gate
+
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.FAILED
+                )
+                assert gate.failed is True
+                assert gate.locked is False
+                assert gate._failure is captured.value
+                assert runtime_graph._session.is_connected is False
+                assert (
+                    runtime_graph._subscription_manager.subscription_keys
+                    == ()
+                )
+                assert application.dispatcher.cancelled.is_set()
+                assert state_store_from(application.runtime).is_empty() is True
+                assert websocket.connection_count == 0
+                assert websocket.subscriptions == ()
+                assert rest.request_count == 0
+                assert application.storage.connection_pool.is_open is False
+                assert application.storage.lifecycle.started is False
+                assert application.bot_session.close_calls == 1
+                await asyncio.to_thread(
+                    restart_harness._assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+        finally:
+            await restart_harness._cleanup_application(application)
+            await asyncio.to_thread(_stop_storage, setup_storage)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(
+        scenario(),
+        timeout_seconds=60.0,
+    )
+
+
+def test_rest_failure_rejects_buffered_live_and_preserves_checkpoint(
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    async def scenario() -> None:
+        recovery_release = threading.Event()
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        websocket = LoopbackTradeWebSocketServer(auto_ack=False)
+        rest = LoopbackTradeRestServer(
+            responses=(
+                LoopbackHttpResponse(
+                    body={"error": "recovery unavailable"},
+                    status=500,
+                    release=recovery_release,
+                ),
+            )
+        )
+        application: restart_harness.PersistentApplication | None = None
+
+        try:
+            async with LoopbackTradeEnvironment(
+                websocket=websocket,
+                rest=rest,
+            ) as environment:
+                settings = restart_harness._settings_for_database(
+                    postgres=postgres_test_settings,
+                    websocket_url=environment.websocket_url,
+                    rest_base_url=rest.base_url,
+                )
+                await asyncio.to_thread(
+                    _seed_checkpoint,
+                    settings,
+                    trade_id=400,
+                    timestamp_ms=base_time_ms,
+                )
+                seeded_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+
+                assert tuple(row[0] for row in seeded_snapshot.trades) == (
+                    400,
+                )
+                assert seeded_snapshot.checkpoint == (400, 1, 1)
+
+                application = restart_harness._build_application(
+                    settings,
+                    task_name="persistent-application-rest-failure",
+                )
+                await application.dispatcher.started.wait()
+                await websocket.wait_for_subscriptions(1)
+                state_store = state_store_from(application.runtime)
+
+                assert state_store.get(SYMBOL).last_trade_id == 400
+
+                await websocket.send_trade(
+                    0,
+                    symbol=SYMBOL,
+                    trade_id=402,
+                    timestamp_ms=base_time_ms + 200,
+                )
+                subscription = websocket.subscriptions[0]
+                await websocket.send_ack(
+                    0,
+                    correlation_id=subscription.correlation_id,
+                )
+                await rest.wait_for_requests(1)
+                runtime_graph: Any = application.runtime
+                gate = runtime_graph._live_processing_gate
+
+                assert gate.locked is True
+                assert gate.failed is False
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.STARTING
+                )
+                assert state_store.get(SYMBOL).last_trade_id == 400
+                assert await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                ) == seeded_snapshot
+
+                recovery_release.set()
+
+                with pytest.raises(TradeTransportError) as captured:
+                    await asyncio.wait_for(
+                        asyncio.shield(application.task),
+                        timeout=restart_harness.APPLICATION_TIMEOUT_SECONDS,
+                    )
+
+                assert type(captured.value) is TradeTransportError
+                assert type(captured.value.__cause__) is ExchangeResponseError
+                failed_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+
+                assert failed_snapshot == seeded_snapshot
+                assert state_store.get(SYMBOL).last_trade_id == 400
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.FAILED
+                )
+                assert gate.failed is True
+                assert gate.locked is False
+                assert gate._failure is captured.value
+                assert runtime_graph._session.is_connected is False
+                assert (
+                    runtime_graph._subscription_manager.subscription_keys
+                    == ()
+                )
+                assert application.dispatcher.cancelled.is_set()
+                assert application.storage.connection_pool.is_open is False
+                assert application.storage.lifecycle.started is False
+                assert application.bot_session.close_calls == 1
+                await asyncio.to_thread(
+                    restart_harness._assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+        finally:
+            recovery_release.set()
+            await restart_harness._cleanup_application(application)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(
+        scenario(),
+        timeout_seconds=60.0,
+    )

app/tests/integration/market_data/storage/test_trade_stream_persistent_transaction_failure_integration.py

@@ -0,0 +1,480 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import psycopg
+import pytest
+
+from src.bootstrap.market_data_storage import (
+    MarketDataStorageBootstrapComposition,
+    build_market_data_storage,
+)
+from src.core.config import Settings
+from src.market_data.acquisition.consistency.trade_observation_sink_protocol import (
+    TradeObservationSinkProtocol,
+)
+from src.market_data.acquisition.models.trade import Trade
+from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
+    TradeStreamProductionRuntimeState,
+)
+from src.market_data.storage.exceptions import (
+    MarketDataStorageOperationError,
+)
+from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
+    LoopbackHttpResponse,
+    LoopbackTradeEnvironment,
+    LoopbackTradeRestServer,
+    LoopbackTradeWebSocketServer,
+    wait_until,
+)
+from tests.integration.market_data.storage import (
+    test_trade_stream_persistent_restart_integration as restart_harness,
+)
+from tests.support.postgres_market_data import (
+    PostgresTestSettings,
+    connect_postgres_test_database,
+)
+from tests.support.trade_stream_runtime import (
+    SYMBOL,
+    assert_no_owned_tasks,
+    run_scenario,
+    state_store_from,
+)
+
+
+pytestmark = pytest.mark.integration
+
+VENUE = "dzengi"
+REJECT_TRIGGER = "reject_runtime_checkpoint_write"
+REJECT_FUNCTION = "market_data.reject_runtime_checkpoint_write"
+
+
+class PostCommitTestError(RuntimeError):
+    """Искусственная ошибка после подтверждённой фиксации в PostgreSQL."""
+
+
+class CommitThenFailObservationSink:
+    """Сохранить принятую сделку и имитировать потерю ответа."""
+
+    __slots__ = (
+        "_accepted_trades",
+        "_delegate",
+        "_expected_error",
+    )
+
+    def __init__(
+        self,
+        *,
+        delegate: TradeObservationSinkProtocol,
+        expected_error: PostCommitTestError,
+    ) -> None:
+        self._delegate = delegate
+        self._expected_error = expected_error
+        self._accepted_trades: list[Trade] = []
+
+    @property
+    def accepted_trades(self) -> tuple[Trade, ...]:
+        return tuple(self._accepted_trades)
+
+    def persist_accepted(
+        self,
+        trade: Trade,
+        *,
+        expected_trade: Trade | None,
+    ) -> None:
+        self._delegate.persist_accepted(
+            trade,
+            expected_trade=expected_trade,
+        )
+        self._accepted_trades.append(trade)
+        raise self._expected_error
+
+    def persist_duplicate(
+        self,
+        trade: Trade,
+    ) -> None:
+        self._delegate.persist_duplicate(trade)
+
+
+def _seed_checkpoint(
+    settings: Settings,
+    *,
+    trade_id: int,
+    timestamp_ms: int,
+) -> None:
+    storage = build_market_data_storage(settings)
+
+    assert storage is not None
+
+    try:
+        storage.lifecycle.start()
+        storage.trade_repository.store_trade_and_advance_checkpoint(
+            venue=VENUE,
+            expected_trade=None,
+            trade=restart_harness._canonical_trade(
+                trade_id=trade_id,
+                timestamp_ms=timestamp_ms,
+            ),
+            observed_at=datetime.now(timezone.utc),
+        )
+    finally:
+        if storage.connection_pool.is_open:
+            storage.lifecycle.stop()
+
+
+def _install_checkpoint_rejection_trigger(
+    postgres: PostgresTestSettings,
+) -> None:
+    with connect_postgres_test_database(postgres) as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                CREATE FUNCTION {REJECT_FUNCTION}()
+                RETURNS trigger
+                LANGUAGE plpgsql
+                AS $$
+                BEGIN
+                    RAISE EXCEPTION 'runtime checkpoint write rejected';
+                END;
+                $$
+                """
+            )
+            cursor.execute(
+                f"""
+                CREATE TRIGGER {REJECT_TRIGGER}
+                BEFORE INSERT OR UPDATE
+                ON market_data.trade_stream_checkpoints
+                FOR EACH ROW
+                EXECUTE FUNCTION {REJECT_FUNCTION}()
+                """
+            )
+
+
+def _remove_checkpoint_rejection_trigger(
+    postgres: PostgresTestSettings,
+) -> None:
+    with connect_postgres_test_database(postgres) as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                DROP TRIGGER IF EXISTS {REJECT_TRIGGER}
+                ON market_data.trade_stream_checkpoints
+                """
+            )
+            cursor.execute(
+                f"DROP FUNCTION IF EXISTS {REJECT_FUNCTION}()"
+            )
+
+
+def test_checkpoint_write_failure_rolls_back_trade_and_runtime_state(
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    async def scenario() -> None:
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        websocket = LoopbackTradeWebSocketServer(auto_ack=False)
+        rest = LoopbackTradeRestServer()
+        application: restart_harness.PersistentApplication | None = None
+        trigger_installed = False
+
+        try:
+            async with LoopbackTradeEnvironment(
+                websocket=websocket,
+                rest=rest,
+            ) as environment:
+                settings = restart_harness._settings_for_database(
+                    postgres=postgres_test_settings,
+                    websocket_url=environment.websocket_url,
+                    rest_base_url=rest.base_url,
+                )
+                await asyncio.to_thread(
+                    _seed_checkpoint,
+                    settings,
+                    trade_id=300,
+                    timestamp_ms=base_time_ms,
+                )
+                seeded_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+                await asyncio.to_thread(
+                    _install_checkpoint_rejection_trigger,
+                    postgres_test_settings,
+                )
+                trigger_installed = True
+                application = restart_harness._build_application(
+                    settings,
+                    task_name="persistent-application-before-commit-failure",
+                )
+                await application.dispatcher.started.wait()
+                await websocket.wait_for_subscriptions(1)
+                subscription = websocket.subscriptions[0]
+                await websocket.send_ack(
+                    0,
+                    correlation_id=subscription.correlation_id,
+                )
+                await wait_until(
+                    lambda: application.runtime.state
+                    is TradeStreamProductionRuntimeState.RUNNING,
+                )
+                state_store = state_store_from(application.runtime)
+                state_before_failure = state_store.get(SYMBOL)
+
+                assert state_before_failure.last_trade_id == 300
+
+                await websocket.send_trade(
+                    0,
+                    symbol=SYMBOL,
+                    trade_id=301,
+                    timestamp_ms=base_time_ms + 100,
+                )
+
+                with pytest.raises(
+                    MarketDataStorageOperationError,
+                ) as captured:
+                    await asyncio.wait_for(
+                        asyncio.shield(application.task),
+                        timeout=restart_harness.APPLICATION_TIMEOUT_SECONDS,
+                    )
+
+                assert type(captured.value) is (
+                    MarketDataStorageOperationError
+                )
+                database_error = captured.value.__cause__
+
+                assert type(database_error) is psycopg.errors.RaiseException
+                assert database_error.sqlstate == "P0001"
+                failed_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+                runtime_graph: Any = application.runtime
+                gate = runtime_graph._live_processing_gate
+
+                assert failed_snapshot == seeded_snapshot
+                assert state_store.get(SYMBOL) is state_before_failure
+                assert state_before_failure.last_trade_id == 300
+                assert application.runtime.state is (
+                    TradeStreamProductionRuntimeState.FAILED
+                )
+                assert gate.failed is True
+                assert gate.locked is False
+                assert gate._failure is captured.value
+                assert runtime_graph._session.is_connected is False
+                assert (
+                    runtime_graph._subscription_manager.subscription_keys
+                    == ()
+                )
+                assert application.dispatcher.cancelled.is_set()
+                assert application.storage.connection_pool.is_open is False
+                assert application.storage.lifecycle.started is False
+                assert application.bot_session.close_calls == 1
+                await asyncio.to_thread(
+                    restart_harness._assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+        finally:
+            await restart_harness._cleanup_application(application)
+
+            if trigger_installed:
+                await asyncio.to_thread(
+                    _remove_checkpoint_rejection_trigger,
+                    postgres_test_settings,
+                )
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(
+        scenario(),
+        timeout_seconds=60.0,
+    )
+
+
+def test_post_commit_error_is_recovered_by_fresh_runtime_graph(
+    postgres_test_settings: PostgresTestSettings,
+) -> None:
+    async def scenario() -> None:
+        base_time_ms = time.time_ns() // 1_000_000 - 10_000
+        expected_error = PostCommitTestError(
+            "caller did not observe committed checkpoint"
+        )
+        websocket = LoopbackTradeWebSocketServer(auto_ack=False)
+        rest = LoopbackTradeRestServer(
+            responses=(
+                LoopbackHttpResponse(
+                    body=[
+                        restart_harness._rest_trade(
+                            trade_id=600,
+                            timestamp_ms=base_time_ms,
+                        ),
+                    ]
+                ),
+            )
+        )
+        first: restart_harness.PersistentApplication | None = None
+        second: restart_harness.PersistentApplication | None = None
+        failing_sink: CommitThenFailObservationSink | None = None
+
+        try:
+            async with LoopbackTradeEnvironment(
+                websocket=websocket,
+                rest=rest,
+            ) as environment:
+                settings = restart_harness._settings_for_database(
+                    postgres=postgres_test_settings,
+                    websocket_url=environment.websocket_url,
+                    rest_base_url=rest.base_url,
+                )
+
+                def failing_sink_factory(
+                    storage: MarketDataStorageBootstrapComposition,
+                ) -> TradeObservationSinkProtocol:
+                    nonlocal failing_sink
+                    failing_sink = CommitThenFailObservationSink(
+                        delegate=storage.trade_observation_sink,
+                        expected_error=expected_error,
+                    )
+                    return failing_sink
+
+                first = restart_harness._build_application(
+                    settings,
+                    task_name="persistent-application-post-commit-failure",
+                    trade_observation_sink_factory=failing_sink_factory,
+                )
+                await first.dispatcher.started.wait()
+                await websocket.wait_for_subscriptions(1)
+                first_subscription = websocket.subscriptions[0]
+                await websocket.send_ack(
+                    0,
+                    correlation_id=first_subscription.correlation_id,
+                )
+                await wait_until(
+                    lambda: first.runtime.state
+                    is TradeStreamProductionRuntimeState.RUNNING,
+                )
+                first_store = state_store_from(first.runtime)
+                first_state = first_store.get(SYMBOL)
+
+                assert first_state.last_trade_id is None
+
+                await websocket.send_trade(
+                    0,
+                    symbol=SYMBOL,
+                    trade_id=600,
+                    timestamp_ms=base_time_ms,
+                )
+
+                with pytest.raises(PostCommitTestError) as captured:
+                    await asyncio.wait_for(
+                        asyncio.shield(first.task),
+                        timeout=restart_harness.APPLICATION_TIMEOUT_SECONDS,
+                    )
+
+                assert captured.value is expected_error
+                assert failing_sink is not None
+                assert len(failing_sink.accepted_trades) == 1
+                committed_object = failing_sink.accepted_trades[0]
+                committed_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+
+                assert tuple(row[0] for row in committed_snapshot.trades) == (
+                    600,
+                )
+                assert committed_snapshot.checkpoint == (600, 1, 1)
+                assert first_state.last_trade_id is None
+                assert first.runtime.state is (
+                    TradeStreamProductionRuntimeState.FAILED
+                )
+                first_graph: Any = first.runtime
+
+                assert first_graph._live_processing_gate._failure is (
+                    expected_error
+                )
+                assert first_graph._session.is_connected is False
+                assert (
+                    first_graph._subscription_manager.subscription_keys
+                    == ()
+                )
+                assert first.dispatcher.cancelled.is_set()
+                assert first.storage.connection_pool.is_open is False
+                await wait_until(
+                    lambda: websocket.active_handler_count == 0,
+                )
+                await assert_no_owned_tasks()
+                await asyncio.to_thread(
+                    restart_harness._assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+
+                second = restart_harness._build_application(
+                    settings,
+                    task_name="persistent-application-post-commit-restart",
+                )
+
+                assert second.storage is not first.storage
+                assert second.storage.trade_repository is not (
+                    first.storage.trade_repository
+                )
+                assert second.runtime is not first.runtime
+
+                await second.dispatcher.started.wait()
+                await websocket.wait_for_subscriptions(2)
+                second_store = state_store_from(second.runtime)
+                second_state = second_store.get(SYMBOL)
+                hydrated_object = second_state.last_trade
+
+                assert second_state is not first_state
+                assert hydrated_object == committed_object
+                assert hydrated_object is not committed_object
+                assert second_state.last_trade_id == 600
+                assert tuple(second_state._trade_window) == (600,)
+
+                second_subscription = websocket.subscriptions[1]
+                await websocket.send_ack(
+                    1,
+                    correlation_id=second_subscription.correlation_id,
+                )
+                await rest.wait_for_requests(1)
+                await wait_until(
+                    lambda: second.runtime.state
+                    is TradeStreamProductionRuntimeState.RUNNING,
+                )
+                recovered_snapshot = await asyncio.to_thread(
+                    restart_harness._snapshot_database,
+                    postgres_test_settings,
+                )
+
+                assert tuple(row[0] for row in recovered_snapshot.trades) == (
+                    600,
+                )
+                assert recovered_snapshot.checkpoint == (600, 1, 1)
+                assert recovered_snapshot.trades[0][2] == [
+                    "dzengi_websocket_trade",
+                    "dzengi",
+                ]
+                assert second_state.last_trade_id == 600
+                assert second_state.last_trade is hydrated_object
+
+                await restart_harness._stop_application(second)
+                await asyncio.to_thread(
+                    restart_harness._assert_no_pool_connections,
+                    postgres_test_settings,
+                )
+        finally:
+            await restart_harness._cleanup_application(second)
+            await restart_harness._cleanup_application(first)
+
+        assert websocket.active_handler_count == 0
+        assert rest.thread_is_alive is False
+        await assert_no_owned_tasks()
+
+    run_scenario(
+        scenario(),
+        timeout_seconds=60.0,
+    )

app/tests/integration/market_data/storage/test_trade_stream_state_hydration_integration.py

@@ -0,0 +1,283 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+
+import pytest
+
+from src.market_data.acquisition.checkpoint import (
+    TradeStreamStateHydrator,
+)
+from src.market_data.acquisition.consistency.trade_stream_consistency_controller import (
+    TradeStreamConsistencyController,
+)
+from src.market_data.acquisition.consistency.trade_stream_state_store import (
+    TradeStreamStateStore,
+)
+from src.market_data.acquisition.models.trade import (
+    Trade,
+    TradeAggressorSide,
+)
+from src.market_data.acquisition.trade_id_sequence import (
+    SIGNED_TRADE_ID_MAX,
+    SIGNED_TRADE_ID_MIN,
+)
+from src.market_data.storage import (
+    MarketDataCheckpointIntegrityError,
+    PostgresTradeRepository,
+    TradeStorageObservationSink,
+)
+from src.storage.postgres_pool import PostgresConnectionPool
+
+
+pytestmark = pytest.mark.integration
+
+VENUE = "dzengi"
+SYMBOL = "BTC/USD_LEVERAGE"
+SECOND_SYMBOL = "ETH/USD_LEVERAGE"
+EVENT_TIME = datetime(2026, 8, 1, 14, 0, tzinfo=timezone.utc)
+OBSERVED_AT = EVENT_TIME + timedelta(seconds=1)
+
+
+def _trade(
+    *,
+    trade_id: int,
+    offset_ms: int = 0,
+    symbol: str = SYMBOL,
+    source: str = "dzengi_websocket_trade",
+) -> Trade:
+    return Trade(
+        symbol=symbol,
+        trade_id=trade_id,
+        price=Decimal("65000.25"),
+        quantity=Decimal("0.001"),
+        executed_at=EVENT_TIME + timedelta(milliseconds=offset_ms),
+        aggressor_side=TradeAggressorSide.BUY,
+        source=source,
+    )
+
+
+def _repository(
+    pool: PostgresConnectionPool,
+) -> PostgresTradeRepository:
+    return PostgresTradeRepository(
+        connection_provider=pool.connection,
+    )
+
+
+def _trade_metadata(
+    pool: PostgresConnectionPool,
+    *,
+    symbol: str,
+) -> tuple[tuple[object, ...], ...]:
+    with pool.connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT
+                    trade_id,
+                    executed_at,
+                    source,
+                    first_observed_at,
+                    last_observed_at,
+                    observation_sources
+                FROM market_data.trades
+                WHERE venue = %s AND symbol = %s
+                ORDER BY first_observed_at, executed_at, trade_id
+                """,
+                (VENUE, symbol),
+            )
+            return tuple(cursor.fetchall())
+
+
+def test_real_hydration_adopts_history_without_rewriting_trades(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=100)
+    second = _trade(trade_id=101, offset_ms=1)
+    repository.store_trade(
+        venue=VENUE,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=second,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+    metadata_before = _trade_metadata(
+        migrated_postgres_pool,
+        symbol=SYMBOL,
+    )
+
+    first_store = TradeStreamStateStore()
+    first_states = TradeStreamStateHydrator(
+        checkpoint_storage=repository,
+        state_store=first_store,
+        venue=VENUE,
+        deduplication_window_size=2,
+    ).hydrate(symbols=(SYMBOL,))
+    first_checkpoint = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+
+    assert len(first_states) == 1
+    assert tuple(first_states[0]._trade_window) == (100, 101)
+    assert first_states[0].last_trade == second
+    assert first_checkpoint is not None
+    assert first_checkpoint.trade == second
+    assert first_checkpoint.revision == 1
+    assert _trade_metadata(
+        migrated_postgres_pool,
+        symbol=SYMBOL,
+    ) == metadata_before
+
+    restart_store = TradeStreamStateStore()
+    restart_state = TradeStreamStateHydrator(
+        checkpoint_storage=repository,
+        state_store=restart_store,
+        venue=VENUE,
+        deduplication_window_size=2,
+    ).hydrate(symbols=(SYMBOL,))[0]
+    repeated_checkpoint = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+
+    assert tuple(restart_state._trade_window) == (100, 101)
+    assert restart_state.accept(
+        replace(first, source="dzengi_recovery"),
+    ) is None
+    assert repeated_checkpoint is not None
+    assert repeated_checkpoint == first_checkpoint
+    assert repeated_checkpoint.revision == 1
+    assert _trade_metadata(
+        migrated_postgres_pool,
+        symbol=SYMBOL,
+    ) == metadata_before
+
+    controller = TradeStreamConsistencyController(
+        state_store=restart_store,
+        trade_observation_sink=TradeStorageObservationSink(
+            trade_storage=repository,
+            venue=VENUE,
+            clock=lambda: OBSERVED_AT + timedelta(seconds=2),
+        ),
+    )
+    third = _trade(trade_id=102, offset_ms=2)
+
+    assert controller.accept(third) is third
+
+    advanced_checkpoint = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    assert advanced_checkpoint is not None
+    assert advanced_checkpoint.trade == third
+    assert advanced_checkpoint.revision == 2
+    assert restart_state.last_trade is third
+
+
+@pytest.mark.parametrize(
+    ("first_trade_id", "second_trade_id", "next_trade_id"),
+    (
+        (SIGNED_TRADE_ID_MAX, SIGNED_TRADE_ID_MIN, SIGNED_TRADE_ID_MIN + 1),
+        (-1, 0, 1),
+    ),
+)
+def test_real_hydration_restores_rollover_aware_deduplication_tail(
+    migrated_postgres_pool: PostgresConnectionPool,
+    first_trade_id: int,
+    second_trade_id: int,
+    next_trade_id: int,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    first = _trade(trade_id=first_trade_id)
+    second = _trade(trade_id=second_trade_id, offset_ms=1)
+    repository.store_trade(
+        venue=VENUE,
+        trade=first,
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=second,
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+    state_store = TradeStreamStateStore()
+
+    state = TradeStreamStateHydrator(
+        checkpoint_storage=repository,
+        state_store=state_store,
+        venue=VENUE,
+        deduplication_window_size=2,
+    ).hydrate(symbols=(SYMBOL,))[0]
+
+    assert tuple(state._trade_window) == (
+        first_trade_id,
+        second_trade_id,
+    )
+    assert state.last_trade == second
+    assert state.accept(
+        replace(first, source="dzengi_recovery"),
+    ) is None
+
+    next_trade = _trade(trade_id=next_trade_id, offset_ms=2)
+    assert state.accept(next_trade) is next_trade
+    assert state.last_trade is next_trade
+
+
+def test_real_multi_symbol_failure_does_not_publish_partial_state(
+    migrated_postgres_pool: PostgresConnectionPool,
+) -> None:
+    repository = _repository(migrated_postgres_pool)
+    repository.store_trade(
+        venue=VENUE,
+        trade=_trade(trade_id=10),
+        observed_at=OBSERVED_AT,
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=_trade(
+            symbol=SECOND_SYMBOL,
+            trade_id=20,
+            offset_ms=1,
+        ),
+        observed_at=OBSERVED_AT + timedelta(seconds=1),
+    )
+    repository.store_trade(
+        venue=VENUE,
+        trade=_trade(
+            symbol=SECOND_SYMBOL,
+            trade_id=19,
+            offset_ms=2,
+        ),
+        observed_at=OBSERVED_AT + timedelta(seconds=2),
+    )
+    state_store = TradeStreamStateStore()
+
+    with pytest.raises(MarketDataCheckpointIntegrityError):
+        TradeStreamStateHydrator(
+            checkpoint_storage=repository,
+            state_store=state_store,
+            venue=VENUE,
+            deduplication_window_size=2,
+        ).hydrate(symbols=(SYMBOL, SECOND_SYMBOL))
+
+    assert state_store.is_empty() is True
+    assert state_store.contains(SYMBOL) is False
+    assert state_store.contains(SECOND_SYMBOL) is False
+
+    adopted_first_symbol = repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SYMBOL,
+    )
+    assert adopted_first_symbol is not None
+    assert adopted_first_symbol.trade.trade_id == 10
+    assert repository.load_checkpoint(
+        venue=VENUE,
+        symbol=SECOND_SYMBOL,
+    ) is None