From ad6c560f6bed6dec017f18463172cfd050c087c6 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 29 Jul 2026 15:07:50 -0400 Subject: [PATCH] Guard `ldshm` from invalid SHM snapshots Do not pass unpublished or corrupt timestamp rows into dedupe, `NativeDB` replacement writes, or live SHM reload. Deats, - infer cadence from finite positive observed steps - choose the smaller observed step when frequency ties - skip invalid or degenerate snapshots as a whole - cover the QQQ zero-slot failure with mutation-failing spies Prompt-IO: ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../run-tests/test-harness-reference.md | 3 + .../20260729T185406Z_d359b1cb_prompt_io.md | 34 +++ ...20260729T185406Z_d359b1cb_prompt_io.raw.md | 58 +++++ piker/storage/cli.py | 76 ++++-- tests/test_ldshm.py | 235 ++++++++++++++++++ 5 files changed, 381 insertions(+), 25 deletions(-) create mode 100644 ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.raw.md create mode 100644 tests/test_ldshm.py diff --git a/.claude/skills/run-tests/test-harness-reference.md b/.claude/skills/run-tests/test-harness-reference.md index 1c951ec0..12ccea7c 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_ldshm.py` - `tests/test_accounting.py::test_account_file_default_empty` - `tests/test_services.py::test_runtime_boot` - `tests/test_services.py::test_datad_spawn` @@ -190,6 +191,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_ldshm.py ldshm unpublished-slot guard test_questrade.py obsolete credentialed tests; skipped test_services.py pikerd/datad/feed/EMS actor lifecycle test_storage_audit.py read-only NativeDB audit and JSON CLI @@ -206,6 +208,7 @@ tests/ | `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/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 | | `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | GUI/config-isolated opt-in | diff --git a/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.md b/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.md new file mode 100644 index 00000000..0a7939da --- /dev/null +++ b/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.md @@ -0,0 +1,34 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: 6409e3bb-1a9a-4ada-b6c4-c554115143e9 +timestamp: 2026-07-29T18:54:06Z +git_ref: d359b1cb +scope: code +substantive: true +raw_file: 20260729T185406Z_d359b1cb_prompt_io.raw.md +--- + +## Prompt + +The user reported that QQQ 60-second zeros appeared as a flat line after +reload, clarified that the outstanding concern was the prior `ldshm` +failure, and instructed the agent to implement the more focused fix. + +## Response summary + +Changed `ldshm` to classify invalid SHM snapshots before downstream +analysis. Cadence is inferred only from observed finite positive steps; +buffers containing non-finite or non-positive timestamps, or lacking a +positive cadence, are skipped rather than partially replacement-written +or reloaded. Added deterministic helper and command-level regressions. + +## Files changed + +- `piker/storage/cli.py` - inspect cadence and reject invalid SHM snapshots before mutation. +- `tests/test_ldshm.py` - cover QQQ-like zeros, corrupt timestamps, cadence selection, degenerate buffers, and mutation prevention. +- `.claude/skills/run-tests/test-harness-reference.md` - register the deterministic regression target. + +## Human edits + +None - generated output remains uncommitted. diff --git a/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.raw.md b/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.raw.md new file mode 100644 index 00000000..cf1b7646 --- /dev/null +++ b/ai/prompt-io/opencode/20260729T185406Z_d359b1cb_prompt_io.raw.md @@ -0,0 +1,58 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-07-29T18:54:06Z +git_ref: d359b1cb +diff_cmd: git diff HEAD~1..HEAD +--- + +The user clarified that the QQQ 60-second chart now displayed a flat +forward-filled segment, then asked whether the previously reported +`piker store ldshm` failure had been fixed. After being told it had not, +the user instructed: "ok but i read above you have a more focussed fix +todo now? so have at it!" + +The implementation focused on the exact failure boundary. It does not +filter invalid rows into a partial replacement and does not mutate live +SHM. Instead, `ldshm` inspects one captured Polars snapshot, infers its +cadence from observed finite positive timestamp steps, and skips the +entire buffer before analysis, persistence, or reload when any timestamp +is non-finite or non-positive. + +> `git diff HEAD~1..HEAD -- piker/storage/cli.py` + +Generated `_shm_period_and_invalid_count()` and integrated its result in +`ldshm`. The helper selects the most frequent observed positive step, +using the smallest observed step as the deterministic tie-break. Short +or cadence-degenerate buffers are skipped. Invalid snapshots are logged +and cannot reach deduplication, NativeDB replacement, or SHM reload. + +> `git diff HEAD~1..HEAD -- tests/test_ldshm.py` + +Generated deterministic regressions for the QQQ-shaped interior zero +run, observed-step cadence selection, all invalid timestamp classes, +short and degenerate buffers, and command-level prevention of dedupe, +storage writes, and reload. The command-level test uses an immutable +Polars snapshot and failing spies, without live SHM mutation or timing. + +> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md` + +Added the deterministic `ldshm` regression target and its change-to-test +mapping to the repository test harness reference. + +Verification output: + +```text +.......................................................... [100%] +58 passed in 0.70s +``` + +`git diff --check` passed. Ruff was unavailable in the worktree's +existing environment. Three adversarial review passes found and drove +fixes for partial-replacement data loss, invented median timeframes, +mixed SHM observations, and pre-guard null-analysis failures. The final +review reported no remaining findings. + +The patch intentionally does not claim that `ldshm` is safe to run as a +general concurrent repair writer. It handles invalid live snapshots by +refusing mutation rather than weakening NativeDB validation. diff --git a/piker/storage/cli.py b/piker/storage/cli.py index f6db254d..8129a4d9 100644 --- a/piker/storage/cli.py +++ b/piker/storage/cli.py @@ -60,6 +60,37 @@ if TYPE_CHECKING: store = typer.Typer() +def _shm_period_and_invalid_count( + times: np.ndarray, + +) -> tuple[float|None, int]: + ''' + Infer cadence without treating unpublished slots as samples. + + ''' + valid: np.ndarray = ( + np.isfinite(times) + & + (times > 0) + ) + invalid_count: int = int(np.count_nonzero(~valid)) + published: np.ndarray = times[valid] + if published.size < 2: + return None, invalid_count + + steps: np.ndarray = np.diff(published) + positive_steps: np.ndarray = steps[steps > 0] + if not positive_steps.size: + return None, invalid_count + + periods, counts = np.unique( + positive_steps, + return_counts=True, + ) + period: float = float(periods[np.argmax(counts)]) + return period, invalid_count + + def _render_audit_report(report: dict) -> None: ''' Render a compact human summary and explicit gap endpoints. @@ -506,38 +537,25 @@ def ldshm( shm_df, ) in tsp.iter_dfs_from_shms(fqme): - times: np.ndarray = shm.array['time'] - d1: float = float(times[-1] - times[-2]) - d2: float = 0 - # XXX, take a median sample rate if sufficient data - if times.size > 2: - d2: float = float(times[-2] - times[-3]) - med: float = np.median(np.diff(times)) - if ( - d1 < 1. - and d2 < 1. - and med < 1. - ): - raise ValueError( - f'Something is wrong with time period for {shm}:\n{times}' - ) - period_s: float = float(max(d1, d2, med)) + times: np.ndarray = shm_df['time'].to_numpy() + ( + period_s, + invalid_count, + ) = _shm_period_and_invalid_count(times) + if period_s is None: + log.warning( + f'Could not infer a positive sample period ' + f'for {shmfile.name}; skipping buffer\n' + ) + continue log.info( f'Processing shm buffer:\n' f' file: {shmfile.name}\n' f' period: {period_s}s\n' ) - null_segs: tuple = tsp.get_null_segs( - frame=shm.array, - period=period_s, - ) - # TODO: call null-seg fixer somehow? - if null_segs: - - if tractor.runtime._state.is_debug_mode(): - await tractor.pause() + if invalid_count: # async with ( # trio.open_nursery() as tn, # mod.open_history_client( @@ -553,6 +571,14 @@ def ldshm( # sampler_stream=sampler_stream, # mkt=mkt, # )) + if tractor.runtime._state.is_debug_mode(): + await tractor.pause() + log.warning( + f'Found {invalid_count} invalid SHM row(s) ' + f'in {shmfile.name}; skipping persistence ' + f'and reload\n' + ) + continue # over-write back to shm? wdts: pl.DataFrame # with dts diff --git a/tests/test_ldshm.py b/tests/test_ldshm.py new file mode 100644 index 00000000..2548f578 --- /dev/null +++ b/tests/test_ldshm.py @@ -0,0 +1,235 @@ +''' +Shared-memory persistence regressions. + +''' +from collections.abc import ( + AsyncIterator, + Iterator, +) +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import numpy as np +import polars as pl +import pytest + +from piker import tsp +from piker.storage import cli as storage_cli +from piker.storage.cli import ( + _shm_period_and_invalid_count, + ldshm, +) + + +def test_ldshm_detects_nulls_without_skewing_period() -> None: + ''' + Null SHM slots must stop persistence without skewing cadence. + + A live QQQ buffer contained an interior run of zero-time rows. + ``ldshm`` included them in period inference and then sent them to + NativeDB, which correctly rejected the replacement. Build a + 60-second series with the same interior hole and enough + bars around the gap. Prove inspection counts every null while + inferring the published series' 60-second period; the + command uses that count to skip both persistence and SHM reload. + + ''' + times = np.array([ + 60, + 120, + 0, + 0, + 300, + 360, + 420, + ]) + + period_s, invalid_count = _shm_period_and_invalid_count(times) + + assert period_s == 60 + assert invalid_count == 2 + + +def test_ldshm_period_is_an_observed_step() -> None: + ''' + Sparse timestamps must not invent a storage timeframe. + + The arithmetic median of two steps can produce a cadence that + occurs nowhere in the source series. Such a value can select the + wrong NativeDB timeframe and hide the gap. Arrange tied 60 and + 120-second steps and prove the deterministic tie-break chooses + the smaller observed period. + + ''' + times = np.array([60, 120, 240]) + + period_s, invalid_count = _shm_period_and_invalid_count(times) + + assert period_s == 60 + assert invalid_count == 0 + + +def test_ldshm_counts_every_invalid_timestamp() -> None: + ''' + Corrupt timestamps stop persistence just like null SHM slots. + + Exact zero denotes an unpublished slot, while negative, NaN, and + infinite values indicate corruption. NativeDB rejects them all, + but allowing a replacement attempt would still abort ``ldshm``. + Mix each invalid class into a regular series and prove the same + snapshot inspection counts all four while preserving cadence. + + ''' + times = np.array([ + 60, + 120, + 0, + -1, + np.nan, + np.inf, + 180, + ]) + + period_s, invalid_count = _shm_period_and_invalid_count(times) + + assert period_s == 60 + assert invalid_count == 4 + + +def test_ldshm_invalid_snapshot_never_reaches_storage( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Invalid snapshots stop before analyzers and storage mutation. + + The original QQQ failure passed zero-time rows through dedupe and + into NativeDB. An intermediate fix called null-segment analysis + before its guard, allowing corrupt indexes to abort the command. + Supply one immutable Polars snapshot containing that combination, + then replace dedupe and writes with failing spies. Completing the + command proves its guard runs first, preventing persistence or + SHM reload without relying on task timing. + + ''' + shm_df = pl.DataFrame({ + 'index': [100, 500, 104], + 'time': [60, 0, 180], + }) + + @asynccontextmanager + async def open_runtime( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[None]: + ''' + Yield a runtime-free command context. + + ''' + yield + + async def fail_write( + *args: Any, + **kwargs: Any, + ) -> None: + ''' + Fail if invalid SHM reaches NativeDB. + + ''' + raise AssertionError('invalid SHM reached NativeDB') + + client = SimpleNamespace(write_ohlcv=fail_write) + + @asynccontextmanager + async def open_storage( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[tuple[SimpleNamespace, SimpleNamespace]]: + ''' + Yield storage objects with a failing writer. + + ''' + yield SimpleNamespace(), client + + @asynccontextmanager + async def open_annotations( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[SimpleNamespace]: + ''' + Yield an inert annotation controller. + + ''' + yield SimpleNamespace() + + def iter_shms( + fqme: str, + ) -> Iterator[tuple[Path, SimpleNamespace, pl.DataFrame]]: + ''' + Yield the single captured invalid snapshot. + + ''' + yield Path('qqq.rt'), SimpleNamespace(), shm_df + + def dedupe(*args: Any, **kwargs: Any) -> None: + ''' + Fail if invalid SHM reaches deduplication. + + ''' + raise AssertionError('invalid SHM reached dedupe') + + async def pause() -> None: + ''' + Replace the command's final interactive pause. + + ''' + + monkeypatch.setattr( + storage_cli, + 'open_piker_runtime', + open_runtime, + ) + monkeypatch.setattr( + storage_cli, + 'open_storage_client', + open_storage, + ) + monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_shms) + monkeypatch.setattr(tsp, 'dedupe_ohlcv_smart', dedupe) + monkeypatch.setattr(storage_cli.tractor, 'pause', pause) + + from piker.ui import _remote_ctl + monkeypatch.setattr( + _remote_ctl, + 'open_annot_ctl', + open_annotations, + ) + + ldshm('qqq.nasdaq.ib') + + +@pytest.mark.parametrize( + 'times', + [ + np.array([0, 0]), + np.array([60]), + np.array([60, 60]), + ], +) +def test_ldshm_skips_frame_without_positive_cadence( + times: np.ndarray, +) -> None: + ''' + Degenerate buffers are skipped before indexing or persistence. + + The old inference indexed final timestamps and left its median + undefined for short buffers. Fully unpublished, one-row, and + duplicate-only frames could therefore fail before the command's + guard. Exercise each shape and prove inspection + returns no cadence, which directs ``ldshm`` to skip the buffer. + + ''' + period_s, _ = _shm_period_and_invalid_count(times) + + assert period_s is None