diff --git a/app/pytest.ini b/app/pytest.ini index d38352d..c353b8f 100644 --- a/app/pytest.ini +++ b/app/pytest.ini @@ -15,4 +15,11 @@ python_functions = addopts = -ra - --strict-markers \ No newline at end of file + --strict-markers + -m + "not integration and not stress and not live" + +markers = + integration: deterministic integration tests using local network endpoints + stress: explicit local stress and soak verification + live: opt-in verification against explicitly configured live endpoints diff --git a/app/src/market_data/acquisition/consistency/trade_stream_state.py b/app/src/market_data/acquisition/consistency/trade_stream_state.py index 81122e8..8f58fe5 100644 --- a/app/src/market_data/acquisition/consistency/trade_stream_state.py +++ b/app/src/market_data/acquisition/consistency/trade_stream_state.py @@ -10,6 +10,10 @@ from src.market_data.acquisition.consistency.trade_stream_exceptions import ( TradeOrderingError, ) from src.market_data.acquisition.models.trade import Trade +from src.market_data.acquisition.trade_id_sequence import ( + is_trade_id_newer, + validate_signed_trade_id, +) DEFAULT_DEDUPLICATION_WINDOW_SIZE = 10_000 @@ -65,15 +69,22 @@ class TradeStreamState: ) trade_id = trade.trade_id + validate_signed_trade_id(trade_id) if self.last_trade_id is not None: - if trade_id < self.last_trade_id: + if not is_trade_id_newer( + trade_id, + self.last_trade_id, + ) and trade_id != self.last_trade_id: previous = self._trades.get(trade_id) if previous is None: raise TradeOrderingError() - if previous == trade: + if self._is_same_market_trade( + previous, + trade, + ): return None raise TradeConsistencyError() @@ -81,7 +92,10 @@ class TradeStreamState: previous = self._trades.get(trade_id) if previous is not None: - if previous == trade: + if self._is_same_market_trade( + previous, + trade, + ): return None raise TradeConsistencyError() @@ -93,6 +107,27 @@ class TradeStreamState: return trade + @staticmethod + def _is_same_market_trade( + first: Trade, + second: Trade, + ) -> bool: + """ + Сравнить биржевой факт независимо от transport source. + + Одна сделка закономерно приходит через WebSocket и повторяется + через REST recovery. Поле ``source`` описывает путь доставки и + поэтому не является частью identity биржевой сделки. + """ + return ( + first.symbol == second.symbol + and first.trade_id == second.trade_id + and first.price == second.price + and first.quantity == second.quantity + and first.executed_at == second.executed_at + and first.aggressor_side is second.aggressor_side + ) + def _append( self, trade: Trade, diff --git a/app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py b/app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py index 15c04a3..fd56da6 100644 --- a/app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py +++ b/app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py @@ -5,6 +5,9 @@ from __future__ import annotations from collections.abc import Iterable from src.market_data.acquisition.models.trade import Trade +from src.market_data.acquisition.trade_id_sequence import ( + trade_id_relative_offset, +) def normalize_recovered_trades( @@ -13,13 +16,18 @@ def normalize_recovered_trades( """ Нормализовать порядок восстановленных сделок. - Сделки возвращаются в возрастающем порядке по ``trade_id``. + Сделки возвращаются в хронологическом порядке signed 32-bit + ``trade_id`` с учётом rollover. Исходная последовательность не изменяется. Нормализатор намеренно не выполняет дедупликацию и не проверяет согласованность сделок. Идентичные и конфликтующие дубликаты должны обрабатываться экземпляром ``TradeStreamConsistencyController``. + Входной набор должен покрывать меньше половины 32-битного цикла. + Это ограничение делает modular-порядок однозначным и существенно + шире максимального часового Recovery window Runtime. + Parameters ---------- trades @@ -29,12 +37,22 @@ def normalize_recovered_trades( ------- tuple[Trade, ...] Неизменяемая последовательность сделок, отсортированная - по возрастанию ``trade_id``. + по rollover-aware порядку ``trade_id``. """ + source = tuple(trades) + + if not source: + return () + + reference_trade_id = source[0].trade_id + return tuple( sorted( - trades, - key=lambda trade: trade.trade_id, + source, + key=lambda trade: trade_id_relative_offset( + trade.trade_id, + reference_trade_id, + ), ) - ) \ No newline at end of file + ) diff --git a/app/src/market_data/acquisition/trade_id_sequence.py b/app/src/market_data/acquisition/trade_id_sequence.py new file mode 100644 index 0000000..98cff41 --- /dev/null +++ b/app/src/market_data/acquisition/trade_id_sequence.py @@ -0,0 +1,75 @@ +from __future__ import annotations + + +SIGNED_TRADE_ID_BITS = 32 +SIGNED_TRADE_ID_HALF_RANGE = 1 << (SIGNED_TRADE_ID_BITS - 1) +SIGNED_TRADE_ID_MODULUS = 1 << SIGNED_TRADE_ID_BITS +SIGNED_TRADE_ID_MIN = -SIGNED_TRADE_ID_HALF_RANGE +SIGNED_TRADE_ID_MAX = SIGNED_TRADE_ID_HALF_RANGE - 1 + + +def validate_signed_trade_id(trade_id: int) -> None: + """Проверить raw Trade ID провайдера как signed 32-bit integer.""" + if isinstance(trade_id, bool) or not isinstance(trade_id, int): + raise TypeError("trade_id must be an integer") + + if not SIGNED_TRADE_ID_MIN <= trade_id <= SIGNED_TRADE_ID_MAX: + raise ValueError("trade_id must fit signed 32-bit range") + + +def trade_id_relative_offset( + candidate_trade_id: int, + reference_trade_id: int, +) -> int: + """ + Вернуть signed modular-смещение candidate относительно reference. + + Положительное значение означает более новый ID, отрицательное — + более старый. Контракт однозначен для расстояний меньше половины + 32-битного цикла. Ровно половина цикла не имеет определимого + направления и отклоняется явно. + """ + validate_signed_trade_id(candidate_trade_id) + validate_signed_trade_id(reference_trade_id) + + forward_distance = ( + candidate_trade_id - reference_trade_id + ) % SIGNED_TRADE_ID_MODULUS + + if forward_distance == SIGNED_TRADE_ID_HALF_RANGE: + raise ValueError( + "Trade ID distance is exactly half of the 32-bit cycle" + ) + + if forward_distance > SIGNED_TRADE_ID_HALF_RANGE: + return forward_distance - SIGNED_TRADE_ID_MODULUS + + return forward_distance + + +def is_trade_id_newer( + candidate_trade_id: int, + reference_trade_id: int, +) -> bool: + """Проверить, следует ли candidate после reference по 32-bit циклу.""" + return ( + trade_id_relative_offset( + candidate_trade_id, + reference_trade_id, + ) + > 0 + ) + + +def is_trade_id_same_or_newer( + candidate_trade_id: int, + reference_trade_id: int, +) -> bool: + """Проверить равенство либо продвижение по 32-bit циклу.""" + return ( + trade_id_relative_offset( + candidate_trade_id, + reference_trade_id, + ) + >= 0 + ) diff --git a/app/src/market_data/acquisition/validation/values.py b/app/src/market_data/acquisition/validation/values.py index 7d51a33..f0e71af 100644 --- a/app/src/market_data/acquisition/validation/values.py +++ b/app/src/market_data/acquisition/validation/values.py @@ -27,6 +27,9 @@ from src.market_data.acquisition.exceptions import ( QuoteValueError, TradeValueError, ) +from src.market_data.acquisition.trade_id_sequence import ( + validate_signed_trade_id, +) _SUPPORTED_WEBSOCKET_OHLC_INTERVALS = frozenset( @@ -831,7 +834,7 @@ def validate_dzengi_websocket_trade_values( значения в Decimal и не выполняет mapping во внутреннюю модель Trade. """ - _trade_positive_int( + _trade_signed_int( event.trade_id, path="$.payload.id", ) @@ -881,7 +884,7 @@ def _validate_rest_agg_trade( *, path: str, ) -> None: - _trade_positive_int( + _trade_signed_int( trade.aggregate_trade_id, path=f"{path}.aggregateTradeId", ) @@ -910,6 +913,20 @@ def _trade_positive_int( ) +def _trade_signed_int( + value: int, + *, + path: str, +) -> None: + try: + validate_signed_trade_id(value) + except (TypeError, ValueError) as error: + raise TradeValueError( + f"{path} должно быть целым числом " + "в диапазоне signed 32-bit." + ) from error + + def _trade_positive_decimal( value: DzengiRawNumeric, *, diff --git a/app/tests/integration/market_data/acquisition/runtime/loopback_trade_exchange.py b/app/tests/integration/market_data/acquisition/runtime/loopback_trade_exchange.py new file mode 100644 index 0000000..8fa0a1d --- /dev/null +++ b/app/tests/integration/market_data/acquisition/runtime/loopback_trade_exchange.py @@ -0,0 +1,904 @@ +from __future__ import annotations + +import asyncio +import json +import threading +from collections import deque +from collections.abc import Callable +from contextlib import AsyncExitStack +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import TracebackType +from typing import Any +from urllib.parse import parse_qs, urlsplit + +from websockets.asyncio.server import ( + Server, + ServerConnection, + serve, +) +from websockets.exceptions import ConnectionClosed +from websockets.protocol import State +from websockets.typing import Subprotocol + +from tests.support.async_wait import wait_until + +RESOURCE_CLEANUP_TIMEOUT_SECONDS = 5.0 +ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS = 20.0 + + +@dataclass(frozen=True, slots=True) +class WebSocketSubscriptionRecord: + connection_index: int + correlation_id: str + symbols: tuple[str, ...] + + +class LoopbackTradeWebSocketServer: + """Управляемый локальный Dzengi-подобный WebSocket endpoint.""" + + def __init__( + self, + *, + events: list[str] | None = None, + auto_ack: bool = True, + ) -> None: + self._events = events if events is not None else [] + self._auto_ack = auto_ack + self._server: Server | None = None + self._host = "127.0.0.1" + self._port: int | None = None + self._connections: list[ServerConnection] = [] + self._handler_tasks: set[asyncio.Task[Any]] = set() + self._subscriptions: list[WebSocketSubscriptionRecord] = [] + self._received_documents: list[dict[str, Any]] = [] + + @property + def url(self) -> str: + if self._port is None: + raise RuntimeError("WebSocket server is not started.") + + return f"ws://{self._host}:{self._port}" + + @property + def port(self) -> int: + if self._port is None: + raise RuntimeError("WebSocket server is not started.") + + return self._port + + @property + def connection_count(self) -> int: + return len(self._connections) + + @property + def active_handler_count(self) -> int: + return sum( + not task.done() + for task in self._handler_tasks + ) + + @property + def active_connection_count(self) -> int: + return sum( + connection.state is State.OPEN + for connection in self._connections + ) + + @property + def subscriptions( + self, + ) -> tuple[WebSocketSubscriptionRecord, ...]: + return tuple(self._subscriptions) + + @property + def received_documents( + self, + ) -> tuple[dict[str, Any], ...]: + return tuple(self._received_documents) + + async def start(self) -> None: + if self._server is not None: + raise RuntimeError("WebSocket server is already started.") + + server = await serve( + self._handle_connection, + self._host, + 0, + subprotocols=(Subprotocol("json"),), + ping_interval=None, + close_timeout=0.2, + ) + self._server = server + + sockets = tuple(server.sockets) + + if not sockets: + server.close() + await server.wait_closed() + self._server = None + raise RuntimeError( + "WebSocket server did not expose a listening socket." + ) + + socket = sockets[0] + address = socket.getsockname() + self._port = int(address[1]) + + async def stop(self) -> None: + server = self._server + + if server is None: + return + + self._server = None + self._port = None + server.close() + + try: + await asyncio.wait_for( + server.wait_closed(), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + except TimeoutError: + for connection in tuple(self._connections): + if connection.state is State.OPEN: + connection.transport.abort() + + await asyncio.wait_for( + server.wait_closed(), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + + if self.active_handler_count: + raise AssertionError( + "WebSocket server handlers did not stop." + ) + + async def wait_for_connections( + self, + count: int, + *, + timeout_seconds: float = 3.0, + ) -> None: + await wait_until( + lambda: self.connection_count >= count, + timeout_seconds=timeout_seconds, + ) + + async def wait_for_subscriptions( + self, + count: int, + *, + timeout_seconds: float = 3.0, + ) -> None: + await wait_until( + lambda: len(self._subscriptions) >= count, + timeout_seconds=timeout_seconds, + ) + + async def send_raw( + self, + connection_index: int, + message: str | bytes, + ) -> None: + await self._connections[connection_index].send(message) + + async def send_trade( + self, + connection_index: int, + *, + symbol: str, + trade_id: int, + timestamp_ms: int, + price: str = "64555.55", + quantity: str = "0.002", + buyer: bool = True, + ) -> None: + await self.send_raw( + connection_index, + json.dumps( + { + "status": "OK", + "destination": "internal.trade", + "payload": { + "id": trade_id, + "price": price, + "size": quantity, + "ts": timestamp_ms, + "symbol": symbol, + "buyer": buyer, + "orderId": f"order-{trade_id}", + }, + } + ), + ) + + async def abort_connection( + self, + connection_index: int, + ) -> None: + connection = self._connections[connection_index] + connection.transport.abort() + await asyncio.sleep(0) + + async def close_connection( + self, + connection_index: int, + *, + code: int = 1012, + reason: str = "integration test reconnect", + ) -> None: + await self._connections[connection_index].close( + code=code, + reason=reason, + ) + + async def _handle_connection( + self, + connection: ServerConnection, + ) -> None: + task = asyncio.current_task() + + if task is None: + raise RuntimeError("WebSocket handler has no asyncio task.") + + self._handler_tasks.add(task) + connection_index = len(self._connections) + self._connections.append(connection) + self._events.append(f"ws.connect:{connection_index}") + + try: + async for raw_message in connection: + if not isinstance(raw_message, str): + continue + + try: + document = json.loads(raw_message) + except json.JSONDecodeError: + continue + + if not isinstance(document, dict): + continue + + self._received_documents.append(document) + + if document.get("destination") != "trades.subscribe": + continue + + correlation_id = document.get("correlationId") + payload = document.get("payload") + + if ( + not isinstance(correlation_id, str) + or not isinstance(payload, dict) + ): + continue + + raw_symbols = payload.get("symbols") + + if not isinstance(raw_symbols, list) or not all( + isinstance(symbol, str) + for symbol in raw_symbols + ): + continue + + record = WebSocketSubscriptionRecord( + connection_index=connection_index, + correlation_id=correlation_id, + symbols=tuple(raw_symbols), + ) + self._subscriptions.append(record) + self._events.append( + f"ws.subscribe:{connection_index}" + ) + + if self._auto_ack: + await connection.send( + json.dumps( + { + "correlationId": correlation_id, + "destination": "trades.subscribe", + "status": "OK", + } + ) + ) + except ConnectionClosed: + pass + finally: + self._events.append( + f"ws.disconnect:{connection_index}" + ) + self._handler_tasks.discard(task) + + +@dataclass(frozen=True, slots=True) +class LoopbackHttpResponse: + body: object + status: int = 200 + release: threading.Event | None = None + + +@dataclass(frozen=True, slots=True) +class LoopbackHttpRequest: + path: str + query: dict[str, tuple[str, ...]] + + +class _LoopbackThreadingHttpServer(ThreadingHTTPServer): + daemon_threads = False + block_on_close = True + + +class LoopbackTradeRestServer: + """Локальный HTTP endpoint для настоящего ExchangeRestClient.""" + + def __init__( + self, + *, + responses: tuple[LoopbackHttpResponse, ...] = (), + events: list[str] | None = None, + ) -> None: + self._events = events if events is not None else [] + self._responses = deque(responses) + self._requests: list[LoopbackHttpRequest] = [] + self._lock = threading.Lock() + self._release_events = { + response.release + for response in responses + if response.release is not None + } + self._server: _LoopbackThreadingHttpServer | None = None + self._thread: threading.Thread | None = None + self._host = "127.0.0.1" + self._port: int | None = None + + @property + def base_url(self) -> str: + if self._port is None: + raise RuntimeError("HTTP server is not started.") + + return f"http://{self._host}:{self._port}" + + @property + def request_count(self) -> int: + with self._lock: + return len(self._requests) + + @property + def requests(self) -> tuple[LoopbackHttpRequest, ...]: + with self._lock: + return tuple(self._requests) + + @property + def thread_is_alive(self) -> bool: + thread = self._thread + return thread is not None and thread.is_alive() + + def start(self) -> None: + if self._server is not None: + raise RuntimeError("HTTP server is already started.") + + controller = self + + class RequestHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self) -> None: # noqa: N802 + controller._handle_get(self) + + def log_message( + self, + format: str, + *args: object, + ) -> None: + del format, args + + server = _LoopbackThreadingHttpServer( + (self._host, 0), + RequestHandler, + ) + self._server = server + self._port = int(server.server_address[1]) + self._thread = threading.Thread( + target=lambda: server.serve_forever( + poll_interval=0.05, + ), + name="loopback-trade-rest-server", + ) + + try: + self._thread.start() + except BaseException: + server.server_close() + self._server = None + self._thread = None + self._port = None + raise + + def stop(self) -> None: + server = self._server + thread = self._thread + + if server is None: + return + + self._server = None + self._thread = None + self._port = None + + for release_event in self._release_events: + release_event.set() + + shutdown_error: BaseException | None = None + + try: + server.shutdown() + except BaseException as error: + shutdown_error = error + finally: + try: + server.server_close() + except BaseException as error: + if shutdown_error is None: + shutdown_error = error + else: + shutdown_error.add_note( + "Loopback HTTP server_close also failed: " + f"{type(error).__name__}." + ) + + if thread is not None: + thread.join(timeout=3) + + if thread.is_alive(): + thread_error = AssertionError( + "Loopback HTTP server thread did not stop." + ) + + if shutdown_error is None: + shutdown_error = thread_error + else: + shutdown_error.add_note(str(thread_error)) + + if shutdown_error is not None: + raise shutdown_error + + async def wait_for_requests( + self, + count: int, + *, + timeout_seconds: float = 3.0, + ) -> None: + await wait_until( + lambda: self.request_count >= count, + timeout_seconds=timeout_seconds, + ) + + def _handle_get( + self, + handler: BaseHTTPRequestHandler, + ) -> None: + parsed = urlsplit(handler.path) + query = { + key: tuple(values) + for key, values in parse_qs( + parsed.query, + keep_blank_values=True, + ).items() + } + request = LoopbackHttpRequest( + path=parsed.path, + query=query, + ) + + self._events.append("rest.request") + + with self._lock: + self._requests.append(request) + response = ( + self._responses.popleft() + if self._responses + else LoopbackHttpResponse(body=[]) + ) + + if response.release is not None: + response.release.wait() + + if isinstance(response.body, bytes): + body = response.body + elif isinstance(response.body, str): + body = response.body.encode("utf-8") + else: + body = json.dumps(response.body).encode("utf-8") + + try: + handler.send_response(response.status) + handler.send_header( + "Content-Type", + "application/json", + ) + handler.send_header( + "Content-Length", + str(len(body)), + ) + handler.send_header( + "Connection", + "close", + ) + handler.end_headers() + handler.wfile.write(body) + except (BrokenPipeError, ConnectionResetError): + pass + + +class LoopbackTcpFaultProxy: + """TCP relay с управляемой потерей client → server traffic.""" + + def __init__( + self, + *, + target_host: str, + target_port: int, + ) -> None: + self._target_host = target_host + self._target_port = target_port + self._server: asyncio.Server | None = None + self._host = "127.0.0.1" + self._port: int | None = None + self._connection_count = 0 + self._blackholed_connections: set[int] = set() + self._handler_tasks: set[asyncio.Task[Any]] = set() + self._relay_tasks: set[asyncio.Task[Any]] = set() + self._writers: set[asyncio.StreamWriter] = set() + + @property + def url(self) -> str: + if self._port is None: + raise RuntimeError("TCP fault proxy is not started.") + + return f"ws://{self._host}:{self._port}" + + @property + def connection_count(self) -> int: + return self._connection_count + + @property + def active_handler_count(self) -> int: + return sum( + not task.done() + for task in self._handler_tasks + ) + + @property + def active_relay_count(self) -> int: + return sum( + not task.done() + for task in self._relay_tasks + ) + + @property + def tracked_writer_count(self) -> int: + return len(self._writers) + + async def start(self) -> None: + if self._server is not None: + raise RuntimeError("TCP fault proxy is already started.") + + server = await asyncio.start_server( + self._handle_client, + self._host, + 0, + ) + self._server = server + sockets = server.sockets + + if not sockets: + server.close() + await server.wait_closed() + self._server = None + raise RuntimeError( + "TCP fault proxy did not expose a listening socket." + ) + + socket = sockets[0] + address = socket.getsockname() + self._port = int(address[1]) + + def blackhole_client_to_server( + self, + connection_index: int, + ) -> None: + self._blackholed_connections.add(connection_index) + + async def stop(self) -> None: + server = self._server + + if server is None: + return + + self._server = None + self._port = None + server.close() + + tasks = tuple( + task + for task in ( + *self._relay_tasks, + *self._handler_tasks, + ) + if not task.done() + ) + writers = tuple(self._writers) + + for task in tasks: + task.cancel() + + for writer in writers: + writer.close() + + async def wait_for_writer( + writer: asyncio.StreamWriter, + ) -> None: + try: + await writer.wait_closed() + except (BrokenPipeError, ConnectionResetError): + pass + + await asyncio.wait_for( + server.wait_closed(), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + + if writers: + try: + await asyncio.wait_for( + asyncio.gather( + *( + wait_for_writer(writer) + for writer in writers + ), + ), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + except TimeoutError: + for writer in writers: + writer.transport.abort() + + await asyncio.wait_for( + asyncio.gather( + *( + wait_for_writer(writer) + for writer in writers + ), + ), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + + for writer in writers: + self._writers.discard(writer) + + if tasks: + await asyncio.wait_for( + asyncio.gather( + *tasks, + return_exceptions=True, + ), + timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS, + ) + + if ( + self.active_handler_count + or self.active_relay_count + or self.tracked_writer_count + ): + raise AssertionError( + "TCP fault proxy resources did not stop." + ) + + async def _handle_client( + self, + client_reader: asyncio.StreamReader, + client_writer: asyncio.StreamWriter, + ) -> None: + task = asyncio.current_task() + + if task is None: + raise RuntimeError("TCP proxy handler has no asyncio task.") + + self._handler_tasks.add(task) + connection_index = self._connection_count + self._connection_count += 1 + + server_writer: asyncio.StreamWriter | None = None + relay_tasks: set[asyncio.Task[None]] = set() + + try: + server_reader, server_writer = await asyncio.open_connection( + self._target_host, + self._target_port, + ) + self._writers.update( + { + client_writer, + server_writer, + } + ) + + upstream = asyncio.create_task( + self._relay( + client_reader, + server_writer, + should_drop=lambda: ( + connection_index + in self._blackholed_connections + ), + ), + name="loopback-proxy-upstream", + ) + downstream = asyncio.create_task( + self._relay( + server_reader, + client_writer, + should_drop=lambda: False, + ), + name="loopback-proxy-downstream", + ) + relay_tasks = { + upstream, + downstream, + } + self._relay_tasks.update(relay_tasks) + + done, pending = await asyncio.wait( + relay_tasks, + return_when=asyncio.FIRST_COMPLETED, + ) + + for relay_task in pending: + relay_task.cancel() + + await asyncio.gather( + *done, + *pending, + return_exceptions=True, + ) + finally: + self._relay_tasks.difference_update(relay_tasks) + + for writer in ( + client_writer, + server_writer, + ): + if writer is None: + continue + + self._writers.discard(writer) + writer.close() + + try: + await writer.wait_closed() + except (BrokenPipeError, ConnectionResetError): + pass + + self._handler_tasks.discard(task) + + @staticmethod + async def _relay( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + *, + should_drop: Callable[[], bool], + ) -> None: + while True: + data = await reader.read(65_536) + + if not data: + return + + if should_drop(): + continue + + writer.write(data) + await writer.drain() + + +class LoopbackTradeEnvironment: + """Exception-safe владелец локальных сетевых ресурсов теста.""" + + def __init__( + self, + *, + websocket: LoopbackTradeWebSocketServer, + rest: LoopbackTradeRestServer, + use_fault_proxy: bool = False, + ) -> None: + self.websocket = websocket + self.rest = rest + self.use_fault_proxy = use_fault_proxy + self.proxy: LoopbackTcpFaultProxy | None = None + self._exit_stack: AsyncExitStack | None = None + + @property + def websocket_url(self) -> str: + proxy = self.proxy + + if proxy is not None: + return proxy.url + + return self.websocket.url + + async def __aenter__(self) -> LoopbackTradeEnvironment: + if self._exit_stack is not None: + raise RuntimeError( + "Loopback environment is already active." + ) + + stack = AsyncExitStack() + await stack.__aenter__() + self._exit_stack = stack + + try: + self.rest.start() + stack.push_async_callback(self._stop_rest) + + await self.websocket.start() + stack.push_async_callback(self._stop_websocket) + + if self.use_fault_proxy: + proxy = LoopbackTcpFaultProxy( + target_host="127.0.0.1", + target_port=self.websocket.port, + ) + self.proxy = proxy + await proxy.start() + stack.push_async_callback(self._stop_proxy) + except BaseException: + self._exit_stack = None + await stack.aclose() + raise + + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> bool | None: + stack = self._exit_stack + self._exit_stack = None + + if stack is None: + return None + + return await stack.__aexit__( + exc_type, + exc_value, + traceback, + ) + + async def _stop_proxy(self) -> None: + proxy = self.proxy + + if proxy is None: + return + + await asyncio.wait_for( + proxy.stop(), + timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS, + ) + + async def _stop_websocket(self) -> None: + await asyncio.wait_for( + self.websocket.stop(), + timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS, + ) + + async def _stop_rest(self) -> None: + await asyncio.wait_for( + asyncio.to_thread( + self.rest.stop, + ), + timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS, + ) diff --git a/app/tests/integration/market_data/acquisition/runtime/test_trade_stream_loopback_integration.py b/app/tests/integration/market_data/acquisition/runtime/test_trade_stream_loopback_integration.py new file mode 100644 index 0000000..432bc37 --- /dev/null +++ b/app/tests/integration/market_data/acquisition/runtime/test_trade_stream_loopback_integration.py @@ -0,0 +1,782 @@ +from __future__ import annotations + +import asyncio +import threading +import time +from typing import Any + +import pytest + +from tests.integration.market_data.acquisition.runtime import ( + loopback_trade_exchange, +) +from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import ( + LoopbackHttpResponse, + LoopbackTcpFaultProxy, + LoopbackTradeEnvironment, + LoopbackTradeRestServer, + LoopbackTradeWebSocketServer, + wait_until, +) +from src.market_data.acquisition.exceptions import ( + TradeTransportError, + WebSocketMessageDecodeError, +) +from src.market_data.acquisition.runtime.runtime_reconnect_recovery_coordinator import ( + RuntimeReconnectRecoveryCoordinator, +) +from src.market_data.acquisition.runtime.trade_stream_production_runtime import ( + TradeStreamProductionRuntime, + TradeStreamProductionRuntimeState, +) +from tests.support.trade_stream_runtime import ( + SYMBOL, + assert_no_owned_tasks, + build_runtime, + reconnect_coordinator_from, + run_scenario, + start_runtime, + state_store_from, + stop_runtime, +) + + +pytestmark = pytest.mark.integration + + +def make_recovered_trade( + *, + trade_id: int, + timestamp_ms: int, +) -> dict[str, object]: + return { + "a": trade_id, + "p": "64555.56", + "q": "0.003", + "T": timestamp_ms, + "m": False, + } + + +class HangingStartupRuntime: + def __init__(self) -> None: + self._state = TradeStreamProductionRuntimeState.STARTING + self.stop_calls = 0 + self.released = asyncio.Event() + self.finished = asyncio.Event() + + @property + def state(self) -> TradeStreamProductionRuntimeState: + return self._state + + async def run(self) -> None: + try: + await self.released.wait() + finally: + self.finished.set() + + async def stop(self) -> None: + self.stop_calls += 1 + self.released.set() + self._state = TradeStreamProductionRuntimeState.STOPPED + + +def test_start_runtime_timeout_cleans_task_it_created() -> None: + async def scenario() -> None: + runtime = HangingStartupRuntime() + + with pytest.raises(TimeoutError): + await start_runtime( + runtime, + timeout_seconds=0.01, + ) + + assert runtime.stop_calls == 1 + assert runtime.finished.is_set() + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_environment_cleans_rest_after_websocket_setup_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + + async def broken_start() -> None: + raise RuntimeError("websocket setup failed") + + monkeypatch.setattr( + websocket, + "start", + broken_start, + ) + + with pytest.raises( + RuntimeError, + match="websocket setup failed", + ): + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ): + raise AssertionError( + "environment entered after setup failure" + ) + + assert rest.thread_is_alive is False + + run_scenario(scenario()) + + +def test_environment_continues_after_websocket_cleanup_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + original_stop = websocket.stop + + async def broken_stop() -> None: + await original_stop() + raise RuntimeError("websocket cleanup failed") + + monkeypatch.setattr( + websocket, + "stop", + broken_stop, + ) + + with pytest.raises( + RuntimeError, + match="websocket cleanup failed", + ): + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ): + assert rest.thread_is_alive is True + + assert websocket.active_handler_count == 0 + assert rest.thread_is_alive is False + + run_scenario(scenario()) + + +def test_proxy_waits_for_aborted_writer_before_removing_tracking( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class RecordingServer: + def __init__(self) -> None: + self.close_calls = 0 + self.wait_closed_calls = 0 + + def close(self) -> None: + self.close_calls += 1 + + async def wait_closed(self) -> None: + self.wait_closed_calls += 1 + + class HangingWriter: + def __init__(self) -> None: + self.close_calls = 0 + self.wait_closed_calls = 0 + self.aborted = False + self.transport = self + + def close(self) -> None: + self.close_calls += 1 + + async def wait_closed(self) -> None: + self.wait_closed_calls += 1 + + if not self.aborted: + await asyncio.Event().wait() + + def abort(self) -> None: + self.aborted = True + + async def scenario() -> None: + proxy = LoopbackTcpFaultProxy( + target_host="127.0.0.1", + target_port=1, + ) + proxy_graph: Any = proxy + server = RecordingServer() + writer = HangingWriter() + proxy_graph._server = server + proxy_graph._port = 1 + proxy_graph._writers.add(writer) + + monkeypatch.setattr( + loopback_trade_exchange, + "RESOURCE_CLEANUP_TIMEOUT_SECONDS", + 0.01, + ) + + await proxy.stop() + + assert server.close_calls == 1 + assert server.wait_closed_calls == 1 + assert writer.close_calls == 1 + assert writer.wait_closed_calls == 2 + assert writer.aborted is True + assert proxy.tracked_writer_count == 0 + + run_scenario(scenario()) + + +def test_real_loopback_websocket_updates_shared_checkpoint() -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + + timestamp_ms = time.time_ns() // 1_000_000 + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=100, + timestamp_ms=timestamp_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 100 + ) + ) + + assert websocket.subscriptions[0].symbols == (SYMBOL,) + assert rest.request_count == 0 + assert runtime_task.done() is False + finally: + if runtime_task is not None: + 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] = [] + recovery_release = threading.Event() + base_time_ms = time.time_ns() // 1_000_000 - 1_000 + websocket = LoopbackTradeWebSocketServer( + events=events, + ) + rest = LoopbackTradeRestServer( + responses=( + LoopbackHttpResponse( + body=[ + make_recovered_trade( + trade_id=201, + timestamp_ms=base_time_ms + 100, + ), + ], + release=recovery_release, + ), + ), + events=events, + ) + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=200, + timestamp_ms=base_time_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 200 + ) + ) + + await websocket.abort_connection(0) + await websocket.wait_for_subscriptions(2) + await rest.wait_for_requests(1) + + await websocket.send_trade( + 1, + symbol=SYMBOL, + trade_id=202, + timestamp_ms=base_time_ms + 200, + ) + await asyncio.sleep(0) + + assert state_store.get(SYMBOL).last_trade_id == 200 + assert events.index("ws.subscribe:1") < events.index( + "rest.request" + ) + + recovery_release.set() + await wait_until( + lambda: state_store.get(SYMBOL).last_trade_id == 202, + ) + + assert runtime_task.done() is False + assert rest.requests[0].path == "/api/v1/aggTrades" + assert rest.requests[0].query["symbol"] == (SYMBOL,) + assert ( + reconnect_coordinator_from(runtime).generation + == 1 + ) + finally: + recovery_release.set() + + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_repeated_network_disconnects_advance_one_generation_each() -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + base_time_ms = time.time_ns() // 1_000_000 - 1_000 + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=300, + timestamp_ms=base_time_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 300 + ) + ) + + for generation in range(1, 4): + previous_connection = generation - 1 + + if generation == 1: + await websocket.close_connection( + previous_connection, + ) + else: + await websocket.abort_connection( + previous_connection, + ) + + await websocket.wait_for_subscriptions( + generation + 1, + ) + await rest.wait_for_requests(generation) + + trade_id = 300 + generation + await websocket.send_trade( + generation, + symbol=SYMBOL, + trade_id=trade_id, + timestamp_ms=( + base_time_ms + generation * 100 + ), + ) + await wait_until( + lambda trade_id=trade_id: ( + state_store.get(SYMBOL).last_trade_id + == trade_id + ) + ) + + assert websocket.connection_count == 4 + assert len(websocket.subscriptions) == 4 + assert rest.request_count == 3 + assert ( + reconnect_coordinator_from(runtime).generation + == 3 + ) + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_http_recovery_error_rejects_buffered_live_and_cleans_up() -> None: + async def scenario() -> None: + recovery_release = threading.Event() + base_time_ms = time.time_ns() // 1_000_000 - 1_000 + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer( + responses=( + LoopbackHttpResponse( + body={ + "error": "recovery unavailable", + }, + status=503, + release=recovery_release, + ), + ), + ) + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=400, + timestamp_ms=base_time_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 400 + ) + ) + + await websocket.abort_connection(0) + await websocket.wait_for_subscriptions(2) + await rest.wait_for_requests(1) + await websocket.send_trade( + 1, + symbol=SYMBOL, + trade_id=402, + timestamp_ms=base_time_ms + 200, + ) + recovery_release.set() + + with pytest.raises( + TradeTransportError, + match="Не удалось получить агрегированные сделки", + ): + await asyncio.wait_for( + runtime_task, + timeout=3, + ) + + assert state_store.get(SYMBOL).last_trade_id == 400 + assert ( + reconnect_coordinator_from(runtime) + .live_processing_gate + .failed + ) + assert ( + runtime.state + is TradeStreamProductionRuntimeState.FAILED + ) + finally: + recovery_release.set() + + if ( + runtime_task is not None + and not runtime_task.done() + ): + await stop_runtime(runtime, runtime_task) + + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_invalid_websocket_json_is_terminal_and_releases_resources() -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_raw(0, "{not-json") + + with pytest.raises( + WebSocketMessageDecodeError, + ): + await asyncio.wait_for( + runtime_task, + timeout=3, + ) + + assert ( + runtime.state + is TradeStreamProductionRuntimeState.FAILED + ) + assert rest.request_count == 0 + finally: + if ( + runtime_task is not None + and not runtime_task.done() + ): + await stop_runtime(runtime, runtime_task) + + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_stop_waits_for_real_http_recovery_worker() -> None: + async def scenario() -> None: + recovery_release = threading.Event() + base_time_ms = time.time_ns() // 1_000_000 - 1_000 + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer( + responses=( + LoopbackHttpResponse( + body=[ + make_recovered_trade( + trade_id=501, + timestamp_ms=base_time_ms + 100, + ), + ], + release=recovery_release, + ), + ), + ) + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=500, + timestamp_ms=base_time_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 500 + ) + ) + + await websocket.abort_connection(0) + await websocket.wait_for_subscriptions(2) + 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 + + recovery_release.set() + await asyncio.wait_for( + stop_task, + timeout=3, + ) + await asyncio.wait_for( + runtime_task, + timeout=3, + ) + + assert state_store.get(SYMBOL).last_trade_id == 501 + assert ( + runtime.state + is TradeStreamProductionRuntimeState.STOPPED + ) + finally: + recovery_release.set() + + if ( + runtime_task is not None + and not runtime_task.done() + ): + await stop_runtime(runtime, runtime_task) + + await assert_no_owned_tasks() + + run_scenario(scenario()) + + +def test_ping_timeout_and_receive_failure_share_one_reconnect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + caller_task_names: list[str] = [] + original_reconnect = ( + RuntimeReconnectRecoveryCoordinator + .reconnect_after_transport_failure + ) + + async def recording_reconnect( + self: RuntimeReconnectRecoveryCoordinator, + *, + observed_generation: int, + ) -> None: + task = asyncio.current_task() + caller_task_names.append( + task.get_name() + if task is not None + else "" + ) + await original_reconnect( + self, + observed_generation=observed_generation, + ) + + monkeypatch.setattr( + RuntimeReconnectRecoveryCoordinator, + "reconnect_after_transport_failure", + recording_reconnect, + ) + + async def scenario() -> None: + base_time_ms = time.time_ns() // 1_000_000 - 1_000 + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + use_fault_proxy=True, + ) as environment: + proxy = environment.proxy + assert proxy is not None + + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + probe_timeout_seconds=0.05, + close_timeout_seconds=0.05, + heartbeat_timeout_seconds=0.15, + scheduler_interval_seconds=0.05, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime(runtime) + await websocket.wait_for_subscriptions(1) + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=600, + timestamp_ms=base_time_ms, + ) + + state_store = state_store_from(runtime) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id == 600 + ) + ) + + proxy.blackhole_client_to_server(0) + + await websocket.wait_for_subscriptions(2) + await rest.wait_for_requests(1) + await websocket.send_trade( + 1, + symbol=SYMBOL, + trade_id=601, + timestamp_ms=base_time_ms + 100, + ) + await wait_until( + lambda: state_store.get(SYMBOL).last_trade_id == 601, + ) + + assert proxy.connection_count == 2 + assert websocket.connection_count == 2 + assert len(websocket.subscriptions) == 2 + assert rest.request_count == 1 + assert ( + reconnect_coordinator_from(runtime).generation + == 1 + ) + assert sorted(caller_task_names) == [ + "trade-stream-receive", + "trade-stream-scheduler", + ] + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + assert proxy.active_handler_count == 0 + assert proxy.active_relay_count == 0 + await assert_no_owned_tasks() + + run_scenario(scenario()) diff --git a/app/tests/live/market_data/acquisition/runtime/test_trade_stream_runtime_live.py b/app/tests/live/market_data/acquisition/runtime/test_trade_stream_runtime_live.py new file mode 100644 index 0000000..20db185 --- /dev/null +++ b/app/tests/live/market_data/acquisition/runtime/test_trade_stream_runtime_live.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from tests.support.live_trade_stream import ( + LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS, + LiveTestConfigurationError, + LiveTradeStreamTestConfig, + build_live_trade_stream_settings, + load_live_trade_stream_test_config, +) + + +pytestmark = pytest.mark.live + +RUNNING_TASK_NAMES = ( + "trade-stream-receive", + "trade-stream-runtime", + "trade-stream-scheduler", +) + + +def require_live_config() -> LiveTradeStreamTestConfig: + try: + config = load_live_trade_stream_test_config() + except LiveTestConfigurationError as error: + pytest.fail( + f"Invalid opt-in live configuration: {error}", + pytrace=False, + ) + + if config is None: + pytest.skip( + "Live verification requires DZENTRA_RUN_LIVE_TESTS=1." + ) + + return config + + +def test_live_runtime_reconnects_recovers_and_resumes_trades() -> None: + config = require_live_config() + + from src.bootstrap.trade_stream_runtime import ( + build_trade_stream_production_runtime, + ) + from src.market_data.acquisition.runtime.trade_stream_production_runtime import ( + TradeStreamProductionRuntimeState, + ) + from src.market_data.acquisition.trade_id_sequence import ( + is_trade_id_newer, + is_trade_id_same_or_newer, + ) + from tests.support.trade_stream_runtime import ( + active_owned_task_names, + assert_no_owned_tasks, + reconnect_coordinator_from, + run_scenario, + start_runtime, + state_store_from, + stop_runtime, + wait_until_or_runtime_exit, + ) + + settings = build_live_trade_stream_settings(config) + runtime = build_trade_stream_production_runtime(settings) + + assert runtime is not None + + async def scenario() -> None: + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime( + runtime, + timeout_seconds=config.trade_timeout_seconds, + ) + state_store = state_store_from(runtime) + + await wait_until_or_runtime_exit( + lambda: ( + state_store.contains(config.symbol) + and state_store.get( + config.symbol + ).last_trade_id + is not None + ), + runtime_task=runtime_task, + timeout_seconds=config.trade_timeout_seconds, + ) + + first_state = state_store.get(config.symbol) + first_trade_id = first_state.last_trade_id + + assert first_trade_id is not None + assert first_state.last_trade is not None + assert first_state.last_trade.symbol == config.symbol + assert runtime.state is ( + TradeStreamProductionRuntimeState.RUNNING + ) + assert runtime_task.done() is False + + coordinator: Any = reconnect_coordinator_from(runtime) + source_generation = coordinator.generation + + await asyncio.wait_for( + coordinator.reconnect(), + timeout=config.trade_timeout_seconds, + ) + + assert coordinator.generation == source_generation + 1 + assert coordinator.live_processing_gate.locked is False + assert coordinator.live_processing_gate.failed is False + assert runtime.state is ( + TradeStreamProductionRuntimeState.RUNNING + ) + assert runtime_task.done() is False + + checkpoint_after_recovery = state_store.get( + config.symbol + ).last_trade_id + + assert checkpoint_after_recovery is not None + assert is_trade_id_same_or_newer( + checkpoint_after_recovery, + first_trade_id, + ) + + def checkpoint_advanced_after_recovery() -> bool: + last_trade_id = state_store.get( + config.symbol + ).last_trade_id + + return ( + last_trade_id is not None + and is_trade_id_newer( + last_trade_id, + checkpoint_after_recovery, + ) + ) + + await wait_until_or_runtime_exit( + checkpoint_advanced_after_recovery, + runtime_task=runtime_task, + timeout_seconds=config.trade_timeout_seconds, + ) + + final_state = state_store.get(config.symbol) + + assert final_state.last_trade_id is not None + assert is_trade_id_newer( + final_state.last_trade_id, + checkpoint_after_recovery, + ) + assert final_state.last_trade is not None + assert final_state.last_trade.symbol == config.symbol + assert active_owned_task_names() == RUNNING_TASK_NAMES + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime( + runtime, + runtime_task, + timeout_seconds=( + LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS + ), + ) + + await assert_no_owned_tasks() + + run_scenario( + scenario(), + timeout_seconds=config.scenario_timeout_seconds, + ) diff --git a/app/tests/stress/market_data/acquisition/runtime/test_trade_stream_runtime_stress.py b/app/tests/stress/market_data/acquisition/runtime/test_trade_stream_runtime_stress.py new file mode 100644 index 0000000..b6a3bba --- /dev/null +++ b/app/tests/stress/market_data/acquisition/runtime/test_trade_stream_runtime_stress.py @@ -0,0 +1,589 @@ +from __future__ import annotations + +import asyncio +import gc +import os +import time +import tracemalloc +from typing import Any + +import pytest + +from src.market_data.acquisition.consistency.trade_stream_state import ( + DEFAULT_DEDUPLICATION_WINDOW_SIZE, + TradeStreamState, +) +from src.market_data.acquisition.runtime.trade_stream_production_runtime import ( + TradeStreamProductionRuntime, + TradeStreamProductionRuntimeState, +) +from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import ( + LoopbackTcpFaultProxy, + LoopbackTradeEnvironment, + LoopbackTradeRestServer, + LoopbackTradeWebSocketServer, + wait_until, +) +from tests.support.trade_stream_runtime import ( + SYMBOL, + active_owned_task_names, + assert_no_owned_tasks, + build_runtime, + reconnect_coordinator_from, + run_scenario, + start_runtime, + state_store_from, + stop_runtime, +) + + +pytestmark = pytest.mark.stress + +TRADE_BURST_COUNT = 20_000 +TRADE_BURST_SIZE = 250 +TRADE_BURST_TIMEOUT_SECONDS = 90.0 +MEMORY_WARMUP_TRADE_COUNT = 12_000 +MEMORY_SAMPLE_TRADE_INTERVAL = 2_000 +MAX_RETAINED_MEMORY_GROWTH_BYTES = 4 * 1024 * 1024 + +RECONNECT_GENERATION_COUNT = 30 +RECONNECT_STORM_TIMEOUT_SECONDS = 90.0 + +DEFAULT_SOAK_SECONDS = 120 +SOAK_TRADES_PER_SECOND = 50 +SOAK_RECONNECT_INTERVAL_SECONDS = 5 +SOAK_CLEANUP_ALLOWANCE_SECONDS = 60.0 +STRESS_CONDITION_TIMEOUT_SECONDS = 10.0 + +RUNNING_TASK_NAMES = ( + "trade-stream-receive", + "trade-stream-runtime", + "trade-stream-scheduler", +) + + +def read_soak_seconds() -> int: + raw_value = os.environ.get( + "DZENTRA_SOAK_SECONDS", + str(DEFAULT_SOAK_SECONDS), + ) + + try: + duration = int(raw_value) + except ValueError as error: + raise ValueError( + "DZENTRA_SOAK_SECONDS must be an integer." + ) from error + + if duration <= 0: + raise ValueError( + "DZENTRA_SOAK_SECONDS must be positive." + ) + + return duration + + +def acquisition_memory_bytes() -> int: + snapshot = tracemalloc.take_snapshot().filter_traces( + ( + tracemalloc.Filter( + True, + "*/src/market_data/acquisition/*", + ), + ) + ) + return sum( + statistic.size + for statistic in snapshot.statistics("filename") + ) + + +def assert_runtime_is_settled( + runtime: TradeStreamProductionRuntime, + *, + expected_generation: int, +) -> None: + coordinator: Any = reconnect_coordinator_from(runtime) + + assert runtime.state is TradeStreamProductionRuntimeState.RUNNING + assert coordinator.generation == expected_generation + assert coordinator.live_processing_gate.locked is False + assert coordinator.live_processing_gate.failed is False + assert coordinator._recovery_task is None + assert active_owned_task_names() == RUNNING_TASK_NAMES + + +async def send_trade_batch( + websocket: LoopbackTradeWebSocketServer, + *, + connection_index: int, + first_trade_id: int, + count: int, + first_timestamp_ms: int, +) -> int: + for offset in range(count): + await websocket.send_trade( + connection_index, + symbol=SYMBOL, + trade_id=first_trade_id + offset, + timestamp_ms=first_timestamp_ms + offset, + ) + + return first_trade_id + count - 1 + + +def test_memory_filter_observes_acquisition_allocations() -> None: + tracing_was_active = tracemalloc.is_tracing() + + if not tracing_was_active: + tracemalloc.start() + + states: tuple[TradeStreamState, ...] = () + + try: + states = tuple( + TradeStreamState( + symbol=f"MEMORY_TEST_{index}", + ) + for index in range(100) + ) + gc.collect() + baseline = acquisition_memory_bytes() + + assert baseline > 0, ( + "Acquisition tracemalloc filter returned an empty baseline." + ) + finally: + del states + gc.collect() + + if not tracing_was_active: + tracemalloc.stop() + + +def test_trade_burst_remains_ordered_and_memory_bounded() -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + memory_samples: list[int] = [] + first_trade_id = 1_000_000 + first_timestamp_ms = time.time_ns() // 1_000_000 + + tracing_was_active = tracemalloc.is_tracing() + + if not tracing_was_active: + tracemalloc.start() + + try: + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime( + runtime, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + await websocket.wait_for_subscriptions( + 1, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + state_store = state_store_from(runtime) + + for sent_count in range( + TRADE_BURST_SIZE, + TRADE_BURST_COUNT + 1, + TRADE_BURST_SIZE, + ): + batch_first_id = ( + first_trade_id + + sent_count + - TRADE_BURST_SIZE + ) + last_trade_id = await send_trade_batch( + websocket, + connection_index=0, + first_trade_id=batch_first_id, + count=TRADE_BURST_SIZE, + first_timestamp_ms=( + first_timestamp_ms + + sent_count + - TRADE_BURST_SIZE + ), + ) + await wait_until( + lambda last_trade_id=last_trade_id: ( + state_store.contains(SYMBOL) + and state_store.get( + SYMBOL + ).last_trade_id + == last_trade_id + ), + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + + if ( + sent_count >= MEMORY_WARMUP_TRADE_COUNT + and ( + sent_count + - MEMORY_WARMUP_TRADE_COUNT + ) + % MEMORY_SAMPLE_TRADE_INTERVAL + == 0 + ): + gc.collect() + memory_samples.append( + acquisition_memory_bytes() + ) + + state: Any = state_store.get(SYMBOL) + expected_last_trade_id = ( + first_trade_id + TRADE_BURST_COUNT - 1 + ) + + assert state.last_trade_id == expected_last_trade_id + assert len(state._trade_window) == ( + DEFAULT_DEDUPLICATION_WINDOW_SIZE + ) + assert len(state._trades) == ( + DEFAULT_DEDUPLICATION_WINDOW_SIZE + ) + assert len(memory_samples) == 5 + assert memory_samples[0] > 0, ( + "Acquisition tracemalloc filter returned an " + f"empty baseline: samples={memory_samples!r}" + ) + retained_memory_growth = ( + memory_samples[-1] - memory_samples[0] + ) + assert ( + retained_memory_growth + <= MAX_RETAINED_MEMORY_GROWTH_BYTES + ), ( + "Acquisition retained memory exceeded the " + "post-warmup limit: " + f"growth={retained_memory_growth} " + f"samples={memory_samples!r}" + ) + assert websocket.connection_count == 1 + assert websocket.active_connection_count == 1 + assert len(websocket.subscriptions) == 1 + assert rest.request_count == 0 + assert_runtime_is_settled( + runtime, + expected_generation=0, + ) + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + assert websocket.active_connection_count == 0 + assert websocket.active_handler_count == 0 + assert rest.thread_is_alive is False + await assert_no_owned_tasks() + finally: + if not tracing_was_active: + tracemalloc.stop() + + run_scenario( + scenario(), + timeout_seconds=TRADE_BURST_TIMEOUT_SECONDS, + ) + + +def test_reconnect_storm_advances_exactly_one_generation_per_fault() -> None: + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + first_trade_id = 2_000_000 + first_timestamp_ms = time.time_ns() // 1_000_000 + proxy: LoopbackTcpFaultProxy | None = None + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + use_fault_proxy=True, + ) as environment: + proxy = environment.proxy + assert proxy is not None + + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + probe_timeout_seconds=0.05, + close_timeout_seconds=0.05, + heartbeat_timeout_seconds=0.15, + scheduler_interval_seconds=0.05, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime( + runtime, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + await websocket.wait_for_subscriptions( + 1, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + state_store = state_store_from(runtime) + + await websocket.send_trade( + 0, + symbol=SYMBOL, + trade_id=first_trade_id, + timestamp_ms=first_timestamp_ms, + ) + await wait_until( + lambda: ( + state_store.contains(SYMBOL) + and state_store.get(SYMBOL).last_trade_id + == first_trade_id + ), + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + + for generation in range( + 1, + RECONNECT_GENERATION_COUNT + 1, + ): + previous_connection = generation - 1 + fault_kind = generation % 3 + + if fault_kind == 1: + await websocket.close_connection( + previous_connection, + ) + elif fault_kind == 2: + await websocket.abort_connection( + previous_connection, + ) + else: + proxy.blackhole_client_to_server( + previous_connection, + ) + + await websocket.wait_for_subscriptions( + generation + 1, + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + await rest.wait_for_requests( + generation, + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + + trade_id = first_trade_id + generation + await websocket.send_trade( + generation, + symbol=SYMBOL, + trade_id=trade_id, + timestamp_ms=( + first_timestamp_ms + generation + ), + ) + await wait_until( + lambda trade_id=trade_id: ( + state_store.get(SYMBOL).last_trade_id + == trade_id + ), + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + assert_runtime_is_settled( + runtime, + expected_generation=generation, + ) + + assert proxy.connection_count == ( + RECONNECT_GENERATION_COUNT + 1 + ) + assert websocket.connection_count == ( + RECONNECT_GENERATION_COUNT + 1 + ) + assert websocket.active_connection_count == 1 + assert len(websocket.subscriptions) == ( + RECONNECT_GENERATION_COUNT + 1 + ) + assert rest.request_count == RECONNECT_GENERATION_COUNT + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + assert proxy is not None + assert proxy.active_handler_count == 0 + assert proxy.active_relay_count == 0 + assert proxy.tracked_writer_count == 0 + assert websocket.active_connection_count == 0 + assert websocket.active_handler_count == 0 + assert rest.thread_is_alive is False + await assert_no_owned_tasks() + + run_scenario( + scenario(), + timeout_seconds=RECONNECT_STORM_TIMEOUT_SECONDS, + ) + + +def test_runtime_remains_stable_during_local_soak() -> None: + soak_seconds = read_soak_seconds() + expected_reconnects = ( + soak_seconds // SOAK_RECONNECT_INTERVAL_SECONDS + ) + expected_trade_count = soak_seconds * SOAK_TRADES_PER_SECOND + + async def scenario() -> None: + websocket = LoopbackTradeWebSocketServer() + rest = LoopbackTradeRestServer() + first_trade_id = 3_000_000 + first_timestamp_ms = time.time_ns() // 1_000_000 + + async with LoopbackTradeEnvironment( + websocket=websocket, + rest=rest, + ) as environment: + runtime = build_runtime( + websocket_url=environment.websocket_url, + rest_base_url=rest.base_url, + ) + runtime_task: asyncio.Task[None] | None = None + + try: + runtime_task = await start_runtime( + runtime, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + await websocket.wait_for_subscriptions( + 1, + timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS, + ) + state_store = state_store_from(runtime) + loop = asyncio.get_running_loop() + started_at = loop.time() + connection_index = 0 + next_trade_id = first_trade_id + + for second in range(1, soak_seconds + 1): + if ( + second + % SOAK_RECONNECT_INTERVAL_SECONDS + == 0 + ): + if connection_index % 2 == 0: + await websocket.close_connection( + connection_index, + ) + else: + await websocket.abort_connection( + connection_index, + ) + + connection_index += 1 + await websocket.wait_for_subscriptions( + connection_index + 1, + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + await rest.wait_for_requests( + connection_index, + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + + last_trade_id = await send_trade_batch( + websocket, + connection_index=connection_index, + first_trade_id=next_trade_id, + count=SOAK_TRADES_PER_SECOND, + first_timestamp_ms=( + first_timestamp_ms + + next_trade_id + - first_trade_id + ), + ) + next_trade_id = last_trade_id + 1 + + await wait_until( + lambda last_trade_id=last_trade_id: ( + state_store.contains(SYMBOL) + and state_store.get( + SYMBOL + ).last_trade_id + == last_trade_id + ), + timeout_seconds=( + STRESS_CONDITION_TIMEOUT_SECONDS + ), + ) + + if ( + second + % SOAK_RECONNECT_INTERVAL_SECONDS + == 0 + ): + assert_runtime_is_settled( + runtime, + expected_generation=connection_index, + ) + + remaining = started_at + second - loop.time() + + if remaining > 0: + await asyncio.sleep(remaining) + + expected_last_trade_id = ( + first_trade_id + expected_trade_count - 1 + ) + state: Any = state_store.get(SYMBOL) + + assert state.last_trade_id == expected_last_trade_id + assert len(state._trade_window) <= ( + DEFAULT_DEDUPLICATION_WINDOW_SIZE + ) + assert len(state._trades) <= ( + DEFAULT_DEDUPLICATION_WINDOW_SIZE + ) + assert websocket.connection_count == ( + expected_reconnects + 1 + ) + assert websocket.active_connection_count == 1 + assert len(websocket.subscriptions) == ( + expected_reconnects + 1 + ) + assert rest.request_count == expected_reconnects + assert_runtime_is_settled( + runtime, + expected_generation=expected_reconnects, + ) + assert runtime_task.done() is False + finally: + if runtime_task is not None: + await stop_runtime(runtime, runtime_task) + + assert websocket.active_connection_count == 0 + assert websocket.active_handler_count == 0 + assert rest.thread_is_alive is False + await assert_no_owned_tasks() + + run_scenario( + scenario(), + timeout_seconds=( + soak_seconds + SOAK_CLEANUP_ALLOWANCE_SECONDS + ), + ) diff --git a/app/tests/support/__init__.py b/app/tests/support/__init__.py new file mode 100644 index 0000000..a14a819 --- /dev/null +++ b/app/tests/support/__init__.py @@ -0,0 +1 @@ +"""Общие вспомогательные компоненты тестов Dzentra.""" diff --git a/app/tests/support/async_wait.py b/app/tests/support/async_wait.py new file mode 100644 index 0000000..7f606ea --- /dev/null +++ b/app/tests/support/async_wait.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Callable + + +Predicate = Callable[[], bool] + + +async def wait_until( + predicate: Predicate, + *, + timeout_seconds: float = 3.0, +) -> None: + """Дождаться наблюдаемого условия без фиксированной длинной паузы.""" + async with asyncio.timeout(timeout_seconds): + while not predicate(): + await asyncio.sleep(0.005) diff --git a/app/tests/support/live_trade_stream.py b/app/tests/support/live_trade_stream.py new file mode 100644 index 0000000..b2a1205 --- /dev/null +++ b/app/tests/support/live_trade_stream.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import math +import os +from collections.abc import Mapping +from dataclasses import dataclass +from typing import TYPE_CHECKING +from urllib.parse import urlsplit + + +if TYPE_CHECKING: + from src.core.config import Settings + + +RUN_LIVE_TESTS_ENV = "DZENTRA_RUN_LIVE_TESTS" +LIVE_REST_URL_ENV = "DZENTRA_LIVE_REST_URL" +LIVE_WEBSOCKET_URL_ENV = "DZENTRA_LIVE_WS_URL" +LIVE_SYMBOLS_ENV = "DZENTRA_LIVE_SYMBOLS" +LIVE_TRADE_TIMEOUT_ENV = "DZENTRA_LIVE_TRADE_TIMEOUT_SECONDS" + +DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS = 600.0 +LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS = 30.0 + + +class LiveTestConfigurationError(ValueError): + """Ошибка явной конфигурации opt-in live verification.""" + + +@dataclass(frozen=True, slots=True) +class LiveTradeStreamTestConfig: + rest_url: str + websocket_url: str + symbol: str + trade_timeout_seconds: float + + @property + def scenario_timeout_seconds(self) -> float: + return ( + self.trade_timeout_seconds * 2 + + LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS * 2 + ) + + +def load_live_trade_stream_test_config( + environment: Mapping[str, str] | None = None, +) -> LiveTradeStreamTestConfig | None: + values = environment if environment is not None else os.environ + opt_in = values.get(RUN_LIVE_TESTS_ENV, "").strip() + + if opt_in in {"", "0"}: + return None + + if opt_in != "1": + raise LiveTestConfigurationError( + f"{RUN_LIVE_TESTS_ENV} must be exactly 1 when enabled." + ) + + rest_url = _require_environment_value( + values, + LIVE_REST_URL_ENV, + ).rstrip("/") + websocket_url = _require_environment_value( + values, + LIVE_WEBSOCKET_URL_ENV, + ) + symbols_value = _require_environment_value( + values, + LIVE_SYMBOLS_ENV, + ) + symbols = tuple( + symbol.strip() + for symbol in symbols_value.split(",") + ) + + if any(not symbol for symbol in symbols): + raise LiveTestConfigurationError( + f"{LIVE_SYMBOLS_ENV} must not contain empty symbols." + ) + + if len(symbols) != 1: + raise LiveTestConfigurationError( + f"{LIVE_SYMBOLS_ENV} must contain exactly one symbol " + "for bounded live verification." + ) + + _validate_secure_url( + rest_url, + name=LIVE_REST_URL_ENV, + expected_scheme="https", + required_path=None, + ) + _validate_secure_url( + websocket_url, + name=LIVE_WEBSOCKET_URL_ENV, + expected_scheme="wss", + required_path="/connect", + ) + + timeout_seconds = _parse_positive_timeout( + values.get( + LIVE_TRADE_TIMEOUT_ENV, + str(DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS), + ), + ) + + return LiveTradeStreamTestConfig( + rest_url=rest_url, + websocket_url=websocket_url, + symbol=symbols[0], + trade_timeout_seconds=timeout_seconds, + ) + + +def build_live_trade_stream_settings( + config: LiveTradeStreamTestConfig, +) -> Settings: + from src.core.config import Settings, TradeStreamSettings + + return Settings( + bot_token="live-verification-does-not-use-telegram", + bot_parse_mode="HTML", + app_env="live-verification", + log_level="INFO", + tz="UTC", + exchange_enabled=True, + exchange_name="dzengi", + exchange_base_url=config.rest_url, + exchange_ws_url="", + exchange_api_key="", + exchange_api_secret="", + exchange_timeout_sec=20, + exchange_testnet=False, + default_symbol=config.symbol, + trade_stream=TradeStreamSettings( + enabled=True, + websocket_url=config.websocket_url, + symbols=(config.symbol,), + open_timeout_seconds=10.0, + probe_timeout_seconds=20.0, + close_timeout_seconds=10.0, + heartbeat_timeout_seconds=20.0, + scheduler_interval_seconds=5.0, + recovery_window_ms=3_599_999, + ), + db_host="localhost", + db_port=5432, + db_name="live-verification", + db_user="live-verification", + db_password="", + debug_enabled=False, + journal_debug_enabled=False, + ) + + +def _require_environment_value( + environment: Mapping[str, str], + name: str, +) -> str: + value = environment.get(name, "").strip() + + if not value: + raise LiveTestConfigurationError( + f"{name} is required when {RUN_LIVE_TESTS_ENV}=1." + ) + + return value + + +def _validate_secure_url( + raw_url: str, + *, + name: str, + expected_scheme: str, + required_path: str | None, +) -> None: + parsed = urlsplit(raw_url) + + if parsed.scheme.lower() != expected_scheme or not parsed.netloc: + raise LiveTestConfigurationError( + f"{name} must be an absolute {expected_scheme} URL." + ) + + if parsed.username is not None or parsed.password is not None: + raise LiveTestConfigurationError( + f"{name} must not contain embedded credentials." + ) + + if parsed.query or parsed.fragment: + raise LiveTestConfigurationError( + f"{name} must not contain a query or fragment." + ) + + if required_path is None: + if parsed.path not in {"", "/"}: + raise LiveTestConfigurationError( + f"{name} must be a base URL without an endpoint path." + ) + return + + if parsed.path.rstrip("/") != required_path: + raise LiveTestConfigurationError( + f"{name} must end with {required_path}." + ) + + +def _parse_positive_timeout(raw_value: str) -> float: + try: + timeout_seconds = float(raw_value.strip()) + except ValueError as error: + raise LiveTestConfigurationError( + f"{LIVE_TRADE_TIMEOUT_ENV} must be a number." + ) from error + + if ( + not math.isfinite(timeout_seconds) + or timeout_seconds <= 0 + ): + raise LiveTestConfigurationError( + f"{LIVE_TRADE_TIMEOUT_ENV} must be positive and finite." + ) + + return timeout_seconds diff --git a/app/tests/support/trade_stream_runtime.py b/app/tests/support/trade_stream_runtime.py new file mode 100644 index 0000000..e0c6dcc --- /dev/null +++ b/app/tests/support/trade_stream_runtime.py @@ -0,0 +1,314 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any, Protocol + +from src.bootstrap.trade_stream_runtime import ( + build_trade_stream_production_runtime, +) +from src.core.config import Settings, TradeStreamSettings +from src.market_data.acquisition.consistency.trade_stream_state_store import ( + TradeStreamStateStore, +) +from src.market_data.acquisition.runtime.trade_stream_production_runtime import ( + TradeStreamProductionRuntime, + TradeStreamProductionRuntimeState, +) +from tests.support.async_wait import wait_until + + +SYMBOL = "BTC/USD_LEVERAGE" +RUNTIME_CLEANUP_TIMEOUT_SECONDS = 5.0 + +OWNED_TASK_NAMES = frozenset( + { + "trade-stream-receive", + "trade-stream-runtime", + "trade-stream-runtime-recovery", + "trade-stream-scheduler", + "trade-stream-startup", + } +) + + +class ManagedRuntimeProtocol(Protocol): + @property + def state(self) -> TradeStreamProductionRuntimeState: + ... + + async def run(self) -> None: + ... + + async def stop(self) -> None: + ... + + +def run_scenario( + scenario: Awaitable[None], + *, + timeout_seconds: float = 30.0, +) -> None: + asyncio.run( + asyncio.wait_for( + scenario, + timeout=timeout_seconds, + ) + ) + + +def make_settings( + *, + websocket_url: str, + rest_base_url: str, + open_timeout_seconds: float = 0.5, + probe_timeout_seconds: float = 0.2, + close_timeout_seconds: float = 0.2, + heartbeat_timeout_seconds: float = 60.0, + scheduler_interval_seconds: float = 60.0, +) -> Settings: + return Settings( + bot_token="integration-test-token", + bot_parse_mode="HTML", + app_env="test", + log_level="INFO", + tz="UTC", + exchange_enabled=True, + exchange_name="dzengi", + exchange_base_url=rest_base_url, + exchange_ws_url="", + exchange_api_key="", + exchange_api_secret="", + exchange_timeout_sec=2, + exchange_testnet=True, + default_symbol=SYMBOL, + trade_stream=TradeStreamSettings( + enabled=True, + websocket_url=websocket_url, + symbols=(SYMBOL,), + open_timeout_seconds=open_timeout_seconds, + probe_timeout_seconds=probe_timeout_seconds, + close_timeout_seconds=close_timeout_seconds, + heartbeat_timeout_seconds=heartbeat_timeout_seconds, + scheduler_interval_seconds=scheduler_interval_seconds, + recovery_window_ms=3_599_999, + ), + db_host="localhost", + db_port=5432, + db_name="integration", + db_user="integration", + db_password="", + debug_enabled=False, + journal_debug_enabled=False, + ) + + +def build_runtime( + *, + websocket_url: str, + rest_base_url: str, + probe_timeout_seconds: float = 0.2, + close_timeout_seconds: float = 0.2, + heartbeat_timeout_seconds: float = 60.0, + scheduler_interval_seconds: float = 60.0, +) -> TradeStreamProductionRuntime: + runtime = build_trade_stream_production_runtime( + make_settings( + websocket_url=websocket_url, + rest_base_url=rest_base_url, + probe_timeout_seconds=probe_timeout_seconds, + close_timeout_seconds=close_timeout_seconds, + heartbeat_timeout_seconds=heartbeat_timeout_seconds, + scheduler_interval_seconds=scheduler_interval_seconds, + ) + ) + + assert runtime is not None + return runtime + + +def state_store_from( + runtime: TradeStreamProductionRuntime, +) -> TradeStreamStateStore: + runtime_graph: Any = runtime + + return ( + runtime_graph + ._reconnect_recovery_coordinator + ._recovery_coordinator + ._state_store + ) + + +def reconnect_coordinator_from( + runtime: TradeStreamProductionRuntime, +) -> Any: + runtime_graph: Any = runtime + return runtime_graph._reconnect_recovery_coordinator + + +def active_owned_task_names() -> tuple[str, ...]: + current_task = asyncio.current_task() + + return tuple( + sorted( + task.get_name() + for task in asyncio.all_tasks() + if task is not current_task + and not task.done() + and task.get_name() in OWNED_TASK_NAMES + ) + ) + + +async def start_runtime( + runtime: ManagedRuntimeProtocol, + *, + timeout_seconds: float = 3.0, +) -> asyncio.Task[None]: + task = asyncio.create_task( + runtime.run(), + name="trade-stream-runtime", + ) + + try: + await wait_until( + lambda: runtime.state + in { + TradeStreamProductionRuntimeState.RUNNING, + TradeStreamProductionRuntimeState.FAILED, + }, + timeout_seconds=timeout_seconds, + ) + + if ( + runtime.state + is TradeStreamProductionRuntimeState.FAILED + ): + await task + except BaseException as error: + try: + await stop_runtime(runtime, task) + except BaseException as cleanup_error: + error.add_note( + "Runtime startup cleanup also failed: " + f"{type(cleanup_error).__name__}." + ) + + raise + + return task + + +async def wait_until_or_runtime_exit( + predicate: Callable[[], bool], + *, + runtime_task: asyncio.Task[None], + timeout_seconds: float, +) -> None: + """ + Дождаться условия, немедленно распространяя завершение Runtime. + + Runtime task не отменяется: helper владеет только внутренней задачей + ожидания условия. Если Runtime завершился без ошибки до выполнения + условия, это считается ошибкой проверочного сценария. + """ + condition_task = asyncio.create_task( + wait_until( + predicate, + timeout_seconds=timeout_seconds, + ) + ) + + try: + await asyncio.wait( + ( + condition_task, + runtime_task, + ), + return_when=asyncio.FIRST_COMPLETED, + ) + + if runtime_task.done(): + condition_task.cancel() + + try: + await condition_task + except asyncio.CancelledError: + pass + + await runtime_task + + raise RuntimeError( + "Trade Stream Runtime exited before the expected " + "live condition was reached." + ) + + await condition_task + finally: + if not condition_task.done(): + condition_task.cancel() + + try: + await condition_task + except asyncio.CancelledError: + pass + + +async def stop_runtime( + runtime: ManagedRuntimeProtocol, + task: asyncio.Task[None], + *, + timeout_seconds: float = RUNTIME_CLEANUP_TIMEOUT_SECONDS, +) -> None: + primary_error: BaseException | None = None + + if not task.done(): + try: + await asyncio.wait_for( + runtime.stop(), + timeout=timeout_seconds, + ) + except BaseException as error: + primary_error = error + + if not task.done() and primary_error is not None: + task.cancel() + + try: + await asyncio.wait_for( + asyncio.shield(task), + timeout=timeout_seconds, + ) + except BaseException as error: + if isinstance(error, TimeoutError): + task.cancel() + + try: + await asyncio.wait_for( + task, + timeout=timeout_seconds, + ) + except asyncio.CancelledError: + pass + except BaseException as cancellation_error: + error.add_note( + "Runtime task cancellation also failed: " + f"{type(cancellation_error).__name__}." + ) + + if primary_error is None: + primary_error = error + else: + primary_error.add_note( + "Runtime task cleanup also failed: " + f"{type(error).__name__}." + ) + + if primary_error is not None: + raise primary_error + + +async def assert_no_owned_tasks() -> None: + await asyncio.sleep(0) + assert active_owned_task_names() == () diff --git a/app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py b/app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py index 4f52b67..66dd6df 100644 --- a/app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py +++ b/app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py @@ -73,6 +73,25 @@ def test_accepts_first_trade() -> None: assert state.last_trade is trade +@pytest.mark.parametrize( + "trade_id", + ( + -(2**31) - 1, + 2**31, + ), +) +def test_rejects_out_of_range_first_trade_id( + trade_id: int, +) -> None: + state = TradeStreamState(symbol="BTCUSD") + + with pytest.raises(ValueError, match="signed 32-bit"): + state.accept(_trade(trade_id=trade_id)) + + assert state.last_trade_id is None + assert state.last_trade is None + + def test_first_accepted_trade_becomes_checkpoint() -> None: state = TradeStreamState(symbol="BTCUSD") trade = _trade() @@ -121,6 +140,49 @@ def test_accepts_trade_with_gap() -> None: assert state.last_trade is trade_after_gap +@pytest.mark.parametrize( + ("first_trade_id", "next_trade_id"), + ( + (2**31 - 1, -(2**31)), + (-1, 0), + ), +) +def test_accepts_next_trade_across_signed_rollover( + first_trade_id: int, + next_trade_id: int, +) -> None: + state = TradeStreamState(symbol="BTCUSD") + first_trade = _trade(trade_id=first_trade_id) + next_trade = _trade(trade_id=next_trade_id) + + state.accept(first_trade) + result = state.accept(next_trade) + + assert result is next_trade + assert state.last_trade_id == next_trade_id + assert state.last_trade is next_trade + + +@pytest.mark.parametrize( + ("current_trade_id", "older_trade_id"), + ( + (-(2**31), 2**31 - 1), + (0, -1), + ), +) +def test_rejects_reverse_direction_across_signed_rollover( + current_trade_id: int, + older_trade_id: int, +) -> None: + state = TradeStreamState(symbol="BTCUSD") + state.accept(_trade(trade_id=current_trade_id)) + + with pytest.raises(TradeOrderingError): + state.accept(_trade(trade_id=older_trade_id)) + + assert state.last_trade_id == current_trade_id + + def test_checkpoint_preserves_trade_identity() -> None: state = TradeStreamState(symbol="BTCUSD") trade = _trade() @@ -158,6 +220,26 @@ def test_identical_duplicate_does_not_change_checkpoint() -> None: assert state.last_trade_id == original_trade.trade_id +def test_duplicate_from_rest_and_websocket_is_same_market_trade() -> None: + state = TradeStreamState(symbol="BTCUSD") + websocket_trade = _trade( + trade_id=-2_036_886_697, + source="dzengi_websocket_trade", + ) + rest_trade = _trade( + trade_id=-2_036_886_697, + source="dzengi", + ) + + state.accept(websocket_trade) + + result = state.accept(rest_trade) + + assert result is None + assert state.last_trade is websocket_trade + assert state.last_trade_id == websocket_trade.trade_id + + def test_raises_consistency_error_for_conflicting_duplicate() -> None: state = TradeStreamState(symbol="BTCUSD") original_trade = _trade( @@ -167,6 +249,7 @@ def test_raises_consistency_error_for_conflicting_duplicate() -> None: conflicting_trade = _trade( trade_id=100, price=Decimal("50001.00"), + source="dzengi_websocket_trade", ) state.accept(original_trade) diff --git a/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py b/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py index a1f9491..47410d3 100644 --- a/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py +++ b/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py @@ -13,6 +13,12 @@ from src.market_data.acquisition.adapters.dzengi.rest import ( from src.market_data.acquisition.consistency.trade_stream_exceptions import ( TradeConsistencyError, ) +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, @@ -281,6 +287,47 @@ def test_returns_trades_in_normalized_order() -> None: ) +@pytest.mark.parametrize( + ("before_rollover", "after_rollover"), + ( + (2**31 - 1, -(2**31)), + (-1, 0), + ), +) +def test_recovers_trades_in_rollover_aware_order( + before_rollover: int, + after_rollover: int, +) -> None: + source = StubTradesDocumentSource( + document=[ + _raw_trade(trade_id=after_rollover), + _raw_trade(trade_id=before_rollover), + ] + ) + consistency_controller = StubConsistencyController() + controller = TradeRecoveryController( + document_source=source, + consistency_controller=consistency_controller, + ) + + result = controller.recover(_request()) + + assert tuple( + trade.trade_id + for trade in result.recovered_trades + ) == ( + before_rollover, + after_rollover, + ) + assert [ + trade.trade_id + for trade in consistency_controller.received_trades + ] == [ + before_rollover, + after_rollover, + ] + + def test_excludes_duplicate_rejected_by_consistency_controller() -> None: source = StubTradesDocumentSource( document=[ @@ -375,6 +422,45 @@ def test_preserves_consistency_controller_returned_instance() -> None: assert result.recovered_trades[0] is replacement_trade +def test_rest_recovery_deduplicates_same_websocket_market_trade() -> None: + trade_id = -2_036_886_697 + timestamp = 1_700_000_000_000 + state_store = TradeStreamStateStore() + consistency_controller = TradeStreamConsistencyController( + state_store, + ) + websocket_trade = Trade( + symbol="BTCUSD", + trade_id=trade_id, + price=Decimal("50000.00"), + quantity=Decimal("0.25"), + executed_at=datetime.fromtimestamp( + timestamp / 1000, + tz=timezone.utc, + ), + aggressor_side=TradeAggressorSide.BUY, + source="dzengi_websocket_trade", + ) + consistency_controller.accept(websocket_trade) + controller = TradeRecoveryController( + document_source=StubTradesDocumentSource( + document=[ + _raw_trade( + trade_id=trade_id, + timestamp=timestamp, + ), + ] + ), + consistency_controller=consistency_controller, + ) + + result = controller.recover(_request()) + + assert result.is_empty is True + state = state_store.get("BTCUSD") + assert state.last_trade is websocket_trade + + def test_propagates_source_error() -> None: expected_error = RuntimeError("source error") @@ -572,4 +658,4 @@ def test_controller_does_not_create_additional_consistency_state() -> None: assert controller.__dict__ == { "_document_source": controller._document_source, "_consistency_controller": consistency_controller, - } \ No newline at end of file + } diff --git a/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py b/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py index 610b1b3..ed7ef66 100644 --- a/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py +++ b/app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py @@ -107,6 +107,68 @@ def test_sorts_arbitrary_order_by_trade_id() -> None: ) +def test_sorts_across_int32_max_to_int32_min_rollover() -> None: + before_rollover = _trade(trade_id=2**31 - 1) + after_rollover = _trade(trade_id=-(2**31)) + + result = normalize_recovered_trades( + ( + after_rollover, + before_rollover, + ) + ) + + assert result == ( + before_rollover, + after_rollover, + ) + + +def test_sorts_sequence_spanning_int32_sign_boundary() -> None: + trades_by_id = { + trade_id: _trade(trade_id=trade_id) + for trade_id in ( + 2**31 - 2, + 2**31 - 1, + -(2**31), + -(2**31) + 1, + ) + } + + result = normalize_recovered_trades( + ( + trades_by_id[-(2**31) + 1], + trades_by_id[2**31 - 1], + trades_by_id[2**31 - 2], + trades_by_id[-(2**31)], + ) + ) + + assert tuple(trade.trade_id for trade in result) == ( + 2**31 - 2, + 2**31 - 1, + -(2**31), + -(2**31) + 1, + ) + + +def test_sorts_across_minus_one_to_zero_rollover() -> None: + before_rollover = _trade(trade_id=-1) + after_rollover = _trade(trade_id=0) + + result = normalize_recovered_trades( + ( + after_rollover, + before_rollover, + ) + ) + + assert result == ( + before_rollover, + after_rollover, + ) + + def test_preserves_stable_order_for_equal_trade_ids() -> None: first_duplicate = _trade( trade_id=100, @@ -221,4 +283,4 @@ def test_returns_new_tuple_for_tuple_input() -> None: result = normalize_recovered_trades(source) assert result == source - assert result is not source \ No newline at end of file + assert result is not source diff --git a/app/tests/unit/market_data/acquisition/test_trade_id_sequence.py b/app/tests/unit/market_data/acquisition/test_trade_id_sequence.py new file mode 100644 index 0000000..2d3cf46 --- /dev/null +++ b/app/tests/unit/market_data/acquisition/test_trade_id_sequence.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import pytest + +from src.market_data.acquisition.trade_id_sequence import ( + SIGNED_TRADE_ID_MAX, + SIGNED_TRADE_ID_MIN, + is_trade_id_newer, + is_trade_id_same_or_newer, + trade_id_relative_offset, + validate_signed_trade_id, +) + + +@pytest.mark.parametrize( + "trade_id", + ( + SIGNED_TRADE_ID_MIN, + -1, + 0, + 1, + SIGNED_TRADE_ID_MAX, + ), +) +def test_validate_signed_trade_id_accepts_full_range( + trade_id: int, +) -> None: + assert validate_signed_trade_id(trade_id) is None + + +@pytest.mark.parametrize( + "trade_id", + ( + SIGNED_TRADE_ID_MIN - 1, + SIGNED_TRADE_ID_MAX + 1, + ), +) +def test_validate_signed_trade_id_rejects_out_of_range( + trade_id: int, +) -> None: + with pytest.raises(ValueError, match="signed 32-bit"): + validate_signed_trade_id(trade_id) + + +def test_validate_signed_trade_id_rejects_boolean() -> None: + with pytest.raises(TypeError, match="must be an integer"): + validate_signed_trade_id(True) + + +@pytest.mark.parametrize( + ("reference_trade_id", "candidate_trade_id"), + ( + (100, 101), + (-100, -99), + (SIGNED_TRADE_ID_MAX, SIGNED_TRADE_ID_MIN), + (-1, 0), + ), +) +def test_rollover_aware_contract_recognizes_next_id( + reference_trade_id: int, + candidate_trade_id: int, +) -> None: + assert ( + trade_id_relative_offset( + candidate_trade_id, + reference_trade_id, + ) + == 1 + ) + assert is_trade_id_newer( + candidate_trade_id, + reference_trade_id, + ) is True + assert is_trade_id_newer( + reference_trade_id, + candidate_trade_id, + ) is False + + +def test_same_id_is_same_or_newer_but_not_newer() -> None: + assert is_trade_id_same_or_newer(-100, -100) is True + assert is_trade_id_newer(-100, -100) is False + + +def test_half_cycle_distance_is_rejected_as_ambiguous() -> None: + with pytest.raises(ValueError, match="exactly half"): + trade_id_relative_offset( + SIGNED_TRADE_ID_MIN, + 0, + ) diff --git a/app/tests/unit/market_data/acquisition/validation/test_values.py b/app/tests/unit/market_data/acquisition/validation/test_values.py index 2401004..9a901d1 100644 --- a/app/tests/unit/market_data/acquisition/validation/test_values.py +++ b/app/tests/unit/market_data/acquisition/validation/test_values.py @@ -535,21 +535,40 @@ def test_validate_empty_rest_agg_trade_values() -> None: @pytest.mark.parametrize( - ("aggregate_trade_id", "timestamp"), + "aggregate_trade_id", [ - (0, 1000), - (-1, 1000), - (1, 0), - (1, -1), + -(2**31), + -2_037_115_004, + -1, + 0, + 1, + 2**31 - 1, ], ) -def test_reject_non_positive_rest_agg_trade_integer_value( +def test_accept_signed_rest_agg_trade_id( aggregate_trade_id: int, - timestamp: int, ) -> None: trades = ( _valid_rest_agg_trade( aggregate_trade_id=aggregate_trade_id, + ), + ) + + assert validate_rest_agg_trade_values(trades) is None + + +@pytest.mark.parametrize( + "timestamp", + [ + 0, + -1, + ], +) +def test_reject_non_positive_rest_agg_trade_timestamp( + timestamp: int, +) -> None: + trades = ( + _valid_rest_agg_trade( timestamp=timestamp, ), ) @@ -561,6 +580,43 @@ def test_reject_non_positive_rest_agg_trade_integer_value( validate_rest_agg_trade_values(trades) +def test_reject_boolean_rest_agg_trade_id() -> None: + trades = ( + _valid_rest_agg_trade( + aggregate_trade_id=True, + ), + ) + + with pytest.raises( + TradeValueError, + match=r"\$\[0\]\.aggregateTradeId должно быть целым числом", + ): + validate_rest_agg_trade_values(trades) + + +@pytest.mark.parametrize( + "aggregate_trade_id", + ( + -(2**31) - 1, + 2**31, + ), +) +def test_reject_out_of_range_rest_agg_trade_id( + aggregate_trade_id: int, +) -> None: + trades = ( + _valid_rest_agg_trade( + aggregate_trade_id=aggregate_trade_id, + ), + ) + + with pytest.raises( + TradeValueError, + match="signed 32-bit", + ): + validate_rest_agg_trade_values(trades) + + @pytest.mark.parametrize( ("field", "value"), [ @@ -684,4 +740,4 @@ def test_rest_agg_trade_value_error_reports_item_index() -> None: TradeValueError, match=r"\$\[1\]\.quantity должно быть больше нуля", ): - validate_rest_agg_trade_values(trades) \ No newline at end of file + validate_rest_agg_trade_values(trades) diff --git a/app/tests/unit/market_data/acquisition/validation/test_websocket_trade_values.py b/app/tests/unit/market_data/acquisition/validation/test_websocket_trade_values.py index 975d5c6..04e0d18 100644 --- a/app/tests/unit/market_data/acquisition/validation/test_websocket_trade_values.py +++ b/app/tests/unit/market_data/acquisition/validation/test_websocket_trade_values.py @@ -85,17 +85,45 @@ def test_validate_websocket_trade_values_does_not_modify_event() -> None: @pytest.mark.parametrize( "trade_id", [ - 0, + -(2**31), + -2_037_129_153, -1, - -123456, + 0, + 1, + 2**31 - 1, ], ) -def test_validate_websocket_trade_values_rejects_non_positive_trade_id( +def test_validate_websocket_trade_values_accepts_signed_trade_id( + trade_id: int, +) -> None: + validate_dzengi_websocket_trade_values( + _event(trade_id=trade_id) + ) + + +def test_validate_websocket_trade_values_rejects_boolean_trade_id() -> None: + with pytest.raises( + TradeValueError, + match=r"\$\.payload\.id должно быть целым числом", + ): + validate_dzengi_websocket_trade_values( + _event(trade_id=True) + ) + + +@pytest.mark.parametrize( + "trade_id", + ( + -(2**31) - 1, + 2**31, + ), +) +def test_validate_websocket_trade_values_rejects_out_of_range_trade_id( trade_id: int, ) -> None: with pytest.raises( TradeValueError, - match=r"\$\.payload\.id должно быть целым числом больше нуля", + match="signed 32-bit", ): validate_dzengi_websocket_trade_values( _event(trade_id=trade_id) @@ -272,4 +300,4 @@ def test_validate_websocket_trade_values_rejects_empty_order_id( ): validate_dzengi_websocket_trade_values( _event(order_id=order_id) - ) \ No newline at end of file + ) diff --git a/app/tests/unit/test_live_trade_stream_support.py b/app/tests/unit/test_live_trade_stream_support.py new file mode 100644 index 0000000..1ec144f --- /dev/null +++ b/app/tests/unit/test_live_trade_stream_support.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import pytest + +from tests.support.live_trade_stream import ( + DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS, + LIVE_REST_URL_ENV, + LIVE_SYMBOLS_ENV, + LIVE_TRADE_TIMEOUT_ENV, + LIVE_WEBSOCKET_URL_ENV, + RUN_LIVE_TESTS_ENV, + LiveTestConfigurationError, + build_live_trade_stream_settings, + load_live_trade_stream_test_config, +) + + +REST_URL = "https://api-adapter.dzengi.com" +WEBSOCKET_URL = "wss://api-adapter.dzengi.com/connect" +SYMBOL = "BTC/USD_LEVERAGE" + + +def make_environment() -> dict[str, str]: + return { + RUN_LIVE_TESTS_ENV: "1", + LIVE_REST_URL_ENV: REST_URL, + LIVE_WEBSOCKET_URL_ENV: WEBSOCKET_URL, + LIVE_SYMBOLS_ENV: SYMBOL, + } + + +@pytest.mark.parametrize( + "opt_in", + ( + "", + "0", + ), +) +def test_live_config_is_disabled_without_exact_opt_in( + opt_in: str, +) -> None: + assert load_live_trade_stream_test_config( + { + RUN_LIVE_TESTS_ENV: opt_in, + } + ) is None + + +def test_invalid_live_opt_in_is_rejected() -> None: + with pytest.raises( + LiveTestConfigurationError, + match="must be exactly 1", + ): + load_live_trade_stream_test_config( + { + RUN_LIVE_TESTS_ENV: "true", + } + ) + + +@pytest.mark.parametrize( + "missing_name", + ( + LIVE_REST_URL_ENV, + LIVE_WEBSOCKET_URL_ENV, + LIVE_SYMBOLS_ENV, + ), +) +def test_enabled_live_config_requires_every_explicit_value( + missing_name: str, +) -> None: + environment = make_environment() + del environment[missing_name] + + with pytest.raises( + LiveTestConfigurationError, + match=missing_name, + ): + load_live_trade_stream_test_config(environment) + + +@pytest.mark.parametrize( + ("name", "value"), + ( + ( + LIVE_REST_URL_ENV, + "http://api-adapter.dzengi.com", + ), + ( + LIVE_REST_URL_ENV, + "https://key@example.com", + ), + ( + LIVE_WEBSOCKET_URL_ENV, + "ws://api-adapter.dzengi.com/connect", + ), + ( + LIVE_WEBSOCKET_URL_ENV, + "wss://api-adapter.dzengi.com/not-connect", + ), + ), +) +def test_live_config_rejects_unsafe_or_wrong_endpoints( + name: str, + value: str, +) -> None: + environment = make_environment() + environment[name] = value + + with pytest.raises(LiveTestConfigurationError): + load_live_trade_stream_test_config(environment) + + +def test_live_config_requires_exactly_one_symbol() -> None: + environment = make_environment() + environment[LIVE_SYMBOLS_ENV] = ( + "BTC/USD_LEVERAGE,ETH/USD_LEVERAGE" + ) + + with pytest.raises( + LiveTestConfigurationError, + match="exactly one symbol", + ): + load_live_trade_stream_test_config(environment) + + +def test_live_config_uses_bounded_default_trade_timeout() -> None: + config = load_live_trade_stream_test_config( + make_environment() + ) + + assert config is not None + assert ( + config.trade_timeout_seconds + == DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS + == 600.0 + ) + + +@pytest.mark.parametrize( + "timeout_value", + ( + "0", + "-1", + "nan", + "inf", + "not-a-number", + ), +) +def test_live_config_rejects_invalid_trade_timeout( + timeout_value: str, +) -> None: + environment = make_environment() + environment[LIVE_TRADE_TIMEOUT_ENV] = timeout_value + + with pytest.raises(LiveTestConfigurationError): + load_live_trade_stream_test_config(environment) + + +def test_live_settings_are_explicit_and_credential_free() -> None: + config = load_live_trade_stream_test_config( + make_environment() + ) + + assert config is not None + settings = build_live_trade_stream_settings(config) + + assert settings.exchange_base_url == REST_URL + assert settings.exchange_api_key == "" + assert settings.exchange_api_secret == "" + assert settings.trade_stream.enabled is True + assert settings.trade_stream.websocket_url == WEBSOCKET_URL + assert settings.trade_stream.symbols == (SYMBOL,) diff --git a/app/tests/unit/test_trade_stream_runtime_support.py b/app/tests/unit/test_trade_stream_runtime_support.py new file mode 100644 index 0000000..1a43650 --- /dev/null +++ b/app/tests/unit/test_trade_stream_runtime_support.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from tests.support.trade_stream_runtime import ( + run_scenario, + wait_until_or_runtime_exit, +) + + +def test_wait_until_or_runtime_exit_returns_when_condition_is_reached() -> None: + async def scenario() -> None: + condition_reached = False + + async def keep_runtime_alive() -> None: + await asyncio.Event().wait() + + runtime_task = asyncio.create_task( + keep_runtime_alive(), + ) + + try: + condition_reached = True + + await wait_until_or_runtime_exit( + lambda: condition_reached, + runtime_task=runtime_task, + timeout_seconds=0.5, + ) + + assert runtime_task.done() is False + finally: + runtime_task.cancel() + + with pytest.raises(asyncio.CancelledError): + await runtime_task + + run_scenario(scenario()) + + +def test_wait_until_or_runtime_exit_propagates_runtime_error_immediately() -> None: + expected_error = RuntimeError("runtime failed") + + async def scenario() -> None: + async def fail_runtime() -> None: + await asyncio.sleep(0) + raise expected_error + + runtime_task = asyncio.create_task( + fail_runtime(), + ) + + with pytest.raises(RuntimeError) as exc_info: + await asyncio.wait_for( + wait_until_or_runtime_exit( + lambda: False, + runtime_task=runtime_task, + timeout_seconds=10.0, + ), + timeout=0.5, + ) + + assert exc_info.value is expected_error + + run_scenario(scenario()) + + +def test_wait_until_or_runtime_exit_rejects_clean_early_runtime_exit() -> None: + async def scenario() -> None: + async def finish_runtime() -> None: + await asyncio.sleep(0) + + runtime_task = asyncio.create_task( + finish_runtime(), + ) + + with pytest.raises( + RuntimeError, + match="exited before the expected live condition", + ): + await asyncio.wait_for( + wait_until_or_runtime_exit( + lambda: False, + runtime_task=runtime_task, + timeout_seconds=10.0, + ), + timeout=0.5, + ) + + run_scenario(scenario()) diff --git a/docs/migrations/build_060_26.md b/docs/migrations/build_060_26.md new file mode 100644 index 0000000..4a1a663 --- /dev/null +++ b/docs/migrations/build_060_26.md @@ -0,0 +1,220 @@ +# Build 060.26 — Integration and Regression + +**Engineering Migration Report** + +--- + +## Контроль документа + +| Свойство | Значение | +|---|---| +| Build | 060.26 | +| Статус | Completed | +| Подсистема | Market Data Acquisition | +| Компонент | Trades Feed / Trade Stream Runtime | +| Дата завершения | 2026-07-31 | +| Версия | 1.0 | + +--- + +## Связанные документы + +- `build_060_25.md` — итог Production Runtime Integration; +- `build_060_26_architecture.md` — архитектура, решения и подробные + результаты подэтапов 060.26.0–060.26.6; +- `dzentra_target_architecture.md` — место Trade Stream в целевой + архитектуре Dzentra; +- `master-roadmap.md` — дальнейшая последовательность Build. + +--- + +## 1. Назначение Build + +Build 060.25 собрал Production Trade Stream Runtime и проверил его на +детерминированных in-process boundaries. Цель Build 060.26 — проверить +тот же production graph через реальные сетевые границы и длительный +lifecycle: + +```text +production factory + → local/live WebSocket connection + → subscription and ACK + → Live Trade processing + → network fault + → single-flight reconnect + → subscription restore + → REST Recovery + → buffered Live processing + → continued Runtime + → graceful shutdown without leaks +``` + +Build выполнялся по verification-first правилу: production-код мог +изменяться только при обнаружении реального дефекта проверочным +сценарием. + +--- + +## 2. Завершённые подэтапы + +| Подэтап | Название | Статус | +|---|---|---| +| 060.26.0 | Verification Contract and Test Taxonomy | Accepted | +| 060.26.1 | Loopback Network Harness | Accepted | +| 060.26.2 | Reconnect and Recovery Integration | Accepted | +| 060.26.3 | Fault, Cancellation and Resource Verification | Accepted | +| 060.26.4 | Stress and Soak Verification | Accepted | +| 060.26.5 | Opt-in Dzengi Live Verification | Accepted | +| 060.26.6 | Final Regression and Acceptance | Accepted | + +Каждая группа проходила отдельный read-only review. Обнаруженные +findings исправлялись и закрывались regression-тестами до принятия. + +--- + +## 3. Test taxonomy и сетевой стенд + +В pytest зарегистрированы три явных набора: + +- `integration` — настоящие локальные TCP, WebSocket и HTTP соединения; +- `stress` — локальная нагрузка, reconnect storm и soak; +- `live` — только явная opt-in проверка публичных Dzengi endpoints. + +Обычная регрессия исключает все три набора. Поэтому локальная сеть, +длительные тесты и внешний exchange не запускаются неявно. + +Loopback harness использует production WebSocket transport и REST +client. Он умеет выполнять graceful close, TCP abort, client-to-server +blackhole, HTTP delay/error и invalid WebSocket message, а также ведёт +точные счётчики соединений, handlers, relay tasks, stream writers и +HTTP server thread. + +--- + +## 4. Проверенные гарантии Runtime + +Интеграционные сценарии подтвердили: + +1. subscription restore происходит до REST Recovery; +2. recovered Trades обрабатываются до buffered live Trades; +3. receive failure и Ping timeout используют одну reconnect generation; +4. graceful close, TCP abort и последовательные reconnect не создают + конкурирующих операций; +5. HTTP error и invalid message завершают Runtime предсказуемо; +6. cancellation ожидает завершения блокирующего REST worker; +7. ошибка cleanup не скрывает primary Runtime error; +8. partial startup не оставляет запущенных задач и серверов; +9. после завершения отсутствуют owned Runtime и harness resources. + +--- + +## 5. Stress и soak + +Fixed stress проверяет: + +- 20 000 последовательных Trades; +- точный ordering и итоговый checkpoint; +- ограничение deduplication window до 10 000 записей; +- ненулевой memory baseline; +- остаточный рост памяти после прогрева не более 4 MiB; +- 30 reconnect generations с чередованием close, abort и Ping timeout. + +Стандартный soak выполняется 120 секунд и обрабатывает 6 000 Trades с +24 reconnect. Финальный расширенный soak выполнялся 900 секунд и +обработал 45 000 Trades со 180 reconnect в одном Runtime lifecycle. + +--- + +## 6. Live verification и обнаруженные дефекты + +Opt-in live harness использует только публичные market-data операции, +явные HTTPS/WSS URL и один явно заданный symbol. Он не использует API +credentials, Telegram, database или торговые операции. + +Live-проверка подтвердила production sequence: + +```text +Live Trade + → checkpoint + → reconnect + → subscription restore + → REST Recovery + → новая Live Trade + → graceful stop +``` + +Первый live-запуск выявил, что Dzengi Trade ID является signed 32-bit +значением. Повторный запуск выявил одну и ту же биржевую сделку через +WebSocket и REST с различным transport `source`. + +Реальные дефекты исправлены единым контрактом: + +- WebSocket и REST принимают полный signed 32-bit диапазон; +- `INT32_MAX → INT32_MIN` и `-1 → 0` являются продвижением ID; +- расстояние ровно в половину цикла отклоняется как неоднозначное; +- Consistency и Recovery используют один rollover-aware порядок; +- identity биржевой сделки не зависит от transport `source`, но + различие цены, количества, времени или стороны остаётся ошибкой. + +Проверка также подтвердила HTTP 200 и одинаковую структуру ответа +`/api/v1/aggTrades` и `/api/v2/aggTrades` для +`BTC/USD_LEVERAGE`. Production Recovery сохраняет текущий endpoint +`/api/v1/aggTrades`. + +--- + +## 7. Финальные результаты + +Итоговая приёмка выполнена 2026-07-31: + +```text +Rollover/Recovery target: 401 passed +Integration repeated ten times: 110 passed +Fixed stress target: 3 passed, 1 deselected in 6.81s +Standard 120-second soak: 1 passed, 3 deselected in 120.09s +Extended 900-second soak: 1 passed, 3 deselected in 900.11s +Full offline regression: 1930 passed, 16 deselected in 3.64s +Opt-in Dzengi live verification: 1 passed in 140.42s +git diff --check: clean +``` + +В integration, stress и soak прогонах `ResourceWarning` считался +ошибкой. После каждого сценария подтверждалось отсутствие оставшихся +Runtime tasks, WebSocket/TCP handlers, relay tasks, tracked writers и +HTTP server threads. + +--- + +## 8. Границы Build + +Build 060.26 не реализует: + +- persistent Market Data Storage; +- persistent checkpoint; +- startup recovery после перезапуска процесса; +- Historical Queries; +- Replay API; +- Analytics API; +- новый reconnect retry/backoff policy. + +Эти задачи относятся к Build 060.27–060.29. Полная итоговая ревизия +документации подсистемы остаётся задачей Build 060.30. + +Постороннее пользовательское изменение `.gitignore` не относится к +Build 060.26 и не должно включаться в его staging. + +--- + +## 9. Итог + +Build 060.26 завершён. + +Production Trade Stream Runtime проверен через настоящие локальные +сетевые соединения, управляемые faults, длительные reconnect/recovery +циклы и ограниченный live Dzengi scenario. Финальный read-only review +не выявил незакрытых findings, а стандартный и расширенный soak не +обнаружили утечек задач, сокетов, потоков или памяти сверх принятого +порога. + +Следующий этап — Build 060.27, посвящённый долговременному хранению +Trades и формированию Persistent Market Data Storage. diff --git a/docs/migrations/build_060_26_architecture.md b/docs/migrations/build_060_26_architecture.md new file mode 100644 index 0000000..68382f1 --- /dev/null +++ b/docs/migrations/build_060_26_architecture.md @@ -0,0 +1,639 @@ +# Build 060.26 — Integration and Regression Architecture + +**Статус:** Completed + +**Build:** 060.26 + +**Подсистема:** Market Data Acquisition / Trade Stream Runtime + +**Дата начала:** 2026-07-31 + +**Дата завершения:** 2026-07-31 + +**Версия документа:** 1.0 + +--- + +## 1. Назначение + +Документ фиксирует архитектуру интеграционной проверки Production +Trade Stream Runtime после завершения Build 060.25. + +Build 060.25 доказал корректность production composition и lifecycle +на детерминированных in-process boundaries. Build 060.26 проверяет +тот же production graph через настоящие сетевые transport +границы: + +- локальное WebSocket-соединение; +- локальный HTTP endpoint Trade Recovery; +- управляемые разрывы, задержки и ошибки сети; +- многократные reconnect/recovery generations; +- отсутствие утечек задач и ресурсов; +- отдельные stress, soak и live exchange проверки. + +Build является verification-first этапом. Production-код не меняется, +если интеграционный сценарий не обнаружит реальный дефект. + +--- + +## 2. Статус подэтапов + +| Подэтап | Название | Статус | +|---|---|---| +| 060.26.0 | Verification Contract and Test Taxonomy | Accepted | +| 060.26.1 | Loopback Network Harness | Accepted | +| 060.26.2 | Reconnect and Recovery Integration | Accepted | +| 060.26.3 | Fault, Cancellation and Resource Verification | Accepted | +| 060.26.4 | Stress and Soak Verification | Accepted | +| 060.26.5 | Opt-in Dzengi Live Verification | Accepted | +| 060.26.6 | Final Regression and Acceptance | Accepted | + +Подэтапы 060.26.0–060.26.3 были реализованы первой группой и прошли +отдельный read-only review до начала длительных stress, soak и live +exchange проверок. + +--- + +## 3. Исходное состояние + +Production Runtime уже включает: + +```text +Application bootstrap + │ + ▼ +TradeStreamProductionRuntime + │ + ├── DzengiWebSocketTransport + ├── WebSocketSession + ├── WebSocketSubscriptionManager + ├── RuntimeReconnectRecoveryCoordinator + ├── RuntimeSupervisor + ├── RuntimeScheduler + └── TradeStreamStateStore + ▲ + │ + Live + Recovery +``` + +Build 060.25.8 проверяет этот граф через управляемые fake WebSocket и +REST зависимости. Такие тесты подтверждают ordering, lifecycle и +cancellation, но не проверяют: + +- реальный WebSocket handshake и frame exchange; +- поведение при TCP close/reset; +- настоящий HTTP request из `ExchangeRestClient`; +- взаимодействие asyncio с блокирующим REST worker; +- освобождение реальных сокетов и серверных handler tasks. + +Существующие диагностические `app/scripts/check_*` обращаются к +протоколам Dzengi напрямую и не являются проверкой Production Runtime. + +--- + +## 4. Классификация тестов + +### 4.1. Unit + +Unit suite: + +- не использует сетевые соединения; +- выполняется обычной командой `pytest`; +- остаётся основной быстрой регрессией проекта; +- сохраняет существующие deterministic fake boundaries. + +### 4.2. Integration + +Маркер: + +```text +integration +``` + +Integration suite: + +- запускает настоящие локальные TCP/WebSocket/HTTP соединения; +- использует production factory и production transport/client; +- не обращается к интернету; +- управляет задержками и ошибками через локальный test harness; +- запускается отдельной явной командой. + +### 4.3. Stress + +Маркер: + +```text +stress +``` + +Stress suite: + +- использует только локальные endpoints; +- проверяет большие последовательности Trade и reconnect generations; +- не входит в обычную unit или integration regression; +- не задаёт хрупких требований к абсолютной производительности. + +### 4.4. Live + +Маркер: + +```text +live +``` + +Live suite: + +- запускается только при явном opt-in; +- требует явно заданные WebSocket URL, REST URL и symbols; +- не использует fallback; +- выполняет только публичные market-data операции; +- не создаёт, не изменяет и не отменяет торговые ордера; +- не входит в обычную регрессию. + +### 4.5. Команды + +```text +# Обычная unit-регрессия +.venv/bin/python -m pytest -q + +# Локальная интеграция +.venv/bin/python -m pytest -q -m integration tests/integration + +# Локальный stress/soak +.venv/bin/python -m pytest -q -m stress tests/stress + +# Расширенный 15-минутный soak +DZENTRA_SOAK_SECONDS=900 \ +.venv/bin/python -m pytest -q -m stress -k soak tests/stress + +# Явная live exchange проверка +DZENTRA_RUN_LIVE_TESTS=1 \ +DZENTRA_LIVE_REST_URL=https://api-adapter.dzengi.com \ +DZENTRA_LIVE_WS_URL=wss://api-adapter.dzengi.com/connect \ +DZENTRA_LIVE_SYMBOLS=BTC/USD_LEVERAGE \ +.venv/bin/python -m pytest -q -m live tests/live +``` + +Конфигурация pytest по умолчанию исключает `integration`, `stress` и +`live`. Поэтому добавление сетевых сценариев не меняет контракт +обычного unit suite. + +--- + +## 5. Loopback Network Harness + +Test harness владеет только тестовой инфраструктурой: + +```text +Production factory + │ + ▼ +TradeStreamProductionRuntime + │ + ├── DzengiWebSocketTransport + │ │ + │ ▼ + │ Loopback WebSocket server + │ + └── ExchangeRestClient + │ + ▼ + Loopback HTTP server +``` + +Дополнительный TCP fault proxy может: + +- пропускать трафик без изменений; +- обрывать конкретное соединение; +- прекращать передачу client → server frames; +- моделировать отсутствие Pong без изменения production transport. + +Harness записывает: + +- номера TCP/WebSocket connections; +- порядок subscription requests; +- correlation ID; +- порядок HTTP recovery requests; +- query parameters recovery window; +- открытые server handlers и relay tasks. + +Harness не: + +- подменяет production parsing или consistency; +- вызывает Recovery Controller напрямую; +- создаёт собственный checkpoint; +- меняет production retry policy; +- хранит market state вместо Runtime. + +--- + +## 6. Основной reconnect/recovery сценарий + +Обязательная последовательность: + +```text +connect + → subscribe + → ACK + → live Trade + → checkpoint + → network disconnect + → reconnect + → restore subscription + → REST Recovery started + → buffered live Trade received + → recovered Trade accepted + → buffered live Trade accepted +``` + +Проверяемые инварианты: + +1. новый connection создаётся один раз для одной generation; +2. subscription restore происходит раньше первого REST request; +3. Live processing закрыт на время Recovery; +4. recovered Trade обрабатывается раньше buffered live Trade; +5. общий checkpoint принадлежит одному `TradeStreamStateStore`; +6. дубликаты на границе Recovery не меняют результат; +7. после успешного Recovery receive loop продолжает работу. + +--- + +## 7. Fault and Cancellation Matrix + +Первая группа Build проверяет: + +- graceful и abrupt WebSocket disconnect; +- повторные успешные reconnect generations; +- одновременное наблюдение одной ошибки receive loop и Heartbeat; +- Ping/Pong timeout через управляемый TCP blackhole; +- HTTP latency; +- HTTP error; +- невалидный WebSocket JSON; +- остановку Runtime во время активного REST worker; +- terminal error без обработки buffered live Trade; +- освобождение Runtime tasks, WebSocket handlers, relay tasks и HTTP + server thread. + +Каждый сценарий имеет внешний timeout. Ожидание состояния выполняется +по наблюдаемому событию или условию, а не через фиксированную длинную +паузу. + +--- + +## 8. Error Policy + +Build 060.26 проверяет принятую политику, но не меняет её неявно: + +- startup failure включённого Trade Stream является terminal; +- неуспешная reconnect/recovery операция является terminal; +- повторные успешные reconnect возможны в следующих generations; +- один transport failure не создаёт конкурирующие reconnect; +- cancellation не открывает Live gate до завершения REST worker; +- cleanup error не скрывает primary Runtime error. + +Если тест обнаруживает несоответствие production-кода этим правилам, +дефект сначала документируется и проходит отдельное согласование. + +--- + +## 9. Resource Ownership + +Production Runtime продолжает владеть: + +- startup task; +- receive task; +- Scheduler task; +- Recovery worker task; +- WebSocket Session lifecycle. + +Test harness владеет: + +- WebSocket listening socket; +- WebSocket server handler tasks; +- HTTP listening socket; +- HTTP server thread; +- TCP proxy listening socket; +- TCP relay tasks и stream writers. + +После каждого теста должны быть закрыты обе группы ресурсов. Проверка +не считается завершённой, если остались: + +- именованные Runtime tasks; +- активные server handlers; +- незавершённые relay tasks; +- незакрытые stream writers; +- живой HTTP server thread. + +--- + +## 10. Границы Build + +Build 060.26 не реализует: + +- persistent Trade, Quote или Candle storage; +- persistent checkpoint; +- startup recovery после перезапуска процесса; +- Historical Queries; +- Replay API; +- Analytics API; +- новый reconnect retry/backoff policy; +- автоматическое включение live exchange tests. + +Эти задачи относятся к Build 060.27–060.29 либо требуют отдельного +архитектурного решения. + +--- + +## 11. Документационная политика + +Во время Build 060.26: + +- этот документ хранит архитектуру и фактический статус подэтапов; +- итоговый результат зафиксирован в `build_060_26.md`; +- в `master-roadmap.md` меняется только фактический статус Build; +- полная ревизия исторических документов и старой нумерации + откладывается до Build 060.30. + +--- + +## 12. Критерии завершения 060.26.0–060.26.3 + +- pytest test taxonomy отделяет network suites от unit regression; +- локальный WebSocket server работает с production transport; +- локальный HTTP server работает с production `ExchangeRestClient`; +- reconnect восстанавливает подписку до Recovery; +- Recovery завершается до обработки buffered live Trade; +- Ping/Pong и network disconnect проходят single-flight coordination; +- HTTP error и invalid message завершают Runtime предсказуемо; +- cancellation дожидается REST worker; +- все тестовые и production ресурсы освобождаются; +- целевой integration suite и полная unit-регрессия проходят; +- отдельный read-only review не содержит незакрытых findings. + +--- + +## 13. Фактическая реализация 060.26.0–060.26.3 + +Добавлены: + +```text +app/tests/integration/market_data/acquisition/runtime/ + loopback_trade_exchange.py + test_trade_stream_loopback_integration.py +``` + +Изменены только для классификации и фактического статуса: + +```text +app/pytest.ini +docs/roadmap/master-roadmap.md +``` + +Loopback harness содержит: + +- настоящий WebSocket server; +- настоящий HTTP server в отдельном управляемом thread; +- TCP fault proxy с client → server blackhole; +- запись connection, subscription и REST request order; +- явный cleanup всех принадлежащих harness ресурсов. + +Production-код Build 060.25 не изменён. + +Реализованы десять integration-сценариев: + +1. production WebSocket transport обновляет общий checkpoint; +2. subscription restore предшествует REST Recovery, а recovered Trade + предшествует buffered live Trade; +3. graceful и abrupt disconnect проходят три последовательных + reconnect generations; +4. HTTP Recovery error блокирует buffered live и завершает Runtime; +5. невалидный WebSocket JSON является terminal error; +6. cancellation ждёт завершения реального HTTP worker; +7. Ping timeout и receive failure используют одну reconnect generation, + при этом явно подтверждено участие Scheduler и receive loop; +8. timeout во время Runtime startup останавливает созданную Runtime + task; +9. ошибка запуска WebSocket server не оставляет запущенным HTTP server; +10. ошибка cleanup WebSocket server не мешает последующему cleanup + HTTP server. + +Проверки на 2026-07-31: + +```text +Integration target: 10 passed +Integration repeated three times: 30 passed +Integration with ResourceWarning as error: 10 passed +Full unit regression: 1869 passed +Default-suite integration deselection: 10 deselected +``` + +Первичный review выявил четыре замечания в test harness: владение +Runtime task при неуспешном startup, ограниченность и +exception-safety cleanup, две ошибки статической типизации и +недостаточно явное доказательство участия обоих single-flight callers. +Все замечания исправлены и закрыты regression-тестами. + +Повторный read-only review не выявил новых findings. Подэтапы +060.26.0–060.26.3 имеют статус `Accepted`. + +--- + +## 14. Фактическая реализация 060.26.4 + +Добавлены: + +```text +app/tests/support/ + __init__.py + trade_stream_runtime.py + +app/tests/stress/market_data/acquisition/runtime/ + test_trade_stream_runtime_stress.py +``` + +Общий test-support владеет только test-side запуском и остановкой +production Runtime. Integration и stress tests используют один +lifecycle contract и не импортируют функции из других `test_*.py`. + +Loopback harness дополнен наблюдаемыми счётчиками открытых WebSocket +connections и TCP stream writers. Production-код не изменён. + +Реализованы четыре локальных stress/soak-сценария: + +1. отдельная проверка подтверждает, что `tracemalloc` filter наблюдает + ненулевые allocations подсистемы; +2. 20 000 последовательных Trades проверяют ordering, итоговый + checkpoint, ограничение deduplication window до 10 000 записей и + остаточный рост памяти после прогрева не более 4 MiB; +3. 30 reconnect generations чередуют graceful close, TCP abort и Ping + timeout и проверяют точное соответствие generations, connections, + subscriptions и REST Recovery requests; +4. стандартный 120-секундный soak обрабатывает 6 000 Trades и 24 + reconnect в одном Runtime lifecycle. Через + `DZENTRA_SOAK_SECONDS=900` доступен отдельный 15-минутный профиль. + +После каждого сценария проверяется отсутствие именованных Runtime +tasks, активных WebSocket/TCP handlers, relay tasks, tracked stream +writers и живого HTTP server thread. Между reconnect проверяются +состояние Runtime, Live gate и отсутствие незавершённой Recovery task. + +Проверки на 2026-07-31: + +```text +Fixed stress target: 3 passed in 6.75s +Standard soak with ResourceWarning error: 1 passed in 120.09s +Integration regression: 11 passed +Full unit regression: 1869 passed +Default-suite network deselection: 15 deselected +``` + +Первичный review выявил три замечания: writer tracking очищался до +доказательства закрытия transport, memory filter мог пройти с пустым +baseline, а stress ожидания неявно наследовали integration timeout. +Все замечания исправлены и закрыты regression-проверками. + +Повторный read-only review не выявил новых findings. Подэтап 060.26.4 +имеет статус `Accepted`. + +--- + +## 15. Фактическая реализация 060.26.5 + +Добавлены: + +```text +app/src/market_data/acquisition/ + trade_id_sequence.py + +app/tests/support/ + async_wait.py + live_trade_stream.py + +app/tests/live/market_data/acquisition/runtime/ + test_trade_stream_runtime_live.py + +app/tests/unit/ + test_live_trade_stream_support.py + test_trade_stream_runtime_support.py + +app/tests/unit/market_data/acquisition/ + test_trade_id_sequence.py +``` + +Live configuration contract требует: + +- точный opt-in `DZENTRA_RUN_LIVE_TESTS=1`; +- отдельный HTTPS REST base URL; +- отдельный WSS URL с endpoint `/connect`; +- ровно один явно заданный symbol; +- отсутствие embedded credentials, query и fragment в URL. + +Без opt-in live test завершается `SKIP` до импорта production +configuration и до создания сетевых компонентов. При включённом opt-in +неполная или небезопасная конфигурация является ошибкой, а не получает +fallback из `.env`, `DEFAULT_SYMBOL` или demo settings. + +Test-side `Settings` создаются явно и всегда содержат пустые API key и +secret. Telegram, application bootstrap, торговые endpoints и order +operations не используются. + +Ожидание рыночного события ограничено 600 секундами. Runtime failure +при этом распространяется немедленно через fail-fast helper и не +маскируется длительным ожиданием внешнего рынка. + +Реализован один ограниченный live-сценарий: + +```text +production factory + → Runtime RUNNING + → первая Live Trade + → checkpoint + → один RuntimeReconnectRecoveryCoordinator.reconnect() + → subscription restore + → production REST Recovery + → новая Live Trade после Recovery + → graceful stop + → отсутствие owned Runtime tasks +``` + +Сценарий намеренно использует текущий production Recovery endpoint +`/api/v1/aggTrades`. Исторические исследования и диагностический +скрипт проекта используют `/api/v2/aggTrades`. Фактическая проверка +подтвердила HTTP 200 и одинаковую структуру ответа обоих endpoint для +`BTC/USD_LEVERAGE`; production endpoint менять не требуется. + +Первый live-запуск выявил, что Dzengi возвращает signed 32-bit Trade ID, +включая отрицательные значения. Следующий запуск выявил закономерный +повтор одной сделки через WebSocket и REST с разным transport +`source`. Оба факта были реальными production-дефектами прежнего +контракта и исправлены: + +- WebSocket и REST validation принимают полный signed 32-bit диапазон; +- одинаковый market fact дедуплицируется независимо от transport + `source`, но различие цены, количества, времени или стороны остаётся + consistency error; +- единый `trade_id_sequence` задаёт modular-порядок по циклу `2^32`; +- переходы `INT32_MAX → INT32_MIN` и `-1 → 0` считаются продвижением; +- Consistency, Recovery normalizer и live assertions используют один + rollover-aware contract; +- расстояние ровно в половину цикла отклоняется как неоднозначное; +- recovery-набор должен занимать меньше половины цикла, что существенно + шире максимального часового Runtime Recovery window. + +Итоговые проверки на 2026-07-31: + +```text +Rollover/Recovery target: 401 passed +Integration regression: 11 passed +Fixed stress regression: 3 passed +Full offline regression: 1930 passed +Default-suite network deselection: 16 deselected +Opt-in Dzengi live verification: 1 passed in 140.42s +``` + +Финальный read-only review проверил modular ordering, signed boundary, +cross-source identity, fail-fast ownership, bounded timeout и отсутствие +обычных сравнений Trade ID в production/live ordering paths. Новых +findings не обнаружено. Подэтап 060.26.5 имеет статус `Accepted`. + +--- + +## 16. Фактическая реализация 060.26.6 + +Финальная приёмка повторно проверила все уровни Build без изменения +production-кода: + +1. десять последовательных запусков локального integration-набора; +2. fixed stress на 20 000 Trades и 30 reconnect generations; +3. стандартный 120-секундный soak на 6 000 Trades и 24 reconnect; +4. расширенный 900-секундный soak на 45 000 Trades и 180 reconnect; +5. полную offline-регрессию проекта; +6. итоговый read-only review production, test и documentation diff. + +Во всех сетевых и нагрузочных сценариях `ResourceWarning` считался +ошибкой. После завершения подтверждено отсутствие принадлежащих Runtime +задач, WebSocket/TCP handlers, relay tasks, tracked stream writers и +живого HTTP server thread. + +Финальные результаты на 2026-07-31: + +```text +Integration repeated ten times: 110 passed +Fixed stress target: 3 passed, 1 deselected in 6.81s +Standard 120-second soak: 1 passed, 3 deselected in 120.09s +Extended 900-second soak: 1 passed, 3 deselected in 900.11s +Full offline regression: 1930 passed, 16 deselected in 3.64s +Opt-in Dzengi live verification: 1 passed in 140.42s +git diff --check: clean +``` + +Live verification не запускалась повторно в 060.26.6: её успешный +результат получен после последних production-изменений signed Trade ID, +а сам acceptance-этап production-код больше не менял. + +Итоговый review не выявил новых findings. Production-изменения Build +ограничены реальными дефектами signed Trade ID, обнаруженными live +verification. Изменение `.gitignore` не относится к Build 060.26 и не +должно включаться в его staging. + +Подэтап 060.26.6 принят. Build 060.26 завершён. diff --git a/docs/roadmap/master-roadmap.md b/docs/roadmap/master-roadmap.md index 2f15117..86cd9e5 100644 --- a/docs/roadmap/master-roadmap.md +++ b/docs/roadmap/master-roadmap.md @@ -8,8 +8,8 @@ | Статус | Active | | Версия | 2.0 | | Дата актуализации | 2026-07-31 | -| Текущий завершённый Build | 060.25 | -| Следующий Build | 060.26 | +| Текущий завершённый Build | 060.26 | +| Следующий Build | 060.27 | --- @@ -44,26 +44,25 @@ Market Data Acquisition ↓ Trades Feed / Time & Sales ↓ -Build 060.25 — Production Runtime Integration +Build 060.26 — Integration and Regression ↓ Completed ``` -060.25.0–060.25.8 приняты. Trade Stream: +060.26.0–060.26.6 приняты. Trade Stream: -- подключён к application bootstrap через отдельный feature flag; -- владеет WebSocket lifecycle; -- восстанавливает подписки и пропущенные Trades; -- сохраняет порядок Recovery → buffered Live; -- использует transport Ping/Pong liveness; -- детерминированно останавливает и ожидает owned tasks; -- прошёл 445 целевых тестов и полную регрессию из 1869 тестов. +- проверен через настоящие локальные WebSocket и HTTP соединения; +- прошёл reconnect, recovery, fault и cancellation scenarios; +- прошёл fixed stress, 120-секундный и 900-секундный soak; +- проверен через opt-in Dzengi live scenario; +- использует единый rollover-aware контракт signed 32-bit Trade ID; +- завершил полную offline-регрессию из 1930 тестов. Подробности: ```text -docs/migrations/build_060_25_architecture.md -docs/migrations/build_060_25.md +docs/migrations/build_060_26_architecture.md +docs/migrations/build_060_26.md ``` --- @@ -82,12 +81,11 @@ docs/migrations/build_060_25.md | 060.23 | Trade Stream Acquisition Integration | Completed | | 060.24 | Runtime Recovery Architecture | Completed | | 060.25 | Production Runtime Integration | Completed | +| 060.26 | Integration and Regression | Completed | -## Следующие Build +## Build 060.26 — Integration and Regression -### Build 060.26 — Integration and Regression - -**Статус:** Next +**Статус:** Completed Назначение: @@ -102,6 +100,8 @@ docs/migrations/build_060_25.md Build не должен превращать сетевые сценарии в обязательную часть обычного unit suite. +## Следующие Build + ### Build 060.27 — Persistent Market Data Storage **Статус:** Planned @@ -1730,6 +1730,6 @@ read-only архитектурного анализа. Старый ориент Актуальная контрольная точка: ```text -Завершён: Build 060.25 — Production Runtime Integration -Следующий: Build 060.26 — Integration and Regression +Завершён: Build 060.26 — Integration and Regression +Следующий: Build 060.27 — Persistent Market Data Storage ```