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

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 #included 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)

// 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);
};
}

// 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 1–3 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 (T1–T3), 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 T1–T5.