From 4af885d90f6e4bcfa963f2f7a1d40436d3c928f3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 19:16:28 -0400 Subject: [PATCH] .ib.feed: cache quote channel contexts Replace the manual `_quote_streams` raw-channel registry with a symbol-keyed `maybe_open_context()` resource whose lifetime extends to the final quote consumer. Give every `open_aio_quote_stream()` caller a lexical child from the cached `LinkedTaskChannel` broadcaster so concurrent users receive the same quotes and close independently. Prompt-IO: ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260813T230744Z_ddac3553_prompt_io.md | 32 +++++ ...20260813T230744Z_ddac3553_prompt_io.raw.md | 39 ++++++ piker/brokers/ib/feed.py | 75 +++++------ tests/test_ib_broadcast.py | 120 ++++++++++++++++++ 4 files changed, 223 insertions(+), 43 deletions(-) create mode 100644 ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md create mode 100644 tests/test_ib_broadcast.py diff --git a/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md new file mode 100644 index 00000000..fba4ae56 --- /dev/null +++ b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md @@ -0,0 +1,32 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: unavailable +timestamp: 2026-08-13T23:07:44Z +git_ref: ddac3553 +scope: code +substantive: true +raw_file: 20260813T230744Z_ddac3553_prompt_io.raw.md +--- + +## Prompt + +Continue the piker broadcast-consumer audit after the Deribit ownership commit +and prepare the next smallest ownership fix. + +## Response summary + +Replaced IB's manual raw quote-channel registry with a symbol-keyed cached +context and gave every caller a lexical child from one retained +`LinkedTaskChannel` broadcaster root. + +## Files changed + +- `piker/brokers/ib/feed.py` - cached raw-channel lifetime and lexical child + ownership. +- `tests/test_ib_broadcast.py` - concurrent fan-out, cache-key and child + lifetime regression. + +## Human edits + +None - generated output follows the requested broadcast-consumer audit. diff --git a/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md new file mode 100644 index 00000000..0a96efbc --- /dev/null +++ b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md @@ -0,0 +1,39 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-13T23:07:44Z +git_ref: ddac3553 +diff_cmd: git diff HEAD~1..HEAD +--- + +The user asked to continue the piker broadcast-consumer audit after committing +the Deribit ownership fix. The next isolated item was IB's actor-local +`_quote_streams` cache, which retained a raw channel owned by its first caller +and constructed competing broadcaster roots for later callers. + +> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py` + +Factored `_open_aio_quote_channel()` as the raw asyncio-channel context and +made `open_aio_quote_stream()` cache that context by symbol through +`maybe_open_context()`. Every caller now receives a lexical child from the +retained `LinkedTaskChannel` broadcaster, while the raw source remains alive +until its final cache user exits. Removed the manual `_quote_streams` registry +and teardown mutation. + +> `git diff HEAD~1..HEAD -- tests/test_ib_broadcast.py` + +Added a concurrent ownership regression using real Tractor broadcast state. +It proves miss and hit callers use the same symbol/context factory, receive the +same quote through distinct children, close independently, and leave the owner +live after the hit exits. + +Verification output: + +```text +. [100%] +1 passed in 0.01s +``` + +The new test passes through a source-isolated IB feed import. Test Ruff, E501, +Python compilation and whitespace checks pass. Normal package collection +remains blocked by the existing Qt environment mismatch. diff --git a/piker/brokers/ib/feed.py b/piker/brokers/ib/feed.py index adeb6361..cfea2b5c 100644 --- a/piker/brokers/ib/feed.py +++ b/piker/brokers/ib/feed.py @@ -738,10 +738,6 @@ async def get_bars( ) -# per-actor cache of inter-eventloop-chans -_quote_streams: dict[str, trio.abc.ReceiveStream] = {} - - async def _setup_quote_stream( chan: tractor.to_asyncio.LinkedTaskChannel, symbol: str, @@ -770,8 +766,6 @@ async def _setup_quote_stream( and is thus run via `tractor.to_asyncio.open_channel_from()`. ''' - global _quote_streams - async with load_aio_clients( disconnect_on_exit=False, ) as accts2clients: @@ -834,9 +828,6 @@ async def _setup_quote_stream( client.ib.cancelMktData(contract) - # decouple broadcast mem chan - _quote_streams.pop(symbol, None) - def push( t: Ticker, tries_before_raise: int = 6, @@ -953,54 +944,52 @@ async def _setup_quote_stream( @acm -async def open_aio_quote_stream( +async def _open_aio_quote_channel( symbol: str, contract: Contract|None = None, -) -> ( - trio.abc.Channel| # iface - tractor.to_asyncio.LinkedTaskChannel # actually -): +) -> tractor.to_asyncio.LinkedTaskChannel: ''' - Open a real-time `Ticker` quote stream from an `asyncio.Task` - spawned via `tractor.to_asyncio.open_channel_from()`, deliver the - inter-event-loop channel to the `trio.Task` caller and cache it - globally for re-use. + Open one raw inter-event-loop quote channel for `symbol`. ''' - from tractor.trionics import broadcast_receiver - global _quote_streams - - from_aio = _quote_streams.get(symbol) - if from_aio: - - # if we already have a cached feed deliver a rx side clone - # to consumer - async with broadcast_receiver( - from_aio, - 2**6, - ) as from_aio: - yield from_aio - return - - from_aio: tractor.to_asyncio.LinkedTaskChannel + chan: tractor.to_asyncio.LinkedTaskChannel + started_contract: Contract async with tractor.to_asyncio.open_channel_from( _setup_quote_stream, symbol=symbol, contract=contract, - ) as (from_aio, contract): + ) as (chan, started_contract): + assert started_contract + yield chan - assert contract - # TODO? de-reg on teardown of last consumer task? - # -> why aren't we using `.trionics.maybe_open_context()` - # here again?? (we are in `open_client_proxies()` tho?) - # - # cache feed for later consumers - _quote_streams[symbol] = from_aio +@acm +async def open_aio_quote_stream( + symbol: str, + contract: Contract|None = None, - yield from_aio +) -> tractor.trionics.BroadcastReceiver: + ''' + Open a caller-owned child of the cached quote channel for `symbol`. + + Keep the raw `LinkedTaskChannel` alive until its last cache user + exits, and retain exactly one broadcaster root on that channel. + + ''' + chan: tractor.to_asyncio.LinkedTaskChannel + async with tractor.trionics.maybe_open_context( + acm_func=_open_aio_quote_channel, + kwargs={ + 'symbol': symbol, + 'contract': contract, + }, + key=symbol, + ) as (_, chan): + bstream: tractor.trionics.BroadcastReceiver + async with chan.subscribe() as bstream: + yield bstream # TODO: cython/mypyc/numba this! diff --git a/tests/test_ib_broadcast.py b/tests/test_ib_broadcast.py new file mode 100644 index 00000000..37250701 --- /dev/null +++ b/tests/test_ib_broadcast.py @@ -0,0 +1,120 @@ +''' +IB cached quote-channel ownership regressions. + +''' +from contextlib import asynccontextmanager as acm + +import tractor +import trio + +from piker.brokers.ib import feed + + +def test_quote_stream_caches_channel_and_owns_child( + monkeypatch, +) -> None: + ''' + IB quote callers must share one cached channel through child cursors. + + `open_aio_quote_stream()` previously stored the first caller's raw + `LinkedTaskChannel` in `_quote_streams`, yielded that root directly, + and wrapped the same source in a new broadcaster for each later + caller. The first caller therefore owned source lifetime while + competing roots split quote receives between concurrent users. + + Stub `maybe_open_context()` with one retained channel and enter a + miss owner plus concurrent hit. Prove both receive the same quote + through distinct children of one broadcaster, the hit closes + independently, and the owner keeps receiving until its own exit. + Also verify both calls request the same symbol cache key. + + ''' + class FakeLinkedTaskChannel: + ''' + Retain one real broadcaster and expose child subscriptions. + + ''' + def __init__(self) -> None: + self._tx: trio.MemorySendChannel + rx: trio.MemoryReceiveChannel + self._tx, rx = trio.open_memory_channel(8) + self._broadcaster: ( + tractor.trionics.BroadcastReceiver + ) = tractor.trionics.broadcast_receiver( + rx, + 8, + ) + + @acm + async def subscribe(self): + ''' + Yield one caller-owned child from the retained root. + + ''' + child: tractor.trionics.BroadcastReceiver + async with self._broadcaster.subscribe() as child: + yield child + + async def push(self, msg: dict) -> None: + ''' + Send one ticker through the retained source channel. + + ''' + await self._tx.send(msg) + + chan: FakeLinkedTaskChannel = FakeLinkedTaskChannel() + context_entries: int = 0 + keys: list[str] = [] + acm_funcs: list = [] + + @acm + async def maybe_open_context(**kwargs): + ''' + Return one retained channel as a miss followed by a hit. + + ''' + nonlocal context_entries + keys.append(kwargs['key']) + acm_funcs.append(kwargs['acm_func']) + cache_hit: bool = context_entries > 0 + context_entries += 1 + yield cache_hit, chan + + monkeypatch.setattr( + feed.tractor.trionics, + 'maybe_open_context', + maybe_open_context, + ) + + async def main() -> None: + owner: tractor.trionics.BroadcastReceiver + async with feed.open_aio_quote_stream( + 'NVDA', + ) as owner: + hit: tractor.trionics.BroadcastReceiver + async with feed.open_aio_quote_stream( + 'NVDA', + ) as hit: + assert hit is not owner + assert len(chan._broadcaster._state.subs) == 3 + + await chan.push({'value': 1}) + assert await owner.receive() == {'value': 1} + assert await hit.receive() == {'value': 1} + + assert hit._closed + assert not owner._closed + assert len(chan._broadcaster._state.subs) == 2 + + await chan.push({'value': 2}) + assert await owner.receive() == {'value': 2} + + assert owner._closed + assert len(chan._broadcaster._state.subs) == 1 + assert keys == ['NVDA', 'NVDA'] + assert acm_funcs == [ + feed._open_aio_quote_channel, + feed._open_aio_quote_channel, + ] + + trio.run(main)