Fix FSP history and sample synchronization

Bootstrap FSP output and timestamps from one source snapshot,
validate the history schema, and compare absolute SHM bounds.
Subscribe before bootstrap, replay history revisions, filter
foreign sampler notices, and append each destination step once.
Close replaced generators before signaling completion.

Use declared `Flume` sample periods so market-closure gaps cannot
be mistaken for the realtime cadence. Cover bounds, snapshot
isolation, backfill routing, sample steps, and targeted redraws.

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

piker/data/flows.py

@@ -24,7 +24,6 @@ real-time data processing data-structures.
 from __future__ import annotations
 
 import tractor
-import pendulum
 import numpy as np
 
 from piker.types import Struct
@@ -73,6 +72,8 @@ class Flume(Struct):
     izero_hist: int = 0
     izero_rt: int = 0
     throttle_rate: int | None = None
+    rt_sample_period_s: float = 1.
+    hist_sample_period_s: float = 60.
 
     @property
     def rt_shm(self) -> ShmArray:
@@ -115,17 +116,9 @@ class Flume(Struct):
         period and ratio between them.
 
         '''
-        times: np.ndarray = self.hist_shm.array['time']
+        rt_step_size_s: float = self.rt_sample_period_s
-        end: float | int = pendulum.from_timestamp(times[-1])
+        hist_step_size_s: float = self.hist_sample_period_s
-        start: float | int = pendulum.from_timestamp(times[times != times[-1]][-1])
+        ratio: float = hist_step_size_s / rt_step_size_s
-        hist_step_size_s: float = (end - start).seconds
-
-        times = self.rt_shm.array['time']
-        end = pendulum.from_timestamp(times[-1])
-        start = pendulum.from_timestamp(times[times != times[-1]][-1])
-        rt_step_size_s = (end - start).seconds
-
-        ratio = hist_step_size_s / rt_step_size_s
         return (
             rt_step_size_s,
             hist_step_size_s,

piker/fsp/_engine.py

@@ -22,7 +22,10 @@ from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from functools import partial
 from typing import (
+    Any,
     AsyncIterator,
+    Awaitable,
+    cast,
     Callable,
     TYPE_CHECKING,
 )
@@ -39,15 +42,20 @@ from ..log import (
     get_console_log,
 )
 from .. import data
+from ..data.ticktools import FeedQuote
 from ..data.flows import Flume
+from ..data._sharedmem import NDTokenMsg
 from tractor.ipc._shm import ShmArray
 from ..data._sampling import (
-    _default_delay_s,
     open_sample_stream,
 )
 from ..accounting import MktPair
 from ._api import (
     Fsp,
+    FspFunc,
+    FspStream,
+    FspYield,
+    FspHistory,
     _load_builtins,
     NDToken,
 )
@@ -58,13 +66,70 @@ if TYPE_CHECKING:
 
 log = get_logger(__name__)
 
+type QuoteFrame = dict[str, FeedQuote]
+type BackfillNotice = tuple[str, float]|list[str|float]
+type SampleValue = int|float|BackfillNotice
+type SampleMsg = dict[str, SampleValue]
+type EdgeFunc = FspFunc
+
+
+def _is_relevant_sample_msg(
+    msg: SampleMsg,
+    fqme: str,
+    period_s: float,
+
+) -> bool:
+    backfill: SampleValue|None = msg.get('backfilling')
+    if backfill is None:
+        return True
+
+    if (
+        not isinstance(backfill, (list, tuple))
+        or
+        len(backfill) != 2
+    ):
+        return False
+
+    backfill_fqme: str | float
+    timeframe: str | float
+    backfill_fqme, timeframe = backfill
+    return (
+        backfill_fqme == fqme
+        and
+        timeframe == period_s
+    )
+
+
+def _should_advance_dst(
+    msg: SampleMsg,
+    step_diff: int,
+
+) -> bool:
+    return (
+        'backfilling' not in msg
+        and
+        step_diff == 1
+    )
+
+
+def _needs_history_resync(
+    msg: SampleMsg,
+    synced: bool,
+
+) -> bool:
+    return (
+        'backfilling' in msg
+        or
+        not synced
+    )
+
 
 async def filter_quotes_by_sym(
 
     sym: str,
-    quote_stream: tractor.MsgStream,
+    quote_stream: AsyncIterator[QuoteFrame],
 
-) -> AsyncIterator[dict]:
+) -> AsyncIterator[FeedQuote]:
     '''
     Filter quote stream by target symbol.
 
