From 61181ea54e701d9eb3d75088f78e308105e4ae22 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 2 Sep 2026 14:52:18 -0400 Subject: [PATCH] Synchronize cancelled gap requests over IPC Use a typed stream receipt as the publication barrier before cancelling the first shared-stream request. Hold its response until the second request arrives, then prove stale reply filtering and exact cancellation without scheduler sleeps or a patched logger. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_gap_overlays.py | 140 ++++++++++++++++++++----------------- 1 file changed, 74 insertions(+), 66 deletions(-) diff --git a/tests/test_gap_overlays.py b/tests/test_gap_overlays.py index 2b1eea50..9481eb3d 100644 --- a/tests/test_gap_overlays.py +++ b/tests/test_gap_overlays.py @@ -2,9 +2,6 @@ Typed chart-local gap-overlay regressions. ''' -from collections.abc import ( - Callable, -) from contextlib import AsyncExitStack from types import SimpleNamespace @@ -799,28 +796,28 @@ async def _delayed_gap_dialog( ) -> None: ''' - Delay the first reply so client cancellation leaves it queued. + Hold the first reply until a second request arrives. ''' await ctx.started([FQME]) stream: tractor.MsgStream async with ctx.open_stream() as stream: with ctx.pld_rx.limit_plds(spec=SetGapOverlay): - req_count: int = 0 - req: SetGapOverlay - async for req in stream: - req_count += 1 - _delayed_request_ids.append(req.request_id) - if req_count == 1: - await stream.send(GapOverlay( - fqme=req.fqme, - timeframe=req.timeframe, - visible=req.visible, - gap_count=0, - request_id='actor-received', - )) - await trio.sleep(0.05) + first: SetGapOverlay = await stream.receive() + _delayed_request_ids.append(first.request_id) + await stream.send(GapOverlay( + fqme=first.fqme, + timeframe=first.timeframe, + visible=first.visible, + gap_count=0, + request_id=( + f'actor-received:{first.request_id}' + ), + )) + second: SetGapOverlay = await stream.receive() + _delayed_request_ids.append(second.request_id) + for req in (first, second): await stream.send(GapOverlay( fqme=req.fqme, timeframe=req.timeframe, @@ -829,12 +826,15 @@ async def _delayed_gap_dialog( request_id=req.request_id, )) + async for unexpected in stream: + raise AssertionError( + 'unexpected third delayed request: ' + f'{unexpected!r}' + ) + @tractor_test(timeout=20) -async def test_remote_gap_dialog_real_actor( - monkeypatch: pytest.MonkeyPatch, - -) -> None: +async def test_remote_gap_dialog_real_actor() -> None: ''' Exchange typed gap state through the production Tractor endpoint. @@ -846,45 +846,22 @@ async def test_remote_gap_dialog_real_actor( the child receives the generated request ID and removes its owner after stream closure. - A second child context emits a receipt before delaying the first - reply. Cancel that exact client task after the receipt, submit a - second request on the shared stream, and prove the client + A second child context emits a typed receipt containing the first + generated request ID, then blocks on receipt of a second request + before publishing either correlated reply. A subscribed real + `MsgStream` receiver gives the parent an explicit publication + barrier. Cancel that exact client task after the receipt, await + its completion, submit the second request, and prove the client discards the queued first reply before returning the second. The child-side request IDs prove cancellation happened after - publication instead of merely preventing the first send. + publication instead of merely preventing the first send, without + an arbitrary scheduler delay or a patched logger. ''' from piker.ui._remote_ctl import ( AnnotClient, remote_gap_overlays, ) - from piker.ui import _remote_ctl - - receipt_seen: trio.Event = trio.Event() - original_warning: Callable[..., None] = ( - _remote_ctl.log.warning - ) - - def observe_warning( - msg: str, - *args: object, - **kwargs: object, - - ) -> None: - ''' - Observe the stale receipt before cancelling its client task. - - ''' - if 'actor-received' in msg: - receipt_seen.set() - original_warning(msg, *args, **kwargs) - - monkeypatch.setattr( - _remote_ctl.log, - 'warning', - observe_warning, - ) - actor_nursery: tractor.ActorNursery async with tractor.open_nursery() as actor_nursery: portal: tractor.Portal = await actor_nursery.start_actor( @@ -955,28 +932,59 @@ async def test_remote_gap_dialog_real_actor( first_scope: trio.CancelScope = ( trio.CancelScope() ) + first_done: trio.Event = trio.Event() async def cancel_first_request() -> None: ''' - Wait for cancellation inside the first - dialog. + Publish once and expose exact cancellation. ''' - with first_scope: - await delayed_client.set_gap_overlay( - SetGapOverlay( - fqme=FQME, - timeframe=60, - specs=[], + try: + with first_scope: + await ( + delayed_client.set_gap_overlay( + SetGapOverlay( + fqme=FQME, + timeframe=60, + specs=[], + ) + ) ) - ) + finally: + first_done.set() nursery: trio.Nursery async with trio.open_nursery() as nursery: - nursery.start_soon(cancel_first_request) - await receipt_seen.wait() + receipt: GapOverlay + async with ( + delayed_stream.subscribe() + as receipt_stream, + ): + nursery.start_soon( + cancel_first_request, + ) + with ( + delayed_ctx.pld_rx.limit_plds( + spec=GapOverlay, + ) + ): + receipt = ( + await receipt_stream.receive() + ) + + receipt_prefix: str = 'actor-received:' + assert receipt.request_id.startswith( + receipt_prefix, + ) + first_request_id: str = ( + receipt.request_id.removeprefix( + receipt_prefix, + ) + ) + assert first_request_id first_scope.cancel() - await wait_all_tasks_blocked() + await first_done.wait() + assert first_scope.cancelled_caught recovered: GapOverlay = ( await delayed_client.set_gap_overlay( @@ -989,7 +997,6 @@ async def test_remote_gap_dialog_real_actor( ) assert recovered.fqme == FQME assert recovered.request_id - nursery.cancel_scope.cancel() delayed_snapshot: dict = await portal.run( _gap_actor_snapshot, @@ -998,6 +1005,7 @@ async def test_remote_gap_dialog_real_actor( delayed_snapshot['delayed_request_ids'] ) assert len(delayed_ids) == 2 + assert delayed_ids[0] == first_request_id assert delayed_ids[0] != delayed_ids[1] assert delayed_ids[1] == recovered.request_id finally: