diff --git a/Client/udpscope/FrameDecoder.cpp b/Client/udpscope/FrameDecoder.cpp index a9afbdf..482a714 100644 --- a/Client/udpscope/FrameDecoder.cpp +++ b/Client/udpscope/FrameDecoder.cpp @@ -1,10 +1,25 @@ #include "FrameDecoder.h" +#include + namespace udpscope { /** Fallback cycle period before the first inter-packet gap is known. */ static constexpr double kDefaultDt = 1.0e-3; +/** + * How far the forward-chained prediction for an accumulated burst may sit from + * where arrival time says it should be before the chain is abandoned. + * + * A kernel draining a backlog of queued datagrams can legitimately put the + * prediction a few hundred milliseconds ahead of arrival, so the threshold has + * to be well clear of that. Anything larger is not delivery jitter: it is lost + * packets or a declared sampling rate that does not match the producer's real + * one, and both must resynchronise rather than accumulate forever. Same value + * and same reasoning as ClockOffset::kRecalibThresholdS. + */ +static constexpr double kBurstResyncThresholdS = 0.5; + void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; state_.assign(signals_.size(), SigState{}); @@ -107,21 +122,25 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, : 0.0; if (d.samplingRate > 0.0) { - /* Forward-chain anchor: t[0] = lastEnd + dt, or wallNow on first - * packet (arrival-time for the very first burst only). */ - double base; + /* Where arrival time says this burst begins: its last element was + * acquired just before the packet landed. */ + const double arrivalAnchor = + wallNow - static_cast(nElems - 1u) * dt; + + /* Chaining from the end of the previous burst is immune to arrival + * jitter — a kernel draining several queued datagrams microseconds + * apart still yields contiguous timestamps. But a pure chain is + * blind: one lost datagram, or a declared rate that does not match + * the producer's real one, displaces every later sample and never + * recovers. So the chain is a PREDICTION, checked each packet + * against arrival and abandoned when the two disagree by more than + * a delivery backlog can explain. That bounds the error instead of + * letting it accumulate. */ + double base = arrivalAnchor; if (st.lastEmittedValid) { - base = st.lastEmittedEnd + dt; - } else { - /* First packet: anchor element 0 at arrival time. - * This one packet may be slightly off, but subsequent packets - * chain from this end and jitter is suppressed thereafter. */ - base = wallNow - static_cast(nElems - 1u) * dt; - /* Calibrate the clock-offset so later hrt-based paths (if any) - * are consistent, but we don't use it in this branch. */ - if (f.hrt != 0u && hrtFit_.ready()) { - const double hrtSec = hrtFit_.toSeconds(f.hrt); - (void) st.offset.map(hrtSec, wallNow); + const double predicted = st.lastEmittedEnd + dt; + if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) { + base = predicted; } } tsOut.resize(nElems); diff --git a/Client/udpscope/FrameDecoder.h b/Client/udpscope/FrameDecoder.h index 43e847e..9f912c5 100644 --- a/Client/udpscope/FrameDecoder.h +++ b/Client/udpscope/FrameDecoder.h @@ -54,8 +54,11 @@ private: bool lastAccValid = false; uint32_t prevAccCount = 0; /** For accumulated scalars with a declared sampling rate: end timestamp - * of the most recently emitted burst, used as a forward-chain anchor - * that is immune to arrival-time jitter. */ + * of the most recently emitted burst. The next burst is PREDICTED to + * start one sample period after it — immune to arrival-time jitter — + * but the prediction is discarded when arrival time disagrees with it + * by more than a delivery backlog can explain, so packet loss cannot + * displace the trace permanently. */ double lastEmittedEnd = 0.0; bool lastEmittedValid = false; }; diff --git a/Client/udpscope/tests/FrameDecoderTest.cpp b/Client/udpscope/tests/FrameDecoderTest.cpp index e923d12..b58e39d 100644 --- a/Client/udpscope/tests/FrameDecoderTest.cpp +++ b/Client/udpscope/tests/FrameDecoderTest.cpp @@ -177,6 +177,45 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { } } +// The counterweight to the test above. Suppressing arrival jitter by chaining +// each burst onto the previous one is only safe while the chain is checked: on +// UDP, packets are lost, and a chain that ignores arrival entirely closes the +// hole silently and dates every later sample a full second early — for the rest +// of the run, because nothing ever pulls it back. The prediction has to be +// abandoned once arrival contradicts it by more than a delivery backlog could. +TEST(FrameDecoder, AccumulatedScalarResynchronisesAfterLostPackets) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 1000.0; /* 10 samples = 10 ms per packet */ + dec.setSignals({m}); + + std::vector ts; + for (int p = 0; p < 10; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 500.0 + p * 0.010, 10); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + } + /* Contiguous so far: burst 9 ends at 500.090. */ + EXPECT_NEAR(ts[9], 500.090, 1e-9); + + /* A full second of packets never arrives. The next one lands at 501.100. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 501.100, 10); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + /* Chaining blindly would put this burst at 500.091..500.100, overlapping + * the gap as though no data were missing. */ + EXPECT_NEAR(ts[0], 501.091, 1e-9); + EXPECT_NEAR(ts[9], 501.100, 1e-9); +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; diff --git a/docs/superpowers/plans/2026-08-27-udpscope.md b/docs/superpowers/plans/2026-08-27-udpscope.md index add1d89..11c4296 100644 --- a/docs/superpowers/plans/2026-08-27-udpscope.md +++ b/docs/superpowers/plans/2026-08-27-udpscope.md @@ -1609,6 +1609,42 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { } } +// ADDED in Task 4 review. The counterweight to the test above: suppressing +// arrival jitter by chaining bursts is only safe while the chain is CHECKED. On +// UDP packets are lost, and an unchecked chain closes the hole silently and +// dates every later sample early for the rest of the run. +TEST(FrameDecoder, AccumulatedScalarResynchronisesAfterLostPackets) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 1000.0; /* 10 samples = 10 ms per packet */ + dec.setSignals({m}); + + std::vector ts; + for (int p = 0; p < 10; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 500.0 + p * 0.010, 10); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + } + EXPECT_NEAR(ts[9], 500.090, 1e-9); + + /* A full second of packets never arrives. The next one lands at 501.100. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 501.100, 10); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + /* Chaining blindly would put this burst at 500.091..500.100, as though no + * data were missing. */ + EXPECT_NEAR(ts[0], 501.091, 1e-9); + EXPECT_NEAR(ts[9], 501.100, 1e-9); +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; @@ -1882,8 +1918,44 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, return true; } - /* Rule 3: accumulated scalar, based on the producer's own hrt. */ + /* Rule 3: accumulated scalar. + * + * AMENDED after Task 4 review. The version below originally sent EVERY + * accumulated scalar through the hrt fit, falling back to packetBurst until + * the fit was ready. That cannot work when a declared samplingRate is + * present: HrtRateFit needs 32 packets, bursty delivery can begin before + * that, and packetBurst then crams a 10 ms burst into a 50 us arrival gap — + * exactly the sawtooth this rule exists to prevent. Worse, HrtRateFit fits + * hrt against ARRIVAL time, so a burst episode corrupts the very rate the + * fallback is waiting on. + * + * With a declared rate none of that is needed: the intra-packet step is + * exact, and bursts are contiguous, so the next burst is PREDICTED at + * lastEmittedEnd + dt. The prediction must be checked, not trusted — a pure + * chain silently closes the hole left by a lost datagram and dates every + * later sample early for the rest of the run. So each packet compares the + * prediction against the arrival anchor and abandons it beyond + * kBurstResyncThresholdS. The hrt path below remains for samplingRate == 0. */ if (d.numElements() == 1u && nElems > 1u) { + const double dtDeclared = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; + if (d.samplingRate > 0.0) { + const double arrivalAnchor = + wallNow - static_cast(nElems - 1u) * dtDeclared; + double base = arrivalAnchor; + if (st.lastEmittedValid) { + const double predicted = st.lastEmittedEnd + dtDeclared; + if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) { + base = predicted; + } + } + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = base + static_cast(e) * dtDeclared; + } + st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedValid = true; + return true; + } if (!hrtFit_.ready()) { return packetBurst(idx, nElems, wallNow, tsOut); } @@ -1891,10 +1963,8 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, const double base = st.offset.map(hrtSec, wallNow); double dt; - if (d.samplingRate > 0.0) { - dt = 1.0 / d.samplingRate; - } else if (st.lastAccValid && st.prevAccCount > 0u && - hrtSec > st.lastAccHrtSec) { + if (st.lastAccValid && st.prevAccCount > 0u && + hrtSec > st.lastAccHrtSec) { /* The flushes carry contiguous RT cycles, so the gap divided by the * previous packet's sample count is exactly one cycle period. */ dt = (hrtSec - st.lastAccHrtSec) / @@ -1945,7 +2015,7 @@ cd Client/udpscope && cmake --build build -j && ./build/udpscope_tests --gtest_f Expected: PASS, 9 tests. -If `AccumulatedScalarSurvivesBurstyDelivery` fails on the first few samples, check that `beginFrame()` is being called before `timestamps()` — the hrt fit needs 32 packets before rule 3 engages, and the packets before that legitimately go through rule 4. +If `AccumulatedScalarSurvivesBurstyDelivery` fails, do NOT reach for the hrt fit: with a declared `samplingRate` rule 3 never consults it, precisely because the fit is not ready for the first 32 packets and is itself corrupted by bursty arrivals. Check instead that `lastEmittedEnd`/`lastEmittedValid` are being updated on every emitted burst. The only test that may legitimately fall through to rule 4 early is `AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared`, whose arrivals are uniform, so `packetBurst` is accurate there. - [ ] **Step 8: Commit**