Skip stale trade ticks in 1s sampler

Queued IB startup trades can arrive after samplerd has advanced
the current 1s bar. Writing them into that bar creates a false
price and volume spike. Compare each stamped trade with the
current bar before updating either SHM buffer. Warn whenever a
next-slot trade must wait for samplerd; skip other mismatches
and leave untimestamped feeds on the receive-time path.

Add focused tests for epoch units, queued trades, and warning
before the next-slot wait. Note that incomplete provider bars
still need a correction path for late trades.

Prompt-IO: ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md

(this patch was generated in some part by `codex` using `gpt-6` (`openai`))

ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md

@@ -0,0 +1,45 @@
+---
+model: gpt-6
+service: codex
+session: sampler_tick_time_guard
+timestamp: 2026-09-25T19:05:23Z
+git_ref: c0dae89a
+scope: code
+substantive: true
+raw_file: 20260925T190523Z_c0dae89a_prompt_io.raw.md
+---
+
+## Prompt
+
+The user observed an oversized MNQ 1s bar just after boot and asked
+to keep queued startup trades out of a different timestamped bar.
+After direct chart and separate pikerd restarts showed no new spike,
+the user requested the fix on a separate branch based on
+`ib_connectivity_harden`, followed by `$commit-plan`.
+
+## Response summary
+
+Normalize provider trade timestamps and compare them with the
+sampler's current 1s bar. Wait briefly for a one-step-ahead tick,
+warn that its price update is delayed, skip other timestamp
+mismatches, and preserve behavior for feeds without tick
+timestamps. Note the remaining late-bar correction race and add
+focused sampler tests.
+
+## Files changed
+
+- `piker/data/_sampling.py` — guard writes by tick and bar time.
+- `tests/test_sampling_tick_times.py` — exercise startup backlog and
+  next-step behavior.
+
+## Human edits
+
+The human identified the oversized MNQ boot bar, proposed a sampler
+startup race, directed the timestamp check, and live-tested the fix
+in both direct-chart and separate-pikerd setups. The human chose to
+isolate this fix from the ongoing segment-repair review on a branch
+based on `ib_connectivity_harden`. During local tuicr review, the
+human required a warning before each next-slot wait because chart
+price latency matters, and called out the race between incomplete
+history bars and late trade ticks. The agent applied the edits and
+added a TODO for IB and provider-neutral bar reconciliation.

ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.raw.md

@@ -0,0 +1,34 @@
+---
+model: gpt-6
+service: codex
+timestamp: 2026-09-25T19:05:23Z
+git_ref: c0dae89a
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt and generated output
+
+The user reported a large MNQ 1s startup bar and asked us to ensure
+queued trades are not aggregated into the current timestep when their
+timestamps indicate otherwise. After a direct chart restart and a
+separate pikerd run showed no spike, the user requested the tested fix
+on a separate worktree branch based on `ib_connectivity_harden`, then
+requested `$commit-plan`.
+
+> `git diff HEAD~1..HEAD -- piker/data/_sampling.py`
+
+Generated `trade_time_s()` to normalize common epoch units. In
+`sample_and_broadcast()`, timestamped trades reach the current 1s bar
+only when their rounded time matches the bar. A trade for the next bar
+waits briefly for samplerd's step. Older or otherwise mismatched
+trades are skipped, with one warning per symbol. Untimestamped trades
+retain the existing receive-time behavior.
+
+> `git diff HEAD~1..HEAD -- tests/test_sampling_tick_times.py`
+
+Generated focused tests for epoch units, startup backlog trades,
+untimestamped trades, and the next-step wait.
+
+The isolated worktree's focused test run passed: 8 tests. Ruff and
+`git diff --check` passed. The user also live-tested the same guard
+on direct chart and separate pikerd starts without seeing the spike.

ai/prompt-io/opencode/20260924_ib_idle_timer.md

