From 931f22d235cbea188e0012056adbbed8e0f91052 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 21 Jul 2026 22:11:10 -0400 Subject: [PATCH] `.tsp._history`: drop orphaned backfill events - let structured nursery joins own child-task completion - stop empty provider frames before timestamp endpoint indexing - cover provider exhaustion and empty frames with bounded tests Prompt-IO: ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260722T001358Z_16f3cd6f_prompt_io.md | 36 ++++++++++ ...20260722T001358Z_16f3cd6f_prompt_io.raw.md | 25 +++++++ piker/tsp/_history.py | 40 +++++------ tests/test_history_backfill.py | 69 +++++++++++++++++++ 4 files changed, 147 insertions(+), 23 deletions(-) create mode 100644 ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md create mode 100644 tests/test_history_backfill.py diff --git a/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md b/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md new file mode 100644 index 00000000..4ba93cc9 --- /dev/null +++ b/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md @@ -0,0 +1,36 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-07-22T00:13:58Z +git_ref: 16f3cd6f +scope: code +substantive: true +raw_file: 20260722T001358Z_16f3cd6f_prompt_io.raw.md +--- + +## Prompt + +Continue the structured backfill phase after the analysis regressions. +Characterize provider exhaustion and genuinely empty frames, then remove +completion paths that can hang after child-task exit. + +## Response summary + +Removed orphan-prone completion events from the structured nursery flow +and checked empty provider frames before indexing timestamp endpoints. +Added deterministic timeout-bounded regressions for both exit paths. + +## Files changed + +- `piker/tsp/_history.py` - structured completion and empty-frame fixes +- `tests/test_history_backfill.py` - deterministic exit regressions +- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md` + - provenance metadata and response summary + +## Human edits + +None - generated changes have not yet been edited by the human. diff --git a/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md b/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md new file mode 100644 index 00000000..d372d4fe --- /dev/null +++ b/ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md @@ -0,0 +1,25 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-22T00:13:58Z +git_ref: 16f3cd6f +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/tsp/_history.py` + +Removed redundant completion events whose consumers waited only after +their owning nurseries had joined. Added empty provider-frame handling +before timestamp endpoint indexing so provider exhaustion exits cleanly. + +> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py` + +Added bounded, actor-free regressions for `DataUnavailable` and empty +provider-frame completion. + +Verification generated with the patch: + +`timeout -k 5 60 python -m pytest -q tests/test_history_backfill.py` + +Result: 2 passed. diff --git a/piker/tsp/_history.py b/piker/tsp/_history.py index dfa5672b..a4006df0 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -202,12 +202,11 @@ async def maybe_fill_null_segments( mkt: MktPair, backfill_until_dt: datetime, - task_status: TaskStatus[trio.Event] = trio.TASK_STATUS_IGNORED, + task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, ) -> list[Frame]: - null_segs_detected = trio.Event() - task_status.started(null_segs_detected) + task_status.started() frame: Frame = shm.array @@ -308,7 +307,6 @@ async def maybe_fill_null_segments( await tractor.pause() raise - null_segs_detected.set() # RECHECK for more null-gaps frame: Frame = shm.array null_segs: tuple | None = get_null_segs( @@ -407,15 +405,13 @@ async def start_backfill( storage: StorageClient|None = None, write_tsdb: bool = True, - task_status: TaskStatus[tuple] = trio.TASK_STATUS_IGNORED, + task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, ) -> int: - # let caller unblock and deliver latest history frame - # and use to signal that backfilling the shm gap until - # the tsdb end is complete! - bf_done = trio.Event() - task_status.started(bf_done) + # Let the caller place storage history while this task + # continues reverse retrieval in the owning nursery. + task_status.started() # based on the sample step size, maybe load a certain amount history update_start_on_prepend: bool = False @@ -535,6 +531,15 @@ async def start_backfill( # charge of solving such faults? yes, right? return + if array.size == 0: + log.warning( + f'Provider {mod.name!r} returned an empty frame\n' + f'fqme: {mkt.fqme}\n' + f'timeframe: {timeframe}\n' + f'end_dt: {end_dt_param}\n' + ) + break + time: np.ndarray = array['time'] assert ( time[0] @@ -800,10 +805,6 @@ async def start_backfill( # memory... # await sampler_stream.send('broadcast_all') - # short-circuit (for now) - bf_done.set() - - # NOTE: originally this was used to cope with a tsdb (marketstore) # which could not delivery very large frames of history over gRPC # (thanks goolag) due to corruption issues. @@ -1190,7 +1191,7 @@ async def tsdb_backfill( trio.open_nursery() as tn, ): - bf_done: trio.Event = await tn.start( + await tn.start( partial( start_backfill, get_hist=get_hist, @@ -1210,8 +1211,6 @@ async def tsdb_backfill( write_tsdb=True, ) ) - nulls_detected: trio.Event|None = None - if last_tsdb_dt is not None: # calc the index from which the tsdb data should be @@ -1288,7 +1287,7 @@ async def tsdb_backfill( # work PREVENTAVELY instead? # -[ ] fill in non-zero epoch time values ALWAYS! # await maybe_fill_null_segments( - nulls_detected: trio.Event = await tn.start(partial( + await tn.start(partial( maybe_fill_null_segments, shm=shm, @@ -1301,11 +1300,6 @@ async def tsdb_backfill( # 2nd nursery END - # TODO: who would want to? - if nulls_detected: - await nulls_detected.wait() - - await bf_done.wait() # TODO: maybe start history anal and load missing "history # gaps" via backend.. diff --git a/tests/test_history_backfill.py b/tests/test_history_backfill.py new file mode 100644 index 00000000..1a58e460 --- /dev/null +++ b/tests/test_history_backfill.py @@ -0,0 +1,69 @@ +''' +Deterministic history-backfill regressions. + +''' +from functools import partial +from types import SimpleNamespace + +import numpy as np +from pendulum import datetime +import trio + +from piker.brokers import DataUnavailable +from piker.data._source import def_iohlcv_fields +from piker.tsp._history import start_backfill + + +def run_exhausted_backfill(get_hist) -> None: + ''' + Run a provider-exhaustion case without actor services. + + ''' + async def main() -> None: + with trio.fail_after(0.5): + async with trio.open_nursery() as nursery: + await nursery.start(partial( + start_backfill, + get_hist=get_hist, + def_frame_duration=None, + mod=SimpleNamespace(name='fake'), + mkt=SimpleNamespace(fqme='x.fake'), + shm=object(), + timeframe=60, + backfill_from_shm_index=100, + backfill_from_dt=datetime(2026, 1, 2), + sampler_stream=object(), + backfill_until_dt=datetime(2026, 1, 1), + storage=None, + write_tsdb=False, + )) + + trio.run(main) + + +def test_data_unavailable_completes() -> None: + ''' + Provider exhaustion exits without orphaned completion waits. + + ''' + async def get_hist(*args, **kwargs): + raise DataUnavailable('history exhausted') + + run_exhausted_backfill(get_hist) + + +def test_empty_frame_completes() -> None: + ''' + An empty provider frame is checked before endpoint indexing. + + ''' + frame = np.empty( + 0, + dtype=np.dtype(def_iohlcv_fields), + ) + + async def get_hist(*args, **kwargs): + end_dt = kwargs['end_dt'] + return frame, end_dt, end_dt + + run_exhausted_backfill(get_hist)