#include "FrameDecoder.h" #include #include #include #include using namespace udpscope; namespace { /** Builds a FrameView over vectors the test owns. */ struct FrameBuilder { std::vector> storage; std::vector ptrs; std::vector counts; FrameView view; void addSignal(std::vector vals) { storage.push_back(std::move(vals)); } /* 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; view.numSignals = static_cast(storage.size()); view.values = ptrs.data(); view.counts = counts.data(); return view; } }; SignalMeta burst(const char* name, uint8_t timeMode, double rate, uint32_t elems, uint32_t timeIdx) { SignalMeta m; m.name = name; m.typeCode = 8; /* float32 */ m.numRows = elems; m.numCols = 1; m.timeMode = timeMode; m.samplingRate = rate; m.timeSignalIdx = timeIdx; return m; } SignalMeta timeSignal(const char* name, uint32_t elems) { SignalMeta m; m.name = name; m.typeCode = 6; /* uint64 -> nanoseconds */ m.numRows = elems; m.numCols = 1; return m; } } /* namespace */ TEST(FrameDecoder, FullArrayTakesOneStampPerElementFromTheTimeSignal) { FrameDecoder dec; dec.setSignals({burst("Sine", kTimeFullArray, 1000.0, 4, 1), timeSignal("Time", 4)}); FrameBuilder fb; fb.addSignal({1.0, 2.0, 3.0, 4.0}); /* Nanoseconds: 5.000, 5.001, 5.002, 5.003 s of producer time. */ fb.addSignal({5.0e9, 5.001e9, 5.002e9, 5.003e9}); const FrameView& f = fb.build(0, 1000.0); dec.beginFrame(f); std::vector ts; ASSERT_TRUE(dec.timestamps(f, 0, ts)); ASSERT_EQ(ts.size(), 4u); /* Element 0 lands on the arrival time; the rest keep the producer spacing. */ EXPECT_NEAR(ts[0], 1000.000, 1e-9); EXPECT_NEAR(ts[1], 1000.001, 1e-9); EXPECT_NEAR(ts[2], 1000.002, 1e-9); EXPECT_NEAR(ts[3], 1000.003, 1e-9); } TEST(FrameDecoder, FirstSampleAnchorsElementZeroAndCountsForward) { FrameDecoder dec; dec.setSignals({burst("Sine", kTimeFirstSample, 1000.0, 4, 1), timeSignal("Time", 1)}); FrameBuilder fb; fb.addSignal({1.0, 2.0, 3.0, 4.0}); fb.addSignal({7.0e9}); const FrameView& f = fb.build(0, 2000.0); dec.beginFrame(f); std::vector ts; ASSERT_TRUE(dec.timestamps(f, 0, ts)); ASSERT_EQ(ts.size(), 4u); EXPECT_NEAR(ts[0], 2000.000, 1e-9); EXPECT_NEAR(ts[3], 2000.003, 1e-9); } TEST(FrameDecoder, LastSampleAnchorsTheFinalElementAndCountsBackward) { FrameDecoder dec; dec.setSignals({burst("Sine", kTimeLastSample, 1000.0, 4, 1), timeSignal("Time", 1)}); FrameBuilder fb; fb.addSignal({1.0, 2.0, 3.0, 4.0}); fb.addSignal({7.0e9}); const FrameView& f = fb.build(0, 3000.0); dec.beginFrame(f); std::vector ts; ASSERT_TRUE(dec.timestamps(f, 0, ts)); ASSERT_EQ(ts.size(), 4u); EXPECT_NEAR(ts[3], 3000.000, 1e-9); EXPECT_NEAR(ts[0], 3000.000 - 0.003, 1e-9); } TEST(FrameDecoder, PlainScalarUsesArrivalTime) { FrameDecoder dec; SignalMeta m; m.name = "Level"; m.typeCode = 9; dec.setSignals({m}); FrameBuilder fb; fb.addSignal({42.0}); const FrameView& f = fb.build(0, 1234.5); dec.beginFrame(f); std::vector ts; ASSERT_TRUE(dec.timestamps(f, 0, ts)); ASSERT_EQ(ts.size(), 1u); EXPECT_DOUBLE_EQ(ts[0], 1234.5); } // This is the failure UDPSourceSession.cpp:560 documents. The kernel delivers // two queued datagrams microseconds apart even though each carries 10 ms of // signal. Dating from arrival crams the second packet's samples into that gap // and the trace becomes a sawtooth; dating from the producer hrt does not. TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { FrameDecoder dec; SignalMeta m; m.name = "Acc"; m.typeCode = 9; m.numRows = 1; m.samplingRate = 1000.0; /* 1 kHz, 10 samples = 10 ms per packet */ dec.setSignals({m}); const double ticks = 1.0e9; std::vector all; for (int p = 0; p < 40; p++) { FrameBuilder fb; fb.addSignal(std::vector(10, static_cast(p))); const double producerSec = 100.0 + p * 0.010; /* Packets 20+ arrive in a burst, all within 50 us of each other. */ 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, static_cast(p + 1)); dec.beginFrame(f); std::vector ts; if (dec.timestamps(f, 0, ts)) { all.insert(all.end(), ts.begin(), ts.end()); } } ASSERT_GT(all.size(), 300u); for (size_t i = 1; i < all.size(); i++) { EXPECT_GT(all[i], all[i - 1]) << "non-monotonic at " << i; EXPECT_NEAR(all[i] - all[i - 1], 0.001, 2e-4) << "spacing collapsed at " << i << " (sawtooth)"; } } 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: 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. 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); dec.beginFrame(f); ASSERT_TRUE(dec.timestamps(f, 0, ts)); /* Chaining blindly gives 500.091; anchoring on arrival gives 501.291. */ EXPECT_NEAR(ts[0], 501.091, 1e-9); EXPECT_NEAR(ts[9], 501.100, 1e-9); } // 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()}); std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/false); FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 501.100, 10, 0u); dec.beginFrame(f); ASSERT_TRUE(dec.timestamps(f, 0, ts)); EXPECT_NEAR(ts[0], 501.091, 1e-9); EXPECT_NEAR(ts[9], 501.100, 1e-9); } // Re-anchoring must never move a signal's timestamps backwards: the ring, the // trigger and the exporter all assume they increase, 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()}); std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); const double prevEnd = ts[9]; FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& f = fb.build(0, 500.000, 10, 2010u); dec.beginFrame(f); ASSERT_TRUE(dec.timestamps(f, 0, ts)); /* Arrival (500.000) is behind our timeline, so there is nothing to spread * into; the burst is 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]); } } // When the chain has to be abandoned but arrival lies just ahead of where the // last burst ended, the correction is made by COMPRESSING this one burst rather // than by stepping back. Rejecting the correction instead would be one-directional // — `predicted` is never below lastEmittedEnd + dt — and a timeline running fast // could then never be pulled back. TEST(FrameDecoder, AccumulatedScalarCompressesOneBurstRatherThanStepBack) { FrameDecoder dec; dec.setSignals({accSignal()}); std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); /* 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); dec.beginFrame(f); ASSERT_TRUE(dec.timestamps(f, 0, ts)); EXPECT_GT(ts[0], 500.090) << "compressed burst must still start after the last one"; EXPECT_NEAR(ts[9], 500.095, 1e-9) << "and end exactly on arrival"; EXPECT_NEAR(ts[1] - ts[0], 0.0005, 1e-9) << "spread over the available room"; } // The whole point of compressing: a declared SamplingRate is a hand-written // config value, and even a correct one is measured against the producer host's // crystal, not ours. Tens of ppm of difference is certain over a long session, // so the reconstructed timeline WILL run away from the wall clock. It has to be // pulled back, and it has to stay monotonic while that happens. TEST(FrameDecoder, AccumulatedScalarDoesNotDriftAwayFromTheWallClockForever) { FrameDecoder dec; dec.setSignals({accSignal()}); /* declares 1 kHz */ /* The producer really runs 1 % fast: 10 samples take 9.9 ms of wall time, * so a chain stepping the declared 10 ms per packet gains 0.1 ms every * packet. This is the direction re-anchoring alone cannot fix: arrival is * always BEHIND the chain, so anchoring on it would step backwards and is * refused. Only compression pulls the timeline back. */ double worstLead = 0.0; double lastEnd = 0.0; for (int p = 0; p < 20000; p++) { FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const double arrival = 500.0 + p * 0.0099; const FrameView& f = fb.build(0, arrival, 10, static_cast(p + 1)); dec.beginFrame(f); std::vector ts; ASSERT_TRUE(dec.timestamps(f, 0, ts)); for (size_t i = 0; i < ts.size(); i++) { ASSERT_GT(ts[i], lastEnd) << "timeline went backwards at packet " << p; lastEnd = ts[i]; } worstLead = std::max(worstLead, ts[9] - arrival); } /* Unchecked, 20000 packets at 0.1 ms each would put the trace 2 s ahead. */ EXPECT_LT(worstLead, 0.6) << "timeline drifted " << worstLead << " s ahead"; } // The 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. TEST(FrameDecoder, AccumulatedScalarDropsADuplicatedDatagram) { FrameDecoder dec; dec.setSignals({accSignal()}); std::vector ts; primeTenBursts(dec, ts, /*withCounter=*/true); const double endBefore = ts[9]; FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); const FrameView& dup = fb.build(0, 500.1001, 10, 10u); /* counter 10 again */ dec.beginFrame(dup); EXPECT_FALSE(dec.timestamps(dup, 0, ts)); /* And the drop must not have disturbed the chain: the genuine next packet * still lands one period after burst 10 ended. */ const FrameView& next = fb.build(0, 500.109, 10, 11u); dec.beginFrame(next); ASSERT_TRUE(dec.timestamps(next, 0, ts)); EXPECT_NEAR(ts[0], endBefore + 0.001, 1e-6); } // 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; m.name = "Acc"; m.typeCode = 9; m.samplingRate = 0.0; /* undeclared */ dec.setSignals({m}); const double ticks = 1.0e9; std::vector last; for (int p = 0; p < 40; p++) { FrameBuilder fb; fb.addSignal(std::vector(10, 1.0)); /* 25 ms per packet, deliberately NOT 10: at 10 the expected 1 ms period * equals kDefaultDt, so a decoder that never derived anything and just * returned the default would pass a test named for the derivation. */ const double producerSec = 100.0 + p * 0.025; /* Zero-mean arrival jitter, so the rate fit still converges but no * single arrival GAP is right. Without it, uniform arrivals make * packetBurst and the hrt path return the same number by construction * and the test cannot tell which branch answered. */ const double jitter[4] = {0.0, 0.003, 0.0, -0.003}; const FrameView& f = fb.build(static_cast(producerSec * ticks), 700.0 + p * 0.025 + jitter[p % 4], 10, static_cast(p + 1)); dec.beginFrame(f); std::vector ts; if (dec.timestamps(f, 0, ts)) { last = ts; } } ASSERT_EQ(last.size(), 10u); /* 25 ms of producer time across 10 samples is a 2.5 ms period, whatever the * datagrams did on the way over. Arrival-spanning the last gap (22 ms) * would give 2.2 ms; defaulting would give 1 ms. */ EXPECT_NEAR(last[1] - last[0], 0.0025, 2e-5); } // 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]; } } } 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 // packet overlap the next one and break ring monotonicity. TEST(FrameDecoder, PacketBurstDropsTheFirstFrameThenSpansBackwards) { FrameDecoder dec; dec.setSignals({burst("Raw", kTimePacket, 0.0, 5, kNoTimeSignal)}); FrameBuilder fb1; fb1.addSignal({1.0, 2.0, 3.0, 4.0, 5.0}); const FrameView& f1 = fb1.build(0, 10.0); dec.beginFrame(f1); std::vector ts; EXPECT_FALSE(dec.timestamps(f1, 0, ts)) << "the first packet has no previous arrival to span from"; FrameBuilder fb2; fb2.addSignal({6.0, 7.0, 8.0, 9.0, 10.0}); const FrameView& f2 = fb2.build(0, 10.05); dec.beginFrame(f2); ASSERT_TRUE(dec.timestamps(f2, 0, ts)); ASSERT_EQ(ts.size(), 5u); EXPECT_GT(ts[0], 10.0); EXPECT_NEAR(ts[4], 10.05, 1e-12); EXPECT_NEAR(ts[1] - ts[0], 0.01, 1e-12); } TEST(FrameDecoder, PacketBurstStaysMonotonicUnderJitteredArrivals) { FrameDecoder dec; dec.setSignals({burst("Raw", kTimePacket, 0.0, 8, kNoTimeSignal)}); const double jitter[] = {0.0, 0.004, -0.003, 0.006, -0.002, 0.0, 0.005, -0.004}; std::vector all; for (int p = 0; p < 8; p++) { FrameBuilder fb; fb.addSignal(std::vector(8, 1.0)); const FrameView& f = fb.build(0, 20.0 + p * 0.05 + jitter[p]); dec.beginFrame(f); std::vector ts; if (dec.timestamps(f, 0, ts)) { all.insert(all.end(), ts.begin(), ts.end()); } } ASSERT_GT(all.size(), 8u); for (size_t i = 1; i < all.size(); i++) { EXPECT_GT(all[i], all[i - 1]) << "packets overlapped at " << i; } } TEST(FrameDecoder, ResetForgetsPerSignalHistory) { FrameDecoder dec; dec.setSignals({burst("Raw", kTimePacket, 0.0, 4, kNoTimeSignal)}); FrameBuilder fb; fb.addSignal({1.0, 2.0, 3.0, 4.0}); const FrameView& f = fb.build(0, 5.0); dec.beginFrame(f); std::vector ts; EXPECT_FALSE(dec.timestamps(f, 0, ts)); const FrameView& f2 = fb.build(0, 5.1); dec.beginFrame(f2); EXPECT_TRUE(dec.timestamps(f2, 0, ts)); dec.reset(); const FrameView& f3 = fb.build(0, 5.2); dec.beginFrame(f3); EXPECT_FALSE(dec.timestamps(f3, 0, ts)) << "after reset the next packet is again the first one"; }