Fix/viewer replay egress tuning

ChangeLog.md

@@ -37,6 +37,9 @@ The entries below cover work done in the IRL fork on top of the upstream `rstula
 - `api_keys` directive supports comma separated keys with constant time comparison.
 - Atomic stats counters and a read locked `put()` so `/stats` no longer stalls the data path.
 - Replaced the blocking HTTP client with an `AsyncHttpClient` (thread pool backed) for stats posting, player key validation, and `on_event_url` callbacks.
+- New per stream `/stats` diagnostics for the jump/replay class of viewer issue: `maxReaderBacklogBytes` and `maxReaderBacklogMs` (furthest any viewer fell behind the ring write head, the size of a potential catch up burst) alongside the existing `ringOverruns`.
+- Fixed `sendBackpressure`, which was read off the publisher role (which never runs the egress write path) and so was structurally always zero. It is now aggregated across a stream's viewers via the shared ring, so it reflects real viewer backpressure.
+- Teardown log lines (`check_invalid_sock`, `get_state`) now carry `stream=<name>` for grep based correlation across many concurrent streams, and a flight recorder line at publisher teardown records the session's peak backlog, overruns, and viewer backpressure before the ring is freed (survives publisher reconnect).
 
 **Logging redesign**
 
@@ -66,6 +69,7 @@ The entries below cover work done in the IRL fork on top of the upstream `rstula
 - Player latency clamped via `SRTO_PEERLATENCY`; bumped UDP buffers; explicit `SRTO_TLPKTDROP`.
 - Disconnect viewers stuck in continuous backpressure; handle `EASYNCSND` backpressure without disconnecting healthy viewers.
 - Race free `m_nDataCount`, write locked `setSize` on the publisher ring; size publisher ring per bitrate with reader overrun detection.
+- Egress tuned to stop live viewers replaying stale video. The publisher ring is sized at 1x the latency window instead of 2x, and `MAX_EGRESS_BATCHES` is reduced from 8 to 2, so a viewer that falls behind is skipped forward per packet by SRT `TLPKTDROP` at the socket rather than catching up in a large burst of old frames (which a viewer perceives as a jump back then skip forward). The ring is treated as a small hand off buffer, not a jitter buffer; the SRT socket holds each viewer's latency window.
 - Compile out per packet `SPDLOG_TRACE` / `SPDLOG_DEBUG` on the data path; drop per packet string allocations.
 - Race and leak fixes on the accept path (close socket and free `addrinfo` on `libsrt_setup` errors; stop leaking `stat_info_t` per accepted role; null check `strdup(sid)`; kick publisher via atomic flag in `disconnect_stream`).
 - Removed HLS recording.

src/core/SLSGroup.cpp

@@ -449,10 +449,10 @@ void CSLSGroup::check_invalid_sock()
         int state = role->get_state(cur_time_ms);
         if (SLS_RS_INVALID == state || SLS_RS_UNINIT == state)
         {
-            spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={}, invalid sock={:d}, "
+            spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={} stream={}, invalid sock={:d}, "
                          "state={:d}, role_map.size={:d}.",
-                         fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()), role->get_fd(),
-                         state, m_map_role.size());
+                         fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()),
+                         role->get_map_data_key(), role->get_fd(), state, m_map_role.size());
             // check relay
             if (role->is_reconnect())
             {
@@ -495,8 +495,9 @@ void CSLSGroup::check_invalid_sock()
             }
             else
             {
-                spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={}, delete.", fmt::ptr(this),
-                             m_worker_number, role->get_role_name(), fmt::ptr(role.get()));
+                spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={} stream={}, delete.",
+                             fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()),
+                             role->get_map_data_key());
             }
             m_map_role.erase(it_erase);
             continue;

src/core/SLSManager.cpp

@@ -445,15 +445,30 @@ json CSLSManager::create_json_stats_for_publisher(CSLSRole *role, int clear)
     // subscriber's read position was forcibly resynced to the write head
     // to avoid handing back corrupted wrapped-around data.
     ret["ringOverruns"] = role->get_ring_overrun_count();
