diff --git a/piker/clearing/_ems.py b/piker/clearing/_ems.py index 1d152926..a9c849e4 100644 --- a/piker/clearing/_ems.py +++ b/piker/clearing/_ems.py @@ -159,11 +159,11 @@ class DarkBook(Struct): _DEFAULT_SIZE: float = 1.0 -async def clear_dark_triggers( +async def _clear_dark_triggers( router: Router, brokerd_orders_stream: tractor.MsgStream, - quote_stream: tractor.MsgStream, + quote_stream: trionics.AsyncReceiver, broker: str, fqme: str, @@ -182,7 +182,6 @@ async def clear_dark_triggers( # - port to the new ringbuf stuff in `tractor.ipc`! # - numba all this! # - this stream may eventually contain multiple symbols - quote_stream._raise_on_lag = False async for quotes in quote_stream: # start = time.time() for sym, quote in quotes.items(): @@ -314,6 +313,32 @@ async def clear_dark_triggers( # print(f'execs scan took: {time.time() - start}') +async def clear_dark_triggers( + router: Router, + brokerd_orders_stream: tractor.MsgStream, + quote_stream: tractor.MsgStream, + broker: str, + fqme: str, + book: DarkBook, + +) -> None: + ''' + Run dark clearing on a dedicated non-strict quote subscription. + + ''' + async with quote_stream.subscribe( + raise_on_lag=False, + ) as quotes_stream: + await _clear_dark_triggers( + router, + brokerd_orders_stream, + quotes_stream, + broker, + fqme, + book, + ) + + class TradesRelay(Struct): # for now we keep only a single connection open with diff --git a/tests/test_ems_broadcast.py b/tests/test_ems_broadcast.py new file mode 100644 index 00000000..8621945b --- /dev/null +++ b/tests/test_ems_broadcast.py @@ -0,0 +1,79 @@ +''' +EMS broadcast ownership regressions. + +''' +from contextlib import asynccontextmanager as acm + +import trio + +from piker.clearing import _ems + + +def test_dark_clearing_owns_non_strict_subscription( + monkeypatch, +) -> None: + ''' + Dark clearing must not mutate another feed consumer's lag policy. + + `clear_dark_triggers()` previously assigned `_raise_on_lag` on + whichever root or temporary receiver was stored in `Flume.stream`. + That changed private shared state and made behavior depend on feed + cache ownership. Supply a stream whose subscription records the + public lag-policy argument and yields a distinct child. Replace + the hot trigger loop with a checkpointing probe, then prove the + non-strict child is entered, passed to the core, and closed only + after processing completes while the source remains untouched. + + ''' + events: list[tuple] = [] + child = object() + + class QuoteStream: + ''' + Record public subscription ownership without quote traffic. + + ''' + @acm + async def subscribe(self, raise_on_lag: bool = True): + ''' + Yield one dedicated child for the wrapper lifetime. + + ''' + events.append(('enter', raise_on_lag)) + try: + yield child + finally: + events.append(('exit', raise_on_lag)) + + async def clear_core(*args) -> None: + ''' + Prove the child remains owned while processing runs. + + ''' + events.append(('core', args[2] is child)) + await trio.lowlevel.checkpoint() + + monkeypatch.setattr( + _ems, + '_clear_dark_triggers', + clear_core, + ) + + async def main() -> None: + stream = QuoteStream() + await _ems.clear_dark_triggers( + object(), + object(), + stream, + 'ib', + 'nvda.nasdaq.ib', + object(), + ) + assert not hasattr(stream, '_raise_on_lag') + + trio.run(main) + assert events == [ + ('enter', False), + ('core', True), + ('exit', False), + ]