fix(liveupdates): add subscription to backlog if failed (#6763)

Reviewed-by: pajlada <rasmus.karlsson@pajlada.com>

CHANGELOG.md

@@ -65,7 +65,7 @@
 - Dev: Added Qt keyword and warning flags project wide. (#6520)
 - Dev: Added REUSE license checking. (#6659)
 - Def: Fixed compilation error in tests with Clang 21. (#6519)
-- Dev: The 7TV and BTTV liveupdates now use Boost.Beast's WebSockets. (#6308)
+- Dev: The 7TV and BTTV liveupdates now use Boost.Beast's WebSockets. (#6308, #6763)
 - Dev: Fixed compilation warnings on clang-cl. (#6528)
 - Dev: Fixed compilation error in tests with Clang 21. (#6519)
 - Dev: Use CMake's `FetchContent` for RapidJSON, PajladaSignals, PajladaSerialize, and PajladaSettings. (#6560, #6567, #6569)

src/providers/liveupdates/BasicPubSubManager.hpp

@@ -107,6 +107,8 @@ public:
 protected:
     void unsubscribe(const Subscription &subscription)
     {
+        assertInGuiThread();
+
         for (auto &client : this->clients_)
         {
             if (client.second->unsubscribe(subscription))
@@ -118,6 +120,8 @@ protected:
 
     void subscribe(const Subscription &subscription)
     {
+        assertInGuiThread();
+
         if (this->trySubscribe(subscription))
         {
             return;
@@ -141,6 +145,8 @@ private:
 
     void onConnectionOpen(size_t id)
     {
+        assertInGuiThread();
+
         DebugCount::increase("LiveUpdates connections");
         this->addingClient_ = false;
         this->diag.connectionsOpened.fetch_add(1, std::memory_order_acq_rel);
@@ -164,8 +170,8 @@ private:
             {
                 qCDebug(chatterinoLiveupdates)
                     << "Failed to subscribe to" << last << "on new client.";
-                // TODO: should we try to add a new client here?
+                this->pendingSubscriptions_.emplace_back(std::move(last));
-                return;
+                break;
             }
             DebugCount::decrease("LiveUpdates subscription backlog");
             pendingSubsToTake--;
@@ -173,12 +179,17 @@ private:
 
         if (!this->pendingSubscriptions_.empty())
         {
+            qCDebug(chatterinoLiveupdates)
+                << "Adding another client for "
+                << this->pendingSubscriptions_.size() << "subs";
             this->addClient();
         }
     }
 
     void onConnectionClose(size_t id)
     {
+        assertInGuiThread();
+
         this->addingClient_ = false;
 
         auto it = this->clients_.find(id);
@@ -216,10 +227,14 @@ private:
         {
             qCWarning(chatterinoLiveupdates)
                 << "Retrying after" << id << "failed";
+            auto nSubs = subs.size();
+            DebugCount::increase("LiveUpdates subscription backlog",
+                                 static_cast<int64_t>(nSubs));
             this->pendingSubscriptions_.insert(
                 this->pendingSubscriptions_.end(),
                 std::make_move_iterator(subs.begin()),
                 std::make_move_iterator(subs.end()));
+
             QTimer::singleShot(this->connectBackoff_.next(), this, [this] {
                 this->addClient();
             });
@@ -234,6 +249,8 @@ private:
 
     void addClient()
     {
+        assertInGuiThread();
+
         if (this->addingClient_ || !this->pool_)
         {
             return;

tests/src/BasicPubSub.cpp

@@ -80,8 +80,9 @@ class MyManager;
 class MyClient : public BasicPubSubClient<DummySubscription, MyClient>
 {
 public:
-    MyClient(MyManager &manager)
+    MyClient(MyManager &manager, size_t limit)
-        : manager(manager)
+        : BasicPubSubClient(limit)
+        , manager(manager)
     {
     }
 
@@ -94,8 +95,9 @@ private:
 class MyManager : public BasicPubSubManager<MyManager, MyClient>
 {
 public:
-    MyManager(QString host)
+    MyManager(QString host, size_t limit = 100)
         : BasicPubSubManager(std::move(host), "Test")
+        , limit_(limit)
     {
     }
 
@@ -126,12 +128,13 @@ public:
 
     std::shared_ptr<MyClient> makeClient()
     {
-        return std::make_shared<MyClient>(*this);
+        return std::make_shared<MyClient>(*this, this->limit_);
     }
 
 private:
     std::mutex messageMtx_;
     std::deque<QString> messageQueue_;
+    size_t limit_;
 
     friend MyClient;
 };
@@ -177,3 +180,46 @@ TEST(BasicPubSub, SubscriptionCycle)
     ASSERT_EQ(manager.diag.connectionsFailed, 0);
     ASSERT_EQ(manager.messagesReceived, 2);
 }
+
+TEST(BasicPubSub, SubLimits)
+{
+    mock::BaseApplication app;
+    const QString host("wss://127.0.0.1:9050/liveupdates/sub-unsub");
+    MyManager manager(host, 1);
+    manager.sub({.type = 1, .condition = "foo"});
+    manager.sub({.type = 2, .condition = "foo"});
+    manager.sub({.type = 3, .condition = "foo"});
+    manager.sub({.type = 4, .condition = "foo"});
+    manager.sub({.type = 5, .condition = "foo"});
+    QTest::qWait(500);
+    ASSERT_EQ(manager.diag.connectionsOpened, 5);
+    ASSERT_EQ(manager.diag.connectionsClosed, 0);
+    ASSERT_EQ(manager.diag.connectionsFailed, 0);
+    ASSERT_EQ(manager.messagesReceived, 5);
+
+    // The messages come from multiple connections, so they're not necessarily
+    // in order.
+    std::vector<QString> messages;
+    for (size_t i = 0; i < 5; i++)
+    {
+        auto msg = manager.popMessage();
+        ASSERT_TRUE(msg.has_value());
+        messages.push_back(*std::move(msg));
+    }
+    std::ranges::sort(messages);
+    ASSERT_EQ(messages[0], QString("ack-sub-1-foo"));
+    ASSERT_EQ(messages[1], QString("ack-sub-2-foo"));
+    ASSERT_EQ(messages[2], QString("ack-sub-3-foo"));
+    ASSERT_EQ(messages[3], QString("ack-sub-4-foo"));
+    ASSERT_EQ(messages[4], QString("ack-sub-5-foo"));
+
+    manager.stop();
+    // after exactly one event loop iteration, we should see updated counters
+    QCoreApplication::processEvents(QEventLoop::AllEvents);
+    QCoreApplication::sendPostedEvents(nullptr, QEvent::DeferredDelete);
+
+    ASSERT_EQ(manager.diag.connectionsOpened, 5);
+    ASSERT_EQ(manager.diag.connectionsClosed, 5);
+    ASSERT_EQ(manager.diag.connectionsFailed, 0);
+    ASSERT_EQ(manager.messagesReceived, 5);
+}