Files
MARTe-Integrated-Components/docs/superpowers/plans/2026-06-25-streamhub-binary-recorder.md
Martino Ferrari e3b458ed94 Add implementation plan for StreamHub binary recorder
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-06-25 00:52:54 +02:00

219 lines
20 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# StreamHub Binary Recorder Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add an option to the headless C++ StreamHub app to record incoming signal data to disk in MARTe2 FileWriter-compatible binary format, one rotating file per source, lossless at packet-decode time, with config + WebSocket control.
**Architecture:** A standalone `BinaryRecorder` class (no dependency on `UDPSourceSession`) owned per session. The session's receive thread serializes each decoded packet into native-type FileWriter rows and appends them to a per-source double buffer; the existing 30 Hz push thread flushes the buffer to disk and handles file rotation, pruning, and fsync (Approach C). Layout (header + per-signal row encoding) is built on the receive thread (same thread as capture, so no layout race); only the staging buffers, the header blob, and control flags cross to the push thread.
**Tech Stack:** C++ (MARTe2 app style; STL allowed but prefer plain arrays + `FastPollingMutexSem`), `pwrite`/`fdatasync`/`statvfs` POSIX I/O, GTest (sources `#include`d into the `StreamHubTest` lib, auto-discovered by `MainGTest`), Python (`validate_binary.py`) + Typst for E2E.
## Global Constraints
- MARTe2 app style in `Source/Applications/StreamHub/`: links MARTe2 core; use `FastPollingMutexSem` on shared paths, not OS mutexes. STL is tolerated here but the existing files avoid it — match the surrounding style (plain arrays, `memcpy`, `snprintf`).
- EUPL v1.1 license header on every new C++ source/header (copy the block from `HistoryWriter.cpp`).
- On-disk format MUST be byte-identical to MARTe2 FileWriter binary output for un-quantized signals: `[u32 numSigs]` then per signal `[u16 TypeDescriptor.all][char[32] name(null-padded)][u32 numElements]`, then rows; each row = all included signals concatenated in order, each signal's `numElements` elements in native type, little-endian. (For quantized signals the stored value is the dequantized reconstruction — lossy by nature — documented, not byte-identical.)
- Type field is the MARTe2 `TypeDescriptor.all` (uint16), mapped from the UDPS `typeCode` (uint8). `validate_binary.py` does not compare the type value across files, but real FileWriter compatibility requires the correct `TypeDescriptor.all`.
- Signal name in the on-disk header is 32 bytes (FileWriter), truncated/padded from the UDPS 64-byte name.
- Config block name is `+Recorder` (StandardParser keeps the `+`; try `+Recorder` then `Recorder`, as `HistoryWriter` does for `+History`).
- WS command dispatch is `strcmp` on the `type` field in `StreamHub::OnWSCommand`.
- `kMaxSessions = 32`, `UDPSS_MAX_SIGNALS = 256`.
---
## File Structure
- Create `Source/Applications/StreamHub/BinaryRecorder.h``RecorderConfig` struct + `BinaryRecorder` class.
- Create `Source/Applications/StreamHub/BinaryRecorder.cpp` — implementation (FileWriter header build, native encode, row serialization, double-buffer, file rotation/prune/fsync, arm/disarm, info).
- Modify `Source/Applications/StreamHub/UDPSourceSession.{h,cpp}` — own a `BinaryRecorder`, build include-mask + layout on CONFIG and on WS-spec epoch change, call `CapturePacket` from `ParseDataPayload`, expose `RecorderFlushTick`/`RequestRecArm`/`RequestRecDisarm`/`SetRecorderConfig`/`SetRecorderSignals`/`GetRecorderInfo`.
- Modify `Source/Applications/StreamHub/StreamHub.{h,cpp}` — parse `+Recorder`, store `RecorderConfig`, pass to sessions, call `RecorderFlushTick` in the push loop, add `recStart`/`recStop`/`recInfo` WS handlers + `recStatus` broadcast.
- Modify `Source/Applications/StreamHub/Makefile.inc` — add `BinaryRecorder.x` to `OBJSX`.
- Create `Test/Applications/StreamHub/BinaryRecorderSrc.cpp``#include "../../../Source/Applications/StreamHub/BinaryRecorder.cpp"`.
- Create `Test/Applications/StreamHub/BinaryRecorderGTest.cpp` — unit tests.
- Modify `Test/Applications/StreamHub/Makefile.inc` — add `BinaryRecorderSrc.x BinaryRecorderGTest.x` to `OBJSX`.
- Modify `run_streamhub.sh` — add a `+Recorder` block to the config heredoc.
- Create `Test/E2E/run_recorder_e2e.sh` (or extend `Test/E2E/datasources/run_e2e_report.sh`) — recorder E2E scenario validated with `validate_binary.py`.
## Key API (defined here, consumed by later tasks)
```cpp
// BinaryRecorder.h
namespace StreamHub {
struct RecorderConfig {
bool enabled; // master enable
bool autoStart; // arm at startup
char directory[512]; // output dir
uint32 maxFileBytes; // roll threshold (MaxFileMB * 1024*1024)
uint32 keepFiles; // newest N kept per source
uint32 stagingBytes; // per-buffer hard cap (StagingMB * 1024*1024)
uint32 flushIntervalSec; // fdatasync cadence
uint32 minDiskFreeMB; // stop-writing guard
char signals[1024]; // "all" or comma-separated "src:sig" keys
};
class BinaryRecorder {
public:
BinaryRecorder();
~BinaryRecorder();
void Init(const RecorderConfig &cfg, const char *sourceId); // once, before use
// Receive thread: (re)build header + per-signal row layout for included signals.
void Configure(const MARTe::UDPSSignalDescriptor *descs, uint32 nSigs,
const bool *includeMask);
// Receive thread: append one packet's rows to the staging buffer.
void CapturePacket(const uint8 *payload, const uint32 *sigOff,
const uint32 *sigElems, uint8 publishMode,
uint32 numSamples);
// Push thread: open pending file, swap+flush staging, rotate, periodic fsync.
void FlushTick(uint32 nowSec);
void RequestArm(); // any thread
void RequestDisarm(); // any thread
bool IsEnabled() const;
// recInfo snapshot.
void GetInfo(bool &recording, char *file, uint32 fileSz,
uint64 &bytesWritten, uint64 &rowsWritten,
uint64 &droppedRows, uint64 &freeMB) const;
static uint16 TypeCodeToDescriptorAll(uint8 udpsTypeCode); // mapping
static uint32 EncodeNative(uint8 typeCode, float64 v, uint8 *dst);
};
}
```
```cpp
// UDPSourceSession additions (consumed by StreamHub)
void SetRecorderConfig(const RecorderConfig &cfg); // before Start
void SetRecorderSignals(const char *spec); // WS thread; epoch-deferred
void RequestRecArm();
void RequestRecDisarm();
void RecorderFlushTick(uint32 nowSec); // push thread
void GetRecorderInfo(...) const; // push thread (recInfo)
```
---
### Task 1: BinaryRecorder header + type mapping + native encode + FileWriter header bytes
**Files:**
- Create: `Source/Applications/StreamHub/BinaryRecorder.h`
- Create: `Source/Applications/StreamHub/BinaryRecorder.cpp`
- Create: `Test/Applications/StreamHub/BinaryRecorderSrc.cpp`
- Create: `Test/Applications/StreamHub/BinaryRecorderGTest.cpp`
- Modify: `Test/Applications/StreamHub/Makefile.inc` (OBJSX += `BinaryRecorderSrc.x BinaryRecorderGTest.x`)
**Interfaces:**
- Produces: `RecorderConfig`, `BinaryRecorder::TypeCodeToDescriptorAll`, `BinaryRecorder::EncodeNative`, `Init`, `Configure`, and a header-on-disk produced by the first `FlushTick` after arm.
- [ ] **Step 1: Write failing tests** for `TypeCodeToDescriptorAll` (UINT32→`UnsignedInteger32Bit.all`, FLOAT32→`Float32Bit.all`, FLOAT64→`Float64Bit.all`, UINT64→`UnsignedInteger64Bit.all`) and `EncodeNative` (FLOAT32 of 1.5 → 4 bytes equal to `float 1.5f`; UINT32 of 7.0 → `uint32 7`; INT16 of -3 → `int16 -3`).
- [ ] **Step 2: Run, verify fail** (`function not defined`).
- [ ] **Step 3: Implement** the two static helpers in `BinaryRecorder.cpp` using MARTe2 type constants (`UnsignedInteger32Bit.all` etc.) and a `switch` on typeCode for `EncodeNative` (mirror `DecodeRawValue`'s type list, casting `float64 v` to the native type then `memcpy`). Add minimal class skeleton + `RecorderConfig`.
- [ ] **Step 4: Run, verify pass.**
- [ ] **Step 5: Add header-write test:** `Init` with a temp dir + `autoStart=true`; `Configure` with 2 signals (e.g. `Sine` float32 numElements=1, `Time` uint32 numElements=4) and `includeMask={true,true}`; `FlushTick(0)` (opens file, writes header). Read the file's first bytes; assert `numSigs==2`, descriptor 0 = `[u16 Float32Bit.all][32B "Sine"][u32 1]`, descriptor 1 = `[u16 UnsignedInteger32Bit.all][32B "Time"][u32 4]`.
- [ ] **Step 6: Run, verify fail.**
- [ ] **Step 7: Implement** `Init` (copy cfg, sourceId, `mkdir` directory), `Configure` (store descs+mask, build header blob into a member buffer: numSigs of included, per-signal u16 type + 32B name + u32 numElements; compute `rowBytes`), and the file-open path inside `FlushTick` (open `<dir>/<sourceId>_<UTC>.bin`, `pwrite` header, set `fileOffset_=headerLen`). Arm state from `autoStart`.
- [ ] **Step 8: Run, verify pass.**
- [ ] **Step 9: Commit** (`feat(streamhub): BinaryRecorder header + native encode`).
### Task 2: Row serialization — subset, native copy, quantized reconstruction, ACCUMULATE expansion
**Files:**
- Modify: `Source/Applications/StreamHub/BinaryRecorder.cpp` (`CapturePacket`)
- Modify: `Test/Applications/StreamHub/BinaryRecorderGTest.cpp`
**Interfaces:**
- Consumes: `Configure` layout from Task 1.
- Produces: staging rows flushed to disk by `FlushTick`.
- [ ] **Step 1: Write failing test (STRICT, un-quantized, all signals):** Configure 2 signals `A`(float32,1 elem), `B`(uint32,2 elem). Build a raw payload of one cycle: `A=2.5f`, `B={10,20}`. `CapturePacket(payload, sigOff={0,4}, sigElems={1,2}, publishMode=STRICT, numSamples=1)`; `FlushTick`. Read file rows; assert exactly 1 row of `rowBytes = 4 + 8 = 12`; bytes equal `float 2.5f` then `uint32 10,20`.
- [ ] **Step 2: Run, verify fail.**
- [ ] **Step 3: Implement** `CapturePacket`: `numRows = (publishMode==ACCUMULATE)?numSamples:1`. For each row r, for each included signal s: if `ACCUMULATE && declaredElems[s]==1` use wire element index r (1 native elem); else copy all `declaredElems[s]` elems (repeated each row). Per element: if `quant==NONE` `memcpy` wire bytes (wire size == native size); else decode quant int → `Dequantize` (replicate the formulas from `UDPSourceSession::DequantizeValue`) → `EncodeNative`. Append into staging `front` under the swap mutex; enforce overflow guard (drop + `droppedRows_++` if `frontLen + rowSpan > stagingBytes`).
- [ ] **Step 4: Run, verify pass.**
- [ ] **Step 5: Add ACCUMULATE test:** scalar `S`(float32, 1 declared elem) accumulated, `numSamples=3`, wire `{1.0f,2.0f,3.0f}`; non-accumulated `C`(uint32,1) value 9. `CapturePacket(..., sigElems={3,1}, publishMode=ACCUMULATE, numSamples=3)`; assert 3 rows, each `[float Sr][uint32 9]` with `S0=1,S1=2,S2=3`.
- [ ] **Step 6: Run, verify fail, implement (if needed), verify pass.**
- [ ] **Step 7: Add subset test:** Configure 3 signals, `includeMask={true,false,true}`; assert header has 2 signals and rows contain only signals 0 and 2 in order.
- [ ] **Step 8: Add quantized test:** signal with `quantType=UDPS_QUANT_UINT16`, `rangeMin=0,rangeMax=10`, wire `uint16 32767`; assert stored float == `0 + (32767/65535)*10` (within 1e-9) encoded as the native type.
- [ ] **Step 9: Run all, verify pass.**
- [ ] **Step 10: Commit** (`feat(streamhub): BinaryRecorder row serialization`).
### Task 3: Push-thread flush — double-buffer swap, pwrite, rotation, prune, fsync, disk guard
**Files:**
- Modify: `Source/Applications/StreamHub/BinaryRecorder.cpp` (`FlushTick`)
- Modify: `Test/Applications/StreamHub/BinaryRecorderGTest.cpp`
**Interfaces:**
- Consumes: staging from Task 2, header blob from Task 1.
- Produces: rotated `.bin` files on disk; `GetInfo` counters.
- [ ] **Step 1: Write failing rotation test:** `maxFileBytes` small (e.g. 200), `keepFiles=2`, row 12 B; capture+flush ~50 rows across several ticks; assert ≤2 `.bin` files remain in the dir, each begins with a valid header, and total data rows across surviving files ≤ capacity but newest rows present.
- [ ] **Step 2: Run, verify fail.**
- [ ] **Step 3: Implement** `FlushTick`: under mutex swap `front`/`back`, read `backLen`, reset `frontLen=0`, copy pending-arm/disarm + pendingReopen + headerBlob; unlock. Apply arm/disarm. If reopen or armed-with-no-file: open new file (write header), reset offset. If armed && backLen: `pwrite(back, backLen, fileOffset_)`, `fileOffset_+=backLen`, `bytesWritten_+=backLen`. If `fileOffset_ >= maxFileBytes`: close, open new file (header), then prune: list `<sourceId>_*.bin` sorted by name (UTC timestamp ⇒ lexicographic = chronological), `unlink` oldest until ≤ `keepFiles`. Every `flushIntervalSec`: `fdatasync(fd)`. Periodic `statvfs`: if free < `minDiskFreeMB` → disarm + log.
- [ ] **Step 4: Run, verify pass.**
- [ ] **Step 5: Add disk-guard test** (set `minDiskFreeMB` huge → first `FlushTick` disarms, no file written / writing stops); **and overflow test** (tiny `stagingBytes`, capture without flushing until guard trips → `GetInfo` `droppedRows>0`).
- [ ] **Step 6: Run, verify pass.**
- [ ] **Step 7: Commit** (`feat(streamhub): BinaryRecorder flush, rotation, guards`).
### Task 4: UDPSourceSession integration
**Files:**
- Modify: `Source/Applications/StreamHub/UDPSourceSession.h` (member `BinaryRecorder recorder_;`, spec/epoch members, new methods)
- Modify: `Source/Applications/StreamHub/UDPSourceSession.cpp`
**Interfaces:**
- Consumes: `BinaryRecorder` API.
- Produces: `SetRecorderConfig`, `SetRecorderSignals`, `RequestRecArm/Disarm`, `RecorderFlushTick`, `GetRecorderInfo` for StreamHub.
- [ ] **Step 1:** Add `#include "BinaryRecorder.h"`, member `BinaryRecorder recorder_;`, `RecorderConfig recCfg_;`, `bool recConfigured_;`, `FastPollingMutexSem recSpecMutex_;`, `char recPendingSpec_[1024];`, `volatile uint32 recPendingEpoch_; uint32 recSeenEpoch_;`. Add the new public methods. Initialize in ctor.
- [ ] **Step 2:** `SetRecorderConfig` (store cfg, `recorder_.Init(cfg, id_.Buffer())`, copy `cfg.signals` into pending spec, bump epoch). `SetRecorderSignals` (lock `recSpecMutex_`, copy spec, bump `recPendingEpoch_`).
- [ ] **Step 3:** Add private `ComputeIncludeMask(const UDPSSignalDescriptor *descs, uint32 n, const char *spec, bool *maskOut)`: if spec=="all" all true; else for each signal build key `"<id_>:<name>"` and set true iff present as a comma-token in spec.
- [ ] **Step 4:** In `ParseConfigPayload`, after descriptors parsed: if `recCfg_.enabled` → compute mask from active spec and `recorder_.Configure(descs, n, mask)` (force `recSeenEpoch_ = recPendingEpoch_` so a CONFIG also adopts any pending spec).
- [ ] **Step 5:** In `ParseDataPayload`, after the pass-1 `sigOff/sigElems` loop and before/after the decode loop: if `recCfg_.enabled`: if `recPendingEpoch_ != recSeenEpoch_` → lock, copy pending spec to active, recompute mask, `recorder_.Configure(...)`, set `recSeenEpoch_`. Then `recorder_.CapturePacket(payload, sigOff, sigElems, pm, numSamples)`.
- [ ] **Step 6:** Implement `RequestRecArm`/`RequestRecDisarm` → forward to recorder; `RecorderFlushTick(nowSec)``recorder_.FlushTick(nowSec)`; `GetRecorderInfo``recorder_.GetInfo`.
- [ ] **Step 7:** Build the app (`make -C Source/Applications/StreamHub -f Makefile.gcc`); fix compile errors. (No new unit test here — behavior covered by Tasks 13 and E2E.)
- [ ] **Step 8: Commit** (`feat(streamhub): wire BinaryRecorder into UDPSourceSession`).
### Task 5: StreamHub integration — config parse, push-loop flush, WS commands
**Files:**
- Modify: `Source/Applications/StreamHub/StreamHub.h` (`RecorderConfig recorderCfg_;` + handler decls)
- Modify: `Source/Applications/StreamHub/StreamHub.cpp`
**Interfaces:**
- Consumes: `UDPSourceSession` recorder methods.
- [ ] **Step 1:** In `Initialise`, after the `+History` block, parse `+Recorder`/`Recorder`: read `Enabled, AutoStart, Directory, MaxFileMB, KeepFiles, StagingMB, FlushIntervalSec, MinDiskFreeMB, Signals` into `recorderCfg_` (defaults: enabled=0, autoStart=1, MaxFileMB=256, KeepFiles=8, StagingMB=8, FlushIntervalSec=5, MinDiskFreeMB=500, Signals="all"). Convert MB→bytes.
- [ ] **Step 2:** Before `sessions_[i].Start()` in both the static-`Sources` loop and `AddSourceInternal`, call `sessions_[i].SetRecorderConfig(recorderCfg_)` when `recorderCfg_.enabled`.
- [ ] **Step 3:** In `PushData`, after `history_.WriteTick`, add `if (recorderCfg_.enabled) sessions_[i].RecorderFlushTick(nowSec);` (compute `nowSec` once per tick from `CLOCK_REALTIME`).
- [ ] **Step 4:** Add WS handlers + dispatch entries: `recStart` (optional `signals``SetRecorderSignals` on each active session, then `RequestRecArm`), `recStop` (`RequestRecDisarm` each), `recInfo` (build JSON from `GetRecorderInfo` per active session, unicast via `wsServer_.SendText(slotIdx,...)`). Add `recStatus` broadcast helper (called on start/stop). Parsing the `signals` array: accept either `"all"` string or a JSON array of `"src:sig"` — reuse `JsonGetString`; for an array, pass the raw substring and let `ComputeIncludeMask` tokenize (strip brackets/quotes).
- [ ] **Step 5:** Build app; fix errors. Manual smoke: run with a `+Recorder` block, confirm a `.bin` appears and grows.
- [ ] **Step 6: Commit** (`feat(streamhub): +Recorder config + recStart/recStop/recInfo WS`).
### Task 6: Build wiring + run script
**Files:**
- Modify: `Source/Applications/StreamHub/Makefile.inc` (OBJSX += `BinaryRecorder.x`)
- Modify: `run_streamhub.sh` (add `+Recorder` block to the config heredoc, default disabled or pointed at a temp dir)
- [ ] **Step 1:** Add `BinaryRecorder.x` to app `OBJSX`. Rebuild app cleanly.
- [ ] **Step 2:** Add a commented `+Recorder` block to `run_streamhub.sh`'s generated cfg (Directory under the run's work dir; `Enabled=0` by default with a note).
- [ ] **Step 3:** Run full unit suite: `make -f Makefile.gcc test && ./Build/x86-linux/GTest/MainGTest.ex`. All BinaryRecorder tests + the existing 38 pass.
- [ ] **Step 4: Commit** (`build(streamhub): wire BinaryRecorder + run script cfg`).
### Task 7: E2E recorder scenario + report
**Files:**
- Create: `Test/E2E/run_recorder_e2e.sh`
- Modify (optional): `Test/E2E/datasources/E2E_Report.typ` (add a recorder result section)
- [ ] **Step 1:** Write `run_recorder_e2e.sh`: build core+apps; launch a UDPStreamer source (reuse the existing E2E source/config or `Test/Configurations`); run StreamHub with `+Recorder{Enabled=1;AutoStart=1;Directory=<out>;Signals="all"}` for a few seconds; stop; locate the produced `<out>/<id>_*.bin`.
- [ ] **Step 2:** Validate the produced file with `Test/E2E/validate_binary.py` against the UDPStreamer's own FileWriter capture (or against a known-good reference), emitting `--json` into the build output dir like the existing datasources E2E.
- [ ] **Step 3:** Run it; confirm PASS (matching rows, 0 mismatching). Persist results into the repo's E2E output dir as the existing scripts do.
- [ ] **Step 4: Commit** (`test(streamhub): recorder E2E validation`).
---
## Self-Review
- **Spec coverage:** ownership (T4), FileWriter format + native types + type/name fields (T1), subset all/list + WS override (T2,T4,T5), rotation size-cap + keep-N + prune (T3), MinDiskFree guard (T3), staging double-buffer + overflow guard (T2,T3), packet-decode capture + ACCUMULATE rows (T2,T4), push-thread flush (T3,T5), config `+Recorder` (T5), WS recStart/recStop/recInfo + recStatus (T5), unit tests (T1T3), E2E (T7). All spec sections map to tasks.
- **Threading note:** layout + capture both run on the receive thread (no layout race); WS spec override is adopted via the receive-thread epoch check (mirrors the existing trigger-epoch pattern at `UDPSourceSession.cpp:293`); file I/O + arm/disarm + rotation all run on the push thread via `FlushTick`; only staging buffers, header blob, control flags, and the spec string cross threads (guarded by `FastPollingMutexSem`).
- **Quantization caveat:** byte-identical only holds for `quant==NONE`; quantized signals store the dequantized reconstruction (lossy). E2E uses un-quantized FileWriter input, so byte-match holds there.
- **Type consistency:** `RecorderConfig`, `Configure(descs,n,mask)`, `CapturePacket(payload,sigOff,sigElems,pm,numSamples)`, `FlushTick(nowSec)` names are used consistently across T1T5.