From 11074238932078e5e17304035ff42d2494080da3 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 11 May 2026 17:34:54 -0500 Subject: [PATCH] fix: Prevent redundant shutdown calls in StreamGenerator and clean up profile managers in ProxyServer --- apps/proxy/live_proxy/output/ts/generator.py | 8 ++++++++ apps/proxy/live_proxy/server.py | 13 +++++++++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index 15596b7c..dcff594f 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -624,6 +624,14 @@ class StreamGenerator: # If no clients left and we're the owner, schedule shutdown using the config value if local_clients == 0 and proxy_server.am_i_owner(self.channel_id): + # When output/profile managers are present, remove_client already spawned + # handle_client_disconnect which will stop them and then stop the channel. + # Spawning a second shutdown greenlet here causes a redundant concurrent + # stop_channel call that can race against the first. + if (proxy_server.output_managers.get(self.channel_id) or + proxy_server.profile_managers.get(self.channel_id)): + return + logger.info(f"No local clients left for channel {self.channel_id}, scheduling shutdown") def delayed_shutdown(): diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index 53958c21..6ccb42b8 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -219,8 +219,9 @@ class ProxyServer: client_id = data.get("client_id") worker_id = data.get("worker_id") logger.debug(f"Owner received {EventType.CLIENT_DISCONNECTED} event for channel {channel_id}, client {client_id} from worker {worker_id}") - # Delegate to dedicated method - self.handle_client_disconnect(channel_id) + # Spawn to avoid blocking the pubsub listener thread + # during the full shutdown path (thread joins, Redis cleanup). + gevent.spawn(self.handle_client_disconnect, channel_id) elif event_type == EventType.STREAM_SWITCH: @@ -1336,6 +1337,14 @@ class ProxyServer: except KeyError: logger.debug(f"Client manager for channel {channel_id} already removed") + # Clean up profile managers and buffers. Owner workers clean profile_managers + # (and their profile_buffers) via stop_all_output_profiles above. Non-owner + # workers only populate profile_buffers (not profile_managers), and that block + # is skipped for them, so we always clean both here to avoid stale entries on + # the next connect. + self.profile_managers.pop(channel_id, None) + self.profile_buffers.pop(channel_id, None) + # Clean up Redis keys self._clean_redis_keys(channel_id)