refactor(eventsub): use backoff over retry count (#6036)

CHANGELOG.md

@@ -37,7 +37,7 @@
 - Bugfix: Fixed color input thinking blue is also red. (#5982)
 - Bugfix: Fixed an issue where commands would sometimes reset if Chatterino was improperly shut down. (#6011)
 - Dev: Subscriptions to PubSub channel points redemption topics now use no auth token, making it continue to work during PubSub shutdown. (#5947)
-- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028, #6035)
+- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028, #6035, #6036)
 - Dev: Remove unneeded platform specifier for toasts. (#5914)
 - Dev: Highlight checks now use non-capturing groups for the boundaries. (#5784)
 - Dev: Removed unused PubSub whisper code. (#5898)

src/providers/twitch/eventsub/Controller.cpp

@@ -191,6 +191,7 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
             case Subscription::State::Subscribing:
             case Subscription::State::Retrying:
             case Subscription::State::Subscribed:
+            case Subscription::State::Unsubscribing:
                 break;
         }
 
@@ -200,7 +201,7 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
             subscription.state = Subscription::State::Subscribing;
 
             // Ensure retries can work as expected since this is a fresh subscription
-            subscription.retryAttempts = 0;
+            subscription.backoff.reset();
 
             assert(subscription.retryTimer == nullptr &&
                    "A new subscription should not have a retry timer created");
@@ -260,7 +261,7 @@ void Controller::reconnectConnection(
             {
                 qCDebug(chatterinoTwitchEventSub) << "Resetting" << it->first;
                 it->second.connection = {};
-                it->second.retryAttempts = 0;
+                it->second.backoff.reset();
                 it->second.state = Subscription::State::Subscribing;
             }
         }
@@ -335,13 +336,6 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
 
                     case Error::Forbidden:
                         qCDebug(LOG) << "Forbidden" << errorString << request;
-                        boost::asio::post(this->ioContext, [this, request]() {
-                            qCDebug(LOG)
-                                << "Calling retrySubscription from Forbidden"
-                                << request;
-                            this->retrySubscription(
-                                request, boost::posix_time::seconds(2), 2);
-                        });
                         return;
 
                     case Error::Conflict:
@@ -358,11 +352,9 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
                         qCWarning(LOG) << "Unhandled error, retrying "
                                           "subscription"
                                        << errorString << request;
-                        boost::asio::post(
+                        boost::asio::post(this->ioContext, [this, request] {
-                            this->ioContext, [this, request]() mutable {
+                            this->retrySubscription(request);
-                                this->retrySubscription(
+                        });
-                                    request, boost::posix_time::seconds(2), 5);
-                            });
                         return;
                 }
 
@@ -376,19 +368,15 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
     {
         // No connection was available to handle this subscription request, create a new connection
         this->createConnection();
-        this->retrySubscription(request, boost::posix_time::millisec(500), 10);
     }
-    else
+    else if (openButNotReadyConnections > 1)
     {
-        if (openButNotReadyConnections > 1)
+        // There should only ever be 0 or 1
-        {
+        qCWarning(LOG) << "We have" << openButNotReadyConnections
-            qCWarning(LOG) << "We have" << openButNotReadyConnections
+                       << "open but no ready connections";
-                           << "open but no ready connections, hmmm";
-        }
-
-        // At least one connection is open, but it has not gotten the welcome message yet
-        this->retrySubscription(request, boost::posix_time::millisec(250), 10);
     }
+
+    this->retrySubscription(request);
 }
 
 std::optional<std::shared_ptr<lib::Session>> Controller::getViableConnection(
@@ -473,16 +461,12 @@ void Controller::registerConnection(std::weak_ptr<lib::Session> &&connection)
     this->connections.emplace_back(std::move(connection));
 }
 
-void Controller::retrySubscription(const SubscriptionRequest &request,
+void Controller::retrySubscription(const SubscriptionRequest &request)
-                                   boost::posix_time::time_duration delay,
-                                   int32_t maxAttempts)
 {
     std::lock_guard lock(this->subscriptionsMutex);
 
     auto &subscription = this->subscriptions[request];
 
-    assert(subscription.retryAttempts >= 0);
-
     if (subscription.refCount == 0)
     {
         qCDebug(LOG) << "No one is interested in this subscription anymore, "
@@ -494,51 +478,23 @@ void Controller::retrySubscription(const SubscriptionRequest &request,
         return;
     }
 
-    if (subscription.retryAttempts == 0 ||
+    if (subscription.state != Subscription::State::Retrying)
-        subscription.retryAttempts > maxAttempts)
     {
-        assert((subscription.state == Subscription::State::Subscribing ||
+        qCDebug(LOG) << "Set state to retrying" << request;
-                subscription.state == Subscription::State::Retrying) &&
+        subscription.state = Subscription::State::Retrying;
-               "new retry must start from Subscribing or Retrying state");
-
-        qCDebug(LOG) << "New retry for subscription" << request
-                     << " - max attempts" << maxAttempts << "from state"
-                     << static_cast<uint8_t>(subscription.state);
-
-        subscription.retryAttempts = maxAttempts;
-    }
-    else
-    {
-        qCDebug(LOG) << "Retrying subscription" << request << " - attempt"
-                     << subscription.retryAttempts << "of" << maxAttempts
-                     << "from state"
-                     << static_cast<uint8_t>(subscription.state);
-        assert(subscription.state == Subscription::State::Retrying &&
-               "retries must come from Retrying state");
-
-        subscription.retryAttempts -= 1;
-
-        if (subscription.retryAttempts == 0)
-        {
-            qCWarning(LOG) << "Reached max amount of retries for" << request;
-            qCDebug(LOG) << "Set state to failed" << request;
-            subscription.state = Subscription::State::Failed;
-            return;
-        }
     }
 
-    qCDebug(LOG) << "Set state to retrying" << request;
+    // we don't need a strong RNG here
-    subscription.state = Subscription::State::Retrying;
+    // NOLINTNEXTLINE(cert-*)
-
+    std::chrono::milliseconds jitter{std::rand() % 256};
-    int attemptNumber = subscription.retryAttempts;
 
     auto retryTimer =
-        std::make_unique<boost::asio::deadline_timer>(this->ioContext);
+        std::make_unique<boost::asio::system_timer>(this->ioContext);
-    retryTimer->expires_from_now(delay);
+    retryTimer->expires_after(subscription.backoff.next() + jitter);
-    retryTimer->async_wait([this, request, attemptNumber](const auto &ec) {
+    retryTimer->async_wait([this, request](const auto &ec) {
         if (!ec)
         {
-            qCDebug(LOG) << "Firing retry" << request << attemptNumber;
+            qCDebug(LOG) << "Firing retry" << request;
             // The timer passed naturally
             this->subscribe(request, true);
         }
@@ -596,6 +552,7 @@ void Controller::markRequestSubscribed(const SubscriptionRequest &request,
     subscription.subscriptionID = subscriptionID;
     qCDebug(LOG) << "Set state to subscribed" << request;
     subscription.state = Subscription::State::Subscribed;
+    subscription.backoff.reset();
 }
 
 void Controller::markRequestFailed(const SubscriptionRequest &request)
@@ -635,7 +592,7 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request)
 
     qCDebug(LOG) << "Set state to unsubscribed" << request;
     subscription.state = Subscription::State::Unsubscribed;
-    subscription.retryAttempts = 0;
+    subscription.backoff.reset();
     auto conn = subscription.connection.lock();
     if (conn)
     {
@@ -646,6 +603,18 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request)
         }
     }
     subscription.connection = {};
+
+    if (subscription.refCount == 0)
+    {
+        // we could remove the subscription here
+        return;
+    }
+
+    // someone subscribed in the meantime
+    subscription.state = Subscription::State::Subscribing;
+    boost::asio::post(this->ioContext, [this, request] {
+        this->subscribe(request, false);
+    });
 }
 
 void Controller::clearConnections()

src/providers/twitch/eventsub/Controller.hpp

@@ -3,6 +3,7 @@
 #include "providers/twitch/eventsub/SubscriptionHandle.hpp"
 #include "providers/twitch/eventsub/SubscriptionRequest.hpp"
 #include "twitch-eventsub-ws/session.hpp"
+#include "util/ExponentialBackoff.hpp"
 #include "util/ThreadGuard.hpp"
 
 #include <boost/asio/executor_work_guard.hpp>
@@ -78,9 +79,7 @@ private:
                           std::unique_ptr<lib::Listener> listener);
     void registerConnection(std::weak_ptr<lib::Session> &&connection);
 
-    void retrySubscription(const SubscriptionRequest &request,
+    void retrySubscription(const SubscriptionRequest &request);
-                           boost::posix_time::time_duration delay,
-                           int32_t maxAttempts);
 
     void markRequestSubscribed(const SubscriptionRequest &request,
                                std::weak_ptr<lib::Session> connection,
@@ -138,8 +137,9 @@ private:
         QString subscriptionID;
 
         /// The timer, if any, for retrying the subscription creation
-        std::unique_ptr<boost::asio::deadline_timer> retryTimer;
+        std::unique_ptr<boost::asio::system_timer> retryTimer;
-        int32_t retryAttempts = 0;
+        // 500ms to 16s backoff
+        ExponentialBackoff<6> backoff{std::chrono::milliseconds{500}};
     };
 
     std::mutex subscriptionsMutex;