From 723d1db3d2551fad7e001b81d50038a48dc71e45 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 8 Oct 2026 16:50:50 +0200 Subject: [PATCH] docs: implementation plan for resequencing and relative timing Twelve tasks in five phases, so relative timing ships before any of the client-side reordering work lands. Co-Authored-By: Claude Opus 4.6 --- ...-10-08-resequencing-and-relative-timing.md | 1551 +++++++++++++++++ 1 file changed, 1551 insertions(+) create mode 100644 docs/superpowers/plans/2026-10-08-resequencing-and-relative-timing.md diff --git a/docs/superpowers/plans/2026-10-08-resequencing-and-relative-timing.md b/docs/superpowers/plans/2026-10-08-resequencing-and-relative-timing.md new file mode 100644 index 0000000..651150b --- /dev/null +++ b/docs/superpowers/plans/2026-10-08-resequencing-and-relative-timing.md @@ -0,0 +1,1551 @@ + + + +# 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.