Merge fix/egress-blocking-send: move egress player send off the QUIC loop

Introduces a bounded PlayerQueue (stale-media drop per draft s11) and a
dedicated writer thread so a stalled RTMP player can no longer wedge the
egress QUIC network thread. on_frame/on_player_play only enqueue; the
writer does the blocking send outside player_mutex. Teardown stops the
QUIC producer, shuts the player fd to unblock a stalled send, closes the
queue, joins the writer, then destroys sessions. Reconnect closes and
flushes the prior player. Stall test pins SO_RCVBUF so drops are
deterministic across host TCP tuning.

Author: mondain

gateway/CMakeLists.txt

@@ -3,6 +3,7 @@ add_library(roqr-gateway STATIC
   src/rtmp_commands.cpp
   src/ingest.cpp
   src/egress.cpp
+  src/player_queue.cpp
 )
 
 target_include_directories(roqr-gateway PUBLIC

gateway/include/roqr/gateway/egress.hpp

@@ -27,6 +27,9 @@ class EgressGateway {
     void stop();
     bool wait_playing(std::chrono::milliseconds timeout);
 
+    // Number of media messages dropped under player backpressure (draft s11).
+    uint64_t frames_dropped() const;
+
 private:
     struct Impl;
     std::unique_ptr<Impl> impl_;

gateway/include/roqr/gateway/player_queue.hpp

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <mutex>
+#include <optional>
+
+#include "roqr/rtmp/message.hpp"
+
+namespace roqr::gateway {
+
+// Thread-safe bounded queue between the RoQR receive thread (producer) and
+// the egress writer thread (consumer). When full, the oldest Coded entry is
+// evicted so decoder config (metadata + sequence headers, Kind::Init) is
+// never dropped while stale coded media is (draft s11). close() unblocks a
+// waiting consumer so the writer thread can exit.
+class PlayerQueue {
+public:
+    enum class Kind { Init, Coded };  // Init = metadata / sequence header
+
+    explicit PlayerQueue(size_t max_messages) : max_(max_messages) {}
+
+    // Enqueue. On overflow, drop the oldest Coded entry to make room; if the
+    // queue holds only Init entries (degenerate) drop the incoming Coded
+    // message instead of growing. Returns false if the message was dropped
+    // (either the incoming one or the evicted victim's slot is the caller's
+    // signal that a drop happened). Never blocks.
+    bool push(roqr::rtmp::RtmpMessage msg, Kind kind);
+
+    // Block until a message is available or the queue is closed. Returns
+    // nullopt only when the queue is closed AND drained.
+    std::optional<roqr::rtmp::RtmpMessage> pop();
+
+    void close();
+    size_t size() const;
+    uint64_t dropped() const;
+
+    // Drop all queued entries without counting them toward dropped() — used
+    // when installing a new player so stale prior-era frames don't linger
+    // ahead of the new player's init frames; this is a deliberate flush, not
+    // the loss dropped() tracks.
+    void clear();
+
+private:
+    struct Entry {
+        roqr::rtmp::RtmpMessage msg;
+        Kind kind;
+    };
+    mutable std::mutex mutex_;
+    std::condition_variable cv_;
+    std::deque<Entry> queue_;
+    size_t max_;
+    uint64_t dropped_ = 0;
+    bool closed_ = false;
+};
+
+}  // namespace roqr::gateway

gateway/src/egress.cpp

@@ -5,10 +5,12 @@
 #include <map>
 #include <mutex>
 #include <string>
+#include <thread>
 #include <vector>
 
 #include "roqr/gateway/bridge.hpp"
 #include "roqr/gateway/gap.hpp"
+#include "roqr/gateway/player_queue.hpp"
 #include "roqr/gateway/rtmp_commands.hpp"
 #include "roqr/quic/client.hpp"
 #include "roqr/rtmp/classify.hpp"
@@ -39,10 +41,34 @@ struct EgressGateway::Impl {
 
     roqr::gateway::GapTracker gaps;  // draft s8, defined in gap.hpp
 
+    // Bounded queue + dedicated writer thread (draft s11): on_frame (QUIC
+    // network thread) and on_player_play (session thread) only enqueue;
+    // the writer thread does the blocking player->send, so a stalled RTMP
+    // player never wedges the QUIC loop. Must be declared before `client`
+    // (see below) so they're destroyed after it re: construction order,
+    // but they're joined/closed explicitly in stop() well before that.
+    static constexpr size_t kMaxQueuedMessages = 512;
+    PlayerQueue queue{kMaxQueuedMessages};
+    std::thread writer;
+    bool writer_started = false;
+
     // Declared last: ~Client joins the network thread before the members
     // its handlers touch are destroyed.
     roqr::quic::Client client;
 
+    void writer_loop() {
+        for (;;) {
+            auto msg = queue.pop();
+            if (!msg) return;  // closed and drained
+            roqr::rtmp::ServerSession* p = nullptr;
+            {
+                std::lock_guard lock(player_mutex);
+                if (player_ready) p = player;
+            }
+            if (p != nullptr) p->send(*msg);
+        }
+    }
+
     static bool is_init(const roqr::rtmp::RtmpMessage& msg) {
         if (msg.type == roqr::rtmp::kTypeDataAmf0 || msg.type == 15 /* AMF3 data */)
             return true;
@@ -67,23 +93,32 @@ struct EgressGateway::Impl {
         if (msg.type == roqr::rtmp::kTypeCommandAmf0) return;  // relay replies
         if (msg.type == 9 && !accept_video(msg)) return;
 
-        std::lock_guard lock(player_mutex);
-        if (is_init(msg)) {
-            if (init_cache.find(msg.type) == init_cache.end()) {
-                init_types.push_back(msg.type);
+        const bool init = is_init(msg);
+        {
+            std::lock_guard lock(player_mutex);
+            if (init) {
+                if (init_cache.find(msg.type) == init_cache.end()) {
+                    init_types.push_back(msg.type);
+                }
+                init_cache[msg.type] = msg;
             }
-            init_cache[msg.type] = msg;
+            if (!player_ready) return;  // pre-play: cache only, don't enqueue
         }
-        if (player != nullptr && player_ready) player->send(msg);
+        queue.push(std::move(msg),
+                   init ? PlayerQueue::Kind::Init : PlayerQueue::Kind::Coded);
     }
 
     // Called on the session thread when the player issues play (after the
-    // RTMP handshake). Primes it with the cached init frames, then opens
-    // live delivery.
+    // RTMP handshake). Enqueues the cached init frames (so the writer
+    // thread delivers them, not the session thread) and sets player_ready
+    // under the lock before releasing, so on_frame cannot interleave a live
+    // coded frame ahead of the init frames.
     void on_player_play() {
         std::lock_guard lock(player_mutex);
         if (player == nullptr) return;
-        for (uint8_t type : init_types) player->send(init_cache.at(type));
+        for (uint8_t type : init_types) {
+            queue.push(init_cache.at(type), PlayerQueue::Kind::Init);
+        }
         player_ready = true;
     }
 
@@ -113,12 +148,24 @@ EgressGateway::~EgressGateway() { stop(); }
 bool EgressGateway::start(const EgressOptions& options) {
     impl_->options = options;
     Impl* impl = impl_.get();
+    if (!impl->writer_started) {
+        impl->writer = std::thread([impl] { impl->writer_loop(); });
+        impl->writer_started = true;
+    }
     impl->begin_play();  // connect + play before accepting the player
     return impl->listener.start(
         options.rtmp_port,
         [impl](roqr::rtmp::ServerSession& s) {
             {
                 std::lock_guard lock(impl->player_mutex);
+                // A previous player may still be alive (e.g. a stalled one
+                // that never disconnected): shut its fd down so any writer
+                // blocked in its send() unblocks, then drop its queued
+                // frames so the new player isn't fed stale, prior-era media
+                // ahead of its own init frames.
+                if (impl->player != nullptr && impl->player != &s)
+                    impl->player->close();
+                impl->queue.clear();
                 impl->player = &s;
                 impl->player_ready = false;
             }
@@ -139,9 +186,26 @@ bool EgressGateway::start(const EgressOptions& options) {
 }
 
 void EgressGateway::stop() {
-    impl_->listener.stop();
+    // 1. Stop the QUIC network thread so on_frame stops enqueuing.
     impl_->client.close();
     impl_->client.wait_closed(std::chrono::seconds(2));
+    // 2. Unblock a writer that may be stuck in a blocking send to a stalled
+    //    player: shut down the player's fd so the send returns.
+    {
+        std::lock_guard lock(impl_->player_mutex);
+        if (impl_->player != nullptr) impl_->player->close();
+    }
+    // 3. Close the queue and join the writer (it drains, sends fail fast on
+    //    the shut-down fd, then exits).
+    impl_->queue.close();
+    if (impl_->writer.joinable()) impl_->writer.join();
+    impl_->writer_started = false;
+    // 4. Now no thread touches the player: safe to destroy the sessions.
+    impl_->listener.stop();
+}
+
+uint64_t EgressGateway::frames_dropped() const {
+    return impl_->queue.dropped();
 }
 
 bool EgressGateway::wait_playing(std::chrono::milliseconds timeout) {

gateway/src/player_queue.cpp

@@ -0,0 +1,64 @@
+#include "roqr/gateway/player_queue.hpp"
+
+#include <algorithm>
+
+namespace roqr::gateway {
+
+bool PlayerQueue::push(roqr::rtmp::RtmpMessage msg, Kind kind) {
+    std::lock_guard lock(mutex_);
+    if (closed_) return false;
+    bool dropped_one = false;
+    if (queue_.size() >= max_) {
+        auto victim = std::find_if(queue_.begin(), queue_.end(),
+                                   [](const Entry& e) {
+                                       return e.kind == Kind::Coded;
+                                   });
+        if (victim != queue_.end()) {
+            queue_.erase(victim);
+            ++dropped_;
+            dropped_one = true;
+        } else {
+            // Only Init entries and still full: drop the incoming rather
+            // than exceed the bound.
+            ++dropped_;
+            return false;
+        }
+    }
+    queue_.push_back(Entry{std::move(msg), kind});
+    cv_.notify_one();
+    return !dropped_one;
+}
+
+std::optional<roqr::rtmp::RtmpMessage> PlayerQueue::pop() {
+    std::unique_lock lock(mutex_);
+    cv_.wait(lock, [&] { return !queue_.empty() || closed_; });
+    if (queue_.empty()) return std::nullopt;  // closed and drained
+    roqr::rtmp::RtmpMessage m = std::move(queue_.front().msg);
+    queue_.pop_front();
+    return m;
+}
+
+void PlayerQueue::close() {
+    {
+        std::lock_guard lock(mutex_);
+        closed_ = true;
+    }
+    cv_.notify_all();
+}
+
+size_t PlayerQueue::size() const {
+    std::lock_guard lock(mutex_);
+    return queue_.size();
+}
+
+uint64_t PlayerQueue::dropped() const {
+    std::lock_guard lock(mutex_);
+    return dropped_;
+}
+
+void PlayerQueue::clear() {
+    std::lock_guard lock(mutex_);
+    queue_.clear();
+}
+
+}  // namespace roqr::gateway

tests/CMakeLists.txt

@@ -93,6 +93,7 @@ if(ROQR_BUILD_EXAMPLES)
     gateway/rtmp_commands_test.cpp
     gateway/media_router_test.cpp
     gateway/gap_recovery_test.cpp
+    gateway/player_queue_test.cpp
   )
   target_link_libraries(roqr-gateway-tests PRIVATE roqr-gateway roqr-relayd-lib Catch2::Catch2WithMain)
   catch_discover_tests(roqr-gateway-tests PROPERTIES TIMEOUT 60)

tests/gateway/player_queue_test.cpp

@@ -0,0 +1,96 @@
+#include <catch2/catch_test_macros.hpp>
+
+#include <thread>
+
+#include "roqr/gateway/player_queue.hpp"
+
+using namespace roqr::gateway;
+using roqr::rtmp::RtmpMessage;
+
+namespace {
+RtmpMessage vid(uint32_t ts) {
+    RtmpMessage m;
+    m.type = 9;
+    m.timestamp = ts;
+    m.payload = {0x27, 0x01};
+    return m;
+}
+RtmpMessage seq_header() {
+    RtmpMessage m;
+    m.type = 9;
+    m.payload = {0x17, 0x00};
+    return m;
+}
+}  // namespace
+
+TEST_CASE("queue is FIFO under the bound") {
+    PlayerQueue q(8);
+    CHECK(q.push(vid(1), PlayerQueue::Kind::Coded));
+    CHECK(q.push(vid(2), PlayerQueue::Kind::Coded));
+    CHECK(q.size() == 2);
+    CHECK(q.pop()->timestamp == 1);
+    CHECK(q.pop()->timestamp == 2);
+}
+
+TEST_CASE("overflow drops the oldest coded frame, stays bounded") {
+    PlayerQueue q(4);
+    for (uint32_t i = 0; i < 4; ++i) {
+        REQUIRE(q.push(vid(i), PlayerQueue::Kind::Coded));
+    }
+    // Now full. Pushing more evicts the oldest coded frame each time.
+    q.push(vid(100), PlayerQueue::Kind::Coded);
+    q.push(vid(101), PlayerQueue::Kind::Coded);
+    CHECK(q.size() == 4);          // bounded
+    CHECK(q.dropped() == 2);       // two evictions
+    // The two oldest (ts 0,1) were evicted; front is now ts 2.
+    CHECK(q.pop()->timestamp == 2);
+}
+
+TEST_CASE("sequence headers are never evicted") {
+    PlayerQueue q(3);
+    REQUIRE(q.push(seq_header(), PlayerQueue::Kind::Init));
+    REQUIRE(q.push(vid(1), PlayerQueue::Kind::Coded));
+    REQUIRE(q.push(vid(2), PlayerQueue::Kind::Coded));
+    // Full. Two more coded frames evict the coded ones, keeping the Init.
+    q.push(vid(3), PlayerQueue::Kind::Coded);
+    q.push(vid(4), PlayerQueue::Kind::Coded);
+    CHECK(q.size() == 3);
+    // The Init (seq header) is still at the front.
+    auto first = q.pop();
+    REQUIRE(first.has_value());
+    CHECK(first->payload == std::vector<uint8_t>{0x17, 0x00});
+}
+
+TEST_CASE("close unblocks a waiting consumer and drains remaining") {
+    PlayerQueue q(8);
+    q.push(vid(1), PlayerQueue::Kind::Coded);
+    q.close();
+    CHECK(q.pop()->timestamp == 1);  // drains the queued item
+    CHECK_FALSE(q.pop().has_value()); // then reports closed+empty
+
+    PlayerQueue q2(8);
+    std::thread consumer([&] {
+        auto m = q2.pop();  // blocks until close()
+        CHECK_FALSE(m.has_value());
+    });
+    q2.close();
+    consumer.join();
+}
+
+TEST_CASE("clear drops all entries without counting them as dropped") {
+    PlayerQueue q(8);
+    REQUIRE(q.push(seq_header(), PlayerQueue::Kind::Init));
+    REQUIRE(q.push(vid(1), PlayerQueue::Kind::Coded));
+    REQUIRE(q.push(vid(2), PlayerQueue::Kind::Coded));
+    CHECK(q.size() == 3);
+    CHECK(q.dropped() == 0);
+
+    q.clear();
+
+    CHECK(q.size() == 0);
+    CHECK(q.dropped() == 0);  // clear() is not a drop-tracked eviction
+
+    // Queue is still usable after clear().
+    REQUIRE(q.push(vid(3), PlayerQueue::Kind::Coded));
+    CHECK(q.pop()->timestamp == 3);
+}