Make Twitch GQL recovery self-healing

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

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"]

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:

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:

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()

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 {

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:

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