From fbf95108c4a2515d49306dc38395f2e621d1f644 Mon Sep 17 00:00:00 2001 From: HimanM <67066047+HimanM@users.noreply.github.com> Date: Mon, 7 Sep 2026 08:12:12 +0530 Subject: [PATCH] Make Twitch GQL recovery self-healing --- core/constants.py | 6 +-- network/twitch.py | 34 +++++++++++++--- tests/test_discord_notifications.py | 8 ++++ tests/test_web_auth.py | 31 +++++++++++++++ tests/test_web_gql_recovery.py | 60 +++++++++++++++++++++++++++++ web/controller.py | 49 ++++++++++++++++++----- web/discord.py | 9 ++++- web/manager.py | 6 +++ 8 files changed, 183 insertions(+), 20 deletions(-) create mode 100644 tests/test_web_gql_recovery.py diff --git a/core/constants.py b/core/constants.py index 0bfbef3..fd5bd98 100644 --- a/core/constants.py +++ b/core/constants.py @@ -345,7 +345,7 @@ def with_variables(self, variables: JsonType) -> GQLPersistedQuery: # returns all in-progress campaigns "Inventory": GQLPersistedQuery( "Inventory", - "d86775d0ef16a63a33ad52e80eaff963b2d5b72fada7c991504a57496e1d8e4b", + "8337eb8541b314040b0edde0c09c5c7a2783ba1960aa9edfbf3bac16d0fec404", variables={ "fetchRewardCampaigns": False, } @@ -362,7 +362,7 @@ def with_variables(self, variables: JsonType) -> GQLPersistedQuery: # returns all available campaigns "Campaigns": GQLPersistedQuery( "ViewerDropsDashboard", - "5a4da2ab3d5b47c9f9ce864e727b2cb346af1e3ea8b897fe8f704a97ff017619", + "d9cae7761dafab85908c85e6683cb4201b449e66ac3bb5e894f15ff12aeafaa7", variables={ "fetchRewardCampaigns": False, } @@ -400,7 +400,7 @@ def with_variables(self, variables: JsonType) -> GQLPersistedQuery: # returns live channels for a particular game "GameDirectory": GQLPersistedQuery( "DirectoryPage_Game", - "cb5dc816e139dcb8a118f14b4b677d59abc224a4b016c4bc2bb00a47fe0ddec4", + "86bcceb4e8b1a51256ff8eed8bd8aae4acacf80d737efe904f84f3aeadf8cafd", variables={ "limit": 30, # limit of channels returned "slug": ..., # game slug diff --git a/network/twitch.py b/network/twitch.py index ca0f95b..fe58fd1 100644 --- a/network/twitch.py +++ b/network/twitch.py @@ -65,6 +65,7 @@ logger = logging.getLogger("TwitchDrops") gql_logger = logging.getLogger("TwitchDrops.gql") +PERSISTED_QUERY_WARNING_AFTER = 15 * 60 class SkipExtraJsonDecoder(json.JSONDecoder): @@ -1379,14 +1380,19 @@ async def gql_request( backoff = ExponentialBackoff(maximum=60) # Use a flag to retry the request a single time, if a specific set of errors is encountered single_retry: bool = True + persisted_query_since: float | None = None + persisted_query_warned = False for delay in backoff: async with self._qgl_limiter: auth_state = await self.get_auth() + headers = auth_state.headers(user_agent=self._client_type.USER_AGENT, gql=True) + if persisted_query_since is not None: + headers["Connection"] = "close" async with self.request( "POST", "https://gql.twitch.tv/gql", json=ops, - headers=auth_state.headers(user_agent=self._client_type.USER_AGENT, gql=True), + headers=headers, ) as response: response_json: JsonType | list[JsonType] = await response.json() gql_logger.debug(f"GQL Response: {response_json}") @@ -1403,10 +1409,7 @@ async def gql_request( if "message" in error_dict: if ( single_retry - and error_dict["message"] in ( - "service error", - "PersistedQueryNotFound", - ) + and error_dict["message"] == "service error" ): logger.error( f"Retrying a {error_dict['message']} for " @@ -1418,6 +1421,27 @@ async def gql_request( delay = 5 force_retry = True break + elif error_dict["message"] == "PersistedQueryNotFound": + operation = response_json.get("extensions", {}).get( + "operationName", "unknown operation" + ) + logger.error(f"Retrying a PersistedQueryNotFound for {operation}") + if persisted_query_since is None: + persisted_query_since = time() + if ( + not persisted_query_warned + and time() - persisted_query_since + >= PERSISTED_QUERY_WARNING_AFTER + ): + self.print( + f"Twitch still rejects persisted query {operation} after " + "15 minutes. DropForge may require an update; retrying " + "automatically." + ) + persisted_query_warned = True + delay = max(delay, 5) + force_retry = True + break elif error_dict["message"] == "server error": # nullify the key the error path points to data_dict: JsonType = response_json["data"] diff --git a/tests/test_discord_notifications.py b/tests/test_discord_notifications.py index ecac9b5..7bcf13e 100644 --- a/tests/test_discord_notifications.py +++ b/tests/test_discord_notifications.py @@ -120,6 +120,14 @@ def test_idle_reason_is_deduplicated_and_recovery_is_reported(self) -> None: "Mining resumed", ]) + def test_operational_events_are_deduplicated(self) -> None: + self.notifier.update({"notify_operational": True}) + + self.notifier.operational("Miner restarting", "Retrying", event_key="restart") + self.notifier.operational("Miner restarting", "Retrying", event_key="restart") + + self.assertEqual(len(self.payloads), 1) + class DiscordNotificationApiTests(unittest.IsolatedAsyncioTestCase): async def test_authenticated_update_never_returns_the_webhook_token(self) -> None: diff --git a/tests/test_web_auth.py b/tests/test_web_auth.py index 501ebee..d7c9cfe 100644 --- a/tests/test_web_auth.py +++ b/tests/test_web_auth.py @@ -1,3 +1,4 @@ +import asyncio import tempfile import unittest from unittest.mock import AsyncMock, patch @@ -67,6 +68,36 @@ def test_stopped_miner_does_not_expose_stale_device_code(self): self.assertEqual(snapshot["login"]["user_code"], "") +class MinerSupervisorTests(unittest.IsolatedAsyncioTestCase): + async def test_failed_engine_restarts_until_stopped(self) -> None: + controller = MinerController() + controller._run_once = AsyncMock(side_effect=[True, False]) + + async def timeout(awaitable, **_kwargs): + awaitable.close() + raise asyncio.TimeoutError + + with patch( + "web.controller.asyncio.wait_for", + new=timeout, + ): + await controller._run() + + self.assertEqual(controller._run_once.await_count, 2) + + async def test_manual_stop_prevents_restart(self) -> None: + controller = MinerController() + + async def stopped_attempt(): + controller._stop_requested.set() + return True + + controller._run_once = AsyncMock(side_effect=stopped_attempt) + await controller._run() + + controller._run_once.assert_awaited_once() + + class WebResponseTests(unittest.IsolatedAsyncioTestCase): async def test_api_responses_are_never_cached(self): with tempfile.TemporaryDirectory() as directory: diff --git a/tests/test_web_gql_recovery.py b/tests/test_web_gql_recovery.py new file mode 100644 index 0000000..f0eafda --- /dev/null +++ b/tests/test_web_gql_recovery.py @@ -0,0 +1,60 @@ +import unittest +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch + +from core.constants import GQLPersistedQuery +from network.twitch import Twitch + + +class AsyncContext: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + +class PersistedQueryRecoveryTests(unittest.IsolatedAsyncioTestCase): + async def test_retries_until_persisted_query_recovers(self) -> None: + twitch = object.__new__(Twitch) + twitch._qgl_limiter = AsyncContext() + twitch._client_type = SimpleNamespace(USER_AGENT="test") + twitch.get_auth = AsyncMock( + return_value=SimpleNamespace(headers=lambda **_kwargs: {}) + ) + twitch.print = Mock() + replies = iter([ + { + "errors": [{"message": "PersistedQueryNotFound"}], + "extensions": {"operationName": "Inventory"}, + }, + { + "errors": [{"message": "PersistedQueryNotFound"}], + "extensions": {"operationName": "Inventory"}, + }, + {"data": {"currentUser": {}}}, + ]) + requests = [] + + @asynccontextmanager + async def request(*_args, **kwargs): + requests.append(kwargs) + yield SimpleNamespace(json=AsyncMock(return_value=next(replies))) + + twitch.request = request + query = GQLPersistedQuery("Inventory", "0" * 64) + with ( + patch("network.twitch.PERSISTED_QUERY_WARNING_AFTER", 0), + patch("network.twitch.asyncio.sleep", new=AsyncMock()), + ): + result = await twitch.gql_request(query) + + self.assertEqual(result, {"data": {"currentUser": {}}}) + self.assertNotIn("Connection", requests[0]["headers"]) + self.assertEqual(requests[1]["headers"]["Connection"], "close") + twitch.print.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/web/controller.py b/web/controller.py index 4567c74..18a6d77 100644 --- a/web/controller.py +++ b/web/controller.py @@ -5,13 +5,14 @@ import logging import traceback from argparse import Namespace +from time import monotonic from typing import Any from core.constants import COOKIES_PATH, FILE_FORMATTER, LOCK_PATH, LOG_PATH from core.exceptions import CaptchaRequired from core.settings import Settings from core.translate import _ -from core.utils import lock_file +from core.utils import ExponentialBackoff, lock_file from network.twitch import Twitch from web.discord import DiscordNotifier from web.manager import WebManager @@ -26,6 +27,7 @@ def __init__(self, notifier: DiscordNotifier | None = None) -> None: self._lock = asyncio.Lock() self._instance_lock: io.TextIOWrapper | None = None self._logging_configured = False + self._stop_requested = asyncio.Event() self.last_error = "" @property @@ -37,18 +39,21 @@ async def start(self) -> bool: if self.running: return False self.last_error = "" + self._stop_requested.clear() self._task = asyncio.create_task(self._run(), name="tdminer") await asyncio.sleep(0) return True async def stop(self, *, notify: bool = True) -> bool: async with self._lock: - if not self.running or self.manager is None: + if not self.running: return False + self._stop_requested.set() task = self._task - if notify and self.notifier is not None: + if notify and self.notifier is not None and self.manager is not None: self.notifier.miner_stopped(self.manager) - self.manager.close() + if self.manager is not None: + self.manager.close() if task is not None: try: await asyncio.wait_for(asyncio.shield(task), timeout=20) @@ -71,11 +76,36 @@ async def reset_auth(self) -> bool: return await self.start() async def _run(self) -> None: + backoff = ExponentialBackoff(variance=0, maximum=60) + while not self._stop_requested.is_set(): + started = monotonic() + restart = await self._run_once() + if not restart or self._stop_requested.is_set(): + return + if monotonic() - started >= 5 * 60: + backoff.reset() + delay = min(5 * next(backoff), 60) + logger = logging.getLogger("TwitchDrops") + logger.warning("Miner engine restarting automatically in %d seconds", delay) + if self.notifier is not None: + self.notifier.operational( + "Miner restarting automatically", + f"DropForge encountered an unexpected error and will retry in {delay:.0f} seconds.", + event_key="miner-restart", + ) + try: + await asyncio.wait_for(self._stop_requested.wait(), timeout=delay) + except asyncio.TimeoutError: + pass + + async def _run_once(self) -> bool: success, instance_lock = lock_file(LOCK_PATH) if not success: self.last_error = f"Another tdminer instance is already running or the lock is busy: {LOCK_PATH}" - return + return False self._instance_lock = instance_lock + restart = False + self.last_error = "" args = Namespace( log=True, tray=False, @@ -109,20 +139,18 @@ async def _run(self) -> None: await client.run() except CaptchaRequired: self.last_error = _("error", "captcha") - client.print(self.last_error) + if self._client is not None: + self._client.print(self.last_error) if self.notifier is not None: self.notifier.operational("Twitch verification required", self.last_error) except asyncio.CancelledError: raise except Exception: + restart = True self.last_error = traceback.format_exc() if self._client is not None: self._client.print("Fatal error encountered:") self._client.print(self.last_error) - if self.notifier is not None: - self.notifier.operational( - "Miner stopped unexpectedly", self.last_error.splitlines()[-1] - ) finally: if self._client is not None: await self._client.shutdown() @@ -131,6 +159,7 @@ async def _run(self) -> None: self._client = None instance_lock.close() self._instance_lock = None + return restart def snapshot(self) -> dict[str, Any]: state = self.manager.snapshot() if self.manager is not None else { diff --git a/web/discord.py b/web/discord.py index 77ac5bc..796c2ad 100644 --- a/web/discord.py +++ b/web/discord.py @@ -234,8 +234,13 @@ def miner_stopped(self, manager: WebManager) -> None: image=_image(drop.campaign.image_url), )) - def operational(self, title: str, detail: str) -> None: - if self._enabled("operational"): + def operational(self, title: str, detail: str, *, event_key: str = "") -> None: + if self._enabled("operational") and ( + not event_key + or self.store.claim_notification_event( + f"discord:operational:{event_key}", cooldown=6 * 60 * 60 + ) + ): self._schedule(self._payload([self._embed(title, detail, _RED)])) def _enabled(self, event: str) -> bool: diff --git a/web/manager.py b/web/manager.py index aaab48b..b959933 100644 --- a/web/manager.py +++ b/web/manager.py @@ -74,6 +74,12 @@ def print(self, message: str) -> None: super().print(message) if self.notifier is not None and message == _("status", "no_channel"): self.notifier.idle(self) + elif self.notifier is not None and message.startswith( + "Twitch still rejects persisted query " + ): + self.notifier.operational( + "Twitch query unavailable", message, event_key=message.split(" after ", 1)[0] + ) def selected_channel_id(self) -> str | None: return self._selected_channel_id