diff --git a/Client/udpscope/FrameDecoder.cpp b/Client/udpscope/FrameDecoder.cpp index 3359bb2..5337ce7 100644 --- a/Client/udpscope/FrameDecoder.cpp +++ b/Client/udpscope/FrameDecoder.cpp @@ -23,21 +23,12 @@ static constexpr double kDefaultDt = 1.0e-3; static constexpr double kBurstResyncThresholdS = 0.5; /** - * Largest counter gap still read as a loss count. - * - * A producer restart returns the counter to zero and a reordered datagram makes - * the unsigned gap wrap to near 2^32; multiplying either by a sample count and - * calling it elapsed time would fabricate centuries. A million lost updates is - * already far beyond any outage worth reconstructing. + * 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 uint32_t kMaxCounterGap = 1000000u; - -/** - * Burst width, as a fraction of nominal, while a leading timeline is being - * pulled back. See the sole use site for why a leading chain cannot be - * corrected in one burst and must be bled off instead. - */ -static constexpr double kLeadBleedFactor = 0.9; +static constexpr double kMinBleedFactor = 0.05; void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; @@ -150,8 +141,16 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * 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. */ - if (d.numElements() == 1u && nElems > 1u) { + * 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 dt = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; @@ -177,11 +176,16 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, double step = dt; if (st.lastEmittedValid) { /* Unsigned subtraction wraps, so this stays right across the - * counter's own 2^32 rollover. A gap far larger than any real - * outage is a restart or a reordered datagram rather than a - * loss count; claim nothing and let the backstop below decide. */ + * 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 would only decide the same question earlier, by a + * second rule that no stream can distinguish from this one. */ const uint32_t gap = f.counter - st.lastCounter; - const double lost = (gap > 1u && gap <= kMaxCounterGap) + const double lost = (gap > 1u) ? static_cast(gap - 1u) * static_cast(st.prevAccCount) : 0.0; @@ -216,17 +220,27 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, base = st.lastEmittedEnd + step; } else { /* The timeline has run PAST arrival: our last burst is - * dated later than the moment this packet landed. There - * is no room to spread into, and no single burst can - * remove the excess without stepping back. So bleed it - * off — draw each burst a fixed fraction narrower than - * nominal until the timeline is back inside the - * threshold, then normal chaining resumes. The factor - * only has to shrink a burst faster than the clock - * mismatch grows it, and a 10 % squeeze outruns the - * tens-of-ppm crystal error that causes this by orders - * of magnitude. */ - step = dt * kLeadBleedFactor; + * 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. + * + * 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. */ + 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; base = st.lastEmittedEnd + step; } } @@ -248,43 +262,51 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, } const double rate = hrtFit_.ticksPerSecond(); - /* Difference raw TICKS, never two toSeconds() results. + /* Integrate short tick DELTAS. Never convert an absolute tick count, and + * never subtract two such conversions. * * hrt counts from the producer's boot, so it is already ~1e11 ticks when * the scope attaches, while the fit is re-estimated on every packet and - * wobbles by a few parts in 1e4. toSeconds() multiplies that relative - * wobble by the whole elapsed epoch: tens of milliseconds of jitter on a - * value whose consecutive difference is a few milliseconds. Subtracting - * two such results measures the wobble, not the interval. + * wobbles by a few parts in 1e4. Any absolute hrt/rate therefore carries + * that relative wobble multiplied by the whole elapsed epoch — tens of + * milliseconds, moving in either direction from one packet to the next. + * As a burst's position that is not merely imprecise, it is + * NON-MONOTONIC: on a 2 h stream with ordinary scheduling jitter a few + * percent of samples land before their own predecessor. * - * Anchoring on the first usable packet keeps the wobble on the elapsed - * interval since attach, which is short, and ClockOffset absorbs the - * arbitrary epoch that anchoring leaves behind exactly as it would - * absorb the producer's boot epoch. */ - if (!st.hrtRefValid) { - st.hrtRef = f.hrt; - st.hrtRefValid = true; + * A delta spans one packet, so its share of the wobble is microseconds, + * and summing deltas keeps it there. ClockOffset then latches the + * 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; } - const double sinceRef = (f.hrt >= st.hrtRef) - ? static_cast(f.hrt - st.hrtRef) / rate - : -static_cast(st.hrtRef - f.hrt) / rate; - const double base = st.offset.map(sinceRef, wallNow); + st.accProdSec += elapsed; + double base = st.offset.map(st.accProdSec, wallNow); - double hrtDt = kDefaultDt; - if (st.lastAccValid && st.prevAccCount > 0u && f.hrt > st.lastAccHrt) { - /* The flushes carry contiguous RT cycles, so the gap divided by the - * previous packet's sample count is exactly one cycle period. */ - hrtDt = (static_cast(f.hrt - st.lastAccHrt) / rate) / - static_cast(st.prevAccCount); + /* The flushes carry contiguous RT cycles, so the gap divided by the + * previous packet's sample count is exactly one cycle period. */ + const double hrtDt = (elapsed > 0.0 && st.prevAccCount > 0u) + ? (elapsed / static_cast(st.prevAccCount)) + : kDefaultDt; + + /* 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. */ + if (st.lastEmittedValid && base <= st.lastEmittedEnd) { + base = st.lastEmittedEnd + hrtDt; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { tsOut[e] = base + static_cast(e) * hrtDt; } - st.lastAccHrt = f.hrt; - st.lastAccValid = true; - st.prevAccCount = nElems; + st.lastAccHrt = f.hrt; + st.lastAccValid = true; + st.prevAccCount = nElems; + st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedValid = true; return true; } diff --git a/Client/udpscope/FrameDecoder.h b/Client/udpscope/FrameDecoder.h index 5f7e382..f83f530 100644 --- a/Client/udpscope/FrameDecoder.h +++ b/Client/udpscope/FrameDecoder.h @@ -8,8 +8,19 @@ * 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. + * + * 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. */ #pragma once @@ -54,8 +65,10 @@ private: * branch for why a tick difference is the only safe way to measure a * producer-side interval while the rate is still being re-estimated. */ uint64_t lastAccHrt = 0u; - uint64_t hrtRef = 0u; - bool hrtRefValid = false; + /** 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 with a declared sampling rate: end timestamp diff --git a/Client/udpscope/tests/FrameDecoderTest.cpp b/Client/udpscope/tests/FrameDecoderTest.cpp index e1cde98..768da6f 100644 --- a/Client/udpscope/tests/FrameDecoderTest.cpp +++ b/Client/udpscope/tests/FrameDecoderTest.cpp @@ -270,9 +270,10 @@ TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { ASSERT_TRUE(dec.timestamps(f, 0, ts)); /* Arrival (500.000) is behind our timeline, so there is nothing to spread - * into; the burst is drawn narrower instead, which starts to bleed the lead - * off while still moving strictly forwards. */ - EXPECT_NEAR(ts[0], 500.0909, 1e-9); + * into; the burst is squeezed instead, which bleeds the lead off while still + * moving strictly forwards. The excess (90 ms) is nine nominal burst widths, + * so the squeeze hits its floor of 0.05 and the step is 50 us. */ + EXPECT_NEAR(ts[0], 500.09005, 1e-9); EXPECT_GT(ts[0], prevEnd) << "resync stepped backwards over the previous burst"; for (size_t i = 1; i < ts.size(); i++) { EXPECT_GT(ts[i], ts[i - 1]); @@ -290,8 +291,13 @@ TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); - /* A counter gap far beyond any real outage: the prediction is unusable, and - * the arrival anchor (500.086) sits behind the previous burst end. */ + /* The compress branch needs the prediction to be rejected while arrival + * still sits between the previous burst's end and one burst beyond it — + * which a plain rate mismatch cannot produce, since the prediction is then + * only a burst away from arrival. It takes a fabricated loss: this gap + * claims 900000 lost packets, putting the prediction 2.5 hours out, while + * the packet itself lands 5 ms after the last burst ended so its arrival + * anchor (500.086) falls just behind that end. */ FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 500.095, 10, 900011u); @@ -364,6 +370,63 @@ TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-9); } +// A producer restart returns the counter to zero mid-stream. The unsigned gap +// then wraps to near 2^32; the loss it implies puts the chained prediction +// centuries out, the arrival backstop rejects it, and arrival becomes the only +// usable reference. +TEST(FrameDecoder, AccumulatedScalarSurvivesAProducerRestart) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + const double prevEnd = ts[9]; + + /* Restarted producer: counter 1 again, and the outage lasted 3 s. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 503.100, 10, 1u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + /* Reading the wrapped gap as a loss count would claim ~4.3e9 lost packets, + * some 5e8 seconds of fabricated signal. */ + EXPECT_NEAR(ts[9], 503.100, 1e-9) << "restart must re-anchor on arrival"; + EXPECT_GT(ts[0], prevEnd); +} + +// Accumulate mode flushes on a timer, so a short cycle legitimately delivers a +// single sample between two full bursts. That packet must stay on the chain: if +// it fell through to the plain-scalar rule it would be dated from arrival while +// its neighbours are chained, and would leave lastCounter behind so the next +// real burst read the skip as a lost datagram. +TEST(FrameDecoder, AccumulatedScalarKeepsShortFlushesOnTheChain) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + + double last = ts[9]; + uint32_t counter = 10u; + double arrival = 500.090; + for (int p = 0; p < 500; p++) { + /* Alternating 10-sample and 1-sample flushes, 10 ms and 1 ms of signal. */ + const uint32_t n = (p % 2 == 0) ? 1u : 10u; + arrival += 0.001 * static_cast(n); + FrameBuilder fb; + fb.addSignal(std::vector(n, 1.0)); + const FrameView& f = fb.build(0, arrival, n, ++counter); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)) << "short flush dropped at " << p; + ASSERT_EQ(ts.size(), n); + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + /* Contiguous: no phantom loss was ever reinstated. */ + ASSERT_NEAR(ts[i] - last, 0.001, 1e-6) << "gap opened at packet " << p; + last = ts[i]; + } + } +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; @@ -401,6 +464,56 @@ TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5); } +// The trap the hrt path fell into once: positioning each burst at +// hrt / ticksPerSecond(). hrt counts from the PRODUCER'S BOOT, so it is already +// ~1e11 ticks for a machine that has been up a day, while the rate is refitted +// on every packet and wobbles by parts in 1e4 as arrival jitter enters and +// leaves the window. The wobble arrives multiplied by that whole epoch — tens of +// milliseconds, in both directions — so bursts land out of order. The producer +// clock here is EXACT; every timestamp inversion this test can see comes from +// the client's own arithmetic. +TEST(FrameDecoder, AccumulatedScalarStaysMonotonicOnALongUndeclaredRunAfterBoot) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.samplingRate = 0.0; /* undeclared: the hrt path */ + dec.setSignals({m}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); /* up 1 day */ + double last = 0.0; + uint32_t seed = 12345u; + for (int p = 0; p < 20000; p++) { /* 500 s of stream */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + /* Exact producer clock: 25 ms per packet, 2.5 ms per sample. */ + const uint64_t hrt = bootHrt + static_cast(p * 0.025 * ticks); + /* Ordinary scheduling jitter, +/- 1 ms, zero mean. */ + seed = seed * 1103515245u + 12345u; + const double jitter = (static_cast((seed >> 16) & 0xFFFFu) / + 65535.0 - 0.5) * 0.002; + const FrameView& f = fb.build(hrt, 700.0 + p * 0.025 + jitter, 10, + static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { continue; } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + /* Ordering alone is too weak to pin this down: clamping a wrong + * absolute position to "just after the last one" restores the + * ordering while leaving the positions wrong, and every forward + * lurch is still accepted. The producer clock is exact, so the + * spacing must be exact too. */ + if (p > 100) { /* past the fit warm-up and its packetBurst fallback */ + ASSERT_NEAR(ts[i] - last, 0.0025, 1e-5) + << "sample spacing wrong at packet " << p; + } + last = ts[i]; + } + } +} + // 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 de8bafc..c37a41c 100644 --- a/docs/superpowers/plans/2026-08-27-udpscope.md +++ b/docs/superpowers/plans/2026-08-27-udpscope.md @@ -1676,9 +1676,10 @@ TEST(FrameDecoder, AccumulatedScalarNeverStepsBackwardsWhenResyncing) { ASSERT_TRUE(dec.timestamps(f, 0, ts)); /* Arrival (500.000) is behind our timeline, so there is nothing to spread - * into; the burst is drawn narrower instead, which starts to bleed the lead - * off while still moving strictly forwards. */ - EXPECT_NEAR(ts[0], 500.0909, 1e-9); + * into; the burst is squeezed instead, which bleeds the lead off while still + * moving strictly forwards. The excess (90 ms) is nine nominal burst widths, + * so the squeeze hits its floor of 0.05 and the step is 50 us. */ + EXPECT_NEAR(ts[0], 500.09005, 1e-9); EXPECT_GT(ts[0], prevEnd) << "resync stepped backwards over the previous burst"; for (size_t i = 1; i < ts.size(); i++) { EXPECT_GT(ts[i], ts[i - 1]); @@ -1695,8 +1696,13 @@ TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); - /* A counter gap far beyond any real outage: the prediction is unusable, and - * the arrival anchor (500.086) sits behind the previous burst end. */ + /* The compress branch needs the prediction to be rejected while arrival + * still sits between the previous burst's end and one burst beyond it — + * which a plain rate mismatch cannot produce, since the prediction is then + * only a burst away from arrival. It takes a fabricated loss: this gap + * claims 900000 lost packets, putting the prediction 2.5 hours out, while + * the packet itself lands 5 ms after the last burst ended so its arrival + * anchor (500.086) falls just behind that end. */ FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 500.095, 10, 900011u); @@ -1708,10 +1714,11 @@ TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { EXPECT_NEAR(ts[1] - ts[0], 0.0005, 1e-9) << "spread over the available room"; } -// The whole point of the bleed: a declared SamplingRate is a hand-written config -// value measured against the PRODUCER host's crystal, not ours. Tens of ppm of -// difference is certain over a long session, so the timeline WILL run away from -// the wall clock. It must be pulled back, and stay monotonic while that happens. +// The whole point of compressing: a declared SamplingRate is a hand-written +// config value, and even a correct one is measured against the producer host's +// crystal, not ours. Tens of ppm of difference is certain over a long session, +// so the reconstructed timeline WILL run away from the wall clock. It has to be +// pulled back, and it has to stay monotonic while that happens. TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { FrameDecoder dec; dec.setSignals({accSignal()}); /* declares 1 kHz */ @@ -1720,7 +1727,7 @@ TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { * so a chain stepping the declared 10 ms per packet gains 0.1 ms every * packet. This is the direction re-anchoring alone cannot fix: arrival is * always BEHIND the chain, so anchoring on it would step backwards and is - * refused. Only the bleed pulls the timeline back. */ + * refused. Only compression pulls the timeline back. */ double worstLead = 0.0; double lastEnd = 0.0; for (int p = 0; p < 20000; p++) { @@ -1768,6 +1775,63 @@ TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-9); } +// A producer restart returns the counter to zero mid-stream. The unsigned gap +// then wraps to near 2^32; the loss it implies puts the chained prediction +// centuries out, the arrival backstop rejects it, and arrival becomes the only +// usable reference. +TEST(FrameDecoder, AccumulatedScalarSurvivesAProducerRestart) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + const double prevEnd = ts[9]; + + /* Restarted producer: counter 1 again, and the outage lasted 3 s. */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const FrameView& f = fb.build(0, 503.100, 10, 1u); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + + /* Reading the wrapped gap as a loss count would claim ~4.3e9 lost packets, + * some 5e8 seconds of fabricated signal. */ + EXPECT_NEAR(ts[9], 503.100, 1e-9) << "restart must re-anchor on arrival"; + EXPECT_GT(ts[0], prevEnd); +} + +// Accumulate mode flushes on a timer, so a short cycle legitimately delivers a +// single sample between two full bursts. That packet must stay on the chain: if +// it fell through to the plain-scalar rule it would be dated from arrival while +// its neighbours are chained, and would leave lastCounter behind so the next +// real burst read the skip as a lost datagram. +TEST(FrameDecoder, AccumulatedScalarKeepsShortFlushesOnTheChain) { + FrameDecoder dec; + dec.setSignals({accSignal()}); + std::vector ts; + primeTenBursts(dec, ts, /*withCounter=*/true); + + double last = ts[9]; + uint32_t counter = 10u; + double arrival = 500.090; + for (int p = 0; p < 500; p++) { + /* Alternating 10-sample and 1-sample flushes, 10 ms and 1 ms of signal. */ + const uint32_t n = (p % 2 == 0) ? 1u : 10u; + arrival += 0.001 * static_cast(n); + FrameBuilder fb; + fb.addSignal(std::vector(n, 1.0)); + const FrameView& f = fb.build(0, arrival, n, ++counter); + dec.beginFrame(f); + ASSERT_TRUE(dec.timestamps(f, 0, ts)) << "short flush dropped at " << p; + ASSERT_EQ(ts.size(), n); + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + /* Contiguous: no phantom loss was ever reinstated. */ + ASSERT_NEAR(ts[i] - last, 0.001, 1e-6) << "gap opened at packet " << p; + last = ts[i]; + } + } +} + TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { FrameDecoder dec; SignalMeta m; @@ -1805,6 +1869,56 @@ TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5); } +// The trap the hrt path fell into once: positioning each burst at +// hrt / ticksPerSecond(). hrt counts from the PRODUCER'S BOOT, so it is already +// ~1e11 ticks for a machine that has been up a day, while the rate is refitted +// on every packet and wobbles by parts in 1e4 as arrival jitter enters and +// leaves the window. The wobble arrives multiplied by that whole epoch — tens of +// milliseconds, in both directions — so bursts land out of order. The producer +// clock here is EXACT; every timestamp inversion this test can see comes from +// the client's own arithmetic. +TEST(FrameDecoder, AccumulatedScalarStaysMonotonicOnALongUndeclaredRunAfterBoot) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.samplingRate = 0.0; /* undeclared: the hrt path */ + dec.setSignals({m}); + + const double ticks = 1.0e9; + const uint64_t bootHrt = static_cast(86400.0 * ticks); /* up 1 day */ + double last = 0.0; + uint32_t seed = 12345u; + for (int p = 0; p < 20000; p++) { /* 500 s of stream */ + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + /* Exact producer clock: 25 ms per packet, 2.5 ms per sample. */ + const uint64_t hrt = bootHrt + static_cast(p * 0.025 * ticks); + /* Ordinary scheduling jitter, +/- 1 ms, zero mean. */ + seed = seed * 1103515245u + 12345u; + const double jitter = (static_cast((seed >> 16) & 0xFFFFu) / + 65535.0 - 0.5) * 0.002; + const FrameView& f = fb.build(hrt, 700.0 + p * 0.025 + jitter, 10, + static_cast(p + 1)); + dec.beginFrame(f); + std::vector ts; + if (!dec.timestamps(f, 0, ts)) { continue; } + for (size_t i = 0; i < ts.size(); i++) { + ASSERT_GT(ts[i], last) << "timeline went backwards at packet " << p; + /* Ordering alone is too weak to pin this down: clamping a wrong + * absolute position to "just after the last one" restores the + * ordering while leaving the positions wrong, and every forward + * lurch is still accepted. The producer clock is exact, so the + * spacing must be exact too. */ + if (p > 100) { /* past the fit warm-up and its packetBurst fallback */ + ASSERT_NEAR(ts[i] - last, 0.0025, 1e-5) + << "sample spacing wrong at packet " << p; + } + last = ts[i]; + } + } +} + // 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 @@ -1951,8 +2065,10 @@ private: * Differencing two toSeconds() results measures the wobble, not the * interval. Difference the ticks and divide once instead. */ uint64_t lastAccHrt = 0u; - uint64_t hrtRef = 0u; - bool hrtRefValid = false; + /* 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; double lastEmittedEnd = 0.0; @@ -1988,16 +2104,10 @@ static constexpr double kDefaultDt = 1.0e-3; */ static constexpr double kBurstResyncThresholdS = 0.5; -/** - * Largest counter gap still read as a loss count. A producer restart returns - * the counter to zero and a reordered datagram wraps the unsigned gap to near - * 2^32; multiplying either by a sample count would fabricate centuries. - */ -static constexpr uint32_t kMaxCounterGap = 1000000u; - -/** Burst width, as a fraction of nominal, while a leading timeline is pulled - * back. See the sole use site. */ -static constexpr double kLeadBleedFactor = 0.9; +/** 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. */ +static constexpr double kMinBleedFactor = 0.05; void FrameDecoder::setSignals(const std::vector& signals) { signals_ = signals; @@ -2121,7 +2231,14 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, * 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. */ - if (d.numElements() == 1u && nElems > 1u) { + /* 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. */ + 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) { const double arrivalAnchor = @@ -2130,11 +2247,14 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, double step = dtDeclared; if (st.lastEmittedValid) { /* Unsigned subtraction wraps, so this is right across the - * counter's own 2^32 rollover. A gap far larger than any real - * outage is a producer restart or a reordered datagram, not a - * loss count; claim nothing and let the backstop decide. */ + * 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. */ const uint32_t gap = f.counter - st.lastCounter; - const double lost = (gap > 1u && gap <= kMaxCounterGap) + const double lost = (gap > 1u) ? static_cast(gap - 1u) * static_cast(st.prevAccCount) : 0.0; @@ -2158,13 +2278,24 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, static_cast(nElems); base = st.lastEmittedEnd + step; } else { - /* We have run PAST arrival, so there is no room to - * spread into and no single burst can remove the excess - * without going backwards. Bleed it off: draw every - * burst a fixed fraction narrower until the timeline is - * back inside the threshold. A 10 % squeeze outruns a - * tens-of-ppm crystal error by orders of magnitude. */ - step = dtDeclared * kLeadBleedFactor; + /* 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 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; + const double excess = st.lastEmittedEnd - wallNow; + double factor = 1.0 - excess / nominal; + if (factor < kMinBleedFactor) { factor = kMinBleedFactor; } + step = dtDeclared * factor; base = st.lastEmittedEnd + step; } } @@ -2179,40 +2310,57 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, st.lastEmittedValid = true; return true; } + /* No declared rate: need hrt-derived dt. */ if (!hrtFit_.ready() || f.hrt == 0u) { return packetBurst(idx, nElems, wallNow, tsOut); } const double rate = hrtFit_.ticksPerSecond(); - /* Difference raw TICKS, never two toSeconds() results — see SigState. - * Anchoring on the first usable packet keeps the fit's wobble on the - * (short) interval since attach instead of on the producer's whole - * uptime. ClockOffset absorbs the arbitrary epoch that leaves behind - * exactly as it would absorb the producer's boot epoch. */ - if (!st.hrtRefValid) { - st.hrtRef = f.hrt; - st.hrtRefValid = true; + /* Integrate short tick DELTAS. Never convert an absolute tick count, and + * never subtract two such conversions. + * + * hrt counts from the producer's boot, so it is already ~1e11 ticks when + * the scope attaches, while the fit is re-estimated on every packet and + * wobbles by a few parts in 1e4. Any absolute hrt/rate therefore carries + * that relative wobble multiplied by the whole elapsed epoch — tens of + * milliseconds, moving in either direction from one packet to the next. + * As a burst's position that is not merely imprecise, it is + * NON-MONOTONIC: on a 2 h stream with ordinary scheduling jitter a few + * percent of samples land before their own predecessor. + * + * A delta spans one packet, so its share of the wobble is microseconds, + * and summing deltas keeps it there. ClockOffset then latches the + * 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; } - const double sinceRef = (f.hrt >= st.hrtRef) - ? static_cast(f.hrt - st.hrtRef) / rate - : -static_cast(st.hrtRef - f.hrt) / rate; - const double base = st.offset.map(sinceRef, wallNow); + st.accProdSec += elapsed; + double base = st.offset.map(st.accProdSec, wallNow); - double dt = kDefaultDt; - if (st.lastAccValid && st.prevAccCount > 0u && f.hrt > st.lastAccHrt) { - /* The flushes carry contiguous RT cycles, so the gap divided by the - * previous packet's sample count is exactly one cycle period. */ - dt = (static_cast(f.hrt - st.lastAccHrt) / rate) / - static_cast(st.prevAccCount); + /* The flushes carry contiguous RT cycles, so the gap divided by the + * previous packet's sample count is exactly one cycle period. */ + const double hrtDt = (elapsed > 0.0 && st.prevAccCount > 0u) + ? (elapsed / static_cast(st.prevAccCount)) + : kDefaultDt; + + /* 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. */ + if (st.lastEmittedValid && base <= st.lastEmittedEnd) { + base = st.lastEmittedEnd + hrtDt; } tsOut.resize(nElems); for (uint32_t e = 0; e < nElems; e++) { - tsOut[e] = base + static_cast(e) * dt; + tsOut[e] = base + static_cast(e) * hrtDt; } - st.lastAccHrt = f.hrt; - st.lastAccValid = true; - st.prevAccCount = nElems; + st.lastAccHrt = f.hrt; + st.lastAccValid = true; + st.prevAccCount = nElems; + st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedValid = true; return true; } @@ -2246,7 +2394,7 @@ set(CORE_SOURCES cd Client/udpscope && cmake --build build -j && ./build/udpscope_tests --gtest_filter='FrameDecoder*' ``` -Expected: PASS, 9 tests. +Expected: PASS, 15 tests. If `AccumulatedScalarSurvivesBurstyDelivery` fails, do NOT reach for the hrt fit: with a declared `samplingRate` rule 3 never consults it, precisely because the fit is not ready for the first 32 packets and — since `HrtRateFit` regresses `hrt` against ARRIVAL time — is itself corrupted by the very bursts it would be asked to survive. Check instead that `lastEmittedEnd`, `lastCounter`, `prevAccCount` and `lastEmittedValid` are updated on every emitted burst. @@ -2254,7 +2402,9 @@ Note for `AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared`: its arri If that test returns exactly `kDefaultDt`, or a value that wanders between runs of different length, the cause is almost certainly a reintroduced `hrtFit_.toSeconds(a) - hrtFit_.toSeconds(b)`. `toSeconds()` divides an ABSOLUTE tick count by a rate refitted on every packet; a producer that has been up for a day is at ~1e11 ticks, so the fit's few-parts-in-1e4 wobble becomes tens of milliseconds of jitter on the result — larger than the interval being measured. Difference the raw ticks and divide once by `ticksPerSecond()`. -If `AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever` fails at ~2 s, the lead bleed is not firing. Note that the "spread out to arrival" compression is unreachable in this case by construction: a LEADING timeline has `lastEmittedEnd > wallNow`, so there is no room to spread into. That branch handles only a bad prediction while arrival is still ahead; the leading case needs the `kLeadBleedFactor` path below it. +If `AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever` fails with a lead that grows without bound, the proportional squeeze is not firing. Note that the "spread out to arrival" compression is unreachable in this case by construction: a LEADING timeline has `lastEmittedEnd > wallNow`, so there is no room to spread into. That branch handles only a bad prediction while arrival is still ahead; the leading case needs the `kMinBleedFactor` branch below it. Get the SIGN of the test's arrival spacing right — the producer must be FAST (arrivals closer together than the declared period, e.g. 0.0099 s for a 10 ms nominal burst). A slow producer makes the chain LAG, which the one-directional backstop already handles, so the test would pass with the squeeze deleted. + +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. - [ ] **Step 8: Commit**