diff --git a/piker/brokers/ib/feed.py b/piker/brokers/ib/feed.py index adeb6361..f5ec0bf3 100644 --- a/piker/brokers/ib/feed.py +++ b/piker/brokers/ib/feed.py @@ -32,7 +32,6 @@ import time from typing import ( Any, Callable, - TYPE_CHECKING, ) from async_generator import aclosing @@ -75,9 +74,6 @@ from ._util import ( ) from .symbols import get_mkt_info -if TYPE_CHECKING: - from trio._core._run import Task - log = get_logger( name=__name__, ) @@ -353,6 +349,7 @@ async def wait_on_data_reset( tuple[ trio.CancelScope, trio.Event, + list[bool], ] ] = trio.TASK_STATUS_IGNORED, ) -> bool: @@ -396,8 +393,9 @@ async def wait_on_data_reset( client: Client = proxy._aio_ns done = trio.Event() + result: list[bool] = [] with trio.move_on_after(timeout) as cs: - task_status.started((cs, done)) + task_status.started((cs, done, result)) log.warning( 'Sending DATA RESET request:\n' @@ -413,6 +411,7 @@ async def wait_on_data_reset( 'NO VNC DETECTED!\n' 'Manually press ctrl-alt-f on your IB java app' ) + result.append(False) done.set() return False @@ -427,6 +426,7 @@ async def wait_on_data_reset( ]: await ev.wait() log.info(f"{name} DATA RESET") + result.append(True) done.set() return True @@ -435,12 +435,12 @@ async def wait_on_data_reset( 'Data reset task canceled?' ) + result.append(False) done.set() return False -_data_resetter_task: Task | None = None -_failed_resets: int = 0 +_data_reset_lock = trio.Lock() async def get_bars( @@ -479,8 +479,11 @@ async def get_bars( `.ib.api.Client` methods. ''' - global _data_resetter_task, _failed_resets nodatas_count: int = 0 + cancelled_count: int = 0 + failed_resets: int = 0 + pacing_reset_pending: bool = False + pacing_reset_completed: bool = False data_cs: trio.CancelScope | None = None result: tuple[ @@ -493,10 +496,16 @@ async def get_bars( async def query(): - global _failed_resets - nonlocal result, data_cs, end_dt, nodatas_count + nonlocal result, data_cs, end_dt + nonlocal nodatas_count, cancelled_count, failed_resets + nonlocal pacing_reset_pending, pacing_reset_completed - while _failed_resets < max_failed_resets: + while True: + if failed_resets >= max_failed_resets: + raise DataUnavailable( + f'IB history reset failed {failed_resets} ' + f'times for {fqme}' + ) try: ( bars, @@ -528,6 +537,10 @@ async def get_bars( ) # NOTE: REQUIRED to pass back value.. result = None + failed_resets = 0 + result_ready.set() + if data_cs: + data_cs.cancel() return None # not enough bars signal, likely due to venue @@ -587,6 +600,7 @@ async def get_bars( first_dt, last_dt, ) + failed_resets = 0 # signal data reset loop parent task result_ready.set() @@ -642,6 +656,12 @@ async def get_bars( 'Query cancelled by IB (:eyeroll:):\n' f'{err.message}' ) + cancelled_count += 1 + if cancelled_count >= max_failed_resets: + raise DataUnavailable( + f'IB cancelled {cancelled_count} history ' + f'queries for {fqme}' + ) continue elif ( @@ -662,6 +682,7 @@ async def get_bars( ) client = proxy._aio_ns.ib.client + pacing_reset_pending = True # cancel any existing reset task if data_cs: @@ -669,69 +690,81 @@ async def get_bars( data_cs.cancel() # spawn new data reset task - data_cs, reset_done = await tn.start( - partial( - wait_on_data_reset, - proxy, - reset_type='connection' + async with _data_reset_lock: + ( + data_cs, + reset_done, + reset_result, + ) = await tn.start( + partial( + wait_on_data_reset, + proxy, + reset_type='connection' + ) ) - ) - if reset_done: - _failed_resets = 0 + await reset_done.wait() + if reset_result[0]: + failed_resets = 0 else: - _failed_resets += 1 + failed_resets += 1 + pacing_reset_pending = False + pacing_reset_completed = True continue else: raise - # TODO: make this global across all history task/requests - # such that simultaneous symbol queries don't try data resettingn - # too fast.. - unset_resetter: bool = False async with ( tractor.trionics.collapse_eg(), trio.open_nursery() as tn ): - # start history request that we allow - # to run indefinitely until a result is acquired + # Run history until a result or bounded reset failure. tn.start_soon(query) - # start history reset loop which waits up to the timeout - # for a result before triggering a data feed reset. + # Serialize farm resets across every history requester. while not result_ready.is_set(): - with trio.move_on_after(feed_reset_timeout): await result_ready.wait() break - if _data_resetter_task: - # don't double invoke the reset hack if another - # requester task already has it covered. - continue + async with _data_reset_lock: + if result_ready.is_set(): + break + if pacing_reset_completed: + pacing_reset_completed = False + continue - else: - _data_resetter_task = trio.lowlevel.current_task() - unset_resetter: bool = True - - # spawn new data reset task - data_cs, reset_done = await tn.start( - partial( - wait_on_data_reset, - proxy, - reset_type='data', + ( + data_cs, + reset_done, + reset_result, + ) = await tn.start( + partial( + wait_on_data_reset, + proxy, + reset_type='data', + ) ) - ) - # sync wait on reset to complete - await reset_done.wait() + await reset_done.wait() - _data_resetter_task = ( - None - if unset_resetter - else _data_resetter_task - ) - assert result + if ( + result_ready.is_set() + or + reset_result[0] + ): + failed_resets = 0 + elif pacing_reset_pending: + # Pacing recovery deliberately cancelled this reset + # and will consume the retry itself. + pass + else: + failed_resets += 1 + if failed_resets >= max_failed_resets: + raise DataUnavailable( + f'IB history reset failed ' + f'{failed_resets} times for {fqme}' + ) return ( result, data_cs is not None, diff --git a/tests/test_ib_history.py b/tests/test_ib_history.py new file mode 100644 index 00000000..a127d726 --- /dev/null +++ b/tests/test_ib_history.py @@ -0,0 +1,516 @@ +''' +IB history adapter regressions backported from 0846cbd4. + +''' +from types import SimpleNamespace + +from ib_async import RequestError +import numpy as np +from pendulum import ( + duration, +) +import pytest +import trio + +from piker.brokers import DataUnavailable +from piker.brokers.ib import feed as ib_feed +from piker.brokers.ib.feed import get_bars + + +def test_empty_history_frame_completes() -> None: + ''' + A blank IB response must release the history request slot. + + ``get_bars()`` previously returned from its query task without + signalling the parent, which then reset the feed forever. That + kept serialized SHM repair from running. Return IB's empty + response and prove the request completes explicitly + with no result instead of relying on a reset timeout. + + ''' + class Proxy: + ''' + Return one empty IB history response. + + ''' + async def bars(self, **kwargs): + ''' + Model ``Client.bars()`` during a venue data gap. + + ''' + return ( + [], + np.empty(0), + duration(seconds=2000), + ) + + async def main() -> None: + with trio.fail_after(0.1): + result, timedout = await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + ) + assert result is None + assert not timedout + + trio.run(main) + + +def test_repeated_ib_cancellation_is_bounded() -> None: + ''' + Repeated IB cancellation must not monopolize history forever. + + IB can answer a historical request with error 162 cancellation. + The old retry loop had no limit, so reverse backfill could retain + the backend's sole request slot and prevent final null repair. + Always raise that response and prove the configured retry bound + terminates with ``DataUnavailable`` deterministically. + + ''' + class Proxy: + ''' + Cancel every IB history request. + + ''' + async def bars(self, **kwargs): + ''' + Raise IB's historical-query cancellation response. + + ''' + raise RequestError( + 1, + 162, + 'API historical data query cancelled', + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=2, + ) + + trio.run(main) + + +def test_query_error_cancels_stalled_reset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Exceptional history exit must cancel its in-flight reset task. + + A timed-out request acquires the reset lock before launching + IB's VNC reset task. If the query then raises while that task is + blocked, stale work must not retain the reset lock. Force that + ordering with explicit Trio checkpoints, then prove bounded query + failure cancels the reset task before propagating. + + ''' + reset_cancelled: list[bool] = [] + + class Proxy: + ''' + Delay once, then return IB cancellation. + + ''' + async def bars(self, **kwargs): + ''' + Let the parent acquire the reset lock before failing. + + ''' + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + 'API historical data query cancelled', + ) + + async def stalled_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Publish reset state, then block until sibling failure. + + ''' + done = trio.Event() + result: list[bool] = [] + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + try: + await trio.sleep_forever() + finally: + reset_cancelled.append(True) + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + stalled_reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=1, + ) + + trio.run(main) + assert reset_cancelled == [True] + + +def test_failed_pacing_resets_reach_retry_limit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Failed pacing resets must increment the request's retry count. + + The reset API returns an ``Event`` and result box. The old + code tested the always-truthy event and reset the failure counter + after each failure. Return immediate failures and prove two + pacing responses terminate instead of looping forever. + + ''' + reset_calls: list[bool] = [] + + class Proxy: + ''' + Return a pacing violation for every request. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Raise IB's pacing response. + + ''' + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def failed_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Publish one completed, unsuccessful reset. + + ''' + done = trio.Event() + result = [False] + reset_calls.append(True) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + failed_reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=2, + ) + + trio.run(main) + assert reset_calls == [True, True] + + +def test_pacing_resets_are_serialized_across_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Concurrent history requests must never reset IB farms together. + + Pacing recovery bypassed reset ownership, allowing 1s and 60s + requests to launch the VNC hack concurrently. Start two pacing + failures in one nursery and hold each fake reset across a Trio + checkpoint. Active-count assertions prove the actor-global lock + serializes reset lifetimes without depending on scheduler timing. + + ''' + active: int = 0 + max_active: int = 0 + reset_calls: int = 0 + + class Proxy: + ''' + Return a pacing violation for every request. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Raise IB's pacing response. + + ''' + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def failed_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Hold one failed reset long enough to expose overlap. + + ''' + nonlocal active, max_active, reset_calls + done = trio.Event() + result: list[bool] = [] + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + active += 1 + reset_calls += 1 + max_active = max(max_active, active) + await trio.sleep(0.01) + active -= 1 + result.append(False) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + failed_reset, + ) + + async def request() -> None: + ''' + Run one independently bounded history request. + + ''' + with pytest.raises(DataUnavailable): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=1, + ) + + async def main() -> None: + with trio.fail_after(0.1): + async with trio.open_nursery() as nursery: + nursery.start_soon(request) + nursery.start_soon(request) + + trio.run(main) + assert reset_calls == 2 + assert max_active == 1 + + +def test_pacing_supersedes_timeout_reset_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Pacing handoff must not count cancellation as another failure. + + A slow request can trigger timeout recovery immediately before IB + returns a pacing violation. Query-side recovery then deliberately + cancels that data reset and waits for the same actor-global lock. + Reproduce this with blocked first and failed second resets. A + one-attempt limit proves cancellation releases the lock + without aborting before the intended connection reset runs. + + ''' + reset_types: list[str] = [] + + class Proxy: + ''' + Return pacing only after timeout recovery starts. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Delay past the reset timeout, then report pacing. + + ''' + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Block data reset and fail connection reset immediately. + + ''' + done = trio.Event() + result: list[bool] = [] + reset_types.append(reset_type) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + if reset_type == 'data': + await trio.sleep_forever() + + result.append(False) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=1, + ) + + trio.run(main) + assert reset_types == ['data', 'connection'] + + +def test_completed_pacing_reset_restarts_timeout_wait( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A queued timeout waiter must consume completed pacing recovery. + + While query-side pacing recovery holds the reset lock, the outer + timeout task can already queue behind it. Without a completion + handoff, that waiter launches a redundant data reset immediately. + Return a blank frame just after a successful connection reset and + prove no third reset runs before the restarted timeout observes + completion. + + ''' + reset_types: list[str] = [] + + class Proxy: + ''' + Pace the first query and return blank data on retry. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + def __init__(self) -> None: + self.calls: int = 0 + + async def bars(self, **kwargs): + ''' + Control the pacing-to-success handoff ordering. + + ''' + self.calls += 1 + if self.calls == 1: + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + await trio.sleep(0.0005) + return ( + [], + np.empty(0), + duration(seconds=2000), + ) + + async def reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Cancel timeout recovery and complete pacing recovery. + + ''' + done = trio.Event() + result: list[bool] = [] + reset_types.append(reset_type) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + if reset_type == 'data': + await trio.sleep_forever() + + result.append(reset_type == 'connection') + done.set() + return result[0] + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + reset, + ) + + async def main() -> None: + with trio.fail_after(0.1): + result, _ = await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=2, + ) + assert result is None + + trio.run(main) + assert reset_types == ['data', 'connection']