diff --git a/.claude/skills/run-tests/test-harness-reference.md b/.claude/skills/run-tests/test-harness-reference.md index 12ccea7c..3b2f9cb6 100644 --- a/.claude/skills/run-tests/test-harness-reference.md +++ b/.claude/skills/run-tests/test-harness-reference.md @@ -107,6 +107,7 @@ Deterministic or local first-pass targets: - `tests/test_storage_audit.py` - `tests/test_backfill_audit_snippet.py` - `tests/test_ib_history.py` +- `tests/test_history_backfill.py` - `tests/test_ldshm.py` - `tests/test_accounting.py::test_account_file_default_empty` - `tests/test_services.py::test_runtime_boot` @@ -191,6 +192,7 @@ tests/ test_ems.py actor, EMS, and paper-position behavior test_feeds.py live Binance/Kraken feeds and shared memory test_ib_history.py deterministic IB history request formatting + test_history_backfill.py deterministic history/SHM orchestration test_ldshm.py ldshm unpublished-slot guard test_questrade.py obsolete credentialed tests; skipped test_services.py pikerd/datad/feed/EMS actor lifecycle @@ -207,7 +209,8 @@ tests/ | `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped | | `piker/storage/_audit.py`, `piker/storage/cli.py` | `tests/test_storage_audit.py` | direct Typer app, no actor | | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only | -| `piker/brokers/ib/api.py` history formatting | `tests/test_ib_history.py` | fake client, no network | +| `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network | +| `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM | | `piker/storage/cli.py` ldshm null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation | | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch | | `piker/accounting/` | targeted accounting node | some cases use live/configured state | diff --git a/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.md b/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.md new file mode 100644 index 00000000..7adb78bd --- /dev/null +++ b/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.md @@ -0,0 +1,37 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: 253ade33-07ba-45b5-8839-be35d17ec164 +timestamp: 2026-07-29T20:40:46Z +git_ref: ad6c560f +scope: code +substantive: true +raw_file: 20260729T204046Z_ad6c560f_prompt_io.raw.md +--- + +## Prompt + +The user reported that the QQQ 1-second chart still contained a gap after +the `ldshm` fix, asked the agent to inspect it, and authorized continued +implementation after a short pause and shutdown of the live workspace. + +## Response summary + +Diagnosed a 653-row active RT SHM null reservation caused by inclusive IB +reverse-query overlaps and startup repair orchestration. Prevented extra +physical SHM overlap rows, serialized and bounded IB history recovery, +made blank/error responses terminate, and constrained synthetic repair to +valid interior timestamp boundaries. Added deterministic concurrency and +data-layout regressions. + +## Files changed + +- `piker/brokers/ib/feed.py` - bound and serialize IB history reset/retry behavior. +- `piker/tsp/_history.py` - prevent SHM overlap skew and safely finalize null repair. +- `tests/test_ib_history.py` - cover blank, cancellation, reset, and pacing interleavings. +- `tests/test_history_backfill.py` - cover overlap and null-repair invariants. +- `.claude/skills/run-tests/test-harness-reference.md` - register deterministic coverage. + +## Human edits + +None - generated output remains uncommitted. diff --git a/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.raw.md b/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.raw.md new file mode 100644 index 00000000..9d17b68c --- /dev/null +++ b/ai/prompt-io/opencode/20260729T204046Z_ad6c560f_prompt_io.raw.md @@ -0,0 +1,72 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-07-29T20:40:46Z +git_ref: ad6c560f +diff_cmd: git diff HEAD~1..HEAD +--- + +The user reported that QQQ still displayed a gap on the 1-second chart +after the focused `ldshm` fix and asked the agent to inspect it. The user +then paused and resumed implementation while shutting down the live piker +trading workspace. + +Read-only live inspection found the active datad generation held 653 +contiguous zero-time rows in QQQ's 1-second RT SHM, while durable Parquet +contained no zero rows. The run occupied absolute indexes 168148 through +168800 between timestamps 1785348398 and 1785349051. The stale actor +generation and active 60-second history were separately identified. + +> `git diff HEAD~1..HEAD -- piker/tsp/_history.py` + +Generated history changes that exclude each inclusive reverse-query +endpoint already present in SHM while retaining the full provider frame +for NativeDB merge. Startup null repair now runs after reverse backfill, +uses bounded UI notifications, handles provider no-data without an +interactive pause, fills only exact interior zero groups, and refuses +synthetic timestamps unless both valid boundaries establish the expected +cadence. + +> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py` + +Generated IB history reset changes that make blank responses complete, +bound repeated cancellation and reset failures per request, serialize +timeout and pacing farm resets with one actor-global Trio lock, carry +explicit reset results, and coordinate pacing handoff with queued timeout +waiters. + +> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py` + +Generated regressions for inclusive endpoint exclusion, full NativeDB +provider deltas, strict synthetic gap boundaries, empty-provider local +fallback, bounded notifications, and preservation of valid margin rows. + +> `git diff HEAD~1..HEAD -- tests/test_ib_history.py` + +Generated deterministic regressions for blank IB responses, cancellation +limits, stalled reset cancellation, pacing reset failure limits, +cross-request reset serialization, timeout-to-pacing handoff, and +successful handoff timeout restart. + +> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md` + +Registered the deterministic history-backfill target and expanded the IB +history mapping to include `feed.py`. + +Verification output: + +```text +........................................................................ [ 93%] +..... [100%] +77 passed in 2.12s +``` + +`git diff --check` passed. Multiple adversarial review rounds identified +and drove fixes for unbounded IB empty/cancellation/reset paths, stale +reset ownership, actor-global failure poisoning, concurrent farm resets, +double-counted pacing handoffs, physical endpoint overlap, and unsafe +synthetic timestamp boundaries. Final review reported no findings. + +Live validation was intentionally deferred because the user shut down the +trading workspace. A fresh datad/chart startup is required to regenerate +SHM and verify that QQQ no longer contains the 653-row null run. diff --git a/piker/brokers/ib/feed.py b/piker/brokers/ib/feed.py index adeb6361..f5ec0bf3 100644 --- a/piker/brokers/ib/feed.py +++ b/piker/brokers/ib/feed.py @@ -32,7 +32,6 @@ import time from typing import ( Any, Callable, - TYPE_CHECKING, ) from async_generator import aclosing @@ -75,9 +74,6 @@ from ._util import ( ) from .symbols import get_mkt_info -if TYPE_CHECKING: - from trio._core._run import Task - log = get_logger( name=__name__, ) @@ -353,6 +349,7 @@ async def wait_on_data_reset( tuple[ trio.CancelScope, trio.Event, + list[bool], ] ] = trio.TASK_STATUS_IGNORED, ) -> bool: @@ -396,8 +393,9 @@ async def wait_on_data_reset( client: Client = proxy._aio_ns done = trio.Event() + result: list[bool] = [] with trio.move_on_after(timeout) as cs: - task_status.started((cs, done)) + task_status.started((cs, done, result)) log.warning( 'Sending DATA RESET request:\n' @@ -413,6 +411,7 @@ async def wait_on_data_reset( 'NO VNC DETECTED!\n' 'Manually press ctrl-alt-f on your IB java app' ) + result.append(False) done.set() return False @@ -427,6 +426,7 @@ async def wait_on_data_reset( ]: await ev.wait() log.info(f"{name} DATA RESET") + result.append(True) done.set() return True @@ -435,12 +435,12 @@ async def wait_on_data_reset( 'Data reset task canceled?' ) + result.append(False) done.set() return False -_data_resetter_task: Task | None = None -_failed_resets: int = 0 +_data_reset_lock = trio.Lock() async def get_bars( @@ -479,8 +479,11 @@ async def get_bars( `.ib.api.Client` methods. ''' - global _data_resetter_task, _failed_resets nodatas_count: int = 0 + cancelled_count: int = 0 + failed_resets: int = 0 + pacing_reset_pending: bool = False + pacing_reset_completed: bool = False data_cs: trio.CancelScope | None = None result: tuple[ @@ -493,10 +496,16 @@ async def get_bars( async def query(): - global _failed_resets - nonlocal result, data_cs, end_dt, nodatas_count + nonlocal result, data_cs, end_dt + nonlocal nodatas_count, cancelled_count, failed_resets + nonlocal pacing_reset_pending, pacing_reset_completed - while _failed_resets < max_failed_resets: + while True: + if failed_resets >= max_failed_resets: + raise DataUnavailable( + f'IB history reset failed {failed_resets} ' + f'times for {fqme}' + ) try: ( bars, @@ -528,6 +537,10 @@ async def get_bars( ) # NOTE: REQUIRED to pass back value.. result = None + failed_resets = 0 + result_ready.set() + if data_cs: + data_cs.cancel() return None # not enough bars signal, likely due to venue @@ -587,6 +600,7 @@ async def get_bars( first_dt, last_dt, ) + failed_resets = 0 # signal data reset loop parent task result_ready.set() @@ -642,6 +656,12 @@ async def get_bars( 'Query cancelled by IB (:eyeroll:):\n' f'{err.message}' ) + cancelled_count += 1 + if cancelled_count >= max_failed_resets: + raise DataUnavailable( + f'IB cancelled {cancelled_count} history ' + f'queries for {fqme}' + ) continue elif ( @@ -662,6 +682,7 @@ async def get_bars( ) client = proxy._aio_ns.ib.client + pacing_reset_pending = True # cancel any existing reset task if data_cs: @@ -669,69 +690,81 @@ async def get_bars( data_cs.cancel() # spawn new data reset task - data_cs, reset_done = await tn.start( - partial( - wait_on_data_reset, - proxy, - reset_type='connection' + async with _data_reset_lock: + ( + data_cs, + reset_done, + reset_result, + ) = await tn.start( + partial( + wait_on_data_reset, + proxy, + reset_type='connection' + ) ) - ) - if reset_done: - _failed_resets = 0 + await reset_done.wait() + if reset_result[0]: + failed_resets = 0 else: - _failed_resets += 1 + failed_resets += 1 + pacing_reset_pending = False + pacing_reset_completed = True continue else: raise - # TODO: make this global across all history task/requests - # such that simultaneous symbol queries don't try data resettingn - # too fast.. - unset_resetter: bool = False async with ( tractor.trionics.collapse_eg(), trio.open_nursery() as tn ): - # start history request that we allow - # to run indefinitely until a result is acquired + # Run history until a result or bounded reset failure. tn.start_soon(query) - # start history reset loop which waits up to the timeout - # for a result before triggering a data feed reset. + # Serialize farm resets across every history requester. while not result_ready.is_set(): - with trio.move_on_after(feed_reset_timeout): await result_ready.wait() break - if _data_resetter_task: - # don't double invoke the reset hack if another - # requester task already has it covered. - continue + async with _data_reset_lock: + if result_ready.is_set(): + break + if pacing_reset_completed: + pacing_reset_completed = False + continue - else: - _data_resetter_task = trio.lowlevel.current_task() - unset_resetter: bool = True - - # spawn new data reset task - data_cs, reset_done = await tn.start( - partial( - wait_on_data_reset, - proxy, - reset_type='data', + ( + data_cs, + reset_done, + reset_result, + ) = await tn.start( + partial( + wait_on_data_reset, + proxy, + reset_type='data', + ) ) - ) - # sync wait on reset to complete - await reset_done.wait() + await reset_done.wait() - _data_resetter_task = ( - None - if unset_resetter - else _data_resetter_task - ) - assert result + if ( + result_ready.is_set() + or + reset_result[0] + ): + failed_resets = 0 + elif pacing_reset_pending: + # Pacing recovery deliberately cancelled this reset + # and will consume the retry itself. + pass + else: + failed_resets += 1 + if failed_resets >= max_failed_resets: + raise DataUnavailable( + f'IB history reset failed ' + f'{failed_resets} times for {fqme}' + ) return ( result, data_cs is not None, diff --git a/piker/tsp/_history.py b/piker/tsp/_history.py index 4ace4e06..df31b849 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -206,6 +206,34 @@ async def notify_backfill( ) +def _synthetic_gap_times( + start_t: float, + right_t: float|None, + count: int, + timeframe: float, + +) -> np.ndarray|None: + ''' + Build bounded synthetic timestamps without crossing valid data. + + ''' + times: np.ndarray = ( + start_t + + + np.arange(1, count + 1)*timeframe + ) + if right_t is None: + return None + + if not np.isclose( + right_t, + start_t + (count + 1)*timeframe, + ): + return None + + return times + + async def shm_push_in_between( shm: ShmArray, to_push: np.ndarray, @@ -262,11 +290,11 @@ async def maybe_fill_null_segments( task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, -) -> list[Frame]: +) -> None: task_status.started() - frame: Frame = shm.array + frame: Frame = shm.array.copy() # TODO, put in parent task/daemon root! import greenback @@ -297,15 +325,26 @@ async def maybe_fill_null_segments( await tractor.pause() break - ( - array, - next_start_dt, - next_end_dt, - ) = await get_hist( - timeframe, - start_dt=start_dt, - end_dt=end_dt, - ) + try: + ( + array, + next_start_dt, + next_end_dt, + ) = await get_hist( + timeframe, + start_dt=start_dt, + end_dt=end_dt, + ) + except ( + NoData, + DataUnavailable, + ) as exc: + log.warning( + f'Provider could not repair SHM gap:\n' + f'{start_dt} -> {end_dt}\n' + f'{exc}\n' + ) + break if array.size == 0: log.warning( @@ -314,7 +353,6 @@ async def maybe_fill_null_segments( ) # ?TODO? do we want to remove the nulls and push # the close price here for the gap duration? - await tractor.pause() break if ( @@ -327,7 +365,7 @@ async def maybe_fill_null_segments( f'frame_start_dt: {frame_start_dt!r}\n' f'backfill_until_dt: {backfill_until_dt!r}\n' ) - await tractor.pause() + break # XXX TODO: pretty sure if i plot tsla, btcusdt.binance # and mnq.cme.ib this causes a Qt crash XXDDD @@ -349,21 +387,11 @@ async def maybe_fill_null_segments( # - remember that in the display side, only refersh this # if the respective history is actually "in view". # loop - try: - await sampler_stream.send({ - 'broadcast_all': { - - # XXX NOTE XXX: see the - # `.ui._display.increment_history_view()` if block - # that looks for this info to FORCE a hard viz - # redraw! - 'backfilling': (mkt.fqme, timeframe), - }, - }) - except tractor.ContextCancelled: - # log.exception - await tractor.pause() - raise + await notify_backfill( + sampler_stream, + mkt, + timeframe, + ) # RECHECK for more null-gaps frame: Frame = shm.array @@ -379,7 +407,7 @@ async def maybe_fill_null_segments( ( iabs_slices, iabs_zero_rows, - zero_t, + _zero_t, ) = null_segs log.warning( f'{len(iabs_slices)} NULL TIME SEGMENTS DETECTED!\n' @@ -398,13 +426,61 @@ async def maybe_fill_null_segments( 'close', ] - for istart, istop in iabs_slices: + zero_indexes: np.ndarray = np.asarray(iabs_zero_rows) + split_at: np.ndarray = ( + np.flatnonzero(np.diff(zero_indexes) != 1) + + + 1 + ) + zero_groups: list[np.ndarray] = np.split( + zero_indexes, + split_at, + ) + repaired: int = 0 + frame_start: int = int(frame['index'][0]) + for zero_group in zero_groups: + istart: int = int(zero_group[0]) + istop: int = int(zero_group[-1]) + 1 + seed_index: int = istart - 1 + if seed_index < frame_start: + log.warning( + f'Can not forward-fill leading SHM nulls:\n' + f'{istart}:{istop}\n' + ) + continue - # get view into buffer for null-segment + seed: np.void = shm._array[seed_index] + start_t: float = seed['time'] + if start_t <= 0: + log.warning( + f'Can not forward-fill SHM nulls from ' + f'zero-time seed at {seed_index}\n' + ) + continue + + # Fill only the null rows, preserving both valid margins. gap: np.ndarray = shm._array[istart:istop] + cls: float = seed['close'] - # copy the oldest OHLC samples forward - cls: float = shm._array[istart]['close'] + frame_end: int = int(frame['index'][-1]) + right_t: float|None = ( + shm._array[istop]['time'] + if istop <= frame_end + else None + ) + gap_times: np.ndarray|None = _synthetic_gap_times( + start_t, + right_t, + len(gap), + timeframe, + ) + if gap_times is None: + log.warning( + f'Can not safely forward-fill SHM nulls:\n' + f'left={start_t} right={right_t} ' + f'rows={len(gap)} period={timeframe}\n' + ) + continue # TODO: how can we mark this range as being a gap tho? # -[ ] maybe pg finally supports nulls in ndarray to @@ -413,30 +489,13 @@ async def maybe_fill_null_segments( # another col/field to denote? gap[ohlc_fields] = cls - start_t: float = shm._array[istart]['time'] - t_diff: float = (istop - istart)*timeframe - - gap['time'] = np.arange( - start=start_t, - stop=start_t + t_diff, - step=timeframe, - ) + gap['time'] = gap_times + repaired += len(gap) # TODO: reimpl using the new `.ui._remote_ctl` ctx # ideally using some kinda decent # tractory-reverse-lookup-connnection from some other # `Context` type thingy? - await sampler_stream.send({ - 'broadcast_all': { - - # XXX NOTE XXX: see the - # `.ui._display.increment_history_view()` if block - # that looks for this info to FORCE a hard viz - # redraw! - 'backfilling': (mkt.fqme, timeframe), - }, - }) - # TODO: interatively step through any remaining # time-gaps/null-segments and spawn piecewise backfiller # tasks in a nursery? @@ -447,6 +506,13 @@ async def maybe_fill_null_segments( # -[ ] fill algo: do queries in alternating "latest, then # earliest, then latest.. etc?" + if repaired: + await notify_backfill( + sampler_stream, + mkt, + timeframe, + ) + async def start_backfill( get_hist, @@ -680,7 +746,12 @@ async def start_backfill( array, prepend_until_dt=backfill_until_dt, ) - to_push: np.ndarray = to_store + # The requested end bar is already the oldest published + # sample. Keep it for idempotent storage merge, but do not + # consume another physical SHM row for the overlap. + to_push: np.ndarray = to_store[ + to_store['time'] < last_start_dt.timestamp() + ] ln: int = len(to_push) if ln: log.info( @@ -1314,28 +1385,26 @@ async def tsdb_backfill( log.info(f'Loaded {to_push.shape} datums from storage') - # NOTE: ASYNC-conduct tsdb timestamp gap detection and backfill any - # seemingly missing (null-time) segments.. - # TODO: ideally these can never exist! - # -[ ] somehow it seems sometimes we're writing zero-ed - # segments to tsdbs during teardown? - # -[ ] can we ensure that the backfiller tasks do this - # work PREVENTAVELY instead? - # -[ ] fill in non-zero epoch time values ALWAYS! - # await maybe_fill_null_segments( - await tn.start(partial( - maybe_fill_null_segments, - - shm=shm, - timeframe=timeframe, - get_hist=get_hist, - sampler_stream=sampler_stream, - mkt=mkt, - backfill_until_dt=last_tsdb_dt, - )) - # 2nd nursery END + if last_tsdb_dt is not None: + # Repair only after reverse backfill releases providers + # such as IB which permit one history request at a time. + # TODO: ideally these null segments can never exist! + # -[ ] somehow it seems sometimes we're writing zero-ed + # segments to tsdbs during teardown? + # -[ ] can we ensure that the backfiller tasks do this + # work PREVENTAVELY instead? + # -[ ] fill in non-zero epoch time values ALWAYS! + await maybe_fill_null_segments( + shm=shm, + timeframe=timeframe, + get_hist=get_hist, + sampler_stream=sampler_stream, + mkt=mkt, + backfill_until_dt=last_tsdb_dt, + ) + # 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 index 56ba7674..099fdc6b 100644 --- a/tests/test_history_backfill.py +++ b/tests/test_history_backfill.py @@ -29,6 +29,8 @@ from piker.storage.nativedb import ( ohlc_key_map, ) from piker.tsp._history import ( + _synthetic_gap_times, + maybe_fill_null_segments, notify_backfill, publish_latest_frame, start_backfill, @@ -92,15 +94,22 @@ def test_empty_frame_completes() -> None: def test_storage_receives_full_provider_delta() -> None: ''' - SHM capacity truncation does not truncate durable history. + SHM drops the published endpoint without truncating storage. + + Reverse history queries include their requested end bar, which is + already the oldest sample in SHM. Publishing the overlap consumed + an extra physical row and made reserved null counts disagree with + elapsed timestamps. Return a frame with the overlap and prove + NativeDB receives the complete provider delta while SHM receives + only bars older than its published boundary. ''' frame = np.zeros( - 2, + 3, dtype=np.dtype(def_iohlcv_fields), ) - frame['index'] = [0, 1] - frame['time'] = [60, 120] + frame['index'] = [0, 1, 2] + frame['time'] = [60, 120, 180] events: list[str] = [] @@ -142,7 +151,7 @@ def test_storage_receives_full_provider_delta() -> None: return ( frame, from_timestamp(60), - from_timestamp(120), + from_timestamp(180), ) async def main() -> None: @@ -154,7 +163,7 @@ def test_storage_receives_full_provider_delta() -> None: mkt=mkt, shm=shm, timeframe=60, - backfill_from_shm_index=1, + backfill_from_shm_index=2, backfill_from_dt=from_timestamp(180), sampler_stream=Sampler(), backfill_until_dt=from_timestamp(60), @@ -163,8 +172,8 @@ def test_storage_receives_full_provider_delta() -> None: ) trio.run(main) - assert shm.pushed[0]['time'].tolist() == [120] - assert storage.frames[0]['time'].tolist() == [60, 120] + assert shm.pushed[0]['time'].tolist() == [60, 120] + assert storage.frames[0]['time'].tolist() == [60, 120, 180] assert events == ['storage', 'shm', 'sampler'] @@ -344,3 +353,163 @@ def test_backfill_notification_timeout_is_bounded( ) trio.run(main) + + +def test_synthetic_gap_times_require_valid_right_boundary() -> None: + ''' + Synthetic rows must remain strictly between valid boundary bars. + + The live QQQ reservation held one more physical zero row than its + timestamp interval could represent. Blind fill would assign the + null the right boundary's timestamp, creating a duplicate. Prove + aligned bounds produce cadence while an overfull segment + is rejected instead of inventing non-monotonic timestamps. + + ''' + aligned = _synthetic_gap_times( + start_t=60, + right_t=300, + count=3, + timeframe=60, + ) + overfull = _synthetic_gap_times( + start_t=60, + right_t=240, + count=3, + timeframe=60, + ) + trailing = _synthetic_gap_times( + start_t=60, + right_t=None, + count=3, + timeframe=60, + ) + + assert aligned is not None + assert aligned.tolist() == [120, 180, 240] + assert overfull is None + assert trailing is None + + +def test_null_repair_falls_back_without_debug_pause( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Empty provider repair must resolve interior zero rows locally. + + QQQ startup exposed 653 zero-time rows between stored history and + an IB frame. The null repair queried IB while reverse backfill + occupied its sole history request slot, then could + enter an interactive debug pause before reaching forward-fill. + Arrange an interior null run, return an empty provider frame, and + wedge sampler notification. A short timeout and fail-fast pause + prove local fallback needs no interactive or IPC progress. + The assertions verify only the + zero rows inherit the preceding close and synthesized timestamps; + both valid boundary rows remain unchanged. + + ''' + monkeypatch.setattr( + 'piker.tsp._history._notify_timeout_s', + 0.01, + ) + + backing = np.zeros( + 105, + dtype=np.dtype(def_iohlcv_fields), + ) + backing['index'] = np.arange(105) + frame = backing[100:105] + frame['index'] = np.arange(100, 105) + frame['time'] = [60, 0, 0, 0, 300] + frame['open'] = [9, 0, 0, 0, 19] + frame['high'] = [11, 0, 0, 0, 21] + frame['low'] = [8, 0, 0, 0, 18] + frame['close'] = [10, 0, 0, 0, 20] + frame['volume'] = [5, 0, 0, 0, 6] + + class Shm: + ''' + Expose the test frame through the SHM repair interface. + + ''' + def __init__(self) -> None: + self._array = backing + + @property + def array(self) -> np.ndarray: + ''' + Return the readable SHM view. + + ''' + return self._array[100:105] + + class Sampler: + ''' + Model a backpressured UI notification stream. + + ''' + async def send(self, msg) -> None: + ''' + Block until the bounded notifier cancels this send. + + ''' + await trio.sleep_forever() + + async def get_hist(*args, **kwargs): + ''' + Return no provider bars for the requested null segment. + + ''' + end_dt = kwargs['end_dt'] + empty = np.empty( + 0, + dtype=np.dtype(def_iohlcv_fields), + ) + return empty, end_dt, end_dt + + async def fail_pause() -> None: + ''' + Reject interactive debugging in normal repair fallbacks. + + ''' + raise AssertionError('null repair entered tractor.pause()') + + monkeypatch.setattr(tractor, 'pause', fail_pause) + + async def main() -> None: + import greenback + + async def ensure_portal() -> None: + ''' + Avoid installing a greenback portal in this unit test. + + ''' + + monkeypatch.setattr( + greenback, + 'ensure_portal', + ensure_portal, + ) + with trio.fail_after(0.1): + await maybe_fill_null_segments( + shm=Shm(), + timeframe=60, + get_hist=get_hist, + sampler_stream=Sampler(), + mkt=SimpleNamespace(fqme='qqq.nasdaq.ib'), + backfill_until_dt=from_timestamp(60), + ) + + trio.run(main) + + assert frame['time'].tolist() == [60, 120, 180, 240, 300] + for field in ('open', 'high', 'low', 'close'): + assert frame[field].tolist() == [ + frame[field][0], + 10, + 10, + 10, + frame[field][-1], + ] + assert frame['volume'].tolist() == [5, 0, 0, 0, 6] diff --git a/tests/test_ib_history.py b/tests/test_ib_history.py index ec3ac694..33acfa86 100644 --- a/tests/test_ib_history.py +++ b/tests/test_ib_history.py @@ -8,11 +8,19 @@ from datetime import ( ) from types import SimpleNamespace -from pendulum import datetime as pdatetime +from ib_async import RequestError +import numpy as np +from pendulum import ( + datetime as pdatetime, + duration, +) import pytest import trio +from piker.brokers import DataUnavailable +from piker.brokers.ib import feed as ib_feed from piker.brokers.ib.api import Client +from piker.brokers.ib.feed import get_bars @pytest.mark.parametrize( @@ -79,3 +87,502 @@ def test_history_end_datetime_uses_ib_utc_format( trio.run(main) assert request['endDateTime'] == expected + + +def test_empty_history_frame_completes() -> None: + ''' + A blank IB response must release the history request slot. + + ``get_bars()`` previously returned from its query task without + signalling the parent, which then reset the feed forever. That + kept serialized SHM repair from running. Return IB's empty + response and prove the request completes explicitly + with no result instead of relying on a reset timeout. + + ''' + class Proxy: + ''' + Return one empty IB history response. + + ''' + async def bars(self, **kwargs): + ''' + Model ``Client.bars()`` during a venue data gap. + + ''' + return ( + [], + np.empty(0), + duration(seconds=2000), + ) + + async def main() -> None: + with trio.fail_after(0.1): + result, timedout = await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + ) + assert result is None + assert not timedout + + trio.run(main) + + +def test_repeated_ib_cancellation_is_bounded() -> None: + ''' + Repeated IB cancellation must not monopolize history forever. + + IB can answer a historical request with error 162 cancellation. + The old retry loop had no limit, so reverse backfill could retain + the backend's sole request slot and prevent final null repair. + Always raise that response and prove the configured retry bound + terminates with ``DataUnavailable`` deterministically. + + ''' + class Proxy: + ''' + Cancel every IB history request. + + ''' + async def bars(self, **kwargs): + ''' + Raise IB's historical-query cancellation response. + + ''' + raise RequestError( + 1, + 162, + 'API historical data query cancelled', + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=2, + ) + + trio.run(main) + + +def test_query_error_cancels_stalled_reset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Exceptional history exit must cancel its in-flight reset task. + + A timed-out request acquires the reset lock before launching + IB's VNC reset task. If the query then raises while that task is + blocked, stale work must not retain the reset lock. Force that + ordering with explicit Trio checkpoints, then prove bounded query + failure cancels the reset task before propagating. + + ''' + reset_cancelled: list[bool] = [] + + class Proxy: + ''' + Delay once, then return IB cancellation. + + ''' + async def bars(self, **kwargs): + ''' + Let the parent acquire the reset lock before failing. + + ''' + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + 'API historical data query cancelled', + ) + + async def stalled_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Publish reset state, then block until sibling failure. + + ''' + done = trio.Event() + result: list[bool] = [] + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + try: + await trio.sleep_forever() + finally: + reset_cancelled.append(True) + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + stalled_reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=1, + ) + + trio.run(main) + assert reset_cancelled == [True] + + +def test_failed_pacing_resets_reach_retry_limit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Failed pacing resets must increment the request's retry count. + + The reset API returns an ``Event`` and result box. The old + code tested the always-truthy event and reset the failure counter + after each failure. Return immediate failures and prove two + pacing responses terminate instead of looping forever. + + ''' + reset_calls: list[bool] = [] + + class Proxy: + ''' + Return a pacing violation for every request. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Raise IB's pacing response. + + ''' + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def failed_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Publish one completed, unsuccessful reset. + + ''' + done = trio.Event() + result = [False] + reset_calls.append(True) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + failed_reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=2, + ) + + trio.run(main) + assert reset_calls == [True, True] + + +def test_pacing_resets_are_serialized_across_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Concurrent history requests must never reset IB farms together. + + Pacing recovery bypassed reset ownership, allowing 1s and 60s + requests to launch the VNC hack concurrently. Start two pacing + failures in one nursery and hold each fake reset across a Trio + checkpoint. Active-count assertions prove the actor-global lock + serializes reset lifetimes without depending on scheduler timing. + + ''' + active: int = 0 + max_active: int = 0 + reset_calls: int = 0 + + class Proxy: + ''' + Return a pacing violation for every request. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Raise IB's pacing response. + + ''' + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def failed_reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Hold one failed reset long enough to expose overlap. + + ''' + nonlocal active, max_active, reset_calls + done = trio.Event() + result: list[bool] = [] + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + active += 1 + reset_calls += 1 + max_active = max(max_active, active) + await trio.sleep(0.01) + active -= 1 + result.append(False) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + failed_reset, + ) + + async def request() -> None: + ''' + Run one independently bounded history request. + + ''' + with pytest.raises(DataUnavailable): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + max_failed_resets=1, + ) + + async def main() -> None: + with trio.fail_after(0.1): + async with trio.open_nursery() as nursery: + nursery.start_soon(request) + nursery.start_soon(request) + + trio.run(main) + assert reset_calls == 2 + assert max_active == 1 + + +def test_pacing_supersedes_timeout_reset_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Pacing handoff must not count cancellation as another failure. + + A slow request can trigger timeout recovery immediately before IB + returns a pacing violation. Query-side recovery then deliberately + cancels that data reset and waits for the same actor-global lock. + Reproduce this with blocked first and failed second resets. A + one-attempt limit proves cancellation releases the lock + without aborting before the intended connection reset runs. + + ''' + reset_types: list[str] = [] + + class Proxy: + ''' + Return pacing only after timeout recovery starts. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + async def bars(self, **kwargs): + ''' + Delay past the reset timeout, then report pacing. + + ''' + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + async def reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Block data reset and fail connection reset immediately. + + ''' + done = trio.Event() + result: list[bool] = [] + reset_types.append(reset_type) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + if reset_type == 'data': + await trio.sleep_forever() + + result.append(False) + done.set() + return False + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + reset, + ) + + async def main() -> None: + with ( + trio.fail_after(0.1), + pytest.raises(DataUnavailable), + ): + await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=1, + ) + + trio.run(main) + assert reset_types == ['data', 'connection'] + + +def test_completed_pacing_reset_restarts_timeout_wait( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A queued timeout waiter must consume completed pacing recovery. + + While query-side pacing recovery holds the reset lock, the outer + timeout task can already queue behind it. Without a completion + handoff, that waiter launches a redundant data reset immediately. + Return a blank frame just after a successful connection reset and + prove no third reset runs before the restarted timeout observes + completion. + + ''' + reset_types: list[str] = [] + + class Proxy: + ''' + Pace the first query and return blank data on retry. + + ''' + _aio_ns = SimpleNamespace( + ib=SimpleNamespace( + client=SimpleNamespace(), + ), + ) + + def __init__(self) -> None: + self.calls: int = 0 + + async def bars(self, **kwargs): + ''' + Control the pacing-to-success handoff ordering. + + ''' + self.calls += 1 + if self.calls == 1: + await trio.sleep(0.02) + raise RequestError( + 1, + 162, + ib_feed._pacing, + ) + + await trio.sleep(0.0005) + return ( + [], + np.empty(0), + duration(seconds=2000), + ) + + async def reset( + proxy, + reset_type='data', + task_status=trio.TASK_STATUS_IGNORED, + ) -> bool: + ''' + Cancel timeout recovery and complete pacing recovery. + + ''' + done = trio.Event() + result: list[bool] = [] + reset_types.append(reset_type) + with trio.CancelScope() as cs: + task_status.started((cs, done, result)) + if reset_type == 'data': + await trio.sleep_forever() + + result.append(reset_type == 'connection') + done.set() + return result[0] + + monkeypatch.setattr( + ib_feed, + 'wait_on_data_reset', + reset, + ) + + async def main() -> None: + with trio.fail_after(0.1): + result, _ = await get_bars( + Proxy(), + 'qqq.nasdaq.ib', + 1, + feed_reset_timeout=0.001, + max_failed_resets=2, + ) + assert result is None + + trio.run(main) + assert reset_types == ['data', 'connection']