@@ -0,0 +1,42 @@
+---
+model: openai/gpt-6-astra
+service: opencode
+scope: code
+substantive: true
+git_ref: efa72980
+raw_file: 20260924_ib_idle_timer.raw.md
+---
+
+## Prompt
+
+The human reported silent IB roaming and requested an immediate fix
+after investigation found the custom nanosecond timestamps incompatible
+with ib_async's datetime-based idle timeout arithmetic.
+
+## Response summary
+
+Override NonShittyWrapper._setTimer to accept integer nanosecond packet
+timestamps while delegating datetime/sentinel states upstream. Keep the
+packet timestamp unchanged on integer-path expiry, clear the timer
+before timeoutEvent emission, and preserve reentrant listener rearming.
+No default heartbeat, reconnect, or idle-timeout arming is introduced.
+
+## Files changed
+
+- piker/brokers/ib/api.py: compatible idle timer arithmetic/lifecycle.
+- tests/test_ib_wrapper_timeout.py: three deterministic offline tests.
+
+## Verification
+
+Nineteen tests passed: new wrapper-timeout regressions and existing IB
+history/method-proxy tests. Restoring the upstream method in memory
+reproduced datetime-minus-int TypeError. No IB service or live network
+was contacted; installed dependencies and original user work were
+unchanged. The isolated worktree index remains unchanged.
+
+## Human contributions
+
+The human supplied rfkill/suspend/roam evidence, requested a session
+health design, and prioritized fixing the discovered timestamp mismatch.
+The implementation deliberately preserves their nanosecond feed schema.
+Acceptance and live validation remain pending.

ai/prompt-io/opencode/20260924_ib_idle_timer.raw.md

@@ -0,0 +1,17 @@
+---
+model: openai/gpt-6-astra
+service: opencode
+git_ref: efa72980
+diff_cmd: git diff HEAD -- piker/brokers/ib/api.py tests/test_ib_wrapper_timeout.py
+---
+
+> git diff HEAD -- piker/brokers/ib/api.py tests/test_ib_wrapper_timeout.py
+
+NonShittyWrapper._setTimer accepts upstream datetime state and piker's
+integer packet-nanosecond state. Expiry clears its handle and timeout
+before notifying listeners, without replacing packet timestamps.
+Tests cover traffic rescheduling, exactly-once expiry, pretraffic and
+sentinel handling, disabling, and callback rearming.
+
+Verification: 19 passed, including existing IB history/proxy tests.
+Wall-clock packet silence remains distinct from session health.

piker/brokers/ib/api.py

@@ -82,6 +82,7 @@ from ib_async import (
     Wrapper,
     RequestError,
 )
+from ib_async.util import getLoop
 import numpy as np
 
 # TODO: in hindsight, probably all imports should be
