feat: a late packet backfills the hole it left, values only

src/acquisition.cpp

@@ -465,20 +465,25 @@ void Acquisition::on_data(const udps_frame_t* frame) {
     have_counter_ = true;
   }
 
-  // A late frame's samples belong in the hole already reserved for them, not at the end of
-  // the timeline. Appending them would walk the time ring's clock backwards, which trips the
-  // non-monotonic guard and can throw the whole stream onto the arrival clock -- a far worse
-  // outcome than the dropped packet we used to have. Until the backfill lands (RingBuffer::amend
-  // and MinMaxPyramid::amend), late frames are dropped here, leaving the reserved hole holding
-  // the NaN it was reserved with -- a value every reader already treats as absent. This is
-  // exactly the behaviour the client had before it started delivering late packets, so nothing is
-  // lost relative to that. This return is the single place the backfill will later hook into.
-  if (frame->late) return;
-
   const Partition& part = model_.partition();
   const std::vector<ChannelInfo>& channels = model_.channels();
   if (part.time_capacity == 0 || channels.empty()) return;
 
+  // A frame delivered behind the sequence: the slots it owns were reserved as NaN by
+  // `reserve_hole()`, so it writes values into them and does nothing else. It is deliberately
+  // not stamped, does not advance `tick_`, and never reaches the trigger or the math channels:
+  // the time ring was pre-stamped at reservation and those consumers are stateful and have
+  // already run past this region. That is the trade the design takes -- the raw trace repairs,
+  // a derived trace keeps the gap, and nothing that was computed from the gap is retrofitted.
+  //
+  // Placed here and not earlier because `part`/`channels` are not in scope above, and not
+  // later because `build_stamps()` is stateful: a late frame's stamps would walk the clock
+  // backwards and can throw the whole stream onto the arrival clock.
+  if (frame->late) {
+    backfill_from(frame, part, channels, model_.live(), part.time_stride);
+    return;
+  }
+
   // The time reference, in order of preference: the signal the first channel's
   // CONFIG entry names via `time_signal_idx`, or -- when the stream instead
   // carries its time base as an explicit full-array signal -- that signal.
@@ -549,7 +554,7 @@ 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->late ? 0u : frame->lost;
+  const std::uint32_t missing = frame->lost;
   if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) {
     reserve_hole(missing, part, channels, store, g);
   } else if (missing > 0 && !warned_outage_) {
@@ -852,6 +857,74 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
   model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed);
 }
 
