# Resequencing and Relative Timing 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:** Recover out-of-order UDP packets instead of discarding them — resequence inside a 20 ms window, and reserve an invalidated hole that a later arrival backfills — and add a relative-time mode whose x comes from the packet counter and the declared rate rather than from a jittery stream clock. **Architecture:** The client gains a bounded reorder ring of *reassembled raw payloads* released in counter order; a packet past the window is delivered flagged late instead of dropped. `Acquisition` reserves a NaN hole of `lost × batch` samples by running them through its existing per-sample loop, so decimation, stamping and the LOD pyramid stay consistent for free, and backfills values through a new amend path. Relative timing replaces `build_stamps()`'s inferred clock with `(slot_base_ + k) / rate`. **Tech Stack:** C++20 (udpscope, `deps/plotter`), C99 (`deps/udpsclient`), GTK4, CMake + Ninja, Python 3 (`tools/mock_udpstreamer.py`). **Spec:** `docs/superpowers/specs/2026-10-08-resequencing-and-relative-timing-design.md` ## Global Constraints - **Every new source file carries the SPDX MIT header** — `// SPDX-License-Identifier: MIT` then `// SPDX-FileCopyrightText: 2026 Martino Ferrari` (`/** */` or `#` comment form in C and Python). - **Two build-warning baselines, read on every build:** exactly **four** warnings on a clean build, **two** on an incremental build. More than that is a regression to fix, not to note. - **Zero `Gtk-CRITICAL`** in any headless run. One is a defect. - **No unit test suite in udpscope.** udpscope changes are verified by the application's own headless self-check (env-var probes under `xvfb-run` against `tools/mock_udpstreamer.py`). `deps/plotter` *does* have a test suite and every library change gets a real test there. - **Verification runs are short and serial.** Batching several `xvfb` runs back to back starves the mock past its 2 s `silence_timeout_s` and reads as a reconnect — a measurement artefact, not a defect. Kill leftover mocks between runs, and put `pkill -f mock_udpstreamer.py` inside a script file rather than typing it in the invoking shell (the pattern matches the invoking shell's own argv and the shell kills itself, exit code 144). - **The latency budget is 20 ms**, as `UDPS_REORDER_WINDOW_S_DEFAULT = 0.020`. A contiguous arrival is released with **zero** added latency. - **The reservation cap is 256 missing packets**, reusing the existing `kGlitchPackets` constant and its reasoning. - **`MinMaxPyramid::amend()`'s precondition is non-finite → finite only.** It is additive, not a rewrite. ### Build and run commands ```bash # udpscope cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo cmake --build build # deps/plotter test suite cmake --build deps/plotter/build/debug xvfb-run -a deps/plotter/build/debug/tests/plot_tests # expect "0 failed" PLOT_TEST_FILTER=amend deps/plotter/build/debug/tests/plot_tests # one test, no EGL needed ``` ### File structure | File | Responsibility | Phase | | --- | --- | --- | | `src/config.hpp`, `src/config.cpp` | `--relative-time` flag | 1 | | `src/acquisition.hpp`, `src/acquisition.cpp` | relative stamps, hole reservation, backfill dispatch | 1, 3, 4 | | `src/model.hpp`, `src/model.cpp` | absolute-time ring, `StreamStats` counters, invalid-span table, amend queue drain | 1, 3, 4 | | `src/channel.hpp`, `src/channel.cpp` | `amend_published()` — applies a queued pyramid amend inside `acquire()` | 4 | | `deps/udpsclient/udps_client.h`, `.c` | `reorder_window_s`, the reorder ring, late delivery, two new stats | 2 | | `deps/plotter/include/plot/data/ring_buffer.hpp` | `amend()` | 4 | | `deps/plotter/include/plot/data/minmax_pyramid.hpp`, `src/data/minmax_pyramid.cpp` | `amend()` | 4 | | `deps/plotter/include/plot/plot.hpp`, `src/render/renderer.cpp` | `set_invalid_spans()` and the band | 5 | | `src/status_text.cpp` | the four new counters in the detail tier | 3 | | `src/ui.cpp` | exit dump, `UDPSCOPE_PROBE_RESEQ_MS` | 3 | | `tools/mock_udpstreamer.py` | `--reorder-delay-ms` | 2 | ### Correction to the spec, carried into this plan The spec says the backfill queue is an SPSC queue. It need not be: `ScopeModel::acquire_all()` is called from `src/ui.cpp:6265` **inside `std::lock_guard lock(model.mutex())`**, and the producer takes that same mutex for the whole of `Acquisition::on_data` (`src/acquisition.cpp:386`). A plain `std::vector` guarded by that mutex is correct and simpler. Task 11 fixes the spec line. --- ## Phase 1 — Relative timing (independently shippable) ### Task 1: The `--relative-time` flag **Files:** - Modify: `src/config.hpp:37` (add the field), `src/config.cpp:55-57` (usage), `src/config.cpp:94-100` (parse) - Test: headless run of `--help` **Interfaces:** - Produces: `Config::relative_time` (`bool`, default `false`), read by Task 2. - [ ] **Step 1: Add the field to `Config`** In `src/config.hpp`, after the `plots` member: ```cpp /// Take the plot's time from the packet counter and the declared sample rate rather than /// from the stream's own time signal, so arrival and producer-clock jitter cannot move a /// sample's x. Absolute stream time is still recorded, for export only (`todo.md` item 2). bool relative_time = false; ``` - [ ] **Step 2: Add the usage line** In `src/config.cpp:print_usage`, between the `--plots` block and the `-h, --help` line: ```c " --relative-time take the plot's time from the packet counter and the\n" " declared rate, not the stream's time signal; absolute\n" " time is then kept for export only\n" ``` - [ ] **Step 3: Add the parse branch** In `src/config.cpp:parse_args`, after the `--plots` branch and before the `else { fail(...) }`: ```cpp } else if (a == "--relative-time") { c.relative_time = true; ``` - [ ] **Step 4: Build and read the help** ```bash cmake --build build && ./build/udpscope --help | grep -A2 relative-time ``` Expected: the three usage lines above. Warning count unchanged (two on this incremental build). - [ ] **Step 5: Commit** ```bash git add src/config.hpp src/config.cpp git commit -m "feat: --relative-time selects the counter-derived clock" ``` --- ### Task 2: Counter-derived stamps **Files:** - Modify: `src/acquisition.hpp:92-93` (state), `src/acquisition.cpp:128-382` (`build_stamps`) - Test: `UDPSCOPE_PROBE_RESEQ_MS` is Task 8's; this task is verified by the existing exit dump plus a one-off stderr line. **Interfaces:** - Consumes: `Config::relative_time` (Task 1). - Produces: `Acquisition::slot_base_` (`std::uint64_t`), `Acquisition::batch_` (`std::uint32_t`, 0 until the first frame), `Acquisition::have_counter_` (`bool`), `Acquisition::last_counter_` (`std::uint32_t`). Task 6 reads `batch_` to size a hole and `slot_base_` is advanced there. - [ ] **Step 1: Add the state** In `src/acquisition.hpp`, after `int time_source_ = 0;`: ```cpp /// Relative-time mode's position counter: samples seen since the stream began, which is /// what the counter names. Advanced by `batch_` for every packet the counter reports /// missing (before the frame's own samples) and then by the frame's own `n`, so it is /// exactly `counter * batch + sample_index` for a constant batch and degrades gracefully /// when a frame's length disagrees. std::uint64_t slot_base_ = 0; /// Samples per DATA packet, fixed by the first frame. Only ever needed to size what is /// *absent*: a delivered frame carries its own length. 0 until the first frame. std::uint32_t batch_ = 0; /// The previous frame's counter, for the signed-delta ordering the client also uses. std::uint32_t last_counter_ = 0; bool have_counter_ = false; ``` - [ ] **Step 2: Branch `build_stamps()` on the mode** In `src/acquisition.cpp`, immediately after `stamps_.assign(n, 0.0);` (line 137) and before the `double period = ...` block, insert: ```cpp // Relative mode: the stamp is a pure function of position and the *declared* rate. Nothing // below this block runs -- not the arrival-cadence override, not the degeneracy guard, not // the glitch bridge, not the monotonic clamp's reason for existing -- because a position // cannot jitter, cannot fail to advance and cannot jump. if (cfg_.relative_time) { if (batch_ == 0) { batch_ = n; std::fprintf(stderr, "[udpscope] relative time: batch=%u samples/packet, rate=%.6g S/s declared\n", batch_, model_.partition().base_dt > 0.0 ? 1.0 / model_.partition().base_dt : 0.0); std::fflush(stderr); } else if (n != batch_) { // A frame whose length disagrees with the first one is an inconsistency in the // producer, not a loss: the counter still names one packet. Warn once and keep the // established batch, which is only used to size holes -- the timeline itself is built // from the lengths actually seen, so it stays exact for this frame either way. static bool warned_batch = false; if (!warned_batch) { warned_batch = true; std::fprintf(stderr, "[udpscope] warning: packet carries %u samples but the stream's first " "carried %u; keeping %u as the packet length a lost packet is worth.\n", n, batch_, batch_); std::fflush(stderr); } } double dt = model_.partition().base_dt; if (!(dt > 0.0)) dt = 1.0e-3; for (std::uint32_t k = 0; k < n; ++k) { stamps_[k] = double(slot_base_ + k) * dt; } slot_base_ += n; return n; } ``` - [ ] **Step 3: Build and run against the mock, clean stream** ```bash cat > /tmp/udps_rel.sh <<'EOF' #!/bin/bash cd /home/martino/Projects/tools/udpscope pkill -f mock_udpstreamer.py; sleep 0.5 nohup python3 tools/mock_udpstreamer.py --port 8130 --rate 100000 --batch 100 --freq 5 \ > /tmp/mock_rel.log 2>&1 & m=$!; sleep 1 xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=8000 UDPSCOPE_ENABLE_ALL_MS=600 UDPSCOPE_EXIT_DUMP=1 \ ./build/udpscope --server 127.0.0.1 --port 8130 --relative-time > /tmp/fault_rel.log 2>&1 kill $m 2>/dev/null grep -E "^\[udpscope\] (frames=|relative time:)" /tmp/fault_rel.log echo "criticals=$(grep -c Gtk-CRITICAL /tmp/fault_rel.log) warnings=$(grep -c 'warning:' /tmp/fault_rel.log)" EOF chmod +x /tmp/udps_rel.sh && /tmp/udps_rel.sh ``` Expected: a `relative time: batch=100 samples/packet` line, `frames=` near 1029 with `gaps=0`, `criticals=0 warnings=0`. A `does not advance across packets` or `jumped implausibly far ahead` warning means the branch was placed after the guard instead of before it. - [ ] **Step 4: Confirm the mode is actually off by default** ```bash sed -i 's/--relative-time //' /tmp/udps_rel.sh && /tmp/udps_rel.sh ``` Expected: no `relative time:` line at all, `frames=` and `gaps=0` as before. Then restore the flag in the script. - [ ] **Step 5: Commit** ```bash git add src/acquisition.hpp src/acquisition.cpp git commit -m "feat: relative mode stamps a sample from its position and the declared rate" ``` --- ### Task 3: An absolute-time ring for export **Files:** - Modify: `src/model.hpp:118-152` (`BufferStore`), `src/model.cpp:140-158`, `src/acquisition.cpp:562-568` (the time push) - Test: an export in relative mode carries absolute stamps **Interfaces:** - Produces: `BufferStore::abs_time()` → `plot::data::RingBuffer&`, same capacity and same group-mean cadence as `time()`. Written only in relative mode; read only by the export path. - [ ] **Step 1: Add the ring to `BufferStore`** In `src/model.hpp`, after the `time()` accessors: ```cpp /// The stream's **absolute** clock, parallel to `time()` and on the same slots. /// /// Relative mode takes `time()` for the counter-derived timeline, and there is exactly one /// time ring per store shared by every channel -- so without this ring, selecting relative /// mode would silently drop absolute time from an export, which is the one output /// `todo.md` item 2 says must keep it. Written only in relative mode (in stream mode /// `time()` already *is* the absolute clock) and read only by the export. plot::data::RingBuffer& abs_time() noexcept { return *abs_time_; } const plot::data::RingBuffer& abs_time() const noexcept { return *abs_time_; } ``` and the member, after `time_`: ```cpp std::shared_ptr> abs_time_; ``` - [ ] **Step 2: Allocate it alongside the time ring** In `src/model.cpp`, in the `BufferStore` constructor, immediately after the line that constructs `time_`: ```cpp // Same capacity as the clock it parallels: the two are written on the same slots, so a // smaller ring would make an export's stamps run out before its values do. abs_time_ = std::make_shared>(part.time_capacity); ``` - [ ] **Step 3: Push the absolute stamp on the same slot** In `src/acquisition.cpp`, replace the time-push block at lines 562-568 with: ```cpp time_acc_ += rel; abs_time_acc_ += abs_stamps_[k] - epoch; ++time_acc_n_; if (tick % g == g - 1) { store.time().push(time_acc_ / static_cast(time_acc_n_)); // The absolute clock on the same slot and under the same group mean, so an export can // pair value, relative stamp and absolute stamp by index with no interpolation. store.abs_time().push(abs_time_acc_ / static_cast(time_acc_n_)); time_acc_ = 0.0; abs_time_acc_ = 0.0; time_acc_n_ = 0; } ``` Add `double abs_time_acc_ = 0.0;` next to `time_acc_` in `src/acquisition.hpp`, reset it in the re-plan block at `src/acquisition.cpp:443`, and add `std::vector abs_stamps_;` next to `stamps_`. - [ ] **Step 4: Fill `abs_stamps_`** In `build_stamps()`, `abs_stamps_` must always carry the stream's own clock regardless of mode. Change `stamps_.assign(n, 0.0);` (line 137) to: ```cpp stamps_.assign(n, 0.0); abs_stamps_.assign(n, 0.0); ``` In the relative branch added by Task 2, before `return n;`, fill it from the declared time signal when there is one and from the relative stamp otherwise: ```cpp // The absolute clock is still read, for export only. When the stream carries no usable // time signal there is no absolute clock to record and the relative stamp is the honest // answer -- an export then says what the counter says, which is all anyone had. if (time_signal >= 0 && time_signal < static_cast(frame->num_signals)) { const udps_signal_t& ts = frame->signals[time_signal]; const bool scalar_time = udps_signal_num_elements(&ts) == 1; for (std::uint32_t k = 0; k < n; ++k) { abs_stamps_[k] = scalar_time ? udps_frame_value(frame, static_cast(time_signal), k, 0) : udps_frame_value(frame, static_cast(time_signal), 0, k); } } else { for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k]; } ``` At the end of the stream-mode path, just before `return n;` at line 381, add: ```cpp // Stream mode: `time()` already holds the absolute clock, so the parallel ring repeats it // rather than carrying a second timeline. One rule for the export either way. for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k]; ``` - [ ] **Step 5: Make the export emit it** Find the export's column assembly and add an `abs_time` column, read from `store.abs_time()` on the same index as the relative stamp: ```bash grep -n "npy\|csv\|export" src/export.cpp | head -20 ``` The column is only added when it differs from the relative stamp — i.e. in relative mode — so a stream-mode export is byte-identical to today's. - [ ] **Step 6: Build, export in both modes, compare** ```bash cmake --build build cat > /tmp/udps_exp.sh <<'EOF' #!/bin/bash cd /home/martino/Projects/tools/udpscope run() { pkill -f mock_udpstreamer.py; sleep 0.5 nohup python3 tools/mock_udpstreamer.py --port 8131 --rate 100000 --batch 100 --freq 5 \ > /tmp/mock_exp.log 2>&1 & local m=$!; sleep 1 xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=6000 UDPSCOPE_ENABLE_ALL_MS=600 \ UDPSCOPE_EXPORT_PATH="/tmp/exp_$1.csv" \ ./build/udpscope --server 127.0.0.1 --port 8131 $2 > "/tmp/exp_$1.log" 2>&1 kill $m 2>/dev/null echo "=== $1"; head -2 "/tmp/exp_$1.csv" echo "criticals=$(grep -c Gtk-CRITICAL /tmp/exp_$1.log)" } run stream "" run rel "--relative-time" EOF chmod +x /tmp/udps_exp.sh && /tmp/udps_exp.sh ``` Expected: `exp_stream.csv`'s header is unchanged from before this task; `exp_rel.csv` carries an extra absolute-time column whose first value is a stream timestamp, not a small multiple of the sample period. `criticals=0` for both. - [ ] **Step 7: Commit** ```bash git add src/model.hpp src/model.cpp src/acquisition.hpp src/acquisition.cpp src/export.cpp git commit -m "feat: relative mode keeps the stream's absolute clock for the export" ``` --- ## Phase 2 — Client resequencing ### Task 4: A reorder ring of reassembled payloads **Files:** - Modify: `deps/udpsclient/udps_client.h:170-184` (stats), `:218-229` (config) - Modify: `deps/udpsclient/udps_client.c:982` (defaults), `:504-514` (the ordering), `:595-599` (the hook), `:1145-1152` (the deadline clamp), `:1222` (the flush) - Test: the mock's `--reorder-every`, measured through udpscope's `gaps` **Interfaces:** - Produces: `udps_client_config_t::reorder_window_s` (`double`, default 0.020; 0 disables), `udps_stats_t::late_resequenced` and `udps_stats_t::late_delivered` (`uint64_t`), and `udps_frame_t::late` (`uint8_t`, 1 when the frame's counter is behind the delivered sequence). Tasks 6 and 7 read all three. - [ ] **Step 1: Extend the public structs** In `deps/udpsclient/udps_client.h`, in `udps_stats_t` after `stale_packets`: ```c /** * DATA packets that arrived out of order and were put *back* in order inside * the reorder window, so the consumer never saw a gap for them. */ uint64_t late_resequenced; /** * DATA packets delivered behind the sequence because the reorder window had * already closed on them. The consumer has a hole for these and may fill it; * before the window existed they were counted in @c stale_packets and dropped. */ uint64_t late_delivered; ``` In `udps_client_config_t` after `max_packet_bytes`: ```c /** * How long a packet may be held waiting for an earlier counter, in seconds. * * A *contiguous* arrival is never held, so this is the latency a gap costs and * not a latency the stream pays. 0 disables resequencing entirely and restores * the drop-on-late behaviour. Default 0.020. */ double reorder_window_s; ``` In `udps_frame_t` after `lost`: ```c /** * 1 when this frame's counter is behind the sequence already delivered: it lost * the race past the reorder window and the consumer has a hole where it belongs. * @c lost is 0 for such a frame -- it is not missing anything, it *is* what was * missing -- so a consumer sizes its backfill from @c counter, not from @c lost. */ uint8_t late; ``` - [ ] **Step 2: Default the window** In `deps/udpsclient/udps_client.c:udps_client_config_init`, beside `cfg->recv_buffer_bytes`: ```c cfg->reorder_window_s = 0.020; ``` - [ ] **Step 3: Add the ring to the client struct** Near the client's other buffers: ```c /** Datagrams held waiting for an earlier counter. */ #define UDPS_REORDER_SLOTS 64 /** Total bytes the reorder ring may hold, so a fast stream cannot grow it without bound. */ #define UDPS_REORDER_BYTES (4u * 1024u * 1024u) typedef struct { uint8_t *buf; /**< owned; grown on demand to the payload's length */ size_t cap; size_t len; uint32_t counter; double recv_time; int used; } udps_held_t; ``` and in `struct udps_client`: ```c udps_held_t held[UDPS_REORDER_SLOTS]; size_t held_bytes; /**< payload bytes currently held */ int held_count; uint32_t next_counter; /**< the counter the sequence is waiting for */ double gap_deadline; /**< wall time the current gap gives up at; 0 = no gap */ ``` Free every `held[i].buf` in `udps_client_destroy`, and reset `held_count`, `held_bytes`, `gap_deadline` and `next_counter` wherever `slot_reset_all()` is called (a reconnect starts a new sequence). - [ ] **Step 4: Replace the ordering test with the reorder policy** `deps/udpsclient/udps_client.c:504-514` becomes a *delivery* decision taken before decode, in the wrapper at `:595-599`. Rename the existing `decode_data()` body's ordering block away and put this in the wrapper: ```c /* Orders the sequence before spending anything on the payload. * * The counter is a wrapping uint32, so it is ordered by the signed difference; * comparing the values directly would call the first packet after the wrap stale * and reject the stream from then on. * * Three cases. A packet that *continues* the sequence is decoded now and costs no * latency at all -- that is the common case and it must stay free. A packet that * runs ahead of the sequence means an earlier one is missing, so it is held and the * gap is given `reorder_window_s` to resolve; a reorder on the wire resolves in * milliseconds, and holding is what turns it back into order. A packet behind the * sequence lost that race: the consumer already has a hole where it belongs, so it * is delivered flagged rather than discarded, which is the whole point -- a dropped * packet cannot fill the hole it left. */ static int order_data(udps_client_t *c, const uint8_t *pl, size_t len, uint32_t counter, double recv_time) { int32_t delta; if (c->cfg.reorder_window_s <= 0.0) { return deliver_data(c, pl, len, counter, recv_time, 0); } if (!c->have_counter) { c->have_counter = 1; c->next_counter = counter + 1u; return deliver_data(c, pl, len, counter, recv_time, 0); } delta = (int32_t)(counter - c->next_counter); if (delta == 0) { int rc = deliver_data(c, pl, len, counter, recv_time, 0); c->next_counter = counter + 1u; if (rc == 0) rc = drain_held(c); return rc; } if (delta > 0) { if (hold_packet(c, pl, len, counter, recv_time) != 0) { /* No room: commit the gap now rather than growing. The counter is * visible, so an undersized window is a reading and not a mystery. */ return commit_gap(c); } if (c->gap_deadline == 0.0) { c->gap_deadline = recv_time + c->cfg.reorder_window_s; } return 0; } /* delta < 0: behind the sequence. */ c->stats.late_delivered++; return deliver_data(c, pl, len, counter, recv_time, 1); } ``` `deliver_data()` is the old `decode_data()` with its ordering block removed, a `late` parameter, and `frame.late = late; frame.lost = late ? 0u : lost;`. `lost` is computed in `drain_held`/`commit_gap` as the number of counters skipped, and is 0 for a contiguous delivery. `hold_packet()` finds a free slot, grows its buffer to `len`, copies the payload, and refuses when `held_count == UDPS_REORDER_SLOTS` or `held_bytes + len > UDPS_REORDER_BYTES`. `drain_held()` repeatedly delivers the held entry whose counter equals `next_counter`, incrementing `stats.late_resequenced` for each, until none matches; it clears `gap_deadline` when `held_count` reaches 0. `commit_gap()` finds the lowest held counter, sets `lost = (uint32_t)(lowest - next_counter)`, advances `next_counter = lowest`, and calls `drain_held()`. `gap_deadline` is cleared by that drain. - [ ] **Step 5: Serve the deadline** In `udps_client_poll`, beside the existing clamps at `:1145-1152`: ```c /* A held packet must flush on time even when nothing further arrives: at 10 pkt/s * the next arrival is 100 ms away, which is five times the whole budget. */ if (c->gap_deadline > 0.0) { double left = c->gap_deadline - now_wall(); if (left < 0.0) left = 0.0; if (left < budget) budget = left; } ``` and after `gc_slots(c, now_wall());` at `:1222`: ```c if (c->gap_deadline > 0.0 && now_wall() >= c->gap_deadline) { (void)commit_gap(c); } ``` - [ ] **Step 6: Build and verify a reorder no longer costs a gap** ```bash cmake --build build cat > /tmp/udps_reseq.sh <<'EOF' #!/bin/bash cd /home/martino/Projects/tools/udpscope run() { local tag="$1"; shift pkill -f mock_udpstreamer.py; sleep 0.5 nohup python3 tools/mock_udpstreamer.py --port 8132 --rate 100000 --batch 100 --freq 5 \ "$@" > "/tmp/mock_$tag.log" 2>&1 & local m=$!; sleep 1 xvfb-run -a env UDPSCOPE_EXIT_AFTER_MS=8000 UDPSCOPE_ENABLE_ALL_MS=600 UDPSCOPE_EXIT_DUMP=1 \ ./build/udpscope --server 127.0.0.1 --port 8132 > "/tmp/fault_$tag.log" 2>&1 kill $m 2>/dev/null echo "=== $tag : $*" grep -E "^\[udpscope\] frames=" "/tmp/fault_$tag.log" echo "criticals=$(grep -c Gtk-CRITICAL /tmp/fault_$tag.log)" } run clean EOF chmod +x /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected for `clean`: `frames=` near 1029, `gaps=0`, `criticals=0` — unchanged, proving the zero-latency path is still free. Then run the reorder case alone: ```bash sed -i 's/^run clean$/run reorder --reorder-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected: `gaps=0` where it previously measured **102**, and `frames` near 1029 rather than 925. That delta is the whole feature: 102 gaps became 0. - [ ] **Step 7: Verify the kill switch** Set `c.reorder_window_s = 0.0;` temporarily in `src/acquisition.cpp:start()`, rebuild, re-run the reorder case. Expected: `gaps=102`, `frames≈925` — exactly today's numbers. Revert the line. - [ ] **Step 8: Commit** ```bash git add deps/udpsclient/udps_client.h deps/udpsclient/udps_client.c git commit -m "feat: hold a packet that runs ahead of the sequence, and deliver a late one flagged" ``` --- ### Task 5: `--reorder-delay-ms` in the mock **Files:** - Modify: `tools/mock_udpstreamer.py` (argparse, the send policy, the `done` line) - Test: the option's own effect, read through udpscope's counters **Interfaces:** - Produces: `--reorder-delay-ms MS` — a withheld packet is released `MS` milliseconds later instead of immediately after the packet that overtook it. Task 8's probe needs a value past 20 ms to reach the backfill path. - [ ] **Step 1: Add the option** Beside `--reorder-every` in the argparse block: ```python p.add_argument( "--reorder-delay-ms", type=float, default=0.0, metavar="MS", help="hold a withheld packet this many milliseconds instead of releasing it " "immediately behind the packet that overtook it. Under the receiver's reorder " "window the packet is resequenced and no hole is ever seen; past it the hole is " "committed first and the packet has to backfill it -- which is the only way to " "reach the backfill path at all, since a 0 ms delay always wins the race.", ) ``` - [ ] **Step 2: Release on a deadline instead of on the next send** `held` becomes `(counter, pkt, release_at)`. Replace the `if held is not None:` release block in the `else:` branch with a time test evaluated every iteration, before the send: ```python if held is not None and now >= held[2]: # Out of order on the wire now. Within the receiver's reorder window this is put # back in order and costs nothing; past it the receiver has already committed a # hole and this packet is what fills it. sock.sendto(held[1], peer) inverted += 1 if not noted_reorder: noted_reorder = True print( "mock udpstreamer: sent packet %d before %d, releasing it %.1f ms late" % (counter, held[0], args.reorder_delay_ms), flush=True, ) held = None ``` and where the packet is withheld: ```python held = (counter, pkt, now + args.reorder_delay_ms / 1000.0) ``` - [ ] **Step 3: Check it compiles and sends** ```bash python3 -m py_compile tools/mock_udpstreamer.py && echo OK cat > /tmp/udps_mock.sh <<'EOF' #!/bin/bash cd /home/martino/Projects/tools/udpscope pkill -f mock_udpstreamer.py; sleep 0.5 timeout 4 python3 tools/mock_udpstreamer.py --port 8133 --rate 100000 --batch 100 \ --reorder-every 10 --reorder-delay-ms 50 2>&1 | grep -E "before|done" EOF chmod +x /tmp/udps_mock.sh && /tmp/udps_mock.sh ``` Expected: a `sent packet N before M, releasing it 50.0 ms late` line and a `done (... inverted)` line with a non-zero inverted count. (`timeout` makes the mock exit without a CONNECT, so a `done` line may be absent; the `before` line is the assertion.) - [ ] **Step 4: Commit** ```bash git add tools/mock_udpstreamer.py git commit -m "test: the mock can release a withheld packet past the reorder window" ``` --- ## Phase 3 — Hole reservation and surfacing ### Task 6: Reserve an invalidated hole **Files:** - Modify: `src/acquisition.cpp:384-416` (before the per-sample loop), `src/acquisition.cpp:553` (the loop) - Modify: `src/model.hpp` (the invalid-span table) - Test: the mock's `--drop-every`, read through the exit dump **Interfaces:** - Consumes: `udps_frame_t::lost`, `udps_frame_t::late`, `Acquisition::batch_` (Task 2), `kGlitchPackets`. - Produces: `ScopeModel::note_hole(f64 x_begin, f64 x_end, u64 slot_begin, u64 slot_count, std::uint32_t counter_first)` and `ScopeModel::holes()` → `const std::vector&` (under `mutex()`). Task 7 counts them, Task 10 addresses them, Task 11 draws them. - [ ] **Step 1: Add the hole table** In `src/model.hpp`, before `StreamStats`: ```cpp /// One reserved, invalidated span: samples the counter says exist and the stream has not /// delivered. Its stored slots hold NaN, so a reader needs no special case -- the table is /// what lets a *late* packet find the slots it belongs to, and what the band draws. struct Hole { f64 x_begin = 0.0; ///< stream seconds of the hole's first slot f64 x_end = 0.0; ///< stream seconds of the hole's last slot u64 slot_begin = 0; ///< absolute stored-slot index of the first slot u64 slot_count = 0; ///< stored slots the hole spans u64 tick_begin = 0; ///< absolute base tick of the hole's first raw sample std::uint32_t counter_first = 0; ///< counter of the first missing packet std::uint32_t packets = 0; ///< packets the hole is worth }; ``` and on `ScopeModel`, with the table capped so a pathological stream cannot grow it: ```cpp /// Records a reserved hole. Caller holds `mutex()` (the producer already does). The table /// is capped at `kMaxHoles`; past that the oldest is dropped, because an unbounded list of /// gaps is a leak and the oldest gap is the one furthest from being backfilled. static constexpr std::size_t kMaxHoles = 64; void note_hole(const Hole& h); const std::vector& holes() const noexcept { return holes_; } /// Drops holes whose slots the ring has already discarded. void forget_holes_before(u64 slot); ``` - [ ] **Step 2: Reserve before the frame's own samples** In `src/acquisition.cpp`, after `const u64 g = part.time_stride;` (line 433) and before the accumulator re-plan block, insert: ```cpp // The gap the counter reports, reserved as NaN before this frame's own samples are // stamped: the hole belongs *ahead* of the data that revealed it. Running the missing // samples through the same per-sample loop below is the whole trick -- `tick_` advances // so the decimation phase does not shift, the group mean lands on the mean stamp, and // `MinMaxPyramid::append` sees NaN, which its non-finite contract already stores as an // empty aggregate every reader treats as absent. // // Capped at `kGlitchPackets`: 256 packet spans in one go is an outage, not loss, and // reserving one would push the entire retained history out of the ring to make room for // samples that are not coming. const std::uint32_t missing = frame->late ? 0u : frame->lost; if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) { reserve_hole(missing, epoch, part, channels, store, g); } else if (missing > 0) { static bool warned_outage = false; if (!warned_outage) { warned_outage = true; std::fprintf(stderr, "[udpscope] warning: %u packets missing in one go, past the %g-packet " "reservation cap; treating it as an outage and carrying on rather than " "reserving %llu samples of hole.\n", missing, kGlitchPackets, static_cast(std::uint64_t(missing) * batch_)); std::fflush(stderr); } } ``` `kGlitchPackets` moves from a function-local `constexpr` in `build_stamps()` to a file-scope one in the anonymous namespace so both call sites read the same constant. - [ ] **Step 3: Write `reserve_hole()`** A new private member of `Acquisition`, factored out so `on_data` stays readable. It is a reduced copy of the per-sample loop that pushes NaN instead of a sample and skips the trigger and the math channels: ```cpp void Acquisition::reserve_hole(std::uint32_t missing, double epoch, const Partition& part, const std::vector& channels, BufferStore& store, u64 g) { const u64 raw = u64(missing) * u64(batch_); Hole h; h.tick_begin = tick_; h.slot_begin = store.values(0).produced(); h.counter_first = last_counter_ + 1u; h.packets = missing; double dt = part.base_dt; if (!(dt > 0.0)) dt = 1.0e-3; for (u64 j = 0; j < raw; ++j) { const u64 tick = tick_++; // The hole is stamped at the declared cadence from the last good stamp -- exact in // relative mode, and in stream mode the same continuation the glitch bridge makes. // It must be *stamped* rather than left short: the time ring is binary-searched by // `ChannelBuffer::neighbours()`, so it has to stay monotone and dense, and backfill // then changes values only and never a stamp. const double rel = last_stamp_ + dt; last_stamp_ = rel; if (j == 0) h.x_begin = rel; h.x_end = rel; time_acc_ += rel; abs_time_acc_ += rel; ++time_acc_n_; if (tick % g == g - 1) { store.time().push(time_acc_ / double(time_acc_n_)); store.abs_time().push(abs_time_acc_ / double(time_acc_n_)); time_acc_ = 0.0; abs_time_acc_ = 0.0; time_acc_n_ = 0; } for (std::size_t i = 0; i < channels.size(); ++i) { if (channels[i].math) continue; // A NaN in the accumulator makes the whole group NaN, which is why a slot straddling // a hole edge is NaN too: its real members were averaged into a sum that is now // non-finite and were never retained, so the backfill cannot recover it and must not // pretend to. One slot per edge, against a hole thousands of slots wide. value_acc_[i] += std::numeric_limits::quiet_NaN(); ++value_acc_n_[i]; const u64 period = g * part.stride[i]; if (tick % period != period - 1) continue; store.values(i).push(static_cast(value_acc_[i] / double(value_acc_n_[i]))); value_acc_[i] = 0.0; value_acc_n_[i] = 0; } // The math channels are fed NaN like everything else, so a derived trace keeps the gap // its source had. The trigger is *not* fed: a NaN comparison is false, so the engine // could not latch here anyway, and feeding it would advance its hold-off clock over // time nothing was sampled in. if (tick % g == g - 1) { std::fill(math_sample_.begin(), math_sample_.end(), std::nanf("")); model_.push_math_sample(math_sample_); } } h.slot_count = store.values(0).produced() - h.slot_begin; model_.note_hole(h); model_.stats().holes.fetch_add(1, std::memory_order_relaxed); model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed); } ``` Declare it in `src/acquisition.hpp` and add `#include `. - [ ] **Step 4: Track the counter** At the top of `on_data`, after the `running_` check: ```cpp // The counter is this frame's identity and the hole's address. It is tracked here rather // than read from `frame->lost` alone because a *late* frame's counter is behind the // sequence and must not move the mark forward. if (!have_counter_ || (std::int32_t)(frame->counter - last_counter_) > 0) { last_counter_ = frame->counter; have_counter_ = true; } ``` - [ ] **Step 5: Build and measure a loss** ```bash cmake --build build sed -i 's/^run reorder.*$/run drop10 --drop-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected: `gaps=102` (unchanged — the packets really are lost), but `frames` now near **1029** rather than 925, because the hole's samples occupy the slots the lost packets would have. `criticals=0`. The channel lines of the exit dump show `stored=` grown by `102 × 100 / stride`. - [ ] **Step 6: Verify the cap** ```bash sed -i 's/^run drop10.*$/run burst300 --drop-every 1000 --drop-burst 300/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh grep "warning:" /tmp/fault_burst300.log | head -2 ``` Expected: the `past the 256-packet reservation cap` warning, and **no** `jumped implausibly far ahead` / `timing samples by arrival instead` pair. Then the same case with `--relative-time` added: the cap warning is still there and the stream stays on its own clock. - [ ] **Step 7: Commit** ```bash git add src/acquisition.hpp src/acquisition.cpp src/model.hpp src/model.cpp git commit -m "fix: a lost packet reserves the samples it owed instead of vanishing from the timeline" ``` --- ### Task 7: Surface the counters **Files:** - Modify: `src/model.hpp:155-162` (`StreamStats`), `src/acquisition.cpp:84-89` (the poll copy) - Modify: `src/status_text.cpp:68-80` (the detail tier), `src/ui.cpp:10718-10724` (the exit dump) - Test: the counters read non-zero in a reorder run and zero in a clean one **Interfaces:** - Consumes: `udps_stats_t::late_resequenced`, `late_delivered`, `stale_packets` (Task 4); `ScopeModel::stats().holes`, `hole_slots` (Task 6). - Produces: `StreamStats::resequenced`, `late`, `stale`, `holes`, `hole_slots`, `backfilled` (all `std::atomic`). Task 10 increments `backfilled`. - [ ] **Step 1: Extend `StreamStats`** ```cpp /// Out-of-order packets put back in order inside the reorder window: no hole was ever /// seen for these. std::atomic resequenced{0}; /// Out-of-order packets delivered behind the sequence, past the window. std::atomic late{0}; /// Packets the client still refused outright (a duplicate, or any late packet while /// resequencing is disabled). Counted since the beginning and displayed nowhere until /// now, which is why an inversion had to be inferred from `packets` and `frames` /// diverging. std::atomic stale{0}; /// Reserved holes, and the stored slots they cost. std::atomic holes{0}; std::atomic hole_slots{0}; /// Stored slots a late packet repaired. std::atomic backfilled{0}; ``` - [ ] **Step 2: Copy them on the poll** In `src/acquisition.cpp:run()`, beside the existing stores: ```cpp out.resequenced.store(s.late_resequenced, std::memory_order_relaxed); out.late.store(s.late_delivered, std::memory_order_relaxed); out.stale.store(s.stale_packets, std::memory_order_relaxed); ``` - [ ] **Step 3: Add them to the detail tier** `StatusSnapshot` gains the six fields, filled where `packets`/`frames`/`gaps` are, and `format_detail`'s format string grows one segment after `gaps %llu`: ```c " | reseq %llu late %llu stale %llu holes %llu (%llu slots, %llu filled)" ``` with the six arguments in that order. The segment is omitted entirely when all six are zero, so a clean stream's status line is unchanged. - [ ] **Step 4: Add them to the exit dump** In `src/ui.cpp`, extend the `frames=... packets=... gaps=...` printf with: ```c " reseq=%llu late=%llu stale=%llu holes=%llu hole_slots=%llu backfilled=%llu" ``` - [ ] **Step 5: Build and read both cases** ```bash cmake --build build sed -i 's/^run burst300.*$/run clean/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected: `reseq=0 late=0 stale=0 holes=0 hole_slots=0 backfilled=0`. ```bash sed -i 's/^run clean$/run reorder --reorder-every 10/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected: `reseq` near **102**, `late=0`, `holes=0`, `gaps=0` — every inversion resequenced, nothing lost. ```bash sed -i 's/^run reorder.*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh && /tmp/udps_reseq.sh ``` Expected: `reseq=0`, `late` near **102**, `holes` near 102, `backfilled=0` (Task 10 is what makes it non-zero). `criticals=0`. - [ ] **Step 6: Commit** ```bash git add src/model.hpp src/acquisition.cpp src/status_text.cpp src/status_text.hpp src/ui.cpp git commit -m "feat: the status bar and the exit dump report resequencing, holes and stale packets" ``` --- ## Phase 4 — Backfill ### Task 8: `MinMaxPyramid::amend()` **Files:** - Modify: `deps/plotter/include/plot/data/minmax_pyramid.hpp:78` (declare), `deps/plotter/src/data/minmax_pyramid.cpp:180` (define) - Test: `deps/plotter/tests/test_minmax_pyramid.cpp` **Interfaces:** - Produces: `void MinMaxPyramid::amend(u64 absolute_sample, f32 v)`. Precondition: the sample at that index was appended **non-finite**. A no-op for an index outside `[window_base(), window_end())`. Task 10 calls it. - [ ] **Step 1: Write the failing test** Append to `deps/plotter/tests/test_minmax_pyramid.cpp`: ```cpp PLOT_TEST(amend_folds_a_backfilled_sample_into_every_covering_level) { // A hole is appended as NaN -- which the pyramid already stores as an empty aggregate -- // and amended later when the packet that owned it finally arrives. The amendment has to // reach every level that covers the index, because the renderer reads whichever level the // zoom picked: repairing level 0 alone would make a trace whole when zoomed in and still // broken when zoomed out. MinMaxPyramid p; p.reserve(8 * kBucket); for (u64 i = 0; i < 8 * kBucket; ++i) { // One NaN, at an index comfortably inside the first bucket of every level. p.append(i == 5 ? std::numeric_limits::quiet_NaN() : 1.0f); } // Before: every covering bucket aggregates kBucket-1 ones and knows nothing of index 5. PLOT_CHECK(p.bucket_count(0, 0) == kBucket - 1); PLOT_CHECK_NEAR(p.bucket(0, 0).max, 1.0); p.amend(5, 7.0f); // Level 0 gained the sample. PLOT_CHECK(p.bucket_count(0, 0) == kBucket); PLOT_CHECK_NEAR(p.bucket(0, 0).max, 7.0); PLOT_CHECK_NEAR(p.bucket_sum(0, 0), f64(kBucket - 1) + 7.0); PLOT_CHECK_NEAR(p.bucket_sumsq(0, 0), f64(kBucket - 1) + 49.0); // And so did every coarser level that covers it. for (u32 l = 1; l < p.level_count(); ++l) { if (p.level_size(l) == 0) continue; const u64 span = u64{kBucket} << l; const u64 b = 5 / span; if (b < p.level_base(l) || b - p.level_base(l) >= p.level_size(l)) continue; PLOT_CHECK_NEAR(p.bucket(l, b).max, 7.0); PLOT_CHECK(p.bucket_count(l, b) == span); } } PLOT_TEST(amend_makes_an_all_nan_bucket_present_again) { // A bucket whose every sample was non-finite is stored with count 0, which every reader // treats as absent. One amendment has to make it present -- min and max become the value // rather than widening from a meaningless zero, which is the case a plain fold gets wrong. MinMaxPyramid p; p.reserve(4 * kBucket); const f32 nan = std::numeric_limits::quiet_NaN(); for (u64 i = 0; i < 2 * kBucket; ++i) p.append(i < kBucket ? nan : 1.0f); PLOT_CHECK(p.bucket_count(0, 0) == 0); p.amend(3, -2.0f); PLOT_CHECK(p.bucket_count(0, 0) == 1); PLOT_CHECK_NEAR(p.bucket(0, 0).min, -2.0); PLOT_CHECK_NEAR(p.bucket(0, 0).max, -2.0); } PLOT_TEST(amend_outside_the_window_is_a_no_op) { // An index the window has already dropped, or one not yet appended, has nowhere to go. // Silently ignoring it is what lets the caller amend without first proving the ring still // holds the slot -- a race it cannot win, because the window slides on the other thread. MinMaxPyramid p; p.reserve(4 * kBucket); for (u64 i = 0; i < 4 * kBucket; ++i) p.append(1.0f); p.drop_before(2 * kBucket); const u64 before = p.bucket_count(0, 2); p.amend(5, 99.0f); // dropped p.amend(100 * kBucket, 99.0f); // never appended PLOT_CHECK(p.bucket_count(0, 2) == before); PLOT_CHECK_NEAR(p.bucket(0, 2).max, 1.0); } ``` Add `#include ` to the test file if it is not already there. - [ ] **Step 2: Run the tests to verify they fail** ```bash cmake --build deps/plotter/build/debug 2>&1 | tail -5 ``` Expected: a compile error, `'amend' is not a member of 'plot::data::MinMaxPyramid'`. - [ ] **Step 3: Declare it** In `deps/plotter/include/plot/data/minmax_pyramid.hpp`, after `append`: ```cpp /// Folds a backfilled sample into every aggregate that covers absolute index /// `absolute_sample`. /// /// **Precondition: the sample at that index was appended non-finite.** That is what makes /// this purely additive -- min/max widen, sum/sumsq accumulate and count rises by one, at /// every covering level -- so no bucket is ever recomputed and the cost is O(levels) /// regardless of how wide the repaired span is. Amending a *finite* sample would need the /// old value removed from sum/sumsq and could not restore min/max at all, which is why /// this is a one-way NaN-to-value operation and not a general write. /// /// An index outside `[window_base(), window_end())` is a no-op: the window slides on the /// consumer thread, so a caller cannot prove the slot is still held before asking. void amend(u64 absolute_sample, f32 v); ``` - [ ] **Step 4: Define it** In `deps/plotter/src/data/minmax_pyramid.cpp`, after `append()`: ```cpp void MinMaxPyramid::amend(u64 absolute_sample, f32 v) { if (!std::isfinite(v)) return; if (absolute_sample < base_ || absolute_sample >= count_) return; const f64 vd = f64(v); const auto fold = [&](Entry& e) { // An empty bucket's `mm` is meaningless (see the Entry note), so the first finite // sample *sets* the extremes instead of widening from a zero that was never a sample. if (e.count == 0) { e.mm.min = v; e.mm.max = v; } else { e.mm.min = std::min(e.mm.min, v); e.mm.max = std::max(e.mm.max, v); } e.sum += vd; e.sumsq += vd * vd; ++e.count; }; // The in-progress level-0 bucket, when the index lies inside it. It is not an entry yet, // so the loop below cannot reach it; leaving it out would lose the amendment for a hole // whose tail is still filling, which is exactly the hole a prompt backfill arrives into. if (count_ % kBucket != 0 && absolute_sample >= (count_ / kBucket) * kBucket) { fold(pending_); } // Every level independently: min/max/sum/sumsq/count are each associative, so folding the // value into a parent gives the same aggregate as re-merging its children would. for (u32 l = 0; l < static_cast(levels_.size()); ++l) { const u64 i = absolute_sample / level_span(l); if (i < levels_[l].base) continue; if (i - levels_[l].base >= static_cast(levels_[l].entries.size())) continue; fold(levels_[l].entries[static_cast(i - levels_[l].base)]); } } ``` - [ ] **Step 5: Run the tests to verify they pass** ```bash cmake --build deps/plotter/build/debug && \ PLOT_TEST_FILTER=amend deps/plotter/build/debug/tests/plot_tests ``` Expected: three tests run, `0 failed`. - [ ] **Step 6: Run the whole suite once** ```bash xvfb-run -a deps/plotter/build/debug/tests/plot_tests 2>&1 | tail -3 ``` Expected: `0 failed`. Run this **once**, in the background if it is slow; do not re-run it to chase a cosmetic difference. - [ ] **Step 7: Commit** ```bash git add deps/plotter/include/plot/data/minmax_pyramid.hpp \ deps/plotter/src/data/minmax_pyramid.cpp \ deps/plotter/tests/test_minmax_pyramid.cpp git commit -m "feat(plotter): a non-finite aggregate can be amended additively at every level" ``` --- ### Task 9: `RingBuffer::amend()` **Files:** - Modify: `deps/plotter/include/plot/data/ring_buffer.hpp:65` (after the batch `push`) - Test: `deps/plotter/tests/test_ring_buffer.cpp` **Interfaces:** - Produces: `bool RingBuffer::amend(u64 absolute, T v)` — `true` when the slot was still retained and written, `false` when the index had aged out or was never produced. Task 10 uses the return to decide whether to queue the pyramid amend. - [ ] **Step 1: Write the failing test** Append to `deps/plotter/tests/test_ring_buffer.cpp`: ```cpp PLOT_TEST(ring_amend_rewrites_a_retained_slot) { // Backfill: a slot pushed as a placeholder is rewritten once the data that belongs there // arrives. The index is absolute, because that is the only coordinate a late packet can // compute -- a retained-window index would have moved by the time it got here. plot::data::RingBuffer r(8); for (int i = 0; i < 6; ++i) r.push(f32(i)); PLOT_CHECK(r.amend(3, 42.0f)); PLOT_CHECK_NEAR(r[3], 42.0); PLOT_CHECK(r.produced() == 6); // an amend is not a push } PLOT_TEST(ring_amend_refuses_an_index_the_ring_no_longer_holds) { // The window slides on the producer, so a late amend may arrive after its slot was // overwritten. Writing it anyway would corrupt whatever sample now owns that slot, so the // bounds test is the whole safety of the operation and the caller reads the result. plot::data::RingBuffer r(8); for (int i = 0; i < 20; ++i) r.push(f32(i)); PLOT_CHECK(r.base() == 12); PLOT_CHECK(!r.amend(3, 42.0f)); // aged out PLOT_CHECK(!r.amend(20, 42.0f)); // never produced PLOT_CHECK_NEAR(r[0], 12.0); // and nothing moved PLOT_CHECK_NEAR(r[7], 19.0); } ``` - [ ] **Step 2: Run the tests to verify they fail** ```bash cmake --build deps/plotter/build/debug 2>&1 | tail -5 ``` Expected: a compile error, `'amend' is not a member`. - [ ] **Step 3: Implement it** In `deps/plotter/include/plot/data/ring_buffer.hpp`, after the batch `push`: ```cpp /// Producer thread only: rewrites the slot at absolute index `absolute`, without pushing. /// Returns false when that index is not retained -- already overwritten, or not yet /// produced -- in which case nothing is written. /// /// This is for *backfill*: a slot pushed as a placeholder and corrected when the data that /// belongs there arrives late. It is the same race class as `push()` -- one relaxed store /// into a slot a consumer may be reading, and T is lock-free by the class's static_assert, /// so a reader sees the old value or the new one and never a torn one. It deliberately does /// not touch `produced_`: the count fixes which indices a snapshot reads, and an amend /// changes a value inside that window rather than extending it. bool amend(u64 absolute, T v) { const u64 p = produced_.load(std::memory_order_relaxed); if (absolute >= p) return false; if (p > cap_ && absolute < p - cap_) return false; buf_[static_cast(absolute & mask_)].store(v, std::memory_order_relaxed); return true; } ``` - [ ] **Step 4: Run the tests to verify they pass** ```bash cmake --build deps/plotter/build/debug && \ PLOT_TEST_FILTER=ring_amend deps/plotter/build/debug/tests/plot_tests ``` Expected: two tests, `0 failed`. - [ ] **Step 5: Commit** ```bash git add deps/plotter/include/plot/data/ring_buffer.hpp deps/plotter/tests/test_ring_buffer.cpp git commit -m "feat(plotter): a retained slot can be backfilled without pushing" ``` --- ### Task 10: Wire the backfill **Files:** - Modify: `src/channel.hpp:48` and `src/channel.cpp:32-63` (`acquire()` drains the queue) - Modify: `src/model.hpp`, `src/model.cpp` (the queue and `BufferStore::acquire_all`) - Modify: `src/acquisition.cpp` (handle a late frame) - Test: the `late` run's `backfilled` counter and a NaN-free trace **Interfaces:** - Consumes: `RingBuffer::amend` (Task 9), `MinMaxPyramid::amend` (Task 8), `ScopeModel::holes()` (Task 6). - Produces: `ChannelBuffer::queue_amend(u64 absolute, f32 v)` (producer side, under `mutex()`) and the drain inside `acquire()`. No new thread primitive: `acquire_all()` runs under `model.mutex()` (`src/ui.cpp:6263-6265`) and the producer holds that same mutex for all of `on_data` (`src/acquisition.cpp:386`), so a plain `std::vector` member is correctly serialised. - [ ] **Step 1: Give `ChannelBuffer` the queue and the drain** In `src/channel.hpp`: ```cpp /// Producer thread, under `ScopeModel::mutex()`: records that absolute sample `i` has been /// backfilled, for `acquire()` to fold into the pyramid. /// /// The pyramid is mutated *only* on the consumer thread -- `acquire()` is called from the /// frame path -- and a bucket is 32 bytes and not atomic, so amending it from the producer /// would let a reader see a half-updated bucket whose `sum` did not match its `count`: a /// wrong mean out of a structure whose readers trust `count == 0` to mean absent. The value /// ring is amended by the producer (one lock-free store); the pyramid waits for here. void queue_amend(u64 absolute, f32 v) { pending_amends_.push_back({absolute, v}); } ``` with the member: ```cpp struct Amend { u64 absolute; f32 value; }; std::vector pending_amends_; ``` In `src/channel.cpp:acquire()`, **after** `published_ = produced;` and the `drop_before` call, so every amended index is already in the pyramid: ```cpp // Backfills, applied after the append pass that published their indices: an amend whose // index the append had not yet reached would be dropped by the pyramid's bounds test, and // a hole's tail can be pushed and backfilled inside one frame interval. if (!pending_amends_.empty()) { for (const Amend& a : pending_amends_) pyramid_.amend(a.absolute, a.value); pending_amends_.clear(); } ``` `invalidate()` clears it too — a rebind's queued amends address slots that no longer exist. - [ ] **Step 2: Handle a late frame in `on_data`** After the counter tracking from Task 6 Step 4, and before the stamping, add an early branch: a late frame does not advance the timeline, it repairs it. ```cpp // A frame delivered behind the sequence: the slots it owns were reserved as NaN, so it // writes values and nothing else. It must not be stamped, must not advance `tick_` and // must not reach the trigger or the math channels -- the time ring was pre-stamped at // reservation and the stateful consumers have already run past this region, which is the // trade the design takes: the raw trace repairs, a derived trace keeps the gap, and the // display therefore says "this arrived too late to be computed with". if (frame->late) { backfill_from(frame, part, channels, store, g); return; } ``` - [ ] **Step 3: Write `backfill_from()`** ```cpp void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part, const std::vector& channels, BufferStore& store, u64 g) { // Which hole this packet belongs to, by counter: that is the whole reason relative timing // and resequencing are one design. A stamp could not answer it -- the packet's own stamps // are a claim about a timeline the ring has already moved past -- but the counter indexes // the reservation table directly. const Hole* hole = nullptr; for (const Hole& h : model_.holes()) { if (frame->counter >= h.counter_first && frame->counter < h.counter_first + h.packets) { hole = &h; break; } } if (hole == nullptr) return; // its hole has already slid out of the ring const u64 offset_raw = u64(frame->counter - hole->counter_first) * u64(batch_); u64 filled = 0; for (std::size_t i = 0; i < channels.size(); ++i) { if (channels[i].math) continue; const u64 period = g * part.stride[i]; // Only slots lying *wholly* inside this packet's raw span are repaired. A slot // straddling the span's edge mixes this packet's samples with ones still missing, and // its real members were never retained -- so it stays NaN, and a displayed hole is at // most one slot wider per edge than the true loss. const u64 first_slot = (offset_raw + period - 1) / period; const u64 last_slot = (offset_raw + u64(batch_)) / period; for (u64 s = first_slot; s < last_slot; ++s) { double acc = 0.0; for (u64 j = 0; j < period; ++j) { const u64 k = s * period + j - offset_raw; const double raw = channels[i].elements > 1 ? udps_frame_value(frame, static_cast(channels[i].signal), 0, std::min(static_cast(k), channels[i].elements - 1)) : udps_frame_value(frame, static_cast(channels[i].signal), static_cast(k), 0); acc += raw * channels[i].gain + channels[i].offset; } const f32 value = static_cast(acc / double(period)); const u64 absolute = hole->slot_begin + offset_raw / period + (s - first_slot); // The ring first, by the producer that owns its writes; the pyramid is queued for the // consumer. A slot the ring has already discarded returns false, and then there is no // aggregate to amend either. if (store.values(i).amend(absolute, value)) { store.channel(i).queue_amend(absolute, value); ++filled; } } } model_.stats().backfilled.fetch_add(filled, std::memory_order_relaxed); } ``` Declare it in `src/acquisition.hpp`. - [ ] **Step 4: Build and measure the repair** ```bash cmake --build build sed -i 's/^run .*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh /tmp/udps_reseq.sh && grep -E "reseq=|backfilled=" /tmp/fault_late.log ``` Expected: `late` near 102, `holes` near 102, and **`backfilled` near `102 × 100 / stride`** — the slots the late packets repaired. `criticals=0`. - [ ] **Step 5: Verify the repair is real, not just counted** The `holes` and `backfilled` counters agreeing is necessary but not sufficient — a wrong `absolute` would amend the wrong slots and still count. Assert on the data: run with `--reorder-delay-ms 50` and compare the exported values against the clean run's, which must agree sample for sample where the hole was. ```bash cat >> /tmp/udps_reseq.sh <<'EOF' python3 - <<'PY' import csv def col(p, i=1): with open(p) as f: return [float(r[i]) for r in csv.reader(f) if r and not r[0].startswith('#')] import math v = col('/tmp/exp_late.csv') print('nan slots in a backfilled export:', sum(1 for x in v if math.isnan(x)), 'of', len(v)) PY EOF ``` Expected: the NaN count is a small multiple of the hole count (the one straddling slot per edge), **not** `102 × 100 / stride`. A NaN count equal to `hole_slots` means the amend wrote nowhere; a NaN count of zero across a boundary means the edge rule was not applied. - [ ] **Step 6: Commit** ```bash git add src/channel.hpp src/channel.cpp src/model.hpp src/model.cpp \ src/acquisition.hpp src/acquisition.cpp git commit -m "feat: a late packet backfills the hole it left, values only" ``` --- ## Phase 5 — The band, the probe and the documentation ### Task 11: A shaded band over an invalid span **Files:** - Modify: `deps/plotter/include/plot/plot.hpp`, `deps/plotter/src/render/renderer.cpp` - Modify: `src/plot_host.cpp` (pass the model's holes to each row) - Test: `deps/plotter/tests/test_render_smoke.cpp` **Interfaces:** - Produces: `void Plot::set_invalid_spans(std::span>)` and `Plot::invalid_spans()`. Drawn behind the trace, in **data** coordinates. - [ ] **Step 1: Write the failing test** Append to `deps/plotter/tests/test_render_smoke.cpp`: ```cpp PLOT_TEST(an_invalid_span_is_shaded_and_moves_with_the_data) { // Unlike the x-note caption, a band marks a *region of the data*, so it must pan and zoom // with it: a band that stayed put under a pan would accuse the wrong samples. It also must // reserve no layout space -- the labels were measured without it, and a hole must not be // able to move them. plottest::HeadlessContext ctx(kW, kH); // ... build a figure with one row and a full-width trace, as the neighbouring tests do ... // 1. ink with no span, 2. ink with a span over the middle third, 3. ink with the same span // after a pan of a known number of pixels. // Assert: (2) differs from (1) only inside the middle third; (3)'s differing region is // (2)'s shifted by exactly the pan; the axes rect is identical in all three. } ``` Fill the body against the neighbouring tests' setup, and against `the_x_note_is_chrome_that_a_pan_or_a_zoom_does_not_move`, which is the *opposite* assertion and shows how the footprint is measured. - [ ] **Step 2: Run it to verify it fails** ```bash cmake --build deps/plotter/build/debug 2>&1 | tail -5 ``` Expected: a compile error, `'set_invalid_spans' is not a member of 'plot::Plot'`. - [ ] **Step 3: Add the API and the draw** `Plot` stores `std::vector> invalid_spans_`. The renderer draws each span as a quad in the axes rect, clipped to it, **before** the traces and after the grid, transformed through the same x mapping the traces use — so it is data-anchored by construction rather than by a second code path. Colour is the readout backplate's tint at low alpha, so it reads as "no data here" and not as a trace. - [ ] **Step 4: Run it to verify it passes** ```bash cmake --build deps/plotter/build/debug && \ xvfb-run -a env PLOT_TEST_FILTER=invalid_span deps/plotter/build/debug/tests/plot_tests ``` Expected: `0 failed`. - [ ] **Step 5: Feed it from the model** In `src/plot_host.cpp`, where each row's per-frame state is written under `model.mutex()`, convert `model.holes()` to `(x_begin, x_end)` pairs in display coordinates (`+ display_epoch`) and call `set_invalid_spans()`. Only in `Live` and `Capturing`: a `Held` or `Paused` snapshot has its own stamps and the live hole table does not address it. - [ ] **Step 6: Build, run and look** ```bash cmake --build build sed -i 's/^run .*$/run late --reorder-every 10 --reorder-delay-ms 50/' /tmp/udps_reseq.sh /tmp/udps_reseq.sh ``` Expected: `criticals=0`, warnings at the baseline. Capture a frame with `UDPSCOPE_SCREENSHOT_PATH` if that probe exists and confirm the band sits over the gap. - [ ] **Step 7: Commit** ```bash git add deps/plotter/include/plot/plot.hpp deps/plotter/src/render/renderer.cpp \ deps/plotter/tests/test_render_smoke.cpp src/plot_host.cpp git commit -m "feat: a reserved hole is shaded, so a sub-pixel gap is still a report" ``` --- ### Task 12: The probe and the documentation **Files:** - Modify: `src/ui.cpp` (`UDPSCOPE_PROBE_RESEQ_MS`) - Modify: `docs/claude/verification.md`, `probe-readings.md`, `invariants-data-path.md`, `invariants-plot-and-layout.md`, `invariants-trigger-and-pinning.md`, `build-and-run.md`, `README.md`, `SPECS.md`, `todo.md` - Modify: `docs/superpowers/specs/2026-10-08-resequencing-and-relative-timing-design.md` (the SPSC correction) - [ ] **Step 1: Add the probe** Following the pattern of `UDPSCOPE_PROBE_STATS_MS` at `src/ui.cpp:10648`, a one-shot that prints: ``` [reseq-probe] reseq=N late=N holes=N hole_slots=N backfilled=N nan_slots=N trigger_in_hole=0 ``` `nan_slots` is counted by scanning the newest channel's retained window; `trigger_in_hole` asserts the engine did not latch inside a reserved span, by comparing the last trigger time against the hole table. - [ ] **Step 2: Run it, three cases, serially** ```bash for t in clean "--reorder-every 10" "--reorder-every 10 --reorder-delay-ms 50"; do : ; done ``` Run each as its own `/tmp/udps_reseq.sh` invocation with `UDPSCOPE_PROBE_RESEQ_MS=5000` added, and record the measured numbers. **Do not batch them** — four `xvfb` runs back to back starve the mock past its 2 s silence timeout and read as a reconnect. - [ ] **Step 3: Write the readings down** Add the measured table to `docs/claude/verification.md` beside the existing fault-injection section, and a per-line entry to `docs/claude/probe-readings.md` saying what each field must read and which defect a wrong reading names. **Every figure must be one measured on this machine** — no estimates. - [ ] **Step 4: Update the invariants** - `invariants-data-path.md`: append-only is no longer the whole truth. State `amend`'s non-finite precondition, that the ring is amended by the producer and the pyramid by the consumer, and why (the 32-byte non-atomic bucket). - `invariants-plot-and-layout.md`: the band is data-anchored and reserves no layout space — the opposite of the `t0` caption, and both rules in one place so neither is applied to the other. - `invariants-trigger-and-pinning.md`: the engine is not fed a reserved hole's samples, and why (a NaN comparison is false anyway, but feeding it would advance the hold-off clock over time nothing was sampled in). - `build-and-run.md`: the vendored `udpsclient` diff — the reorder ring, late delivery replacing the discard, the third `clamp_deadline()`, and `reorder_window_s = 0` as the exact-restore switch. - [ ] **Step 5: Update `README.md`, `SPECS.md` and `todo.md`** `README.md` gains `--relative-time` and a paragraph on what recovery does and does not repair (values yes, math and trigger no). `SPECS.md` gets the deviation note. `todo.md` item 2 is struck. - [ ] **Step 6: Correct the spec's SPSC line** In the design document's Backfill section, replace *"the producer enqueues `(slot, value)` on an SPSC queue"* with the measured truth: `acquire_all()` is called under `model.mutex()` (`src/ui.cpp:6263`) and the producer holds that same mutex for all of `on_data` (`src/acquisition.cpp:386`), so a plain `std::vector` member is correctly serialised and no lock-free queue is needed. - [ ] **Step 7: Full clean build, read both baselines** ```bash rm -rf build && cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=RelWithDebInfo && \ cmake --build build 2>&1 | grep -c "warning:" ``` Expected: **4**. More is a regression introduced by this work and is fixed before the commit, not noted. - [ ] **Step 8: Commit** ```bash git add src/ui.cpp docs/ README.md SPECS.md todo.md git commit -m "docs: the reseq probe's readings, and the invariants resequencing changes" ``` --- ## Self-review **Spec coverage.** Relative timing → Tasks 1, 2; the declared-rate-only rule → Task 2 Step 2; the absolute ring for export → Task 3; the reorder ring and the three arrival cases → Task 4; `clamp_deadline` → Task 4 Step 5; the ring bounds and early commit → Task 4 Step 4; `reorder_window_s` including the 0 kill switch → Task 4 Steps 1, 2, 7; hole reservation through the existing loop → Task 6 Step 3; the 256-packet cap → Task 6 Steps 2, 6; math NaN-through and the trigger not latching → Task 6 Step 3 and Task 12 Step 1; the pyramid amend → Task 8; the ring amend → Task 9; the thread split → Task 10 Step 1; counter-addressed backfill and the whole-slot-only edge rule → Task 10 Step 3; the counters including `stale_packets` → Task 7; the band → Task 11; the mock's missing option → Task 5; probe and docs → Task 12. **Two spec items deliberately deferred, both named in the spec's own out-of-scope list:** replaying math/trigger over a backfill, and the `(min, max, mean, rms)` gather mode. **Type consistency.** `amend` returns `bool` on `RingBuffer` (the caller branches on it, Task 10 Step 3) and `void` on `MinMaxPyramid` (bounds failure is a no-op by contract, Task 8 Step 3) — deliberately different, and both call sites match. `Hole::counter_first`/`packets` are the fields Task 10 searches on. `StreamStats::backfilled` is declared in Task 7 and incremented in Task 10. **Known soft spots, flagged rather than papered over.** Task 3 Step 5 and Task 11 Steps 1 and 3 point at code I have not read line by line (`src/export.cpp`'s column assembly, `Plot`'s state block, the smoke test's figure setup) and say what to read rather than quoting code that might not compile. Everything else is complete.