From 9527421bcd162dca553fba711732ece52c88cca5 Mon Sep 17 00:00:00 2001 From: pajlada Date: Sun, 9 Feb 2025 13:53:31 +0100 Subject: [PATCH] feat(eventsub): properly unsubscribe once no more handles are interested (#5943) --- CHANGELOG.md | 2 +- mocks/include/mocks/Helix.hpp | 5 + src/providers/twitch/api/Helix.cpp | 40 ++++++ src/providers/twitch/api/Helix.hpp | 10 ++ src/providers/twitch/eventsub/Controller.cpp | 131 +++++++++++++------ src/providers/twitch/eventsub/Controller.hpp | 28 ++-- 6 files changed, 167 insertions(+), 49 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e5f4e0e..94bd573f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ - Bugfix: Fixed the reply button showing for inline whispers and announcements. (#5863) - Bugfix: Fixed suspicious user treatment update messages not being searchable. (#5865) - Bugfix: Ensure miniaudio backend exits even if it doesn't exit cleanly. (#5896) -- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932) +- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943) - 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/mocks/include/mocks/Helix.hpp b/mocks/include/mocks/Helix.hpp index 06f75ab2..3c55d3dd 100644 --- a/mocks/include/mocks/Helix.hpp +++ b/mocks/include/mocks/Helix.hpp @@ -436,6 +436,11 @@ public: failureCallback)), (override)); + MOCK_METHOD(void, deleteEventSubSubscription, + (const QString &request, ResultCallback<> successCallback, + (FailureCallback failureCallback)), + (override)); + MOCK_METHOD(void, update, (QString clientId, QString oauthToken), (override)); diff --git a/src/providers/twitch/api/Helix.cpp b/src/providers/twitch/api/Helix.cpp index cef4515d..3ececeda 100644 --- a/src/providers/twitch/api/Helix.cpp +++ b/src/providers/twitch/api/Helix.cpp @@ -3334,6 +3334,46 @@ QDebug &operator<<(QDebug &dbg, return dbg; } +void Helix::deleteEventSubSubscription(const QString &subscriptionID, + ResultCallback<> successCallback, + FailureCallback failureCallback) +{ + QUrlQuery query; + query.addQueryItem("id", subscriptionID); + + this->makeDelete("eventsub/subscriptions", query) + .onSuccess([successCallback](const auto &result) { + if (result.status() != 204) + { + qCWarning(chatterinoTwitchEventSub) + << "Success result for deleting eventsub subscription was " + << result.formatError() << "but we expected it to be 204"; + } + + successCallback(); + }) + .onError([failureCallback](const NetworkResult &result) { + if (!result.status()) + { + failureCallback(result.formatError()); + return; + } + + const auto obj = result.parseJson(); + auto message = obj["message"].toString(); + + if (message.isEmpty()) + { + failureCallback("Twitch internal server error"); + } + else + { + failureCallback(message); + } + }) + .execute(); +} + NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery, NetworkRequestType type) { diff --git a/src/providers/twitch/api/Helix.hpp b/src/providers/twitch/api/Helix.hpp index 00dc6e7f..8298a520 100644 --- a/src/providers/twitch/api/Helix.hpp +++ b/src/providers/twitch/api/Helix.hpp @@ -1217,6 +1217,11 @@ public: FailureCallback failureCallback) = 0; + // https://dev.twitch.tv/docs/api/reference/#delete-eventsub-subscription + virtual void deleteEventSubSubscription( + const QString &subscriptionID, ResultCallback<> successCallback, + FailureCallback failureCallback) = 0; + virtual void update(QString clientId, QString oauthToken) = 0; protected: @@ -1567,6 +1572,11 @@ public: FailureCallback failureCallback) final; + // https://dev.twitch.tv/docs/api/reference/#delete-eventsub-subscription + void deleteEventSubSubscription( + const QString &subscriptionID, ResultCallback<> successCallback, + FailureCallback failureCallback) final; + void update(QString clientId, QString oauthToken) final; static void initialize(); diff --git a/src/providers/twitch/eventsub/Controller.cpp b/src/providers/twitch/eventsub/Controller.cpp index 8f53f878..bb853b99 100644 --- a/src/providers/twitch/eventsub/Controller.cpp +++ b/src/providers/twitch/eventsub/Controller.cpp @@ -60,7 +60,6 @@ Controller::Controller() Controller::~Controller() { qCInfo(LOG) << "Controller dtor start"; - this->queuedSubscriptions.clear(); for (const auto &weakConnection : this->connections) { @@ -73,6 +72,8 @@ Controller::~Controller() connection->close(); } + this->subscriptions.clear(); + this->work.reset(); if (this->thread->joinable()) @@ -91,15 +92,35 @@ void Controller::removeRef(const SubscriptionRequest &request) { std::lock_guard lock(this->subscriptionsMutex); - assert(this->activeSubscriptions.contains(request)); + assert(this->subscriptions.contains(request)); - auto &xd = this->activeSubscriptions[request]; - xd.refCount--; - qCInfo(LOG) << "Remove ref for" << request << xd.refCount; - // todo use actual atomic things here to be smart - if (xd.refCount <= 0) + auto &subscription = this->subscriptions[request]; + subscription.refCount--; + qCInfo(LOG) << "Removed ref for" << request << subscription.refCount; + + if (subscription.refCount <= 0) { - qCInfo(LOG) << "TODO: Unsubscribe from" << request; + 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?"; + return; + } + + qCInfo(LOG) << "Unsubscribing from" << request; + getHelix()->deleteEventSubSubscription( + subscription.subscriptionID, + [request] { + qCInfo(LOG) << "Successfully unsubscribed from" << request; + }, + [request](const auto &errorMessage) { + qCInfo(LOG) + << "An error occurred while attempting to unsubscribe from" + << request << errorMessage; + }); + + subscription.subscriptionID.clear(); } } @@ -110,13 +131,13 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request) { std::lock_guard lock(this->subscriptionsMutex); - auto &xd = this->activeSubscriptions[request]; - if (xd.refCount == 0) + auto &subscription = this->subscriptions[request]; + if (subscription.refCount == 0) { needToSubscribe = true; } - xd.refCount++; - qCInfo(LOG) << "Add ref for" << request << xd.refCount; + subscription.refCount++; + qCInfo(LOG) << "Added ref for" << request << subscription.refCount; } auto handle = std::make_unique(request); @@ -131,24 +152,29 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request) return handle; } -void Controller::subscribe(const SubscriptionRequest &request, bool isQueued) +void Controller::subscribe(const SubscriptionRequest &request, bool isRetry) { qCInfo(LOG) << "Subscribe request for" << request.subscriptionType; - boost::asio::post(this->ioContext, [this, request, isQueued] { + boost::asio::post(this->ioContext, [this, request, isRetry] { // 1. Flush dead connections (maybe this should not be done here) // TODO: implement - if (isQueued) { - qCInfo(LOG) << "Removing subscription from queued list"; - this->queuedSubscriptions.erase(request); - } + std::lock_guard lock(this->subscriptionsMutex); + auto &subscription = this->subscriptions[request]; + if (isRetry) + { + qCInfo(LOG) << "Removing subscription from queued list"; - if (this->queuedSubscriptions.contains(request)) - { - qCWarning(LOG) << "We already have a queued subscription for this, " - "let's chill :)"; - return; + subscription.retryTimer.reset(); + } + else if (subscription.retryTimer) + { + qCWarning(LOG) + << "We already have a queued subscription for this, " + "let's chill :)"; + return; + } } uint32_t openButNotReadyConnections = 0; @@ -188,7 +214,8 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued) [this, request, connection, weakConnection{weakConnection}](const auto &res) { qCInfo(LOG) << "success" << res; - this->markRequestSubscribed(request, weakConnection); + this->markRequestSubscribed(request, weakConnection, + res.subscriptionID); }, [this, request](const auto &error, const auto &errorString) { using Error = HelixCreateEventSubSubscriptionError; @@ -204,6 +231,10 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued) 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: @@ -221,8 +252,8 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued) << "Unhandled error, retrying subscription" << errorString; boost::asio::post(this->ioContext, [this, request] { - this->queueSubscription( - request, boost::posix_time::seconds(2)); + this->retrySubscription( + request, boost::posix_time::seconds(2), 5); }); break; } @@ -234,18 +265,20 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued) { // No connection was available to handle this subscription request, create a new connection this->createConnection(); - this->queueSubscription(request, boost::posix_time::millisec(500)); + this->retrySubscription(request, boost::posix_time::millisec(500), + 10); } else { if (openButNotReadyConnections > 1) { qCWarning(LOG) << "We have" << openButNotReadyConnections - << "open but not ready connections, hmmm"; + << "open but no ready connections, hmmm"; } // At least one connection is open, but it has not gotten the welcome message yet - this->queueSubscription(request, boost::posix_time::millisec(250)); + this->retrySubscription(request, boost::posix_time::millisec(250), + 10); } }); } @@ -288,15 +321,30 @@ void Controller::registerConnection(std::weak_ptr &&connection) this->connections.emplace_back(std::move(connection)); } -void Controller::queueSubscription(const SubscriptionRequest &request, - boost::posix_time::time_duration delay) +void Controller::retrySubscription(const SubscriptionRequest &request, + boost::posix_time::time_duration delay, + int32_t maxAttempts) { - this->threadGuard->guard(); + std::lock_guard lock(this->subscriptionsMutex); - auto resubTimer = + auto &connection = this->subscriptions[request]; + + if (connection.retryAttempts <= 0) + { + connection.retryAttempts = maxAttempts; + } + else if (--connection.retryAttempts == 0) + { + qCWarning(LOG) << "Reached max amount of retries for" << request; + return; + } + + qCInfo(LOG) << "Retrying subscription" << request << " - attempt" + << connection.retryAttempts; + auto retryTimer = std::make_unique(this->ioContext); - resubTimer->expires_from_now(delay); - resubTimer->async_wait([this, request](const auto &ec) { + retryTimer->expires_from_now(delay); + retryTimer->async_wait([this, request](const auto &ec) { if (!ec) { // The timer passed naturally @@ -304,15 +352,22 @@ void Controller::queueSubscription(const SubscriptionRequest &request, } }); - this->queuedSubscriptions.emplace(request, std::move(resubTimer)); + assert(connection.retryTimer == nullptr && + "Timer should not already be set"); + + connection.retryTimer = std::move(retryTimer); } void Controller::markRequestSubscribed(const SubscriptionRequest &request, - std::weak_ptr connection) + std::weak_ptr connection, + const QString &subscriptionID) { std::lock_guard lock(this->subscriptionsMutex); - this->activeSubscriptions[request].connection = std::move(connection); + auto &subscription = this->subscriptions[request]; + + subscription.connection = std::move(connection); + subscription.subscriptionID = subscriptionID; } } // namespace chatterino::eventsub diff --git a/src/providers/twitch/eventsub/Controller.hpp b/src/providers/twitch/eventsub/Controller.hpp index 113f4f12..c1c12337 100644 --- a/src/providers/twitch/eventsub/Controller.hpp +++ b/src/providers/twitch/eventsub/Controller.hpp @@ -22,6 +22,9 @@ class IController public: virtual ~IController() = default; + /// Removes one reference for the given subscription request + /// + /// Should realistically only be called in the dtor of SubscriptionHandle virtual void removeRef(const SubscriptionRequest &request) = 0; /// Subscribe will make a request to each open connection and ask them to @@ -47,16 +50,18 @@ public: const SubscriptionRequest &request) override; private: - void subscribe(const SubscriptionRequest &request, bool isQueued); + void subscribe(const SubscriptionRequest &request, bool isRetry); void createConnection(); void registerConnection(std::weak_ptr &&connection); - void queueSubscription(const SubscriptionRequest &request, - boost::posix_time::time_duration delay); + void retrySubscription(const SubscriptionRequest &request, + boost::posix_time::time_duration delay, + int32_t maxAttempts); void markRequestSubscribed(const SubscriptionRequest &request, - std::weak_ptr connection); + std::weak_ptr connection, + const QString &subscriptionID); const std::string userAgent; @@ -72,17 +77,20 @@ private: std::vector> connections; - struct XD { + struct Subscription { int32_t refCount = 0; std::weak_ptr connection; + + /// The ID of the subscription the Twitch Helix API has given us + QString subscriptionID; + + /// The timer, if any, for retrying the subscription creation + std::unique_ptr retryTimer; + int32_t retryAttempts = 0; }; std::mutex subscriptionsMutex; - std::unordered_map activeSubscriptions; - - std::unordered_map> - queuedSubscriptions; + std::unordered_map subscriptions; }; class DummyController : public IController