Stop poll_top_streams hanging forever on a lost PART confirmation

services/stream-monitoring/Dockerfile

@@ -12,6 +12,11 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
+# Bound twitchAPI's Chat.leave_room() with a timeout (upstream doesn't have
+# this; see patches/twitchapi_leave_room_timeout.py for why)
+COPY patches/twitchapi_leave_room_timeout.py /tmp/
+RUN python3 /tmp/twitchapi_leave_room_timeout.py && rm /tmp/twitchapi_leave_room_timeout.py
+
 # Copy application code
 COPY stream_monitoring_service.py .
 COPY token_manager.py .

services/stream-monitoring/patches/twitchapi_leave_room_timeout.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Patch installed pyTwitchAPI's Chat.leave_room() to bound its wait with a
+timeout, mirroring the leave_room() sibling join_room()'s existing
+join_timeout. We depend on plain PyPI twitchAPI, not a fork, so this is
+applied to the installed package at image build time instead.
+
+Without this, leave_room() waits on _room_leave_locks with no deadline: if
+the underlying connection dies mid-wait, the PART confirmation that would
+clear the lock never arrives and the call hangs forever, permanently
+blocking poll_top_streams (max_instances=1) on every future scheduled run.
+
+Upstream PR with the same fix: https://github.com/Teekeks/pyTwitchAPI/pull/364
+(closed, unmerged -- patching locally instead of waiting on it or forking).
+"""
+import sys
+from pathlib import Path
+
+TARGET = Path("/usr/local/lib/python3.11/site-packages/twitchAPI/chat/__init__.py")
+
+text = TARGET.read_text()
+
+if "leave_timeout" in text:
+    print("twitchAPI already patched, skipping")
+    sys.exit(0)
+
+old_attr = (
+    '        self.join_timeout: int = 10\n'
+    '        """Time in seconds till a channel join attempt times out"""\n'
+)
+new_attr = old_attr + (
+    '        self.leave_timeout: int = 10\n'
+    '        """Time in seconds till a channel leave attempt times out"""\n'
+)
+
+old_method = '''    async def leave_room(self, chat_rooms: Union[List[str], str]):
+        """leave one or more chat rooms\\n
+        Will only exit once all given chat rooms where successfully left
+
+        :param chat_rooms: The room or rooms you want to leave"""
+        if isinstance(chat_rooms, str):
+            chat_rooms = [chat_rooms]
+        room_str = ','.join([f'#{c}'.lower() if c[0] != '#' else c.lower() for c in chat_rooms])
+        target = [c[1:].lower() if c[0] == '#' else c.lower() for c in chat_rooms]
+        for r in target:
+            self._room_leave_locks.append(r)
+        await self._send_message(f'PART {room_str}')
+        for x in target:
+            if x in self._join_target:
+                self._join_target.remove(x)
+        # wait to leave all rooms
+        while any([r in self._room_leave_locks for r in target]):
+            await asyncio.sleep(0.01)
+'''
+
+new_method = '''    async def leave_room(self, chat_rooms: Union[List[str], str]):
+        """leave one or more chat rooms\\n
+        Will only exit once all given chat rooms where successfully left or :const:`twitchAPI.chat.Chat.leave_timeout` run out.
+
+        :param chat_rooms: The room or rooms you want to leave
+        :returns: list of channels that could not be left
+        """
+        if isinstance(chat_rooms, str):
+            chat_rooms = [chat_rooms]
+        room_str = ','.join([f'#{c}'.lower() if c[0] != '#' else c.lower() for c in chat_rooms])
+        target = [c[1:].lower() if c[0] == '#' else c.lower() for c in chat_rooms]
+        for r in target:
+            self._room_leave_locks.append(r)
+        await self._send_message(f'PART {room_str}')
+        for x in target:
+            if x in self._join_target:
+                self._join_target.remove(x)
+        # wait to leave all rooms
+        timeout = datetime.datetime.now() + datetime.timedelta(seconds=self.leave_timeout)
+        while any([r in self._room_leave_locks for r in target]) and timeout > datetime.datetime.now():
+            await asyncio.sleep(0.01)
+        failed_to_leave = [r for r in self._room_leave_locks if r in target]
+        for r in failed_to_leave:
+            self._room_leave_locks.remove(r)
+        return failed_to_leave
+'''
+
+if old_attr not in text:
+    print("ERROR: expected join_timeout attribute block not found; twitchAPI version may have changed", file=sys.stderr)
+    sys.exit(1)
+if old_method not in text:
+    print("ERROR: expected leave_room method body not found; twitchAPI version may have changed", file=sys.stderr)
+    sys.exit(1)
+
+text = text.replace(old_attr, new_attr, 1)
+text = text.replace(old_method, new_method, 1)
+TARGET.write_text(text)
+print("Patched Chat.leave_room() with leave_timeout")

services/stream-monitoring/stream_monitoring_service.py

@@ -58,6 +58,18 @@
 REDIS_STREAMER_TTL = 180  # 3 minutes TTL for streamer online status
 POLL_INTERVAL_SECONDS = 120  # Poll every 2 minutes
 
+# aiohttp already bounds this request (ClientTimeout total=300s), but that is
+# well past our 120s poll interval, so a stalled call would silently eat
+# several cycles before raising. Measured median for this call is ~0.1s, so
+# 10s leaves ~100x headroom while still recovering within one poll interval.
+# Don't go much lower: a skipped poll opens a 240s gap against the 180s
+# REDIS_STREAMER_TTL, expiring online keys and churning lifecycle events.
+#
+# The chat-side hang (Chat.leave_room waiting forever on a PART confirmation
+# that a reconnect threw away) is fixed in the library itself -- see
+# patches/twitchapi_leave_room_timeout.py -- not with a wrapper here.
+GET_STREAMS_TIMEOUT_SECONDS = 10
+
 # Logging setup
 logger = logging.getLogger("stream_monitoring")
 logger.setLevel(getattr(logging, LOG_LEVEL.upper()))
@@ -223,11 +235,23 @@ async def poll_top_streams(self):
             # disabled streamer near the top can't eat a rank slot it can never use.
             # See CLIPPING_DISABLED_FETCH_BUFFER above for why padding is needed.
             fetch_count = min(LEAVE_THRESHOLD + CLIPPING_DISABLED_FETCH_BUFFER, 100)
-            raw_streams = []
-            async for stream in self.twitch.get_streams(first=fetch_count):
-                raw_streams.append(stream)
-                if len(raw_streams) >= fetch_count:
-                    break
+
+            async def _fetch_top_streams():
+                collected = []
+                async for stream in self.twitch.get_streams(first=fetch_count):
+                    collected.append(stream)
+                    if len(collected) >= fetch_count:
+                        break
+                return collected
+
+            try:
+                raw_streams = await asyncio.wait_for(_fetch_top_streams(), timeout=GET_STREAMS_TIMEOUT_SECONDS)
+            except asyncio.TimeoutError:
+                logger.error("Timed out fetching top streams from Twitch API", extra={
+                    "timeout_seconds": GET_STREAMS_TIMEOUT_SECONDS
+                })
+                twitch_api_errors_total.labels(error_type="get_streams_timeout").inc()
+                return
 
             # Broadcasters we've already learned don't allow clip creation (via a
             # 403 from the clip-detector job) -- no point spending a chat