Bound how long the packet queue holds a frame back

app/videonative/src/main/cpp/BufferedPacketQueue.h

@@ -5,20 +5,39 @@
 #include <cstdio>
 #endif
 #include <algorithm>
+#include <chrono>
 #include <cstdarg>
 #include <cstddef>
 #include <cstdint>
+#include <cstdlib>
 #include <limits>
 
 #include <unordered_map>
 #include <vector>
 
 // Define logging tag and maximum buffer size
 #define BUFFERED_QUEUE_LOG_TAG "BufferedPacketQueue"
-// Considering the packet rate about 100 packets per second, 10 packets should be enough
+// Last-resort cap on the buffer. How long this is in wall clock time depends entirely on the
+// packet rate, which is why it cannot be the only bound - see MAX_BUFFER_AGE.
 constexpr size_t MAX_BUFFER_SIZE = 15;
 // Number of monotonically increasing packets
 constexpr size_t MONOTONIC_THRESHOLD = 5;
+// A monotonic run is only tracked while the gap is plausibly a reorder. Kept separate from
+// MONOTONIC_THRESHOLD: the gap grows by one with every packet held back, so gating the counter
+// on the same value it is compared against means it can never reach it.
+constexpr size_t MONOTONIC_MAX_DISTANCE = 64;
+
+// Clock used to bound how long a packet may be held back.
+using QueueClock     = std::chrono::steady_clock;
+using QueueTimePoint = QueueClock::time_point;
+
+// A gap in the sequence numbers on this path is almost always a packet that FEC could not
+// recover, not a reorder: by the time packets get here they have come through wfb-ng and a
+// loopback socket, where reordering takes microseconds. Waiting for a packet that will never
+// arrive is pure added latency, so the wait is bounded in time rather than in packets - the
+// packet bound alone is worth ~20ms on a 1080p video stream but ~300ms on the audio stream,
+// which runs at a fraction of the packet rate.
+constexpr auto MAX_BUFFER_AGE = std::chrono::milliseconds(20);
 
 // Type definition for sequence numbers
 using SeqType   = uint16_t;
@@ -45,14 +64,32 @@ class BufferedPacketQueue
      * @param callback Callable to handle processed packets.
      */
     template <typename Callback>
-    void processPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, Callback& callback)
+    void processPacket(
+        SeqType         currPacketIdx,
+        const uint8_t*  data,
+        std::size_t     data_length,
+        Callback&       callback,
+        QueueTimePoint  now = QueueClock::now())
     {
         logDebug(
             "Processing packet with Sequence=%u, lastPacketIdx=%u, firstPacket=%s",
             currPacketIdx,
             mLastPacketIdx,
             mFirstPacket ? "true" : "false");
 
+        // Before anything else: give up on a gap we have been waiting on for too long. Done
+        // here rather than in handleOutOfOrderPacket so that an in-order packet arriving after
+        // a stall does not get delivered ahead of what is already buffered.
+        if (!mPackets.empty() && (now - mOldestBufferedAt) >= MAX_BUFFER_AGE)
+        {
+            logWarning(
+                "Held %zu packet(s) for more than %lldms waiting on Sequence=%u. Flushing.",
+                mPackets.size(),
+                (long long) MAX_BUFFER_AGE.count(),
+                static_cast<unsigned>(static_cast<SeqType>(mLastPacketIdx + 1)));
+            mLastPacketIdx = drainBufferInOrder(callback);
+        }
+
         if (isFirstPacket(currPacketIdx))
         {
             handleFirstPacket(currPacketIdx);
@@ -73,7 +110,7 @@ class BufferedPacketQueue
         else
         {
             // Out-of-order packet
-            handleOutOfOrderPacket(currPacketIdx, data, data_length, callback);
+            handleOutOfOrderPacket(currPacketIdx, data, data_length, callback, now);
         }
     }
 
@@ -83,6 +120,9 @@ class BufferedPacketQueue
 
     std::unordered_map<SeqType, std::vector<uint8_t>> mPackets;
 
+    // When the current stall started, i.e. when mPackets last went from empty to non-empty.
+    QueueTimePoint mOldestBufferedAt{};
+
     // This variable is used to track a situation where the sequence number is increasing monotonically while packets
     // are out of order. if this counter reaches MONOTONIC_THRESHOLD, we will restart buffering and update lastPacketIdx
     // to the highest sequence index received.
@@ -174,7 +214,12 @@ class BufferedPacketQueue
      * @param callback Callable to handle processed packets.
      */
     template <typename Callback>
-    void handleOutOfOrderPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, Callback& callback)
+    void handleOutOfOrderPacket(
+        SeqType        currPacketIdx,
+        const uint8_t* data,
+        std::size_t    data_length,
+        Callback&      callback,
+        QueueTimePoint now)
     {
         logDebug("Out-of-order packet detected. Sequence=%u", currPacketIdx);
 
@@ -184,10 +229,15 @@ class BufferedPacketQueue
             // return;
         }
 