@@ -147,9 +148,45 @@ _bar_sizes = {
 
 _show_wap_in_history: bool = False
 
-# overrides to sidestep pretty questionable design decisions in
-# ``ib_async``:
 class NonShittyWrapper(Wrapper):
+    '''
+    Overrides to sidestep pretty questionable design decisions in
+    ``ib_async``.
+
+    '''
+
+    def _setTimer(self, delay: float = 0):
+        '''
+        Accept packet epoch nanoseconds alongside upstream datetimes.
+
+        This measures wall-clock packet silence, not session health.
+
+        '''
+        if isinstance(self.lastTime, datetime):
+            return super()._setTimer(delay)
+
+        if not self._timeout:
+            return
+
+        diff = (time.time_ns() - self.lastTime) / 1_000_000_000
+        if not delay:
+            delay = self._timeout - diff
+
+        if delay > 0:
+            self._timeoutHandle = getLoop().call_later(
+                delay,
+                self._setTimer,
+            )
+        else:
+            self._logger.debug('Timeout')
+            # Match Wrapper.setTimeout(0)'s timer teardown, but keep
+            # lastTime in ns for quote/fill and timeoutEvent readers.
+            if self._timeoutHandle:
+                self._timeoutHandle.cancel()
+            self._timeoutHandle = None
+            self._timeout = 0
+            self.ib.timeoutEvent.emit(diff)
+
     def tcpDataArrived(self):
         """Override time stamps to be floats for now.
         """

piker/data/_sampling.py

@@ -25,6 +25,7 @@ from collections import (
     defaultdict,
 )
 from contextlib import asynccontextmanager as acm
+from math import isfinite
 import time
 from typing import (
     Any,
@@ -67,6 +68,42 @@ if TYPE_CHECKING:
 # tick-event stream.
 _default_delay_s: float = 1.0
 
+# Give the independent samplerd incrementer one step to catch up
+# before deciding a tick stamped for the next 1s bar is unusable.
+_next_tick_step_wait_s: float = 1.5
+
+
+def trade_time_s(tick: dict) -> float|None:
+    '''
+    Read a trade's epoch time in seconds, if one was supplied.
+
+    Broker feeds may omit timestamps. IB's `TickData.time` uses
+    nanoseconds; other feeds may use seconds, milliseconds, or
+    microseconds. An absent or invalid value retains the existing
+    receive-time sampling behavior.
+
+    '''
+    raw: Any = tick.get('time')
+    if (
+        isinstance(raw, bool)
+        or
+        not isinstance(raw, (int, float))
+    ):
+        return None
+
+    seconds: float = float(raw)
+    if not isfinite(seconds) or seconds <= 0:
+        return None
+
+    if seconds >= 1e17:
+        seconds /= 1e9
+    elif seconds >= 1e14:
+        seconds /= 1e6
+    elif seconds >= 1e11:
+        seconds /= 1e3
+
+    return seconds
+
 
 class Sampler:
     '''
@@ -610,6 +647,7 @@ async def sample_and_broadcast(
     log.info("Started shared mem bar writer")
 
     overruns = Counter()
+    mismatched_ticks: Counter[str] = Counter()
 
     # NOTE, only used for debugging live-data-feed issues, though
     # this should be resolved more correctly in the future using the
@@ -658,6 +696,60 @@ async def sample_and_broadcast(
                 # write trade events to shm last OHLC sample
                 if ticktype in ('trade', 'utrade'):
 
+                    tick_s: float|None = trade_time_s(tick)
+                    bar_s: float = float(rt_shm.array[-1]['time'])
+                    if (
+                        tick_s is not None
+                        and
+                        round(tick_s) == round(bar_s) + 1
+                    ):
+                        # The chart cannot see this trade price until
+                        # `Sampler.increment_ohlc_buffer()` advances.
+                        # TODO: emit a `FeedDelayed` event to feed
+                        # subscribers if this occurs beyond startup.
+                        log.warning(
+                            'DELAYING LIVE TRADE SHM UPDATE until '
+                            f'next 1s OHLC slot for '
+                            f'{broker_symbol}!\n'
+                            f'tick_time_s: {tick_s}\n'
+                            f'current_bar_s: {bar_s}\n'
+                            f'max_wait_s: {_next_tick_step_wait_s}'
+                        )
+                        with trio.move_on_after(
+                            _next_tick_step_wait_s
+                        ):
+                            while round(tick_s) > round(bar_s):
+                                await trio.sleep(0.01)
+                                bar_s = float(
+                                    rt_shm.array[-1]['time']
+                                )
+                    if (
+                        tick_s is not None
+                        and
+                        round(tick_s) != round(bar_s)
+                    ):
+                        # The newest provider frame may already hold
+                        # queued startup trades. Replaying them into
+                        # the sampler's current bar misdates them and
+                        # can double-count provider history. Leave
+                        # past rows to the history/backfill path.
+                        # Sampler.increment_ohlc_buffer() rounds its
+                        # 1s step timestamp, so use the same bucket.
+                        # TODO: `Client.bars()` fetches a one-shot
+                        # latest frame; a late trade may be absent
+                        # from its unfinished last bar. Reconcile IB
+                        # `barUpdateEvent` from `keepUpToDate=True`,
+                        # or add a provider-neutral back-bar
+                        # correction path.
+                        mismatched_ticks[broker_symbol] += 1
+                        if mismatched_ticks[broker_symbol] == 1:
+                            log.warning(
+                                'Skipping trade outside current 1s '
+                                f'bar for {broker_symbol}: '
+                                f'tick={tick_s}, bar={bar_s}'
+                            )
+                        continue
+
                     last = tick['price']
 
                     # more compact inline-way to do this assignment
@@ -781,7 +873,7 @@ async def sample_and_broadcast(
                                     # channel-ipc-pipe is still up.
                                     if (
                                         not chan.connected()
-                                        or ctx._cancel_called
+                                        or ctx.cancel_called
                                     ):
                                         log.warning(
                                             'Dropping broken consumer:\n'

tests/providers/__init__.py

@@ -0,0 +1,4 @@
+'''
+Provider-specific regression tests.
+
+'''

tests/providers/test_ib_wrapper_timeout.py

@@ -0,0 +1,183 @@
+'''
+Offline packet-idle timer regressions using real eventkit events.
+
+'''
+import asyncio
+from datetime import datetime
+from types import SimpleNamespace
+
+from eventkit import Event
+from ib_async import wrapper as upstream
+import pytest
+
+from piker.brokers.ib import api
+
+
+@pytest.fixture
+def timer(monkeypatch):
+    '''
+    Drive real asyncio TimerHandles with a deterministic wall clock.
+
+    '''
+    class ClockLoop:
+        now = 1_800_000_000.0
+
+        def __init__(self):
+            self.handles = []
+
+        def get_debug(self):
+            return False
+
+        def _timer_handle_cancelled(self, handle):
+            pass
+
+        def call_exception_handler(self, context):
+            raise context['exception']
+
+        def call_later(self, delay, callback):
+            handle = asyncio.TimerHandle(
+                self.now + delay, callback, (), self,
+            )
+            self.handles.append(handle)
+            return handle
+
+        def advance(self, seconds):
+            target = self.now + seconds
+            while self.handles:
+                handle = min(self.handles, key=lambda h: h.when())
+                if handle.when() > target:
+                    break
+                self.handles.remove(handle)
+                self.now = handle.when()
+                if not handle.cancelled():
+                    handle._run()
+            self.now = target
+
+    loop = ClockLoop()
+
+    class ClockDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return cls.fromtimestamp(loop.now, tz)
+
+    monkeypatch.setattr(upstream, 'datetime', ClockDateTime)
+    monkeypatch.setattr(upstream, 'getLoop', lambda: loop)
+    monkeypatch.setattr(api, 'getLoop', lambda: loop)
+    monkeypatch.setattr(
+        api.time, 'time_ns', lambda: int(loop.now * 1_000_000_000),
+    )
+    ib = SimpleNamespace(timeoutEvent=Event('timeoutEvent'))
+    wrapper = api.NonShittyWrapper(ib)
+    events = []
+
+    def record(diff):
+        events.append((diff, wrapper.lastTime))
+
+    ib.timeoutEvent.connect(record)
+    return loop, wrapper, events
+
+
+def test_packet_timeout_reschedules_and_emits_once(timer):
+    '''
+    Upstream datetime subtraction raised TypeError after a packet
+    replaced NonShittyWrapper.lastTime with epoch nanoseconds.
+    Arm the real wrapper, deliver packets around the first deadline,
+    and drive its TimerHandles without sleeping or opening sockets.
+    New traffic must postpone expiry by the remaining idle interval;
+    the real eventkit listener must see seconds and the intact ns
+    timestamp exactly once, with the timer already disabled.
+
+    '''
+    loop, wrapper, events = timer
+    wrapper.setTimeout(5)
+    loop.advance(1)
+    wrapper.tcpDataArrived()
+    assert isinstance(wrapper.lastTime, int)
+    loop.advance(3)
+    wrapper.tcpDataArrived()
+    packet_time = wrapper.lastTime
+    loop.advance(1)
+    assert events == []
+    assert wrapper.lastTime == packet_time
+    assert wrapper._timeoutHandle.when() == loop.now + 4
+    loop.advance(4)
+    assert events == [(5.0, packet_time)]
+    assert wrapper._timeoutHandle is None
+    assert wrapper._timeout == 0
+    assert wrapper.lastTime == packet_time
+    wrapper._setTimer()
+    loop.advance(20)
+    assert events == [(5.0, packet_time)]
+
+
+def test_datetime_timeout_and_sentinel(timer):
+    '''
+    Before any packet, Wrapper.setTimeout uses datetime timestamps.
+    The ns fix must preserve that upstream path and datetime.min's
+    sentinel short circuit. Run an armed pre-traffic timer to expiry,
+    then invoke the sentinel path and assert no callback is queued.
+
+    '''
+    loop, wrapper, events = timer
+    assert isinstance(wrapper.lastTime, datetime)
+    assert wrapper._timeoutHandle is None
+    wrapper.setTimeout(3)
+    loop.advance(3)
+    assert len(events) == 1
+    assert events[0][0] == 3.0
+    assert isinstance(events[0][1], datetime)
+    assert wrapper._timeoutHandle is None
+    wrapper.lastTime = datetime.min
+    wrapper._setTimer(2)
+    loop.advance(10)
+    assert len(events) == 1
+    assert wrapper.lastTime == datetime.min
+    assert wrapper._timeoutHandle is None
+
+
+def test_rearm_disable_and_reentrant_listener(timer):
+    '''
+    Replacing or disabling a packet-idle timer must cancel its old
+    handle. Expiry must also finish teardown before timeoutEvent
+    listeners rearm it, otherwise the new timer can be lost.
+    Advance cancelled deadlines deterministically, then rearm from
+    a real eventkit listener and deliver another packet. Both arms
+    must emit once, with packet timestamps intact during emission.
+
+    '''
+    loop, wrapper, events = timer
+    wrapper.setTimeout(2)
+    wrapper.tcpDataArrived()
+    old = wrapper._timeoutHandle
+    wrapper.setTimeout(4)
+    assert old.cancelled()
+    replacement = wrapper._timeoutHandle
+    wrapper.tcpDataArrived()
+    wrapper.setTimeout(0)
+    assert replacement.cancelled()
+    assert wrapper._timeoutHandle is None
+    wrapper.tcpDataArrived()
+    loop.advance(10)
+    assert events == []
+
+    observed = []
+
+    def rearm(diff):
+        observed.append((wrapper._timeout, wrapper._timeoutHandle))
+        if len(observed) == 1:
+            wrapper.setTimeout(3)
+            wrapper.tcpDataArrived()
+
+    wrapper.ib.timeoutEvent.connect(rearm)
+    wrapper.setTimeout(2)
+    wrapper.tcpDataArrived()
+    first_packet = wrapper.lastTime
+    loop.advance(2)
+    second_packet = wrapper.lastTime
+    assert wrapper._timeout == 3
+    assert wrapper._timeoutHandle is not None
+    loop.advance(3)
+    loop.advance(10)
+    assert events == [(2.0, first_packet), (3.0, second_packet)]
+    assert observed == [(0, None), (0, None)]
+    assert wrapper.lastTime == second_packet

tests/test_sampling_tick_times.py

@@ -0,0 +1,242 @@
+# piker: trading gear for hackers Copyright (C) 2018-present  Tyler
+# Goodlet (in stewardship of pikers)
+
+# This program is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public License
+# as published by the Free Software Foundation, either version 3 of
+# the License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public
+# License along with this program.  If not, see
+# <https://www.gnu.org/licenses/>.
+
+'''
+Live sampler checks for timestamped startup trade backlogs.
+
+'''
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import trio
+
+from piker.data._sampling import (
+    sample_and_broadcast,
+    trade_time_s,
+)
+from piker.data import _sampling
+
+
+@pytest.mark.parametrize(
+    ('raw', 'expected'),
+    [
+        (1_790_360_484, 1_790_360_484.),  # seconds
+        (1_790_360_484_000, 1_790_360_484.),  # milliseconds
+        (1_790_360_484_000_000, 1_790_360_484.),
+        (1_790_360_484_000_000_000, 1_790_360_484.),
+        (None, None),  # provider without a tick timestamp
+    ],
+)
+def test_trade_time_s_accepts_common_epoch_units(
+    raw: int|None,
+    expected: float|None,
+) -> None:
+    assert trade_time_s({'time': raw}) == expected
+
+
+@pytest.mark.parametrize(
+    ('bar_time', 'current_time_ns'),
+    [
+        (
+            1_790_360_484.,
+            1_790_360_484_250_000_000,
+        ),
+        (
+            1_790_360_485.,
+            1_790_360_484_750_000_000,
+        ),
+    ],
+)
+def test_queued_trade_is_not_written_to_a_later_bar(
+    bar_time: float,
+    current_time_ns: int,
+) -> None:
+    '''
+    A 13:21:19 IB trade queued until the 13:21:24 1s bar must not
+    add its price or volume to either live SHM tail. A current tick
+    in the same input channel still updates both bars.
+
+    '''
+    dtype = [
+        ('time', 'f8'),
+        ('open', 'f8'),
+        ('high', 'f8'),
+        ('low', 'f8'),
+        ('close', 'f8'),
+        ('volume', 'f8'),
+    ]
+    rt = np.array(
+        [(bar_time, 30_918., 30_918.,
+          30_918., 30_918., 19.)],
+        dtype=dtype,
+    )
+    hist = np.array(
+        [(1_790_360_460., 30_918., 30_918.,
+          30_918., 30_918., 248.)],
+        dtype=dtype,
+    )
+    rt_shm = SimpleNamespace(array=rt)
+    hist_shm = SimpleNamespace(array=hist)
+    bus = SimpleNamespace(get_subs=lambda key: set())
+
+    async def replay() -> None:
+        send, receive = trio.open_memory_channel(4)
+        async with send:
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': 1_790_360_479_217_767_927,
+                        'price': 30_923.,
+                        'size': 12_013.,
+                    }],
+                    'volume': 2_222_721.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': current_time_ns,
+                        'price': 30_918.25,
+                        'size': 2.,
+                    }],
+                    'volume': 2_222_723.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': int(
+                            (bar_time + 2.) * 1_000_000_000
+                        ),
+                        'price': 30_925.,
+                        'size': 100.,
+                    }],
+                    'volume': 2_222_823.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'price': 30_918.5,
+                        'size': 1.,
+                    }],
+                    'volume': 2_222_824.,
+                },
+            })
+        async with receive:
+            await sample_and_broadcast(
+                bus,
+                rt_shm,
+                hist_shm,
+                receive,
+                'ib',
+            )
+
+    trio.run(replay)
+    assert float(rt['high'][-1]) == 30_918.5
+    assert float(rt['volume'][-1]) == 22.
+    assert float(hist['high'][-1]) == 30_918.5
+    assert float(hist['volume'][-1]) == 251.
+
+
+def test_next_second_trade_waits_for_sampler_step(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A trade for the next 1s slot must not update the older SHM bar.
+
+    The independent `Sampler.increment_ohlc_buffer()` can lag the
+    tick stream. Waiting avoids a misdated price but delays chart
+    updates, so a warning must fire before the slot advances. Hold
+    the old slot until that warning, then advance it and verify the
+    trade reaches both SHM tails without being dropped.
+
+    '''
+    dtype = [
+        ('time', 'f8'),
+        ('open', 'f8'),
+        ('high', 'f8'),
+        ('low', 'f8'),
+        ('close', 'f8'),
+        ('volume', 'f8'),
+    ]
+    rt = np.array(
+        [(1_790_360_484., 30_918., 30_918.,
+          30_918., 30_918., 0.)],
+        dtype=dtype,
+    )
+    hist = rt.copy()
+    rt_shm = SimpleNamespace(array=rt)
+    hist_shm = SimpleNamespace(array=hist)
+    bus = SimpleNamespace(get_subs=lambda key: set())
+    warnings: list[str] = []
+    wait_started = trio.Event()
+
+    def record_warning(message: str) -> None:
+        warnings.append(message)
+        wait_started.set()
+
+    monkeypatch.setattr(
+        _sampling.log,
+        'warning',
+        record_warning,
+    )
+
+    async def replay() -> None:
+        send, receive = trio.open_memory_channel(1)
+        async with send:
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': 1_790_360_485_250_000_000,
+                        'price': 30_919.,
+                        'size': 2.,
+                    }],
+                    'volume': 2_222_723.,
+                },
+            })
+
+        async def step() -> None:
+            with trio.fail_after(2):
+                await wait_started.wait()
+            rt['time'][-1] = 1_790_360_485.
+
+        async with (
+            receive,
+            trio.open_nursery() as nursery,
+        ):
+            nursery.start_soon(step)
+            await sample_and_broadcast(
+                bus,
+                rt_shm,
+                hist_shm,
+                receive,
+                'ib',
+            )
+
+    trio.run(replay)
+    assert float(rt['high'][-1]) == 30_919.
+    assert float(rt['volume'][-1]) == 2.
+    assert float(hist['volume'][-1]) == 2.
+    assert len(warnings) == 1
+    assert 'DELAYING LIVE TRADE SHM UPDATE' in warnings[0]