`.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`))backfiller_deep_fixes
parent
976db6c33c
commit
931f22d235
|
|
@ -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.
|
||||
|
|
@ -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.
|
||||
|
|
@ -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..
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue