fix: a claimed gap must be corroborated before it is believed, and its hole must fit

The user's real host reported "connected and live" with one missing-data band over the whole window
and no trace in it. The counter is the only thing that claims a packet existed, and a reservation
was booked on its word alone: a source whose DATA counter counts samples, or frames of an inner
layer, advances by more than 1 per datagram, the client cannot check that claim (it has no clock of
its own to compare against), so it holds each packet, saturates its reorder ring, delivers it in
order and reports the skipped counters as lost. Every frame therefore reported a phantom gap, and
`reserve_hole()` wrote `lost * batch_` NaN samples with no bound but the 256-PACKET cap -- the
sample form of that rule ("the number to weigh against the ring is 256 * batch_, not the 256") was
never implemented. With relative time on -- which that session carries -- each phantom gap is
materialised as NaN slots by construction: the window filled with reservations, there was never
anything missing to repair them, and the gaps also stretched the relative clock, which is why the
status bar read live at ten times the true rate.

`Acquisition::gap_plausible()` now gates every reservation on two independent tests: the arrival
clock must corroborate the claim (that many missing packets would take that many frame intervals;
deliberately permissive, because a false refusal costs a shading and never data), and the ring must
be able to hold the hole. The verdict is taken before the stamps, and relative mode's slot advance
is skipped with it, since the advance and the reservation are the same claim. Refusals are counted
in `StreamStats::gaps_refused` and said once per connection with the numbers, naming the wire shape
they describe.

tools/mock_udpstreamer.py gains `--counter-step N`: a counter that is not a packet counter on an
intact link, which is the one shape the mock could not produce and the shape the report came from.

Verified in both directions. A counter stepping 1000 per datagram reads `holes=0 hole_slots=0
backfilled=0` with the traces present (3 traces, 1.4 M samples) against 7.9 M claimed gaps, and the
warning above names the cause. Every other packet withheld -- a real loss -- still reads
`holes=3579`, so genuine gaps are still reserved: a fix that stopped believing real losses would
have traded one wrong picture for another. The user's control read as "the math channel is the
problem" and was confounded -- disabling it is a structural change, which rebuilds the buffer and
clears the accumulated NaN.

src/acquisition.cpp

@@ -203,6 +203,7 @@ void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count,
   // next session is the failure mode worth avoiding in a feature whose whole job is diagnosis.
   warned_batch_ = false;
   warned_outage_ = false;
+  warned_phantom_gap_ = false;
   warned_bridge_ = false;
   warned_time_source_ = false;
   // The clock choice is per connection too: a reconnect re-derives it from the frames
@@ -243,9 +244,61 @@ void Acquisition::reset_session_state() noexcept {
   have_arrival_interval_ = false;
   arrival_interval_ = 0.0;
   arrival_prev_ = 0.0;
+  have_prev_frame_recv_ = false;
+  prev_frame_recv_ = 0.0;
 }
 
-std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal) {
+void Acquisition::note_frame_arrival(double recv_time) noexcept {
+  const bool first = !have_prev_frame_recv_;
+  const double elapsed = first ? 0.0 : recv_time - prev_frame_recv_;
+  prev_frame_recv_ = recv_time;
+  have_prev_frame_recv_ = true;
+  // The arrival *cadence* belongs to the arrival clock, and `build_stamps()`'s arrival branch keeps
+  // it whenever that clock is the one in play (`time_source_ == 2`, decided once per connection and
+  // sticky). Every other case needs it here: relative mode returns before that branch, and a stream
+  // whose declared time signal checks out never reaches it either -- while the reservation test
+  // below weighs the counter's claim against a measured cadence whatever clock the display uses,
+  // because a counter that does not count packets lies in all of them. The two writers are therefore
+  // exclusive by `time_source_`, and this is the branch's own update (0.8/0.2) so the estimate does
+  // not jump at the one frame where the clock changes hands.
+  if (first || time_source_ == 2) return;
+  if (elapsed > 0.0 && std::isfinite(elapsed)) {
+    arrival_interval_ =
+        have_arrival_interval_ ? 0.8 * arrival_interval_ + 0.2 * elapsed : elapsed;
+    have_arrival_interval_ = true;
+  }
+}
+
+bool Acquisition::gap_plausible(std::uint32_t missing, double elapsed, const Partition& part,
+                                u64 g) const {
+  if (missing == 0) return true;
+  // The ring cannot hold a hole bigger than itself, and this test needs no measurement: the
+  // invariant's own rule is that the reservation's *sample* cost -- `missing * batch_`, not the
+  // packet cap -- is what has to be weighed against the ring. A hole that laps the ring erases the
+  // history it is meant to annotate, can never be backfilled (its slots are already overwritten),
+  // and is never forgotten (`forget_holes_before()` waits for slots the ring has passed), so it
+  // leaves a band over a buffer that is all NaN.
+  if (g == 0 || part.time_capacity == 0) return false;
+  if (u64(missing) * u64(batch_) / g >= part.time_capacity) return false;
+  // And the counter has to agree with the wire about how long the loss took. The counter is a claim
+  // about the stream; the arrival clock is a *measurement* of it: `missing` packets would have taken
+  // `missing` frame intervals to arrive, and the frame before this one is the witness of how long
+  // actually passed. A device whose counter counts samples, or frames of an inner layer, reports a
+  // forward jump on *every* datagram while the wire delivers on time -- and believing that booked a
+  // NaN hole per packet, i.e. a band over the whole window filled with samples that were never
+  // missing, which no backfill can repair because no packet was ever lost. The factor is a half and
+  // not one: the socket clock jitters and a drain can compress two frames' arrival times, so only a
+  // claim the wire flatly contradicts (half the time it needs, or less) is refused. A refused gap
+  // costs a shading, never data.
+  if (have_arrival_interval_ && arrival_interval_ > 0.0 && elapsed > 0.0 &&
+      elapsed < 0.5 * double(missing) * arrival_interval_) {
+    return false;
+  }
+  return true;
+}
+
+std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal,
+                                        bool counter_gap_refused) {
   // The frame's span in base ticks: a scalar signal carries `num_samples` values
   // (one per batched cycle), a block signal (rows*cols > 1) carries its whole
   // element block. The largest of those is how many base ticks this frame spans --
@@ -294,8 +347,11 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
     if (!(dt > 0.0)) dt = 1.0e-3;
     // The reserved hole for the missing packets advances `slot_base_` so this frame's
     // own stamps land *after* the hole rather than on top of it. The same conditions
-    // the reservation call uses: not late, gap present, batch known, within the cap.
-    if (!frame->late && frame->lost > 0 && batch_ > 0 &&
+    // the reservation call uses: not late, gap present, batch known, within the cap --
+    // and not one the arrival clock or the ring refused, because the advance and the
+    // reservation are the *same* claim about the wire: believing a phantom gap here
+    // would stretch the timeline by exactly the loss that did not happen.
+    if (!frame->late && frame->lost > 0 && batch_ > 0 && !counter_gap_refused &&
         double(frame->lost) <= kGlitchPackets) {
       slot_base_ += u64(frame->lost) * batch_;
     }
@@ -615,7 +671,20 @@ void Acquisition::on_data(const udps_frame_t* frame) {
       }
     }
   }
