Clean test-owned SHM leaks in `pytest`

Track successful `create=True` calls through Tractor's
current-process factory instead of scanning `/dev/shm` or trusting
the attachment cache.

- cooperate with normal actor unlink and restore token-cache state
- verify POSIX object identity before fallback cleanup
- fail after cleaning exact surviving segments
- reproduce the pre-`current_actor()` allocation leak

Prompt-IO: ai/prompt-io/opencode/20260727T215906Z_689df816_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

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

@@ -166,8 +166,13 @@ 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, shared-memory, or socket reaper. Never apply
-  historical broad `pkill -f tractor._child` guidance automatically.
+  repository-local process or socket reaper. Never apply 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

@@ -0,0 +1,40 @@
+---
+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

@@ -0,0 +1,30 @@
+---
+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

@@ -0,0 +1,42 @@
+---
+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

@@ -0,0 +1,34 @@
+---
+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

@@ -0,0 +1,41 @@
+---
+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

@@ -0,0 +1,25 @@
+---
+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,26 +51,19 @@ 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
@@ -127,7 +120,13 @@ def unpack_fqme_from_parquet_filepath(path: Path) -> str:
     return fqme
 
 
-ohlc_key_map = None
+# Only hydrate fields shared by canonical storage and provider SHM.
+# 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:
@@ -151,10 +150,6 @@ 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:
@@ -273,49 +268,56 @@ class NativeStorageClient:
             {},
         )[fqme] = df
 
-    def _get_write_lock(
+    def _canonicalize_ohlcv(
         self,
-        fqme: str,
-        timeframe: int,
+        df: pl.DataFrame,
 
-    ) -> trio.Lock:
+    ) -> pl.DataFrame:
         '''
-        Return the actor-local lock for one durable series.
+        Normalize a provider frame to the durable OHLCV schema.
+
+        Provider frames may omit the derived ``index`` field and add
+        provider-only fields. Durable frames contain only canonical
+        fields in their declared order.
 
         '''
-        key: tuple[str, int] = (fqme, timeframe)
-        return self._write_locks.setdefault(
-            key,
-            trio.Lock(),
+        required: tuple[str, ...] = tuple(
+            name
+            for name, _ in def_iohlcv_fields
+        )
+        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)}'
             )
 
-    @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,
+        times: np.ndarray = df['time'].to_numpy()
+        if (
+            not np.issubdtype(times.dtype, np.number)
+            or
+            not np.all(np.isfinite(times))
+        ):
+            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 (
+            df
+            .with_columns(
+                pl.Series('index', np.arange(df.height))
+            )
+            .select(required)
+            .cast(schema)
         )
-                    break
-                except BlockingIOError:
-                    await trio.sleep(0.01)
-            try:
-                yield
-            finally:
-                flock(lock_file.fileno(), LOCK_UN)
 
     async def read_ohlcv(
         self,
@@ -381,9 +383,7 @@ class NativeStorageClient:
             df: pl.DataFrame = tsp.np2pl(ohlcv)
         else:
             df = ohlcv
-        df = df.with_columns(
-            pl.Series('index', np.arange(df.height))
-        )
+        df = self._canonicalize_ohlcv(df)
         self._validate_ohlcv(df)
 
         # TODO: in terms of managing the ultra long term data
@@ -499,25 +499,21 @@ class NativeStorageClient:
         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.
+        Writer ownership belongs to the persistent ``datad`` feed task.
 
         '''
-        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 = pl.read_parquet(path)
+            stored: pl.DataFrame = self._canonicalize_ohlcv(
+                pl.read_parquet(path)
+            )
             merged: pl.DataFrame = pl.concat(
                 [stored, incoming],
                 how='diagonal_relaxed',
@@ -552,12 +548,6 @@ 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,10 +682,7 @@ def pl2np(
         df.height,
         dtype,
     )
-    for field, col in zip(
-        dtype.fields,
-        df.columns,
-    ):
-        array[field] = df.get_column(col).to_numpy()
+    for field in dtype.fields:
+        array[field] = df.get_column(field).to_numpy()
 
     return array

tests/conftest.py

@@ -1,8 +1,13 @@
 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
@@ -89,6 +94,151 @@ 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,18 +3,31 @@ 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,
@@ -194,6 +207,118 @@ 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

@@ -0,0 +1,109 @@
+'''
+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,11 +2,6 @@
 NativeDB durability and history-preservation regressions.
 
 '''
-from fcntl import (
-    flock,
-    LOCK_EX,
-    LOCK_UN,
-)
 import os
 from pathlib import Path
 
@@ -74,6 +69,42 @@ 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:
@@ -267,6 +298,30 @@ 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:
@@ -309,15 +364,22 @@ def test_series_do_not_interfere(
     assert y_60['time'].to_list() == [60, 120]
 
 
-def test_index_files_ignores_lock_and_stale_temp_files(
+def test_index_files_ignores_sidecar_files(
     tmp_path: Path,
 ) -> None:
     '''
-    Crash leftovers and writer locks are not durable series entries.
+    Legacy writer locks and crash leftovers are not 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()
@@ -326,39 +388,25 @@ def test_index_files_ignores_lock_and_stale_temp_files(
     assert index['x.test']['period'] == 60
 
 
-def test_contended_file_lock_yields_to_trio(
+def test_writes_create_no_lock_sidecars(
     tmp_path: Path,
 ) -> None:
     '''
-    Cross-client lock contention does not block the actor loop.
+    Actor-owned NativeDB writes must not create lock sidecars.
+
+    ``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)
-    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 not list(tmp_path.glob('*.lock'))
     assert stored['time'].to_list() == [60, 120]