diff --git a/ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md b/ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md new file mode 100644 index 00000000..3a4f1680 --- /dev/null +++ b/ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md @@ -0,0 +1,43 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-07-22T02:41:34Z +git_ref: 3a4f6737 +scope: code +substantive: true +raw_file: 20260722T024134Z_3a4f6737_prompt_io.raw.md +--- + +## Prompt + +The user asked why the in-progress `piker/storage` changes were omitted +from the earlier commit plan. After receiving an inventory of unfinished +Phase 3 requirements, the user instructed the agent to complete the work. + +## Response summary + +Finished non-destructive, atomic NativeDB updates and integrated them +with history startup and reverse backfill. Expanded tests through two +adversarial review rounds that caught and corrected cross-client lock +blocking, stale temp indexing, discontinuous persisted indexes, startup +snapshot races, missing latest-frame persistence, notification wedges, +and mutable/truncated SHM persistence. + +## Files changed + +- `piker/storage/__init__.py` - incremental storage contract +- `piker/storage/nativedb.py` - validated atomic merge persistence +- `piker/storage/marketstore/__init__.py` - non-duplicate update mode +- `piker/tsp/_history.py` - provider-delta and startup persistence +- `tests/test_storage_nativedb.py` - NativeDB durability regressions +- `tests/test_history_backfill.py` - persistence lifecycle regressions +- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_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/20260722T024134Z_3a4f6737_prompt_io.raw.md b/ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md new file mode 100644 index 00000000..16f28afd --- /dev/null +++ b/ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md @@ -0,0 +1,43 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-22T02:41:34Z +git_ref: 3a4f6737 +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/storage/__init__.py piker/storage/nativedb.py piker/storage/marketstore/__init__.py` + +Added an incremental storage update contract. NativeDB now validates +canonical numeric OHLCV data, serializes writers both actor-locally and +through cancellable filesystem locks, merges with incoming-wins seam +resolution, regenerates contiguous indexes, reopens and validates a +same-directory temporary parquet, fsyncs it, atomically replaces the +target, fsyncs the directory, and publishes cache only after replacement. + +NativeDB indexing ignores lock and crash-left temporary files. +MarketStore uses its non-duplicate write mode for incremental updates. + +> `git diff HEAD~1..HEAD -- piker/tsp/_history.py` + +Separated full provider deltas from capacity-truncated SHM pushes. +Backfill persists before SHM mutation, publishes bounded notifications, +and delays latest-frame publication until the concurrent pre-update +storage snapshot completes. + +> `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py tests/test_history_backfill.py` + +Added deterministic coverage for merge preservation, conflict policy, +atomic failures, reopened-candidate validation, cache consistency, +schema and timestamp invariants, contiguous index regeneration, +separate series, stale temp indexing, cancellable file-lock contention, +full provider-delta persistence, latest-frame ordering, and bounded +notification teardown. + +Verification generated with the patch: + +- deterministic analysis/backfill/storage set: 31 passed +- Ruff: passed +- `git diff --check`: passed +- final adversarial review: no findings diff --git a/piker/storage/__init__.py b/piker/storage/__init__.py index 3256dbf2..46e18626 100644 --- a/piker/storage/__init__.py +++ b/piker/storage/__init__.py @@ -139,6 +139,15 @@ class StorageClient( ) -> None: ... + async def update_ohlcv( + self, + fqme: str, + ohlcv: np.ndarray, + timeframe: int, + + ) -> None: + ... + class TimeseriesNotFound(Exception): ''' diff --git a/piker/storage/marketstore/__init__.py b/piker/storage/marketstore/__init__.py index de42de44..a41f315f 100644 --- a/piker/storage/marketstore/__init__.py +++ b/piker/storage/marketstore/__init__.py @@ -340,6 +340,24 @@ class MktsStorageClient: if err: raise MarketStoreError(err) + async def update_ohlcv( + self, + fqme: str, + ohlcv: np.ndarray, + timeframe: int, + + ) -> None: + ''' + Append an incremental frame using MarketStore semantics. + + ''' + await self.write_ohlcv( + fqme, + ohlcv, + timeframe, + append_and_duplicate=False, + ) + # XXX: currently the only way to do this is through the CLI: # sudo ./marketstore connect --dir ~/.config/piker/data diff --git a/piker/storage/nativedb.py b/piker/storage/nativedb.py index edb3a362..73fd5850 100644 --- a/piker/storage/nativedb.py +++ b/piker/storage/nativedb.py @@ -51,9 +51,18 @@ YET! # - https://github.com/spslater/borgapi # - https://nixos.wiki/wiki/ZFS from __future__ import annotations +from collections.abc import AsyncIterator from contextlib import asynccontextmanager as acm from datetime import datetime +from fcntl import ( + flock, + LOCK_EX, + LOCK_NB, + LOCK_UN, +) +import os from pathlib import Path +from tempfile import NamedTemporaryFile import time import numpy as np @@ -61,6 +70,7 @@ import polars as pl from pendulum import ( from_timestamp, ) +import trio from piker import config from piker import tsp @@ -141,6 +151,10 @@ class NativeStorageClient: # series' cache from tsdb reads self._dfs: dict[str, dict[str, pl.DataFrame]] = {} + self._write_locks: dict[ + tuple[str, int], + trio.Lock, + ] = {} @property def address(self) -> str: @@ -162,13 +176,13 @@ class NativeStorageClient: if ( path.is_dir() or - '.parquet' not in str(path) + path.suffix != '.parquet' # or # path.name in {'borked', 'expired',} ): continue - key: str = path.name.rstrip('.parquet') + key: str = path.name.removesuffix('.parquet') fqme, _, descr = key.rpartition('.') prefix, _, suffix = descr.partition('ohlcv') period: int = int(suffix.strip('s')) @@ -259,6 +273,50 @@ class NativeStorageClient: {}, )[fqme] = df + def _get_write_lock( + self, + fqme: str, + timeframe: int, + + ) -> trio.Lock: + ''' + Return the actor-local lock for one durable series. + + ''' + key: tuple[str, int] = (fqme, timeframe) + return self._write_locks.setdefault( + key, + trio.Lock(), + ) + + @acm + async def _open_file_lock( + self, + fqme: str, + timeframe: int, + + ) -> AsyncIterator[None]: + ''' + Serialize a series read-merge-write across client processes. + + ''' + path: Path = self.mk_path(fqme, timeframe) + lock_path: Path = path.with_name(f'.{path.name}.lock') + with lock_path.open('a+b') as lock_file: + while True: + try: + flock( + lock_file.fileno(), + LOCK_EX | LOCK_NB, + ) + break + except BlockingIOError: + await trio.sleep(0.01) + try: + yield + finally: + flock(lock_file.fileno(), LOCK_UN) + async def read_ohlcv( self, fqme: str, @@ -323,12 +381,10 @@ class NativeStorageClient: df: pl.DataFrame = tsp.np2pl(ohlcv) else: df = ohlcv - - self._cache_df( - fqme=fqme, - df=df, - timeframe=timeframe, + df = df.with_columns( + pl.Series('index', np.arange(df.height)) ) + self._validate_ohlcv(df) # TODO: in terms of managing the ultra long term data # -[ ] use a proper profiler to measure all this IO and @@ -338,7 +394,41 @@ class NativeStorageClient: # -[ ] try out ``fastparquet``'s append writing: # https://fastparquet.readthedocs.io/en/latest/api.html#fastparquet.write start = time.time() - df.write_parquet(path) + with NamedTemporaryFile( + dir=path.parent, + prefix=f'.{path.name}.', + suffix='.tmp', + delete=False, + ) as tmp_file: + tmp_path = Path(tmp_file.name) + + try: + df.write_parquet(tmp_path) + committed: pl.DataFrame = pl.read_parquet(tmp_path) + self._validate_ohlcv(committed) + if not committed.equals(df): + raise IOError( + 'Temporary parquet differs from input frame' + ) + + with tmp_path.open('rb') as tmp_file: + os.fsync(tmp_file.fileno()) + tmp_path.replace(path) + + self._cache_df( + fqme=fqme, + df=committed, + timeframe=timeframe, + ) + + dir_fd: int = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(dir_fd) + finally: + os.close(dir_fd) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise delay: float = round( time.time() - start, ndigits=6, @@ -349,6 +439,107 @@ class NativeStorageClient: ) return path + def _validate_ohlcv( + self, + df: pl.DataFrame, + + ) -> None: + ''' + Validate timestamp invariants before publishing a series. + + ''' + if df.is_empty(): + raise ValueError('Can not write an empty OHLCV series') + + required: tuple[str, ...] = tuple( + name + for name, _ in def_iohlcv_fields + ) + missing: set[str] = set(required).difference(df.columns) + if missing: + raise ValueError( + f'OHLCV frame is missing columns: {sorted(missing)}' + ) + + for field in required: + values: np.ndarray = df[field].to_numpy() + if ( + not np.issubdtype(values.dtype, np.number) + or + not np.all(np.isfinite(values)) + ): + raise ValueError( + f'OHLCV column {field!r} must contain ' + f'finite numeric data' + ) + + times: np.ndarray = df['time'].to_numpy() + if ( + not np.all(np.isfinite(times)) + or + np.any(times <= 0) + ): + raise ValueError( + 'OHLCV timestamps must be finite and positive' + ) + + if np.any(np.diff(times) <= 0): + raise ValueError( + 'OHLCV timestamps must be strictly increasing' + ) + + async def update_ohlcv( + self, + fqme: str, + ohlcv: np.ndarray | pl.DataFrame, + timeframe: int, + + ) -> Path: + ''' + Merge an incremental frame into durable OHLCV history. + + Incoming rows replace stored rows at matching timestamps. + Writes for each series are serialized so concurrent backfills + can not overwrite each other's read-merge-write cycle. + + ''' + lock: trio.Lock = self._get_write_lock( + fqme, + timeframe, + ) + async with lock: + async with self._open_file_lock(fqme, timeframe): + if isinstance(ohlcv, np.ndarray): + incoming: pl.DataFrame = tsp.np2pl(ohlcv) + else: + incoming = ohlcv + self._validate_ohlcv(incoming) + + path: Path = self.mk_path(fqme, timeframe) + if path.exists(): + stored: pl.DataFrame = pl.read_parquet(path) + merged: pl.DataFrame = pl.concat( + [stored, incoming], + how='diagonal_relaxed', + ) + merged = ( + merged + .unique( + subset='time', + keep='last', + ) + .sort('time') + ) + else: + merged = incoming + + self._validate_ohlcv(merged) + return self._write_ohlcv( + fqme, + merged, + timeframe, + ) + async def write_ohlcv( self, fqme: str, @@ -361,11 +552,17 @@ class NativeStorageClient: to (local) disk. ''' - return self._write_ohlcv( + lock: trio.Lock = self._get_write_lock( fqme, - ohlcv, timeframe, ) + async with lock: + async with self._open_file_lock(fqme, timeframe): + return self._write_ohlcv( + fqme, + ohlcv, + timeframe, + ) async def delete_ts( self, diff --git a/piker/tsp/_history.py b/piker/tsp/_history.py index a4006df0..4ace4e06 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -78,7 +78,6 @@ from piker.brokers._util import ( ) from piker.storage import TimeseriesNotFound from ._anal import ( - dedupe, get_null_segs, iter_null_segs, Frame, @@ -123,6 +122,7 @@ _default_rt_size: int = _days_worth * _secs_in_day # NOTE: start the append index in rt buffer such that 1 day's worth # can be appenened before overrun. _rt_buffer_start = int((_days_worth - 1) * _secs_in_day) +_notify_timeout_s: float = 1 def diff_history( @@ -148,6 +148,64 @@ def diff_history( return array[times >= prepend_until_dt.timestamp()] +def mk_storage_key(mkt: MktPair) -> str: + ''' + Build the historical storage key for a market. + + Always drop the source-asset token for non-currency-pair market + types. For historical reasons the table-key schema stores + `tsla.nasdaq.ib`, not `tsla/usd.nasdaq.ib`, while currency pairs + and crypto-settled futures retain both assets. + + ''' + if ( + mkt.dst.atype not in { + 'crypto', + 'crypto_currency', + 'fiat', # a "forex pair" + 'perpetual_future', # stupid "perps" from cex land + } + and not ( + mkt.src.atype == 'crypto_currency' + and + mkt.dst.atype == 'future' + ) + ): + return mkt.get_fqme( + delim_char='', + without_src=True, + ) + + return mkt.get_fqme(delim_char='') + + +async def notify_backfill( + sampler_stream: tractor.MsgStream, + mkt: MktPair, + timeframe: float, + +) -> None: + ''' + Broadcast an SHM update without making teardown unbounded. + + ''' + with trio.move_on_after( + _notify_timeout_s, + shield=True, + ) as cs: + await sampler_stream.send({ + 'broadcast_all': { + 'backfilling': (mkt.fqme, timeframe), + }, + }) + + if cs.cancelled_caught: + log.warning( + f'Timed out broadcasting backfill for ' + f'{timeframe}@{mkt.fqme}' + ) + + async def shm_push_in_between( shm: ShmArray, to_push: np.ndarray, @@ -618,10 +676,11 @@ async def start_backfill( ) # await tractor.pause() - to_push = diff_history( + to_store: np.ndarray = diff_history( array, prepend_until_dt=backfill_until_dt, ) + to_push: np.ndarray = to_store ln: int = len(to_push) if ln: log.info( @@ -661,6 +720,23 @@ async def start_backfill( ) break + if ( + storage is not None + and + write_tsdb + ): + log.info( + f'Writing {len(to_store)} frame to storage:\n' + f'{next_start_dt} -> {last_start_dt}' + ) + await storage.update_ohlcv( + mk_storage_key(mkt), + to_store, + timeframe, + ) + + shm_full: bool = False + # bail gracefully on shm allocation overrun/full # condition try: @@ -671,11 +747,6 @@ async def start_backfill( backfill_until_dt=backfill_until_dt, update_start_on_prepend=update_start_on_prepend, ) - await sampler_stream.send({ - 'broadcast_all': { - 'backfilling': (mkt.fqme, timeframe), - }, - }) # decrement next prepend point next_prepend_index = next_prepend_index - ln @@ -687,7 +758,7 @@ async def start_backfill( f'Reached buffer start (index={next_prepend_index}), ' f'stopping backfill' ) - break + shm_full = True except ValueError as ve: _ve = ve @@ -702,6 +773,7 @@ async def start_backfill( ) to_push = to_push[-next_prepend_index + 1:] + ln = len(to_push) await shm_push_in_between( shm, to_push, @@ -709,89 +781,23 @@ async def start_backfill( backfill_until_dt=backfill_until_dt, update_start_on_prepend=update_start_on_prepend, ) - await sampler_stream.send({ - 'broadcast_all': { - 'backfilling': (mkt.fqme, timeframe), - }, - }) - - # XXX, can't push the entire frame? so - # push only the amount that can fit.. - break + next_prepend_index = next_prepend_index - ln + last_start_dt = next_start_dt + shm_full = True log.info( f'Shm pushed {ln} frame:\n' f'{next_start_dt} -> {last_start_dt}' ) - # FINALLY, maybe write immediately to the tsdb backend for - # long-term storage. - if ( - storage is not None - and - write_tsdb - ): - log.info( - f'Writing {ln} frame to storage:\n' - f'{next_start_dt} -> {last_start_dt}' - ) + await notify_backfill( + sampler_stream, + mkt, + timeframe, + ) - # NOTE, always drop the src asset token for - # non-currency-pair like market types (for now) - # - # THAT IS, for now our table key schema is NOT - # including the dst[/src] source asset token. SO, - # 'tsla.nasdaq.ib' over 'tsla/usd.nasdaq.ib' for - # historical reasons ONLY. - if ( - mkt.dst.atype not in { - 'crypto', - 'crypto_currency', - 'fiat', # a "forex pair" - 'perpetual_future', # stupid "perps" from cex land - } - and not ( - mkt.src.atype == 'crypto_currency' - and - mkt.dst.atype in { - 'future', - } - ) - ): - col_sym_key: str = mkt.get_fqme( - delim_char='', - without_src=True, - ) - else: - col_sym_key: str = mkt.get_fqme( - delim_char='', - ) - - await storage.write_ohlcv( - col_sym_key, - shm.array, - timeframe, - ) - df: pl.DataFrame = await storage.as_df( - fqme=mkt.fqme, - period=timeframe, - load_from_offline=False, - ) - ( - wdts, - deduped, - diff, - ) = dedupe(df) - if diff: - log.warning( - f'Found {diff!r} duplicates in tsdb! ' - f'=> Overwriting with `deduped` data !! <=\n' - ) - await storage.write_ohlcv( - col_sym_key, - deduped, - timeframe, - ) + if shm_full: + break else: # finally filled gap @@ -950,11 +956,11 @@ async def back_load_from_tsdb( # await sampler_stream.send('broadcast_all') -async def push_latest_frame( +async def query_latest_frame( # box-type only that should get packed with the datetime # objects received for the latest history frame dt_eps: list[DateTime, DateTime], - shm: ShmArray, + frames: list[np.ndarray], get_hist: Callable[ [int, datetime, datetime], tuple[np.ndarray, str] @@ -983,7 +989,7 @@ async def push_latest_frame( mr_start_dt, mr_end_dt, ]) - task_status.started(dt_eps) + frames.append(array) # XXX: timeframe not supported for backend (since # above exception type), terminate immediately since @@ -997,18 +1003,38 @@ async def push_latest_frame( # prolly tf not supported return None - # NOTE: on the first history, most recent history - # frame we PREPEND from the current shm ._last index - # and thus a gap between the earliest datum loaded here - # and the latest loaded from the tsdb may exist! + task_status.started(dt_eps) + + return dt_eps + + +async def publish_latest_frame( + storage: StorageClient, + mkt: MktPair, + shm: ShmArray, + array: np.ndarray, + timeframe: float, + +) -> None: + ''' + Persist then publish the most-recent provider frame. + + ''' + await storage.update_ohlcv( + mk_storage_key(mkt), + array, + timeframe, + ) + + # NOTE: on the first history, most recent history frame we + # PREPEND from the current shm ._last index. A gap may exist + # between its earliest datum and the latest loaded from the tsdb. log.info(f'Pushing {array.size} to shm!') shm.push( array, prepend=True, # append on first frame ) - return dt_eps - async def load_tsdb_hist( storage: StorageClient, @@ -1094,14 +1120,15 @@ async def tsdb_backfill( # concurrently load the provider's most-recent-frame AND any # pre-existing tsdb history already saved in `piker` storage. dt_eps: list[DateTime, DateTime] = [] + latest_frames: list[np.ndarray] = [] async with ( tractor.trionics.collapse_eg(), trio.open_nursery() as tn ): tn.start_soon( - push_latest_frame, + query_latest_frame, dt_eps, - shm, + latest_frames, get_hist, timeframe, config, @@ -1112,6 +1139,15 @@ async def tsdb_backfill( timeframe, ) + if latest_frames: + await publish_latest_frame( + storage, + mkt, + shm, + latest_frames[0], + timeframe, + ) + # tell parent task to continue # TODO: really we'd want this the other way with the # tsdb load happening asap and the since the latest diff --git a/tests/test_history_backfill.py b/tests/test_history_backfill.py index 1a58e460..0a26ea52 100644 --- a/tests/test_history_backfill.py +++ b/tests/test_history_backfill.py @@ -6,12 +6,20 @@ from functools import partial from types import SimpleNamespace import numpy as np -from pendulum import datetime +from pendulum import ( + datetime, + from_timestamp, +) +import pytest import trio from piker.brokers import DataUnavailable from piker.data._source import def_iohlcv_fields -from piker.tsp._history import start_backfill +from piker.tsp._history import ( + notify_backfill, + publish_latest_frame, + start_backfill, +) def run_exhausted_backfill(get_hist) -> None: @@ -67,3 +75,147 @@ def test_empty_frame_completes() -> None: return frame, end_dt, end_dt run_exhausted_backfill(get_hist) + + +def test_storage_receives_full_provider_delta() -> None: + ''' + SHM capacity truncation does not truncate durable history. + + ''' + frame = np.zeros( + 2, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['index'] = [0, 1] + frame['time'] = [60, 120] + + events: list[str] = [] + + class Shm: + def __init__(self) -> None: + self.pushed: list[np.ndarray] = [] + + def push(self, array, **kwargs) -> None: + self.pushed.append(array.copy()) + events.append('shm') + + class Sampler: + async def send(self, msg) -> None: + events.append('sampler') + + class Storage: + def __init__(self) -> None: + self.frames: list[np.ndarray] = [] + + async def update_ohlcv( + self, + fqme: str, + ohlcv: np.ndarray, + timeframe: int, + ) -> None: + self.frames.append(ohlcv.copy()) + events.append('storage') + + shm = Shm() + storage = Storage() + mkt = SimpleNamespace( + fqme='x.test', + dst=SimpleNamespace(atype='crypto'), + src=SimpleNamespace(atype='crypto_currency'), + get_fqme=lambda **kwargs: 'x.test', + ) + + async def get_hist(*args, **kwargs): + return ( + frame, + from_timestamp(60), + from_timestamp(120), + ) + + async def main() -> None: + with trio.fail_after(0.5): + await start_backfill( + get_hist=get_hist, + def_frame_duration=None, + mod=SimpleNamespace(name='fake'), + mkt=mkt, + shm=shm, + timeframe=60, + backfill_from_shm_index=1, + backfill_from_dt=from_timestamp(180), + sampler_stream=Sampler(), + backfill_until_dt=from_timestamp(60), + storage=storage, + write_tsdb=True, + ) + + trio.run(main) + assert shm.pushed[0]['time'].tolist() == [120] + assert storage.frames[0]['time'].tolist() == [60, 120] + assert events == ['storage', 'shm', 'sampler'] + + +def test_latest_frame_is_persisted_before_shm() -> None: + ''' + Startup persists the most-recent provider frame before publication. + + ''' + frame = np.zeros( + 2, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['time'] = [60, 120] + events: list[str] = [] + + class Storage: + async def update_ohlcv(self, *args) -> None: + events.append('storage') + + class Shm: + def push(self, array, **kwargs) -> None: + events.append('shm') + + async def main() -> None: + with trio.fail_after(0.5): + await publish_latest_frame( + storage=Storage(), + mkt=SimpleNamespace( + fqme='x.test', + dst=SimpleNamespace(atype='crypto'), + src=SimpleNamespace(atype='crypto_currency'), + get_fqme=lambda **kwargs: 'x.test', + ), + shm=Shm(), + array=frame, + timeframe=60, + ) + + trio.run(main) + assert events == ['storage', 'shm'] + + +def test_backfill_notification_timeout_is_bounded( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A wedged sampler can not indefinitely shield actor teardown. + + ''' + monkeypatch.setattr( + 'piker.tsp._history._notify_timeout_s', + 0.01, + ) + + class Sampler: + async def send(self, msg) -> None: + await trio.sleep_forever() + + async def main() -> None: + with trio.fail_after(0.1): + await notify_backfill( + Sampler(), + SimpleNamespace(fqme='x.test'), + 60, + ) + + trio.run(main) diff --git a/tests/test_storage_nativedb.py b/tests/test_storage_nativedb.py new file mode 100644 index 00000000..a1f4b674 --- /dev/null +++ b/tests/test_storage_nativedb.py @@ -0,0 +1,364 @@ +''' +NativeDB durability and history-preservation regressions. + +''' +from fcntl import ( + flock, + LOCK_EX, + LOCK_UN, +) +import os +from pathlib import Path + +import numpy as np +import polars as pl +import pytest +import trio + +from piker import tsp +from piker.data._source import def_iohlcv_fields +from piker.storage.nativedb import NativeStorageClient + + +def mk_ohlcv( + times: tuple[float, ...], + closes: tuple[float, ...] | None = None, + +) -> np.ndarray: + ''' + Build a minimal structured OHLCV frame. + + ''' + array = np.zeros( + len(times), + dtype=np.dtype(def_iohlcv_fields), + ) + array['index'] = np.arange(len(times)) + array['time'] = times + array['close'] = closes or times + return array + + +def run(coro) -> None: + ''' + Run a NativeDB operation with a bounded Trio clock. + + ''' + async def main() -> None: + with trio.fail_after(1): + await coro + + trio.run(main) + + +@pytest.mark.parametrize('use_polars', [False, True]) +def test_numpy_and_polars_round_trip( + tmp_path: Path, + use_polars: bool, +) -> None: + ''' + Both supported frame types survive durable serialization. + + ''' + client = NativeStorageClient(tmp_path) + array = mk_ohlcv( + (60, 120, 180), + (1, 2, 3), + ) + payload = tsp.np2pl(array) if use_polars else array + + run(client.write_ohlcv('x.test', payload, 60)) + loaded = trio.run(client.read_ohlcv, 'x.test', 60) + + assert loaded['time'].tolist() == [60, 120, 180] + assert loaded['close'].tolist() == [1, 2, 3] + + +def test_update_preserves_history_and_resolves_conflicts( + tmp_path: Path, +) -> None: + ''' + Incremental writes retain old rows and prefer incoming bars. + + ''' + client = NativeStorageClient(tmp_path) + old = mk_ohlcv( + (60, 120, 180), + (1, 2, 3), + ) + incoming = mk_ohlcv( + (180, 240), + (30, 4), + ) + + run(client.write_ohlcv('x.test', old, 60)) + run(client.update_ohlcv('x.test', incoming, 60)) + stored = pl.read_parquet(client.mk_path('x.test', 60)) + + assert stored['time'].to_list() == [60, 120, 180, 240] + assert stored['close'].to_list() == [1, 2, 30, 4] + assert stored['index'].to_list() == [0, 1, 2, 3] + + +def test_write_ohlcv_remains_an_explicit_replacement( + tmp_path: Path, +) -> None: + ''' + Full repair writes can intentionally remove persisted rows. + + ''' + client = NativeStorageClient(tmp_path) + old = mk_ohlcv((60, 120, 180)) + replacement = mk_ohlcv((120, 180)) + + run(client.write_ohlcv('x.test', old, 60)) + run(client.write_ohlcv('x.test', replacement, 60)) + stored = pl.read_parquet(client.mk_path('x.test', 60)) + + assert stored['time'].to_list() == [120, 180] + + +def test_failed_write_preserves_file_and_cache( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A failed parquet write leaves durable and cached data intact. + + ''' + client = NativeStorageClient(tmp_path) + old = mk_ohlcv((60, 120)) + incoming = mk_ohlcv((180,)) + run(client.write_ohlcv('x.test', old, 60)) + + def fail_write( + df: pl.DataFrame, + file: Path, + *args, + **kwargs, + ) -> None: + Path(file).write_bytes(b'partial parquet') + raise OSError('simulated write failure') + + monkeypatch.setattr( + pl.DataFrame, + 'write_parquet', + fail_write, + ) + with pytest.raises(OSError, match='simulated write failure'): + run(client.update_ohlcv('x.test', incoming, 60)) + + stored = pl.read_parquet(client.mk_path('x.test', 60)) + cached = trio.run(client.as_df, 'x.test', 60, False) + assert stored['time'].to_list() == [60, 120] + assert cached['time'].to_list() == [60, 120] + assert not list(tmp_path.glob('*.tmp')) + + +def test_invalid_temporary_parquet_is_not_committed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + The reopened candidate must validate before atomic replacement. + + ''' + client = NativeStorageClient(tmp_path) + old = mk_ohlcv((60, 120)) + run(client.write_ohlcv('x.test', old, 60)) + real_read = pl.read_parquet + + def corrupt_candidate(source, *args, **kwargs): + if Path(source).suffix == '.tmp': + return pl.DataFrame({'time': [180]}) + return real_read(source, *args, **kwargs) + + monkeypatch.setattr(pl, 'read_parquet', corrupt_candidate) + with pytest.raises(ValueError, match='missing columns'): + run(client.update_ohlcv( + 'x.test', + mk_ohlcv((180,)), + 60, + )) + + path = client.mk_path('x.test', 60) + stored = real_read(path) + cached = trio.run(client.as_df, 'x.test', 60, False) + assert stored['time'].to_list() == [60, 120] + assert cached['time'].to_list() == [60, 120] + assert not list(tmp_path.glob('*.tmp')) + + +def test_directory_sync_failure_keeps_visible_cache_consistent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A post-replace durability error leaves cache matching the file. + + ''' + client = NativeStorageClient(tmp_path) + run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60)) + real_fsync = os.fsync + calls: int = 0 + + def fail_directory_sync(fd: int) -> None: + nonlocal calls + calls += 1 + if calls == 2: + raise OSError('simulated directory sync failure') + real_fsync(fd) + + monkeypatch.setattr(os, 'fsync', fail_directory_sync) + with pytest.raises(OSError, match='directory sync failure'): + run(client.update_ohlcv( + 'x.test', + mk_ohlcv((120,)), + 60, + )) + + stored = pl.read_parquet(client.mk_path('x.test', 60)) + cached = trio.run(client.as_df, 'x.test', 60, False) + assert stored['time'].to_list() == [60, 120] + assert cached['time'].to_list() == [60, 120] + + +@pytest.mark.parametrize( + 'times', + [ + (0, 60), + (60, 60), + (120, 60), + ], +) +def test_update_rejects_invalid_timestamps( + tmp_path: Path, + times: tuple[float, ...], +) -> None: + ''' + Incremental input must have positive, increasing timestamps. + + ''' + client = NativeStorageClient(tmp_path) + incoming = mk_ohlcv(times) + + with pytest.raises(ValueError): + run(client.update_ohlcv('x.test', incoming, 60)) + + assert not client.mk_path('x.test', 60).exists() + + +def test_replacement_rejects_invalid_timestamps( + tmp_path: Path, +) -> None: + ''' + Explicit replacement enforces the same durable invariants. + + ''' + client = NativeStorageClient(tmp_path) + + with pytest.raises(ValueError): + run(client.write_ohlcv( + 'x.test', + mk_ohlcv((60, 60)), + 60, + )) + + assert not client.mk_path('x.test', 60).exists() + + +def test_write_rejects_invalid_schema_and_values( + tmp_path: Path, +) -> None: + ''' + Durable OHLCV columns must exist and contain finite numbers. + + ''' + client = NativeStorageClient(tmp_path) + with pytest.raises(ValueError, match='missing columns'): + run(client.write_ohlcv( + 'x.test', + pl.DataFrame({'time': [60]}), + 60, + )) + + invalid = mk_ohlcv((60,)) + invalid['close'] = np.nan + with pytest.raises(ValueError, match='finite numeric'): + run(client.write_ohlcv('x.test', invalid, 60)) + + +def test_series_do_not_interfere( + tmp_path: Path, +) -> None: + ''' + FQME and timeframe keys isolate merge and cache state. + + ''' + client = NativeStorageClient(tmp_path) + run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60)) + run(client.write_ohlcv('x.test', mk_ohlcv((1, 2)), 1)) + run(client.write_ohlcv('y.test', mk_ohlcv((60, 120)), 60)) + run(client.update_ohlcv('x.test', mk_ohlcv((120,)), 60)) + + x_60 = pl.read_parquet(client.mk_path('x.test', 60)) + x_1 = pl.read_parquet(client.mk_path('x.test', 1)) + y_60 = pl.read_parquet(client.mk_path('y.test', 60)) + assert x_60['time'].to_list() == [60, 120] + assert x_1['time'].to_list() == [1, 2] + assert y_60['time'].to_list() == [60, 120] + + +def test_index_files_ignores_lock_and_stale_temp_files( + tmp_path: Path, +) -> None: + ''' + Crash leftovers and writer locks are not durable series entries. + + ''' + client = NativeStorageClient(tmp_path) + run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60)) + (tmp_path / 'x.test.ohlcv60s.parquet.crash.tmp').touch() + + index = client.index_files() + + assert list(index) == ['x.test'] + assert index['x.test']['period'] == 60 + + +def test_contended_file_lock_yields_to_trio( + tmp_path: Path, +) -> None: + ''' + Cross-client lock contention does not block the actor loop. + + ''' + client = NativeStorageClient(tmp_path) + run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60)) + path = client.mk_path('x.test', 60) + lock_path = path.with_name(f'.{path.name}.lock') + + async def main() -> None: + done = trio.Event() + + async def update() -> None: + await client.update_ohlcv( + 'x.test', + mk_ohlcv((120,)), + 60, + ) + done.set() + + with lock_path.open('a+b') as lock_file: + flock(lock_file.fileno(), LOCK_EX) + async with trio.open_nursery() as nursery: + nursery.start_soon(update) + await trio.sleep(0.03) + assert not done.is_set() + flock(lock_file.fileno(), LOCK_UN) + with trio.fail_after(0.5): + await done.wait() + + trio.run(main) + stored = pl.read_parquet(path) + assert stored['time'].to_list() == [60, 120]