From 6f626a9c456f304b4355b23f812db0c6e5750ce7 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 23 Oct 2026 10:50:20 -0400 Subject: [PATCH] .ib.feed: backport bounded reset coordination PARTIAL BACKPORT -- IB adapter and seven adapter regressions only. The source commit also repairs startup gaps in `ShmArray`; this patch does NOT deliver that complete history/storage fix. Import the exact `piker/brokers/ib/feed.py` delta: serialize farm resets, publish completion and result separately, keep failure counters per request, bound IB cancellation retries, and complete empty replies without leaving the parent waiting forever. Deats, - Extract seven existing cases into `tests/test_ib_history.py`. Their function bodies match the source; the module header and imports are reduced for this older landing tree. - Omit `piker/tsp/_history.py`: inclusive endpoint handling, SHM layout, final null-sweep ordering and synthetic gap boundaries. - Omit the source's history-layer regressions, harness update and historical prompt files. The source SHA retains that provenance. Downstream replay, Only the adapter hunks and these seven cases are already present. Retain the omitted history/SHM changes and their regressions; do NOT skip the whole source commit when its adapter delta is empty. Partial-backport-of: 0846cbd4a4197e47415689cf5e9d146ba3174c32 Source-author: goodboy Prompt-IO: ai/prompt-io/opencode/20261022T212248Z_dc618229_ib_recovery_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-6-astra` (`openai`)) --- piker/brokers/ib/feed.py | 137 +++++++---- tests/test_ib_history.py | 516 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 601 insertions(+), 52 deletions(-) create mode 100644 tests/test_ib_history.py 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']