feat: decouple storage sinks from USB drain — revised (supersedes #96)

[boringethan]

Refs #116. (Refs, not Fixes — the ticket must stay open through pre-release validation.)

What

This is #96 plus the fixes from its review, rebased onto current next (through #111). It carries the original two commits (AsyncSink decoupling + backpressure instrumentation) and adds one commit addressing the review findings. If this merges, #96 should be closed.

Review fixes on top of #96

1. Batch-queue sizing was off ~16× (the PR's own success criterion). A batch row is one per-camera histogram sample — parse_histogram_stream fires on_row once per camera per capture — so a dual-sensor 8+8 scan pushes ~640 rows/s into the shared queue, and the original ceil(3.0 × CAPTURE_HZ / batch_size) bought ~0.19 s of buffering, not the documented 3 s. The depth is now derived from CAPTURE_HZ × active_cameras, where the camera count comes from the mask popcounts of the sides that actually stream (composed with #108's connected+masked gate; the source worst-cases 8 cams/side when unknown). The sizing test now asserts wall-clock capacity at the true row rate instead of validating the formula against itself. Memory note: the full 3 s budget can transiently pin ~125 MB during a worst-case dual-sensor stall (dense (N,2,8,1024) batch layout — see follow-up below); bounded and released as the runner drains.

2. AsyncSink join-timeout finalize race. If the worker failed to drain within 60 s, on_complete() previously finalized the wrapped sink on the runner thread while the worker could be mid-consume() on the same sink — racing the in-flight write, and for ScanDBSink potentially deleting the session as "empty" while its data was still buffered. The worker now owns finalization (runs on_complete after the sentinel, always serialized with consume); on a join timeout the runner defers to the worker instead of racing it, so a late-unwedging disk still persists everything.

3. AsyncSink reuse poisoning. The queue was created once and never drained, so reusing an instance after a timed-out scan could replay stale payloads into the new session or kill the new worker with the stale sentinel, after which consume()'s blocking put would wedge the runner thread permanently. Each on_scan_start now gets a fresh queue, and workers bind their queue as an argument so a swap can never redirect a still-draining previous worker.

4. Drop warnings could be suppressed by stall warnings. Both warning sites shared one 5 s rate-limit token, so a benign 100 ms-stall warning could swallow the first real data-loss warning for up to 5 s. They now rate-limit independently.

5. Docs (the convention → contract gap). docs/API.md §6 advertised custom sinks with no mention that consume() runs on the drain-coupled runner thread — a user storage sink would silently reintroduce the exact frame-drop failure this PR root-causes. API.md and the Sink protocol docstring now state the latency budget (wrap I/O sinks in AsyncSink) and that payloads are dispatched by reference and consumed concurrently — never mutate them.

Tests

• New: join-timeout deferred finalize, fresh-queue reuse, drop-warning-not-suppressed, row-rate sizing (8/16 cams × 10/20 batch), worst-case default sizing.
• After rebasing onto next @ c1f7b70: tests/test_pipeline/, tests/test_scan_workflow.py, tests/test_stream_interface_backpressure.py — 274 passed, 30 skipped, 0 failed (the fix: exclude unconnected/unmasked sensors from LiveUsbSource #108/fix: always emit OFF TriggerStateEvent even when stop_trigger raises #109 test fakes needed **kwargs on their fake LiveUsbSource factories to accept the new active_cameras parameter).

Follow-ups (out of scope)

• Dense one-hot (N,2,8,1024) FrameBatch layout inflates buffered-batch memory ~16×; a sparse per-row layout would cut the worst-case stall footprint from ~125 MB to ~8 MB.
• Runner-level async dispatch (or a Sink.blocking flag) would make the decoupling structural instead of wrap-by-convention.

🤖 Generated with Claude Code

[boringethan]

For the reviewer: #293 is a smaller alternative for the data-loss part of #116. It sizes the per-side packet queue, which holds raw bytes, to 10 s and leaves the sinks inline. The two were compared head to head in the same harness. Both prevent frame loss in every scenario; this PR additionally keeps the live display within ~0.15 s during a DB stall, where #293's display lags by the stall length. Suggest deciding between them together: merge #293 and either close this or cut it down to the AsyncSink part.