From 440b805afd290d5a24a3f3a99143eaacaca05700 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 27 Aug 2026 22:43:27 +0200 Subject: [PATCH] fix(udpscope): make rule 3 converge in both branches and survive reordering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eight review findings on FrameDecoder's accumulated-scalar rule. The squeeze that pulls a leading timeline back was expressed as a fraction of the NOMINAL burst width, which cannot converge: inside one timestamps() call the wall clock is frozen, so any positive step raises the lead measured at that instant, and the lead only falls because the wall advances between packets. At the kMinBleedFactor floor the timeline still gained 0.05 * nominal per packet, so a declared SamplingRate of 30 against a producer really flushing 10 samples at 1 kHz ran away without bound (667 s of lead after 1000 s of stream). Cap the burst's total advance at half the wall time really elapsed since this signal's previous burst instead, and the lead strictly falls for any declared rate. SigState gained lastEmittedWall for that reference; lastPacketWall could not be reused because it belongs to packetBurst. The hrt branch contributed zero elapsed for a late datagram but still wrote the hrt reference back to it, so the next packet's delta spanned two intervals and fabricated a whole extra packet of producer time — permanently, since the monotonic clamp discards the correction ClockOffset would have made. Reordering is reachable in production: udps_client.c only counts counter gaps. Simply never regressing the reference is not the fix either, because a producer restart would then freeze the signal forever, so the two are now separated by the size of the backward jump. The hrt branch's clamp was also one-directional, reintroducing on that branch exactly the defect the declared branch's squeeze exists to prevent: a backward wall step (NTP, suspend/resume) left a permanent lead. It now shares the same wall-elapsed cap. Also: anchor an hrt-branch burst's LAST element on arrival, matching the declared branch, so two accumulated scalars in one scope do not sit a burst apart on the shared X axis; treat a non-finite samplingRate off the wire as undeclared, since +inf produced a 0.0/0.0 factor the floor could not catch and turned every stamp NaN; write lastCounter on the hrt branch so duplicate datagrams are dropped there too; and correct two comments that argued for the current code with claims that are false (a counter-gap clamp reaches the opposite outcome, not the same one earlier, and the squeeze's steady state is a sawtooth, not a fixed offset). Seven new tests, each proven non-vacuous by sabotage; 54 pass. Plan document Task 4 re-synced and its stale test count and "agree on the same stream" claim corrected. --- Client/udpscope/FrameDecoder.cpp | 227 ++++- Client/udpscope/FrameDecoder.h | 53 +- Client/udpscope/tests/FrameDecoderTest.cpp | 341 ++++++- docs/superpowers/plans/2026-08-27-udpscope.md | 862 +++++++++++++++--- 4 files changed, 1317 insertions(+), 166 deletions(-) diff --git a/Client/udpscope/FrameDecoder.cpp b/Client/udpscope/FrameDecoder.cpp index 5337ce7..241b022 100644 --- a/Client/udpscope/FrameDecoder.cpp +++ b/Client/udpscope/FrameDecoder.cpp @@ -30,6 +30,59 @@ static constexpr double kBurstResyncThresholdS = 0.5; */ static constexpr double kMinBleedFactor = 0.05; +/** + * Largest share of the wall time elapsed since a signal's previous burst that + * that signal's next burst may advance its own timeline by, while a lead is + * being pulled back. + * + * This, and not kMinBleedFactor, is what makes a lead converge. A fraction of + * the NOMINAL burst width cannot: the floor still advances the timeline by + * kMinBleedFactor * nominal per packet while the wall advances one packet + * interval, so it diverges outright whenever nominal exceeds + * (1 / kMinBleedFactor) packet intervals — a declared SamplingRate of 30 against + * a producer really flushing 10 samples at 1 kHz put the trace 667 s ahead after + * 1000 s of stream. Measuring the allowance against elapsed WALL time instead + * bounds the advance below the wall's own advance for any declared rate, so the + * lead strictly falls whatever the config says. Any fraction under 1 converges; + * a half both converges quickly and leaves the burst visibly compressed rather + * than frozen. + */ +static constexpr double kWallBleedFraction = 0.5; + +/** + * A backward jump in producer hrt larger than this is a producer RESTART; a + * smaller one is a reordered datagram. + * + * The two need opposite handling — a reorder must leave the hrt reference + * untouched (its interval was already counted by the packet that overtook it), + * a restart must rebase onto the new epoch or the signal never advances again — + * and nothing but the size of the jump distinguishes them. A second of producer + * time is orders of magnitude more than any reordering window a UDP path can + * produce (a few packet intervals) and orders of magnitude less than a restart, + * which drops hrt from the producer's whole uptime back to near zero. + * + * Deliberately NOT kBurstResyncThresholdS: that one asks how far a WALL-clock + * prediction may sit from arrival, a different quantity in a different clock + * that happens to be tuned for delivery jitter. Sharing the number would couple + * two unrelated tunings. + */ +static constexpr double kProducerRestartS = 1.0; + +/** + * The declared sampling rate, or 0 when there is none to trust. + * + * samplingRate arrives unvalidated from a signal descriptor on the wire. A + * malformed +inf reaches the reciprocal as dt == 0, which makes a burst's + * nominal width zero and the proportional squeeze compute 0.0/0.0 — and a NaN + * factor is not caught by the floor, since every comparison against NaN is + * false, so the whole burst is emitted as NaN. Rejecting it at the boundary + * costs one test and removes the entire class. + */ +static double DeclaredRate(double samplingRate) { + return (std::isfinite(samplingRate) && samplingRate > 0.0) ? samplingRate + : 0.0; +} + void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; state_.assign(signals_.size(), SigState{}); @@ -120,7 +173,8 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, if ((d.timeMode == kTimeFirstSample || d.timeMode == kTimeLastSample) && hasTimeSig && f.counts[tIdx] >= 1u && f.values[tIdx] != nullptr) { const double anchor = st.offset.map(f.values[tIdx][0] * tScale, wallNow); - const double dt = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; + const double rate = DeclaredRate(d.samplingRate); + const double dt = (rate > 0.0) ? (1.0 / rate) : 0.0; tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { tsOut[e] = (d.timeMode == kTimeFirstSample) @@ -151,11 +205,12 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * datagram and reinstated a hole that never existed. A signal that has never * burst is a genuine scalar and is left to rule 5. */ if (d.numElements() == 1u && (nElems > 1u || st.lastEmittedValid)) { - const double dt = (d.samplingRate > 0.0) - ? (1.0 / d.samplingRate) - : 0.0; + /* A rate that is not a finite positive number is no rate at all; see + * DeclaredRate(). Such a signal takes the hrt branch below. */ + const double declared = DeclaredRate(d.samplingRate); + const double dt = (declared > 0.0) ? (1.0 / declared) : 0.0; - if (d.samplingRate > 0.0) { + if (declared > 0.0) { /* Where arrival time says this burst begins: its last element was * acquired just before the packet landed. */ const double arrivalAnchor = @@ -181,9 +236,23 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * A producer restart or a reordered datagram makes the wrapped * gap enormous, and this deliberately does NOT special-case * that: an absurd gap yields an absurd prediction, which the - * arrival backstop below then rejects on its own. Clamping the - * gap first would only decide the same question earlier, by a - * second rule that no stream can distinguish from this one. */ + * arrival backstop below then rejects on its own. + * + * Clamping the gap first is not a harmless earlier version of + * the same decision — it reaches the OPPOSITE answer. A clamp + * that treats gap > kMaxCounterGap as unknowable has to fall + * back to lost == 0, so the prediction becomes + * lastEmittedEnd + dt: one sample period after the last burst, + * which is exactly the shape of a healthy chain and therefore + * lands INSIDE the arrival backstop, is accepted, and silently + * closes an outage of arbitrary length. Letting the absurd gap + * through produces an absurd prediction that the backstop + * catches, and the burst re-anchors on arrival — which is the + * right answer, and what + * AccumulatedScalarSurvivesAProducerRestart pins down. The + * arithmetic cannot overflow: gap and prevAccCount are both + * bounded by 2^32-1, so lost is at most ~1.8e19, finite, and + * always rejected. */ const uint32_t gap = f.counter - st.lastCounter; const double lost = (gap > 1u) ? static_cast(gap - 1u) * @@ -223,24 +292,52 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * dated later than the moment this packet landed, so * there is no room to spread into and no burst can end * on arrival without starting before it. Squeeze this - * one by exactly the excess instead. That lands its end - * one nominal burst ahead of arrival — the closest a - * forward-only timeline can legally get — and the excess - * settles at (nominal width - true burst period), a - * couple of hundred microseconds for the ppm-scale - * crystal mismatch that causes this. + * one instead, by the excess and by the wall time that + * has really elapsed since this signal's last burst. * - * The floor keeps the step positive when the excess is - * larger than a whole burst (a declared rate that is - * wrong by a factor, not by ppm). It only slows the - * recovery: each burst then advances by almost nothing - * while arrival keeps advancing, so the excess still - * falls to zero, just over several packets. */ + * Both terms are needed, and only the second one + * converges. Within this call the wall clock is frozen + * at wallNow, so ANY positive step increases the lead + * measured at this instant; the lead falls only because + * the wall advances BETWEEN packets. A step expressed + * purely as a fraction of the nominal burst width + * therefore diverges as soon as the nominal width + * outruns the packet interval — with the floor alone, a + * declared 30 Hz against a producer really flushing 10 + * samples at 1 kHz gained ~0.67 s of lead per second of + * stream, without bound. Capping the burst's total + * advance at kWallBleedFraction of the elapsed wall time + * makes it advance strictly slower than the wall for any + * declared rate, so the lead strictly falls. + * + * The proportional term still does the fine work: when + * the excess is smaller than a burst it removes it in + * one packet. kMinBleedFactor only keeps that term + * positive when the excess exceeds a whole burst. + * + * Steady state is a sawtooth, not a fixed offset: the + * squeeze pulls the lead down, ordinary chaining resumes + * on the very next packet and pushes it back up until + * the prediction misses arrival by more than + * kBurstResyncThresholdS. So the lead cycles between a + * fraction of a millisecond and roughly that threshold — + * bounded, which is what matters, but not zero. */ const double nominal = static_cast(nElems) * dt; const double excess = st.lastEmittedEnd - wallNow; double factor = 1.0 - excess / nominal; if (factor < kMinBleedFactor) { factor = kMinBleedFactor; } - step = dt * factor; + double advance = nominal * factor; + + const double wallElapsed = wallNow - st.lastEmittedWall; + if (wallElapsed > 0.0) { + const double cap = kWallBleedFraction * wallElapsed; + if (cap < advance) { advance = cap; } + } + step = advance / static_cast(nElems); + /* Unreachable with a finite positive dt — kept because + * downstream monotonicity must not depend on that + * argument holding for every value off the wire. */ + if (!(step > 0.0)) { step = dt * kMinBleedFactor; } base = st.lastEmittedEnd + step; } } @@ -250,6 +347,7 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, tsOut[e] = base + static_cast(e) * step; } st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedWall = wallNow; st.lastCounter = f.counter; st.prevAccCount = nElems; st.lastEmittedValid = true; @@ -279,11 +377,47 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * arbitrary epoch that leaves behind, exactly as it would have latched * the producer's boot epoch. */ double elapsed = 0.0; - if (st.lastAccValid && f.hrt > st.lastAccHrt) { - elapsed = static_cast(f.hrt - st.lastAccHrt) / rate; + /* Whether lastAccHrt should take this packet's value. Only a packet + * that legitimately defines the new front of producer time may move it; + * see the backward case below. */ + bool takeHrt = true; + if (st.lastAccValid) { + if (f.hrt > st.lastAccHrt) { + elapsed = static_cast(f.hrt - st.lastAccHrt) / rate; + } else { + /* hrt went backwards. Two entirely different events look like + * this and only the SIZE of the jump separates them. + * + * A small one is a reordered datagram: the packet that overtook + * it already counted the interval it covers, so it must + * contribute nothing — and must also leave lastAccHrt alone. + * Letting it write lastAccHrt anyway (which is what this code + * used to do unconditionally) rolls the reference back one + * interval, so the NEXT packet's delta spans two and fabricates + * a whole extra packet of producer time. It never heals: + * ClockOffset would correct it, but the monotonic clamp below + * discards every backward correction. A hundred swaps on a + * 25 ms stream left the trace 3.5 s ahead, permanently. The C + * client does not reorder for us — udps_client.c only COUNTS + * counter gaps — so this is reachable on any real network. + * + * A large one is a producer restart: hrt drops from the + * producer's whole uptime to near zero. Here the unconditional + * write was the right behaviour and must be kept, because + * refusing to regress would leave every subsequent packet below + * lastAccHrt forever, elapsed permanently zero and the signal + * frozen. Rebase, and reset the offset so it re-latches against + * the new epoch instead of being dragged there by recalibration. */ + const double backward = + static_cast(st.lastAccHrt - f.hrt) / rate; + if (backward > kProducerRestartS) { + st.offset.reset(); + } else { + takeHrt = false; + } + } } st.accProdSec += elapsed; - double base = st.offset.map(st.accProdSec, wallNow); /* The flushes carry contiguous RT cycles, so the gap divided by the * previous packet's sample count is exactly one cycle period. */ @@ -291,21 +425,58 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, ? (elapsed / static_cast(st.prevAccCount)) : kDefaultDt; + /* Anchor the burst's LAST element on arrival, not its first. The + * packet's hrt is the tick count of sample 0 (UDPSourceSession.cpp:574), + * so stepping forward from it is right — but ClockOffset latches + * offset = wall - producerSec on its first call, and passing the raw + * arrival would put sample 0 at the instant the packet carrying the + * whole burst LANDED, dating every sample in it late by a burst. The + * declared-rate branch above already anchors on the burst end + * (arrivalAnchor), and two accumulated scalars in one scope, one with a + * declared rate and one without, would otherwise sit a burst apart on a + * shared X axis — 9 ms for 10 samples at 1 kHz, plain to see at a 200 ms + * window. Since map() latches once, this is a constant shift applied at + * latch and recalibration only; it changes no spacing. */ + double base = st.offset.map( + st.accProdSec, + wallNow - static_cast(nElems - 1u) * hrtDt); + double step = hrtDt; + /* ClockOffset recalibrates once true drift passes its threshold, and a * recalibration can land behind where this signal already is. - * Downstream requires increasing stamps, so step forward minimally. */ + * Downstream requires increasing stamps, so step forward minimally — + * but a bare forward step is one-directional, exactly the defect the + * declared branch's squeeze exists to avoid. A backward wall step (an + * NTP correction, a suspend/resume) would otherwise leave this signal + * permanently ahead of the wall clock, since the recalibrated base is + * behind lastEmittedEnd on every later packet too and the clamp keeps + * discarding it. So cap the burst's total advance against the wall time + * elapsed since this signal's previous burst, for the reason spelled out + * at kWallBleedFraction: only that makes the lead bleed off. */ if (st.lastEmittedValid && base <= st.lastEmittedEnd) { - base = st.lastEmittedEnd + hrtDt; + const double wallElapsed = wallNow - st.lastEmittedWall; + if (wallElapsed > 0.0) { + const double cap = kWallBleedFraction * wallElapsed / + static_cast(nElems); + if (cap < step) { step = cap; } + } + base = st.lastEmittedEnd + step; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { - tsOut[e] = base + static_cast(e) * hrtDt; + tsOut[e] = base + static_cast(e) * step; } - st.lastAccHrt = f.hrt; + if (takeHrt) { st.lastAccHrt = f.hrt; } st.lastAccValid = true; st.prevAccCount = nElems; st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedWall = wallNow; + /* Same duplicate-datagram exposure as the declared branch: a host joined + * on two interfaces receives every unfragmented update twice, and the + * guard at the top of timestamps() can only fire if this branch leaves a + * counter behind for it to compare against. */ + st.lastCounter = f.counter; st.lastEmittedValid = true; return true; } diff --git a/Client/udpscope/FrameDecoder.h b/Client/udpscope/FrameDecoder.h index f83f530..7f6f2d9 100644 --- a/Client/udpscope/FrameDecoder.h +++ b/Client/udpscope/FrameDecoder.h @@ -10,17 +10,30 @@ * Source/Applications/StreamHub/UDPSourceSession.cpp documents this failure and * solves it; these are the same rules, computed from udps_frame_t's own fields. * - * One rule deliberately differs. StreamHub anchors every accumulated-scalar - * burst on the packet's own hrt, converted with the LOCAL MARTe - * HighResolutionTimer frequency — correct only because StreamHub runs on the - * producer's host. A bench scope attaches over the network and has no access to - * that frequency; it can only regress hrt against arrival time, which is - * exactly what the bursty delivery above corrupts. So when a SamplingRate is - * declared this decoder chains bursts instead, using the packet counter to - * account for loss and arrival time only as a backstop. The consequence is that - * a declared rate measured against the producer's crystal rather than ours makes - * the reconstructed timeline drift, and drift that only arrival time can - * observe must be corrected against arrival time — see rule 3. + * Two rules deliberately differ, both in the accumulated-scalar case (rule 3). + * + * First, the anchor. StreamHub anchors every accumulated-scalar burst on the + * packet's own hrt, converted with the LOCAL MARTe HighResolutionTimer + * frequency — correct only because StreamHub runs on the producer's host. A + * bench scope attaches over the network and has no access to that frequency; it + * can only regress hrt against arrival time, which is exactly what the bursty + * delivery above corrupts. So when a SamplingRate is declared this decoder + * chains bursts instead, using the packet counter to account for loss and + * arrival time only as a backstop. The consequence is that a declared rate + * measured against the producer's crystal rather than ours makes the + * reconstructed timeline drift, and drift that only arrival time can observe + * must be corrected against arrival time — see rule 3. + * + * Second, the entry condition. UDPSourceSession.cpp:554 routes any update + * carrying nElems <= 1 to plain arrival time. That is safe for a host-local + * consumer whose arrival time is the producer's own clock, but wrong here: + * Accumulate mode flushes on a TIMER, so a short RT cycle legitimately delivers + * a single sample between two full bursts. Dating that one sample from arrival + * while its neighbours are chained puts it off the chain, and — worse — leaves + * lastCounter behind, so the next full burst reads the skipped counter as a lost + * datagram and reinstates a hole that never existed. So a signal that has + * already burst keeps every later update on rule 3 regardless of its length; a + * signal that has never burst is a genuine scalar and is left to rule 5. */ #pragma once @@ -71,12 +84,22 @@ private: double accProdSec = 0.0; bool lastAccValid = false; uint32_t prevAccCount = 0; - /** For accumulated scalars with a declared sampling rate: end timestamp - * of the most recently emitted burst, and the packet counter it came - * from. The next burst is chained onto that end, with the counter gap - * reinstating the exact duration of any lost datagrams. */ + /** For accumulated scalars (rule 3, either branch): end timestamp of the + * most recently emitted burst, and the packet counter it came from. The + * next burst is chained onto that end, with the counter gap reinstating + * the exact duration of any lost datagrams. */ double lastEmittedEnd = 0.0; uint32_t lastCounter = 0u; + /** ARRIVAL time of the packet that produced lastEmittedEnd. Valid + * exactly when lastEmittedValid is, so it needs no flag of its own. + * Deliberately not lastPacketWall, which belongs to packetBurst() and + * is updated on frames rule 3 never emits. This is the only reference + * against which a leading timeline can be pulled back: the correction + * has to be expressed as a fraction of the wall time that has really + * elapsed since this signal's previous burst, because within a single + * timestamps() call the wall clock is frozen and every forward step, + * however small, increases the lead measured at that instant. */ + double lastEmittedWall = 0.0; bool lastEmittedValid = false; }; diff --git a/Client/udpscope/tests/FrameDecoderTest.cpp b/Client/udpscope/tests/FrameDecoderTest.cpp index 768da6f..078e9ba 100644 --- a/Client/udpscope/tests/FrameDecoderTest.cpp +++ b/Client/udpscope/tests/FrameDecoderTest.cpp @@ -2,6 +2,8 @@ #include +#include +#include #include using namespace udpscope; @@ -346,6 +348,91 @@ TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { EXPECT_LT(worstLead, 0.6) << "timeline drifted " << worstLead << " s ahead"; } +// The test above only exercises a rate that is wrong by ppm, where the squeeze's +// proportional term does all the work. A rate wrong by a FACTOR is the case the +// kMinBleedFactor floor cannot handle on its own: at the floor the timeline +// still advances kMinBleedFactor * nominal per packet, so whenever the nominal +// burst is wider than 1/kMinBleedFactor packet intervals the lead grows without +// bound rather than bleeding off (measured: 27 s of lead after 40 s of stream, +// 667 s after 1000 s). Only capping the advance against the wall time really +// elapsed since this signal's previous burst converges for every declared rate. +TEST(FrameDecoder, AccumulatedScalarConvergesWhenTheDeclaredRateIsFarTooLow) { + FrameDecoder dec; + SignalMeta m = accSignal(); + m.samplingRate = 30.0; /* config says 30 Hz... */ + dec.setSignals({m}); + + /* ...while the producer really flushes 10 samples at 1 kHz, so a packet is + * 10 ms of wall time and 333 ms of nominal, declared time. */ + double worstLead = 0.0; + double last = 0.0; + for (int p = 0; p < 5000; p++) { /* 50 s of stream */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double arrival = 500.0 + p * 0.010; + 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], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + worstLead = std::max(worstLead, ts[9] - arrival); + } + + /* Bounded, not zero: normal chaining resumes the moment the squeeze stops, + * so the lead sawtooths up to about kBurstResyncThresholdS and back. */ + EXPECT_LT(worstLead, 1.0) << "timeline ran " << worstLead << " s ahead"; +} + +// samplingRate is unvalidated wire data. A malformed +inf makes the declared +// period zero, so a burst's nominal width is zero and the proportional squeeze +// evaluates 0.0/0.0 — and a NaN factor slips past the floor, because every +// comparison against NaN is false. The burst, and then every burst after it, +// comes out NaN. Treating a non-finite rate as no rate at all removes the class. +TEST(FrameDecoder, AccumulatedScalarWithANonFiniteRateFallsBackToTheHrtPath) { + FrameDecoder dec; + SignalMeta m = accSignal(); + m.samplingRate = std::numeric_limits::infinity(); + dec.setSignals({m}); + + const double ticks = 1.0e9; + std::vector last; + for (int p = 0; p < 60; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double producerSec = 100.0 + p * 0.025; + /* Packet 40 lands at exactly the same instant as packet 39. With the + * degenerate zero period the previous burst ends precisely on its own + * arrival, so this makes the squeeze's excess exactly zero — the 0.0/0.0 + * that produces the NaN. */ + const int q = (p == 40) ? 39 : p; + /* Zero-mean jitter so the answer also identifies WHICH branch replied: + * a degenerate declared branch spans the jittered arrival gap, the hrt + * branch returns the producer's exact 2.5 ms whatever delivery did. */ + const double jitter[4] = {0.0, 0.003, 0.0, -0.003}; + const double arrival = 700.0 + q * 0.025 + jitter[q % 4]; + const FrameView& f = + fb.build(static_cast(producerSec * ticks), arrival, 10, + static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_TRUE(std::isfinite(ts[i])) + << "packet " << p << " element " << i; + } + last = ts; + } + } + + ASSERT_EQ(last.size(), 10u); + EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5) + << "an unusable declared rate must fall through to the hrt path"; +} + // 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. @@ -367,7 +454,7 @@ TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { 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); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-6); } // A producer restart returns the counter to zero mid-stream. The unsigned gap @@ -514,6 +601,258 @@ TEST(FrameDecoder, AccumulatedScalarStaysMonotonicOnALongUndeclaredRunAfterBoot) } } +namespace { + +/** One delivered datagram of an undeclared-rate accumulated scalar. */ +struct HrtPacket { + uint64_t hrt; + double arrival; + uint32_t counter; +}; + +SignalMeta undeclaredAcc() { + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 0.0; /* undeclared: the hrt branch */ + return m; +} + +/** Runs a delivery schedule of 10-sample bursts; returns the last stamp emitted. */ +double runUndeclared(const std::vector& pkts) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + double lastTs = 0.0; + for (const HrtPacket& p : pkts) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(p.hrt, p.arrival, 10, p.counter); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { lastTs = ts.back(); } + } + return lastTs; +} + +/** 300 clean packets, 25 ms apart, from a producer that has been up a day. */ +std::vector cleanUndeclaredStream() { + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + std::vector pkts; + for (int p = 0; p < 300; p++) { + pkts.push_back(HrtPacket{ + bootHrt + static_cast(p * 0.025 * ticks), + 700.0 + p * 0.025, + static_cast(p + 1)}); + } + return pkts; +} + +} /* namespace */ + +// A datagram that overtakes its neighbour arrives with an hrt BEHIND the one +// already recorded. It must contribute no producer time — the packet that +// overtook it already counted the interval — and it must also leave the hrt +// reference alone. Writing the reference back is what the code used to do, and +// it makes the NEXT packet's delta span two intervals, fabricating a whole extra +// packet of producer time per reorder. That error never heals: ClockOffset would +// correct it but the monotonic clamp discards every backward correction. +TEST(FrameDecoder, UndeclaredAccumulatedScalarIgnoresReorderedDatagrams) { + const std::vector clean = cleanUndeclaredStream(); + + /* Ten swaps: each pair is delivered in the opposite order, so the arrival + * times stay increasing (delivery order is what the socket saw) while the + * hrt and counter they carry are exchanged. */ + std::vector reordered = clean; + for (int k = 100; k < 200; k += 10) { + std::swap(reordered[k].hrt, reordered[k + 1].hrt); + std::swap(reordered[k].counter, reordered[k + 1].counter); + } + + const double cleanEnd = runUndeclared(clean); + const double reorderedEnd = runUndeclared(reordered); + + /* Each swap used to add about one packet of producer time (25 ms); ten of + * them left the trace a quarter of a second ahead, for good. */ + EXPECT_NEAR(reorderedEnd, cleanEnd, 1.0e-3) + << "reordering left " << (reorderedEnd - cleanEnd) << " s of offset"; +} + +// The counterweight. A producer restart drops hrt from the machine's whole +// uptime back to near zero, and that is the one case where the hrt reference +// MUST be allowed to regress: refusing every backward step would leave each +// later packet below the reference forever, the elapsed producer time +// permanently zero, and the signal frozen at the fallback period. +TEST(FrameDecoder, UndeclaredAccumulatedScalarSurvivesAProducerRestart) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + double last = 0.0; + for (int p = 0; p < 120; p++) { + const bool restarted = (p >= 60); + /* After the restart hrt counts from one second of uptime, and the + * outage cost two seconds of wall time. */ + const uint64_t hrt = restarted + ? static_cast((1.0 + (p - 60) * 0.025) * ticks) + : bootHrt + static_cast(p * 0.025 * ticks); + const double arrival = restarted + ? (700.0 + 59 * 0.025 + 2.0 + (p - 60) * 0.025) + : (700.0 + p * 0.025); + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { + /* Only the very first packet, which has no previous arrival for the + * pre-fit fallback to span from. */ + ASSERT_EQ(p, 0) << "packet " << p << " produced nothing"; + continue; + } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + /* The restart packet itself has no measurable interval and falls back to + * the default period; from the next one on the producer's own 2.5 ms + * must be back. A decoder that could not regress the reference would sit + * at the 1 ms fallback for the rest of the run. */ + if (p >= 62) { + EXPECT_NEAR(ts[1] - ts[0], 0.0025, 1e-5) + << "spacing not recovered at packet " << p; + } + } +} + +// The hrt branch's clamp used to be one-directional, which is the same defect +// the declared branch's squeeze exists to prevent. A wall clock that steps +// BACKWARDS — an NTP correction, a suspend/resume — leaves the emitted timeline +// permanently ahead, because the recalibrated position is behind lastEmittedEnd +// on every later packet too and the clamp keeps discarding it. +TEST(FrameDecoder, UndeclaredAccumulatedScalarRecoversFromABackwardWallStep) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + double last = 0.0; + double lead = 0.0; + double worstAfter = 0.0; + /* 100 ms packets of 10 samples. The step is 0.6 s — just past + * ClockOffset::kRecalibThresholdS, which is what makes the recalibrated + * position land behind lastEmittedEnd and the clamp fire at all — and it + * comes after the rate fit's 256-sample window is full, so the fit + * redistributes it slowly enough not to be mistaken for this recovery. */ + for (int p = 0; p < 340; p++) { + const uint64_t hrt = bootHrt + static_cast(p * 0.1 * ticks); + const double arrival = 700.0 + p * 0.1 - ((p >= 300) ? 0.6 : 0.0); + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { + ASSERT_EQ(p, 0) << "packet " << p << " produced nothing"; + continue; + } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + lead = ts.back() - arrival; + /* Twenty packets is a generous allowance: the cap bleeds half a packet + * interval per packet, so the 0.6 s step is gone in twelve. */ + if (p >= 320) { worstAfter = std::max(worstAfter, std::fabs(lead)); } + } + + EXPECT_LT(worstAfter, 0.1) + << "still " << worstAfter << " s from the wall clock long after the step"; +} + +// The two branches must place a burst the same way round or two accumulated +// scalars in one scope, one with a declared rate and one without, sit a whole +// burst apart on the shared X axis. The declared branch anchors the LAST element +// on arrival, which is right: the samples were acquired before the packet +// carrying them landed. The hrt branch used to latch its offset against raw +// arrival, putting the FIRST element there instead. +TEST(FrameDecoder, UndeclaredAccumulatedScalarEndsItsBurstOnArrival) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + /* 10 ms per packet of 10 samples, so the derived period is 1 ms — equal to + * the fallback the very first hrt-branch packet has to use, which is what + * ClockOffset latches against. Any other period would bake that one packet's + * fallback into the offset and blur the convention this test is pinning. */ + double lastArrival = 0.0; + std::vector last; + for (int p = 0; p < 60; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const uint64_t hrt = bootHrt + static_cast(p * 0.010 * ticks); + const double arrival = 700.0 + p * 0.010; + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { last = ts; lastArrival = arrival; } + } + + ASSERT_EQ(last.size(), 10u); + EXPECT_NEAR(last[9], lastArrival, 1e-9) << "burst must END on arrival"; + EXPECT_NEAR(last[0], lastArrival - 0.009, 1e-9); +} + +// The same double delivery that the declared branch guards against — a host +// joined on two interfaces receives every unfragmented update twice — reaches an +// undeclared-rate signal identically. The guard can only fire if this branch +// leaves a counter behind for it to compare against. +TEST(FrameDecoder, UndeclaredAccumulatedScalarDropsADuplicatedDatagram) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + std::vector ts; + for (int p = 0; p < 50; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(bootHrt + static_cast(p * 0.010 * ticks), + 700.0 + p * 0.010, 10, static_cast(p + 1)); + dec.beginFrame(f); + const bool ok = dec.timestamps(f, 0, ts); + ASSERT_EQ(ok, p != 0) << "at packet " << p; + } + const double endBefore = ts[9]; + + /* Counter 50 again, the same update off the second interface. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& dup = + fb.build(bootHrt + static_cast(49 * 0.010 * ticks), + 700.0 + 49 * 0.010 + 0.0001, 10, 50u); + dec.beginFrame(dup); + EXPECT_FALSE(dec.timestamps(dup, 0, ts)) << "duplicate was emitted twice"; + + /* And the drop left the chain alone: the genuine next update still lands one + * period after the last burst ended. */ + const FrameView& next = + fb.build(bootHrt + static_cast(50 * 0.010 * ticks), + 700.0 + 50 * 0.010, 10, 51u); + dec.beginFrame(next); + ASSERT_TRUE(dec.timestamps(next, 0, ts)); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-6); +} + // A PACKET burst has no per-element time at all. Elements span // (lastPacket, thisPacket] — backwards from arrival, because the samples were // acquired before the packet landed. Forward extrapolation would let a jittered diff --git a/docs/superpowers/plans/2026-08-27-udpscope.md b/docs/superpowers/plans/2026-08-27-udpscope.md index c37a41c..6173cce 100644 --- a/docs/superpowers/plans/2026-08-27-udpscope.md +++ b/docs/superpowers/plans/2026-08-27-udpscope.md @@ -1434,6 +1434,8 @@ Create `Client/udpscope/tests/FrameDecoderTest.cpp`: #include +#include +#include #include using namespace udpscope; @@ -1451,13 +1453,17 @@ struct FrameBuilder { storage.push_back(std::move(vals)); } - const FrameView& build(uint64_t hrt, double recvTime, uint32_t numSamples = 1) { + /* Real frames carry a per-update counter; leaving it at zero would hide + * whichever rules depend on it, so it must be passed explicitly. */ + const FrameView& build(uint64_t hrt, double recvTime, uint32_t numSamples = 1, + uint32_t counter = 0) { ptrs.clear(); counts.clear(); for (const auto& s : storage) { ptrs.push_back(s.data()); counts.push_back(static_cast(s.size())); } + view.counter = counter; view.hrt = hrt; view.recvTime = recvTime; view.numSamples = numSamples; @@ -1593,7 +1599,7 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { const double arrival = (p < 20) ? (500.0 + p * 0.010) : (500.2 + (p - 20) * 0.00005); const FrameView& f = fb.build(static_cast(producerSec * ticks), - arrival, 10); + arrival, 10, static_cast(p + 1)); dec.beginFrame(f); std::vector ts; if (dec.timestamps(f, 0, ts)) { @@ -1609,27 +1615,46 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { } } -// 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); +namespace { + +/** Ten contiguous 10-sample bursts at 1 kHz, counters 1..10, ending at 500.090. */ +SignalMeta accSignal() { + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 1000.0; /* 10 samples = 10 ms per packet */ + return m; +} + +void primeTenBursts(FrameDecoder& dec, std::vector& ts, bool withCounter) { + 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, + withCounter ? static_cast(p + 1) : 0u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + } + ASSERT_NEAR(ts[9], 500.090, 1e-9); +} + +} /* namespace */ // 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. +// jitter is only safe if loss is accounted for: a bare chain closes the hole a +// dropped datagram left and dates every later sample early for the rest of the +// run. 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. */ + /* Counter 111 after 10: 100 packets lost, 1000 samples, exactly 1 s. The + * packet lands 200 ms later than that truth would predict. */ FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 501.300, 10, 111u); @@ -1641,8 +1666,9 @@ 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. +// A producer that never advances the counter, or restarts it, leaves nothing to +// reconstruct from. Arrival time is then the better of two bad answers, and the +// chain has to be abandoned rather than left to drift forever. TEST(FrameDecoder, AccumulatedScalarResyncsOnArrivalWhenTheCounterSaysNothing) { FrameDecoder dec; dec.setSignals({accSignal()}); @@ -1660,8 +1686,10 @@ TEST(FrameDecoder, AccumulatedScalarResyncsOnArrivalWhenTheCounterSaysNothing) { } // 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. +// trigger and the exporter all assume they increase, and a backward step is +// indistinguishable from corruption downstream. Here the counter claims a +// 20 s hole while the packet arrives 10 ms after the last one, so the +// prediction and arrival disagree wildly and arrival points into the past. TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { FrameDecoder dec; dec.setSignals({accSignal()}); @@ -1686,10 +1714,11 @@ TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { } } -// 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. +// 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()}); @@ -1751,6 +1780,91 @@ TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { EXPECT_LT(worstLead, 0.6) << "timeline drifted " << worstLead << " s ahead"; } +// The test above only exercises a rate that is wrong by ppm, where the squeeze's +// proportional term does all the work. A rate wrong by a FACTOR is the case the +// kMinBleedFactor floor cannot handle on its own: at the floor the timeline +// still advances kMinBleedFactor * nominal per packet, so whenever the nominal +// burst is wider than 1/kMinBleedFactor packet intervals the lead grows without +// bound rather than bleeding off (measured: 27 s of lead after 40 s of stream, +// 667 s after 1000 s). Only capping the advance against the wall time really +// elapsed since this signal's previous burst converges for every declared rate. +TEST(FrameDecoder, AccumulatedScalarConvergesWhenTheDeclaredRateIsFarTooLow) { + FrameDecoder dec; + SignalMeta m = accSignal(); + m.samplingRate = 30.0; /* config says 30 Hz... */ + dec.setSignals({m}); + + /* ...while the producer really flushes 10 samples at 1 kHz, so a packet is + * 10 ms of wall time and 333 ms of nominal, declared time. */ + double worstLead = 0.0; + double last = 0.0; + for (int p = 0; p < 5000; p++) { /* 50 s of stream */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double arrival = 500.0 + p * 0.010; + 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], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + worstLead = std::max(worstLead, ts[9] - arrival); + } + + /* Bounded, not zero: normal chaining resumes the moment the squeeze stops, + * so the lead sawtooths up to about kBurstResyncThresholdS and back. */ + EXPECT_LT(worstLead, 1.0) << "timeline ran " << worstLead << " s ahead"; +} + +// samplingRate is unvalidated wire data. A malformed +inf makes the declared +// period zero, so a burst's nominal width is zero and the proportional squeeze +// evaluates 0.0/0.0 — and a NaN factor slips past the floor, because every +// comparison against NaN is false. The burst, and then every burst after it, +// comes out NaN. Treating a non-finite rate as no rate at all removes the class. +TEST(FrameDecoder, AccumulatedScalarWithANonFiniteRateFallsBackToTheHrtPath) { + FrameDecoder dec; + SignalMeta m = accSignal(); + m.samplingRate = std::numeric_limits::infinity(); + dec.setSignals({m}); + + const double ticks = 1.0e9; + std::vector last; + for (int p = 0; p < 60; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double producerSec = 100.0 + p * 0.025; + /* Packet 40 lands at exactly the same instant as packet 39. With the + * degenerate zero period the previous burst ends precisely on its own + * arrival, so this makes the squeeze's excess exactly zero — the 0.0/0.0 + * that produces the NaN. */ + const int q = (p == 40) ? 39 : p; + /* Zero-mean jitter so the answer also identifies WHICH branch replied: + * a degenerate declared branch spans the jittered arrival gap, the hrt + * branch returns the producer's exact 2.5 ms whatever delivery did. */ + const double jitter[4] = {0.0, 0.003, 0.0, -0.003}; + const double arrival = 700.0 + q * 0.025 + jitter[q % 4]; + const FrameView& f = + fb.build(static_cast(producerSec * ticks), arrival, 10, + static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_TRUE(std::isfinite(ts[i])) + << "packet " << p << " element " << i; + } + last = ts; + } + } + + ASSERT_EQ(last.size(), 10u); + EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5) + << "an unusable declared rate must fall through to the hrt path"; +} + // 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. @@ -1772,7 +1886,7 @@ TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { 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); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-6); } // A producer restart returns the counter to zero mid-stream. The unsigned gap @@ -1919,6 +2033,258 @@ TEST(FrameDecoder, AccumulatedScalarStaysMonotonicOnALongUndeclaredRunAfterBoot) } } +namespace { + +/** One delivered datagram of an undeclared-rate accumulated scalar. */ +struct HrtPacket { + uint64_t hrt; + double arrival; + uint32_t counter; +}; + +SignalMeta undeclaredAcc() { + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 0.0; /* undeclared: the hrt branch */ + return m; +} + +/** Runs a delivery schedule of 10-sample bursts; returns the last stamp emitted. */ +double runUndeclared(const std::vector& pkts) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + double lastTs = 0.0; + for (const HrtPacket& p : pkts) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(p.hrt, p.arrival, 10, p.counter); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { lastTs = ts.back(); } + } + return lastTs; +} + +/** 300 clean packets, 25 ms apart, from a producer that has been up a day. */ +std::vector cleanUndeclaredStream() { + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + std::vector pkts; + for (int p = 0; p < 300; p++) { + pkts.push_back(HrtPacket{ + bootHrt + static_cast(p * 0.025 * ticks), + 700.0 + p * 0.025, + static_cast(p + 1)}); + } + return pkts; +} + +} /* namespace */ + +// A datagram that overtakes its neighbour arrives with an hrt BEHIND the one +// already recorded. It must contribute no producer time — the packet that +// overtook it already counted the interval — and it must also leave the hrt +// reference alone. Writing the reference back is what the code used to do, and +// it makes the NEXT packet's delta span two intervals, fabricating a whole extra +// packet of producer time per reorder. That error never heals: ClockOffset would +// correct it but the monotonic clamp discards every backward correction. +TEST(FrameDecoder, UndeclaredAccumulatedScalarIgnoresReorderedDatagrams) { + const std::vector clean = cleanUndeclaredStream(); + + /* Ten swaps: each pair is delivered in the opposite order, so the arrival + * times stay increasing (delivery order is what the socket saw) while the + * hrt and counter they carry are exchanged. */ + std::vector reordered = clean; + for (int k = 100; k < 200; k += 10) { + std::swap(reordered[k].hrt, reordered[k + 1].hrt); + std::swap(reordered[k].counter, reordered[k + 1].counter); + } + + const double cleanEnd = runUndeclared(clean); + const double reorderedEnd = runUndeclared(reordered); + + /* Each swap used to add about one packet of producer time (25 ms); ten of + * them left the trace a quarter of a second ahead, for good. */ + EXPECT_NEAR(reorderedEnd, cleanEnd, 1.0e-3) + << "reordering left " << (reorderedEnd - cleanEnd) << " s of offset"; +} + +// The counterweight. A producer restart drops hrt from the machine's whole +// uptime back to near zero, and that is the one case where the hrt reference +// MUST be allowed to regress: refusing every backward step would leave each +// later packet below the reference forever, the elapsed producer time +// permanently zero, and the signal frozen at the fallback period. +TEST(FrameDecoder, UndeclaredAccumulatedScalarSurvivesAProducerRestart) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + double last = 0.0; + for (int p = 0; p < 120; p++) { + const bool restarted = (p >= 60); + /* After the restart hrt counts from one second of uptime, and the + * outage cost two seconds of wall time. */ + const uint64_t hrt = restarted + ? static_cast((1.0 + (p - 60) * 0.025) * ticks) + : bootHrt + static_cast(p * 0.025 * ticks); + const double arrival = restarted + ? (700.0 + 59 * 0.025 + 2.0 + (p - 60) * 0.025) + : (700.0 + p * 0.025); + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { + /* Only the very first packet, which has no previous arrival for the + * pre-fit fallback to span from. */ + ASSERT_EQ(p, 0) << "packet " << p << " produced nothing"; + continue; + } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + /* The restart packet itself has no measurable interval and falls back to + * the default period; from the next one on the producer's own 2.5 ms + * must be back. A decoder that could not regress the reference would sit + * at the 1 ms fallback for the rest of the run. */ + if (p >= 62) { + EXPECT_NEAR(ts[1] - ts[0], 0.0025, 1e-5) + << "spacing not recovered at packet " << p; + } + } +} + +// The hrt branch's clamp used to be one-directional, which is the same defect +// the declared branch's squeeze exists to prevent. A wall clock that steps +// BACKWARDS — an NTP correction, a suspend/resume — leaves the emitted timeline +// permanently ahead, because the recalibrated position is behind lastEmittedEnd +// on every later packet too and the clamp keeps discarding it. +TEST(FrameDecoder, UndeclaredAccumulatedScalarRecoversFromABackwardWallStep) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + double last = 0.0; + double lead = 0.0; + double worstAfter = 0.0; + /* 100 ms packets of 10 samples. The step is 0.6 s — just past + * ClockOffset::kRecalibThresholdS, which is what makes the recalibrated + * position land behind lastEmittedEnd and the clamp fire at all — and it + * comes after the rate fit's 256-sample window is full, so the fit + * redistributes it slowly enough not to be mistaken for this recovery. */ + for (int p = 0; p < 340; p++) { + const uint64_t hrt = bootHrt + static_cast(p * 0.1 * ticks); + const double arrival = 700.0 + p * 0.1 - ((p >= 300) ? 0.6 : 0.0); + + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { + ASSERT_EQ(p, 0) << "packet " << p << " produced nothing"; + continue; + } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + last = ts[i]; + } + lead = ts.back() - arrival; + /* Twenty packets is a generous allowance: the cap bleeds half a packet + * interval per packet, so the 0.6 s step is gone in twelve. */ + if (p >= 320) { worstAfter = std::max(worstAfter, std::fabs(lead)); } + } + + EXPECT_LT(worstAfter, 0.1) + << "still " << worstAfter << " s from the wall clock long after the step"; +} + +// The two branches must place a burst the same way round or two accumulated +// scalars in one scope, one with a declared rate and one without, sit a whole +// burst apart on the shared X axis. The declared branch anchors the LAST element +// on arrival, which is right: the samples were acquired before the packet +// carrying them landed. The hrt branch used to latch its offset against raw +// arrival, putting the FIRST element there instead. +TEST(FrameDecoder, UndeclaredAccumulatedScalarEndsItsBurstOnArrival) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + /* 10 ms per packet of 10 samples, so the derived period is 1 ms — equal to + * the fallback the very first hrt-branch packet has to use, which is what + * ClockOffset latches against. Any other period would bake that one packet's + * fallback into the offset and blur the convention this test is pinning. */ + double lastArrival = 0.0; + std::vector last; + for (int p = 0; p < 60; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const uint64_t hrt = bootHrt + static_cast(p * 0.010 * ticks); + const double arrival = 700.0 + p * 0.010; + const FrameView& f = + fb.build(hrt, arrival, 10, static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { last = ts; lastArrival = arrival; } + } + + ASSERT_EQ(last.size(), 10u); + EXPECT_NEAR(last[9], lastArrival, 1e-9) << "burst must END on arrival"; + EXPECT_NEAR(last[0], lastArrival - 0.009, 1e-9); +} + +// The same double delivery that the declared branch guards against — a host +// joined on two interfaces receives every unfragmented update twice — reaches an +// undeclared-rate signal identically. The guard can only fire if this branch +// leaves a counter behind for it to compare against. +TEST(FrameDecoder, UndeclaredAccumulatedScalarDropsADuplicatedDatagram) { + FrameDecoder dec; + dec.setSignals({undeclaredAcc()}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); + std::vector ts; + for (int p = 0; p < 50; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = + fb.build(bootHrt + static_cast(p * 0.010 * ticks), + 700.0 + p * 0.010, 10, static_cast(p + 1)); + dec.beginFrame(f); + const bool ok = dec.timestamps(f, 0, ts); + ASSERT_EQ(ok, p != 0) << "at packet " << p; + } + const double endBefore = ts[9]; + + /* Counter 50 again, the same update off the second interface. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& dup = + fb.build(bootHrt + static_cast(49 * 0.010 * ticks), + 700.0 + 49 * 0.010 + 0.0001, 10, 50u); + dec.beginFrame(dup); + EXPECT_FALSE(dec.timestamps(dup, 0, ts)) << "duplicate was emitted twice"; + + /* And the drop left the chain alone: the genuine next update still lands one + * period after the last burst ended. */ + const FrameView& next = + fb.build(bootHrt + static_cast(50 * 0.010 * ticks), + 700.0 + 50 * 0.010, 10, 51u); + dec.beginFrame(next); + ASSERT_TRUE(dec.timestamps(next, 0, ts)); + EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-6); +} + // A PACKET burst has no per-element time at all. Elements span // (lastPacket, thisPacket] — backwards from arrival, because the samples were // acquired before the packet landed. Forward extrapolation would let a jittered @@ -2015,8 +2381,32 @@ Create `Client/udpscope/FrameDecoder.h`: * though each represents ~10 ms of signal, and arrival-time interpolation then * crams a packet's samples into that tiny gap — the trace renders as a sawtooth. * Source/Applications/StreamHub/UDPSourceSession.cpp documents this failure and - * solves it; these are the same rules, computed from udps_frame_t's own fields - * so the scope and StreamHub agree on the same stream. + * solves it; these are the same rules, computed from udps_frame_t's own fields. + * + * Two rules deliberately differ, both in the accumulated-scalar case (rule 3). + * + * First, the anchor. StreamHub anchors every accumulated-scalar burst on the + * packet's own hrt, converted with the LOCAL MARTe HighResolutionTimer + * frequency — correct only because StreamHub runs on the producer's host. A + * bench scope attaches over the network and has no access to that frequency; it + * can only regress hrt against arrival time, which is exactly what the bursty + * delivery above corrupts. So when a SamplingRate is declared this decoder + * chains bursts instead, using the packet counter to account for loss and + * arrival time only as a backstop. The consequence is that a declared rate + * measured against the producer's crystal rather than ours makes the + * reconstructed timeline drift, and drift that only arrival time can observe + * must be corrected against arrival time — see rule 3. + * + * Second, the entry condition. UDPSourceSession.cpp:554 routes any update + * carrying nElems <= 1 to plain arrival time. That is safe for a host-local + * consumer whose arrival time is the producer's own clock, but wrong here: + * Accumulate mode flushes on a TIMER, so a short RT cycle legitimately delivers + * a single sample between two full bursts. Dating that one sample from arrival + * while its neighbours are chained puts it off the chain, and — worse — leaves + * lastCounter behind, so the next full burst reads the skipped counter as a lost + * datagram and reinstates a hole that never existed. So a signal that has + * already burst keeps every later update on rule 3 regardless of its length; a + * signal that has never burst is a genuine scalar and is left to rule 5. */ #pragma once @@ -2055,24 +2445,34 @@ private: struct SigState { ClockOffset offset; - double lastPacketWall = 0.0; + double lastPacketWall = 0.0; bool lastPacketValid = false; - /* 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; - /* Producer seconds since this signal's first usable packet, built by - * SUMMING short tick deltas -- never recomputed from an absolute tick - * count. */ - double accProdSec = 0.0; - bool lastAccValid = false; - uint32_t prevAccCount = 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; + /** Producer seconds since this signal's first usable packet, built by + * SUMMING short tick deltas. Never recomputed from an absolute tick + * count; see the samplingRate == 0 branch. */ + double accProdSec = 0.0; + bool lastAccValid = false; + uint32_t prevAccCount = 0; + /** For accumulated scalars (rule 3, either branch): end timestamp of the + * most recently emitted burst, and the packet counter it came from. The + * next burst is chained onto that end, with the counter gap reinstating + * the exact duration of any lost datagrams. */ double lastEmittedEnd = 0.0; uint32_t lastCounter = 0u; + /** ARRIVAL time of the packet that produced lastEmittedEnd. Valid + * exactly when lastEmittedValid is, so it needs no flag of its own. + * Deliberately not lastPacketWall, which belongs to packetBurst() and + * is updated on frames rule 3 never emits. This is the only reference + * against which a leading timeline can be pulled back: the correction + * has to be expressed as a fraction of the wall time that has really + * elapsed since this signal's previous burst, because within a single + * timestamps() call the wall clock is frozen and every forward step, + * however small, increases the lead measured at that instant. */ + double lastEmittedWall = 0.0; bool lastEmittedValid = false; }; @@ -2091,6 +2491,8 @@ Create `Client/udpscope/FrameDecoder.cpp`: ```cpp #include "FrameDecoder.h" +#include + namespace udpscope { /** Fallback cycle period before the first inter-packet gap is known. */ @@ -2098,17 +2500,80 @@ 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. + * should be before the chain is abandoned and time is re-anchored on arrival. + * + * This is a backstop, not the primary mechanism: the packet counter normally + * accounts for lost datagrams exactly, so the prediction and arrival agree. + * It catches what the counter cannot describe — a producer restart (the + * counter returns to zero), a counter that never advances, and a declared + * sampling rate that does not match the producer's real one. A kernel draining + * a backlog of queued datagrams can legitimately put the prediction a couple of + * hundred milliseconds from arrival, so the threshold sits well clear of that. + * Same value and same reasoning as ClockOffset::kRecalibThresholdS. */ static constexpr double kBurstResyncThresholdS = 0.5; -/** Narrowest a burst may be drawn, as a fraction of nominal, while a leading - * timeline is pulled back. Only a floor: the squeeze is normally proportional - * to the excess and removes it in one burst. See the sole use site. */ +/** + * Narrowest a burst may be drawn, as a fraction of its nominal width, while a + * leading timeline is being pulled back. Only a floor: the squeeze is normally + * proportional to the excess and removes it in a single burst. See the sole use + * site. + */ static constexpr double kMinBleedFactor = 0.05; +/** + * Largest share of the wall time elapsed since a signal's previous burst that + * that signal's next burst may advance its own timeline by, while a lead is + * being pulled back. + * + * This, and not kMinBleedFactor, is what makes a lead converge. A fraction of + * the NOMINAL burst width cannot: the floor still advances the timeline by + * kMinBleedFactor * nominal per packet while the wall advances one packet + * interval, so it diverges outright whenever nominal exceeds + * (1 / kMinBleedFactor) packet intervals — a declared SamplingRate of 30 against + * a producer really flushing 10 samples at 1 kHz put the trace 667 s ahead after + * 1000 s of stream. Measuring the allowance against elapsed WALL time instead + * bounds the advance below the wall's own advance for any declared rate, so the + * lead strictly falls whatever the config says. Any fraction under 1 converges; + * a half both converges quickly and leaves the burst visibly compressed rather + * than frozen. + */ +static constexpr double kWallBleedFraction = 0.5; + +/** + * A backward jump in producer hrt larger than this is a producer RESTART; a + * smaller one is a reordered datagram. + * + * The two need opposite handling — a reorder must leave the hrt reference + * untouched (its interval was already counted by the packet that overtook it), + * a restart must rebase onto the new epoch or the signal never advances again — + * and nothing but the size of the jump distinguishes them. A second of producer + * time is orders of magnitude more than any reordering window a UDP path can + * produce (a few packet intervals) and orders of magnitude less than a restart, + * which drops hrt from the producer's whole uptime back to near zero. + * + * Deliberately NOT kBurstResyncThresholdS: that one asks how far a WALL-clock + * prediction may sit from arrival, a different quantity in a different clock + * that happens to be tuned for delivery jitter. Sharing the number would couple + * two unrelated tunings. + */ +static constexpr double kProducerRestartS = 1.0; + +/** + * The declared sampling rate, or 0 when there is none to trust. + * + * samplingRate arrives unvalidated from a signal descriptor on the wire. A + * malformed +inf reaches the reciprocal as dt == 0, which makes a burst's + * nominal width zero and the proportional squeeze compute 0.0/0.0 — and a NaN + * factor is not caught by the floor, since every comparison against NaN is + * false, so the whole burst is emitted as NaN. Rejecting it at the boundary + * costs one test and removes the entire class. + */ +static double DeclaredRate(double samplingRate) { + return (std::isfinite(samplingRate) && samplingRate > 0.0) ? samplingRate + : 0.0; +} + void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; state_.assign(signals_.size(), SigState{}); @@ -2162,7 +2627,7 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t 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 de-duplicates FRAGMENTS only, so an unfragmented update + * 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. */ @@ -2170,8 +2635,8 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, 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 + /* 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) && @@ -2199,7 +2664,8 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, if ((d.timeMode == kTimeFirstSample || d.timeMode == kTimeLastSample) && hasTimeSig && f.counts[tIdx] >= 1u && f.values[tIdx] != nullptr) { const double anchor = st.offset.map(f.values[tIdx][0] * tScale, wallNow); - const double dt = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; + const double rate = DeclaredRate(d.samplingRate); + const double dt = (rate > 0.0) ? (1.0 / rate) : 0.0; tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { tsOut[e] = (d.timeMode == kTimeFirstSample) @@ -2209,93 +2675,160 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, return true; } - /* Rule 3: accumulated scalar. + /* Rule 3: accumulated scalar, based on declared sampling rate or hrt. * - * 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. + * When samplingRate is declared the inter-element step is exact and we + * anchor from the end of the previous burst rather than from arrival time + * or hrt. This makes the output immune to arrival jitter: even when the + * kernel delivers two packets microseconds apart each burst starts exactly + * one sample period after the previous burst ended. * - * With a declared rate none of that is needed: the intra-packet step is - * exact and bursts are contiguous, so the next burst chains onto the end of - * the previous one. The one thing a bare chain gets wrong is LOSS — it - * closes the hole a dropped datagram left, dating every later sample early - * for the rest of the run — and the wire already says exactly how much is - * missing: FrameView::counter increments once per update, so a gap of g - * means g-1 lost packets. Reinstating that duration needs no estimate and - * no threshold. The arrival-anchor comparison is only a BACKSTOP for what - * the counter cannot express (producer restart, counter stuck at zero, a - * declared rate that is simply wrong), and it must never move time - * backwards. The hrt path below remains for samplingRate == 0. */ - /* A signal that has already burst stays on this rule even when a later - * packet carries ONE sample: Accumulate mode flushes on a timer, so a short - * cycle legitimately yields one. Letting it fall to rule 5 would date it - * from arrival while its neighbours are chained, and would leave lastCounter - * behind so the next real burst read the skip as a lost datagram and - * reinstated a hole that never existed. A signal that has never burst is a - * genuine scalar and is left to rule 5. */ + * When samplingRate is absent we must derive dt from the hrt gap, which + * requires the HrtRateFit to be ready. Until then we fall back to + * packetBurst (arrival-time spanning), which is accurate during the normal + * pre-burst delivery phase that precedes the fit becoming ready. + * + * A signal that has already produced a burst stays on this rule even when a + * later packet carries a single sample — Accumulate mode flushes on a timer, + * so a short cycle legitimately yields one. Dropping such a packet to rule 5 + * would date it from arrival while its neighbours are chained, and would + * leave lastCounter behind so the next real burst read the skip as a lost + * datagram and reinstated a hole that never existed. A signal that has never + * burst is a genuine scalar and is left to rule 5. */ if (d.numElements() == 1u && (nElems > 1u || st.lastEmittedValid)) { - const double dtDeclared = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; - if (d.samplingRate > 0.0) { + /* A rate that is not a finite positive number is no rate at all; see + * DeclaredRate(). Such a signal takes the hrt branch below. */ + const double declared = DeclaredRate(d.samplingRate); + const double dt = (declared > 0.0) ? (1.0 / declared) : 0.0; + + if (declared > 0.0) { + /* 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) * dtDeclared; + wallNow - static_cast(nElems - 1u) * dt; + + /* Chaining onto the end of the previous burst is immune to arrival + * jitter — a kernel draining several queued datagrams microseconds + * apart still yields contiguous timestamps. What a bare chain gets + * wrong is loss: it closes the hole a dropped datagram left, and + * every later sample is then dated early for the rest of the run. + * + * The wire says exactly how much is missing. counter increments + * once per update, so a gap of g means g-1 lost packets, each + * carrying (as far as we can tell) as many samples as the last one + * we saw. Reinstating that duration keeps the chain honest without + * consulting arrival time at all. */ double base = arrivalAnchor; - double step = dtDeclared; + double step = dt; if (st.lastEmittedValid) { - /* Unsigned subtraction wraps, so this is right across the - * counter's own 2^32 rollover. A producer restart or reordered - * datagram makes the wrapped gap enormous, and that is NOT - * special-cased: an absurd gap yields an absurd prediction, - * which the arrival backstop rejects on its own. Clamping the - * gap first would decide the same question earlier, by a second - * rule no stream can distinguish from this one. */ + /* Unsigned subtraction wraps, so this stays right across the + * counter's own 2^32 rollover. + * + * A producer restart or a reordered datagram makes the wrapped + * gap enormous, and this deliberately does NOT special-case + * that: an absurd gap yields an absurd prediction, which the + * arrival backstop below then rejects on its own. + * + * Clamping the gap first is not a harmless earlier version of + * the same decision — it reaches the OPPOSITE answer. A clamp + * that treats gap > kMaxCounterGap as unknowable has to fall + * back to lost == 0, so the prediction becomes + * lastEmittedEnd + dt: one sample period after the last burst, + * which is exactly the shape of a healthy chain and therefore + * lands INSIDE the arrival backstop, is accepted, and silently + * closes an outage of arbitrary length. Letting the absurd gap + * through produces an absurd prediction that the backstop + * catches, and the burst re-anchors on arrival — which is the + * right answer, and what + * AccumulatedScalarSurvivesAProducerRestart pins down. The + * arithmetic cannot overflow: gap and prevAccCount are both + * bounded by 2^32-1, so lost is at most ~1.8e19, finite, and + * always rejected. */ const uint32_t gap = f.counter - st.lastCounter; const double lost = (gap > 1u) ? static_cast(gap - 1u) * static_cast(st.prevAccCount) : 0.0; - const double predicted = st.lastEmittedEnd + dtDeclared * (1.0 + lost); + const double predicted = st.lastEmittedEnd + dt * (1.0 + lost); + + /* Backstop for what the counter cannot express: a producer + * restart, a counter stuck at zero, or a declared rate that is + * simply wrong. Beyond this the chain is not recoverable and + * arrival time is the better of two bad answers. */ if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) { base = predicted; } + if (base <= st.lastEmittedEnd) { - /* 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. */ + /* 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) { - /* 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 no burst can end on - * arrival without starting before it. Squeeze this one - * by exactly the excess instead: its end lands one - * nominal width ahead of arrival -- the closest a - * forward-only timeline can legally get -- and the - * excess settles at (nominal width - true period), - * microseconds for a ppm-scale crystal mismatch. + /* The timeline has run PAST arrival: our last burst is + * dated later than the moment this packet landed, so + * there is no room to spread into and no burst can end + * on arrival without starting before it. Squeeze this + * one instead, by the excess and by the wall time that + * has really elapsed since this signal's last burst. * - * The floor keeps the step positive when the excess - * exceeds a whole burst (a declared rate wrong by a - * factor, not by ppm). It only slows recovery: each - * burst then advances by almost nothing while arrival - * keeps advancing, so the excess still reaches zero. */ - const double nominal = static_cast(nElems) * dtDeclared; + * Both terms are needed, and only the second one + * converges. Within this call the wall clock is frozen + * at wallNow, so ANY positive step increases the lead + * measured at this instant; the lead falls only because + * the wall advances BETWEEN packets. A step expressed + * purely as a fraction of the nominal burst width + * therefore diverges as soon as the nominal width + * outruns the packet interval — with the floor alone, a + * declared 30 Hz against a producer really flushing 10 + * samples at 1 kHz gained ~0.67 s of lead per second of + * stream, without bound. Capping the burst's total + * advance at kWallBleedFraction of the elapsed wall time + * makes it advance strictly slower than the wall for any + * declared rate, so the lead strictly falls. + * + * The proportional term still does the fine work: when + * the excess is smaller than a burst it removes it in + * one packet. kMinBleedFactor only keeps that term + * positive when the excess exceeds a whole burst. + * + * Steady state is a sawtooth, not a fixed offset: the + * squeeze pulls the lead down, ordinary chaining resumes + * on the very next packet and pushes it back up until + * the prediction misses arrival by more than + * kBurstResyncThresholdS. So the lead cycles between a + * fraction of a millisecond and roughly that threshold — + * bounded, which is what matters, but not zero. */ + const double nominal = static_cast(nElems) * dt; const double excess = st.lastEmittedEnd - wallNow; double factor = 1.0 - excess / nominal; if (factor < kMinBleedFactor) { factor = kMinBleedFactor; } - step = dtDeclared * factor; + double advance = nominal * factor; + + const double wallElapsed = wallNow - st.lastEmittedWall; + if (wallElapsed > 0.0) { + const double cap = kWallBleedFraction * wallElapsed; + if (cap < advance) { advance = cap; } + } + step = advance / static_cast(nElems); + /* Unreachable with a finite positive dt — kept because + * downstream monotonicity must not depend on that + * argument holding for every value off the wire. */ + if (!(step > 0.0)) { step = dt * kMinBleedFactor; } base = st.lastEmittedEnd + step; } } @@ -2305,11 +2838,13 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, tsOut[e] = base + static_cast(e) * step; } st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedWall = wallNow; st.lastCounter = f.counter; st.prevAccCount = nElems; st.lastEmittedValid = true; return true; } + /* No declared rate: need hrt-derived dt. */ if (!hrtFit_.ready() || f.hrt == 0u) { return packetBurst(idx, nElems, wallNow, tsOut); @@ -2333,11 +2868,47 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * arbitrary epoch that leaves behind, exactly as it would have latched * the producer's boot epoch. */ double elapsed = 0.0; - if (st.lastAccValid && f.hrt > st.lastAccHrt) { - elapsed = static_cast(f.hrt - st.lastAccHrt) / rate; + /* Whether lastAccHrt should take this packet's value. Only a packet + * that legitimately defines the new front of producer time may move it; + * see the backward case below. */ + bool takeHrt = true; + if (st.lastAccValid) { + if (f.hrt > st.lastAccHrt) { + elapsed = static_cast(f.hrt - st.lastAccHrt) / rate; + } else { + /* hrt went backwards. Two entirely different events look like + * this and only the SIZE of the jump separates them. + * + * A small one is a reordered datagram: the packet that overtook + * it already counted the interval it covers, so it must + * contribute nothing — and must also leave lastAccHrt alone. + * Letting it write lastAccHrt anyway (which is what this code + * used to do unconditionally) rolls the reference back one + * interval, so the NEXT packet's delta spans two and fabricates + * a whole extra packet of producer time. It never heals: + * ClockOffset would correct it, but the monotonic clamp below + * discards every backward correction. A hundred swaps on a + * 25 ms stream left the trace 3.5 s ahead, permanently. The C + * client does not reorder for us — udps_client.c only COUNTS + * counter gaps — so this is reachable on any real network. + * + * A large one is a producer restart: hrt drops from the + * producer's whole uptime to near zero. Here the unconditional + * write was the right behaviour and must be kept, because + * refusing to regress would leave every subsequent packet below + * lastAccHrt forever, elapsed permanently zero and the signal + * frozen. Rebase, and reset the offset so it re-latches against + * the new epoch instead of being dragged there by recalibration. */ + const double backward = + static_cast(st.lastAccHrt - f.hrt) / rate; + if (backward > kProducerRestartS) { + st.offset.reset(); + } else { + takeHrt = false; + } + } } st.accProdSec += elapsed; - double base = st.offset.map(st.accProdSec, wallNow); /* The flushes carry contiguous RT cycles, so the gap divided by the * previous packet's sample count is exactly one cycle period. */ @@ -2345,21 +2916,58 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, ? (elapsed / static_cast(st.prevAccCount)) : kDefaultDt; + /* Anchor the burst's LAST element on arrival, not its first. The + * packet's hrt is the tick count of sample 0 (UDPSourceSession.cpp:574), + * so stepping forward from it is right — but ClockOffset latches + * offset = wall - producerSec on its first call, and passing the raw + * arrival would put sample 0 at the instant the packet carrying the + * whole burst LANDED, dating every sample in it late by a burst. The + * declared-rate branch above already anchors on the burst end + * (arrivalAnchor), and two accumulated scalars in one scope, one with a + * declared rate and one without, would otherwise sit a burst apart on a + * shared X axis — 9 ms for 10 samples at 1 kHz, plain to see at a 200 ms + * window. Since map() latches once, this is a constant shift applied at + * latch and recalibration only; it changes no spacing. */ + double base = st.offset.map( + st.accProdSec, + wallNow - static_cast(nElems - 1u) * hrtDt); + double step = hrtDt; + /* ClockOffset recalibrates once true drift passes its threshold, and a * recalibration can land behind where this signal already is. - * Downstream requires increasing stamps, so step forward minimally. */ + * Downstream requires increasing stamps, so step forward minimally — + * but a bare forward step is one-directional, exactly the defect the + * declared branch's squeeze exists to avoid. A backward wall step (an + * NTP correction, a suspend/resume) would otherwise leave this signal + * permanently ahead of the wall clock, since the recalibrated base is + * behind lastEmittedEnd on every later packet too and the clamp keeps + * discarding it. So cap the burst's total advance against the wall time + * elapsed since this signal's previous burst, for the reason spelled out + * at kWallBleedFraction: only that makes the lead bleed off. */ if (st.lastEmittedValid && base <= st.lastEmittedEnd) { - base = st.lastEmittedEnd + hrtDt; + const double wallElapsed = wallNow - st.lastEmittedWall; + if (wallElapsed > 0.0) { + const double cap = kWallBleedFraction * wallElapsed / + static_cast(nElems); + if (cap < step) { step = cap; } + } + base = st.lastEmittedEnd + step; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { - tsOut[e] = base + static_cast(e) * hrtDt; + tsOut[e] = base + static_cast(e) * step; } - st.lastAccHrt = f.hrt; + if (takeHrt) { st.lastAccHrt = f.hrt; } st.lastAccValid = true; st.prevAccCount = nElems; st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedWall = wallNow; + /* Same duplicate-datagram exposure as the declared branch: a host joined + * on two interfaces receives every unfragmented update twice, and the + * guard at the top of timestamps() can only fire if this branch leaves a + * counter behind for it to compare against. */ + st.lastCounter = f.counter; st.lastEmittedValid = true; return true; } @@ -2394,7 +3002,7 @@ set(CORE_SOURCES cd Client/udpscope && cmake --build build -j && ./build/udpscope_tests --gtest_filter='FrameDecoder*' ``` -Expected: PASS, 15 tests. +Expected: PASS, 25 `FrameDecoder` tests — 54 across the whole `udpscope_tests` binary. 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. @@ -2406,6 +3014,16 @@ If `AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever` fails with a lead If `AccumulatedScalarStaysMonotonicOnALongUndeclaredRunAfterBoot` fails, the hrt path has been rewritten to position bursts from an ABSOLUTE tick conversion. Note that this test asserts spacing as well as order: the `base <= lastEmittedEnd` guard alone restores order while leaving positions wrong, so an order-only assertion would pass against a broken decoder. +If `AccumulatedScalarConvergesWhenTheDeclaredRateIsFarTooLow` fails with a lead in the tens of seconds, the `kWallBleedFraction` cap has been dropped or expressed against the wrong reference. `kMinBleedFactor` alone CANNOT converge, and the difference is not a matter of speed: inside one `timestamps()` call the wall clock is frozen, so every positive step increases the lead measured at that instant; the lead falls only because the wall advances between packets. A step that is a fraction of the NOMINAL burst therefore outruns the wall whenever the nominal burst is wider than `1 / kMinBleedFactor` packet intervals — declared 30 Hz against a producer really flushing 10 samples at 1 kHz gained 0.67 s per second of stream, unbounded, and declared 50 Hz was exactly marginal. The cap must be a fraction of `wallNow - st.lastEmittedWall`, i.e. of wall time really elapsed since THIS signal's previous burst. Do not reuse `lastPacketWall` for that reference: it belongs to `packetBurst()` and is updated on frames rule 3 never emits. + +If `UndeclaredAccumulatedScalarIgnoresReorderedDatagrams` fails, `st.lastAccHrt` is being written for a packet whose `hrt` is behind it. The elapsed-time guard alone is not enough — it correctly contributes zero for the late packet, but rolling the reference back makes the NEXT packet's delta span two intervals and fabricate a whole extra packet of producer time, permanently (`ClockOffset` would correct it; the monotonic clamp discards every backward correction). Note this test is also sensitive to the `kWallBleedFraction` cap on the hrt branch, which is what reabsorbs the extra clamped burst a reorder emits; `UndeclaredAccumulatedScalarSurvivesAProducerRestart` is the test that isolates the reference-update rule on its own. + +If `UndeclaredAccumulatedScalarSurvivesAProducerRestart` fails with the spacing stuck at `kDefaultDt`, the backward-jump handling has been collapsed into "never regress". A restart drops `hrt` from the producer's whole uptime to near zero, so every later packet is below the reference forever, `elapsed` is permanently zero and the signal freezes. Only the SIZE of the jump separates a restart from a reorder — hence `kProducerRestartS`. Do not reuse `kBurstResyncThresholdS` for it: that one answers how far a wall-clock prediction may sit from arrival, a different quantity in a different clock, tuned for delivery jitter. + +If `UndeclaredAccumulatedScalarRecoversFromABackwardWallStep` fails, the hrt branch's `base <= lastEmittedEnd` clamp has gone back to a bare `lastEmittedEnd + hrtDt`, which is one-directional and leaves an NTP correction or a suspend/resume as a permanent lead. Note the test's geometry is deliberate: the step is 0.6 s (it must exceed `ClockOffset::kRecalibThresholdS` or the recalibration that puts `base` behind `lastEmittedEnd` never happens at all), and it lands after the 256-sample rate-fit window is full. `HrtRateFit` regresses `hrt` against ARRIVAL, so it eventually absorbs the step too, at roughly `step / kWindow` per packet — a much slower second correction that would swamp the measurement if the step were placed early or the run continued for hundreds of packets past it. + +If `UndeclaredAccumulatedScalarEndsItsBurstOnArrival` fails by exactly `(nElems - 1) * hrtDt`, `ClockOffset::map()` is being latched against raw `wallNow` again, which puts the burst's FIRST element on arrival while the declared branch puts its LAST one there — two accumulated scalars in one scope, one with a declared rate and one without, then sit a whole burst apart on the shared X axis. The test's 10 ms packet of 10 samples is chosen so the derived period equals `kDefaultDt`: the very first hrt-branch packet has no measurable interval and latches the offset using that fallback, and any other period would bake the difference into the offset for the rest of the run. + - [ ] **Step 8: Commit** ```bash