pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 13 Packages Projects Releases Wiki Activity

.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: 0846cbd4a4
Source-author: goodboy <goodboy_foss@protonmail.com>
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`))
wkt/land_ib_recovery
Gud Boi 2026-10-23 10:50:20 -04:00
parent 72d0bdad5e
commit 6f626a9c45
2 changed files with 601 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
piker/brokers/ib/feed.py
View File

@ -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,

516
tests/test_ib_history.py 100644
View File

@ -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']