-        bufferPacket(currPacketIdx, data, data_length);
+        bufferPacket(currPacketIdx, data, data_length, now);
 
-        auto dist = calculateDistance(currPacketIdx, mLastPacketIdx);
-        if (std::abs(dist) < MONOTONIC_THRESHOLD)
+        // calculateDistance(a, b) is how far b is ahead of a - see seqLessThan below - so the
+        // question "is this packet ahead of the last one we delivered" has to be asked in that
+        // order. Reversed, dist is negative for exactly the case this heuristic exists for (a
+        // gap ahead of us), the else branch below clears the counter every time, and the
+        // buffer only ever drains on the MAX_BUFFER_SIZE cap.
+        auto dist = calculateDistance(mLastPacketIdx, currPacketIdx);
+        if (static_cast<size_t>(std::abs(dist)) < MONOTONIC_MAX_DISTANCE)
         {
             // Check for monotonic increases
             if (dist > 0)
@@ -196,10 +246,8 @@ class BufferedPacketQueue
                 logDebug("Monotonic increase count: %zu", mMonotonicOutOfOrderIncreaseCount);
                 if (mMonotonicOutOfOrderIncreaseCount >= MONOTONIC_THRESHOLD)
                 {
-                    restartBuffering(callback, currPacketIdx);
-                    // Update lastPacketIdx to the highest sequence index received
-                    SeqType newLastIdx = currPacketIdx;
-                    logWarning("Monotonic threshold reached. Updating lastPacketIdx to %u", newLastIdx);
+                    mLastPacketIdx = drainBufferInOrder(callback);
+                    logWarning("Monotonic threshold reached. Updating lastPacketIdx to %u", mLastPacketIdx);
                 }
             }
             else
@@ -214,7 +262,7 @@ class BufferedPacketQueue
         {
             logWarning(
                 "Buffer size exceeded MAX_BUFFER_SIZE (%zu). Processing in-order buffered packets.", MAX_BUFFER_SIZE);
-            restartBuffering(callback, currPacketIdx);
+            mLastPacketIdx = drainBufferInOrder(callback);
         }
     }
 
@@ -231,19 +279,27 @@ class BufferedPacketQueue
      * @param data Pointer to the packet data.
      * @param data_length Size of the packet data.
      */
-    void bufferPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length)
+    void bufferPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, QueueTimePoint now)
     {
+        // Only the start of a stall is recorded, because the buffer is always drained as a
+        // whole. A partial drain leaves the mark where it was, which errs towards flushing
+        // early - the safe direction on a live link.
+        if (mPackets.empty())
+        {
+            mOldestBufferedAt = now;
+        }
         mPackets[currPacketIdx] = std::vector<uint8_t>(data, data + data_length);
         logDebug("Buffered out-of-order packet. Buffer size: %zu", mPackets.size());
     }
 
     /**
-     * @brief Handles buffer overflow by processing in-order packets and discarding others.
+     * @brief Delivers everything currently held back, in sequence order, and empties the buffer.
      * @tparam Callback A callable type that processes the packet data.
      * @param callback Callable to handle processed packets.
+     * @return The highest sequence number delivered, or mLastPacketIdx if nothing was held.
      */
     template <typename Callback>
-    void restartBuffering(Callback& callback, SeqType currPacketIdx)
+    SeqType drainBufferInOrder(Callback& callback)
     {
         // Process as many in-order buffered packets as possible
         processBufferedPackets(callback);
@@ -262,25 +318,40 @@ class BufferedPacketQueue
                 sortedPackets.push_back(it);
             }
 