-    // Egress send-buffer backpressure events. Counts how often
-    // srt_sendmsg to this role returned EASYNCSND (SRT send buffer
-    // full). Each event means the viewer's link could not absorb a
-    // write burst and SLS deferred the remainder to the next epoll
-    // wake instead of disconnecting the viewer. Steady growth on a
-    // player role indicates that viewer's link is under-provisioned
-    // for the stream bitrate or that the player negotiated too small
-    // a latency window.
-    ret["sendBackpressure"] = role->get_send_backpressure_count();
+    // Egress send-buffer backpressure events, aggregated across every viewer of
+    // this stream. Counts how often srt_sendmsg to a viewer returned EASYNCSND
+    // (SRT send buffer full) — the viewer's link could not absorb a write burst,
+    // so SLS deferred the remainder to the next epoll wake instead of dropping
+    // the viewer. Sourced from the shared publisher ring, NOT the publisher
+    // role's own per-role counter: /stats enumerates only publishers, and a
+    // publisher never runs the egress write path, so that counter was always 0.
+    // Steady growth means viewers are falling behind — under-provisioned links
+    // for the stream bitrate, or a viewer latency window that is too small.
+    ret["sendBackpressure"] = role->get_viewer_backpressure_events(clear);
+    // Furthest any viewer of this stream fell behind the publisher ring write
+    // head (bytes) since the last clear. This is the catch-up burst a viewer
+    // will drain when it recovers, which the viewer perceives as a time-skip /
+    // "replay". Divide by the stream bitrate for a millisecond figure
+    // (maxReaderBacklogMs below). 0 on a healthy stream where every viewer keeps
+    // up with the write head.
+    int64_t max_backlog_bytes = role->get_max_reader_backlog(clear);
+    ret["maxReaderBacklogBytes"] = max_backlog_bytes;
+    // Same figure expressed as playout time, so operators do not have to do the
+    // bitrate math. ms = bytes * 8 / kbps. Only meaningful once a bitrate is
+    // known; reported as 0 otherwise.
+    int bitrate_kbps = role->get_bitrate();
+    ret["maxReaderBacklogMs"] =
+        (max_backlog_bytes > 0 && bitrate_kbps > 0) ? (int64_t)(max_backlog_bytes * 8 / bitrate_kbps) : 0;
 
     ret["audioGapFill"] = json::object();
     ret["audioGapFill"]["enabled"] = audio_gap_stats.enabled;

src/core/SLSMapData.cpp

@@ -80,19 +80,25 @@ int CSLSMapData::add(char *key, int max_bitrate_kbps, int latency_ms)
     // Decide the ring size up front (without resizing yet) so the memory cap
     // can be enforced before we commit the (potentially large) allocation.
     // The CSLSRecycleArray default (get_data_size()) is the floor; a bitrate +
-    // latency hint grows it so a subscriber falling up to one full SRT latency
-    // window behind is still safe from overruns. We multiply by 2 to give a
-    // viewer's effective lag headroom against publisher-side jitter. Skip the
-    // growth if the caller passed no hint — the default is enough for typical
-    // bitrates.
+    // latency hint grows it so a subscriber falling up to one SRT latency window
+    // behind is still safe from overruns. Sized at 1x the latency window (was
+    // 2x): the ring is a hand-off buffer, NOT the jitter buffer — the viewer's
+    // own SRT socket holds its latency window and drops late packets via
+    // TLPKTDROP. Oversizing the ring only lets a slow viewer bank multiple
+    // seconds of stale data that then replays as a rewind. One latency window is
+    // enough to ride out publisher-side jitter without overrun; anything beyond
+    // it should be skipped forward at the socket, not stored. Skip the growth if
+    // the caller passed no hint — the default is enough for typical bitrates.
+    // Watch /stats ringOverruns after changing this: 0 means the ring is still
+    // large enough; sustained growth means it was cut too small for the bitrate.
     int64_t intended_size = (int64_t)data_array->get_data_size();
     if (max_bitrate_kbps > 0 && latency_ms > 0)
     {
         // bytes_per_sec is bounded by ~max_bitrate_kbps (config caps in
         // SLSPublisher.hpp limit it to 1_000_000 kbps = 125 MB/s).
         int64_t bytes_per_sec = (int64_t)max_bitrate_kbps * 1000 / 8;
-        // 2x the latency window in seconds, clamped to a minimum of 1s.
-        int64_t window_secs = (int64_t)latency_ms * 2 / 1000;
+        // 1x the latency window in seconds, clamped to a minimum of 1s.
+        int64_t window_secs = (int64_t)latency_ms / 1000;
         if (window_secs < 1)
             window_secs = 1;
         int64_t target_bytes = bytes_per_sec * window_secs;
@@ -177,6 +183,39 @@ int64_t CSLSMapData::get_overrun_count(const char *key)
     return it->second->get_overrun_count();
 }
 
