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