-  const std::uint32_t n = build_stamps(frame, time_signal);
+  // The wire's own cadence, and the verdict on the counter's claim, both taken *before* the
+  // stamps: relative mode's `slot_base_` advance and the reservation below are the same claim
+  // about the wire (see `build_stamps`), so they have to agree about it. The verdict is taken
+  // before the cadence is updated, too: folding this frame's own arrival into the estimate would
+  // let one burst raise the very number that is supposed to corroborate it -- a 200-packet outage
+  // smooths the interval to 40 ms and then refuses itself.
+  const bool first_frame = !have_prev_frame_recv_;
+  const double since_prev = first_frame ? 0.0 : frame->recv_time - prev_frame_recv_;
+  const std::uint32_t missing = frame->lost;
+  const bool gap_refused =
+      missing > 0 && !gap_plausible(missing, since_prev, part, part.time_stride);
+  note_frame_arrival(frame->recv_time);
+
+  const std::uint32_t n = build_stamps(frame, time_signal, gap_refused);
   if (n == 0) return;
 
   if (model_.epoch() == 0.0) {
@@ -695,8 +764,29 @@ void Acquisition::on_data(const udps_frame_t* frame) {
   // through the same decimation accumulators (value_acc_, value_acc_n_, math_sample_),
   // which must already be sized to the current channel count and stride before any
   // per-sample write.
-  const std::uint32_t missing = frame->lost;
-  if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) {
+  if (missing > 0 && gap_refused) {
+    // A claim the wire contradicted, or a hole the ring cannot hold: framing it as NaN is what
+    // turns one phantom gap into a band over the whole window (and, at a batch big enough, into
+    // minutes of the producer's time under this lock). The samples are kept, no hole is booked,
+    // and the timeline above was not advanced either -- so the picture shows the data and the gap
+    // shows as the x discontinuity it is. Said once per connection with the numbers, because it is
+    // otherwise invisible: the status bar's `gaps` climbs either way.
+    model_.stats().gaps_refused.fetch_add(1, std::memory_order_relaxed);
+    if (!warned_phantom_gap_) {
+      warned_phantom_gap_ = true;
+      std::fprintf(stderr,
+                   "[udpscope] warning: the counter reports %u packet(s) missing (%llu samples, "
+                   "%.3f s at the measured cadence of %.6f s/frame) but only %.3f s passed since "
+                   "the previous frame, or the hole is larger than the ring (%llu slots); "
+                   "refusing the reservation and keeping the samples. A counter that does not "
+                   "count packets -- samples, or frames of an inner layer -- reports a jump on "
+                   "every datagram and reads exactly like this.\n",
+                   missing, static_cast<unsigned long long>(std::uint64_t(missing) * batch_),
+                   double(missing) * arrival_interval_, arrival_interval_, since_prev,
+                   static_cast<unsigned long long>(u64(missing) * u64(batch_) / part.time_stride));
+      std::fflush(stderr);
+    }
+  } else if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) {
     reserve_hole(missing, part, channels, store, g);
   } else if (missing > 0 && !warned_outage_) {
     warned_outage_ = true;