From dca48729766c4838e97940192a167e68975054f7 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 00:45:25 +0200 Subject: [PATCH] Add design spec for StreamHub binary recorder Lossless packet-decode capture to FileWriter-compatible per-source binary files with size-capped rotation, config + WS control, and double-buffered flush on the push thread (Approach C). Co-Authored-By: Claude Opus 4.6 --- ...-06-25-streamhub-binary-recorder-design.md | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 docs/superpowers/specs/2026-06-25-streamhub-binary-recorder-design.md diff --git a/docs/superpowers/specs/2026-06-25-streamhub-binary-recorder-design.md b/docs/superpowers/specs/2026-06-25-streamhub-binary-recorder-design.md new file mode 100644 index 0000000..026c5b0 --- /dev/null +++ b/docs/superpowers/specs/2026-06-25-streamhub-binary-recorder-design.md @@ -0,0 +1,188 @@ +# StreamHub Binary Recorder — Design + +**Date:** 2026-06-25 +**Component:** `Source/Applications/StreamHub/` +**Status:** Approved (design phase) + +## Goal + +Add an option to the headless C++ StreamHub app to store incoming signal data to +disk in MARTe2 FileWriter-compatible binary format. The operator can record **all +signals** ("full packets") or a **user-specified subset**. Capture is **lossless** +(full source rate) and produces files **byte-identical to a MARTe2 FileWriter +capture**, so they validate against the existing `Test/E2E/.../validate_binary.py` +tooling. + +## Locked Decisions + +| Decision | Choice | +|---|---| +| Tap point | Packet-decode time (lossless, full source rate) | +| Data types | Native packet types (byte-identical to FileWriter) | +| File layout | One `.bin` file per source/session | +| Retention | Rotating files: size cap (`MaxFileMB`) + keep newest `KeepFiles` | +| Control | Auto-start on launch (config) + WS `recStart`/`recStop` | +| Signal subset | Config default (`"all"` or list) + WS override per run | +| Disk-write threading | **Approach C**: receive thread serializes into a per-source double-buffer; the existing 30 Hz push thread flushes to disk | + +## Architecture + +### Component & ownership + +New `BinaryRecorder` class (`BinaryRecorder.{h,cpp}`), **one instance owned by each +`UDPSourceSession`** — the per-source analog of the global `HistoryWriter`. + +``` +UDPSourceSession + ├─ rings_[] (float64, live/zoom/trigger) ← unchanged + ├─ stats_ ← unchanged + └─ recorder_ (BinaryRecorder) ← NEW: native-type, lossless, file-per-source +``` + +- The session configures its recorder when it knows its signal layout (first CONFIG, + in `ParseConfigPayload`), since the on-disk header is derived from the signal + descriptors and the resolved subset. +- Lifecycle mirrors the session: `Configure(descs, subset) → [recording] → Stop()`. + A CONFIG change closes the current file and re-opens with a new header. +- The recorder object always exists; it only writes when **armed** (auto-armed at + startup if config enables it; toggled by WS). + +### On-disk format (byte-identical to MARTe2 FileWriter) + +``` +[u32 numSigs] +per signal: [u16 typeCode][char[32] name][u32 numElements] (38 B) +then rows: each row = one cycle, all subset signals, signal-major, + native types, little-endian +``` + +- `numSigs`, descriptors, and `numElements` come from the resolved subset, so a + subset capture is a valid standalone FileWriter file containing only the chosen + signals. +- **Native-type reconstruction:** the wire payload may be quantized + (`quantType != NONE`). The recorder dequantizes to the physical value then + re-encodes to the descriptor's `typeCode` (float32/uint32/int16/...). For + `quantType == NONE` it is a straight byte copy. + +### Rotation (size cap + keep N) + +- Files named `_.bin` in the configured directory. +- When the active file reaches `MaxFileMB`: close it, open a new one (fresh header), + delete the oldest until at most `KeepFiles` remain for that source. +- `MinDiskFreeMB` guard (like HistoryWriter): if free space drops below it, stop + writing (disarm) and log, rather than filling the disk. + +### ACCUMULATE publish mode + +A packet can carry `numSamples` values for an accumulated scalar while other signals +carry 1. To keep fixed-width FileWriter rows, the recorder emits **`numSamples` rows +per packet**, repeating the non-accumulated signals' values across those rows (the +natural "one row per cycle" expansion, matching how the rings already receive +`numSamples` writes). + +### Capture path (receive thread) + +In `ParseDataPayload`, after the existing decode/ring-write loop, if the recorder is +armed: + +- Serialize the packet into native-type FileWriter rows (1 row, or `numSamples` rows + in ACCUMULATE) and append the bytes into the **active staging buffer**. +- Staging is a per-source **double buffer** (two growable byte buffers, + `front`/`back`). The receive thread always appends to `front`. A short + `FastPollingMutexSem` protects only the buffer-swap and append bookkeeping — never + the disk. +- **Overflow guard:** each staging buffer has a soft cap (`StagingMB`, default 8 MB). + If the push thread has fallen so far behind that `front` would exceed the cap, the + recorder increments a `droppedRows` counter and skips the append (keeping reception + alive). Reported via stats/WS so loss is observable, not silent. +- No disk syscalls ever run on the receive thread. + +### Flush path (push thread) + +Inside the existing 30 Hz push loop, after rings/zoom for each session, the recorder +does its disk work on the **push thread**: + +1. **Swap** `front`/`back` under the short mutex (receive thread keeps appending to + the new `front` immediately). +2. **Write** the filled `back` buffer with `pwrite()` at the running offset; clear + `back` for reuse. +3. **Rotate** if the file offset crossed `MaxFileMB` (close, open new with header, + prune to `KeepFiles`, check `MinDiskFreeMB`). +4. **fsync cadence:** `fdatasync()` every `FlushIntervalSec` (like HistoryWriter), + not every tick. + +Arm/disarm/rotate requests from WS (other threads) are applied here via a small +pending-flag, so all file open/close/unlink happens on one thread — no cross-thread +file races. Durability latency is bounded to one push tick (~33 ms) plus the fsync +interval. + +## Configuration + +New `+Recorder` block in the StreamHub cfg (parsed in `Initialise`, try +`+Recorder`/`Recorder`): + +``` ++Recorder = { + Enabled = 1 // master enable; arms at startup if AutoStart=1 + AutoStart = 1 // begin recording on launch + Directory = "/var/streamhub/rec" // required + Signals = "all" // "all" OR comma-separated "src:sig" keys + MaxFileMB = 256 // size cap → roll + KeepFiles = 8 // newest N kept per source + StagingMB = 8 // per-source staging soft cap (overflow guard) + FlushIntervalSec= 5 // fdatasync cadence + MinDiskFreeMB = 500 // stop-writing guard +} +``` + +- `Signals = "all"` → every signal of every source. A subset list selects per-source + signals by `src:sig` key; a source with no selected signals does not record. + +## WebSocket control + +Dispatched in `OnWSCommand` via the existing strcmp-on-`type` pattern; replies +unicast. + +- `recStart` — `{type:"recStart", signals?:"all"|["src:sig",...]}` → arm; optional + `signals` overrides the config subset for this run. +- `recStop` — disarm, flush, close files. +- `recInfo` — returns + `{enabled, recording, perSource:[{id, file, bytesWritten, rowsWritten, droppedRows, freeMB}]}`. +- `recStatus` event broadcast on arm/disarm/rotate/overflow so clients reflect state + without polling. + +## Testing + +### Unit tests (GTest, `Test/Applications/StreamHub/`) + +- **Header/format:** configure with a known descriptor set + subset → assert written + header bytes match the FileWriter layout (`numSigs`, 38-byte descriptors, native + `typeCode`/`numElements`). +- **Native-type re-encode:** feed quantized and non-quantized inputs → assert on-disk + bytes equal the expected native-type encoding. +- **ACCUMULATE expansion:** packet with `numSamples>1` for a scalar → assert + `numSamples` rows emitted, non-accumulated signals repeated. +- **Rotation:** drive past `MaxFileMB` → assert roll, new header, prune to + `KeepFiles`, oldest deleted. +- **Overflow guard:** stall the flush, overfill staging → assert `droppedRows` + increments and the reception path does not block. +- **Subset selection:** `"all"` vs explicit `src:sig` list → assert only chosen + signals appear in the file. + +### E2E (extend `Test/E2E/`) + +- Add a recorder scenario: run StreamHub with `+Recorder` against a live UDPStreamer + source, then validate the produced `.bin` with the existing `validate_binary.py` + (checks signal count/sizes + row matching). +- Reuse `run_e2e_report.sh`'s output-dir + JSON + Typst-report plumbing so the + recorder result lands in the report alongside the unicast/multicast results. + +## Risks / Notes + +- **Native re-encode vs quantization:** correctness depends on the dequantize + + re-encode path matching the original type exactly; covered by a dedicated unit test. +- **Staging memory:** `StagingMB` per source bounds RAM; overflow is observable via + `droppedRows` rather than unbounded growth. +- **Rotation hiccups:** file open/unlink on the push thread (non-RT), never on the + receive path. +- **CONFIG mid-recording:** re-headers a new file; old file is closed cleanly.