+int64_t CSLSMapData::get_max_reader_backlog(const char *key, bool clear)
+{
+    if (!key)
+        return -1;
+    CSLSLock lock(&m_rwclock, false);
+    auto it = m_map_array.find(std::string_view{key});
+    if (it == m_map_array.end() || it->second == NULL)
+        return -1;
+    return it->second->get_max_reader_backlog(clear);
+}
+
+void CSLSMapData::report_viewer_backpressure(const char *key)
+{
+    if (!key)
+        return;
+    CSLSLock lock(&m_rwclock, false);
+    auto it = m_map_array.find(std::string_view{key});
+    if (it == m_map_array.end() || it->second == NULL)
+        return;
+    it->second->report_viewer_backpressure();
+}
+
+int64_t CSLSMapData::get_viewer_backpressure_events(const char *key, bool clear)
+{
+    if (!key)
+        return -1;
+    CSLSLock lock(&m_rwclock, false);
+    auto it = m_map_array.find(std::string_view{key});
+    if (it == m_map_array.end() || it->second == NULL)
+        return -1;
+    return it->second->get_viewer_backpressure_events(clear);
+}
+
 int CSLSMapData::remove(char *key)
 {
     int ret = SLS_ERROR;

src/core/SLSMapData.hpp

@@ -99,6 +99,14 @@ class CSLSMapData
     // (writer lapped the reader). Returns -1 if `key` is unknown.
     int64_t get_overrun_count(const char *key);
 
+    // Diagnostic gauges for the stream `key`, forwarded to its ring. See the
+    // matching CSLSRecycleArray accessors. Return -1 (getters) / no-op
+    // (reporter) if `key` is unknown. `clear` resets the high-water so /stats
+    // can report a per-poll-interval peak.
+    int64_t get_max_reader_backlog(const char *key, bool clear = false);
+    void report_viewer_backpressure(const char *key);
+    int64_t get_viewer_backpressure_events(const char *key, bool clear = false);
+
     int put(char *key, char *data, int len, int64_t *last_read_time = NULL);
     void set_audio_gap_fill(const char *key, bool enabled);
     bool get_audio_gap_stats(const char *key, AudioGapStreamStats &stats, int clear = 0);

src/core/SLSPublisher.cpp

@@ -111,6 +111,16 @@ int CSLSPublisher::uninit()
 
     if (m_map_data)
     {
+        // Flight recorder: the diagnostic gauges live on the ring we are about
+        // to free, so publisher takeover / reconnect would otherwise erase the
+        // one session's data an operator most wants (the streamer got fed up and
+        // reconnected). Flush the ring's final peaks to a single greppable line
+        // (stream=<name>) before remove() drops it. One line per session end, so
+        // it stays quiet even with many concurrent streams.
+        spdlog::info("[{}] CSLSPublisher::uninit, session end stream={}, peakReaderBacklogBytes={}, "
+                     "ringOverruns={}, viewerBackpressure={}.",
+                     fmt::ptr(this), m_map_data_key, get_max_reader_backlog(false), get_ring_overrun_count(),
+                     get_viewer_backpressure_events(false));
         ret = m_map_data->remove(m_map_data_key);
         spdlog::info("[{}] CSLSPublisher::uninit, removed publisher from m_map_data, ret={:d}.", fmt::ptr(this), ret);
     }

src/core/SLSRecycleArray.cpp

@@ -39,12 +39,25 @@
 // fit (see CSLSMapData::add overload).
 const int DEFAULT_MAX_DATA_SIZE = 8 * 1024 * 1024;
 
+// Monotonic-max update of an atomic without a lock. Used off the read path to
+// keep a high-water gauge; relaxed ordering is fine (the value is diagnostic and
+// has no happens-before relationship with the data it describes).
+static void atomic_store_max(std::atomic<int64_t> &slot, int64_t candidate)
+{
+    int64_t prev = slot.load(std::memory_order_relaxed);
+    while (candidate > prev && !slot.compare_exchange_weak(prev, candidate, std::memory_order_relaxed))
+    {
+    }
+}
+
 CSLSRecycleArray::CSLSRecycleArray()
 {
     m_nDataSize = DEFAULT_MAX_DATA_SIZE;
     m_nWritePos = 0;
     m_nDataCount.store(0, std::memory_order_relaxed);
     m_overrun_count.store(0, std::memory_order_relaxed);
+    m_max_reader_backlog.store(0, std::memory_order_relaxed);
+    m_viewer_backpressure_events.store(0, std::memory_order_relaxed);
 
     m_last_read_time.store(sls_gettime_ms(), std::memory_order_relaxed);
 
@@ -183,6 +196,13 @@ int CSLSRecycleArray::get(char *data, int size, SLSRecycleArrayID *read_id, int
         return SLS_OK;
     }
 
+    // Diagnostic high-water: how far behind the write head this reader was on
+    // entry (i.e. the backlog it is about to drain). A large value means the
+    // reader's next drains will burst out a catch-up, which a viewer perceives
+    // as a time-skip. Non-overrun path only — the overrun case is capped at the
+    // buffer size and already has its own counter. See get_max_reader_backlog.
+    atomic_store_max(m_max_reader_backlog, bytes_since_last_read);
+
     SPDLOG_TRACE(
         "[{}] CSLSRecycleArray::get, read_id->nReadPos={:d}, m_nWritePos={:d}, m_nDataCount={:d}, m_nDataSize={:d}.",
         fmt::ptr(this), read_id->nReadPos, m_nWritePos, cur_data_count, m_nDataSize);
@@ -260,3 +280,34 @@ int64_t CSLSRecycleArray::get_last_read_time()
 {
     return m_last_read_time.load(std::memory_order_acquire);
 }
+
+int64_t CSLSRecycleArray::get_reader_backlog(const SLSRecycleArrayID *read_id) const
+{
+    if (NULL == read_id || read_id->bFirst)
+        return 0;
+    int64_t behind = m_nDataCount.load(std::memory_order_relaxed) - read_id->nDataCount;
+    if (behind < 0)
+        behind = 0;
+    if (behind > m_nDataSize)
+        behind = m_nDataSize;
+    return behind;
+}
+
+int64_t CSLSRecycleArray::get_max_reader_backlog(bool clear)
+{
+    if (clear)
+        return m_max_reader_backlog.exchange(0, std::memory_order_relaxed);
+    return m_max_reader_backlog.load(std::memory_order_relaxed);
+}
+
+void CSLSRecycleArray::report_viewer_backpressure()
+{
+    m_viewer_backpressure_events.fetch_add(1, std::memory_order_relaxed);
+}
+
+int64_t CSLSRecycleArray::get_viewer_backpressure_events(bool clear)
+{
+    if (clear)
+        return m_viewer_backpressure_events.exchange(0, std::memory_order_relaxed);
+    return m_viewer_backpressure_events.load(std::memory_order_relaxed);
+}

src/core/SLSRecycleArray.hpp

@@ -70,6 +70,30 @@ class CSLSRecycleArray
         return m_overrun_count.load(std::memory_order_relaxed);
     }
 
+    // How many bytes this specific reader is behind the write head right now,
+    // i.e. the backlog it would burst out on its next drain. 0 for a reader
+    // that has not yet anchored (bFirst) or has caught up. Clamped to the
+    // buffer size because a reader further behind than that has already been
+    // (or is about to be) overrun-resynced. Lock-free: m_nDataCount is atomic
+    // and read_id is owned by the calling worker thread. Diagnostic only — this
+    // is the metric that answers "is SLS holding a catch-up burst for a viewer".
+    int64_t get_reader_backlog(const SLSRecycleArrayID *read_id) const;
+
+    // High-water of get_reader_backlog seen across ALL readers since the last
+    // clear. This is the per-stream signal an operator polls to see whether any
+    // viewer fell far enough behind that its eventual catch-up drain is bursty
+    // (visible to that viewer as a time-skip). clear=true resets it so /stats
+    // can report a per-interval peak instead of a lifetime max.
+    int64_t get_max_reader_backlog(bool clear = false);
+
+    // Aggregate egress backpressure across the viewers reading this ring. The
+    // per-role counter lived on the player role, but /stats only enumerates
+    // publishers, so it was never surfaced. Players report here (a shared,
+    // publisher-visible object) on each EASYNCSND so publisher /stats can show
+    // real viewer backpressure instead of the publisher role's always-zero one.
+    void report_viewer_backpressure();
+    int64_t get_viewer_backpressure_events(bool clear = false);
+
 private:
     char *m_arrayData;
     int m_nDataSize;
@@ -91,5 +115,11 @@ class CSLSRecycleArray
     std::atomic<int64_t> m_last_read_time{0};
     std::atomic<int64_t> m_overrun_count;
 
+    // Diagnostic gauges (see the public accessors). Written off the read path in
+    // get() and by viewer roles reporting backpressure; read by the /stats HTTP
+    // thread. Atomic, relaxed — purely observational, no ordering requirements.
+    std::atomic<int64_t> m_max_reader_backlog{0};
+    std::atomic<int64_t> m_viewer_backpressure_events{0};
+
     CSLSRWLock m_rwclock;
 };