feat: decouple storage sinks from the USB drain; surface host backpressure

omotion/ScanWorkflow.py

@@ -378,6 +378,7 @@ def start_scan(
         from omotion.pipeline.sinks import (
             CsvSink as PipelineCsvSink, DiagnosticsLogSink, ScanDBSink, ScanMetadata,
         )
+        from omotion.pipeline.async_sink import AsyncSink
         from omotion.pipeline.pedestal import SensorPedestals, pedestal_for_fw
 
         # Clear the prior scan's outcome up front so a refusal below (busy, or
@@ -479,8 +480,12 @@ def _safe_pedestal(sensor) -> float:
                     True if scan_db_path is None
                     else bool(request.write_corrected_csv)
                 )
+                # AsyncSink: run the CSV writes on a worker thread so a disk
+                # flush can't stall the runner's USB-drain loop and overflow
+                # the firmware histo queue (2026-06 soak root-cause).
                 default_sinks.append(
-                    PipelineCsvSink(output_dir=data_dir, write_corrected=write_corrected)
+                    AsyncSink(PipelineCsvSink(output_dir=data_dir,
+                                              write_corrected=write_corrected))
                 )
             if scan_db_path is not None:
                 # Pre-flight the scan DB before the laser fires. The DB is the
@@ -513,7 +518,11 @@ def _safe_pedestal(sensor) -> float:
                     with self._lock:
                         self._running = False
                     return False
-                default_sinks.append(ScanDBSink(db_path=scan_db_path))
+                # AsyncSink: keep the SQLite commit/fsync off the USB-drain
+                # thread. AsyncSink mirrors `critical`, so a DB on_scan_start
+                # failure still aborts the scan (and the pre-flight above
+                # already caught the common case with the laser off).
+                default_sinks.append(AsyncSink(ScanDBSink(db_path=scan_db_path)))
             # Telemetry CSV: per-scan snapshots from ConsoleTelemetryPoller.
             # Not a pipeline sink — the poller is its own daemon thread that
             # predates the sink-based architecture, so we register a listener

omotion/StreamInterface.py

@@ -119,6 +119,19 @@ def __init__(self, dev, interface_index, desc="Stream"):
         self.expected_size = None
         self.isStreaming = False
         self.packets_received: int = 0  # USB transfers queued since last start_streaming
+        # Host-side backpressure stats since last start_streaming. A put that
+        # blocks because the parser/runner stalled is the host-side cause of
+        # the sensor firmware's "HISTO enqueue fail" drops; surfacing it per
+        # scan makes that previously-invisible stall measurable.
+        self.backpressure_stalls: int = 0      # puts that blocked >= warn threshold
+        self.backpressure_stall_s: float = 0.0  # total seconds blocked on those puts
+        self.dropped_chunks: int = 0            # chunks dropped (queue full past timeout)
+        self._last_backpressure_warn_s: float = 0.0
+        # A put blocking this long is a sub-second stall worth counting/warning.
+        self._backpressure_warn_s: float = 0.1
+        # Drop a chunk only after the queue has been full this long (the loop's
+        # original >1 s drop policy). Kept as an attribute so tests can shorten it.
+        self._put_timeout_s: float = 1.0
 
     def start_streaming(self, queue_obj, expected_size):
         # Recover from a stale thread left over by a previous scan whose
@@ -143,6 +156,10 @@ def start_streaming(self, queue_obj, expected_size):
         self.data_queue = queue_obj
         self.expected_size = expected_size
         self.packets_received = 0
+        self.backpressure_stalls = 0
+        self.backpressure_stall_s = 0.0
+        self.dropped_chunks = 0
+        self._last_backpressure_warn_s = 0.0
         self.stop_event.clear()
         self.thread = threading.Thread(target=self._stream_loop, daemon=True)
         self.thread.start()
@@ -168,10 +185,19 @@ def stop_streaming(self):
         self.isStreaming = False
         self.data_queue = None
         self.expected_size = None
