pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 12 Packages Projects Releases Wiki Activity

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`))
wkt/replay_provider_e2e
Gud Boi 2026-09-02 14:52:18 -04:00
parent a36c444b78
commit 61181ea54e
1 changed files with 74 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
tests/test_gap_overlays.py
View File

@ -2,9 +2,6 @@
Typed chart-local gap-overlay regressions. Typed chart-local gap-overlay regressions.
''' '''
from collections.abc import (
Callable,
)
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from types import SimpleNamespace from types import SimpleNamespace
@ -799,28 +796,28 @@ async def _delayed_gap_dialog(
) -> None: ) -> None:
''' '''
Delay the first reply so client cancellation leaves it queued. Hold the first reply until a second request arrives.
''' '''
await ctx.started([FQME]) await ctx.started([FQME])
stream: tractor.MsgStream stream: tractor.MsgStream
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
with ctx.pld_rx.limit_plds(spec=SetGapOverlay): with ctx.pld_rx.limit_plds(spec=SetGapOverlay):
req_count: int = 0 first: SetGapOverlay = await stream.receive()
req: SetGapOverlay _delayed_request_ids.append(first.request_id)
async for req in stream: await stream.send(GapOverlay(
req_count += 1 fqme=first.fqme,
_delayed_request_ids.append(req.request_id) timeframe=first.timeframe,
if req_count == 1: visible=first.visible,
await stream.send(GapOverlay( gap_count=0,
fqme=req.fqme, request_id=(
timeframe=req.timeframe, f'actor-received:{first.request_id}'
visible=req.visible, ),
gap_count=0, ))
request_id='actor-received',
))
await trio.sleep(0.05)
second: SetGapOverlay = await stream.receive()
_delayed_request_ids.append(second.request_id)
for req in (first, second):
await stream.send(GapOverlay( await stream.send(GapOverlay(
fqme=req.fqme, fqme=req.fqme,
timeframe=req.timeframe, timeframe=req.timeframe,
@ -829,12 +826,15 @@ async def _delayed_gap_dialog(
request_id=req.request_id, request_id=req.request_id,
)) ))
async for unexpected in stream:
raise AssertionError(
'unexpected third delayed request: '
f'{unexpected!r}'
)
@tractor_test(timeout=20) @tractor_test(timeout=20)
async def test_remote_gap_dialog_real_actor( async def test_remote_gap_dialog_real_actor() -> None:
monkeypatch: pytest.MonkeyPatch,
) -> None:
''' '''
Exchange typed gap state through the production Tractor endpoint. 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 the child receives the generated request ID and removes its owner
after stream closure. after stream closure.
A second child context emits a receipt before delaying the first A second child context emits a typed receipt containing the first
reply. Cancel that exact client task after the receipt, submit a generated request ID, then blocks on receipt of a second request
second request on the shared stream, and prove the client 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 discards the queued first reply before returning the second. The
child-side request IDs prove cancellation happened after 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 ( from piker.ui._remote_ctl import (
AnnotClient, AnnotClient,
remote_gap_overlays, 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 actor_nursery: tractor.ActorNursery
async with tractor.open_nursery() as actor_nursery: async with tractor.open_nursery() as actor_nursery:
portal: tractor.Portal = await actor_nursery.start_actor( portal: tractor.Portal = await actor_nursery.start_actor(
@ -955,28 +932,59 @@ async def test_remote_gap_dialog_real_actor(
first_scope: trio.CancelScope = ( first_scope: trio.CancelScope = (
trio.CancelScope() trio.CancelScope()
) )
first_done: trio.Event = trio.Event()
async def cancel_first_request() -> None: async def cancel_first_request() -> None:
''' '''
Wait for cancellation inside the first Publish once and expose exact cancellation.
dialog.
''' '''
with first_scope: try:
await delayed_client.set_gap_overlay( with first_scope:
SetGapOverlay( await (
fqme=FQME, delayed_client.set_gap_overlay(
timeframe=60, SetGapOverlay(
specs=[], fqme=FQME,
timeframe=60,
specs=[],
)
)
) )
) finally:
first_done.set()
nursery: trio.Nursery nursery: trio.Nursery
async with trio.open_nursery() as nursery: async with trio.open_nursery() as nursery:
nursery.start_soon(cancel_first_request) receipt: GapOverlay
await receipt_seen.wait() 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() first_scope.cancel()
await wait_all_tasks_blocked() await first_done.wait()
assert first_scope.cancelled_caught
recovered: GapOverlay = ( recovered: GapOverlay = (
await delayed_client.set_gap_overlay( await delayed_client.set_gap_overlay(
@ -989,7 +997,6 @@ async def test_remote_gap_dialog_real_actor(
) )
assert recovered.fqme == FQME assert recovered.fqme == FQME
assert recovered.request_id assert recovered.request_id
nursery.cancel_scope.cancel()
delayed_snapshot: dict = await portal.run( delayed_snapshot: dict = await portal.run(
_gap_actor_snapshot, _gap_actor_snapshot,
@ -998,6 +1005,7 @@ async def test_remote_gap_dialog_real_actor(
delayed_snapshot['delayed_request_ids'] delayed_snapshot['delayed_request_ids']
) )
assert len(delayed_ids) == 2 assert len(delayed_ids) == 2
assert delayed_ids[0] == first_request_id
assert delayed_ids[0] != delayed_ids[1] assert delayed_ids[0] != delayed_ids[1]
assert delayed_ids[1] == recovered.request_id assert delayed_ids[1] == recovered.request_id
finally: finally: