fix(eventsub): connection race (#6017)

This commit is contained in:
pajlada
2025-03-02 16:48:08 +01:00
committed by GitHub
parent 1812f1bb15
commit 58859d439c
7 changed files with 404 additions and 156 deletions
+1 -1
View File
@@ -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)
+3 -1
View File
@@ -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;
}
+1 -1
View File
@@ -151,6 +151,7 @@ private:
const std::unique_ptr<Logging> logging;
std::unique_ptr<Emotes> emotes;
std::unique_ptr<AccountController> accounts;
std::unique_ptr<eventsub::IController> eventSub;
std::unique_ptr<HotkeyController> hotkeys;
std::unique_ptr<WindowManager> windows;
std::unique_ptr<Toasts> toasts;
@@ -178,7 +179,6 @@ private:
std::unique_ptr<IStreamerMode> streamerMode;
std::unique_ptr<ITwitchUsers> twitchUsers;
std::unique_ptr<pronouns::Pronouns> pronouns;
std::unique_ptr<eventsub::IController> eventSub;
#ifdef CHATTERINO_HAVE_PLUGINS
std::unique_ptr<PluginController> plugins;
#endif
+335 -143
View File
@@ -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<uint8_t>(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<uint8_t>(subscription.state);
}
auto handle = std::make_unique<RawSubscriptionHandle>(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 *>(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 *>(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<lib::Session>(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<std::shared_ptr<lib::Session>> 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 *>(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<uint8_t>(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<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;
subscription.state = Subscription::State::Retrying;
int attemptNumber = subscription.retryAttempts;
auto retryTimer =
std::make_unique<boost::asio::deadline_timer>(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<lib::Session> 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<uint8_t>(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
+47 -2
View File
@@ -11,7 +11,10 @@
#include <QJsonObject>
#include <QString>
#include <atomic>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <thread>
@@ -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<lib::Session> 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<std::weak_ptr<lib::Session>> connections;
[[nodiscard]] std::optional<std::shared_ptr<lib::Session>>
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<lib::Session> connection;
@@ -91,6 +129,8 @@ private:
std::mutex subscriptionsMutex;
std::unordered_map<SubscriptionRequest, Subscription> subscriptions;
std::atomic<bool> 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
@@ -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;
}
@@ -21,7 +21,7 @@ struct SubscriptionRequest {
/// Optional list of conditions for the subscription
std::vector<std::pair<QString, QString>> 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);