-            // Sort the vector based on the keys
+            // Sorted by distance from the last delivered packet, not by raw value: a block
+            // that straddles the wrap point (65534, 65535, 0, 1) sorts to 0, 1, 65534, 65535
+            // by value, and would be handed to the parser in that order.
+            const SeqType from = mLastPacketIdx;
             std::sort(
                 sortedPackets.begin(),
                 sortedPackets.end(),
-                [](const auto& a, const auto& b) { return a->first < b->first; });
-
-            // Iterate over the sorted packets and invoke the callback
+                [from](const auto& a, const auto& b) {
+                    return static_cast<SeqType>(a->first - from) < static_cast<SeqType>(b->first - from);
+                });
+
+            // Seeded from a packet that is actually in the buffer rather than from
+            // mLastPacketIdx. RTP starts at a random sequence number, so a VTX that reboots
+            // mid-session can land more than half the sequence space away, where
+            // calculateDistance() reads as negative - seeded from mLastPacketIdx nothing
+            // would ever move and the queue would never resync.
+            SeqType highest = sortedPackets.front()->first;
             for (const auto& it : sortedPackets)
             {
                 const auto& packet = it->second;
                 logDebug("Processing possibly out-of-order buffered packet with Sequence=%u.", it->first);
                 callback(packet.data(), packet.size());
+                if (calculateDistance(highest, it->first) > 0)
+                {
+                    highest = it->first;
+                }
             }
 
             mPackets.clear();
             // Reset the monotonic increase counter
             mMonotonicOutOfOrderIncreaseCount = 0;
+            return highest;
         }
-        mLastPacketIdx = currPacketIdx;
+        return mLastPacketIdx;
     }
 
     /**

app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp

@@ -1,6 +1,7 @@
 #include "BufferedPacketQueue.h"  // the class under test
 #include <gtest/gtest.h>
 #include <cstdint>
+#include <iostream>
 #include <vector>
 
 // ---------- Test fixture ----------------------------------------------------
@@ -10,7 +11,17 @@ class BufferedPacketQueueTest : public ::testing::Test
     BufferedPacketQueue   q;
     std::vector<uint16_t> delivered;
 
-    void SetUp() override { delivered.clear(); }
+    // The queue bounds how long it will hold a packet back, so the tests drive the clock
+    // themselves instead of letting a loaded machine decide whether the bound was hit.
+    QueueTimePoint now{};
+
+    void SetUp() override
+    {
+        delivered.clear();
+        now = QueueTimePoint{};
+    }
+
+    void advance(int ms) { now += std::chrono::milliseconds(ms); }
 
     /* Helper: feed one packet and record what the queue actually delivers. */
     void feed(uint16_t seq)
@@ -23,7 +34,7 @@ class BufferedPacketQueueTest : public ::testing::Test
             delivered.push_back(*(uint16_t*) seq);
         };
 
-        q.processPacket(seq, (uint8_t*) &dummy, 2, cb);
+        q.processPacket(seq, (uint8_t*) &dummy, 2, cb, now);
     }
 };
 
@@ -57,6 +68,124 @@ TEST_F(BufferedPacketQueueTest, ReorderedDeliversInOrder)
     ASSERT_EQ(delivered, expected) << "Overflow flush should deliver the entire block in one shot";
 }
 
