292 lines
9.7 KiB
Python
292 lines
9.7 KiB
Python
'''
|
|
IB quote restart regressions using the real Tractor asyncio bridge.
|
|
|
|
'''
|
|
import asyncio
|
|
from contextlib import asynccontextmanager as acm
|
|
from types import SimpleNamespace
|
|
|
|
from async_generator import aclosing
|
|
from ib_async import Contract, IB, RequestError, Ticker
|
|
import pytest
|
|
import tractor
|
|
import trio
|
|
|
|
from piker.brokers.ib import feed
|
|
from piker.brokers.ib import _util
|
|
from piker.brokers.ib.api import Client, open_client_proxy
|
|
|
|
|
|
async def exercise_quote_restart() -> list[str]:
|
|
'''
|
|
Cancel an active quote consumer, then reopen in the same actor.
|
|
|
|
Only the external IB connection is replaced. The production quote
|
|
producer, eventkit callback, bridge, broadcaster and context cache
|
|
all run unchanged in an asyncio-infected Tractor child. The farm
|
|
OK event in stream_quotes() cancels its local scope after a reset;
|
|
reproduce that cancellation after an actual ticker delivery.
|
|
|
|
A cached owner can turn this into a normal bridge exit, feeding
|
|
TrioTaskExited back into the consumer and killing brokerd. Two
|
|
generations must close their IB subscriptions and asyncio tasks.
|
|
Return callback failures separately so actor survival cannot be
|
|
mistaken for clean teardown. This proves local restart semantics,
|
|
not connectivity to a real Gateway.
|
|
|
|
'''
|
|
contract = Contract(symbol='QQQ')
|
|
tasks: list[asyncio.Task] = []
|
|
tickers: list[Ticker] = []
|
|
cancelled: list[Contract] = []
|
|
callback_errors: list[dict] = []
|
|
|
|
class ReplayIB:
|
|
'''
|
|
Deliver one real eventkit ticker per subscription.
|
|
|
|
'''
|
|
def reqMktData(self, con: Contract, opts: str) -> Ticker:
|
|
'''
|
|
Emit once the production producer installs its callback.
|
|
|
|
'''
|
|
task = asyncio.current_task()
|
|
assert task is not None
|
|
tasks.append(task)
|
|
ticker = Ticker(contract=con)
|
|
tickers.append(ticker)
|
|
asyncio.get_running_loop().call_soon(
|
|
ticker.updateEvent.emit,
|
|
ticker,
|
|
)
|
|
return ticker
|
|
|
|
def cancelMktData(self, con: Contract) -> None:
|
|
'''
|
|
Record the production producer's subscription teardown.
|
|
|
|
'''
|
|
cancelled.append(con)
|
|
|
|
client = SimpleNamespace(ib=ReplayIB())
|
|
|
|
@acm
|
|
async def load_clients(**kwargs):
|
|
'''
|
|
Supply an offline IB peer without opening broker connections.
|
|
|
|
'''
|
|
yield {'ib.replay': client}
|
|
|
|
loop = asyncio.get_running_loop()
|
|
previous_handler = loop.get_exception_handler()
|
|
loop.set_exception_handler(
|
|
lambda loop, context: callback_errors.append(context)
|
|
)
|
|
try:
|
|
with pytest.MonkeyPatch.context() as patch:
|
|
patch.setattr(feed, 'load_aio_clients', load_clients)
|
|
patch.setattr(
|
|
feed,
|
|
'get_preferred_data_client',
|
|
lambda clients: ('ib.replay', client),
|
|
)
|
|
for generation in range(2):
|
|
with trio.CancelScope() as scope:
|
|
async with feed.open_aio_quote_stream(
|
|
'qqq.nasdaq',
|
|
contract=contract,
|
|
) as quotes:
|
|
async with aclosing(quotes):
|
|
ticker = await quotes.receive()
|
|
assert ticker is tickers[generation]
|
|
scope.cancel()
|
|
await trio.lowlevel.checkpoint()
|
|
|
|
assert scope.cancelled_caught
|
|
assert len(cancelled) == generation + 1
|
|
assert tasks[generation].done()
|
|
assert len(tickers[generation].updateEvent) == 0
|
|
|
|
# Flush done callbacks before accepting a clean restart.
|
|
await tractor.to_asyncio.run_task(asyncio.sleep, delay=0)
|
|
return [
|
|
repr(context['exception'])
|
|
for context in callback_errors
|
|
]
|
|
finally:
|
|
loop.set_exception_handler(previous_handler)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'check_callbacks',
|
|
[False, True],
|
|
ids=['actor-survival', 'callback-cleanliness'],
|
|
)
|
|
def test_quote_restart_preserves_actor(
|
|
reg_addr,
|
|
check_callbacks: bool,
|
|
) -> None:
|
|
'''
|
|
A farm-style quote restart must leave its broker actor usable.
|
|
|
|
Use a real TCP actor nursery and an asyncio-infected child. Run
|
|
two subscription generations, then invoke another RPC on the same
|
|
actor. Both cases reject a propagated bridge exception, leaked
|
|
quote task or lost actor. The strict case also rejects asyncio
|
|
callback failures; the survival case reports those independently.
|
|
The backout can restore restart survival while the strict case
|
|
still exposes Tractor's completion-callback defect.
|
|
|
|
'''
|
|
async def main() -> None:
|
|
with trio.fail_after(10):
|
|
async with (
|
|
tractor.open_root_actor(registry_addrs=[reg_addr]),
|
|
tractor.open_nursery() as nursery,
|
|
):
|
|
portal = await nursery.start_actor(
|
|
'ib_quote_replay',
|
|
enable_modules=[__name__],
|
|
infect_asyncio=True,
|
|
)
|
|
errors = await portal.run(exercise_quote_restart)
|
|
errors += await portal.run(exercise_quote_restart)
|
|
await portal.cancel_actor()
|
|
if check_callbacks:
|
|
assert not errors, errors
|
|
|
|
trio.run(main)
|
|
|
|
|
|
async def exercise_history_recovery(timeframe: int) -> None:
|
|
'''
|
|
Replay pacing and farm recovery across the production IB adapter.
|
|
|
|
Keep Client.bars(), MethodProxy, both relay tasks, get_bars(),
|
|
wait_on_data_reset() and data_reset_hack() intact. Replace only
|
|
external contract discovery, historical replies and VNC input.
|
|
Deliver the farm event through IB.errorEvent after the real reset
|
|
task is waiting, and reject a retry before that recovery signal.
|
|
|
|
This covers the bridge missing from the imported policy tests;
|
|
it neither sends VNC keys nor connects to a broker or storage.
|
|
|
|
'''
|
|
requests: int = 0
|
|
reset_calls: int = 0
|
|
recovery_sent: bool = False
|
|
reset_started = trio.Event()
|
|
contract = Contract(symbol='QQQ')
|
|
reason = 'HMDS data farm connection is OK:ushmds'
|
|
|
|
class ReplayIB(IB):
|
|
'''
|
|
Use real IB events with scripted historical responses.
|
|
|
|
'''
|
|
async def reqHistoricalDataAsync(self, **kwargs) -> list:
|
|
'''
|
|
Reject the first request; return empty data after reset.
|
|
|
|
'''
|
|
nonlocal requests
|
|
requests += 1
|
|
if requests == 1:
|
|
raise RequestError(1, 162, feed._pacing)
|
|
assert recovery_sent
|
|
assert requests == 2
|
|
return []
|
|
|
|
ib = ReplayIB()
|
|
client = Client(ib=ib, config={'vnc_addrs': []})
|
|
|
|
async def find_contracts(fqme: str) -> list[Contract]:
|
|
'''
|
|
Return the replay contract without network qualification.
|
|
|
|
'''
|
|
return [contract]
|
|
|
|
async def vnc_input(client: Client) -> None:
|
|
'''
|
|
Record completion of external reset input on asyncio.
|
|
|
|
'''
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
assert reset_calls == 1
|
|
reset_started.set()
|
|
await asyncio.sleep(0)
|
|
|
|
async def emit_recovery() -> None:
|
|
'''
|
|
Deliver the Gateway status through Client.inline_errors().
|
|
|
|
'''
|
|
nonlocal recovery_sent
|
|
recovery_sent = True
|
|
ib.errorEvent.emit(-1, 2106, reason, contract)
|
|
|
|
with pytest.MonkeyPatch.context() as patch:
|
|
patch.setattr(client, 'find_contracts', find_contracts)
|
|
patch.setattr(_util, '_reset_tech', 'vnc')
|
|
patch.setattr(_util, 'vnc_click_hack', vnc_input)
|
|
async with open_client_proxy(client) as proxy:
|
|
async def gateway_status() -> None:
|
|
'''
|
|
Wait for the actual farm-event consumer to be ready.
|
|
|
|
'''
|
|
await reset_started.wait()
|
|
event = proxy.event_table[reason]
|
|
while not event.statistics().tasks_waiting:
|
|
await trio.lowlevel.checkpoint()
|
|
await tractor.to_asyncio.run_task(emit_recovery)
|
|
|
|
async with trio.open_nursery() as nursery:
|
|
nursery.start_soon(gateway_status)
|
|
result, reset_used = await feed.get_bars(
|
|
proxy,
|
|
'qqq.nasdaq',
|
|
timeframe,
|
|
)
|
|
assert result is None
|
|
assert reset_used
|
|
assert not feed._data_reset_lock.locked()
|
|
|
|
assert requests == 2
|
|
assert reset_calls == 1
|
|
|
|
|
|
def test_history_pacing_through_real_bridge(reg_addr) -> None:
|
|
'''
|
|
Pacing recovery must complete before retrying either bar period.
|
|
|
|
Run the production proxy and reset tasks in an infected actor for
|
|
both 1s and 60s history. A returned empty frame must release the
|
|
request rather than trigger more resets. Repeat on the same actor
|
|
to expose leaked ownership from the first request.
|
|
|
|
'''
|
|
async def main() -> None:
|
|
with trio.fail_after(10):
|
|
async with (
|
|
tractor.open_root_actor(registry_addrs=[reg_addr]),
|
|
tractor.open_nursery() as nursery,
|
|
):
|
|
portal = await nursery.start_actor(
|
|
'ib_history_replay',
|
|
enable_modules=[__name__],
|
|
infect_asyncio=True,
|
|
)
|
|
for timeframe in (1, 60):
|
|
await portal.run(
|
|
exercise_history_recovery,
|
|
timeframe=timeframe,
|
|
)
|
|
await portal.cancel_actor()
|
|
|
|
trio.run(main)
|