Rate limit outgoing JOIN messages (#3115)

Co-authored-by: Rasmus Karlsson <rasmus.karlsson@pajlada.com>
Co-authored-by: Tal Neoran <talneoran@gmail.com>

src/CMakeLists.txt

@@ -292,6 +292,8 @@ set(SOURCE_FILES
         util/NuulsUploader.hpp
         util/RapidjsonHelpers.cpp
         util/RapidjsonHelpers.hpp
+        util/RatelimitBucket.cpp
+        util/RatelimitBucket.hpp
         util/SplitCommand.cpp
         util/SplitCommand.hpp
         util/StreamLink.cpp

src/providers/irc/AbstractIrcServer.cpp

@@ -15,6 +15,10 @@ const int RECONNECT_BASE_INTERVAL = 2000;
 // 60 falloff counter means it will try to reconnect at most every 60*2 seconds
 const int MAX_FALLOFF_COUNTER = 60;
 
+// Ratelimits for joinBucket_
+const int JOIN_RATELIMIT_BUDGET = 18;
+const int JOIN_RATELIMIT_COOLDOWN = 10500;
+
 AbstractIrcServer::AbstractIrcServer()
 {
     // Initialize the connections
@@ -23,6 +27,17 @@ AbstractIrcServer::AbstractIrcServer()
     this->writeConnection_->moveToThread(
         QCoreApplication::instance()->thread());
 
+    // Apply a leaky bucket rate limiting to JOIN messages
+    auto actuallyJoin = [&](QString message) {
+        if (!this->channels.contains(message))
+        {
+            return;
+        }
+        this->readConnection_->sendRaw("JOIN #" + message);
+    };
+    this->joinBucket_.reset(new RatelimitBucket(
+        JOIN_RATELIMIT_BUDGET, JOIN_RATELIMIT_COOLDOWN, actuallyJoin, this));
+
     QObject::connect(this->writeConnection_.get(),
                      &Communi::IrcConnection::messageReceived, this,
                      [this](auto msg) {
@@ -224,7 +239,7 @@ ChannelPtr AbstractIrcServer::getOrAddChannel(const QString &dirtyChannelName)
         {
             if (this->readConnection_->isConnected())
             {
-                this->readConnection_->sendRaw("JOIN #" + channelName);
+                this->joinBucket_->send(channelName);
             }
         }
     }
@@ -284,7 +299,7 @@ void AbstractIrcServer::onReadConnected(IrcConnection *connection)
     {
         if (auto channel = weak.lock())
         {
-            connection->sendRaw("JOIN #" + channel->getName());
+            this->joinBucket_->send(channel->getName());
         }
     }
 

src/providers/irc/AbstractIrcServer.hpp

@@ -8,6 +8,7 @@
 
 #include "common/Common.hpp"
 #include "providers/irc/IrcConnection2.hpp"
+#include "util/RatelimitBucket.hpp"
 
 namespace chatterino {
 
@@ -88,6 +89,10 @@ private:
     QObjectPtr<IrcConnection> writeConnection_ = nullptr;
     QObjectPtr<IrcConnection> readConnection_ = nullptr;
 
+    // Our rate limiting bucket for the Twitch join rate limits
+    // https://dev.twitch.tv/docs/irc/guide#rate-limits
+    QObjectPtr<RatelimitBucket> joinBucket_;
+
     QTimer reconnectTimer_;
     int falloffCounter_ = 1;
 

src/util/RatelimitBucket.cpp

@@ -0,0 +1,45 @@
+#include "RatelimitBucket.hpp"
+
+#include <QTimer>
+
+namespace chatterino {
+
+RatelimitBucket::RatelimitBucket(int budget, int cooldown,
+                                 std::function<void(QString)> callback,
+                                 QObject *parent)
+    : QObject(parent)
+    , budget_(budget)
+    , cooldown_(cooldown)
+    , callback_(callback)
+{
+}
+
+void RatelimitBucket::send(QString channel)
+{
+    this->queue_.append(channel);
+
+    if (this->budget_ > 0)
+    {
+        this->handleOne();
+    }
+}
+
+void RatelimitBucket::handleOne()
+{
+    if (queue_.isEmpty())
+    {
+        return;
+    }
+
+    auto item = queue_.takeFirst();
+
+    this->budget_--;
+    callback_(item);
+
+    QTimer::singleShot(cooldown_, this, [this] {
+        this->budget_++;
+        this->handleOne();
+    });
+}
+
+}  // namespace chatterino

src/util/RatelimitBucket.hpp

@@ -0,0 +1,40 @@
+#pragma once
+
+#include <QList>
+#include <QObject>
+#include <QString>
+
+namespace chatterino {
+
+class RatelimitBucket : public QObject
+{
+public:
+    RatelimitBucket(int budget, int cooldown,
+                    std::function<void(QString)> callback, QObject *parent);
+
+    void send(QString channel);
+
+private:
+    /**
+     * @brief budget_ denotes the amount of calls that can be handled before we need to wait for the cooldown
+     **/
+    int budget_;
+
+    /**
+     * @brief This is the amount of time in milliseconds it takes for one used up budget to be put back into the bucket for use elsewhere
+     **/
+    const int cooldown_;
+
+    std::function<void(QString)> callback_;
+    QList<QString> queue_;
+
+    /**
+     * @brief Run the callback on one entry in the queue.
+     *
+     * This will start a timer that runs after cooldown_ milliseconds that
+     * gives back one "token" to the bucket and calls handleOne again.
+     **/
+    void handleOne();
+};
+
+}  // namespace chatterino