From e28c3d3f267efa1973d752151fd691aa5b24dbac Mon Sep 17 00:00:00 2001 From: John Novak Date: Fri, 14 Aug 2026 07:38:53 -0700 Subject: [PATCH] Stop poll_top_streams hanging forever on a lost PART confirmation poll_top_streams runs under APScheduler with max_instances=1, so a single hung call silently stops every subsequent poll ("maximum number of running instances reached") until the process restarts. Throughput then decays over hours rather than failing outright. Two independent causes, both fixed here. Chat side, the real one. twitchAPI's Chat.leave_room() adds the channel to _room_leave_locks, sends PART, then waits for the confirmation to clear it -- with no timeout. Chat.__connect() does not clear those locks on reconnect; only the full _stop() does. So a reconnect during leave_room() throws away the confirmation and the wait loop never ends. The trigger is our own no_message_reset_time=0.5, which drops the websocket receive timeout to 30s, so any 30-second lull reconnects. The failure compounds: less chat means more reconnects, which freezes the channel set, which means less chat. We depend on plain PyPI twitchAPI rather than a fork, so patches/ rewrites the installed package at image build time to add a leave_timeout, mirroring the join_timeout that join_room already has. The script is idempotent and exits non-zero if the upstream method body has moved, so a version bump fails the build instead of silently shipping unpatched. Same fix is open upstream as Teekeks/pyTwitchAPI#365; drop patches/ once that lands in a release. API side, defence in depth. get_streams is bounded only by aiohttp's 300s default, well past the 120s poll interval, so a stall eats several cycles before raising. Measured median is ~0.1s; 10s leaves ~100x headroom while still recovering inside one interval. Not lower: a skipped poll opens a 240s gap against the 180s REDIS_STREAMER_TTL, expiring online keys and churning lifecycle events. Deliberately no wait_for around join_room/leave_room. The library already bounds joins, a legal rate-limited join can take up to 20s, and cancelling from outside leaks the very lock entry this patch exists to release. Co-Authored-By: Claude Opus 5 --- services/stream-monitoring/Dockerfile | 5 + .../patches/twitchapi_leave_room_timeout.py | 92 +++++++++++++++++++ .../stream_monitoring_service.py | 34 ++++++- 3 files changed, 126 insertions(+), 5 deletions(-) create mode 100644 services/stream-monitoring/patches/twitchapi_leave_room_timeout.py diff --git a/services/stream-monitoring/Dockerfile b/services/stream-monitoring/Dockerfile index b5076b3..d8ba9e2 100644 --- a/services/stream-monitoring/Dockerfile +++ b/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 . diff --git a/services/stream-monitoring/patches/twitchapi_leave_room_timeout.py b/services/stream-monitoring/patches/twitchapi_leave_room_timeout.py new file mode 100644 index 0000000..e8d1f0f --- /dev/null +++ b/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") diff --git a/services/stream-monitoring/stream_monitoring_service.py b/services/stream-monitoring/stream_monitoring_service.py index b5aa8a5..369bf47 100644 --- a/services/stream-monitoring/stream_monitoring_service.py +++ b/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