diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b2bdf14..20bc483c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,7 +32,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) +- 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) - 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/Application.cpp b/src/Application.cpp index 93442df1..2838bb29 100644 --- a/src/Application.cpp +++ b/src/Application.cpp @@ -165,6 +165,7 @@ Application::Application(Settings &_settings, const Paths &paths, , logging(new Logging(_settings)) , emotes(new Emotes) , accounts(new AccountController) + , eventSub(makeEventSubController(_settings)) , hotkeys(new HotkeyController) , windows(new WindowManager(paths, _settings, *this->themes, *this->fonts)) , toasts(new Toasts) @@ -193,7 +194,6 @@ Application::Application(Settings &_settings, const Paths &paths, , streamerMode(new StreamerMode) , twitchUsers(new TwitchUsers) , pronouns(new pronouns::Pronouns) - , eventSub(makeEventSubController(_settings)) #ifdef CHATTERINO_HAVE_PLUGINS , plugins(new PluginController(paths)) #endif @@ -203,6 +203,8 @@ Application::Application(Settings &_settings, const Paths &paths, Application::~Application() { + this->eventSub->setQuitting(); + // we do this early to ensure getApp isn't used in any dtors INSTANCE = nullptr; } diff --git a/src/Application.hpp b/src/Application.hpp index 334485f4..66aa03c4 100644 --- a/src/Application.hpp +++ b/src/Application.hpp @@ -151,6 +151,7 @@ private: const std::unique_ptr logging; std::unique_ptr emotes; std::unique_ptr accounts; + std::unique_ptr eventSub; std::unique_ptr hotkeys; std::unique_ptr windows; std::unique_ptr toasts; @@ -178,7 +179,6 @@ private: std::unique_ptr streamerMode; std::unique_ptr twitchUsers; std::unique_ptr pronouns; - std::unique_ptr eventSub; #ifdef CHATTERINO_HAVE_PLUGINS std::unique_ptr plugins; #endif diff --git a/src/providers/twitch/eventsub/Controller.cpp b/src/providers/twitch/eventsub/Controller.cpp index bb853b99..f7c141c9 100644 --- a/src/providers/twitch/eventsub/Controller.cpp +++ b/src/providers/twitch/eventsub/Controller.cpp @@ -59,6 +59,9 @@ Controller::Controller() Controller::~Controller() { + assert(this->quitting && "Application should call setQuitting() before " + "destroying the controller"); + qCInfo(LOG) << "Controller dtor start"; for (const auto &weakConnection : this->connections) @@ -90,54 +93,121 @@ Controller::~Controller() void Controller::removeRef(const SubscriptionRequest &request) { + if (this->quitting) + { + // We're quitting - we don't care to unsub + return; + } + std::lock_guard lock(this->subscriptionsMutex); assert(this->subscriptions.contains(request)); auto &subscription = this->subscriptions[request]; subscription.refCount--; - qCInfo(LOG) << "Removed ref for" << request << subscription.refCount; + assert(subscription.refCount >= 0); + if (subscription.refCount == 0) + { + qCDebug(LOG) << "Removed last ref for" << request; + } + else + { + qCDebug(LOG) << "Removed ref for" << request << "(" + << subscription.refCount << "remaining)"; + } if (subscription.refCount <= 0) { + // No longer interested in this topic, ensure we don't have a retry in flight + subscription.retryTimer.reset(); if (subscription.subscriptionID.isEmpty()) { - qCWarning(LOG) << "Refcount fell to zero for" << request - << "but we had no subscription ID attached - was a " - "successful subscription never made?"; + qCDebug(LOG) + << "Refcount fell to zero for" << request + << "but we had no subscription ID attached - a " + "successful subscription was never made. From state " + << static_cast(subscription.state); return; } - qCInfo(LOG) << "Unsubscribing from" << request; + qCDebug(LOG) << "Unsubscribing from" << request; + subscription.state = Subscription::State::Unsubscribing; + getHelix()->deleteEventSubSubscription( subscription.subscriptionID, - [request] { - qCInfo(LOG) << "Successfully unsubscribed from" << request; + [this, request] { + qCDebug(LOG) << "Successfully unsubscribed from" << request; + this->markRequestUnsubscribed(request); }, - [request](const auto &errorMessage) { - qCInfo(LOG) + [this, request](const auto &errorMessage) { + qCWarning(LOG) << "An error occurred while attempting to unsubscribe from" << request << errorMessage; + this->markRequestUnsubscribed(request); }); subscription.subscriptionID.clear(); } } +void Controller::setQuitting() +{ + this->quitting = true; +} + SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request) { + assert(!this->quitting && + "Subscribe cannot be called while we are quitting"); + bool needToSubscribe = false; { + // TODO: Investigate if this scope can be done in boost::asio::post instead + // Basically, if the SubscriptionHandle can be built & returned entirely without waiting for the subscriptionsMutex lock std::lock_guard lock(this->subscriptionsMutex); auto &subscription = this->subscriptions[request]; - if (subscription.refCount == 0) + + assert(subscription.refCount >= 0); + + switch (subscription.state) { - needToSubscribe = true; + case Subscription::State::Unsubscribed: + needToSubscribe = true; + assert(subscription.refCount == 0 && + "An unsubscribed subscription should have 0 references"); + break; + + case Subscription::State::Failed: + qCDebug(LOG) + << "New subscription attempt to previously-failed request" + << request; + needToSubscribe = true; + break; + + case Subscription::State::Subscribing: + case Subscription::State::Retrying: + case Subscription::State::Subscribed: + break; } + + if (needToSubscribe) + { + qCDebug(LOG) << "Set state to subscribing" << request; + subscription.state = Subscription::State::Subscribing; + + // Ensure retries can work as expected since this is a fresh subscription + subscription.retryAttempts = 0; + + assert(subscription.retryTimer == nullptr && + "A new subscription should not have a retry timer created"); + } + subscription.refCount++; - qCInfo(LOG) << "Added ref for" << request << subscription.refCount; + qCDebug(LOG) << "Added ref for" << request << subscription.refCount + << needToSubscribe + << "state:" << static_cast(subscription.state); } auto handle = std::make_unique(request); @@ -154,133 +224,153 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request) void Controller::subscribe(const SubscriptionRequest &request, bool isRetry) { - qCInfo(LOG) << "Subscribe request for" << request.subscriptionType; - boost::asio::post(this->ioContext, [this, request, isRetry] { - // 1. Flush dead connections (maybe this should not be done here) - // TODO: implement + // 1. Flush dead connections (maybe this should not be done here) + // TODO: implement + { + std::lock_guard lock(this->subscriptionsMutex); + auto &subscription = this->subscriptions[request]; + if (isRetry) { - std::lock_guard lock(this->subscriptionsMutex); - auto &subscription = this->subscriptions[request]; - if (isRetry) - { - qCInfo(LOG) << "Removing subscription from queued list"; + qCDebug(LOG) << "Retry subscribe request for" << request; - subscription.retryTimer.reset(); - } - else if (subscription.retryTimer) - { - qCWarning(LOG) - << "We already have a queued subscription for this, " - "let's chill :)"; - return; - } - } + assert(subscription.retryTimer != nullptr); - uint32_t openButNotReadyConnections = 0; - - // 2. Check if any currently open connection can handle this subscription - for (const auto &weakConnection : this->connections) - { - auto connection = weakConnection.lock(); - if (!connection) - { - // TODO: remove it here? - continue; - } - - auto *listener = - dynamic_cast(connection->getListener()); - - if (listener == nullptr) - { - // something really goofy is going on - qCWarning(LOG) << "listener was not the correct type"; - continue; - } - - if (listener->getSessionID().isEmpty()) - { - // This connection is open but it's not ready (i.e. no welcome has been posted yet) - ++openButNotReadyConnections; - continue; - } - - // TODO: Check if this listener has room for another subscription - - // TODO: Don't hardcode the subscription version - getHelix()->createEventSubSubscription( - request, listener->getSessionID(), - [this, request, connection, - weakConnection{weakConnection}](const auto &res) { - qCInfo(LOG) << "success" << res; - this->markRequestSubscribed(request, weakConnection, - res.subscriptionID); - }, - [this, request](const auto &error, const auto &errorString) { - using Error = HelixCreateEventSubSubscriptionError; - switch (error) - { - case Error::BadRequest: - qCWarning(LOG) << "BadRequest" << errorString; - break; - - case Error::Unauthorized: - qCWarning(LOG) << "Unauthorized" << errorString; - break; - - case Error::Forbidden: - qCWarning(LOG) << "Forbidden" << errorString; - boost::asio::post(this->ioContext, [this, request] { - this->retrySubscription( - request, boost::posix_time::seconds(2), 5); - }); - break; - - case Error::Conflict: - // This session ID is already subscribed to this request, some logic of ours is wrong - qCWarning(LOG) << "Conflict" << errorString; - break; - - case Error::Ratelimited: - qCWarning(LOG) << "Ratelimited" << errorString; - break; - - case Error::Forwarded: - default: - qCWarning(LOG) - << "Unhandled error, retrying subscription" - << errorString; - boost::asio::post(this->ioContext, [this, request] { - this->retrySubscription( - request, boost::posix_time::seconds(2), 5); - }); - break; - } - }); - return; - } - - if (openButNotReadyConnections == 0) - { - // No connection was available to handle this subscription request, create a new connection - this->createConnection(); - this->retrySubscription(request, boost::posix_time::millisec(500), - 10); + subscription.retryTimer.reset(); } else { - 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); + qCDebug(LOG) << "New subscribe request for" << request; } - }); + + assert(subscription.retryTimer == nullptr); + } + + uint32_t openButNotReadyConnections = 0; + + // 2. Check if any currently open connection can handle this subscription + auto viableConnection = + this->getViableConnection(openButNotReadyConnections); + + if (viableConnection.has_value()) + { + const auto &connection = *viableConnection; + auto *listener = dynamic_cast(connection->getListener()); + + assert(listener != nullptr && "Something goofy has gone wrong, Session " + "listener must be our Connection type"); + + qCDebug(LOG) << "Make helix request for" << request; + getHelix()->createEventSubSubscription( + request, listener->getSessionID(), + [this, request, connection, + weakConnection{std::weak_ptr(connection)}]( + const auto &res) { + qCDebug(LOG) << "Subscription success" << request; + this->markRequestSubscribed(request, weakConnection, + res.subscriptionID); + }, + [this, request](const auto &error, const auto &errorString) { + using Error = HelixCreateEventSubSubscriptionError; + switch (error) + { + case Error::BadRequest: + qCDebug(LOG) << "Bad request" << errorString << request; + break; + + case Error::Unauthorized: + qCDebug(LOG) + << "Unauthorized" << errorString << request; + break; + + 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: + // This session ID is already subscribed to this request, some logic of ours is wrong + qCWarning(LOG) << "Conflict" << errorString << request; + break; + + case Error::Ratelimited: + qCDebug(LOG) << "Ratelimited" << errorString << request; + break; + + case Error::Forwarded: + default: + 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); + }); + return; + } + + this->markRequestFailed(request); + }); + + return; + } + + if (openButNotReadyConnections == 0) + { + // 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 + { + 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); + } +} + +std::optional> Controller::getViableConnection( + uint32_t &openButNotReadyConnections) +{ + for (const auto &weakConnection : this->connections) + { + auto connection = weakConnection.lock(); + if (!connection) + { + // TODO: remove it here? + continue; + } + + auto *listener = dynamic_cast(connection->getListener()); + + assert(listener != nullptr && "Something goofy has gone wrong, Session " + "listener must be our Connection type"); + + if (listener->getSessionID().isEmpty()) + { + // This connection is open but it's not ready (i.e. no welcome has been posted yet) + ++openButNotReadyConnections; + continue; + } + + // TODO: Check if this listener has room for another subscription + + return connection; + } + + return {}; } void Controller::createConnection() @@ -327,47 +417,149 @@ void Controller::retrySubscription(const SubscriptionRequest &request, { std::lock_guard lock(this->subscriptionsMutex); - auto &connection = this->subscriptions[request]; + auto &subscription = this->subscriptions[request]; - if (connection.retryAttempts <= 0) + assert(subscription.retryAttempts >= 0); + + if (subscription.refCount == 0) { - connection.retryAttempts = maxAttempts; - } - else if (--connection.retryAttempts == 0) - { - qCWarning(LOG) << "Reached max amount of retries for" << request; + qCDebug(LOG) << "No one is interested in this subscription anymore, " + "stop trying" + << request << "from state" + << static_cast(subscription.state); + qCDebug(LOG) << "Set state to unsubscribed" << request; + subscription.state = Subscription::State::Unsubscribed; return; } - qCInfo(LOG) << "Retrying subscription" << request << " - attempt" - << connection.retryAttempts; + if (subscription.retryAttempts == 0 || + subscription.retryAttempts > maxAttempts) + { + 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; + + int attemptNumber = subscription.retryAttempts; + auto retryTimer = std::make_unique(this->ioContext); retryTimer->expires_from_now(delay); - retryTimer->async_wait([this, request](const auto &ec) { + retryTimer->async_wait([this, request, attemptNumber](const auto &ec) { if (!ec) { + qCDebug(LOG) << "Firing retry" << request << attemptNumber; // The timer passed naturally this->subscribe(request, true); } + else + { + qCDebug(LOG) << "Retry timer for" << request << "was cancelled"; + // If we mark the request as unsubscribed here, and we had to actually unsubscribe, + // if an actual unsubscribe happens then it'll go from Unsubscribed -> ACTUALLY unsubscribed + // + // We might still need to update the state here, but we might want some new state for that + // e.g. RetryCancelled or something + // this->markRequestUnsubscribed(request); + } }); - assert(connection.retryTimer == nullptr && + assert(subscription.retryTimer == nullptr && "Timer should not already be set"); - connection.retryTimer = std::move(retryTimer); + subscription.retryTimer = std::move(retryTimer); } void Controller::markRequestSubscribed(const SubscriptionRequest &request, std::weak_ptr connection, const QString &subscriptionID) { + if (this->quitting) + { + return; + } + std::lock_guard lock(this->subscriptionsMutex); auto &subscription = this->subscriptions[request]; + assert((subscription.state == Subscription::State::Subscribing || + subscription.state == Subscription::State::Retrying) && + "A subscription can only be marked subscribed from the Subscribing " + "or Retrying state"); + subscription.connection = std::move(connection); subscription.subscriptionID = subscriptionID; + qCDebug(LOG) << "Set state to subscribed" << request; + subscription.state = Subscription::State::Subscribed; +} + +void Controller::markRequestFailed(const SubscriptionRequest &request) +{ + if (this->quitting) + { + return; + } + + qCWarning(LOG) << "Request" << request << "marked as failed"; + + std::lock_guard lock(this->subscriptionsMutex); + + auto &subscription = this->subscriptions[request]; + + qCDebug(LOG) << "Set state to failed" << request; + subscription.state = Subscription::State::Failed; +} + +void Controller::markRequestUnsubscribed(const SubscriptionRequest &request) +{ + if (this->quitting) + { + return; + } + + std::lock_guard lock(this->subscriptionsMutex); + + auto &subscription = this->subscriptions[request]; + + qCDebug(LOG) << "Request" << request << "marked as unsubscribed from state" + << static_cast(subscription.state); + + assert(subscription.state == Subscription::State::Unsubscribing || + subscription.state == Subscription::State::Retrying); + assert(subscription.retryTimer == nullptr); + + qCDebug(LOG) << "Set state to unsubscribed" << request; + subscription.state = Subscription::State::Unsubscribed; + subscription.retryAttempts = 0; } } // namespace chatterino::eventsub diff --git a/src/providers/twitch/eventsub/Controller.hpp b/src/providers/twitch/eventsub/Controller.hpp index c1c12337..fd6bd817 100644 --- a/src/providers/twitch/eventsub/Controller.hpp +++ b/src/providers/twitch/eventsub/Controller.hpp @@ -11,7 +11,10 @@ #include #include +#include +#include #include +#include #include #include @@ -27,6 +30,11 @@ public: /// Should realistically only be called in the dtor of SubscriptionHandle virtual void removeRef(const SubscriptionRequest &request) = 0; + /// Mark the controller as quitting + /// + /// This lets us simplify some logic with unsubscriptions (i.e. we ignore it instead) + virtual void setQuitting() = 0; + /// Subscribe will make a request to each open connection and ask them to /// add this subscription. /// @@ -46,6 +54,8 @@ public: void removeRef(const SubscriptionRequest &request) override; + void setQuitting() override; + [[nodiscard]] SubscriptionHandle subscribe( const SubscriptionRequest &request) override; @@ -63,6 +73,10 @@ private: std::weak_ptr connection, const QString &subscriptionID); + void markRequestFailed(const SubscriptionRequest &request); + + void markRequestUnsubscribed(const SubscriptionRequest &request); + const std::string userAgent; std::string eventSubHost; @@ -77,7 +91,31 @@ private: std::vector> connections; + [[nodiscard]] std::optional> + getViableConnection(uint32_t &openButNotReadyConnections); + struct Subscription { + enum class State : uint8_t { + /// No subscription attempt has been made, or we have unsubscribed after all references + /// were released + Unsubscribed, + + /// The subscription attempt failed, maxing out our retry attempts + Failed, + + /// The initial subscription is currently in progress + Subscribing, + + /// A retry is currently in progress + Retrying, + + /// The subscription has been established + Subscribed, + + /// We've lost interested in this subscription - currently unsubscribing + Unsubscribing, + } state = State::Unsubscribed; + int32_t refCount = 0; std::weak_ptr connection; @@ -91,6 +129,8 @@ private: std::mutex subscriptionsMutex; std::unordered_map subscriptions; + + std::atomic quitting = false; }; class DummyController : public IController @@ -101,14 +141,19 @@ public: void removeRef(const SubscriptionRequest &request) override { (void)request; - }; + } + + void setQuitting() override + { + // + } [[nodiscard]] SubscriptionHandle subscribe( const SubscriptionRequest &request) override { (void)request; return {}; - }; + } }; } // namespace chatterino::eventsub diff --git a/src/providers/twitch/eventsub/SubscriptionRequest.cpp b/src/providers/twitch/eventsub/SubscriptionRequest.cpp index 81216d94..f49a7a73 100644 --- a/src/providers/twitch/eventsub/SubscriptionRequest.cpp +++ b/src/providers/twitch/eventsub/SubscriptionRequest.cpp @@ -4,20 +4,29 @@ namespace chatterino::eventsub { -QDebug &operator<<(QDebug &dbg, const SubscriptionRequest &v) +QDebug operator<<(QDebug dbg, const SubscriptionRequest &v) { - dbg << "eventsub::SubscriptionRequest{ type:" << v.subscriptionType - << "version:" << v.subscriptionVersion; + QDebugStateSaver saver(dbg); + + dbg.nospace().noquote() + << "eventsub::SubscriptionRequest[" << v.subscriptionType << '.' + << v.subscriptionVersion << "]{"; + bool first = true; if (!v.conditions.empty()) { - dbg << "conditions:["; for (const auto &[conditionKey, conditionValue] : v.conditions) { - dbg << conditionKey << "=" << conditionValue << ','; + if (!first) + { + dbg.nospace() << ", "; + } + dbg.nospace().noquote() << conditionKey << "=" << conditionValue; + + first = false; } - dbg << ']'; } - dbg << '}'; + dbg.nospace() << "} "; + return dbg; } diff --git a/src/providers/twitch/eventsub/SubscriptionRequest.hpp b/src/providers/twitch/eventsub/SubscriptionRequest.hpp index dde8090d..377aa3e0 100644 --- a/src/providers/twitch/eventsub/SubscriptionRequest.hpp +++ b/src/providers/twitch/eventsub/SubscriptionRequest.hpp @@ -21,7 +21,7 @@ struct SubscriptionRequest { /// Optional list of conditions for the subscription std::vector> conditions; - friend QDebug &operator<<(QDebug &dbg, const SubscriptionRequest &v); + friend QDebug operator<<(QDebug dbg, const SubscriptionRequest &v); }; bool operator==(const SubscriptionRequest &lhs, const SubscriptionRequest &rhs);