fix(core): track consumed bytes so reader overrun detects true lag

the overrun check compared the writer's byte counter against a reader
counter that was snapped to the writer's value on every get(), so it
measured writer progress since the last call, not accumulated lag. a
reader draining less than the writer produces per cycle (ingest above
the ~33 mbps per-viewer egress ceiling) could be lapped without the
check firing, silently reading wrapped garbage, while the backlog
gauges undercounted the same way. advance the counter by bytes actually
consumed instead: the check becomes an exact true-lag invariant and
maxReaderBacklog reports real backlog under partial drains.

Author: datagutt

src/core/SLSRecycleArray.cpp

@@ -208,14 +208,16 @@ int CSLSRecycleArray::get(char *data, int size, SLSRecycleArrayID *read_id, int
         return SLS_OK;
     }
 
-    // Overrun detection: if the writer has produced more than m_nDataSize
-    // bytes since this reader last sampled the buffer, the contents the
-    // reader was about to consume have already been overwritten by newer
-    // data. Without this check we'd silently hand back a wrapped-around
-    // region of the ring containing bytes that don't belong to the
-    // reader's logical position — producing corrupt TS / out-of-order
-    // delivery to the subscriber. Force the reader to resync to the
-    // current write head and count the event for diagnostics.
+    // Overrun detection: read_id->nDataCount tracks bytes CONSUMED by this
+    // reader (advanced by copy_data_len below), so the delta against the
+    // writer's monotonic byte counter is the reader's true accumulated lag —
+    // including lag built up across partial drains, where each get() copies
+    // only a slice of a larger backlog. If that lag reaches the buffer size
+    // the writer has physically overwritten the reader's unread region, and
+    // reading on would hand back a wrapped-around mix of newer bytes at the
+    // reader's stale logical position — corrupt TS / out-of-order delivery.
+    // Force the reader to resync to the current write head and count the
+    // event for diagnostics.
     //
     // A negative delta is equally disqualifying: m_nDataCount is monotonic
     // for a given ring incarnation, so a reader counter ahead of the ring's
@@ -310,7 +312,12 @@ int CSLSRecycleArray::get(char *data, int size, SLSRecycleArrayID *read_id, int
                      read_id->nReadPos, m_nDataSize);
         read_id->nReadPos = 0;
     }
-    read_id->nDataCount = cur_data_count;
+    // Advance by what was actually consumed, NOT to cur_data_count: snapping
+    // to the writer's counter would silently forgive any backlog this call
+    // didn't drain, letting a persistently under-draining reader be lapped
+    // without the overrun check above ever firing (and making the backlog
+    // gauges undercount in exactly that case).
+    read_id->nDataCount += copy_data_len;
     SPDLOG_TRACE("[{}] CSLSRecycleArray::get, copy_data_lens={:d}.", fmt::ptr(this), copy_data_len);
     return copy_data_len;
 }

tests/test_recycle_array.cpp

@@ -391,6 +391,55 @@ TEST_CASE("CSLSRecycleArray: setSize invalidates readers anchored on the old buf
     CHECK(std::memcmp(out, chunk, sizeof(chunk)) == 0);
 }
 
+// A reader that is CALLED every cycle but drains less than the writer
+// produces accumulates lag across partial drains. The overrun check compares
+// the writer's counter against bytes CONSUMED, so this lap must be detected
+// even though every per-call writer delta is far below the ring size. (The
+// old code snapped the reader counter to the writer's on every call, so this
+// reader was lapped silently and read wrapped garbage forever.)
+TEST_CASE("CSLSRecycleArray: partially draining reader is overrun-resynced at true lag")
+{
+    const int RING = 64;
+    CSLSRecycleArray ring;
+    ring.setSize(RING);
+
+    SLSRecycleArrayID id = fresh_reader();
+    anchor(ring, id);
+
+    // Writer adds 16/cycle, reader drains at most 8/cycle: true lag grows by
+    // 8 per iteration and passes RING at iteration 8, while each per-call
+    // writer delta stays at 16 (well under RING).
+    char chunk[16];
+    char out[8];
+    bool overran = false;
+    int64_t peak_backlog = 0;
+    for (int i = 0; i < 40 && !overran; i++)
+    {
+        std::memset(chunk, (char)('A' + (i % 26)), sizeof(chunk));
+        CHECK(ring.put(chunk, sizeof(chunk)) == (int)sizeof(chunk));
+        int got = ring.get(out, sizeof(out), &id, 0);
+        CHECK(got >= 0);
+        int64_t backlog = ring.get_max_reader_backlog(false);
+        if (backlog > peak_backlog)
+            peak_backlog = backlog;
+        if (ring.get_overrun_count() > 0)
+            overran = true;
+    }
+    CHECK(overran);
+    // The backlog gauge must have tracked the true accumulated lag, not the
+    // per-call writer delta (which never exceeded 16).
+    CHECK(peak_backlog > 16);
+
+    // After the resync the reader is at the live head and receives exactly
+    // the fresh data.
+    char fresh[16];
+    std::memset(fresh, 'Z', sizeof(fresh));
+    CHECK(ring.put(fresh, sizeof(fresh)) == (int)sizeof(fresh));
+    char big[64] = {0};
+    CHECK(ring.get(big, sizeof(big), &id, 0) == (int)sizeof(fresh));
+    CHECK(std::memcmp(big, fresh, sizeof(fresh)) == 0);
+}
+
 // Backstop for the generation check: a reader whose byte counter is ahead of
 // the ring's monotonic counter cannot belong to this incarnation. It must be
 // resynced, never used to index the buffer.