-        logger.info(
+        summary = (
             f"{self.desc}: Streaming stopped — "
             f"{self.packets_received} USB read chunk(s) received"
         )
+        if self.backpressure_stalls or self.dropped_chunks:
+            summary += (
+                f"; host backpressure: {self.backpressure_stalls} stall(s) "
+                f">{int(self._backpressure_warn_s * 1000)}ms "
+                f"totalling {self.backpressure_stall_s:.2f}s, "
+                f"{self.dropped_chunks} chunk(s) dropped"
+            )
+        log = logger.warning if self.dropped_chunks else logger.info
+        log(summary)
 
     def flush_stale_data(
         self,
@@ -359,6 +385,51 @@ def _process_packet(self, raw, pkt_count, cmp_count, cmp_errors):
             self.data_queue.put(raw)
         return cmp_count, cmp_errors
 
+    def _enqueue_chunk(self, data_queue, data) -> bool:
+        """Put one USB chunk onto the packet queue, measuring host-side
+        backpressure.
+
+        A put that returns immediately costs nothing; one that blocks means
+        the parser/runner has stalled and is holding the queue full — the
+        host-side cause of the firmware's histo-queue overflow. We time the
+        block: any wait >= ``_backpressure_warn_s`` is counted (and rate-limit
+        warned) so sub-second stalls are visible, not just the >1 s drops.
+
+        Returns True if the chunk was enqueued, False if it was dropped
+        because the queue stayed full past ``_put_timeout_s``.
+        """
+        t0 = time.monotonic()
+        try:
+            data_queue.put(data, timeout=self._put_timeout_s)
+        except queue.Full:
+            self.dropped_chunks += 1
+            # Real host-side data loss — warn (rate-limited) so it's visible
+            # mid-scan, not only in the stop summary.
+            if t0 - self._last_backpressure_warn_s >= 5.0:
+                self._last_backpressure_warn_s = t0
+                logger.warning(
+                    "%s: packet queue full for >%.0fs (parser falling behind) "
+                    "— dropping %d-byte chunk; %d dropped so far this scan",
+                    self.desc, self._put_timeout_s, len(data), self.dropped_chunks,
+                )
+            return False
+        blocked = time.monotonic() - t0
+        self.packets_received += 1
+        if blocked >= self._backpressure_warn_s:
+            self.backpressure_stalls += 1
+            self.backpressure_stall_s += blocked
+            # Rate-limit to at most once per 5 s so a sustained stall doesn't
+            # flood the log; every stall is still counted above.
+            if t0 - self._last_backpressure_warn_s >= 5.0:
+                self._last_backpressure_warn_s = t0
+                logger.warning(
+                    "%s: USB read stalled %.0f ms on a full packet queue "
+                    "(parser/runner falling behind — risks firmware histo "
+                    "drops); %d stall(s) so far this scan",
+                    self.desc, blocked * 1000.0, self.backpressure_stalls,
+                )
+        return True
+
     def _stream_loop(self):
         # Read timeout must exceed the worst-case USB transfer latency for the
         # final frame.  Normal cadence is ~25 ms; the last frame of a scan can
@@ -392,22 +463,15 @@ def _stream_loop(self):
                     timeout=_READ_TIMEOUT_MS,
                 )
                 if data and data_queue is self.data_queue:
-                    # Use a bounded put so the loop can never block forever
-                    # on a stopped/slow parser. With self.stop_event set the
+                    # Bounded put (see _enqueue_chunk) so the loop can never
+                    # block forever on a stopped/slow parser, and so host-side
+                    # backpressure is measured. With self.stop_event set the
                     # parser also drains until empty (see parse_histogram_stream),
-                    # so this drop window is only ever 1s of backlog at scan
-                    # teardown — small price for a guaranteed loop exit.
-                    try:
-                        data_queue.put(bytes(data), timeout=1.0)
-                        self.packets_received += 1
-                    except queue.Full:
-                        if self.stop_event.is_set():
-                            break
-                        logger.warning(
-                            "%s: data_queue full for >1s during streaming "
-                            "(parser falling behind?); dropping %d-byte chunk",
-                            self.desc, len(data),
-                        )
+                    # so a drop here is only ever <=_put_timeout_s of backlog at
+                    # scan teardown — small price for a guaranteed loop exit.
+                    if not self._enqueue_chunk(data_queue, bytes(data)) \
+                            and self.stop_event.is_set():
+                        break
             except usb.core.USBError as e:
                 if e.errno in (110, 10060):
                     # Timeout — no data arrived within the read window.

omotion/pipeline/__init__.py

@@ -36,6 +36,7 @@
 from .factory import default_pipeline
 from .pedestal import SensorPedestals
 from .sinks import Sink, ScanMetadata, CsvSink, ScanDBSink, DiagnosticsLogSink
+from .async_sink import AsyncSink
 from .sources import Source, LiveUsbSource, CsvReplaySource
 from .telemetry import (
     TelemetryAggregator, TelemetryFeeder, TelemetryIngestStage, TelemetrySample,
@@ -51,7 +52,7 @@
     "default_pipeline",
     "SensorPedestals",
     "Sink", "ScanMetadata",
-    "CsvSink", "ScanDBSink", "DiagnosticsLogSink",
+    "CsvSink", "ScanDBSink", "DiagnosticsLogSink", "AsyncSink",
     "Source", "LiveUsbSource", "CsvReplaySource",
     "TelemetryAggregator", "TelemetryFeeder", "TelemetryIngestStage",
     "TelemetrySample",

omotion/pipeline/async_sink.py

@@ -0,0 +1,119 @@
+"""AsyncSink — run a slow sink's consume() on its own worker thread.
+
+The ScanRunner consume loop runs on the same thread that drains the live
+USB batch queue (see ``LiveUsbSource`` / ``StreamInterface``). A sink that
+blocks there — an fsync on a SQLite commit, a CSV buffer flushing to disk —
+stalls that drain, which backpressures the histogram parser and then
+``dev.read``, overflowing the sensor firmware's shallow 4-deep histo queue
+and dropping frames. (Root-caused on the 2026-06 long-scan soak: the
+firmware "HISTO enqueue fail: queue full" drops were brief host-side drain
+stalls, not a USB fault.)
+
+``AsyncSink`` wraps a sink and moves ``consume()`` onto a dedicated worker
+thread fed by a bounded FIFO queue, so a *transient* sink stall is absorbed
+by the buffer instead of the USB path. Sustained overload (a sink that can
+never keep up) still backpressures once the buffer fills — that is correct;
+you cannot outrun the disk forever and dropping persisted data is worse.
+
+Lifecycle:
+  * ``on_scan_start`` runs the wrapped sink synchronously and propagates any
+    exception, so a ``critical`` sink whose start fails still aborts the scan
+    (ScanRunner reads ``.critical``) and the sink is fully initialized before
+    any ``consume`` is enqueued.
+  * ``consume`` enqueues ``(channel, payload)`` (cheap) for the worker.
+  * ``on_complete`` enqueues a sentinel, waits for the worker to drain every
+    buffered item, then calls the wrapped sink's ``on_complete`` — so no
+    buffered data is lost.
+"""
+
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+from typing import Any
+
+logger = logging.getLogger("openmotion.sdk.pipeline.async_sink")
+
+_SENTINEL = object()
+
+
+class AsyncSink:
+    """Threaded buffer in front of a Sink. Duck-types the Sink protocol."""
+
+    def __init__(self, sink: Any, *, max_queue: int = 64,
+                 join_timeout_s: float = 60.0) -> None:
+        self._sink = sink
+        self._queue: queue.Queue = queue.Queue(maxsize=max(1, int(max_queue)))
+        self._join_timeout_s = float(join_timeout_s)
+        self._worker: threading.Thread | None = None
+        self._started = False
+        self._closed = False
+        self._name = type(sink).__name__
+        # Mirror the attributes the runner reads off a sink so AsyncSink is a
+        # drop-in stand-in (channels for routing, critical for abort policy).
+        self.channels = getattr(sink, "channels", set())
+        self.critical = getattr(sink, "critical", False)
+
+    @property
+    def wrapped(self):
+        """The underlying sink, for callers that need to introspect past the
+        async wrapper (e.g. ``isinstance`` checks on the concrete sink type)."""
+        return self._sink
+
+    def on_scan_start(self, meta) -> None:
+        # Synchronous on purpose: a critical sink that fails here must still
+        # raise so ScanRunner aborts before the laser fires, and the sink must
+        # be ready before the first enqueued consume runs on the worker.
+        self._sink.on_scan_start(meta)
+        self._closed = False
+        self._worker = threading.Thread(
+            target=self._run, name=f"AsyncSink-{self._name}", daemon=True,
+        )
+        self._started = True
+        self._worker.start()
+
+    def consume(self, channel: str, payload: Any) -> None:
+        if not self._started or self._closed:
+            # No worker to service the item (start failed/skipped or already
+            # finalized). Dropping is correct: there is nowhere to deliver it.
+            return
+        # Blocking put: absorbs transient stalls up to max_queue, and applies
+        # honest backpressure only on sustained overload.
+        self._queue.put((channel, payload))
+
+    def on_complete(self) -> None:
+        if not self._started:
+            # on_scan_start never ran (or raised) — no worker was created;
+            # finalize the wrapped sink directly so it still tears down.
+            self._sink.on_complete()
+            return
+        if self._closed:
+            return
+        self._closed = True
+        self._queue.put(_SENTINEL)
+        self._worker.join(timeout=self._join_timeout_s)
+        if self._worker.is_alive():
+            logger.error(
+                "AsyncSink(%s): worker did not drain within %.0fs; "
+                "%d buffered item(s) may be lost",
+                self._name, self._join_timeout_s, self._queue.qsize(),
+            )
+        self._sink.on_complete()
+
+    def _run(self) -> None:
+        while True:
+            item = self._queue.get()
+            try:
+                if item is _SENTINEL:
+                    return
+                channel, payload = item
+                try:
+                    self._sink.consume(channel, payload)
+                except Exception:
+                    logger.exception(
+                        "AsyncSink(%s): consume raised on channel %s; "
+                        "continuing", self._name, channel,
+                    )
+            finally:
+                self._queue.task_done()

omotion/pipeline/sources.py

@@ -11,6 +11,7 @@
 
 import csv
 import logging
+import math
 import queue
 import threading
 import time
@@ -20,12 +21,20 @@
 
 import numpy as np
 
+from omotion.config import CAPTURE_HZ
 from .batch import FrameBatch
 from .sinks import ScanMetadata
 
 
 logger = logging.getLogger("openmotion.sdk.pipeline.sources")
 
+# How long a downstream stall (a slow sink, a GC pause, a pipeline CPU spike)
+# the host should absorb before the USB reader is starved and the sensor
+# firmware's 4-deep (~100 ms) histo queue overflows. The batch queue default
+# is sized to hold this many seconds of frames so the stall is buffered here
+# instead of backpressuring dev.read. See the 2026-06 long-scan soak.
+_TARGET_BUFFER_SECONDS = 3.0
+
 
 @runtime_checkable
 class Source(Protocol):
@@ -195,6 +204,7 @@ def __init__(self, *,
                  batch_size_frames: int = 10,
                  flush_interval_s: float = 0.25,
                  packet_queue_size: int = 64,
+                 batch_queue_size: Optional[int] = None,
                  metadata: ScanMetadata):
         super().__init__(metadata=metadata)
         self._console = console
@@ -205,7 +215,20 @@ def __init__(self, *,
             side: queue.Queue(maxsize=packet_queue_size)
             for side, sensor in self._sensors.items() if sensor is not None
         }
-        self._batch_queue: queue.Queue = queue.Queue(maxsize=4)
+        # Parsed-batch buffer between the per-side reader threads and the
+        # runner. Was a hardcoded 4 (~1 s); a transient runner stall (sink
+        # fsync, GC) filled it and backpressured the parser -> dev.read ->
+        # the firmware's 4-deep histo queue overflowed and dropped frames.
+        # Default it to ~_TARGET_BUFFER_SECONDS of frames, derived from the
+        # 40 Hz capture rate and the batch size (so the intent survives a
+        # different batch_size_frames), and keep it tunable. The runner stalls
+        # before the parser does, so this batch buffer is the slack that keeps
+        # dev.read flowing; the 64-deep packet queue adds ~1.6 s more behind
+        # it. See the 2026-06 long-scan soak root-cause + AsyncSink.
+        if batch_queue_size is None:
+            frames = _TARGET_BUFFER_SECONDS * CAPTURE_HZ
+            batch_queue_size = math.ceil(frames / max(1, self._batch_size))
+        self._batch_queue: queue.Queue = queue.Queue(maxsize=max(1, int(batch_queue_size)))
         self._stop = threading.Event()
         self._reader_threads: list[threading.Thread] = []