+// ---------- A gap that will never be filled --------------------------------
+// The common case on a lossy link: FEC could not recover one packet, and every packet after it
+// is held back waiting for it. Nothing is lost by waiting, but everything behind the gap gets
+// later and later, so the queue has to give up at some point.
+TEST_F(BufferedPacketQueueTest, PermanentGapDoesNotHoldTheStreamForFifteenPackets)
+{
+    for (uint16_t s = 1; s <= 5; ++s) feed(s);
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 3, 4, 5}));
+
+    // 6 never arrives.
+    for (uint16_t s = 7; s <= 11; ++s) feed(s);
+    feed(12);
+
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12}))
+        << "A monotonic run past the gap should release the buffer, not wait for MAX_BUFFER_SIZE";
+}
+
+// The monotonic run is counted in packets, so how much latency it costs depends on the packet
+// rate - which on the audio stream is a fraction of the video one. The age bound is what makes
+// the wait the same on both.
+TEST_F(BufferedPacketQueueTest, StaleBufferIsFlushedOnTimeout)
+{
+    feed(1);
+    // 2 never arrives.
+    feed(3);
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1})) << "3 is held back waiting for 2";
+
+    advance(25);
+    feed(4);
+
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 3, 4}))
+        << "Once the buffer is older than MAX_BUFFER_AGE it has to be released";
+}
+
+// The other side of that bound: a reorder that resolves quickly must still be reordered, not
+// flushed out of sequence.
+TEST_F(BufferedPacketQueueTest, ReorderWithinTimeoutIsStillPutBackInOrder)
+{
+    feed(1);
+    feed(3);
+    advance(5);
+    feed(2);
+
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 3}))
+        << "A packet that arrives late but within MAX_BUFFER_AGE must not be flushed early";
+}
+
+// A large jump is not a reorder - it is a stream that restarted somewhere else. It must not be
+// mistaken for a monotonic run, and the buffer cap has to catch it.
+TEST_F(BufferedPacketQueueTest, LargeJumpFallsBackToTheBufferCap)
+{
+    feed(1);
+    for (uint16_t s = 30000; s < 30000 + MAX_BUFFER_SIZE; ++s) feed(s);
+
+    ASSERT_EQ(delivered.size(), 1u + MAX_BUFFER_SIZE)
+        << "The buffer cap should have released the jumped-to block";
+    ASSERT_EQ(delivered.front(), 1);
+    ASSERT_EQ(delivered.back(), static_cast<uint16_t>(30000 + MAX_BUFFER_SIZE - 1));
+}
+
+// A flush delivers everything it is holding, so the pointer has to end up past all of it.
+// Rewinding it to whichever packet happened to trigger the flush re-creates the stall that
+// was just cleared.
+TEST_F(BufferedPacketQueueTest, FlushAdvancesPastEverythingItDelivered)
+{
+    for (uint16_t s = 1; s <= 5; ++s) feed(s);
+
+    // 6 never arrives, and 11 is missing from the run so the flush is triggered by 10 while
+    // 12 is already buffered behind it.
+    feed(7);
+    feed(8);
+    feed(9);
+    feed(12);
+    feed(10);
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 3, 4, 5, 7, 8, 9, 10, 12}));
+
+    feed(13);
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 3, 4, 5, 7, 8, 9, 10, 12, 13}))
+        << "13 follows the highest packet already delivered and must not be held back";
+}
+
+// RTP starts at a random sequence number, so a VTX rebooting mid-session shows up as a jump
+// that can be more than half the sequence space - where a signed 16 bit distance reads as
+// "behind us" and the queue would never move again.
+TEST_F(BufferedPacketQueueTest, StreamRestartFarAheadResyncs)
+{
+    feed(1);
+    feed(2);
+
+    feed(40000);
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2})) << "40000 is held back at first";
+
+    advance(25);
+    feed(40001);
+    feed(40002);
+
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{1, 2, 40000, 40001, 40002}))
+        << "after the age bound releases 40000 the queue has to follow the new base";
+}
+
+// The buffer is a hash map, so the flush has to sort it - and sorting by raw value puts a
+// block that straddles the wrap point in the wrong order.
+TEST_F(BufferedPacketQueueTest, FlushAcrossTheWrapDeliversInOrder)
+{
+    feed(65532);
+
+    // 65533 never arrives; the rest of the run crosses zero.
+    feed(65534);
+    feed(65535);
+    feed(0);
+    feed(1);
+    feed(2);
+    feed(3);
+
+    ASSERT_EQ(delivered, (std::vector<uint16_t>{65532, 65534, 65535, 0, 1, 2, 3}))
+        << "sorted by value this comes out as 0, 1, 2, 65534, 65535";
+}
+
 // ---------- gtest boilerplate main -----------------------------------------
 int main(int argc, char** argv)
 {

app/videonative/src/main/cpp/tests/CMakeLists.txt

@@ -25,7 +25,7 @@ enable_testing()
 
 # ---------- Test executable --------------------------------------------------
 add_executable(queue_test
-    BufferedPacketqueue_test.cpp
+    BufferedPacketQueue_test.cpp
 )
 
 target_include_directories(queue_test PUBLIC