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)