.claude/skills/run-tests/test-harness-reference.md

@@ -166,13 +166,8 @@ whole suite supports UDS merely because the Tractor plugin exposes it.
   before fixtures can isolate config. If explicitly requested, isolate
   `XDG_CONFIG_HOME` before Python starts and use the proper Qt/Nix shell.
 - Piker and the installed Tractor pytest plugin do not provide a
-  repository-local process or socket reaper. Never apply historical broad
+  repository-local process, shared-memory, or socket reaper. Never apply
-  `pkill -f tractor._child` guidance automatically.
+  historical broad `pkill -f tractor._child` guidance automatically.
-- The function-scoped autouse `shm_leak_tracker` fixture wraps Tractor's
-  current-process `SharedMemory` factory. It tracks only successful
-  `create=True` calls, restores the pre-test `_known_tokens` cache, and
-  unlinks exact surviving names before failing the leaking test. It never
-  scans `/dev/shm` or unlinks attachments created by another process.
 
 ## Test Layout
 

ai/prompt-io/opencode/20260727T204711Z_84a6d47b_prompt_io.md

@@ -1,40 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: unavailable
-timestamp: 2026-07-27T20:47:11Z
-git_ref: 84a6d47b
-scope: code
-substantive: true
-raw_file: 20260727T204711Z_84a6d47b_prompt_io.raw.md
----
-
-## Prompt
-
-The user identified that `backfiller_deep_fixes` introduced persistent
-`.parquet.lock` files, questioned whether Parquet or Arrow already handled
-writer coordination, and clarified that piker's `tractor` structured
-concurrency design already provides single-writer ownership through the
-persistent `datad` feed task. The user then asked to open the backfiller
-worktree and resolve the redundant locking.
-
-## Response summary
-
-Removed redundant actor-local and filesystem locking while preserving
-atomic crash-safe Parquet replacement. Replaced the synthetic contention
-test with regressions for the actual cross-branch failure: writes create
-no lock sidecars, and legacy lock artifacts are ignored during indexing.
-
-## Files changed
-
-- `piker/storage/nativedb.py` - rely on `datad` writer ownership
-- `tests/test_storage_nativedb.py` - lock-sidecar regressions
-- `ai/prompt-io/opencode/20260727T204711Z_84a6d47b_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260727T204711Z_84a6d47b_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260727T204711Z_84a6d47b_prompt_io.raw.md

@@ -1,30 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-27T20:47:11Z
-git_ref: 84a6d47b
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py`
-
-Removed actor-local and filesystem writer locks from NativeDB. The
-persistent `datad` feed task already owns writes for each series, while
-its timeframe children write distinct files. NativeDB retains validated
-temporary writes, file and directory synchronization, and atomic target
-replacement for crash durability. The exact `.parquet` suffix filter is
-retained so legacy lock sidecars are harmless.
-
-> `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py`
-
-Removed the synthetic cross-client lock-contention test. Added regression
-coverage proving replacement and incremental writes create no lock
-sidecars. Extended indexing coverage with the legacy
-`.parquet.lock` artifact that caused upstream `flake_update` to crash.
-
-Verification generated with the patch:
-
-- NativeDB and history regression set: 20 passed
-- Ruff: passed
-- `git diff --check`: passed

ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md

