From 299edfdc627f36d70f765d7435103cdb454d890d Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Thu, 11 Jun 2026 18:35:43 -0600 Subject: [PATCH 1/2] fix(execution): bound alpaca-py websocket mutators behind async facades (ALP-946) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit alpaca-py's stream mutators (subscribe_quotes / unsubscribe_quotes / subscribe_trade_updates) are loop-affined sync APIs: on a running stream they marshal the send onto the stream's own event loop and block the calling thread on an untimed Future.result(). Called from the monitor's own loop thread (the periodic subscription diff) that is a guaranteed self-deadlock — four production wedges, one per position-set-changing invocation since 2026-06-10. - bounded_streams.bounded_call: shared off-thread + wait_for-bounded offload, generalizing queries._rest_call (deleted); the safety-core quote fetcher collapses onto it. - BoundedStockDataStream / BoundedTradingStream: async facades routing every vendor mutator through bounded_call; the stock-stream factory and client_factory.build_trading_stream() now hand out only facades. - StockDataStreamProtocol / _SubscribableStream mutators are async by contract, so a bare sync call on the loop is unrepresentable at the seam; all call sites await. - A facade TimeoutError from the diff path rides the existing budget-counted reconnect; the rebuilt connection re-subscribes the full target set (pinned by a new test). Deviation from the issue's literal signature: bounded_call's keyword is timeout_seconds (not timeout) — ruff ASYNC109 flags async defs with a `timeout` parameter, and the facades' constructor keyword in the spec already uses timeout_seconds. Co-Authored-By: Claude Fable 5 --- .../execution/broker_adapter/__init__.py | 2 + .../broker_adapter/bounded_streams.py | 145 +++++++++++++++++ .../broker_adapter/client_factory.py | 22 ++- .../execution/broker_adapter/fill_stream.py | 13 +- .../execution/broker_adapter/queries.py | 32 ++-- .../safety_core/price_feed.py | 12 +- .../underlying_stream/task.py | 34 ++-- .../broker_adapter/test_bounded_streams.py | 152 ++++++++++++++++++ .../broker_adapter/test_client_factory.py | 7 +- .../broker_adapter/test_fill_stream.py | 6 +- .../fill_stream_consumer/test_task.py | 2 +- .../test_paper_enrichment_end_to_end.py | 2 +- .../underlying_stream/test_task.py | 77 ++++++++- 13 files changed, 441 insertions(+), 65 deletions(-) create mode 100644 src/alphamind/execution/broker_adapter/bounded_streams.py create mode 100644 tests/execution/broker_adapter/test_bounded_streams.py 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..5d6d2aa28 --- /dev/null +++ b/src/alphamind/execution/broker_adapter/bounded_streams.py @@ -0,0 +1,145 @@ +"""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: ... + + +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() 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..1b02f8bf9 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") @@ -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/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..b24250525 100644 --- a/src/alphamind/execution/continuous_monitor/underlying_stream/task.py +++ b/src/alphamind/execution/continuous_monitor/underlying_stream/task.py @@ -70,16 +70,20 @@ @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 +122,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 +137,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 +329,11 @@ 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) + 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 +425,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..444817e1a --- /dev/null +++ b/tests/execution/broker_adapter/test_bounded_streams.py @@ -0,0 +1,152 @@ +"""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 + + 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 + + +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_delegates(self) -> None: + vendor = _RecordingVendorTradingStream() + facade = BoundedTradingStream(vendor) + + await facade._run_forever() + + assert vendor.run_forever_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( From 42ed4b35a7d777989a284b064c1255d84c43dff7 Mon Sep 17 00:00:00 2001 From: Jackson Atassi Date: Thu, 11 Jun 2026 18:48:01 -0600 Subject: [PATCH 2/2] fix: address code-review findings on ALP-946 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BoundedTradingStream gains stop_ws delegation (vendor has it; the stock facade exposes it — asymmetry would AttributeError a future teardown path) - task.py imports QuoteHandler from bounded_streams instead of redefining it - retry.bounded_broker_call delegates to bounded_call (same offload idiom) - corrected the false no-symbol subscribe comment (vendor registers nothing; _run_forever waits for the first delta, connect-time send carries the set) - stale _rest_call / wait_for mechanism comments now reference bounded_call Co-Authored-By: Claude Fable 5 --- .../execution/broker_adapter/bounded_streams.py | 5 +++++ .../execution/broker_adapter/queries.py | 8 ++++---- src/alphamind/execution/broker_adapter/retry.py | 17 +++++++++-------- .../underlying_stream/task.py | 11 +++++------ .../broker_adapter/test_bounded_streams.py | 8 +++++++- 5 files changed, 30 insertions(+), 19 deletions(-) diff --git a/src/alphamind/execution/broker_adapter/bounded_streams.py b/src/alphamind/execution/broker_adapter/bounded_streams.py index 5d6d2aa28..cb70ad116 100644 --- a/src/alphamind/execution/broker_adapter/bounded_streams.py +++ b/src/alphamind/execution/broker_adapter/bounded_streams.py @@ -117,6 +117,8 @@ 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). @@ -143,3 +145,6 @@ async def subscribe_trade_updates(self, handler: Any) -> None: 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/queries.py b/src/alphamind/execution/broker_adapter/queries.py index 1b02f8bf9..f62f55d71 100644 --- a/src/alphamind/execution/broker_adapter/queries.py +++ b/src/alphamind/execution/broker_adapter/queries.py @@ -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 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/underlying_stream/task.py b/src/alphamind/execution/continuous_monitor/underlying_stream/task.py index b24250525..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,9 +66,6 @@ # --------------------------------------------------------------------------- -QuoteHandler = Callable[[Any], Awaitable[None]] - - @runtime_checkable class StockDataStreamProtocol(Protocol): """Async surface of the bounded stock-stream facade this task uses. @@ -331,8 +329,9 @@ async def _handler(payload: Any) -> None: if 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. + # 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 diff --git a/tests/execution/broker_adapter/test_bounded_streams.py b/tests/execution/broker_adapter/test_bounded_streams.py index 444817e1a..d863227f4 100644 --- a/tests/execution/broker_adapter/test_bounded_streams.py +++ b/tests/execution/broker_adapter/test_bounded_streams.py @@ -122,6 +122,7 @@ class _RecordingVendorTradingStream: 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)) @@ -129,6 +130,9 @@ def subscribe_trade_updates(self, handler: Any) -> None: 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: @@ -143,10 +147,12 @@ async def handler(_update: Any) -> None: ... assert thread_id != threading.get_ident() assert seen_handler is handler - async def test_run_forever_delegates(self) -> None: + 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