From 12b0d66e854247e92cb5a8668be71d816c72d856 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 13:48:32 -0400 Subject: [PATCH 1/4] .clearing: own dark quote broadcast subscription Give `clear_dark_triggers()` a dedicated `MsgStream.subscribe()` handle with `raise_on_lag=False` instead of mutating the private lag policy on whichever root or cached child sits in `Flume.stream`. Also, - factor the hot quote loop into `_clear_dark_triggers()` so the public task owns the child receiver for its full lifetime. - add an EMS regression proving the lag policy, child lifetime, and source-stream state remain isolated. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- piker/clearing/_ems.py | 31 +++++++++++++-- tests/test_ems_broadcast.py | 79 +++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 3 deletions(-) create mode 100644 tests/test_ems_broadcast.py 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), + ] -- 2.34.1 From b691422b2678e1e7b3d4b7e1d70c25d6c644c6d3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 18:53:44 -0400 Subject: [PATCH 2/4] .data: isolate cached feed stream consumers Keep provider root streams private to the cached `Feed` and give every `maybe_open_feed()` caller, including the first owner, a local subscription view with copied `Flume` descriptors. Also, - key cached resources by the normalized full symbol set and stream policy instead of only the first FQME. - subscribe each `Feed.open_multi_stream()` provider relay and wait for all relays to start before yielding the merged channel. - give each provider relay an independent send-channel clone so one provider EOC cannot terminate its siblings. - document `Feed.pause()` and `Feed.resume()` as shared remote-provider controls rather than caller-local cursor operations. - cover miss/hit ownership, multi-market provider mapping, relay fan-out, lexical teardown and cache-key separation. Prompt-IO: ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260813T202402Z_12b0d66e_prompt_io.md | 34 ++ ...20260813T202402Z_12b0d66e_prompt_io.raw.md | 51 +++ piker/data/feed.py | 174 +++++++-- tests/test_feed_broadcast.py | 347 ++++++++++++++++++ 4 files changed, 580 insertions(+), 26 deletions(-) create mode 100644 ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md create mode 100644 tests/test_feed_broadcast.py diff --git a/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md new file mode 100644 index 00000000..c9d82f5a --- /dev/null +++ b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md @@ -0,0 +1,34 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: unavailable +timestamp: 2026-08-13T20:24:02Z +git_ref: 12b0d66e +scope: code +substantive: true +raw_file: 20260813T202402Z_12b0d66e_prompt_io.raw.md +--- + +## Prompt + +Continue the paired piker broadcast-consumer work after Tractor PR 490 +stabilization. Stop cached feed callers from mutating or consuming shared +quote-stream roots, then review, verify and prepare a commit boundary. + +## Response summary + +Made provider root streams private to the cached `Feed`, gave every caller a +local subscribed view, corrected feed cache identity, and made multi-provider +relays own synchronized child subscriptions and independent send-channel +clones. + +## Files changed + +- `piker/data/feed.py` - caller-local feed views, complete cache keys, and + subscribed multi-provider relays with independent channel lifetimes. +- `tests/test_feed_broadcast.py` - real broadcast ownership, lifecycle and + cache-identity regressions, including sibling survival after provider EOC. + +## Human edits + +None - generated output follows the requested cached-feed ownership item. diff --git a/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md new file mode 100644 index 00000000..a168dcc1 --- /dev/null +++ b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md @@ -0,0 +1,51 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-13T20:24:02Z +git_ref: 12b0d66e +diff_cmd: git diff HEAD~1..HEAD +--- + +The user asked to continue the paired piker broadcast-consumer work after +Tractor PR 490 stabilization. The next item was to stop cached feed callers +from mutating or consuming shared `Feed`/`Flume` quote-stream roots, then +review, verify and prepare a commit boundary. + +> `git diff HEAD~1..HEAD -- piker/data/feed.py` + +Changed `maybe_open_feed()` so the cached `Feed` exclusively owns provider +root streams and every caller, including the first cache owner, receives a +caller-local `Feed`/`Flume` view backed by one child subscription per provider. +The cache identity now includes normalized full symbol shape and stream policy. + +Changed `Feed.open_multi_stream()` so every provider relay subscribes instead +of directly consuming a shared stream, and synchronizes relay startup before +yielding the merged channel. Each relay owns a cloned send channel so one +provider EOC cannot terminate its siblings. The module-level relay returns its +live child receiver through `TaskStatus` for typed retention and introspection. +Documented `Feed.pause()`/`Feed.resume()` as provider-context-wide controls +shared by local cache users. + +> `git diff HEAD~1..HEAD -- tests/test_feed_broadcast.py` + +Added real `Feed`, `Flume`, and Tractor broadcast regressions covering both +cache-miss owner and cache-hit paths, one-provider-many-market mapping, +multi-provider relay fan-out and startup, duplex controls, lexical child +closure, sibling survival after provider EOC, cached-root immutability, and +full cache-key differentiation. + +Verification output: + +```text +... [100%] +3 passed in 0.02s + +. [100%] +1 passed in 0.01s +``` + +The second result is the existing EMS broadcast ownership regression. Ruff, +E501, compilation and whitespace checks passed. Repeated adversarial review +found and resolved shared `Feed.streams`, incomplete cache-key, fake-lifecycle, +relay-startup and first-owner asymmetry issues; final review reported no +findings. diff --git a/piker/data/feed.py b/piker/data/feed.py index d31d58a3..28b8d589 100644 --- a/piker/data/feed.py +++ b/piker/data/feed.py @@ -30,6 +30,7 @@ from collections import ( defaultdict, abc, ) +from copy import copy from contextlib import asynccontextmanager as acm from functools import partial import time @@ -677,6 +678,36 @@ async def open_feed_bus( bus.remove_subs(bs_fqme, subs) +async def relay_to_common_memchan( + stream: ( + tractor.MsgStream + |trionics.BroadcastReceiver + ), + tx: trio.MemorySendChannel[dict[str, Any]], + task_status: TaskStatus[ + trionics.BroadcastReceiver, + ] = trio.TASK_STATUS_IGNORED, + +) -> None: + ''' + Relay one provider subscription into a common memory channel. + + Deliver the live child receiver through `task_status` so the + caller can retain and inspect each provider relay's cursor. + + ''' + bstream: trionics.BroadcastReceiver + async with ( + stream.subscribe() as bstream, + tx, + ): + task_status.started(bstream) + + msg: dict[str, Any] + async for msg in bstream: + await tx.send(msg) + + class Feed(Struct): ''' A per-provider API for client-side consumption from real-time data @@ -726,29 +757,51 @@ class Feed(Struct): if len(mods) == 1: # just pass the datad stream directly if only one provider # was detected. - stream = self.streams[list(brokers)[0]] + stream: ( + tractor.MsgStream + |trionics.BroadcastReceiver + ) = self.streams[list(brokers)[0]] + bstream: trionics.BroadcastReceiver async with stream.subscribe() as bstream: yield bstream return # start multiplexing task tree + tx: trio.MemorySendChannel[dict[str, Any]] + rx: trio.MemoryReceiveChannel[dict[str, Any]] tx, rx = trio.open_memory_channel(616) - async def relay_to_common_memchan(stream: tractor.MsgStream): - async with tx: - async for msg in stream: - await tx.send(msg) - async with ( tractor.trionics.collapse_eg(), trio.open_nursery() as nurse ): # spawn a relay task for each stream so that they all # multiplex to a common channel. - for brokername in mods: - stream = self.streams[brokername] - nurse.start_soon(relay_to_common_memchan, stream) + broker2bstreams: dict[ + str, + trionics.BroadcastReceiver, + ] = {} + broker: str + for broker in mods: + stream: ( + tractor.MsgStream + |trionics.BroadcastReceiver + ) = self.streams[broker] + relay_tx: trio.MemorySendChannel[ + dict[str, Any] + ] = tx.clone() + + bstream: trionics.BroadcastReceiver = await nurse.start( + relay_to_common_memchan, + stream, + relay_tx, + ) + broker2bstreams[broker] = bstream + + # Each relay now owns one `tx` clone. Closing the original + # lets `rx` reach EOC only after every provider relay exits. + await tx.aclose() try: yield rx finally: @@ -757,10 +810,23 @@ class Feed(Struct): _max_sample_rate: int = 1 async def pause(self) -> None: + ''' + Pause the shared remote provider context for this cached feed. + + This controls provider publication for every local cache user; + it does not pause only this caller's receive cursor. + + ''' + stream: tractor.MsgStream|trionics.BroadcastReceiver for stream in set(self.streams.values()): await stream.send('pause') async def resume(self) -> None: + ''' + Resume the shared remote provider context for this cached feed. + + ''' + stream: tractor.MsgStream|trionics.BroadcastReceiver for stream in set(self.streams.values()): await stream.send('resume') @@ -820,7 +886,17 @@ async def maybe_open_feed( in a tractor broadcast receiver. ''' - fqme = fqmes[0] + cache_key: tuple[ + tuple[str, ...], + float|None, + bool, + bool, + ] = ( + tuple(sorted(fqmes)), + kwargs.get('tick_throttle'), + kwargs.get('allow_overruns', True), + kwargs.get('start_stream', True), + ) async with trionics.maybe_open_context( acm_func=open_feed, @@ -833,27 +909,73 @@ async def maybe_open_feed( 'allow_overruns': kwargs.get('allow_overruns', True), 'start_stream': kwargs.get('start_stream', True), }, - key=fqme, + key=cache_key, ) as (cache_hit, feed): + feed: Feed - if cache_hit: - log.info(f'Using cached feed for {fqme}') - # add a new broadcast subscription for the quote stream - # if this feed is likely already in use + if ( + cache_hit + and + log.at_least_level('info') + ): + log.info( + f'Using cached feed for key:\n' + f'{cache_key!r}' + ) - async with trionics.gather_contexts( - mngrs=[stream.subscribe() for stream in feed.streams.values()] - ) as bstreams: - for bstream, flume in zip(bstreams, feed.flumes.values()): - # XXX: TODO: horrible hackery that needs fixing.. - # i guess we have to create context proxies? - bstream._ctx = flume.stream._ctx - flume.stream = bstream + # The cached `Feed` exclusively owns provider root streams. + # Every caller, including the first cache owner, gets a local + # child cursor and copied descriptors for its lexical lifetime. + broker_names: tuple[str, ...] = tuple(feed.streams) + bstreams: tuple[ + trionics.BroadcastReceiver, + ..., + ] + async with trionics.gather_contexts( + mngrs=[ + feed.streams[broker].subscribe() + for broker in broker_names + ] + ) as bstreams: + broker2bstreams: dict[ + str, + trionics.BroadcastReceiver, + ] = dict(zip( + broker_names, + bstreams, + strict=True, + )) + flumes: dict[str, Flume] = {} + fqme: str + flume: Flume + for fqme, flume in feed.flumes.items(): + broker: str = flume.mkt.broker - yield feed - else: - yield feed + bstream: trionics.BroadcastReceiver = ( + broker2bstreams[broker] + ) + + # XXX: TODO: horrible hackery that needs fixing.. + # i guess we have to create context proxies? + root_stream: tractor.MsgStream = feed.streams[broker] + + ctx: tractor.Context = root_stream.ctx + bstream._ctx = ctx + + local_flume: Flume = copy(flume) + local_flume.stream = bstream + flumes[fqme] = local_flume + + # `feed` is the cache-owned resource and must retain only + # provider root streams. A shallow `Feed` copy gives this + # caller its own descriptor mappings while intentionally + # sharing immutable market metadata, SHM handles and + # provider controls for the cache resource's lifetime. + local_feed: Feed = copy(feed) + local_feed.flumes = flumes + local_feed.streams = broker2bstreams + yield local_feed @acm diff --git a/tests/test_feed_broadcast.py b/tests/test_feed_broadcast.py new file mode 100644 index 00000000..9f19b4cb --- /dev/null +++ b/tests/test_feed_broadcast.py @@ -0,0 +1,347 @@ +''' +Cached feed broadcast ownership regressions. + +''' +from contextlib import asynccontextmanager as acm +from types import SimpleNamespace + +import tractor +import trio +import pytest + +from piker.data import feed as feed_mod +from piker.data.flows import Flume + + +class FakeMsgStream: + ''' + Duplex test stream backed by a real Tractor broadcaster. + + ''' + def __init__(self, broker: str) -> None: + self.broker: str = broker + self._ctx: object = object() + self._tx, rx = trio.open_memory_channel(8) + self._broadcaster = tractor.trionics.broadcast_receiver( + rx, + 8, + ) + self.children: list[ + tractor.trionics.BroadcastReceiver, + ] = [] + self.controls: list[str] = [] + + @property + def ctx(self) -> object: + ''' + Expose the public context ref provided by `MsgStream.ctx`. + + ''' + return self._ctx + + @acm + async def subscribe(self): + ''' + Yield a real child receiver with duplex send delegation. + + ''' + child: tractor.trionics.BroadcastReceiver + async with self._broadcaster.subscribe() as child: + child.send = self.send + self.children.append(child) + yield child + + async def send(self, msg: str) -> None: + ''' + Record a pause or resume control sent through a child. + + ''' + self.controls.append(msg) + + async def push(self, msg: dict) -> None: + ''' + Send one provider quote into the broadcast source. + + ''' + await self._tx.send(msg) + + async def close_source(self) -> None: + ''' + End this provider without closing sibling relay channels. + + ''' + await self._tx.aclose() + + +def make_feed() -> tuple[ + feed_mod.Feed, + dict[str, FakeMsgStream], +]: + ''' + Build a real `Feed` and `Flume` graph for two providers. + + ''' + roots: dict[str, FakeMsgStream] = { + broker: FakeMsgStream(broker) + for broker in ('ib', 'deribit') + } + flumes: dict[str, Flume] = { + fqme: Flume( + mkt=SimpleNamespace(broker=broker), + first_quote={}, + _rt_shm_token=SimpleNamespace(), + stream=roots[broker], + ) + for fqme, broker in ( + ('nvda.nasdaq.ib', 'ib'), + ('aapl.nasdaq.ib', 'ib'), + ('btc-usd.deribit', 'deribit'), + ) + } + return ( + feed_mod.Feed( + mods={ + broker: SimpleNamespace(name=broker) + for broker in roots + }, + portals={}, + flumes=flumes, + streams=roots, + status={}, + ), + roots, + ) + + +@pytest.mark.parametrize('cache_hit', [False, True]) +def test_cached_feed_returns_local_streams( + monkeypatch, + cache_hit: bool, +) -> None: + ''' + A cached feed caller must not consume or replace shared streams. + + `maybe_open_feed()` previously paired provider subscriptions with + `Feed.flumes.values()` using `zip()` and installed each child on the + cached `Flume`. One caller therefore exposed its lexical receiver + to every holder of the shared `Feed`; after exit those descriptors + pointed at closed children. The positional pairing also dropped a + market whenever one provider supplied multiple flumes. + + Model two real provider broadcasters with two IB markets and one + Deribit market. Enter both first-owner and cache-hit paths, then + prove the yielded `Feed`, its `Feed.streams` mapping and all + `Flume`s are caller-local views. Exercise provider-context-wide + pause/resume and `Feed.open_multi_stream()` to show duplex sends + still reach the roots while each relay owns another child. Finally + prove every lexical child closes and all cached descriptors still + point at live roots. + + ''' + cached_feed: feed_mod.Feed + roots: dict[str, FakeMsgStream] + cached_feed, roots = make_feed() + cached_flumes: dict[str, Flume] = cached_feed.flumes + + @acm + async def maybe_open_context(**kwargs): + ''' + Return the controlled shared feed as a cache hit. + + ''' + yield cache_hit, cached_feed + + monkeypatch.setattr( + feed_mod.trionics, + 'maybe_open_context', + maybe_open_context, + ) + + local_feed: feed_mod.Feed|None = None + local_streams: dict[ + str, + tractor.trionics.BroadcastReceiver, + ]|None = None + + async def main() -> None: + nonlocal local_feed, local_streams + feed: feed_mod.Feed + async with feed_mod.maybe_open_feed( + list(cached_flumes), + ) as feed: + local_feed = feed + local_streams = feed.streams + assert feed is not cached_feed + assert feed.streams is not roots + assert feed.flumes is not cached_flumes + + broker: str + stream: tractor.trionics.BroadcastReceiver + for broker, stream in feed.streams.items(): + assert stream is roots[broker].children[0] + assert stream._ctx is roots[broker].ctx + + fqme: str + flume: Flume + for fqme, flume in feed.flumes.items(): + cached: Flume = cached_flumes[fqme] + + stream: tractor.trionics.BroadcastReceiver = ( + feed.streams[flume.mkt.broker] + ) + + assert flume is not cached + assert flume.stream is stream + assert cached.stream is roots[flume.mkt.broker] + + assert ( + feed.flumes['nvda.nasdaq.ib'].stream + is feed.flumes['aapl.nasdaq.ib'].stream + ) + + await feed.pause() + await feed.resume() + assert all( + root.controls == ['pause', 'resume'] + for root in roots.values() + ) + + stream: trio.MemoryReceiveChannel + async with feed.open_multi_stream() as stream: + assert all( + len(child._state.subs) == 3 + for child in feed.streams.values() + ) + await roots['ib'].push({'provider': 'ib'}) + await roots['deribit'].push({ + 'provider': 'deribit', + }) + with trio.fail_after(1): + msgs = { + (await stream.receive())['provider'], + (await stream.receive())['provider'], + } + assert msgs == {'ib', 'deribit'} + + # One provider's EOC must close only its `tx` clone; + # the sibling relay continues publishing to `stream`. + await roots['ib'].close_source() + with trio.fail_after(1): + while ( + len(feed.streams['ib']._state.subs) + != 2 + ): + await trio.lowlevel.checkpoint() + + await roots['deribit'].push({ + 'provider': 'deribit-after-ib-eoc', + }) + msg: dict + msg = await stream.receive() + + assert msg['provider'] == 'deribit-after-ib-eoc' + + assert all( + len(child._state.subs) == 2 + for child in feed.streams.values() + ) + + trio.run(main) + + assert local_feed is not None + assert local_streams is not None + assert all(stream._closed for stream in local_streams.values()) + assert all( + flume.stream is roots[flume.mkt.broker] + for flume in cached_flumes.values() + ) + assert all( + len(root._broadcaster._state.subs) == 1 + for root in roots.values() + ) + + +def test_feed_cache_key_includes_shape_and_policy( + monkeypatch, +) -> None: + ''' + Feed cache identity must include every resource-defining input. + + The old `fqmes[0]` key aliased `[A]` with `[A, B]` and ignored + `tick_throttle`, `allow_overruns` and `start_stream`. Whichever + caller entered first silently determined later feed shape and + stream behavior, including reuse of a history-only remote feed for + a live quote request. + + Capture the key passed to `maybe_open_context()` for reordered and + varied requests. Symbol order must normalize to one key, while the + full symbol set and each stream policy must produce distinct keys. + + ''' + feed: feed_mod.Feed = feed_mod.Feed( + mods={}, + portals={}, + flumes={}, + streams={}, + status={}, + ) + keys: list[tuple] = [] + + @acm + async def maybe_open_context(**kwargs): + ''' + Capture cache identity without opening a remote feed. + + ''' + keys.append(kwargs['key']) + yield False, feed + + @acm + async def gather_contexts(mngrs): + ''' + Permit the metadata-only empty feed used by this key probe. + + ''' + assert not mngrs + yield () + + monkeypatch.setattr( + feed_mod.trionics, + 'maybe_open_context', + maybe_open_context, + ) + monkeypatch.setattr( + feed_mod.trionics, + 'gather_contexts', + gather_contexts, + ) + + async def open_once( + fqmes: list[str], + **kwargs, + ) -> None: + async with feed_mod.maybe_open_feed( + fqmes, + **kwargs, + ): + pass + + async def main() -> None: + await open_once(['a.ib', 'b.ib']) + await open_once(['b.ib', 'a.ib']) + await open_once(['a.ib']) + await open_once(['a.ib', 'b.ib'], tick_throttle=10) + await open_once( + ['a.ib', 'b.ib'], + allow_overruns=False, + ) + await open_once( + ['a.ib', 'b.ib'], + start_stream=False, + ) + + trio.run(main) + + assert keys[0] == keys[1] + assert len(set(keys[2:])) == 4 + assert keys[0] not in keys[2:] -- 2.34.1 From ddac3553a3bd4e2c29a7d2188dffb2c10ee065da Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 19:04:19 -0400 Subject: [PATCH 3/4] .deribit: own cached price-feed subscriptions Give every `maybe_open_price_feed()` caller a lexical child from the cached `LinkedTaskChannel` broadcaster instead of yielding the raw channel to the first owner and constructing competing roots for hits. Add a concurrent regression proving miss and hit callers receive the same quote, close independently, and leave the retained root alive. Prompt-IO: ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260813T225940Z_b691422b_prompt_io.md | 32 +++++ ...20260813T225940Z_b691422b_prompt_io.raw.md | 37 ++++++ piker/brokers/deribit/api.py | 13 +- tests/test_deribit_broadcast.py | 112 ++++++++++++++++++ 4 files changed, 188 insertions(+), 6 deletions(-) create mode 100644 ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.raw.md create mode 100644 tests/test_deribit_broadcast.py diff --git a/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md b/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md new file mode 100644 index 00000000..7fb87d11 --- /dev/null +++ b/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.md @@ -0,0 +1,32 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: unavailable +timestamp: 2026-08-13T22:59:40Z +git_ref: b691422b +scope: code +substantive: true +raw_file: 20260813T225940Z_b691422b_prompt_io.raw.md +--- + +## Prompt + +Continue the piker broadcast-consumer audit after the cached feed ownership +commit and prepare the next smallest ownership fix. + +## Response summary + +Made every Deribit cached price-feed caller own a lexical child subscription +from the retained `LinkedTaskChannel` broadcaster instead of constructing +competing roots around one source channel. + +## Files changed + +- `piker/brokers/deribit/api.py` - uniform child subscription ownership for + price-feed misses and cache hits. +- `tests/test_deribit_broadcast.py` - concurrent fan-out and child lifetime + regression. + +## Human edits + +None - generated output follows the requested broadcast-consumer audit. diff --git a/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.raw.md b/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.raw.md new file mode 100644 index 00000000..8859b242 --- /dev/null +++ b/ai/prompt-io/opencode/20260813T225940Z_b691422b_prompt_io.raw.md @@ -0,0 +1,37 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-13T22:59:40Z +git_ref: b691422b +diff_cmd: git diff HEAD~1..HEAD +--- + +The user asked to continue the piker broadcast-consumer audit after committing +the cached feed ownership fix. The next isolated item was Deribit's cached +price feed, which wrapped one retained `LinkedTaskChannel` in competing +broadcaster roots for cache hits. + +> `git diff HEAD~1..HEAD -- piker/brokers/deribit/api.py` + +Changed `maybe_open_price_feed()` so both the first cache owner and every cache +hit enter `LinkedTaskChannel.subscribe()` and receive a lexical child from the +channel's single retained broadcaster root. + +> `git diff HEAD~1..HEAD -- tests/test_deribit_broadcast.py` + +Added a concurrent ownership regression using real Tractor broadcast state. +It proves a miss owner and cache-hit caller receive the same quote, closing the +hit child leaves the owner live, and both children clean up independently while +the channel root remains registered. + +Verification output: + +```text +. [100%] +1 passed in 0.01s +``` + +The new test passes through a source-isolated Deribit import. Test Ruff, +Python compilation and whitespace checks pass. Full package import and +full-file Deribit Ruff remain blocked by pre-existing Qt, msgspec model, and +legacy Deribit lint errors unrelated to this patch. diff --git a/piker/brokers/deribit/api.py b/piker/brokers/deribit/api.py index 85d490be..fc0c3a4e 100644 --- a/piker/brokers/deribit/api.py +++ b/piker/brokers/deribit/api.py @@ -36,6 +36,7 @@ from trio_typing import TaskStatus from rapidfuzz import process as fuzzy import numpy as np from tractor.trionics import ( + BroadcastReceiver, broadcast_receiver, maybe_open_context, collapse_eg, @@ -589,20 +590,20 @@ async def open_price_feed( @acm async def maybe_open_price_feed( instrument: str -) -> trio.abc.ReceiveStream: +) -> BroadcastReceiver: # TODO: add a predicate to maybe_open_context + feed: to_asyncio.LinkedTaskChannel async with maybe_open_context( acm_func=open_price_feed, kwargs={ 'instrument': instrument }, key=f'{instrument}-price', - ) as (cache_hit, feed): - if cache_hit: - yield broadcast_receiver(feed, 10) - else: - yield feed + ) as (_, feed): + bstream: BroadcastReceiver + async with feed.subscribe() as bstream: + yield bstream diff --git a/tests/test_deribit_broadcast.py b/tests/test_deribit_broadcast.py new file mode 100644 index 00000000..44ffdd6e --- /dev/null +++ b/tests/test_deribit_broadcast.py @@ -0,0 +1,112 @@ +''' +Deribit cached feed broadcast ownership regressions. + +''' +from contextlib import asynccontextmanager as acm + +import tractor +import trio + +from piker.brokers.deribit import api + + +def test_price_feed_owns_child_subscription( + monkeypatch, +) -> None: + ''' + Every cached price-feed caller must own one lexical child receiver. + + `maybe_open_price_feed()` previously yielded the raw + `LinkedTaskChannel` to the first cache owner, then constructed a new + broadcaster root around that same channel for every cache hit. + Concurrent roots competed for source receives, so quotes could be + split across callers and exiting one wrapper did not express child + ownership on the retained channel broadcaster. + + Return one real `LinkedTaskChannel`-shaped fake first as a miss and + then as a concurrent hit. Nest both caller contexts and prove each + gets a distinct child from the channel's single retained root and + receives the same quote. Close the hit child, then prove the first + owner continues receiving before its own lexical child closes. + + ''' + class FakeLinkedTaskChannel: + ''' + Retain one real broadcaster and expose its public subscription. + + ''' + 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 quote through the retained source channel. + + ''' + await self._tx.send(msg) + + channel: FakeLinkedTaskChannel = FakeLinkedTaskChannel() + context_entries: int = 0 + + @acm + async def maybe_open_context(**kwargs): + ''' + Return the controlled channel with either cache state. + + ''' + nonlocal context_entries + cache_hit: bool = context_entries > 0 + context_entries += 1 + yield cache_hit, channel + + monkeypatch.setattr( + api, + 'maybe_open_context', + maybe_open_context, + ) + + async def main() -> None: + owner: tractor.trionics.BroadcastReceiver + async with api.maybe_open_price_feed( + 'btc-usd', + ) as owner: + hit: tractor.trionics.BroadcastReceiver + async with api.maybe_open_price_feed( + 'btc-usd', + ) as hit: + assert hit is not owner + assert len(channel._broadcaster._state.subs) == 3 + + await channel.push({'value': 1}) + assert await owner.receive() == {'value': 1} + assert await hit.receive() == {'value': 1} + + assert hit._closed + assert not owner._closed + assert len(channel._broadcaster._state.subs) == 2 + + await channel.push({'value': 2}) + assert await owner.receive() == {'value': 2} + + assert owner._closed + assert len(channel._broadcaster._state.subs) == 1 + + trio.run(main) -- 2.34.1 From 4af885d90f6e4bcfa963f2f7a1d40436d3c928f3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 19:16:28 -0400 Subject: [PATCH 4/4] .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) -- 2.34.1