@@ -1,42 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-27T21:24:54Z
-git_ref: 689df816
-scope: code
-substantive: true
-raw_file: 20260727T212454Z_689df816_prompt_io.raw.md
----
-
-## Prompt
-
-The user reran the known IB FQME on `backfiller_deep_fixes` and reported
-that `publish_latest_frame()` still failed because NativeDB required an
-`index` absent from IB's provider frame. The user called out the missed
-tabular schema boundary and requested an actual end-to-end test suite
-rather than storage fakes.
-
-## Response summary
-
-Repaired the IB-provider-to-NativeDB boundary by canonicalizing durable
-fields and mapping reloads into provider-specific SHM buffers. Added a
-local actor/SHM/Parquet integration regression covering first startup and
-restart, plus focused conversion and timestamp-coercion tests. Two review
-rounds found and resolved restart hydration and fractional-time hazards.
-
-## Files changed
-
-- `piker/storage/nativedb.py` - canonical storage and SHM field mapping
-- `piker/tsp/_anal.py` - name-based Polars-to-NumPy conversion
-- `tests/test_history_backfill.py` - actor/SHM/Parquet integration test
-- `tests/test_storage_nativedb.py` - schema conversion regressions
-- `ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md

@@ -1,34 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-27T21:24:54Z
-git_ref: 689df816
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py piker/tsp/_anal.py`
-
-Normalized index-less provider frames to NativeDB's canonical names,
-order, and dtypes before merge and publication. Provider-only columns
-are excluded from durable files, derived indexes are regenerated, and
-fractional timestamps are rejected before integer conversion. Added a
-canonical field map for hydrating provider-specific SHM buffers while
-leaving provider-only fields at their defaults. Changed Polars-to-NumPy
-conversion to select columns by field name rather than position.
-
-> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py tests/test_storage_nativedb.py`
-
-Added deterministic coverage using IB's actual provider dtype and a real
-local `tractor` root actor, IB-typed shared-memory buffers, NativeDB,
-Parquet persistence, first-start publication, and restart hydration.
-Added focused regressions for reordered/extra Polars columns and
-fractional timestamp rejection.
-
-Verification generated with the patch:
-
-- NativeDB and history integration set: 23 passed
-- Ruff: passed
-- `git diff --check`: passed
-- second adversarial review: no findings
-- no live gateway or network tests run

ai/prompt-io/opencode/20260727T215906Z_689df816_prompt_io.md

@@ -1,41 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-27T21:59:06Z
-git_ref: 689df816
-scope: code
-substantive: true
-raw_file: 20260727T215906Z_689df816_prompt_io.raw.md
----
-
-## Prompt
-
-After the actor/SHM/Parquet integration test exposed a stale SHM triplet,
-the user requested leak-cleaner machinery in the pytest harness using
-Tractor's existing ownership and lifetime patterns.
-
-## Response summary
-
-Added creator-scoped SHM leak tracking to pytest without scanning
-`/dev/shm` or treating attachments as owned. The fixture cooperates with
-normal Tractor teardown, verifies POSIX object identity before fallback
-cleanup, restores process-local token state, and fails after cleaning a
-leak. Regressions cover the exact pre-actor-registration allocation window,
-positional creators, and external attachment safety.
-
-## Files changed
-
-- `tests/conftest.py` - creator-scoped SHM leak tracking fixture
-- `tests/test_shm_cleanup.py` - ownership and failure-window regressions
-- `.claude/skills/run-tests/test-harness-reference.md`
-  - document current-process SHM cleanup behavior
-- `ai/prompt-io/opencode/20260727T215906Z_689df816_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260727T215906Z_689df816_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not been edited by the human.

ai/prompt-io/opencode/20260727T215906Z_689df816_prompt_io.raw.md