+void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part,
+                                const std::vector<ChannelInfo>& channels, BufferStore& store,
+                                u64 g) {
+  // `batch_` is learned in `build_stamps()`, which a late frame never reaches. A late frame
+  // before any good one therefore has no packet length to size its span with, and no hole
+  // could have been reserved for it either.
+  if (batch_ == 0) return;
+
+  // Which hole this packet belongs to, by counter: that is the whole reason relative timing
+  // and resequencing are one design. A stamp could not answer it -- the packet's own stamps
+  // are a claim about a timeline the ring has moved past -- but the counter indexes the
+  // reservation table directly.
+  const Hole* hole = nullptr;
+  for (const Hole& h : model_.holes()) {
+    if (frame->counter >= h.counter_first && frame->counter < h.counter_first + h.packets) {
+      hole = &h;
+      break;
+    }
+  }
+  if (hole == nullptr) return;  // its hole has already slid out of the ring, or never existed
+
+  // The packet's own span in absolute base ticks. `tick_begin` is the hole's first raw sample
+  // and the packets inside a hole are consecutive, so the counter's distance from
+  // `counter_first` places this one exactly.
+  const u64 lo = hole->tick_begin + u64(frame->counter - hole->counter_first) * u64(batch_);
+  const u64 hi = lo + u64(batch_);
+
+  u64 filled = 0;
+  for (std::size_t i = 0; i < channels.size(); ++i) {
+    // A derived channel has no packet signal to read, and the design does not retrofit it:
+    // its gap stays.
+    if (channels[i].math) continue;
+    const u64 period = g * part.stride[i];
+    // Only slots lying *wholly* inside this packet's span are repaired. A slot straddling the
+    // span's edge mixes this packet's samples with ones still missing, and its real members
+    // were never retained -- `reserve_hole()` averaged a NaN into them -- so it stays NaN, and
+    // a displayed hole is at most one slot wider per edge than the true loss.
+    const u64 first_slot = (lo + period - 1) / period;
+    const u64 last_slot = hi / period;
+    for (u64 s = first_slot; s < last_slot; ++s) {
+      // The same boxcar mean the live path stores, over the same group of raw samples: the
+      // whole group is inside this packet, so the average is exact and not an approximation
+      // of what the live path would have written.
+      double acc = 0.0;
+      for (u64 j = 0; j < period; ++j) {
+        const u64 k = s * period + j - lo;  // this packet's own sample index, 0..batch_-1
+        const double raw =
+            channels[i].elements > 1
+                ? udps_frame_value(frame, static_cast<std::uint32_t>(channels[i].signal), 0,
+                                   std::min<std::uint32_t>(static_cast<std::uint32_t>(k),
+                                                           channels[i].elements - 1))
+                : udps_frame_value(frame, static_cast<std::uint32_t>(channels[i].signal),
+                                   static_cast<std::uint32_t>(k), 0);
+        acc += raw * channels[i].gain + channels[i].offset;
+      }
+      const f32 value = static_cast<f32>(acc / double(period));
+      // The ring first, by the producer that owns its writes; the pyramid is queued for the
+      // consumer (see `ChannelBuffer::queue_amend`). A slot the ring has already discarded
+      // returns false, and then there is no aggregate to amend either.
+      if (store.values(i).amend(s, value)) {
+        store.channel(i).queue_amend(s, value);
+        ++filled;
+      }
+    }
+  }
+  model_.stats().backfilled.fetch_add(filled, std::memory_order_relaxed);
+}
+
 // Static trampolines: the client's C ABI cannot call a member function directly.
 void Acquisition::config_trampoline(const udps_signal_t* signals, std::uint32_t count,
                                     std::uint8_t publish_mode, void* user) {

src/acquisition.hpp

@@ -61,6 +61,12 @@ class Acquisition {
   void reserve_hole(std::uint32_t missing, const Partition& part,
                     const std::vector<ChannelInfo>& channels, BufferStore& store, u64 g);
 
+  /// Writes a late frame's values into the NaN-filled holes their counter addresses,
+  /// amending the value ring directly (producer path) and queuing pyramid amends for
+  /// `ChannelBuffer::acquire()` to apply on the consumer thread.
+  void backfill_from(const udps_frame_t* frame, const Partition& part,
+                     const std::vector<ChannelInfo>& channels, BufferStore& store, u64 g);
+
   ScopeModel& model_;
   Config cfg_;
   udps_client* client_ = nullptr;

src/channel.cpp

@@ -31,41 +31,50 @@ ChannelBuffer::ChannelBuffer(std::shared_ptr<const plot::data::RingBuffer<f32>>
 
 void ChannelBuffer::acquire() noexcept {
   const u64 produced = values_->produced();
-  if (produced <= published_) {
+  if (produced > published_) {
+    // A producer that lapped this ring -- a long trigger wait, or a store that was
+    // not being published for a while -- makes the incremental append meaningless:
+    // the samples between `published_` and `produced_` are no longer in the ring.
+    // Rebuilding from the retained window is O(capacity) once, and it is the only
+    // way the published window stays coherent (an empty or garbled window is what
+    // "the trace disappeared" looks like from outside).
     // Nothing new; still allow the window to slide if another reader dropped it.
-    return;
+    if (produced - published_ > values_->capacity()) {
+      published_ = 0;
+      pyramid_.clear();
+    }
+    f32 chunk[kChunk];
+    for (u64 i = published_; i < produced;) {
+      const u64 take = std::min<u64>(kChunk, produced - i);
+      values_->copy_from(i, take, chunk);
+      for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]);
+      i += take;
+    }
+    published_ = produced;
+    // The pyramid's window follows the ring's: sample index `base` is the oldest
+    // one still retained. drop_before() is monotonic and cheap.
+    pyramid_.drop_before(values_->base());
+    // And the window's first stamp with it, here and nowhere else (see the member's note): the
+    // window is pinned by the line above, so its origin is a fixed sample, and this is the one
+    // place in the object's life where that sample's slot is safe to read.
+    if (pyramid_.size() > 0) window_x0_ = time_of(absolute(0));
   }
-  // A producer that lapped this ring -- a long trigger wait, or a store that was
-  // not being published for a while -- makes the incremental append meaningless:
-  // the samples between `published_` and `produced_` are no longer in the ring.
-  // Rebuilding from the retained window is O(capacity) once, and it is the only
-  // way the published window stays coherent (an empty or garbled window is what
-  // "the trace disappeared" looks like from outside).
-  if (produced - published_ > values_->capacity()) {
-    published_ = 0;
-    pyramid_.clear();
+  // Backfills, applied after the append pass above: an amend is only ever queued for a slot
+  // the value ring already produced, so the append has reached it and the pyramid's bounds
+  // test will accept it. Unconditional, outside the `produced > published_` guard -- a late
+  // packet can land in a frame interval that pushed nothing new, and the plan's placement
+  // behind the old early return would have left the pyramid stale until the next push.
+  if (!pending_amends_.empty()) {
+    for (const Amend& a : pending_amends_) pyramid_.amend(a.absolute, a.value);
+    pending_amends_.clear();
   }
-  f32 chunk[kChunk];
-  for (u64 i = published_; i < produced;) {
-    const u64 take = std::min<u64>(kChunk, produced - i);
-    values_->copy_from(i, take, chunk);
-    for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]);
-    i += take;
-  }
-  published_ = produced;
-  // The pyramid's window follows the ring's: sample index `base` is the oldest
-  // one still retained. drop_before() is monotonic and cheap.
-  pyramid_.drop_before(values_->base());
-  // And the window's first stamp with it, here and nowhere else (see the member's note): the
-  // window is pinned by the line above, so its origin is a fixed sample, and this is the one
-  // place in the object's life where that sample's slot is safe to read.
-  if (pyramid_.size() > 0) window_x0_ = time_of(absolute(0));
 }
 
 void ChannelBuffer::invalidate() noexcept {
   published_ = 0;
   window_x0_ = 0.0;
   pyramid_.clear();
+  pending_amends_.clear();
 }
 
 f64 ChannelBuffer::time_of(u64 absolute_index) const noexcept {

src/channel.hpp

@@ -49,6 +49,17 @@ class ChannelBuffer final : public plot::data::Series {
   /// Consumer thread: forgets the published window (after a rebind or a reset).
   void invalidate() noexcept;
 
+  /// Producer thread, under `ScopeModel::mutex()`: records that absolute sample `absolute`
+  /// has been backfilled, for `acquire()` to fold into the pyramid.
+  ///
+  /// The pyramid is mutated *only* on the consumer thread, because a bucket is a 32-byte
+  /// non-atomic aggregate the renderer reads unlocked at paint time: amending it from the
+  /// producer would let a reader see a half-updated bucket whose `sum` did not match its
+  /// `count`, which is a wrong mean out of a structure whose readers trust `count == 0` to
+  /// mean absent. The value ring is amended by the producer, which owns its writes; the
+  /// pyramid waits for here.
+  void queue_amend(u64 absolute, f32 v) { pending_amends_.push_back({absolute, v}); }
+
   void set_nominal_dt(f64 dt) noexcept { dt_ = dt; }
   /// Re-bases the x axis: `x0()`/`x_at()` become `epoch + timestamp`. Used to make
   /// the axis read seconds from the trigger event instead of from the stream.
@@ -100,6 +111,13 @@ class ChannelBuffer final : public plot::data::Series {
   /// Timestamp of channel sample `absolute_index`, in seconds since `epoch_`.
   f64 time_of(u64 absolute_index) const noexcept;
 
+  struct Amend {
+    u64 absolute;
+    f32 value;
+  };
+  /// Backfills the producer has written into the value ring and the pyramid has not seen yet.
+  std::vector<Amend> pending_amends_;
+
   std::shared_ptr<const plot::data::RingBuffer<f32>> values_;
   std::shared_ptr<const plot::data::RingBuffer<f64>> times_;
   u64 stride_ = 1;