@@ -75,10 +140,91 @@ async def filter_quotes_by_sym(
     yield {}
 
     async for quotes in quote_stream:
-        quote = quotes.get(sym)
+        quote: FeedQuote|None = quotes.get(sym)
         if quote:
             yield quote
 
+
+class _HistoryShm:
+    '''
+    Present an immutable history view before switching to live SHM.
+
+    '''
+    def __init__(self, shm: ShmArray) -> None:
+        self._shm: ShmArray = shm
+        self.first: int = shm._first.value
+        self.last_index: int = shm._last.value
+        self.history: np.ndarray = shm._array[
+            self.first:self.last_index
+        ].copy()
+        self.live: bool = False
+
+    @property
+    def array(self) -> np.ndarray:
+        if self.live:
+            return self._shm.array
+        return self.history
+
+    @property
+    def index(self) -> int:
+        if self.live:
+            return self._shm.index
+        return self.last_index % self._shm._len
+
+    def last(self, length: int = 1) -> np.ndarray:
+        return self.array[-length:]
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._shm, name)
+
+
+async def _open_history_snapshot(
+    edge_func: EdgeFunc,
+    quote_stream: AsyncIterator[QuoteFrame],
+    fqme: str,
+    src_shm: ShmArray,
+
+) -> tuple[
+    FspStream,
+    FspHistory,
+    tuple[int, int],
+    np.ndarray,
+]:
+    '''
+    Start an FSP against an immutable source-history snapshot.
+
+    '''
+    history_shm: _HistoryShm = _HistoryShm(src_shm)
+    out_stream: FspStream = edge_func(
+        filter_quotes_by_sym(fqme, quote_stream),
+        cast(ShmArray, history_shm),
+    )
+    try:
+        first_yield: FspYield = await anext(out_stream)
+    except BaseException:
+        with trio.CancelScope(shield=True):
+            await out_stream.aclose()
+        raise
+
+    if isinstance(first_yield, tuple):
+        await out_stream.aclose()
+        raise TypeError(
+            f'FSP `{edge_func.__name__}` yielded a realtime update '
+            f'before its historical output'
+        )
+
+    history_output: FspHistory = first_yield
+    history_shm.live = True
+    return (
+        out_stream,
+        history_output,
+        (
+            history_shm.first,
+            history_shm.last_index,
+        ),
+        history_shm.history,
+    )
+
 # TODO: unifying the abstractions in this FSP subsys/layer:
 # -[ ] move the `.data.flows.Flume` type into this
 #   module/subsys/pkg?
@@ -124,24 +270,33 @@ class Cascade(Struct):
     fsp: Fsp  # UI-side middleware ctl API
 
     # filled during cascade/.bind_func() (fsp_compute) init phases
-    bind_func: Callable | None = None
+    bind_func: Callable[..., Awaitable[None]]|None = None
-    complete: trio.Event | None = None
+    complete: trio.Event|None = None
-    cs: trio.CancelScope | None = None
+    cs: trio.CancelScope|None = None
-    client_stream: tractor.MsgStream | None = None
+    client_stream: tractor.MsgStream|None = None
 
     async def resync(self) -> int:
         # TODO: adopt an incremental update engine/approach
         # where possible here eventually!
         log.info(f're-syncing fsp {self.fsp.name} to source')
-        self.cs.cancel()
+        cs: trio.CancelScope|None = self.cs
-        await self.complete.wait()
+        complete: trio.Event|None = self.complete
-        index: int = await self.tn.start(self.bind_func)
+        bind_func = self.bind_func
+        client_stream = self.client_stream
+        assert cs is not None
+        assert complete is not None
+        assert bind_func is not None
+        assert client_stream is not None
+
+        cs.cancel()
+        await complete.wait()
+        index: int = await self.tn.start(bind_func)
 
         # always trigger UI refresh after history update,
         # see ``piker.ui._fsp.FspAdmin.open_chain()`` and
         # ``piker.ui._display.trigger_update()``.
         dst_shm: ShmArray = self.dst.rt_shm