@@ -1,25 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-27T21:59:06Z
-git_ref: 689df816
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- tests/conftest.py tests/test_shm_cleanup.py .claude/skills/run-tests/test-harness-reference.md`
-
-Added function-scoped SHM ownership tracking around Tractor's test-process
-factory. The fixture records only successful creators, removes ownership
-records during normal Tractor teardown, verifies POSIX object identity
-before fallback unlink, restores the token-cache baseline, and fails after
-cleaning any leak. Regressions reproduce allocation failure before actor
-lifetime registration and prove external attachments remain untouched.
-
-Verification generated with the patch:
-
-- SHM cleanup regressions: 2 passed
-- combined SHM, NativeDB, and history set: 25 passed
-- Ruff: passed
-- `git diff --check`: passed
-- final adversarial review: no findings

piker/storage/nativedb.py

@@ -51,19 +51,26 @@ 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
 
-from bidict import bidict
 import numpy as np
 import polars as pl
 from pendulum import (
     from_timestamp,
 )
+import trio
 
 from piker import config
 from piker import tsp
@@ -120,13 +127,7 @@ def unpack_fqme_from_parquet_filepath(path: Path) -> str:
     return fqme
 
 
-# Only hydrate fields shared by canonical storage and provider SHM.
+ohlc_key_map = None
-# Provider-only fields remain at their buffer defaults on restart.
-ohlc_key_map = bidict({
-    name: name
-    for name, _ in def_iohlcv_fields
-    if name != 'index'
-})
 
 
 class NativeStorageClient:
@@ -150,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:
@@ -268,56 +273,49 @@ class NativeStorageClient:
             {},
         )[fqme] = df
 
-    def _canonicalize_ohlcv(
+    def _get_write_lock(
         self,
-        df: pl.DataFrame,
+        fqme: str,
+        timeframe: int,
 
-    ) -> pl.DataFrame:
+    ) -> trio.Lock:
         '''
-        Normalize a provider frame to the durable OHLCV schema.
+        Return the actor-local lock for one durable series.
-
-        Provider frames may omit the derived ``index`` field and add
-        provider-only fields. Durable frames contain only canonical
-        fields in their declared order.
 
         '''
-        required: tuple[str, ...] = tuple(
+        key: tuple[str, int] = (fqme, timeframe)
-            name
+        return self._write_locks.setdefault(
-            for name, _ in def_iohlcv_fields
+            key,
-        )
+            trio.Lock(),
-        schema = {
-            name: pl.Int64 if field_type is int else pl.Float64
-            for name, field_type in def_iohlcv_fields
-        }
-        missing: set[str] = set(required).difference(df.columns)
-        missing.discard('index')
-        if missing:
-            raise ValueError(
-                f'OHLCV frame is missing columns: {sorted(missing)}'
         )
 
-        times: np.ndarray = df['time'].to_numpy()
+    @acm
-        if (
+    async def _open_file_lock(
-            not np.issubdtype(times.dtype, np.number)
+        self,
-            or
+        fqme: str,
-            not np.all(np.isfinite(times))
+        timeframe: int,
-        ):
-            raise ValueError(
-                "OHLCV column 'time' must contain finite numeric data"
-            )
-        if np.any(times != np.floor(times)):
-            raise ValueError(
-                'OHLCV timestamps must use whole-second values'
-            )
 
-        return (
+    ) -> AsyncIterator[None]:
-            df
+        '''
-            .with_columns(
+        Serialize a series read-merge-write across client processes.
-                pl.Series('index', np.arange(df.height))
+
-            )
+        '''
-            .select(required)
+        path: Path = self.mk_path(fqme, timeframe)
-            .cast(schema)
+        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,
@@ -383,7 +381,9 @@ class NativeStorageClient:
             df: pl.DataFrame = tsp.np2pl(ohlcv)
         else:
             df = ohlcv
-        df = self._canonicalize_ohlcv(df)
+        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
@@ -499,21 +499,25 @@ class NativeStorageClient:
         Merge an incremental frame into durable OHLCV history.
 
         Incoming rows replace stored rows at matching timestamps.
-        Writer ownership belongs to the persistent ``datad`` feed task.
+        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
-        incoming = self._canonicalize_ohlcv(incoming)
                 self._validate_ohlcv(incoming)
 
                 path: Path = self.mk_path(fqme, timeframe)
                 if path.exists():
-            stored: pl.DataFrame = self._canonicalize_ohlcv(
+                    stored: pl.DataFrame = pl.read_parquet(path)
-                pl.read_parquet(path)
-            )
                     merged: pl.DataFrame = pl.concat(
                         [stored, incoming],
                         how='diagonal_relaxed',
@@ -548,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,

piker/tsp/_anal.py

@@ -682,7 +682,10 @@ def pl2np(
         df.height,
         dtype,
     )
-    for field in dtype.fields:
+    for field, col in zip(
-        array[field] = df.get_column(field).to_numpy()
+        dtype.fields,
+        df.columns,
+    ):
+        array[field] = df.get_column(col).to_numpy()
 
     return array

tests/conftest.py

@@ -1,13 +1,8 @@
 from contextlib import asynccontextmanager as acm
-from collections.abc import Callable
 from functools import partial
 import logging
 import os
 from pathlib import Path
-from weakref import (
-    ReferenceType,
-    ref,
-)
 
 import pytest
 import tractor
@@ -94,151 +89,6 @@ def log(
     )
 
 
-@pytest.fixture(autouse=True)
-def shm_leak_tracker(
-    monkeypatch: pytest.MonkeyPatch,
-
-) -> Callable[[], set[str]]:
-    '''
-    Clean SHM created but not released by the current test process.
-
-    Tractor's token cache includes attachments, so cache membership
-    can not prove ownership. Track only successful ``create=True``
-    calls through Tractor's own factory. Normal actor-stack teardown
-    unlinks these names first; any survivors are exact test-owned leaks
-    which this fixture removes before failing the test.
-
-    '''
-    from tractor.ipc import _shm
-
-    known_tokens: dict = dict(_shm._known_tokens)
-    owned_segments: dict[
-        str,
-        tuple[
-            tuple[int, int]|None,
-            ReferenceType,
-        ],
-    ] = {}
-    open_shm = _shm.SharedMemory
-    unlink_shm = getattr(_shm, 'shm_unlink', None)
-
-    def normalize_name(name: str) -> str:
-        return str(name).lstrip('/')
-
-    def segment_id(segment) -> tuple[int, int]|None:
-        fd: int = getattr(segment, '_fd', -1)
-        if fd < 0:
-            return None
-        stat = os.fstat(fd)
-        return stat.st_dev, stat.st_ino
-
-    def open_test_shm(*args, **kwargs):
-        segment = open_shm(*args, **kwargs)
-        create: bool = kwargs.get(
-            'create',
-            args[1] if len(args) > 1 else False,
-        )
-        if create:
-            name: str = normalize_name(segment.name)
-            owned_segments[name] = (
-                segment_id(segment),
-                ref(segment),
-            )
-        return segment
-
-    if unlink_shm is not None:
-        def unlink_owned_name(name: str) -> None:
-            normalized: str = normalize_name(name)
-            try:
-                unlink_shm(name)
-            except FileNotFoundError:
-                owned_segments.pop(normalized, None)
-                raise
-            else:
-                owned_segments.pop(normalized, None)
-
-        monkeypatch.setattr(
-            _shm,
-            'shm_unlink',
-            unlink_owned_name,
-        )
-
-    def cleanup() -> set[str]:
-        leaked: set[str] = set()
-        errors: list[Exception] = []
-        try:
-            for name, record in list(owned_segments.items()):
-                expected_id, segment_ref = record
-                try:
-                    segment = open_shm(
-                        name=name,
-                        create=False,
-                    )
-                except FileNotFoundError:
-                    segment = None
-                except Exception as err:
-                    errors.append(err)
-                    segment = None
-                else:
-                    try:
-                        actual_id: tuple[int, int]|None = segment_id(
-                            segment
-                        )
-                        if expected_id is None:
-                            errors.append(RuntimeError(
-                                f'Can not verify SHM ownership: {name}'
-                            ))
-                        elif actual_id != expected_id:
-                            errors.append(RuntimeError(
-                                f'SHM name changed ownership: {name}'
-                            ))
-                        else:
-                            leaked.add(name)
-                            segment.unlink()
-                    except FileNotFoundError:
-                        pass
-                    except Exception as err:
-                        errors.append(err)
-                    finally:
-                        try:
-                            segment.close()
-                        except Exception as err:
-                            errors.append(err)
-
-                original = segment_ref()
-                if original is not None:
-                    try:
-                        original.close()
-                    except Exception as err:
-                        errors.append(err)
-        finally:
-            owned_segments.clear()
-            _shm._known_tokens.clear()
-            _shm._known_tokens.update(known_tokens)
-
-        if errors:
-            raise ExceptionGroup(
-                'Failed to clean test-owned SHM',
-                errors,
-            )
-        return leaked
-
-    monkeypatch.setattr(
-        _shm,
-        'SharedMemory',
-        open_test_shm,
-    )
-    yield cleanup
-
-    leaked: set[str] = cleanup()
-    if leaked:
-        names: str = '\n'.join(sorted(leaked))
-        pytest.fail(
-            f'Test leaked SHM segments; cleaned:\n'
-            f'{names}'
-        )
-
-
 @acm
 async def _open_test_pikerd(
     tmpconfdir: str,

tests/test_history_backfill.py

@@ -3,31 +3,18 @@ Deterministic history-backfill regressions.
 
 '''
 from functools import partial
-from pathlib import Path
 from types import SimpleNamespace
-from uuid import uuid4
 
 import numpy as np
 from pendulum import (
     datetime,
     from_timestamp,
 )
-import polars as pl
 import pytest
-import tractor
 import trio
 
 from piker.brokers import DataUnavailable
-from piker.brokers.ib.api import (
-    _bar_load_dtype,
-    _ohlc_dtype,
-)
-from piker.data._sharedmem import maybe_open_shm_array
 from piker.data._source import def_iohlcv_fields
-from piker.storage.nativedb import (
-    NativeStorageClient,
-    ohlc_key_map,
-)
 from piker.tsp._history import (
     notify_backfill,
     publish_latest_frame,
@@ -207,118 +194,6 @@ def test_latest_frame_is_persisted_before_shm() -> None:
     assert events == ['storage', 'shm']
 
 
-def test_ib_latest_frame_round_trips_through_nativedb(
-    tmp_path: Path,
-) -> None:
-    '''
-    Persist and reload IB's provider schema through history startup.
-
-    ``publish_latest_frame()`` previously passed IB's index-less bars
-    directly to NativeDB, whose durable-schema validator rejected the
-    missing derived ``index``. IB also appends ``count`` after OHLCV,
-    which exposed positional Polars-to-NumPy conversion to field
-    corruption. Build the actual IB load dtype with distinct values,
-    run the real history publication, NativeDB Parquet, and ``ShmArray``
-    paths, then hydrate a fresh IB buffer from storage. Assertions prove
-    first-start publication preserves provider fields while restart maps
-    canonical fields and leaves provider-only ``count`` at its default.
-
-    '''
-    frame = np.zeros(
-        2,
-        dtype=np.dtype(_bar_load_dtype),
-    )
-    frame['time'] = [60, 120]
-    frame['open'] = [1.1, 2.1]
-    frame['high'] = [1.2, 2.2]
-    frame['low'] = [1.0, 2.0]
-    frame['close'] = [1.15, 2.15]
-    frame['volume'] = [10, 20]
-    frame['count'] = [3, 4]
-
-    storage = NativeStorageClient(tmp_path)
-    shm_key = f'test_ib_history_{uuid4().hex}'
-    mkt = SimpleNamespace(
-        fqme='mnq.cme.20260918.ib',
-        dst=SimpleNamespace(atype='continuous_future'),
-        src=SimpleNamespace(atype='fiat'),
-        get_fqme=lambda **kwargs: 'mnq.cme.20260918.ib',
-    )
-
-    async def main() -> None:
-        with trio.fail_after(2):
-            async with tractor.open_root_actor(
-                name=shm_key,
-                tpt_bind_addrs=[('127.0.0.1', 0)],
-            ):
-                shm, opened = maybe_open_shm_array(
-                    key=f'{shm_key}_first',
-                    size=16,
-                    dtype=np.dtype(_ohlc_dtype),
-                    append_start_index=8,
-                )
-                restart_shm, restart_opened = maybe_open_shm_array(
-                    key=f'{shm_key}_restart',
-                    size=16,
-                    dtype=np.dtype(_ohlc_dtype),
-                    append_start_index=8,
-                )
-                assert opened
-                assert restart_opened
-
-                await publish_latest_frame(
-                    storage=storage,
-                    mkt=mkt,
-                    shm=shm,
-                    array=frame,
-                    timeframe=60,
-                )
-                loaded = await storage.read_ohlcv(
-                    mkt.fqme,
-                    timeframe=60,
-                )
-                restart_shm.push(
-                    loaded,
-                    prepend=True,
-                    field_map=ohlc_key_map,
-                )
-                stored = pl.read_parquet(
-                    storage.mk_path(mkt.fqme, 60)
-                )
-                canonical = [
-                    name
-                    for name, _ in def_iohlcv_fields
-                ]
-                canonical_schema = {
-                    name: (
-                        pl.Int64
-                        if field_type is int
-                        else pl.Float64
-                    )
-                    for name, field_type in def_iohlcv_fields
-                }
-
-                assert stored.columns == canonical
-                assert dict(stored.schema) == canonical_schema
-                assert list(loaded.dtype.fields) == canonical
-                assert loaded['index'].tolist() == [0, 1]
-                for field in canonical[1:]:
-                    assert (
-                        loaded[field].tolist()
-                        ==
-                        frame[field].tolist()
-                    )
-                    assert (
-                        restart_shm.array[field].tolist()
-                        ==
-                        frame[field].tolist()
-                    )
-                assert shm.array['count'].tolist() == [3, 4]
-                assert restart_shm.array['count'].tolist() == [0, 0]
-
-    trio.run(main)
-
-
 def test_backfill_notification_timeout_is_bounded(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/test_shm_cleanup.py

@@ -1,109 +0,0 @@
-'''
-Pytest shared-memory ownership and leak-cleanup regressions.
-
-'''
-from collections.abc import Callable
-from uuid import uuid4
-
-import numpy as np
-import pytest
-from tractor._exceptions import NoRuntime
-from tractor.ipc import _shm
-from tractor.ipc._mp_bs import disable_mantracker
-
-from piker.data._sharedmem import maybe_open_shm_array
-from piker.data._source import def_iohlcv_fields
-
-
-def test_shm_tracker_catches_pre_actor_failure(
-    shm_leak_tracker: Callable[[], set[str]],
-) -> None:
-    '''
-    Clean allocations which fail before actor-lifetime registration.
-
-    ``open_shm_ndarray()`` creates its data and index segments before it
-    asks ``current_actor()`` for the lifetime stack. Without a runtime,
-    that lookup raises after all three OS names exist and the old harness
-    leaked them. Reproduce that ordering through Piker's real wrapper,
-    invoke tracked cleanup, and prove the data, first, and last segments
-    plus their process-local token are all removed.
-
-    '''
-    key = f'test_shm_failure_{uuid4().hex}'
-    with pytest.raises(NoRuntime):
-        maybe_open_shm_array(
-            key=key,
-            size=4,
-            dtype=np.dtype(def_iohlcv_fields),
-        )
-
-    token = _shm.NDToken.from_msg(_shm._known_tokens[key])
-    names = {
-        token.shm_name,
-        token.shm_first_index_name,
-        token.shm_last_index_name,
-    }
-    assert shm_leak_tracker() == names
-    assert key not in _shm._known_tokens
-    for name in names:
-        with pytest.raises(FileNotFoundError):
-            _shm.SharedMemory(
-                name=name,
-                create=False,
-            )
-
-
-def test_shm_leak_tracker_unlinks_only_owned_segments(
-    shm_leak_tracker: Callable[[], set[str]],
-) -> None:
-    '''
-    Test cleanup must unlink creators without harming attachments.
-
-    A failed SHM allocation can escape before Tractor registers actor
-    lifetime callbacks. Broad cache or filesystem cleanup is unsafe
-    because tests may attach to a live Piker segment they do not own.
-    Create one external segment through an unpatched factory, attach to
-    it through Tractor, and create one test-owned segment through the
-    tracked factory. Invoke cleanup and prove only the owned name was
-    removed while the external segment remains attachable.
-
-    '''
-    open_external_shm = disable_mantracker()
-    external = None
-    attachment = None
-    owned = None
-    try:
-        external = open_external_shm(
-            create=True,
-            size=8,
-        )
-        attachment = _shm.SharedMemory(
-            name=external.name,
-            create=False,
-        )
-        owned = _shm.SharedMemory(None, True, 8)
-        cleaned: set[str] = shm_leak_tracker()
-
-        assert cleaned == {owned.name}
-        with pytest.raises(FileNotFoundError):
-            _shm.SharedMemory(
-                name=owned.name,
-                create=False,
-            )
-
-        survivor = _shm.SharedMemory(
-            name=external.name,
-            create=False,
-        )
-        survivor.close()
-    finally:
-        if owned is not None:
-            owned.close()
-        if attachment is not None:
-            attachment.close()
-        if external is not None:
-            try:
-                external.unlink()
-            except FileNotFoundError:
-                pass
-            external.close()

tests/test_storage_nativedb.py

@@ -2,6 +2,11 @@
 NativeDB durability and history-preservation regressions.
 
 '''
+from fcntl import (
+    flock,
+    LOCK_EX,
+    LOCK_UN,
+)
 import os
 from pathlib import Path
 
@@ -69,42 +74,6 @@ def test_numpy_and_polars_round_trip(
     assert loaded['close'].tolist() == [1, 2, 3]
 
 
-def test_pl2np_maps_fields_by_name() -> None:
-    '''
-    Polars conversion must not depend on DataFrame column positions.
-
-    Provider and legacy Parquet frames can include extra columns or
-    present canonical fields in a different order. The old ``zip()``
-    conversion paired NumPy field names with DataFrame positions,
-    silently assigning unrelated values. Build distinct canonical
-    values, prepend provider-only ``count``, reverse canonical order,
-    and prove every structured-array field is selected by its name.
-
-    '''
-    expected = mk_ohlcv(
-        (60, 120),
-        (1.15, 2.15),
-    )
-    expected['open'] = [1.1, 2.1]
-    expected['high'] = [1.2, 2.2]
-    expected['low'] = [1.0, 2.0]
-    expected['volume'] = [10, 20]
-    canonical = [name for name, _ in def_iohlcv_fields]
-    reordered = (
-        tsp.np2pl(expected)
-        .with_columns(pl.Series('count', [3, 4]))
-        .select(['count', *reversed(canonical)])
-    )
-
-    actual = tsp.pl2np(
-        reordered,
-        dtype=expected.dtype,
-    )
-
-    for field in canonical:
-        assert actual[field].tolist() == expected[field].tolist()
-
-
 def test_update_preserves_history_and_resolves_conflicts(
     tmp_path: Path,
 ) -> None:
@@ -298,30 +267,6 @@ def test_replacement_rejects_invalid_timestamps(
     assert not client.mk_path('x.test', 60).exists()
 
 
-def test_fractional_timestamps_are_not_truncated(
-    tmp_path: Path,
-) -> None:
-    '''
-    Durable timestamp coercion must not silently alter provider data.
-
-    NativeDB declares integer epoch seconds, while IB delivers its
-    timestamps in a floating dtype. Casting before validation truncated
-    fractional values and could collapse distinct rows at one second.
-    Supply otherwise-valid half-second values and prove the write rejects
-    them before publishing any Parquet path.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    frame = tsp.np2pl(mk_ohlcv((60, 120))).with_columns(
-        pl.Series('time', [60.5, 120.5])
-    )
-
-    with pytest.raises(ValueError, match='whole-second'):
-        run(client.write_ohlcv('x.test', frame, 60))
-
-    assert not client.mk_path('x.test', 60).exists()
-
-
 def test_write_rejects_invalid_schema_and_values(
     tmp_path: Path,
 ) -> None:
@@ -364,22 +309,15 @@ def test_series_do_not_interfere(
     assert y_60['time'].to_list() == [60, 120]
 
 
-def test_index_files_ignores_sidecar_files(
+def test_index_files_ignores_lock_and_stale_temp_files(
     tmp_path: Path,
 ) -> None:
     '''
-    Legacy writer locks and crash leftovers are not series entries.
+    Crash leftovers and writer locks are not durable series entries.
-
-    Earlier deep-fix revisions created persistent
-    ``.parquet.lock`` files beside each series. The upstream indexer
-    mistakes those sidecars for Parquet data and crashes while parsing
-    their period. Arrange both a legacy lock and stale temporary file,
-    then prove exact-suffix indexing exposes only the durable series.
 
     '''
     client = NativeStorageClient(tmp_path)
     run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    (tmp_path / '.x.test.ohlcv60s.parquet.lock').touch()
     (tmp_path / 'x.test.ohlcv60s.parquet.crash.tmp').touch()
 
     index = client.index_files()
@@ -388,25 +326,39 @@ def test_index_files_ignores_sidecar_files(
     assert index['x.test']['period'] == 60
 
 
-def test_writes_create_no_lock_sidecars(
+def test_contended_file_lock_yields_to_trio(
     tmp_path: Path,
 ) -> None:
     '''
-    Actor-owned NativeDB writes must not create lock sidecars.
+    Cross-client lock contention does not block the actor loop.
-
-    ``datad`` already gives each persistent feed one parent history
-    writer, with its child tasks writing distinct timeframe files. A
-    redundant filesystem lock previously leaked ``.parquet.lock``
-    files into NativeDB and made upstream ``flake_update`` crash during
-    startup. Exercise replacement and incremental writes, then prove
-    no lock artifact exists and the merged history remains intact.
 
     '''
     client = NativeStorageClient(tmp_path)
     run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    run(client.update_ohlcv('x.test', mk_ohlcv((120,)), 60))
     path = client.mk_path('x.test', 60)
-    stored = pl.read_parquet(path)
+    lock_path = path.with_name(f'.{path.name}.lock')
 
-    assert not list(tmp_path.glob('*.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]