From ee33492c81b9af7f538a00db7d387c451fbcf426 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 3 Sep 2026 15:32:15 -0400 Subject: [PATCH] Synchronize `replay` with feed subscribers Deats, - notify optional backend observers after subscriber mutations - gate replay advancement on real `Feed.pause()` state changes - exercise `pikerd`, `datad`, `samplerd`, SHM and typed IPC - prove repeatable transcripts, acknowledged faults and teardown Prompt-IO: ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260903T192743Z_aadfd1cf_prompt_io.md | 42 ++ ...20260903T192743Z_aadfd1cf_prompt_io.raw.md | 35 ++ piker/data/feed.py | 27 ++ tests/replay/conftest.py | 31 ++ tests/replay/test_provider.py | 370 ++++++++++++++++++ 5 files changed, 505 insertions(+) create mode 100644 ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.raw.md create mode 100644 tests/replay/conftest.py create mode 100644 tests/replay/test_provider.py diff --git a/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md b/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md new file mode 100644 index 00000000..0e75bbc1 --- /dev/null +++ b/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md @@ -0,0 +1,42 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: 8737d0a9-98da-4fa1-ba35-ca8a3a42bbc9 +timestamp: 2026-09-03T19:27:43Z +git_ref: wkt/replay_provider_e2e +scope: code +substantive: true +raw_file: 20260903T192743Z_aadfd1cf_prompt_io.raw.md +--- + +## Prompt + +Prove the replay provider through the real Piker feed runtime rather +than monkeypatching `open_feed()`: exercise datad, samplerd, history and +quote SHM, search, pause/resume, typed failures, repeatability, and clean +actor/resource teardown without sleeps or log assertions. + +## Response summary + +Connects feed subscription mutations to an optional provider observer +and adds real Tractor integration tests. The suite compares complete +replay generations, verifies production pause/resume barriers, exercises +typed producer controls, and proves acknowledged failure ordering and +named-SHM cleanup. + +## Files changed + +- `piker/data/feed.py` - notify an optional backend subscriber-state + observer after add/remove operations. +- `tests/replay/conftest.py` - select tracked replay scenarios for child + actors. +- `tests/replay/test_provider.py` - exercise real feed services, SHM, + controls, repeatability, failures, and teardown. + +## Human edits + +The human reviewed the completed integration, questioned whether fake +provider traffic offered enough real bug-detection value, and directed +the work to stop before additional replay features. The human requires +future E2E effort to use live broker machinery wherever practical and +to derive focused data tests from observed real-world defects. diff --git a/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.raw.md b/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.raw.md new file mode 100644 index 00000000..dd8cf9b7 --- /dev/null +++ b/ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.raw.md @@ -0,0 +1,35 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-09-03T19:27:43Z +git_ref: wkt/replay_provider_e2e +diff_cmd: git diff HEAD~1..HEAD +--- + +## Generated changes + +> `git diff HEAD~1..HEAD -- piker/data/feed.py tests/replay/` + +Adds an optional backend notification at real feed-bus subscriber +mutations so replay can acknowledge the production `Feed.pause()` and +`Feed.resume()` state. Integration fixtures select tracked scenarios +through inherited configuration, while real actor tests exercise +pikerd, datad, samplerd, history and quote SHM, search, typed replay +control, failure acknowledgement, transcript equality, and exact +resource teardown. + +## Verification + +- The basic scenario passes twice through fresh actor-tree generations + with identical control, quote, and history transcripts. +- The failure scenario acknowledges its declared fault before any + failed quote reaches SHM. +- Exact datad-created SHM names disappear after each generation. +- The combined replay suite passes four tests; remaining warnings are + existing Tractor and Piker deprecations. + +## Design discussion + +The user concluded this middleware-focused work should not delay real +Qt/Trio application testing. The integration remains a completed parked +experiment rather than the foundation for the immediate chart roadmap. diff --git a/piker/data/feed.py b/piker/data/feed.py index d31d58a3..dd608e28 100644 --- a/piker/data/feed.py +++ b/piker/data/feed.py @@ -38,6 +38,7 @@ from typing import ( Any, AsyncContextManager, Awaitable, + Callable, Sequence, TYPE_CHECKING, ) @@ -506,6 +507,28 @@ async def open_feed_bus( assert brokername in servicename bus: _FeedsBus = get_feed_bus(brokername) + try: + mod: ModuleType = get_brokermod(brokername) + except ImportError: + mod = get_ingestormod(brokername) + + on_sub_change: Callable[[str, bool], None] | None = getattr( + mod, + 'on_feed_subscription_change', + None, + ) + + def notify_sub_change(bs_fqme: str) -> None: + ''' + Notify an optional backend subscriber-state observer. + + ''' + if on_sub_change is not None: + on_sub_change( + bs_fqme, + bool(bus.get_subs(bs_fqme)), + ) + sub_registered = trio.Event() flumes: dict[str, Flume] = {} @@ -638,6 +661,7 @@ async def open_feed_bus( bs_fqme, {sub} ) + notify_sub_change(bs_fqme) # sync caller with all subs registered state sub_registered.set() @@ -654,12 +678,14 @@ async def open_feed_bus( log.info( f'Pausing {bs_fqme} feed for {uid}') bus.remove_subs(bs_fqme, subs) + notify_sub_change(bs_fqme) elif msg == 'resume': for bs_fqme, subs in local_subs.items(): log.info( f'Resuming {bs_fqme} feed for {uid}') bus.add_subs(bs_fqme, subs) + notify_sub_change(bs_fqme) else: raise ValueError(msg) @@ -675,6 +701,7 @@ async def open_feed_bus( # drop all subs for this task from the bus for bs_fqme, subs in local_subs.items(): bus.remove_subs(bs_fqme, subs) + notify_sub_change(bs_fqme) class Feed(Struct): diff --git a/tests/replay/conftest.py b/tests/replay/conftest.py new file mode 100644 index 00000000..adb98dbd --- /dev/null +++ b/tests/replay/conftest.py @@ -0,0 +1,31 @@ +''' +Offline replay integration fixtures. + +''' +from pathlib import Path + +import pytest + + +@pytest.fixture +def replay_scenario( + request: pytest.FixtureRequest, + monkeypatch: pytest.MonkeyPatch, +) -> Path: + ''' + Select one tracked scenario for parent and child actors. + + ''' + inputs: Path = ( + Path(__file__).parents[1] + / '_inputs' + / 'replay' + ) + path: Path = inputs / f'{request.param}.json' + if not path.is_file(): + raise FileNotFoundError(path) + monkeypatch.setenv( + 'PIKER_REPLAY_SCENARIO', + str(path), + ) + return path diff --git a/tests/replay/test_provider.py b/tests/replay/test_provider.py new file mode 100644 index 00000000..549db3eb --- /dev/null +++ b/tests/replay/test_provider.py @@ -0,0 +1,370 @@ +''' +Real service-tree coverage for the offline replay provider. + +''' +from contextlib import AbstractAsyncContextManager +from multiprocessing.shared_memory import SharedMemory +from pathlib import Path +from typing import ( + Any, + Callable, +) + +import msgspec +import numpy as np +import pytest +import tractor +import trio + +from piker.brokers import replay +from piker.data import open_feed +from piker.service import Services + + +FQME: str = 'btcusd.test.replay' + + +async def _request( + stream: tractor.MsgStream, + command: replay.ReplayCommand, +) -> replay.ReplayAck: + ''' + Send one replay command and receive its correlated reply. + + ''' + await stream.send(command) + ack: replay.ReplayAck = await stream.receive() + assert ack.command_id == command.command_id + return ack + + +async def _search_replay_symbols( + portal: tractor.Portal, +) -> dict[str, Any]: + ''' + Exercise the backend's normal symbol-search context. + + ''' + ctx: tractor.Context + async with portal.open_context( + replay.open_symbol_search, + ) as (ctx, _): + async with ctx.open_stream() as stream: + await stream.send('btc') + return await stream.receive() + + +async def _run_basic_generation( + open_test_pikerd: Callable[ + ..., + AbstractAsyncContextManager, + ], + loglevel: str, +) -> tuple[bytes, list[dict[str, Any]], list[float]]: + ''' + Run one complete replay through real data services. + + ''' + quote_transcript: list[dict[str, Any]] = [] + shm_names: list[str] = [] + services: Services + async with ( + open_test_pikerd() as (_, _, _, services), + open_feed( + [FQME], + loglevel=loglevel, + ) as feed, + ): + assert 'datad.replay' in services.service_tasks + assert 'samplerd' in services.service_tasks + + stream: tractor.MsgStream = feed.streams['replay'] + first_msg: dict[str, Any] = await stream.receive() + quote_transcript.append(first_msg) + assert first_msg[FQME]['replay_seq'] == 1 + + flume = feed.flumes[FQME] + history_closes: list[float] = ( + flume.hist_shm.array['close'].tolist() + ) + shm_names.extend([ + flume.hist_shm._shm.name, + flume.rt_shm._shm.name, + ]) + assert history_closes == [ + 99.2, + 99.4, + 99.6, + 99.8, + 100.2, + 100.6, + ] + + portal: tractor.Portal = feed.portals[replay] + matches: dict[str, Any] = await _search_replay_symbols( + portal + ) + assert list(matches) == [FQME] + + control_ctx: tractor.Context + started: replay.ReplaySnapshot + async with portal.open_context( + replay.open_replay_control, + ) as (control_ctx, started): + assert started.state == 'ready' + assert started.event_sequence == 1 + + async with control_ctx.open_stream() as control: + with control_ctx.pld_rx.limit_plds( + spec=replay.ReplayAck, + ): + await feed.pause() + paused: replay.ReplayAck = await _request( + control, + replay.AwaitState( + command_id=1, + state='paused', + ), + ) + assert paused.ok + assert not paused.snapshot.subscriber_active + + rejected: replay.ReplayAck = await _request( + control, + replay.Advance( + command_id=2, + event_sequence=2, + ), + ) + assert not rejected.ok + assert rejected.snapshot.event_sequence == 1 + + await feed.resume() + ready: replay.ReplayAck = await _request( + control, + replay.AwaitState( + command_id=3, + state='ready', + ), + ) + assert ready.ok + assert ready.snapshot.subscriber_active + + producer_paused = await _request( + control, + replay.Pause(command_id=4), + ) + assert producer_paused.snapshot.state == 'paused' + gate_rejected = await _request( + control, + replay.Advance( + command_id=5, + event_sequence=2, + ), + ) + assert not gate_rejected.ok + producer_ready = await _request( + control, + replay.Resume(command_id=6), + ) + assert producer_ready.snapshot.state == 'ready' + + for command_id, event_sequence in ( + (7, 2), + (8, 3), + (9, 4), + ): + advanced: replay.ReplayAck = ( + await _request( + control, + replay.Advance( + command_id=command_id, + event_sequence=( + event_sequence + ), + ), + ) + ) + assert advanced.ok + quote: dict[str, Any] = ( + await stream.receive() + ) + quote_transcript.append(quote) + assert ( + quote[FQME]['replay_seq'] + == event_sequence + ) + + snap_ack: replay.ReplayAck = await _request( + control, + replay.Snapshot(command_id=10), + ) + assert snap_ack.snapshot.state == 'exhausted' + transcript: bytes = msgspec.json.encode( + snap_ack.snapshot + ) + + np.testing.assert_allclose( + flume.rt_shm.array['close'][-1], + 101.0, + ) + + assert not services.service_tasks + for shm_name in shm_names: + with pytest.raises(FileNotFoundError): + SharedMemory(name=shm_name) + + return transcript, quote_transcript, history_closes + + +@pytest.mark.parametrize( + 'replay_scenario', + ['basic-v1'], + indirect=True, +) +def test_real_feed_replay_is_repeatable( + replay_scenario: Path, + open_test_pikerd: Callable[ + ..., + AbstractAsyncContextManager, + ], + loglevel: str, +) -> None: + ''' + Replay identical typed transcripts through fresh actor trees. + + Mocked feed calls can hide backend discovery, datad and samplerd + startup, history publication, SHM writes, stream subscription + control, and context teardown failures. Run the same tracked + scenario through two fresh `open_test_pikerd()` generations. + Each generation receives the initial quote, pauses the production + feed, awaits the typed subscriber-state barrier, proves + advancement is rejected, resumes, repeats that gate check + through typed producer controls, and explicitly advances every + remaining event. Exact control bytes, quote dictionaries, and + history values must match, while service and named-SHM checks + prove each generation fully tears down before the next one + starts. + + ''' + assert replay_scenario.name == 'basic-v1.json' + + async def main() -> None: + first = await _run_basic_generation( + open_test_pikerd, + loglevel, + ) + second = await _run_basic_generation( + open_test_pikerd, + loglevel, + ) + assert first == second + + trio.run(main) + + +@pytest.mark.parametrize( + 'replay_scenario', + ['failure-v1'], + indirect=True, +) +def test_failure_injection_is_acknowledged( + replay_scenario: Path, + open_test_pikerd: Callable[ + ..., + AbstractAsyncContextManager, + ], + loglevel: str, +) -> None: + ''' + Stop at a fixture failure without publishing its quote. + + An injected provider disconnect used to require timing a task + crash and inferring its position from logs. Start the real feed, + arm the fixture's third event through typed control, publish + event two, then request event three. Its correlated negative + acknowledgement must identify the fixture error, retain event + sequence two, and expose `failed` state. Reading SHM after the + acknowledgement proves the rejected event was not hidden in the + sampler before normal actor and SHM teardown. + + ''' + assert replay_scenario.name == 'failure-v1.json' + + async def main() -> None: + services: Services + shm_names: list[str] = [] + async with ( + open_test_pikerd() as (_, _, _, services), + open_feed( + [FQME], + loglevel=loglevel, + ) as feed, + ): + quotes: tractor.MsgStream = feed.streams['replay'] + first: dict[str, Any] = await quotes.receive() + assert first[FQME]['replay_seq'] == 1 + + flume = feed.flumes[FQME] + shm_names.extend([ + flume.hist_shm._shm.name, + flume.rt_shm._shm.name, + ]) + portal: tractor.Portal = feed.portals[replay] + ctx: tractor.Context + async with portal.open_context( + replay.open_replay_control, + ) as (ctx, _): + async with ctx.open_stream() as control: + with ctx.pld_rx.limit_plds( + spec=replay.ReplayAck, + ): + armed: replay.ReplayAck = await _request( + control, + replay.FailAt( + command_id=1, + event_sequence=3, + ), + ) + assert armed.ok + + advanced: replay.ReplayAck = ( + await _request( + control, + replay.Advance( + command_id=2, + event_sequence=2, + ), + ) + ) + assert advanced.ok + second: dict[str, Any] = ( + await quotes.receive() + ) + assert second[FQME]['replay_seq'] == 2 + + failed: replay.ReplayAck = await _request( + control, + replay.Advance( + command_id=3, + event_sequence=3, + ), + ) + assert not failed.ok + assert failed.snapshot.state == 'failed' + assert failed.snapshot.event_sequence == 2 + assert failed.error == ( + 'fixture_disconnect: offline provider ' + 'disconnected' + ) + assert ( + flume.rt_shm.array['close'][-1] + == 100.5 + ) + + assert not services.service_tasks + for shm_name in shm_names: + with pytest.raises(FileNotFoundError): + SharedMemory(name=shm_name) + + trio.run(main)