-        await self.client_stream.send({
+        await client_stream.send({
             'fsp_update': {
                 'key': dst_shm.token,
                 'first': dst_shm._first.value,
@@ -158,27 +313,48 @@ class Cascade(Struct):
         '''
         src_shm: ShmArray = self.src.rt_shm
         dst_shm: ShmArray = self.dst.rt_shm
-        step_diff = src_shm.index - dst_shm.index
+        src_first = src_shm._first.value
-        len_diff = abs(len(src_shm.array) - len(dst_shm.array))
+        src_last = src_shm._last.value
-        synced: bool = not (
+        dst_first = dst_shm._first.value
-            # the source is likely backfilling and we must
+        dst_last = dst_shm._last.value
-            # sync history calculations
-            len_diff > 2
 
-            # we aren't step synced to the source and may be
+        # Compare absolute bounds instead of ``ShmArray.index``, which
-            # leading/lagging by a step
+        # wraps at the allocation size. A one-step source lead is the
-            or step_diff > 1
+        # normal state before the destination sample is appended.
-            or step_diff < 0
+        step_diff = src_last - dst_last
+        len_diff = abs(
+            (src_last - src_first)
+            - (dst_last - dst_first)
+        )
+        synced: bool = (
+            src_first == dst_first
+            and
+            step_diff in (0, 1)
         )
         if not synced:
             fsp: Fsp = self.fsp
+            report = (
+                f'ns-path: {fsp.ns_path!r}\n'
+                f'shm-token: {src_shm.token}\n'
+                f'src-bounds: {(src_first, src_last)}\n'
+                f'dst-bounds: {(dst_first, dst_last)}\n'
+                f'step_diff: {step_diff}\n'
+                f'len_diff: {len_diff}\n'
+            )
+            if (
+                src_first < dst_first
+                and
+                src_last == dst_last
+            ):
+                log.info(
+                    f'FSP source history grew:\n'
+                    f'{report}'
+                )
+            else:
                 log.warning(
                     f'***DESYNCED fsp***\n'
                     f'------------------\n'
-                f'ns-path: {fsp.ns_path!r}\n'
+                    f'{report}'
-                f'shm-token: {src_shm.token}\n'
-                f'step_diff: {step_diff}\n'
-                f'len_diff: {len_diff}\n'
                 )
         return (
             synced,
@@ -187,17 +363,20 @@ class Cascade(Struct):
         )
 
     async def poll_and_sync_to_step(self) -> int:
-        synced, step_diff, _ = self.is_synced()
+        while True:
-        while not synced:
             await self.resync()
-            synced, step_diff, _ = self.is_synced()
+            (
-
+                synced,
+                step_diff,
+                _,
+            ) = self.is_synced()
+            if synced:
                 return step_diff
 
     @acm
     async def open_edge(
         self,
-        bind_func: Callable,
+        bind_func: Callable[..., Awaitable[None]],
     ) -> int:
         self.bind_func = bind_func
         index = await self.tn.start(bind_func)
@@ -206,14 +385,123 @@ class Cascade(Struct):
         # -[ ] dynamic reconnection after update?
 
 
+async def _close_fsp_stream(out_stream: FspStream) -> None:
+    with trio.CancelScope(shield=True):
+        await out_stream.aclose()
+
+
+def _write_history_output(
+    dst_shm: ShmArray,
+    history_output: FspHistory,
+    src_bounds: tuple[int, int],
+    src_history: np.ndarray,
+    func_name: str,
+    profiler: Profiler,
+
+) -> int:
+    src_first, src_last = src_bounds
+    fields = dst_shm.array.dtype.fields
+    if fields is None:
+        raise TypeError('FSP destination SHM must be a structured array')
+    fields = fields.copy()
+    fields.pop('index')
+    fields.pop('time')
+    history_by_field: np.ndarray|None = None
+
+    if len(fields) > 1:
+        if not isinstance(history_output, dict):
+            raise ValueError(
+                f'`{func_name}` is a multi-output FSP and should '
+                f'yield a `dict[str, np.ndarray]` for history'
+            )
+
+        output_fields: list[str] = [
+            key
+            for key in fields
+            if key in history_output
+        ]
+        if not output_fields:
+            raise ValueError(
+                f'`{func_name}` produced no declared history fields'
+            )
+
+        for key in output_fields:
+            output: np.ndarray|None = history_output[key]
+            if history_by_field is None:
+                length: int = (
+                    len(src_history)
+                    if output is None
+                    else len(output)
+                )
+
+                # using the first output, determine the length of the
+                # struct-array that will be pushed to shm.
+                history_by_field = np.zeros(
+                    length,
+                    dtype=dst_shm.array.dtype,
+                )
+
+            if output is not None:
+                history_by_field[key] = output
+
+    else:
+        if not isinstance(history_output, np.ndarray):
+            raise ValueError(
+                f'`{func_name}` is a single output FSP and should '
+                f'yield an `np.ndarray` for history'
+            )
+        history_by_field = np.zeros(
+            len(history_output),
+            dtype=dst_shm.array.dtype,
+        )
+        history_by_field[func_name] = history_output
+
+    if history_by_field is None:
+        raise ValueError(f'`{func_name}` produced no history fields')
+    if len(history_by_field) != len(src_history):
+        raise ValueError(
+            f'`{func_name}` history length '
+            f'{len(history_by_field)} does not match source length '
+            f'{len(src_history)}'
+        )
+
+    history_by_field['time'] = src_history['time']
+
+    # TODO: XXX:
+    # THERE'S A BIG BUG HERE WITH THE `index` field since we're
+    # prepending a copy of the first value a few times to make
+    # sub-curves align with the parent bar chart.
+    # This likely needs to be fixed either by,
+    # - manually assigning the index and historical data
+    #   seperately to the shm array (i.e. not using .push())
+    # - developing some system on top of the shared mem array that
+    #   is `index` aware such that historical data can be indexed
+    #   relative to the true first datum? Not sure if this is sane
+    #   for incremental compuations.
+    first: int = src_first
+    dst_shm._first.value = first
+
+    # TODO: can we use this `start` flag instead of the manual
+    # setting above?
+    index: int = dst_shm.push(
+        history_by_field,
+        start=first,
+    )
+
+    profiler(f'{func_name} pushed history')
+    profiler.finish()
+    assert index == src_last
+    return index
+
+
 async def connect_streams(
     casc: Cascade,
     mkt: MktPair,
-    quote_stream: trio.abc.ReceiveChannel,
+    quote_stream: AsyncIterator[QuoteFrame],
     src: Flume,
     dst: Flume,
 
-    edge_func: Callable,
+    edge_func: EdgeFunc,
 
     # attach_stream: bool = False,
     task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
@@ -239,119 +527,38 @@ async def connect_streams(
     # fqme: str = mkt.fqme
     fqme: str = src.mkt.fqme
 
-    # TODO: dynamic introspection of what the underlying (vertex)
-    # function actually requires from input node (flumes) then
-    # deliver those inputs as part of a graph "compilation" step?
-    out_stream = edge_func(
-
-        # TODO: do we even need this if we do the feed api right?
-        # shouldn't a local stream do this before we get a handle
-        # to the async iterable? it's that or we do some kinda
-        # async itertools style?
-        filter_quotes_by_sym(fqme, quote_stream),
-
-        # XXX: currently the ``ohlcv`` arg, but we should allow
-        # (dynamic) requests for src flume (node) streams?
-        src.rt_shm,
-    )
-
     # HISTORY COMPUTE PHASE
     # conduct a single iteration of fsp with historical bars input
     # and get historical output.
-    history_output: (
+    history_output: FspHistory
-        dict[str, np.ndarray]  # multi-output case
+    src_shm: ShmArray = src.rt_shm
-        | np.ndarray,  # single output case
+    (
+        out_stream,
+        history_output,
+        src_bounds,
+        src_history,
+    ) = await _open_history_snapshot(
+        edge_func,
+        quote_stream,
+        fqme,
+        src_shm,
     )
-    history_output = await anext(out_stream)
+    func_name: str = edge_func.__name__
-
-    func_name = edge_func.__name__
     profiler(f'{func_name} generated history')
 
-    # build struct array with an 'index' field to push as history
-
-    # TODO: push using a[['f0', 'f1', .., 'fn']] = .. syntax no?
-    # if the output array is multi-field then push
-    # each respective field.
     dst_shm: ShmArray = dst.rt_shm
-    fields = getattr(dst_shm.array.dtype, 'fields', None).copy()
+    try:
-    fields.pop('index')
+        index: int = _write_history_output(
-    history_by_field: np.ndarray | None = None
+            dst_shm,
-    src_shm: ShmArray = src.rt_shm
+            history_output,
-    src_time = src_shm.array['time']
+            src_bounds,
-
+            src_history,
-    if (
+            func_name,
-        fields and
+            profiler,
-        len(fields) > 1
-    ):
-        if not isinstance(history_output, dict):
-            raise ValueError(
-                f'`{func_name}` is a multi-output FSP and should yield a '
-                '`dict[str, np.ndarray]` for history'
         )
-
+    except BaseException:
-        for key in fields.keys():
+        await _close_fsp_stream(out_stream)
-            if key in history_output:
+        raise
-                output = history_output[key]
-
-                if history_by_field is None:
-
-                    if output is None:
-                        length = len(src_shm.array)
-                    else:
-                        length = len(output)
-
-                    # using the first output, determine
-                    # the length of the struct-array that
-                    # will be pushed to shm.
-                    history_by_field = np.zeros(
-                        length,
-                        dtype=dst_shm.array.dtype
-                    )
-
-                if output is None:
-                    continue
-
-                history_by_field[key] = output
-
-    # single-key output stream
-    else:
-        if not isinstance(history_output, np.ndarray):
-            raise ValueError(
-                f'`{func_name}` is a single output FSP and should yield an '
-                '`np.ndarray` for history'
-            )
-        history_by_field = np.zeros(
-            len(history_output),
-            dtype=dst_shm.array.dtype
-        )
-        history_by_field[func_name] = history_output
-
-    history_by_field['time'] = src_time[-len(history_by_field):]
-
-    history_output['time'] = src_shm.array['time']
-
-    # TODO: XXX:
-    # THERE'S A BIG BUG HERE WITH THE `index` field since we're
-    # prepending a copy of the first value a few times to make
-    # sub-curves align with the parent bar chart.
-    # This likely needs to be fixed either by,
-    # - manually assigning the index and historical data
-    #   seperately to the shm array (i.e. not using .push())
-    # - developing some system on top of the shared mem array that
-    #   is `index` aware such that historical data can be indexed
-    #   relative to the true first datum? Not sure if this is sane
-    #   for incremental compuations.
-    first = dst_shm._first.value = src_shm._first.value
-
-    # TODO: can we use this `start` flag instead of the manual
-    # setting above?
-    index = dst_shm.push(
-        history_by_field,
-        start=first,
-    )
-
-    profiler(f'{func_name} pushed history')
-    profiler.finish()
 
     # setup a respawn handle
     with trio.CancelScope() as cs:
@@ -398,6 +605,9 @@ async def connect_streams(
                 # if hz > 60:
                 #     log.info(f'FSP quote too fast: {hz}')
                 # last = time.time()
+        finally:
+            try:
+                await _close_fsp_stream(out_stream)
             finally:
                 casc.complete.set()
 
@@ -410,11 +620,13 @@ async def cascade(
     fqme: str,
 
     # flume pair cascaded using an "edge function"
-    src_flume_addr: dict,
+    src_flume_addr: dict[str, Any],
-    dst_flume_addr: dict,
+    dst_flume_addr: dict[str, Any],
     ns_path: NamespacePath,
 
-    shm_registry: dict[str, NDToken],
+    shm_registry: list[
+        tuple[NDTokenMsg, str, NDTokenMsg]
+    ],
 
     zero_on_step: bool = False,
     loglevel: str|None = None,
@@ -452,7 +664,7 @@ async def cascade(
     # src: ShmArray = attach_shm_array(token=src_shm_token)
     # dst: ShmArray = attach_shm_array(readonly=False, token=dst_shm_token)
 
-    reg = _load_builtins()
+    reg: dict[NamespacePath, Fsp] = _load_builtins()
     lines = '\n'.join([f'{key.rpartition(":")[2]} => {key}' for key in reg])
     log.info(
         f'Registered FSP set:\n{lines}'
@@ -465,18 +677,21 @@ async def cascade(
     # not sure how else to do it.
     for (token, fsp_name, dst_token) in shm_registry:
         Fsp._flow_registry[(
-            NDToken.from_msg(token),
+            NDToken.from_msg(dict(token)),
             fsp_name,
-        )] = NDToken.from_msg(dst_token), None
+        )] = NDToken.from_msg(dict(dst_token)), None
 
-    fsp: Fsp = reg.get(
+    fsp: Fsp|None = reg.get(
         NamespacePath(ns_path)
     )
-    func: Callable = fsp.func
+    if (
-
+        fsp is None
-    if not func:
+        or
+        not fsp.func
+    ):
         # TODO: assume it's a func target path
         raise ValueError(f'Unknown fsp target: {ns_path}')
+    func: FspFunc = fsp.func
 
     _fqme: str = src.mkt.fqme
     assert _fqme == fqme
@@ -521,7 +736,7 @@ async def cascade(
             # the target task is spawned implicitly and then the event is
             # set via some higher level api? At that poing we might as well
             # be writing a one-cancels-one nursery though right?
-            casc = Cascade(
+            casc: Cascade = Cascade(
                 src,
                 dst,
                 tn,
@@ -529,7 +744,7 @@ async def cascade(
             )
 
             # TODO: this seems like it should be wrapped somewhere?
-            fsp_target = partial(
+            fsp_target: Callable[..., Awaitable[None]] = partial(
                 connect_streams,
                 casc=casc,
                 mkt=mkt,
@@ -543,9 +758,20 @@ async def cascade(
                 # and renders dst flume output(s)
                 edge_func=func
             )
-            async with casc.open_edge(
+
+            # Subscribe before history bootstrap so in-place repairs
+            # with unchanged SHM bounds can not lose invalidations.
+            period_s: float = src.rt_sample_period_s
+
+            async with (
+                open_sample_stream(
+                    period_s=period_s,
+                    loglevel=loglevel,
+                ) as istream,
+                casc.open_edge(
                     bind_func=fsp_target,
-            ) as index:
+                ) as index,
+            ):
                 # casc.bind_func = fsp_target
                 # index = await tn.start(fsp_target)
                 dst_shm: ShmArray = dst.rt_shm
@@ -567,45 +793,51 @@ async def cascade(
                 async with ctx.open_stream() as client_stream:
                     casc.client_stream: tractor.MsgStream = client_stream
 
-                    s, step, ld = casc.is_synced()
-
-                    # detect sample period step for subscription to increment
-                    # signal
-                    times = src.rt_shm.array['time']
-                    if len(times) > 1:
-                        last_ts = times[-1]
-                        delay_s: float = float(last_ts - times[times != last_ts][-1])
-                    else:
-                        # our default "HFT" sample rate.
-                        delay_s: float = _default_delay_s
-
-                    # sub and increment the underlying shared memory buffer
-                    # on every step msg received from the global `samplerd`
-                    # service.
-                    async with open_sample_stream(
-                        period_s=float(delay_s),
-                        loglevel=loglevel,
-                    ) as istream:
-
                     profiler(f'{func_name}: sample stream up')
                     profiler.finish()
 
-                        async for i in istream:
+                    async for sample_msg in istream:
-                            # print(f'FSP incrementing {i}')
+                        # print(f'FSP incrementing {sample_msg}')
+                        msg: SampleMsg = cast(
+                            SampleMsg,
+                            sample_msg,
+                        )
 
-                            # respawn the compute task if the source
+                        # ``samplerd`` broadcasts history events for
-                            # array has been updated such that we compute
+                        # every feed and period. Only this cascade's
-                            # new history from the (prepended) source.
+                        # source can invalidate its destination.
-                            synced, step_diff, _ = casc.is_synced()
+                        if not _is_relevant_sample_msg(
-                            if not synced:
+                            msg,
-                                step_diff: int = await casc.poll_and_sync_to_step()
+                            fqme,
-
+                            period_s,
-                                # skip adding a last bar since we should already
+                        ):
-                                # be step alinged
+                            continue
-                                if step_diff == 0:
+
+                        # Respawn the compute task after source history
+                        # revisions or an actual bounds desync.
+                        (
+                            synced,
+                            step_diff,
+                            _,
+                        ) = casc.is_synced()
+                        if _needs_history_resync(
+                            msg,
+                            synced,
+                        ):
+                            step_diff = (
+                                await casc.poll_and_sync_to_step()
+                            )
+
+                        # Backfill broadcasts only announce source
+                        # history revisions. They never represent a
+                        # new sample row. Duplicate sample wakeups are
+                        # likewise non-advancing when already aligned.
+                        if not _should_advance_dst(
+                            msg,
+                            step_diff,
+                        ):
                             continue
 
-                            # read out last shm row, copy and write new row
                         array = dst_shm.array
 
                         # some metrics like vlm should be reset
@@ -619,8 +851,12 @@ async def cascade(
 
                         # sync with source buffer's time step
                         src_l2 = src_shm.array[-2:]
-                            src_li, src_lt = src_l2[-1][['index', 'time']]
+                        src_li, src_lt = src_l2[-1][
-                            src_2li, src_2lt = src_l2[-2][['index', 'time']]
+                            ['index', 'time']
+                        ]
+                        src_2li, src_2lt = src_l2[-2][
+                            ['index', 'time']
+                        ]
                         dst_shm._array['time'][src_li] = src_lt
                         dst_shm._array['time'][src_2li] = src_2lt
 

tests/test_fsp_sync.py

@@ -3,12 +3,22 @@ FSP history synchronization regressions.
 
 '''
 from collections.abc import AsyncIterator
+from types import SimpleNamespace
 from typing import cast
 
 import numpy as np
 import pytest
 import trio
 
+from piker.fsp._engine import (
+    _is_relevant_sample_msg,
+    _needs_history_resync,
+    _open_history_snapshot,
+    _should_advance_dst,
+    Cascade,
+    EdgeFunc,
+    QuoteFrame,
+)
 from piker.fsp._momo import (
     rsi,
     wma,
@@ -16,14 +26,56 @@ from piker.fsp._momo import (
 from piker.fsp._volume import (
     tina_vwap,
 )
+from piker.accounting import MktPair
+from piker.data.flows import Flume
 from piker.data._sharedmem import NDTokenMsg
 from piker.data.ticktools import FeedQuote
 from piker.fsp._api import Fsp
+from piker.ui._chart import LinkedSplits
+from piker.ui._fsp import update_fsp_vizs
 from tractor.ipc._shm import (
     NDToken,
     ShmArray,
 )
 
+type SampleMsg = dict[
+    str,
+    int|float|tuple[str, float],
+]
+
+
+def test_sample_period_ignores_market_closure_gap() -> None:
+    '''
+    Keep FSP cascades subscribed to their regular sampler period.
+
+    Starting a cascade immediately after a market closure left the
+    source SHM tail with a large gap between its last two distinct
+    timestamps. Using only that pair subscribed the cascade to the gap
+    duration instead of the regular one-second stream. Realtime FSP
+    writes then repeatedly replaced one row without advancing its SHM
+    bound. Give both source SHMs closure-gap tail deltas and prove
+    `Flume.get_ds_info()` returns its declared one- and 60-second
+    periods instead of deriving cadence from those rows.
+
+    '''
+    rt_shm = OhlcvShm(length=4)
+    hist_shm = OhlcvShm(length=4)
+    rt_shm._array['time'][:4] = [100, 101, 102, 3600]
+    hist_shm._array['time'][:4] = [100, 160, 220, 3600]
+    flume = Flume(
+        mkt=cast(MktPair, SimpleNamespace()),
+        first_quote={},
+        _rt_shm_token=rt_shm._token,
+        _hist_shm_token=hist_shm._token,
+    )
+    flume._rt_shm = cast(ShmArray, rt_shm)
+    flume._hist_shm = cast(ShmArray, hist_shm)
+
+    assert np.diff(rt_shm.array['time'])[-1] == 3498
+    assert np.diff(hist_shm.array['time'])[-1] == 3380
+    assert flume.get_ds_info() == (1, 60, 60)
+
+
 class Value:
     def __init__(self, value: int) -> None:
         self.value: int = value
@@ -96,6 +148,236 @@ class OhlcvShm(Shm):
         )
 
 
+class FspShm(Shm):
+    def __init__(
+        self,
+        first: int = 0,
+        last: int = 4,
+        token: str = 'fsp',
+
+    ) -> None:
+        dtype = np.dtype([
+            ('index', '<i8'),
+            ('time', '<f8'),
+            ('flow', '<f8'),
+            ('dark_flow', '<f8'),
+        ])
+        size: int = max(last + 2, 4096)
+        self._first = Value(first)
+        self._last = Value(last)
+        self._array = np.zeros(size, dtype=dtype)
+        self._array['index'] = np.arange(size)
+        self._array['time'] = np.arange(size)
+        self._array['flow'] = np.arange(size)
+        self._len = len(self._array)
+        self._token = NDToken(
+            shm_name=token,
+            shm_first_index_name=f'{token}_first',
+            shm_last_index_name=f'{token}_last',
+            dtype_descr=tuple(dtype.descr),
+            size=self._len,
+        )
+
+
+def mk_cascade(
+    src_bounds: tuple[int, int],
+    dst_bounds: tuple[int, int],
+
+) -> Cascade:
+    fsp: Fsp = cast(
+        Fsp,
+        SimpleNamespace(
+            name='test_fsp',
+            ns_path='tests:test_fsp',
+        ),
+    )
+    src: Flume = cast(
+        Flume,
+        SimpleNamespace(
+            rt_shm=Shm(*src_bounds, token='src'),
+        ),
+    )
+    dst: Flume = cast(
+        Flume,
+        SimpleNamespace(
+            rt_shm=Shm(*dst_bounds, token='dst'),
+        ),
+    )
+    nursery: trio.Nursery = cast(trio.Nursery, None)
+    return Cascade(src, dst, nursery, fsp)
+
+
+@pytest.mark.parametrize(
+    'src_bounds,dst_bounds,expected',
+    [
+        ((10, 20), (10, 20), (True, 0, 0)),
+        ((10, 21), (10, 20), (True, 1, 1)),
+        ((0, 3000), (2000, 3000), (False, 0, 2000)),
+        ((9, 20), (10, 20), (False, 0, 1)),
+        ((8, 20), (10, 20), (False, 0, 2)),
+        ((10, 20), (9, 20), (False, 0, 1)),
+        ((11, 21), (10, 20), (False, 1, 0)),
+        ((10, 22), (10, 20), (False, 2, 2)),
+        ((10, 19), (10, 20), (False, -1, 1)),
+    ],
+)
+def test_cascade_sync_uses_absolute_bounds(
+    src_bounds: tuple[int, int],
+    dst_bounds: tuple[int, int],
+    expected: tuple[bool, int, int],
+
+) -> None:
+    '''
+    Detect every history-bound mismatch without modulo-index skew.
+
+    Source prepends move only its first bound, while a normal realtime
+    step moves only its last bound. The former requires a historical
+    recompute even for one or two rows; the latter permits exactly one
+    destination append. This matrix also shifts equal-length bounds and
+    puts the destination ahead to prove neither state is accepted.
+
+    '''
+    cascade = mk_cascade(src_bounds, dst_bounds)
+
+    assert cascade.is_synced() == expected
+
+
+@pytest.mark.parametrize(
+    'msg,expected',
+    [
+        ({'index': 1}, True),
+        ({'backfilling': ('tsla.nasdaq.ib', 1)}, True),
+        ({'backfilling': ('btcusdt.binance', 1)}, False),
+        ({'backfilling': ('tsla.nasdaq.ib', 60)}, False),
+    ],
+)
+def test_cascade_filters_foreign_backfill_events(
+    msg: SampleMsg,
+    expected: bool,
+
+) -> None:
+    '''
+    Ignore backfill wakeups from overlay markets and other periods.
+
+    Samplerd broadcasts every backfill marker to every period and FSP
+    subscriber. With multiple markets overlaid, an unrelated provider
+    frame used to wake each primary-market cascade and race its own SHM
+    inspection. Feed regular sample steps and matching history events
+    through, but reject both a foreign FQME and the 60-second period.
+
+    '''
+    assert _is_relevant_sample_msg(
+        msg,
+        fqme='tsla.nasdaq.ib',
+        period_s=1,
+    ) is expected
+
+
+@pytest.mark.parametrize(
+    'msg,step_diff,expected',
+    [
+        ({'index': 1}, 1, True),
+        ({'index': 1}, 0, False),
+        ({'backfilling': ('tsla.nasdaq.ib', 1)}, 1, False),
+        ({'backfilling': ('tsla.nasdaq.ib', 1)}, 0, False),
+    ],
+)
+def test_only_new_sample_steps_advance_destination(
+    msg: SampleMsg,
+    step_diff: int,
+    expected: bool,
+
+) -> None:
+    '''
+    Keep queued history and duplicate wakeups from adding FSP rows.
+
+    A recomputation can consume several prepends before their sampler
+    markers are delivered. Those stale markers then observe aligned SHM
+    bounds and previously appended duplicate destination rows. Model the
+    queued marker and regular duplicate cases, and prove only a genuine
+    one-step source lead authorizes destination advancement.
+
+    '''
+    assert _should_advance_dst(msg, step_diff) is expected
+
+
+@pytest.mark.parametrize(
+    'msg,synced,expected',
+    [
+        ({'index': 1}, True, False),
+        ({'index': 1}, False, True),
+        ({'backfilling': ('tsla.nasdaq.ib', 1)}, True, True),
+        ({'backfilling': ('tsla.nasdaq.ib', 1)}, False, True),
+    ],
+)
+def test_backfill_always_invalidates_fsp_history(
+    msg: SampleMsg,
+    synced: bool,
+    expected: bool,
+
+) -> None:
+    '''
+    Recompute in-place history repairs with unchanged SHM bounds.
+
+    Gap repair can replace null source rows without moving either SHM
+    bound. A bounds-only predicate therefore reports synchronization
+    even though derived values are stale. Prove every relevant backfill
+    marker forces history replay while an aligned sample event does not.
+
+    '''
+    assert _needs_history_resync(msg, synced) is expected
+
+
+def test_history_compute_uses_immutable_source_snapshot() -> None:
+    '''
+    Do not publish output and timestamps from different source ranges.
+
+    A source append or prepend can occur while the FSP's first yield is
+    computing. The old path read timestamps only afterward and could pair
+    an N-row result with a shifted N-row timestamp tail. Mutate the live
+    last bound during the first yield, then prove the accepted output and
+    timestamps retain the original range while later reads switch to live
+    SHM.
+
+    '''
+    shm = Shm(10, 20, token='source')
+    live_lengths: list[int] = []
+
+    async def edge(
+        _source: AsyncIterator[FeedQuote],
+        src_shm: ShmArray,
+
+    ) -> AsyncIterator[np.ndarray]:
+        shm._last.value += 1
+        yield np.ones(len(src_shm.array))
+        live_lengths.append(len(src_shm.array))
+
+    async def main() -> None:
+        (
+            out_stream,
+            history,
+            bounds,
+            src_history,
+        ) = await _open_history_snapshot(
+            cast(EdgeFunc, edge),
+            quote_stream=cast(
+                AsyncIterator[QuoteFrame],
+                object(),
+            ),
+            fqme='tsla.nasdaq.ib',
+            src_shm=cast(ShmArray, shm),
+        )
+        assert bounds == (10, 20)
+        assert len(history) == 10
+        assert len(src_history) == 10
+        with pytest.raises(StopAsyncIteration):
+            await anext(out_stream)
+
+    trio.run(main)
+
+    assert live_lengths == [11]
+
+
 @pytest.mark.parametrize('target', [wma, rsi, tina_vwap])
 def test_builtin_fsp_stream_contract(target: Fsp) -> None:
     '''
@@ -131,3 +413,98 @@ def test_builtin_fsp_stream_contract(target: Fsp) -> None:
         await stream.aclose()
 
     trio.run(main)
+
+
+class Viz:
+    def __init__(self, name: str, shm: Shm) -> None:
+        self.name: str = name
+        self.shm: Shm = shm
+        self.updates: int = 0
+        self.force_redraws: list[bool] = []
+        self._last_fsp_update_sig: (
+            tuple[int, int, bytes]|None
+        ) = None
+        self._mxmns: dict[
+            tuple[int, int],
+            tuple[int, int],
+        ] = {(10, 20): (1, 2)}
+        self.view = SimpleNamespace(
+            rescale_count=0,
+        )
+
+        def rescale(
+            *,
+            do_linked_charts: bool,
+            do_overlay_scaling: bool,
+
+        ) -> None:
+            assert not do_linked_charts
+            assert do_overlay_scaling
+            self.view.rescale_count += 1
+
+        self.view.interact_graphics_cycle = rescale
+        self.plot: SimpleNamespace = SimpleNamespace(
+            getAxis=lambda name: SimpleNamespace(_stickies={}),
+            vb=self.view,
+        )
+
+    def update_graphics(
+        self,
+        force_redraw: bool = False,
+
+    ) -> None:
+        self.updates += 1
+        self.force_redraws.append(force_redraw)
+
+
+def test_fsp_history_update_redraws_only_derived_vizs() -> None:
+    '''
+    Keep a source-history prepend from refreshing the whole chart.
+
+    FSP cascades recompute after each near-term backfill frame. The old
+    notification called the linked chart's full graphics cycle, which
+    redrew the primary market and every overlay and disturbed the live
+    view repeatedly. Arrange two visualizations sharing the rebuilt FSP
+    SHM plus unrelated source and overlay SHMs, then prove only the two
+    derived curves receive an update.
+
+    '''
+    fsp_shm = FspShm(10, 20, token='derived')
+    source_viz = Viz('source', Shm(10, 20, token='source'))
+    first_fsp_viz = Viz('flow', fsp_shm)
+    second_fsp_viz = Viz('dark_flow', fsp_shm)
+    overlay_viz = Viz('overlay', Shm(10, 20, token='overlay'))
+
+    linked: LinkedSplits = cast(
+        LinkedSplits,
+        SimpleNamespace(
+            chart=SimpleNamespace(
+                _vizs={
+                    'source': source_viz,
+                    'overlay': overlay_viz,
+                },
+            ),
+            subplots={
+                'volume': SimpleNamespace(
+                    _vizs={
+                        'flow': first_fsp_viz,
+                        'dark_flow': second_fsp_viz,
+                    },
+                ),
+            },
+        ),
+    )
+
+    updated = update_fsp_vizs(linked, fsp_shm._token)
+
+    assert updated == 2
+    assert first_fsp_viz.updates == 1
+    assert second_fsp_viz.updates == 1
+    assert first_fsp_viz.force_redraws == [True]
+    assert second_fsp_viz.force_redraws == [True]
+    assert first_fsp_viz._mxmns == {}
+    assert second_fsp_viz._mxmns == {}
+    assert first_fsp_viz.view.rescale_count == 1
+    assert second_fsp_viz.view.rescale_count == 1
+    assert source_viz.updates == 0
+    assert overlay_viz.updates == 0