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