diff --git a/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md b/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md new file mode 100644 index 00000000..e09327b2 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md @@ -0,0 +1,34 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-08-12T21:31:17Z +git_ref: 51185487 +scope: code +substantive: true +raw_file: 20260812T213117Z_51185487_prompt_io.raw.md +--- + +## Prompt + +For Tractor PR 490 item four, expose `raise_on_lag` through the public IPC and +asyncio linked-channel subscription wrappers. Review, verify and return a +complete commit plan before applying the API downstream in piker. + +## Response summary + +Added public per-subscription lag policy to both wrappers, preserved first-call +root policy, documented the semantics, and covered forwarding plus real fan-out +paths. + +## Files changed + +- `tractor/_streaming.py` - `MsgStream` lag policy forwarding. +- `tractor/to_asyncio.py` - linked-channel lag policy forwarding. +- `docs/guide/streaming.rst` - IPC fan-out policy docs. +- `docs/guide/asyncio.rst` - linked-channel fan-out policy docs. +- `tests/test_task_broadcasting.py` - wrapper policy regression. + +## Human edits + +None - generated output follows the requested fourth iterative item. diff --git a/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.raw.md b/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.raw.md new file mode 100644 index 00000000..a1725ad8 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.raw.md @@ -0,0 +1,53 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-12T21:31:17Z +git_ref: 51185487 +diff_cmd: git diff HEAD~1..HEAD +--- + +The user requested the fourth iterative refinement for Tractor PR 490: expose +subscriber lag policy through the public `MsgStream.subscribe()` and +`LinkedTaskChannel.subscribe()` wrappers, then stop for a complete commit +plan. This enables piker to replace private receiver mutation. + +> `git diff HEAD~1..HEAD -- tractor/_streaming.py` + +Added `raise_on_lag: bool = True` to `MsgStream.subscribe()`. The first call +passes the policy to the irreversibly allocated root broadcaster and its +child; later calls configure each child independently while retaining the +root's first-call policy. + +> `git diff HEAD~1..HEAD -- tractor/to_asyncio.py` + +Added equivalent lag-policy forwarding to `LinkedTaskChannel.subscribe()`. + +> `git diff HEAD~1..HEAD -- docs/guide/streaming.rst` + +> `git diff HEAD~1..HEAD -- docs/guide/asyncio.rst` + +Documented strict versus warn/drop/resume behavior, independent child policy, +and first-call root policy for both wrapper types. + +> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py` + +Added a parameterized wrapper-level regression using minimal receive-compatible +handles. It verifies a first non-raising subscription configures root and +child, then a later strict child does not mutate the sticky root policy. + +Verification output: + +```text +................... [100%] +19 passed in 5.71s + +.................... [100%] +20 passed in 6.88s + +. [100%] +1 passed in 0.86s +``` + +The second and third runs cover actual `MsgStream` and infected-asyncio +`LinkedTaskChannel` fan-out respectively. Python compilation and +`git diff --check` passed. Adversarial review found no actionable issues. diff --git a/docs/guide/asyncio.rst b/docs/guide/asyncio.rst index 27e1d05f..9877e7a4 100644 --- a/docs/guide/asyncio.rst +++ b/docs/guide/asyncio.rst @@ -209,6 +209,11 @@ The underlying broadcast machinery is lazily allocated on first use and is *not* reversible for the channel's remaining lifetime, so only reach for it when you actually want the fan-out. +As with ``MsgStream``, pass ``raise_on_lag=False`` for a consumer +which may warn, drop old values and resume from the retained window. +Each child chooses independently; the first subscription also fixes +the linked channel's root receive policy. + One-shot calls with ``run_task()`` ---------------------------------- When you just want a single ``asyncio`` result and no streaming diff --git a/docs/guide/streaming.rst b/docs/guide/streaming.rst index 1c071268..17d35ef8 100644 --- a/docs/guide/streaming.rst +++ b/docs/guide/streaming.rst @@ -171,6 +171,12 @@ keeps pace with the *fastest* subscriber; a task falling more than the buffered window behind has its next receive raise ``tractor.trionics.Lagged`` to say it lost data. +Pass ``raise_on_lag=False`` when a consumer may drop old values and +resume from the oldest retained item instead. The receiver logs the +overrun rather than raising. Each child subscription chooses its own +policy; the first call also fixes the policy of the stream's root +receive handle because broadcaster allocation is irreversible. + The broadcast handle stays duplex btw: it proxies ``send()`` through to the underlying stream, so each subscriber task can keep talking upstream while consuming its fan-out copy. diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index 8d331f31..f6d0483e 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -8,6 +8,7 @@ from contextlib import ( from functools import partial from itertools import cycle import time +from types import SimpleNamespace from typing import Optional import warnings @@ -15,6 +16,7 @@ import pytest import trio from trio.lowlevel import current_task import tractor +from tractor.to_asyncio import LinkedTaskChannel from tractor.trionics import ( broadcast_receiver, BroadcastReceiveError, @@ -952,3 +954,74 @@ def test_no_raise_on_lag(): with pytest.raises(KeyboardInterrupt): trio.run(main) + + +@pytest.mark.parametrize( + ('subscribe', 'chan_attr'), + [ + (tractor.MsgStream.subscribe, '_rx_chan'), + (LinkedTaskChannel.subscribe, '_from_aio'), + ], + ids=['msg-stream', 'linked-task-channel'], +) +def test_stream_subscribe_forwards_lag_policy( + subscribe, + chan_attr: str, +) -> None: + ''' + Stream wrappers must expose per-subscriber lag policy. + + `MsgStream.subscribe()` and `LinkedTaskChannel.subscribe()` + previously omitted `BroadcastReceiver.raise_on_lag`, forcing + downstream users to mutate a private receiver attribute. Invoke + each public wrapper against a minimal receive-compatible handle. + Prove the first non-raising subscription configures both the + irreversible root broadcaster and its child, while a later strict + child selects its own policy without changing that root. + + ''' + class StreamHandle: + ''' + Provide the wrapper fields needed for local fan-out. + + ''' + def __init__(self) -> None: + self._broadcaster = None + setattr( + self, + chan_attr, + SimpleNamespace( + _state=SimpleNamespace(max_buffer_size=1), + ), + ) + + async def receive(self): + ''' + Block if a regression unexpectedly enters source receive. + + ''' + await trio.sleep_forever() + + async def send(self, value) -> None: + ''' + Satisfy `MsgStream` duplex-handle patching. + + ''' + + async def main() -> None: + stream = StreamHandle() + async with subscribe( + stream, + raise_on_lag=False, + ) as first: + assert not stream._broadcaster._raise_on_lag + assert not first._raise_on_lag + + async with subscribe( + stream, + raise_on_lag=True, + ) as second: + assert not stream._broadcaster._raise_on_lag + assert second._raise_on_lag + + trio.run(main) diff --git a/tractor/_streaming.py b/tractor/_streaming.py index adc77dcd..a5e94f19 100644 --- a/tractor/_streaming.py +++ b/tractor/_streaming.py @@ -512,6 +512,7 @@ class MsgStream(trio.abc.Channel): @acm async def subscribe( self, + raise_on_lag: bool = True, ) -> AsyncIterator[BroadcastReceiver]: ''' @@ -526,6 +527,11 @@ class MsgStream(trio.abc.Channel): value from the far end via the internally created broudcast receiver wrapper. + ``raise_on_lag=False`` makes this subscription warn and resume + at the oldest retained value after an overrun. The first call + also sets that policy for this stream's root receive handle; + later child subscriptions choose their policy independently. + ''' # NOTE: This operation is indempotent and non-reversible, so be # sure you can deal with any (theoretical) overhead of the the @@ -541,6 +547,7 @@ class MsgStream(trio.abc.Channel): # TODO: can remove this kwarg right since # by default behaviour is to do this anyway? receive_afunc=self.receive, + raise_on_lag=raise_on_lag, ) # NOTE: we override the original stream instance's receive @@ -552,7 +559,9 @@ class MsgStream(trio.abc.Channel): # seems there's no graceful way to type this with ``mypy``? # https://github.com/python/mypy/issues/708 - async with self._broadcaster.subscribe() as bstream: + async with self._broadcaster.subscribe( + raise_on_lag=raise_on_lag, + ) as bstream: assert bstream.key != self._broadcaster.key assert bstream._recv == self._broadcaster._recv diff --git a/tractor/to_asyncio.py b/tractor/to_asyncio.py index e2b51e17..28756052 100644 --- a/tractor/to_asyncio.py +++ b/tractor/to_asyncio.py @@ -324,6 +324,7 @@ class LinkedTaskChannel( @acm async def subscribe( self, + raise_on_lag: bool = True, ) -> AsyncIterator[BroadcastReceiver]: ''' @@ -335,6 +336,11 @@ class LinkedTaskChannel( See ``tractor._streaming.MsgStream.subscribe()`` for further similar details. + + ``raise_on_lag=False`` makes this subscription warn and resume + at the oldest retained value after an overrun. The first call + also sets that policy for this channel's root receive handle; + later child subscriptions choose their policy independently. ''' if self._broadcaster is None: @@ -343,11 +349,14 @@ class LinkedTaskChannel( # use memory channel size by default self._from_aio._state.max_buffer_size, # type: ignore receive_afunc=self.receive, + raise_on_lag=raise_on_lag, ) self.receive = bcast.receive # type: ignore - async with self._broadcaster.subscribe() as bstream: + async with self._broadcaster.subscribe( + raise_on_lag=raise_on_lag, + ) as bstream: assert bstream.key != self._broadcaster.key assert bstream._recv == self._broadcaster._recv yield bstream