From 3270284cfef3f961c56f135f9ec7715d0fc794d1 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 27 Aug 2026 21:55:33 +0200 Subject: [PATCH] fix(udpscope): bound the reconstructed timeline against the wall clock Round 3 of the Task 4 review. Three defects, all in FrameDecoder rule 3. The resync backstop was one-directional. `predicted` is never below lastEmittedEnd + dt, so rejecting a correction that would step backwards meant only a LAGGING chain could ever be pulled back; a chain running fast drifted ahead without bound. Two hosts' crystals differ by tens of ppm, so a declared SamplingRate is always slightly wrong in one direction or the other and this is certain on a long session. A leading timeline cannot be corrected in one burst without going backwards -- lastEmittedEnd is by definition past arrival -- so the excess is bled off by drawing each burst 10 % narrower until the timeline is back inside the threshold. A repeated packet counter was treated as a normal packet. The C client de-duplicates fragments only, so an unfragmented update reaching a host that joined the group on two interfaces was emitted twice, doubling the values and advancing the timeline by a burst that never existed. The samplingRate == 0 path differenced two HrtRateFit::toSeconds() results. toSeconds() divides an absolute tick count -- ~1e11 on a producer that has been up a day -- by a rate refitted on every packet, so its few-parts-in-1e4 wobble arrives multiplied by the whole elapsed epoch: tens of milliseconds of jitter on a value whose consecutive difference is a few milliseconds. Raw ticks are differenced instead, anchored on the first usable packet so the wobble applies only to the interval since attach. The existing hrt-gap test could not have caught the last one: its 10 ms producer period made the expected answer exactly kDefaultDt, so a decoder that derived nothing passed. It now uses 25 ms. Four tests added, all sabotage-proven. The plan is updated to match. Co-Authored-By: Claude Opus 4.6 --- Client/udpscope/FrameDecoder.cpp | 116 +++++-- Client/udpscope/FrameDecoder.h | 7 +- Client/udpscope/TimeBase.h | 9 + Client/udpscope/tests/FrameDecoderTest.cpp | 112 ++++++- docs/superpowers/plans/2026-08-27-udpscope.md | 300 +++++++++++++++--- 5 files changed, 478 insertions(+), 66 deletions(-) diff --git a/Client/udpscope/FrameDecoder.cpp b/Client/udpscope/FrameDecoder.cpp index b217420..3359bb2 100644 --- a/Client/udpscope/FrameDecoder.cpp +++ b/Client/udpscope/FrameDecoder.cpp @@ -22,6 +22,23 @@ static constexpr double kDefaultDt = 1.0e-3; */ static constexpr double kBurstResyncThresholdS = 0.5; +/** + * Largest counter gap still read as a loss count. + * + * A producer restart returns the counter to zero and a reordered datagram makes + * the unsigned gap wrap to near 2^32; multiplying either by a sample count and + * calling it elapsed time would fabricate centuries. A million lost updates is + * already far beyond any outage worth reconstructing. + */ +static constexpr uint32_t kMaxCounterGap = 1000000u; + +/** + * Burst width, as a fraction of nominal, while a leading timeline is being + * pulled back. See the sole use site for why a leading chain cannot be + * corrected in one burst and must be bled off instead. + */ +static constexpr double kLeadBleedFactor = 0.9; + void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; state_.assign(signals_.size(), SigState{}); @@ -73,6 +90,16 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, const double wallNow = f.recvTime; SigState& st = state_[idx]; + /* A repeated counter is a duplicated datagram — the same update arriving + * twice because the host joined the multicast group on two interfaces, say. + * The C client only de-duplicates fragments, so an unfragmented update + * reaches us intact both times; emitting it again would double the values + * and advance the timeline by a burst that never existed. Counter zero is + * excluded because a producer that never sets one leaves it there. */ + if (st.lastEmittedValid && f.counter != 0u && f.counter == st.lastCounter) { + return false; + } + /* hasTimeSignal() bounds the index against the FRAME's signal count, but * the time signal's type code is read from our own table, whose size is * independent — a frame carrying more signals than the installed table @@ -147,11 +174,14 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * we saw. Reinstating that duration keeps the chain honest without * consulting arrival time at all. */ double base = arrivalAnchor; + double step = dt; if (st.lastEmittedValid) { /* Unsigned subtraction wraps, so this stays right across the - * counter's own 2^32 rollover. */ + * counter's own 2^32 rollover. A gap far larger than any real + * outage is a restart or a reordered datagram rather than a + * loss count; claim nothing and let the backstop below decide. */ const uint32_t gap = f.counter - st.lastCounter; - const double lost = (gap > 1u) + const double lost = (gap > 1u && gap <= kMaxCounterGap) ? static_cast(gap - 1u) * static_cast(st.prevAccCount) : 0.0; @@ -164,17 +194,46 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) { base = predicted; } - /* Re-anchoring must never move time backwards: the ring, the - * trigger and the exporter all assume a signal's timestamps - * increase. A backward resync would be indistinguishable from - * corruption downstream, so give up the correction instead. */ + if (base <= st.lastEmittedEnd) { - base = st.lastEmittedEnd + dt; + /* Re-anchoring here would step backwards, and the ring, the + * trigger and the exporter all require a signal's stamps to + * increase. Rejecting the correction outright is not an + * option either: `predicted` is never less than + * lastEmittedEnd + dt, so rejection would make the backstop + * one-directional and let a timeline that runs FAST — two + * hosts' crystals differ by tens of ppm, so this is certain + * on a long session, not hypothetical — drift ahead of the + * wall clock without bound. + * + * So compress instead of stepping back: start immediately + * after the previous burst and spread this one out to + * arrival. A single packet is drawn narrower than its true + * width, and in exchange the timeline is back in step. */ + if (wallNow > st.lastEmittedEnd) { + step = (wallNow - st.lastEmittedEnd) / + static_cast(nElems); + base = st.lastEmittedEnd + step; + } else { + /* The timeline has run PAST arrival: our last burst is + * dated later than the moment this packet landed. There + * is no room to spread into, and no single burst can + * remove the excess without stepping back. So bleed it + * off — draw each burst a fixed fraction narrower than + * nominal until the timeline is back inside the + * threshold, then normal chaining resumes. The factor + * only has to shrink a burst faster than the clock + * mismatch grows it, and a 10 % squeeze outruns the + * tens-of-ppm crystal error that causes this by orders + * of magnitude. */ + step = dt * kLeadBleedFactor; + base = st.lastEmittedEnd + step; + } } } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { - tsOut[e] = base + static_cast(e) * dt; + tsOut[e] = base + static_cast(e) * step; } st.lastEmittedEnd = tsOut[nElems - 1u]; st.lastCounter = f.counter; @@ -184,29 +243,48 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, } /* No declared rate: need hrt-derived dt. */ - if (!hrtFit_.ready()) { + if (!hrtFit_.ready() || f.hrt == 0u) { return packetBurst(idx, nElems, wallNow, tsOut); } - const double hrtSec = hrtFit_.toSeconds(f.hrt); - const double base = st.offset.map(hrtSec, wallNow); + const double rate = hrtFit_.ticksPerSecond(); - double hrtDt; - if (st.lastAccValid && st.prevAccCount > 0u && hrtSec > st.lastAccHrtSec) { + /* Difference raw TICKS, never two toSeconds() results. + * + * hrt counts from the producer's boot, so it is already ~1e11 ticks when + * the scope attaches, while the fit is re-estimated on every packet and + * wobbles by a few parts in 1e4. toSeconds() multiplies that relative + * wobble by the whole elapsed epoch: tens of milliseconds of jitter on a + * value whose consecutive difference is a few milliseconds. Subtracting + * two such results measures the wobble, not the interval. + * + * Anchoring on the first usable packet keeps the wobble on the elapsed + * interval since attach, which is short, and ClockOffset absorbs the + * arbitrary epoch that anchoring leaves behind exactly as it would + * absorb the producer's boot epoch. */ + if (!st.hrtRefValid) { + st.hrtRef = f.hrt; + st.hrtRefValid = true; + } + const double sinceRef = (f.hrt >= st.hrtRef) + ? static_cast(f.hrt - st.hrtRef) / rate + : -static_cast(st.hrtRef - f.hrt) / rate; + const double base = st.offset.map(sinceRef, wallNow); + + double hrtDt = kDefaultDt; + if (st.lastAccValid && st.prevAccCount > 0u && f.hrt > st.lastAccHrt) { /* The flushes carry contiguous RT cycles, so the gap divided by the * previous packet's sample count is exactly one cycle period. */ - hrtDt = (hrtSec - st.lastAccHrtSec) / + hrtDt = (static_cast(f.hrt - st.lastAccHrt) / rate) / static_cast(st.prevAccCount); - } else { - hrtDt = kDefaultDt; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { tsOut[e] = base + static_cast(e) * hrtDt; } - st.lastAccHrtSec = hrtSec; - st.lastAccValid = true; - st.prevAccCount = nElems; + st.lastAccHrt = f.hrt; + st.lastAccValid = true; + st.prevAccCount = nElems; return true; } diff --git a/Client/udpscope/FrameDecoder.h b/Client/udpscope/FrameDecoder.h index a00e8b4..5f7e382 100644 --- a/Client/udpscope/FrameDecoder.h +++ b/Client/udpscope/FrameDecoder.h @@ -50,7 +50,12 @@ private: ClockOffset offset; double lastPacketWall = 0.0; bool lastPacketValid = false; - double lastAccHrtSec = 0.0; + /** Raw ticks, not seconds — see the comment in the samplingRate == 0 + * branch for why a tick difference is the only safe way to measure a + * producer-side interval while the rate is still being re-estimated. */ + uint64_t lastAccHrt = 0u; + uint64_t hrtRef = 0u; + bool hrtRefValid = false; bool lastAccValid = false; uint32_t prevAccCount = 0; /** For accumulated scalars with a declared sampling rate: end timestamp diff --git a/Client/udpscope/TimeBase.h b/Client/udpscope/TimeBase.h index 513f6d6..c9a7f20 100644 --- a/Client/udpscope/TimeBase.h +++ b/Client/udpscope/TimeBase.h @@ -71,6 +71,15 @@ public: * from its own boot, not from the Unix epoch. Pass the result to * ClockOffset::map() to land it on the wall clock; latching that arbitrary * epoch difference is precisely what ClockOffset is for. + * + * @warning Never subtract two of these results to measure a short interval. + * The rate is refitted on every add() and wobbles by a few parts in 1e4, + * while hrt is already ~1e11 ticks by the time a scope attaches to a + * long-running producer — so the division carries that relative wobble + * multiplied by the entire elapsed epoch, tens of milliseconds of jitter on + * an absolute value. The jitter is common to both operands only if the rate + * did not change between them, which is exactly what it does. To measure an + * interval, difference the raw ticks and divide once by ticksPerSecond(). */ double toSeconds(uint64_t hrt) const; void reset(); diff --git a/Client/udpscope/tests/FrameDecoderTest.cpp b/Client/udpscope/tests/FrameDecoderTest.cpp index 63245ad..e1cde98 100644 --- a/Client/udpscope/tests/FrameDecoderTest.cpp +++ b/Client/udpscope/tests/FrameDecoderTest.cpp @@ -269,12 +269,101 @@ TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { dec.beginFrame(f); ASSERT_TRUE(dec.timestamps(f, 0, ts)); + /* Arrival (500.000) is behind our timeline, so there is nothing to spread + * into; the burst is drawn narrower instead, which starts to bleed the lead + * off while still moving strictly forwards. */ + EXPECT_NEAR(ts[0], 500.0909, 1e-9); EXPECT_GT(ts[0], prevEnd) << "resync stepped backwards over the previous burst"; for (size_t i = 1; i < ts.size(); i++) { EXPECT_GT(ts[i], ts[i - 1]); } } +// When the chain has to be abandoned but arrival lies just ahead of where the +// last burst ended, the correction is made by COMPRESSING this one burst rather +// than by stepping back. Rejecting the correction instead would be one-directional +// — `predicted` is never below lastEmittedEnd + dt — and a timeline running fast +// could then never be pulled back. +TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + + /* A counter gap far beyond any real outage: the prediction is unusable, and + * the arrival anchor (500.086) sits behind the previous burst end. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 500.095, 10, 900011u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + EXPECT_GT(ts[0], 500.090) << "compressed burst must still start after the last one"; + EXPECT_NEAR(ts[9], 500.095, 1e-9) << "and end exactly on arrival"; + EXPECT_NEAR(ts[1] - ts[0], 0.0005, 1e-9) << "spread over the available room"; +} + +// The whole point of compressing: a declared SamplingRate is a hand-written +// config value, and even a correct one is measured against the producer host's +// crystal, not ours. Tens of ppm of difference is certain over a long session, +// so the reconstructed timeline WILL run away from the wall clock. It has to be +// pulled back, and it has to stay monotonic while that happens. +TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { + FrameDecoder dec; + dec.setSignals({accSignal()}); /* declares 1 kHz */ + + /* The producer really runs 1 % fast: 10 samples take 9.9 ms of wall time, + * so a chain stepping the declared 10 ms per packet gains 0.1 ms every + * packet. This is the direction re-anchoring alone cannot fix: arrival is + * always BEHIND the chain, so anchoring on it would step backwards and is + * refused. Only compression pulls the timeline back. */ + double worstLead = 0.0; + double lastEnd = 0.0; + for (int p = 0; p < 20000; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double arrival = 500.0 + p * 0.0099; + const FrameView& f = + fb.build(0, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], lastEnd) << "timeline went backwards at packet " << p; + lastEnd = ts[i]; + } + worstLead = std::max(worstLead, ts[9] - arrival); + } + + /* Unchecked, 20000 packets at 0.1 ms each would put the trace 2 s ahead. */ + EXPECT_LT(worstLead, 0.6) << "timeline drifted " << worstLead << " s ahead"; +} + +// The C client de-duplicates fragments but not whole unfragmented updates, so a +// host subscribed on two interfaces sees each datagram twice. Emitting the +// repeat would double the values and advance time by a burst that never was. +TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + const double endBefore = ts[9]; + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& dup = fb.build(0, 500.1001, 10, 10u); /* counter 10 again */ + dec.beginFrame(dup); + EXPECT_FALSE(dec.timestamps(dup, 0, ts)); + + /* And the drop must not have disturbed the chain: the genuine next packet + * still lands one period after burst 10 ended. */ + const FrameView& next = fb.build(0, 500.109, 10, 11u); + dec.beginFrame(next); + ASSERT_TRUE(dec.timestamps(next, 0, ts)); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-9); +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; @@ -288,15 +377,17 @@ TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) for (int p = 0; p < 40; p++) { FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); - const double producerSec = 100.0 + p * 0.010; /* 10 ms per packet */ - /* Zero-mean arrival jitter, so the rate fit still converges but any - * single arrival GAP is wrong. Without it, uniform arrivals make - * packetBurst and the hrt path return the same number and the test - * cannot tell which branch produced it. The last packet's gap is - * 7 ms, which arrival-spanning would render as a 0.7 ms period. */ + /* 25 ms per packet, deliberately NOT 10: at 10 the expected 1 ms period + * equals kDefaultDt, so a decoder that never derived anything and just + * returned the default would pass a test named for the derivation. */ + const double producerSec = 100.0 + p * 0.025; + /* Zero-mean arrival jitter, so the rate fit still converges but no + * single arrival GAP is right. Without it, uniform arrivals make + * packetBurst and the hrt path return the same number by construction + * and the test cannot tell which branch answered. */ const double jitter[4] = {0.0, 0.003, 0.0, -0.003}; const FrameView& f = fb.build(static_cast(producerSec * ticks), - 700.0 + p * 0.010 + jitter[p % 4], 10, + 700.0 + p * 0.025 + jitter[p % 4], 10, static_cast(p + 1)); dec.beginFrame(f); std::vector ts; @@ -304,9 +395,10 @@ TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) } ASSERT_EQ(last.size(), 10u); - /* 10 ms of producer time across 10 samples is a 1 ms period, whatever the - * datagrams did on the way over. */ - EXPECT_NEAR(last[1] - last[0], 0.001, 1e-5); + /* 25 ms of producer time across 10 samples is a 2.5 ms period, whatever the + * datagrams did on the way over. Arrival-spanning the last gap (22 ms) + * would give 2.2 ms; defaulting would give 1 ms. */ + EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5); } // A PACKET burst has no per-element time at all. Elements span diff --git a/docs/superpowers/plans/2026-08-27-udpscope.md b/docs/superpowers/plans/2026-08-27-udpscope.md index 849fa6f..de8bafc 100644 --- a/docs/superpowers/plans/2026-08-27-udpscope.md +++ b/docs/superpowers/plans/2026-08-27-udpscope.md @@ -1609,24 +1609,27 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { } } -// ADDED in Task 4 review, together with two siblings. See the shipped -// tests/FrameDecoderTest.cpp for the full set — accSignal()/primeTenBursts() -// helpers plus: -// * AccumulatedScalarReinstatesLostPacketsFromTheCounterGap — counter 10 → -// 111 means 100 lost packets = exactly 1 s; the packet deliberately lands -// 200 ms off that truth so the test fails if the answer comes from arrival. -// * AccumulatedScalarResyncsOnArrivalWhenTheCounterSaysNothing — counter -// stuck at 0, so only the arrival backstop can recover. -// * AccumulatedScalarNeverStepsBackwardsWhenResyncing — a resync that would -// move a signal's timestamps into the past must be given up instead. -// FrameBuilder::build() gained a `counter` parameter for these; leaving it at -// zero, as the original harness did, hides the counter rules entirely. +// ADDED in the Task 4 review rounds. FrameBuilder::build() gained a `counter` +// parameter for these; leaving it at zero, as the original harness did, hides +// the counter rules entirely. Two shared helpers: +// +// /** 1 kHz accumulated scalar: 10 samples = 10 ms per packet. */ +// SignalMeta accSignal(); +// /** Ten contiguous bursts, counters 1..10, leaving ts[9] == 500.090. */ +// void primeTenBursts(FrameDecoder&, std::vector& ts, bool withCounter); + +// The counterweight to the test above. Chaining bursts to suppress arrival +// jitter is only safe if loss is accounted for. The wire says exactly how much +// is missing, so no estimate is needed — and this test deliberately makes +// arrival time a LIAR (200 ms off) to prove the reconstruction comes from the +// counter and not from when the packet landed. TEST(FrameDecoder, AccumulatedScalarReinstatesLostPacketsFromTheCounterGap) { FrameDecoder dec; dec.setSignals({accSignal()}); std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); + /* Counter 111 after 10: 100 packets lost, 1000 samples, exactly 1 s. */ FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 501.300, 10, 111u); @@ -1638,6 +1641,133 @@ TEST(FrameDecoder, AccumulatedScalarReinstatesLostPacketsFromTheCounterGap) { EXPECT_NEAR(ts[9], 501.100, 1e-9); } +// A producer that never advances the counter leaves nothing to reconstruct +// from. Arrival time is then the better of two bad answers. +TEST(FrameDecoder, AccumulatedScalarResyncsOnArrivalWhenTheCounterSaysNothing) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/false); + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 501.100, 10, 0u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + EXPECT_NEAR(ts[0], 501.091, 1e-9); + EXPECT_NEAR(ts[9], 501.100, 1e-9); +} + +// Re-anchoring must never move a signal's timestamps backwards: the ring, the +// trigger and the exporter all assume they increase. Here the counter claims a +// 20 s hole while the packet arrives BEFORE our timeline reached. +TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + const double prevEnd = ts[9]; + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 500.000, 10, 2010u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + /* Arrival (500.000) is behind our timeline, so there is nothing to spread + * into; the burst is drawn narrower instead, which starts to bleed the lead + * off while still moving strictly forwards. */ + EXPECT_NEAR(ts[0], 500.0909, 1e-9); + EXPECT_GT(ts[0], prevEnd) << "resync stepped backwards over the previous burst"; + for (size_t i = 1; i < ts.size(); i++) { + EXPECT_GT(ts[i], ts[i - 1]); + } +} + +// When the chain must be abandoned but arrival lies just ahead of where the last +// burst ended, the correction COMPRESSES this one burst rather than stepping +// back. Rejecting the correction instead would be one-directional — `predicted` +// is never below lastEmittedEnd + dt — so a fast timeline could never return. +TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + + /* A counter gap far beyond any real outage: the prediction is unusable, and + * the arrival anchor (500.086) sits behind the previous burst end. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 500.095, 10, 900011u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + EXPECT_GT(ts[0], 500.090) << "compressed burst must still start after the last one"; + EXPECT_NEAR(ts[9], 500.095, 1e-9) << "and end exactly on arrival"; + EXPECT_NEAR(ts[1] - ts[0], 0.0005, 1e-9) << "spread over the available room"; +} + +// The whole point of the bleed: a declared SamplingRate is a hand-written config +// value measured against the PRODUCER host's crystal, not ours. Tens of ppm of +// difference is certain over a long session, so the timeline WILL run away from +// the wall clock. It must be pulled back, and stay monotonic while that happens. +TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { + FrameDecoder dec; + dec.setSignals({accSignal()}); /* declares 1 kHz */ + + /* The producer really runs 1 % fast: 10 samples take 9.9 ms of wall time, + * so a chain stepping the declared 10 ms per packet gains 0.1 ms every + * packet. This is the direction re-anchoring alone cannot fix: arrival is + * always BEHIND the chain, so anchoring on it would step backwards and is + * refused. Only the bleed pulls the timeline back. */ + double worstLead = 0.0; + double lastEnd = 0.0; + for (int p = 0; p < 20000; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double arrival = 500.0 + p * 0.0099; + const FrameView& f = + fb.build(0, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], lastEnd) << "timeline went backwards at packet " << p; + lastEnd = ts[i]; + } + worstLead = std::max(worstLead, ts[9] - arrival); + } + + /* Unchecked, 20000 packets at 0.1 ms each would put the trace 2 s ahead. */ + EXPECT_LT(worstLead, 0.6) << "timeline drifted " << worstLead << " s ahead"; +} + +// The C client de-duplicates fragments but not whole unfragmented updates, so a +// host subscribed on two interfaces sees each datagram twice. Emitting the +// repeat would double the values and advance time by a burst that never was. +TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + const double endBefore = ts[9]; + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& dup = fb.build(0, 500.1001, 10, 10u); /* counter 10 again */ + dec.beginFrame(dup); + EXPECT_FALSE(dec.timestamps(dup, 0, ts)); + + /* And the drop must not have disturbed the chain: the genuine next packet + * still lands one period after burst 10 ended. */ + const FrameView& next = fb.build(0, 500.109, 10, 11u); + dec.beginFrame(next); + ASSERT_TRUE(dec.timestamps(next, 0, ts)); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-9); +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; @@ -1651,17 +1781,28 @@ TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) for (int p = 0; p < 40; p++) { FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); - const double producerSec = 100.0 + p * 0.010; /* 10 ms per packet */ + /* 25 ms per packet, deliberately NOT 10: at 10 the expected 1 ms period + * equals kDefaultDt, so a decoder that never derived anything and just + * returned the default would pass a test named for the derivation. */ + const double producerSec = 100.0 + p * 0.025; + /* Zero-mean arrival jitter, so the rate fit still converges but no + * single arrival GAP is right. Without it, uniform arrivals make + * packetBurst and the hrt path return the same number by construction + * and the test cannot tell which branch answered. */ + const double jitter[4] = {0.0, 0.003, 0.0, -0.003}; const FrameView& f = fb.build(static_cast(producerSec * ticks), - 700.0 + p * 0.010, 10); + 700.0 + p * 0.025 + jitter[p % 4], 10, + static_cast(p + 1)); dec.beginFrame(f); std::vector ts; if (dec.timestamps(f, 0, ts)) { last = ts; } } ASSERT_EQ(last.size(), 10u); - /* 10 ms of producer time across 10 samples is a 1 ms period. */ - EXPECT_NEAR(last[1] - last[0], 0.001, 1e-5); + /* 25 ms of producer time across 10 samples is a 2.5 ms period, whatever the + * datagrams did on the way over. Arrival-spanning the last gap (22 ms) + * would give 2.2 ms; defaulting would give 1 ms. */ + EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5); } // A PACKET burst has no per-element time at all. Elements span @@ -1802,9 +1943,21 @@ private: ClockOffset offset; double lastPacketWall = 0.0; bool lastPacketValid = false; - double lastAccHrtSec = 0.0; + /* Raw ticks, not seconds. HrtRateFit::toSeconds() divides an absolute + * tick count (~1e11 on a producer that has been up a while) by a rate + * refitted every packet, so its result carries the fit's few-parts-in- + * 1e4 wobble multiplied by the whole elapsed epoch — tens of ms of + * jitter on a value whose consecutive difference is a few ms. + * Differencing two toSeconds() results measures the wobble, not the + * interval. Difference the ticks and divide once instead. */ + uint64_t lastAccHrt = 0u; + uint64_t hrtRef = 0u; + bool hrtRefValid = false; bool lastAccValid = false; uint32_t prevAccCount = 0; + double lastEmittedEnd = 0.0; + uint32_t lastCounter = 0u; + bool lastEmittedValid = false; }; std::vector signals_; @@ -1827,6 +1980,25 @@ namespace udpscope { /** Fallback cycle period before the first inter-packet gap is known. */ static constexpr double kDefaultDt = 1.0e-3; +/** + * How far a chained burst prediction may sit from where arrival time says it + * should be before the chain is abandoned. A backstop only: the counter + * normally accounts for loss exactly. Same value and reasoning as + * ClockOffset::kRecalibThresholdS. + */ +static constexpr double kBurstResyncThresholdS = 0.5; + +/** + * Largest counter gap still read as a loss count. A producer restart returns + * the counter to zero and a reordered datagram wraps the unsigned gap to near + * 2^32; multiplying either by a sample count would fabricate centuries. + */ +static constexpr uint32_t kMaxCounterGap = 1000000u; + +/** Burst width, as a fraction of nominal, while a leading timeline is pulled + * back. See the sole use site. */ +static constexpr double kLeadBleedFactor = 0.9; + void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; state_.assign(signals_.size(), SigState{}); @@ -1866,7 +2038,8 @@ bool FrameDecoder::packetBurst(uint32_t idx, uint32_t nElems, double wallNow, bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, std::vector& tsOut) { tsOut.clear(); - if (idx >= signals_.size() || idx >= f.numSignals || f.counts == nullptr) { + if (idx >= signals_.size() || idx >= f.numSignals || + f.counts == nullptr || f.values == nullptr) { return false; } @@ -1877,7 +2050,22 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, const double wallNow = f.recvTime; SigState& st = state_[idx]; - const bool hasTimeSig = d.hasTimeSignal(f.numSignals); + /* A repeated counter is a duplicated datagram — the same update arriving + * twice because the host joined the multicast group on two interfaces, say. + * The C client de-duplicates FRAGMENTS only, so an unfragmented update + * reaches us intact both times; emitting it again would double the values + * and advance the timeline by a burst that never existed. Counter zero is + * excluded because a producer that never sets one leaves it there. */ + if (st.lastEmittedValid && f.counter != 0u && f.counter == st.lastCounter) { + return false; + } + + /* hasTimeSignal() bounds the index against the FRAME's signal count, but the + * time signal's type code is read from our own table, whose size is + * independent — a frame carrying more signals than the installed table + * (briefly possible after a CONFIG change) would otherwise read past it. */ + const bool hasTimeSig = d.hasTimeSignal(f.numSignals) && + d.timeSignalIdx < signals_.size(); const uint32_t tIdx = hasTimeSig ? d.timeSignalIdx : 0u; const double tScale = hasTimeSig ? TimeSignalScale(signals_[tIdx].typeCode) @@ -1939,11 +2127,14 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, const double arrivalAnchor = wallNow - static_cast(nElems - 1u) * dtDeclared; double base = arrivalAnchor; + double step = dtDeclared; if (st.lastEmittedValid) { /* Unsigned subtraction wraps, so this is right across the - * counter's own 2^32 rollover. */ + * counter's own 2^32 rollover. A gap far larger than any real + * outage is a producer restart or a reordered datagram, not a + * loss count; claim nothing and let the backstop decide. */ const uint32_t gap = f.counter - st.lastCounter; - const double lost = (gap > 1u) + const double lost = (gap > 1u && gap <= kMaxCounterGap) ? static_cast(gap - 1u) * static_cast(st.prevAccCount) : 0.0; @@ -1952,12 +2143,35 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, base = predicted; } if (base <= st.lastEmittedEnd) { - base = st.lastEmittedEnd + dtDeclared; + /* Re-anchoring would step backwards, which the ring, the + * trigger and the exporter all forbid. But simply rejecting + * the correction makes the backstop ONE-DIRECTIONAL: + * `predicted` is never below lastEmittedEnd + dt, so a + * timeline running FAST — certain over a long session, two + * hosts' crystals differ by tens of ppm — would drift ahead + * without bound. Correct without stepping back instead. */ + if (wallNow > st.lastEmittedEnd) { + /* Arrival is still ahead of us: start immediately after + * the last burst and spread this one out to arrival. + * One packet is drawn narrow; the timeline is in step. */ + step = (wallNow - st.lastEmittedEnd) / + static_cast(nElems); + base = st.lastEmittedEnd + step; + } else { + /* We have run PAST arrival, so there is no room to + * spread into and no single burst can remove the excess + * without going backwards. Bleed it off: draw every + * burst a fixed fraction narrower until the timeline is + * back inside the threshold. A 10 % squeeze outruns a + * tens-of-ppm crystal error by orders of magnitude. */ + step = dtDeclared * kLeadBleedFactor; + base = st.lastEmittedEnd + step; + } } } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { - tsOut[e] = base + static_cast(e) * dtDeclared; + tsOut[e] = base + static_cast(e) * step; } st.lastEmittedEnd = tsOut[nElems - 1u]; st.lastCounter = f.counter; @@ -1965,30 +2179,40 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, st.lastEmittedValid = true; return true; } - if (!hrtFit_.ready()) { + if (!hrtFit_.ready() || f.hrt == 0u) { return packetBurst(idx, nElems, wallNow, tsOut); } - const double hrtSec = hrtFit_.toSeconds(f.hrt); - const double base = st.offset.map(hrtSec, wallNow); + const double rate = hrtFit_.ticksPerSecond(); - double dt; - if (st.lastAccValid && st.prevAccCount > 0u && - hrtSec > st.lastAccHrtSec) { + /* Difference raw TICKS, never two toSeconds() results — see SigState. + * Anchoring on the first usable packet keeps the fit's wobble on the + * (short) interval since attach instead of on the producer's whole + * uptime. ClockOffset absorbs the arbitrary epoch that leaves behind + * exactly as it would absorb the producer's boot epoch. */ + if (!st.hrtRefValid) { + st.hrtRef = f.hrt; + st.hrtRefValid = true; + } + const double sinceRef = (f.hrt >= st.hrtRef) + ? static_cast(f.hrt - st.hrtRef) / rate + : -static_cast(st.hrtRef - f.hrt) / rate; + const double base = st.offset.map(sinceRef, wallNow); + + double dt = kDefaultDt; + if (st.lastAccValid && st.prevAccCount > 0u && f.hrt > st.lastAccHrt) { /* 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) / + dt = (static_cast(f.hrt - st.lastAccHrt) / rate) / static_cast(st.prevAccCount); - } else { - dt = kDefaultDt; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { tsOut[e] = base + static_cast(e) * dt; } - st.lastAccHrtSec = hrtSec; - st.lastAccValid = true; - st.prevAccCount = nElems; + st.lastAccHrt = f.hrt; + st.lastAccValid = true; + st.prevAccCount = nElems; return true; } @@ -2026,7 +2250,11 @@ Expected: PASS, 9 tests. 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 — since `HrtRateFit` regresses `hrt` against ARRIVAL time — is itself corrupted by the very bursts it would be asked to survive. Check instead that `lastEmittedEnd`, `lastCounter`, `prevAccCount` and `lastEmittedValid` are updated on every emitted burst. -Note for `AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared`: its arrivals carry zero-mean jitter on purpose. Under UNIFORM arrivals the hrt path and `packetBurst` return the same number by construction (the fit expresses `hrt` in arrival-clock seconds), so the test could not tell which branch answered. +Note for `AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared`: its arrivals carry zero-mean jitter on purpose. Under UNIFORM arrivals the hrt path and `packetBurst` return the same number by construction (the fit expresses `hrt` in arrival-clock seconds), so the test could not tell which branch answered. Its producer period is 25 ms, not 10 ms, for the same reason: at 10 ms the expected 1 ms answer equals `kDefaultDt`, so a decoder that derived nothing would pass. + +If that test returns exactly `kDefaultDt`, or a value that wanders between runs of different length, the cause is almost certainly a reintroduced `hrtFit_.toSeconds(a) - hrtFit_.toSeconds(b)`. `toSeconds()` divides an ABSOLUTE tick count by a rate refitted on every packet; a producer that has been up for a day is at ~1e11 ticks, so the fit's few-parts-in-1e4 wobble becomes tens of milliseconds of jitter on the result — larger than the interval being measured. Difference the raw ticks and divide once by `ticksPerSecond()`. + +If `AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever` fails at ~2 s, the lead bleed is not firing. Note that the "spread out to arrival" compression is unreachable in this case by construction: a LEADING timeline has `lastEmittedEnd > wallNow`, so there is no room to spread into. That branch handles only a bad prediction while arrival is still ahead; the leading case needs the `kLeadBleedFactor` path below it. - [ ] **Step 8: Commit**