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:]