MARTe-Integrated-Components/docs/superpowers/specs/2026-06-25-streamhub-binary-recorder-design.md

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 <sourceId>_<UTC-timestamp>.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.