.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`))

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.

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.

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
 
 
 

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)