From 0235451c5603d5088ef9d8f9d8545ffa3fb2563a Mon Sep 17 00:00:00 2001 From: nerix Date: Fri, 7 Mar 2025 19:09:10 +0100 Subject: [PATCH] refactor(eventsub): use backoff over retry count (#6036) --- CHANGELOG.md | 2 +- src/providers/twitch/eventsub/Controller.cpp | 105 +++++++------------ src/providers/twitch/eventsub/Controller.hpp | 10 +- 3 files changed, 43 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dd19617f..68a811a1 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/src/providers/twitch/eventsub/Controller.cpp b/src/providers/twitch/eventsub/Controller.cpp index b5958711..12a192a2 100644 --- a/src/providers/twitch/eventsub/Controller.cpp +++ b/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( - this->ioContext, [this, request]() mutable { - this->retrySubscription( - request, boost::posix_time::seconds(2), 5); - }); + boost::asio::post(this->ioContext, [this, request] { + this->retrySubscription(request); + }); 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) - { - qCWarning(LOG) << "We have" << openButNotReadyConnections - << "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); + // There should only ever be 0 or 1 + qCWarning(LOG) << "We have" << openButNotReadyConnections + << "open but no ready connections"; } + + this->retrySubscription(request); } std::optional> Controller::getViableConnection( @@ -473,16 +461,12 @@ void Controller::registerConnection(std::weak_ptr &&connection) this->connections.emplace_back(std::move(connection)); } -void Controller::retrySubscription(const SubscriptionRequest &request, - boost::posix_time::time_duration delay, - int32_t maxAttempts) +void Controller::retrySubscription(const SubscriptionRequest &request) { 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 || - subscription.retryAttempts > maxAttempts) + if (subscription.state != Subscription::State::Retrying) { - assert((subscription.state == Subscription::State::Subscribing || - 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(subscription.state); - - subscription.retryAttempts = maxAttempts; - } - else - { - qCDebug(LOG) << "Retrying subscription" << request << " - attempt" - << subscription.retryAttempts << "of" << maxAttempts - << "from state" - << static_cast(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; + subscription.state = Subscription::State::Retrying; } - qCDebug(LOG) << "Set state to retrying" << request; - subscription.state = Subscription::State::Retrying; - - int attemptNumber = subscription.retryAttempts; + // we don't need a strong RNG here + // NOLINTNEXTLINE(cert-*) + std::chrono::milliseconds jitter{std::rand() % 256}; auto retryTimer = - std::make_unique(this->ioContext); - retryTimer->expires_from_now(delay); - retryTimer->async_wait([this, request, attemptNumber](const auto &ec) { + std::make_unique(this->ioContext); + retryTimer->expires_after(subscription.backoff.next() + jitter); + 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() diff --git a/src/providers/twitch/eventsub/Controller.hpp b/src/providers/twitch/eventsub/Controller.hpp index 2916cbc4..e484996e 100644 --- a/src/providers/twitch/eventsub/Controller.hpp +++ b/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 @@ -78,9 +79,7 @@ private: std::unique_ptr listener); void registerConnection(std::weak_ptr &&connection); - void retrySubscription(const SubscriptionRequest &request, - boost::posix_time::time_duration delay, - int32_t maxAttempts); + void retrySubscription(const SubscriptionRequest &request); void markRequestSubscribed(const SubscriptionRequest &request, std::weak_ptr connection, @@ -138,8 +137,9 @@ private: QString subscriptionID; /// The timer, if any, for retrying the subscription creation - std::unique_ptr retryTimer; - int32_t retryAttempts = 0; + std::unique_ptr retryTimer; + // 500ms to 16s backoff + ExponentialBackoff<6> backoff{std::chrono::milliseconds{500}}; }; std::mutex subscriptionsMutex;