pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 11 Packages Projects Releases Wiki Activity

Make `NativeDB` backfills atomic and non-destructive 

- merge provider deltas without dropping unaffected history
- validate reopened temp parquet before atomic replacement
- serialize writers with actor-local and cancellable file locks
- publish cache only after disk replacement
- persist full startup/backfill frames before SHM publication
- bound sampler notifications and avoid MarketStore duplicates

Prompt-IO: ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
backfiller_deep_fixes
Gud Boi 2026-07-21 22:50:09 -04:00
parent c60a5a45f3
commit b000db94e0
8 changed files with 969 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md 100644
View File

@ -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.

43
ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md 100644
View File

@ -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

9
piker/storage/__init__.py
View File

@ -139,6 +139,15 @@ class StorageClient(
) -> None:
...
async def update_ohlcv(
self,
fqme: str,
ohlcv: np.ndarray,
timeframe: int,
) -> None:
...
class TimeseriesNotFound(Exception):
'''

18
piker/storage/marketstore/__init__.py
View File

@ -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

213
piker/storage/nativedb.py
View File

@ -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,6 +552,12 @@ class NativeStorageClient:
to (local) disk.
'''
lock: trio.Lock = self._get_write_lock(
fqme,
timeframe,
)
async with lock:
async with self._open_file_lock(fqme, timeframe):
return self._write_ohlcv(
fqme,
ohlcv,

224
piker/tsp/_history.py
View File

@ -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,90 +781,24 @@ 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}'
)
# 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,
await notify_backfill(
sampler_stream,
mkt,
timeframe,
)
if shm_full:
break
else:
# finally filled gap
log.info(
@ -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

156
tests/test_history_backfill.py
View File

@ -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)

364
tests/test_storage_nativedb.py 100644
View File

@ -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]