diff --git a/src/alphamind/execution/broker_adapter/__init__.py b/src/alphamind/execution/broker_adapter/__init__.py index 4f52ef496..4b7c31f17 100644 --- a/src/alphamind/execution/broker_adapter/__init__.py +++ b/src/alphamind/execution/broker_adapter/__init__.py @@ -25,6 +25,7 @@ (`ALP-121 `_). """ +from alphamind.execution.broker_adapter.bounded_streams import bounded_call from alphamind.execution.broker_adapter.client_factory import ( AlpacaClientFactory, ExecutionMode, @@ -150,6 +151,7 @@ "Submitted", "TouchQuote", "TradeAccountSnapshot", + "bounded_call", "build_occ_symbol", "classify_alpaca_error", "derive_capital_floor_client_order_id", diff --git a/src/alphamind/execution/broker_adapter/bounded_streams.py b/src/alphamind/execution/broker_adapter/bounded_streams.py new file mode 100644 index 000000000..cb70ad116 --- /dev/null +++ b/src/alphamind/execution/broker_adapter/bounded_streams.py @@ -0,0 +1,150 @@ +"""Bounded async facades over alpaca-py's synchronous seams (ALP-946). + +alpaca-py is synchronous in two distinct ways and both can freeze the calling +event loop: + +* **REST calls** issue blocking ``requests`` round-trips with no socket + timeout — a hung connection blocks the calling thread forever (the ALP-841 + wedge: a bare-sync ``get_orders`` froze the monitor loop and its watchdog + for ~4.5h). +* **Stream mutators** (``subscribe_quotes`` / ``unsubscribe_quotes`` / + ``subscribe_trade_updates``) are *loop-affined*: on a running stream they + marshal the actual send onto the stream's own event loop via + ``run_coroutine_threadsafe`` and block the calling thread on an untimed + ``Future.result()``. Called from that loop's own thread the scheduled + coroutine can never run — a guaranteed self-deadlock (the ALP-946 wedge: + every position-set-changing invocation froze the monitor for ~150 s). + +:func:`bounded_call` is the one safe calling convention for both: offloading +to a worker thread keeps the loop turning (and gives the loop-affined +mutators a free loop to marshal onto), and the :func:`asyncio.wait_for` +bound guarantees the await resolves even if the callable never returns — +raising :class:`TimeoutError` for the caller's reconnect/retry machinery. +The client factory's socket-level timeout remains defence-in-depth: it lets +the orphaned worker thread eventually unwind rather than leak. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any, Protocol + +# Wall-clock budget (seconds) for one vendor stream-mutator call run on a +# worker thread. Mirrors ``queries._REST_TIMEOUT_SECONDS`` (30s) and sits +# below the client factory's 60s socket-timeout floor, so this loop-level +# bound is the one that normally fires and a wedged send surfaces as a +# TimeoutError-driven reconnect rather than a frozen await. +_STREAM_CALL_TIMEOUT_SECONDS = 30.0 + + +async def bounded_call[T](fn: Callable[..., T], /, *args: object, timeout_seconds: float) -> T: + """Run a synchronous alpaca-py call off the event loop, time-bounded. + + Offloading to a worker thread keeps the calling loop running — for REST + calls that means the loop survives a hung socket (ALP-841); for the + loop-affined stream mutators it means the send they marshal back onto + the loop can actually drain (ALP-946). The :func:`asyncio.wait_for` + bound guarantees the await resolves within *timeout_seconds* even if the + callable never returns — raising :class:`TimeoutError`, which callers + treat as transient (retry on the next sweep interval, or tear down and + rebuild the stream). The client factory's socket-level timeout is the + matching thread-side floor so the orphaned worker eventually unwinds. + """ + return await asyncio.wait_for(asyncio.to_thread(fn, *args), timeout=timeout_seconds) + + +QuoteHandler = Callable[[Any], Awaitable[None]] + + +class _SyncStockDataStream(Protocol): + """The vendor ``StockDataStream`` surface the facade wraps. + + The mutators are the loop-affined synchronous APIs; ``_run_forever`` is + alpaca-py's documented async entry point (the underscore prefix is the + library's own convention) and ``stop_ws`` its async teardown. + """ + + def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None: ... + + def unsubscribe_quotes(self, *symbols: str) -> None: ... + + async def _run_forever(self) -> None: ... + + async def stop_ws(self) -> None: ... + + +class BoundedStockDataStream: + """Async facade over a vendor ``StockDataStream`` (ALP-946). + + The vendor mutators must run on a thread *other than* the stream's own + event-loop thread (see module docstring); the facade routes every call + through :func:`bounded_call` so callers can simply ``await`` from the + loop. ``_run_forever`` / ``stop_ws`` are already async — plain awaited + delegations. + """ + + def __init__( + self, + stream: _SyncStockDataStream, + *, + timeout_seconds: float = _STREAM_CALL_TIMEOUT_SECONDS, + ) -> None: + self._stream = stream + self._timeout_seconds = timeout_seconds + + async def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None: + await bounded_call( + self._stream.subscribe_quotes, handler, *symbols, timeout_seconds=self._timeout_seconds + ) + + async def unsubscribe_quotes(self, *symbols: str) -> None: + await bounded_call( + self._stream.unsubscribe_quotes, *symbols, timeout_seconds=self._timeout_seconds + ) + + async def _run_forever(self) -> None: + await self._stream._run_forever() + + async def stop_ws(self) -> None: + await self._stream.stop_ws() + + +class _SyncTradingStream(Protocol): + """The vendor ``TradingStream`` surface the facade wraps.""" + + def subscribe_trade_updates(self, handler: Any) -> None: ... + + async def _run_forever(self) -> None: ... + + async def stop_ws(self) -> None: ... + + +class BoundedTradingStream: + """Async facade over a vendor ``TradingStream`` (ALP-946). + + ``subscribe_trade_updates`` carries the same loop-affined blocking branch + as the stock-stream mutators (``alpaca/trading/stream.py``); the facade + routes it through :func:`bounded_call` so the fill stream's subscribe can + never block the event loop, however the stream's run state interleaves. + """ + + def __init__( + self, + stream: _SyncTradingStream, + *, + timeout_seconds: float = _STREAM_CALL_TIMEOUT_SECONDS, + ) -> None: + self._stream = stream + self._timeout_seconds = timeout_seconds + + async def subscribe_trade_updates(self, handler: Any) -> None: + await bounded_call( + self._stream.subscribe_trade_updates, handler, timeout_seconds=self._timeout_seconds + ) + + async def _run_forever(self) -> None: + await self._stream._run_forever() + + async def stop_ws(self) -> None: + await self._stream.stop_ws() diff --git a/src/alphamind/execution/broker_adapter/client_factory.py b/src/alphamind/execution/broker_adapter/client_factory.py index 42eaed4aa..0e83ce4c2 100644 --- a/src/alphamind/execution/broker_adapter/client_factory.py +++ b/src/alphamind/execution/broker_adapter/client_factory.py @@ -22,6 +22,7 @@ from alpaca.trading.stream import TradingStream from alphamind.config.models.venue import AlpacaCredentials, VenueConfig +from alphamind.execution.broker_adapter.bounded_streams import BoundedTradingStream ExecutionMode = Literal["paper", "live"] @@ -127,13 +128,20 @@ def build_trading_client(self) -> TradingClient: _install_socket_timeout(client) return client - def build_trading_stream(self) -> TradingStream: - """Return a fresh ``TradingStream`` bound to the resolved credentials.""" - return TradingStream( - api_key=self._credentials.api_key, - secret_key=self._credentials.api_secret, - paper=(self._credentials.mode == "paper"), - url_override=self._credentials.ws_url, + def build_trading_stream(self) -> BoundedTradingStream: + """Return a fresh ``TradingStream`` wrapped in the async facade. + + The vendor stream's ``subscribe_trade_updates`` is a loop-affined sync + API (ALP-946); the facade is the only surface handed out so no caller + can issue the bare blocking call on the event loop. + """ + return BoundedTradingStream( + TradingStream( + api_key=self._credentials.api_key, + secret_key=self._credentials.api_secret, + paper=(self._credentials.mode == "paper"), + url_override=self._credentials.ws_url, + ) ) def build_stock_data_client(self) -> StockHistoricalDataClient: diff --git a/src/alphamind/execution/broker_adapter/fill_stream.py b/src/alphamind/execution/broker_adapter/fill_stream.py index 152b922d8..03e6ae0e1 100644 --- a/src/alphamind/execution/broker_adapter/fill_stream.py +++ b/src/alphamind/execution/broker_adapter/fill_stream.py @@ -321,13 +321,18 @@ def _coerce_float(value: object) -> float: class _SubscribableStream(Protocol): - """Subset of ``alpaca.trading.stream.TradingStream`` we depend on. + """Async surface of the bounded trading-stream facade we depend on. Defined as a protocol so the test suite can swap a fake without inheriting - from the real (network-touching) class. + from the real (network-touching) class. The runtime object is a + :class:`~alphamind.execution.broker_adapter.bounded_streams.BoundedTradingStream` + wrapping the vendor stream. ``subscribe_trade_updates`` is ``async`` by + contract (ALP-946): alpaca-py's bare sync mutator is loop-affined — + calling it on the stream's own loop thread self-deadlocks the process + loop — so the seam only admits the bounded off-thread form. """ - def subscribe_trade_updates(self, handler: Any) -> None: ... + async def subscribe_trade_updates(self, handler: Any) -> None: ... async def _run_forever(self) -> None: ... @@ -378,7 +383,7 @@ async def subscribe_trade_updates( async def _handler(update: TradeUpdate) -> None: await queue.put(update) - stream.subscribe_trade_updates(_handler) + await stream.subscribe_trade_updates(_handler) # ``asyncio.TaskGroup`` is incompatible with async-generator cleanup: # ``gen.aclose()`` injects ``GeneratorExit`` into the body, which a # surrounding ``async with TaskGroup()`` re-raises as diff --git a/src/alphamind/execution/broker_adapter/queries.py b/src/alphamind/execution/broker_adapter/queries.py index f843b6c9e..f62f55d71 100644 --- a/src/alphamind/execution/broker_adapter/queries.py +++ b/src/alphamind/execution/broker_adapter/queries.py @@ -15,10 +15,9 @@ from __future__ import annotations -import asyncio import datetime as dt import functools -from collections.abc import AsyncGenerator, Callable +from collections.abc import AsyncGenerator from decimal import Decimal from typing import Any, Literal from zoneinfo import ZoneInfo @@ -46,6 +45,7 @@ from pydantic import BaseModel, ConfigDict, TypeAdapter from alphamind._kernel.money import Money, Price, money, price, signed_money +from alphamind.execution.broker_adapter.bounded_streams import bounded_call _ET = ZoneInfo("America/New_York") @@ -61,10 +61,10 @@ # endpoints cannot share ``_PAGE_SIZE`` — requesting 500 here is a hard 422 (ALP-932). _ACTIVITIES_PAGE_SIZE = 100 -# Per-page wall-clock budget (seconds) for a synchronous Alpaca REST call run on -# a worker thread. alpaca-py's ``TradingClient`` issues blocking ``requests`` -# calls with no socket timeout, so a hung connection would otherwise block its -# worker thread forever; bounding the ``to_thread`` await with ``wait_for`` +# Per-page wall-clock budget (seconds) for a synchronous Alpaca REST call run +# through ``bounded_call``. alpaca-py's ``TradingClient`` issues blocking +# ``requests`` calls with no socket timeout, so a hung connection would +# otherwise block its worker thread forever; the ``bounded_call`` bound # guarantees the *calling event loop* never freezes regardless of how long the # socket hangs (the ALP-841 lesson — a bare-sync ``get_orders`` froze the # monitor loop and its watchdog for ~4.5h). The client factory also installs a @@ -75,24 +75,6 @@ _REST_TIMEOUT_SECONDS = 30.0 -async def _rest_call[T](fn: Callable[[], T]) -> T: - """Run a synchronous Alpaca REST call off the event loop, time-bounded. - - The paginating query generators (:meth:`AccountStateQueries.get_orders`, - :meth:`~AccountStateQueries.get_account_activities`) are consumed from the - continuous monitor's recovery sweep and the account-activities poll — both - on an event loop shared with the fill stream and (pre-isolation) the safety - watchdog. alpaca-py is synchronous, so calling it inline would block that - loop for the full network round-trip; a *hung* call would freeze it - indefinitely (ALP-841). Offloading to a worker thread keeps the loop - running, and the :func:`asyncio.wait_for` bound guarantees the await resolves - within :data:`_REST_TIMEOUT_SECONDS` even if the socket never returns — - raising :class:`TimeoutError`, which the adapter's error classifier treats as - transient so the sweep retries on its next interval. - """ - return await asyncio.wait_for(asyncio.to_thread(fn), timeout=_REST_TIMEOUT_SECONDS) - - # --------------------------------------------------------------------------- # Typed response records — all frozen # --------------------------------------------------------------------------- @@ -545,7 +527,10 @@ async def get_orders( # a hung connection must not freeze the loop (ALP-841). ``partial`` # binds this page's request by value (the call is awaited before the # next iteration rebinds ``req``). - page_result = await _rest_call(functools.partial(self._client.get_orders, filter=req)) + page_result = await bounded_call( + functools.partial(self._client.get_orders, filter=req), + timeout_seconds=_REST_TIMEOUT_SECONDS, + ) if not isinstance(page_result, list) or not page_result: break page: list[Order] = page_result @@ -587,8 +572,9 @@ async def get_account_activities( # a hung connection must not freeze the loop (ALP-841). ``partial`` # binds this page's params by value (the call is awaited before the # next iteration mutates them). - result = await _rest_call( - functools.partial(self._client.get, "/account/activities", params) + result = await bounded_call( + functools.partial(self._client.get, "/account/activities", params), + timeout_seconds=_REST_TIMEOUT_SECONDS, ) if not isinstance(result, list) or not result: break diff --git a/src/alphamind/execution/broker_adapter/retry.py b/src/alphamind/execution/broker_adapter/retry.py index 96ba52bb9..be8c74cdc 100644 --- a/src/alphamind/execution/broker_adapter/retry.py +++ b/src/alphamind/execution/broker_adapter/retry.py @@ -22,6 +22,7 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass +from alphamind.execution.broker_adapter.bounded_streams import bounded_call from alphamind.execution.broker_adapter.errors import is_transient @@ -63,8 +64,8 @@ class GatewaySubmissionFailed: # client factory installs a matching 60s socket-level floor so an orphaned worker # eventually unwinds, but a single hung attempt would otherwise park the calling # task — including the continuous monitor's fire-close path — for the full socket -# timeout. Mirroring ``queries._rest_call`` (the read side), this ``wait_for`` -# is the event-loop-level bound that resolves the await within a tight budget and +# timeout. Mirroring ``queries._REST_TIMEOUT_SECONDS`` (the read side), this is +# the event-loop-level bound that resolves the await within a tight budget and # raises ``TimeoutError`` (classified transient → retried) — the ALP-841 lesson # applied to the write side. Below the 60s socket floor so this bound fires first. _SUBMIT_TIMEOUT_SECONDS = 30.0 @@ -73,13 +74,13 @@ class GatewaySubmissionFailed: async def bounded_broker_call[T](fn: Callable[[], T]) -> T: """Run a synchronous Alpaca write call off the event loop, time-bounded. - Offloads the blocking SDK call to a worker thread and bounds the await with - :func:`asyncio.wait_for` at :data:`_SUBMIT_TIMEOUT_SECONDS`, so a hung socket - cannot park the calling task (e.g. the monitor's fire-close submission) for - the full client-factory socket timeout. Raises :class:`TimeoutError` on - expiry, which :func:`submit_with_retry`'s transient classifier retries. + Delegates to :func:`~alphamind.execution.broker_adapter.bounded_streams.bounded_call` + at :data:`_SUBMIT_TIMEOUT_SECONDS`, so a hung socket cannot park the calling + task (e.g. the monitor's fire-close submission) for the full client-factory + socket timeout. Raises :class:`TimeoutError` on expiry, which + :func:`submit_with_retry`'s transient classifier retries. """ - return await asyncio.wait_for(asyncio.to_thread(fn), timeout=_SUBMIT_TIMEOUT_SECONDS) + return await bounded_call(fn, timeout_seconds=_SUBMIT_TIMEOUT_SECONDS) # Backoff schedule per the story: 0.5s, 1s, 2s, 4s, capped at half the diff --git a/src/alphamind/execution/continuous_monitor/safety_core/price_feed.py b/src/alphamind/execution/continuous_monitor/safety_core/price_feed.py index 544b1510b..d4f682739 100644 --- a/src/alphamind/execution/continuous_monitor/safety_core/price_feed.py +++ b/src/alphamind/execution/continuous_monitor/safety_core/price_feed.py @@ -27,6 +27,7 @@ from alpaca.data.historical import StockHistoricalDataClient from alpaca.data.requests import StockLatestQuoteRequest +from alphamind.execution.broker_adapter import bounded_call from alphamind.execution.continuous_monitor.underlying_stream.cache import ( UnderlyingPriceCache, UnderlyingQuote, @@ -106,9 +107,9 @@ class AlpacaLatestQuoteFetcher: Wraps a ``StockHistoricalDataClient`` (IEX feed, sharing the trading credentials, socket-timeout installed by the client factory). The synchronous - SDK call runs off the event loop via :func:`asyncio.to_thread`, time-bounded - by :func:`asyncio.wait_for` so a hung socket never wedges the poll loop. The - keyed response is mapped through the pure :func:`_latest_quotes_to_underlying`. + SDK call runs off the event loop via :func:`bounded_call` so a hung socket + never wedges the poll loop. The keyed response is mapped through the pure + :func:`_latest_quotes_to_underlying`. """ client: StockHistoricalDataClient @@ -117,8 +118,7 @@ class AlpacaLatestQuoteFetcher: async def __call__(self, symbols: frozenset[str]) -> list[UnderlyingQuote]: ordered = sorted(symbols) request = StockLatestQuoteRequest(symbol_or_symbols=ordered, feed=self.feed) - response = await asyncio.wait_for( - asyncio.to_thread(self.client.get_stock_latest_quote, request), - timeout=_FETCH_TIMEOUT_SECONDS, + response = await bounded_call( + self.client.get_stock_latest_quote, request, timeout_seconds=_FETCH_TIMEOUT_SECONDS ) return _latest_quotes_to_underlying(response, ordered) diff --git a/src/alphamind/execution/continuous_monitor/underlying_stream/task.py b/src/alphamind/execution/continuous_monitor/underlying_stream/task.py index dd2c88542..8c184c6c9 100644 --- a/src/alphamind/execution/continuous_monitor/underlying_stream/task.py +++ b/src/alphamind/execution/continuous_monitor/underlying_stream/task.py @@ -40,13 +40,14 @@ import logging import os import time -from collections.abc import Awaitable, Callable +from collections.abc import Callable from datetime import UTC, datetime from typing import Any, Protocol, runtime_checkable from alphamind._kernel.exception_group import first_non_cancelled from alphamind.config.models.continuous_monitor import ContinuousMonitorConfig from alphamind.execution.broker_adapter import StreamActivityMonitor, StreamStalledError +from alphamind.execution.broker_adapter.bounded_streams import QuoteHandler from alphamind.execution.continuous_monitor.session import MonitorMode, MonitorSession from alphamind.execution.continuous_monitor.underlying_stream.cache import ( UnderlyingPriceCache, @@ -65,21 +66,22 @@ # --------------------------------------------------------------------------- -QuoteHandler = Callable[[Any], Awaitable[None]] - - @runtime_checkable class StockDataStreamProtocol(Protocol): - """Subset of ``alpaca.data.live.stock.StockDataStream`` this task uses. + """Async surface of the bounded stock-stream facade this task uses. Defined as a Protocol so the test suite can swap a fake without - inheriting from alpaca-py's network-touching class. The runtime real - object satisfies this surface by construction. + inheriting from alpaca-py's network-touching class. The runtime object + is a :class:`~alphamind.execution.broker_adapter.bounded_streams.BoundedStockDataStream` + wrapping the vendor stream. The mutators are ``async`` by contract + (ALP-946): alpaca-py's bare sync mutators are loop-affined — calling one + on the stream's own loop thread self-deadlocks the whole process loop — + so the seam only admits the bounded off-thread form. """ - def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None: ... + async def subscribe_quotes(self, handler: QuoteHandler, *symbols: str) -> None: ... - def unsubscribe_quotes(self, *symbols: str) -> None: ... + async def unsubscribe_quotes(self, *symbols: str) -> None: ... async def _run_forever(self) -> None: ... @@ -118,6 +120,8 @@ def build(self, *, mode: str) -> StockDataStreamProtocol: from alpaca.data.enums import DataFeed from alpaca.data.live.stock import StockDataStream + from alphamind.execution.broker_adapter.bounded_streams import BoundedStockDataStream + if mode == "live": key_env, secret_env = "ALPACA_LIVE_KEY", "ALPACA_LIVE_SECRET" else: @@ -131,10 +135,12 @@ def build(self, *, mode: str) -> StockDataStreamProtocol: f"{missing} are unset or empty" ) raise RuntimeError(msg) - return StockDataStream( - api_key=api_key, - secret_key=api_secret, - feed=DataFeed.IEX, + return BoundedStockDataStream( + StockDataStream( + api_key=api_key, + secret_key=api_secret, + feed=DataFeed.IEX, + ) ) @@ -321,11 +327,12 @@ async def _handler(payload: Any) -> None: # Subscribe to the initial target set. current: frozenset[str] = await compute_target_underlyings(repository) if current: - stream.subscribe_quotes(_handler, *sorted(current)) + await stream.subscribe_quotes(_handler, *sorted(current)) else: - # ``subscribe_quotes`` with no symbols still registers the handler so - # late-arriving subscribe calls reuse it. - stream.subscribe_quotes(_handler) + # With no symbols the vendor registers nothing; ``_run_forever`` then + # waits (by design) until the first diff delta registers a symbol, at + # which point its connect-time subscribe message carries the full set. + await stream.subscribe_quotes(_handler) # ``_run_forever`` is alpaca-py's documented async entry point — the # underscore-prefixed name is the library's own convention, not a @@ -417,9 +424,9 @@ async def _periodic_subscription_diff( added = target - live removed = live - target if added: - stream.subscribe_quotes(handler, *sorted(added)) + await stream.subscribe_quotes(handler, *sorted(added)) if removed: - stream.unsubscribe_quotes(*sorted(removed)) + await stream.unsubscribe_quotes(*sorted(removed)) live = set(target) diff --git a/tests/execution/broker_adapter/test_bounded_streams.py b/tests/execution/broker_adapter/test_bounded_streams.py new file mode 100644 index 000000000..d863227f4 --- /dev/null +++ b/tests/execution/broker_adapter/test_bounded_streams.py @@ -0,0 +1,158 @@ +"""Tests for ``broker_adapter.bounded_streams`` (ALP-946). + +alpaca-py's stream-mutator methods (``subscribe_quotes`` / +``unsubscribe_quotes`` / ``subscribe_trade_updates``) are loop-affined +synchronous APIs: on a running stream they marshal the send onto the stream's +own event loop via ``run_coroutine_threadsafe`` and block the calling thread +on an untimed ``Future.result()``. Called from the loop's own thread that is +a guaranteed self-deadlock — the production wedge behind ALP-946. + +``bounded_call`` and the facade classes make the safe calling convention the +only one expressible: every vendor sync call runs on a worker thread (so the +scheduled coroutine can drain on the free loop) under an ``asyncio.wait_for`` +bound (so the await always resolves). These tests pin the loop-affinity +contract — off-loop-thread execution with identical arguments — so no +per-call-site thread assertion is needed elsewhere. +""" + +from __future__ import annotations + +import threading +from collections.abc import Awaitable, Callable +from typing import Any + +import pytest + +from alphamind.execution.broker_adapter import bounded_call +from alphamind.execution.broker_adapter.bounded_streams import ( + BoundedStockDataStream, + BoundedTradingStream, +) + + +class _RecordingVendorStockStream: + """Fake vendor ``StockDataStream`` recording call thread ids + arguments.""" + + def __init__(self) -> None: + self.subscribe_calls: list[tuple[int, tuple[object, ...]]] = [] + self.unsubscribe_calls: list[tuple[int, tuple[str, ...]]] = [] + self.run_forever_called = False + self.stop_ws_called = False + + def subscribe_quotes(self, handler: Callable[[Any], Awaitable[None]], *symbols: str) -> None: + self.subscribe_calls.append((threading.get_ident(), (handler, *symbols))) + + def unsubscribe_quotes(self, *symbols: str) -> None: + self.unsubscribe_calls.append((threading.get_ident(), symbols)) + + async def _run_forever(self) -> None: + self.run_forever_called = True + + async def stop_ws(self) -> None: + self.stop_ws_called = True + + +class TestBoundedCall: + async def test_runs_callable_off_loop_thread_passing_args_and_return(self) -> None: + loop_thread_id = threading.get_ident() + seen: dict[str, object] = {} + + def record(*args: object) -> str: + seen["thread_id"] = threading.get_ident() + seen["args"] = args + return "payload" + + result = await bounded_call(record, "AAPL", 7, timeout_seconds=5.0) + + assert result == "payload" + assert seen["args"] == ("AAPL", 7) + assert seen["thread_id"] != loop_thread_id + + async def test_raises_timeout_error_when_callable_blocks_past_bound(self) -> None: + release = threading.Event() + + def blocks_until_released() -> None: + release.wait() + + try: + with pytest.raises(TimeoutError): + await bounded_call(blocks_until_released, timeout_seconds=0.05) + finally: + # Unwind the orphaned worker thread so it doesn't outlive the test. + release.set() + + +class TestBoundedStockDataStream: + async def test_subscribe_quotes_invokes_vendor_off_loop_thread_with_args(self) -> None: + vendor = _RecordingVendorStockStream() + facade = BoundedStockDataStream(vendor) + + async def handler(_payload: Any) -> None: ... + + await facade.subscribe_quotes(handler, "SPY", "AAPL") + + ((thread_id, args),) = vendor.subscribe_calls + assert thread_id != threading.get_ident() + assert args == (handler, "SPY", "AAPL") + + async def test_unsubscribe_quotes_invokes_vendor_off_loop_thread_with_args(self) -> None: + vendor = _RecordingVendorStockStream() + facade = BoundedStockDataStream(vendor) + + await facade.unsubscribe_quotes("SPY", "AAPL") + + ((thread_id, symbols),) = vendor.unsubscribe_calls + assert thread_id != threading.get_ident() + assert symbols == ("SPY", "AAPL") + + async def test_run_forever_and_stop_ws_delegate(self) -> None: + vendor = _RecordingVendorStockStream() + facade = BoundedStockDataStream(vendor) + + await facade._run_forever() + await facade.stop_ws() + + assert vendor.run_forever_called + assert vendor.stop_ws_called + + +class _RecordingVendorTradingStream: + """Fake vendor ``TradingStream`` recording call thread ids + arguments.""" + + def __init__(self) -> None: + self.subscribe_calls: list[tuple[int, object]] = [] + self.run_forever_called = False + self.stop_ws_called = False + + def subscribe_trade_updates(self, handler: Any) -> None: + self.subscribe_calls.append((threading.get_ident(), handler)) + + async def _run_forever(self) -> None: + self.run_forever_called = True + + async def stop_ws(self) -> None: + self.stop_ws_called = True + + +class TestBoundedTradingStream: + async def test_subscribe_trade_updates_invokes_vendor_off_loop_thread(self) -> None: + vendor = _RecordingVendorTradingStream() + facade = BoundedTradingStream(vendor) + + async def handler(_update: Any) -> None: ... + + await facade.subscribe_trade_updates(handler) + + ((thread_id, seen_handler),) = vendor.subscribe_calls + assert thread_id != threading.get_ident() + assert seen_handler is handler + + async def test_run_forever_and_stop_ws_delegate(self) -> None: + vendor = _RecordingVendorTradingStream() + facade = BoundedTradingStream(vendor) + + await facade._run_forever() + await facade.stop_ws() + + assert vendor.run_forever_called + assert vendor.stop_ws_called diff --git a/tests/execution/broker_adapter/test_client_factory.py b/tests/execution/broker_adapter/test_client_factory.py index bb98b622e..e61f86ab3 100644 --- a/tests/execution/broker_adapter/test_client_factory.py +++ b/tests/execution/broker_adapter/test_client_factory.py @@ -174,10 +174,12 @@ def test_build_trading_client_returns_live_client_when_mode_live( assert "paper" not in str(base_url) -def test_build_trading_stream_returns_stream_instance( +def test_build_trading_stream_returns_bounded_facade_over_vendor_stream( monkeypatch: pytest.MonkeyPatch, venue: VenueConfig ) -> None: + """The factory hands out the async facade, never a bare vendor stream (ALP-946).""" from alphamind.execution.broker_adapter import AlpacaClientFactory + from alphamind.execution.broker_adapter.bounded_streams import BoundedTradingStream monkeypatch.setenv("ALPACA_PAPER_KEY", "paper-key") monkeypatch.setenv("ALPACA_PAPER_SECRET", "paper-secret") @@ -185,7 +187,8 @@ def test_build_trading_stream_returns_stream_instance( factory = AlpacaClientFactory(venue, mode="paper") stream = factory.build_trading_stream() - assert isinstance(stream, TradingStream) + assert isinstance(stream, BoundedTradingStream) + assert isinstance(stream._stream, TradingStream) def test_build_corporate_actions_client_returns_paper_client( diff --git a/tests/execution/broker_adapter/test_fill_stream.py b/tests/execution/broker_adapter/test_fill_stream.py index 081ae7a6a..1bd534f8a 100644 --- a/tests/execution/broker_adapter/test_fill_stream.py +++ b/tests/execution/broker_adapter/test_fill_stream.py @@ -445,7 +445,7 @@ def __init__(self) -> None: self.run_cancelled = False self._park = asyncio.Event() - def subscribe_trade_updates(self, handler: Any) -> None: + async def subscribe_trade_updates(self, handler: Any) -> None: self.handler = handler async def inject(self, update: TradeUpdate) -> None: @@ -594,7 +594,7 @@ async def test_stream_task_oserror_unblocks_consumer_and_propagates(self) -> Non """ class _ErrorStream: - def subscribe_trade_updates(self, handler: Any) -> None: + async def subscribe_trade_updates(self, handler: Any) -> None: pass # no real events — the error fires before any arrive async def _run_forever(self) -> None: @@ -624,7 +624,7 @@ async def test_stream_task_clean_exit_unblocks_consumer_without_error(self) -> N """ class _CleanStream: - def subscribe_trade_updates(self, handler: Any) -> None: + async def subscribe_trade_updates(self, handler: Any) -> None: pass async def _run_forever(self) -> None: diff --git a/tests/execution/continuous_monitor/fill_stream_consumer/test_task.py b/tests/execution/continuous_monitor/fill_stream_consumer/test_task.py index f8c7ffc90..a67475c83 100644 --- a/tests/execution/continuous_monitor/fill_stream_consumer/test_task.py +++ b/tests/execution/continuous_monitor/fill_stream_consumer/test_task.py @@ -182,7 +182,7 @@ def __init__(self) -> None: self.handler: Any = None self.run_cancelled = False - def subscribe_trade_updates(self, handler: Any) -> None: + async def subscribe_trade_updates(self, handler: Any) -> None: self.handler = handler async def inject(self, update: TradeUpdate) -> None: diff --git a/tests/execution/continuous_monitor/test_paper_enrichment_end_to_end.py b/tests/execution/continuous_monitor/test_paper_enrichment_end_to_end.py index 8e8658bcb..3504fb079 100644 --- a/tests/execution/continuous_monitor/test_paper_enrichment_end_to_end.py +++ b/tests/execution/continuous_monitor/test_paper_enrichment_end_to_end.py @@ -162,7 +162,7 @@ class _FakeStream: def __init__(self) -> None: self.handler: Any = None - def subscribe_trade_updates(self, handler: Any) -> None: + async def subscribe_trade_updates(self, handler: Any) -> None: self.handler = handler async def inject(self, update: TradeUpdate) -> None: diff --git a/tests/execution/continuous_monitor/underlying_stream/test_task.py b/tests/execution/continuous_monitor/underlying_stream/test_task.py index 7f84e52ab..575161ab3 100644 --- a/tests/execution/continuous_monitor/underlying_stream/test_task.py +++ b/tests/execution/continuous_monitor/underlying_stream/test_task.py @@ -153,14 +153,16 @@ def __init__(self) -> None: self._run_started = asyncio.Event() self._stop_run = asyncio.Event() - def subscribe_quotes(self, handler: Callable[[Any], Awaitable[None]], *symbols: str) -> None: + async def subscribe_quotes( + self, handler: Callable[[Any], Awaitable[None]], *symbols: str + ) -> None: # alpaca-py uses the handler from the first ``subscribe_quotes`` call. if self._handler is None: self._handler = handler self.subscribed.update(symbols) self.subscribe_calls.append(tuple(symbols)) - def unsubscribe_quotes(self, *symbols: str) -> None: + async def unsubscribe_quotes(self, *symbols: str) -> None: self.subscribed.difference_update(symbols) self.unsubscribe_calls.append(tuple(symbols)) @@ -193,16 +195,18 @@ class _FakeFactory: """Yields a fresh ``_FakeStream`` on each ``build`` call. Reconnect cycles create a new ``_FakeStream``; tests can inspect the - full sequence via ``streams``. + full sequence via ``streams``. ``stream_cls`` lets a test substitute a + misbehaving stream variant. """ - def __init__(self) -> None: + def __init__(self, stream_cls: type[_FakeStream] = _FakeStream) -> None: self.streams: list[_FakeStream] = [] self.build_calls: list[str] = [] # mode tags for assertions + self._stream_cls = stream_cls def build(self, *, mode: str) -> _FakeStream: self.build_calls.append(mode) - stream = _FakeStream() + stream = self._stream_cls() self.streams.append(stream) return stream @@ -427,6 +431,69 @@ async def test_removed_position_triggers_unsubscribe_quotes(self) -> None: await asyncio.gather(task, return_exceptions=True) +class TestFacadeTimeoutReconnect: + """ALP-946 — a facade ``TimeoutError`` on a diff delta rides the existing + budget-counted reconnect, and the rebuilt connection re-subscribes the + full target set (healing any half-applied subscription state).""" + + async def test_timeout_on_subscribe_delta_reconnects_and_resubscribes_full_set( + self, caplog: pytest.LogCaptureFixture + ) -> None: + class _TimeoutOnDeltaStream(_FakeStream): + """Times out the AAPL delta; connect-time full-set calls succeed.""" + + async def subscribe_quotes( + self, handler: Callable[[Any], Awaitable[None]], *symbols: str + ) -> None: + if symbols == ("AAPL",): + raise TimeoutError + await super().subscribe_quotes(handler, *symbols) + + reader = _MutableReader( + (_equity_position(position_id=PositionId("p1"), ticker=Symbol("SPY")),) + ) + factory = _FakeFactory(stream_cls=_TimeoutOnDeltaStream) + cache = UnderlyingPriceCache() + task = asyncio.create_task( + run_underlying_stream( + _session(), + _config(subscription_refresh_seconds=1, max_reconnect_attempts=3), + repository=reader, + cache=cache, + factory=factory, + ) + ) + try: + await asyncio.wait_for(_eventually(lambda: bool(factory.streams)), timeout=1.0) + stream1 = factory.streams[0] + await asyncio.wait_for( + _eventually(lambda: stream1.subscribed == {"SPY"}), + timeout=1.0, + ) + # A new position makes the next diff issue the fatal AAPL delta. + reader.set_positions( + ( + _equity_position(position_id=PositionId("p1"), ticker=Symbol("SPY")), + _equity_position(position_id=PositionId("p2"), ticker=Symbol("AAPL")), + ) + ) + # The TimeoutError tears down connection 1 and a fresh stream is built. + await asyncio.wait_for(_eventually(lambda: len(factory.streams) >= 2), timeout=5.0) + stream2 = factory.streams[1] + # The rebuilt connection subscribes the FULL current target set at + # connect time — not just the failed delta. + await asyncio.wait_for( + _eventually(lambda: ("AAPL", "SPY") in stream2.subscribe_calls), + timeout=2.0, + ) + # Budget-counted (not budget-neutral): the disconnect branch logged + # the remaining-attempts countdown, not the budget-neutral refresh. + assert any("reconnect attempts remaining" in rec.message for rec in caplog.records) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + class TestReconnect: async def test_disconnect_triggers_reconnect_and_cache_survives(self) -> None: reader = _MutableReader(