Bug Fix: Refactored channel teardown to kill ffmpeg earlier in the process. Removed redundant cleanup task. Added guard to prevent multiple cleanups for the same channel.

CHANGELOG.md

@@ -100,6 +100,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Plugin monkey-patches and module-level hooks were never applied in uWSGI workers.** Both uWSGI configs use `lazy-apps=true`, meaning each worker boots independently and never inherits state from the master. `should_skip_initialization()` correctly skipped one-shot startup tasks in workers, but also blocked `discover_plugins`, so plugin modules were never imported in any of the 4 request-serving workers. Plugins relying on patching request handling (monkey-patches, signal registrations) were silently inactive until a Connect event lazily triggered discovery in that specific worker. Discovery is now run in every process that serves requests, gated only for Celery processes (which use `worker_ready`) and management commands that don't serve requests.
 - **PostgreSQL connection pool exhausted under load in gevent workers.** uWSGI's gevent pool runs many greenlets concurrently on a single OS thread. With `CONN_MAX_AGE = 60`, each greenlet that touched the database retained its own open connection for the full 60 seconds (gevent thread-locals are greenlet-locals), rapidly exhausting PostgreSQL's `max_connections` limit under moderate concurrency. `DATABASE_CONN_MAX_AGE` is now `0` so connections are closed after each request. `close_old_connections()` is also called at the top of the long-running stream manager and cleanup watchdog loops so stale handles accumulated across greenlet switches are released promptly.
 - **Stream proxy race: cleanup watchdog could stop a channel still in the connecting phase.** When the first viewer triggered channel initialization, the client was registered only after the connect-wait loop completed. The cleanup watchdog runs concurrently and stops channels with zero connected clients after a grace period; if the grace period elapsed during the connect-wait, the watchdog killed the channel and left the viewer stuck. The client is now registered before the connect-wait loop begins so the watchdog always sees at least one client.
+- **2-second "stream thread did not terminate within timeout" warning on every channel stop.** `_close_socket` in `input/manager.py` was closing the relay pipe read-end (`self.socket`) before killing the ffmpeg process. The stream OS thread blocks in `select()` on that fd; on Linux, closing an fd from another thread while a `select()` is in progress on it does not reliably interrupt the call (POSIX allows this to be undefined). The thread stayed blocked for the full chunk-timeout (5 s), and `stream_thread.join(timeout=2.0)` always expired first. Fixed by killing ffmpeg first: when ffmpeg dies its copy of the relay write-end closes, delivering EOF to `select()` immediately. `self.socket` is closed afterward as cleanup only.
+- **Duplicate `stop_channel` calls on every channel shutdown.** `StreamGenerator._cleanup` triggered channel shutdown via two parallel paths: `client_manager.remove_client()` (which fires `handle_client_disconnect` on the owning worker, the correct path) and `_schedule_channel_shutdown_if_needed` (which independently spawned a delayed `stop_channel` greenlet). The duplicate call was suppressed by the `_stopping_channels` guard but produced a redundant log entry and unnecessary greenlet on every shutdown. `_schedule_channel_shutdown_if_needed` has been removed; `handle_client_disconnect` is the sole shutdown trigger.
+- **Concurrent greenlets could re-enter `_close_socket` during `proc.wait()`.** `self.transcode_process` was set to `None` at the end of the `if proc:` block rather than immediately after capturing the reference. Under gevent, `proc.wait(timeout=0.5)` yields the hub, allowing a second greenlet to enter `_close_socket`, find `self.transcode_process` still set, and attempt a second kill+close. `self.transcode_process = None` is now assigned immediately after `proc = self.transcode_process` so concurrent callers see `None` and skip the block.
 
 ### Performance
 

apps/proxy/live_proxy/input/manager.py

@@ -1440,23 +1440,20 @@ class StreamManager:
             except Exception as e:
                 logger.debug(f"Error stopping HTTP reader for channel {self.channel_id}: {e}")
 
-        # Otherwise handle socket and transcode resources
-        if self.socket:
-            try:
-                self.socket.close()
-            except Exception as e:
-                logger.debug(f"Error closing socket for channel {self.channel_id}: {e}")
-                pass
-
-        # Enhanced transcode process cleanup with immediate termination
-        if self.transcode_process:
+        # Kill proc before closing self.socket. Closing relay_read while the stream
+        # OS thread is blocked in select() on it does not reliably wake that select()
+        # on Linux. Killing ffmpeg closes its relay_write, sending EOF to the stream
+        # thread naturally. We close self.socket afterward as cleanup only.
+        proc = self.transcode_process
+        self.transcode_process = None  # claim early so concurrent greenlets skip this block
+        if proc:
             try:
                 logger.debug(f"Killing transcode process for channel {self.channel_id}")
-                self.transcode_process.kill()
+                proc.kill()
 
                 # Give it a very short time to die
                 try:
-                    self.transcode_process.wait(timeout=0.5)
+                    proc.wait(timeout=0.5)
                 except subprocess.TimeoutExpired:
                     logger.error(f"Failed to kill transcode process even with force for channel {self.channel_id}")
             except Exception as e:
@@ -1464,18 +1461,26 @@ class StreamManager:
 
                 # Final attempt: try to kill directly
                 try:
-                    self.transcode_process.kill()
+                    proc.kill()
                 except Exception as e:
                     logger.error(f"Final kill attempt failed for channel {self.channel_id}: {e}")
 
+        # Close relay socket after proc death; stream thread has already unblocked via EOF.
+        if self.socket:
+            try:
+                self.socket.close()
+            except Exception as e:
+                logger.debug(f"Error closing socket for channel {self.channel_id}: {e}")
+
+        if proc:
             # Explicitly close all subprocess pipes to prevent file descriptor leaks
             try:
-                if self.transcode_process.stdin:
-                    self.transcode_process.stdin.close()
-                if self.transcode_process.stdout:
-                    self.transcode_process.stdout.close()
-                if self.transcode_process.stderr:
-                    self.transcode_process.stderr.close()
+                if proc.stdin:
+                    proc.stdin.close()
+                if proc.stdout:
+                    proc.stdout.close()
+                if proc.stderr:
+                    proc.stderr.close()
                 logger.debug(f"Closed all subprocess pipes for channel {self.channel_id}")
             except Exception as e:
                 logger.debug(f"Error closing subprocess pipes for channel {self.channel_id}: {e}")
@@ -1492,8 +1497,7 @@ class StreamManager:
                 finally:
                     self.stderr_reader_thread = None
 
-            self.transcode_process = None
-            self.transcode_process_active = False  # Reset the flag
+            self.transcode_process_active = False
 
             # Clear transcode active key in Redis if available
             if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
@@ -1558,6 +1562,10 @@ class StreamManager:
                     try:
                         chunk = _os.read(fd, Config.CHUNK_SIZE)
                     except OSError as e:
+                        import errno as _errno
+                        if e.errno == _errno.EAGAIN and (self.stop_requested or not self.running):
+                            self.connected = False
+                            return False
                         logger.warning(f"Read error for channel {self.channel_id}: {e}")
                         self.connected = False
                         return False

apps/proxy/live_proxy/output/ts/generator.py

@@ -597,43 +597,6 @@ class StreamGenerator:
             except Exception as e:
                 logger.error(f"Could not log client disconnect event: {e}")
 
-            # Schedule channel shutdown if no clients left
-            self._schedule_channel_shutdown_if_needed(local_clients)
-
-    def _schedule_channel_shutdown_if_needed(self, local_clients):
-        """
-        Schedule channel shutdown if there are no clients left and we're the owner.
-        """
-        proxy_server = ProxyServer.get_instance()
-
-        # 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():
-                # Use the config setting instead of hardcoded value
-                shutdown_delay = ConfigHelper.channel_shutdown_delay()  # Use ConfigHelper
-                logger.info(f"Waiting {shutdown_delay}s before checking if channel should be stopped")
-                gevent.sleep(shutdown_delay)  # Replace time.sleep
-
-                # After delay, check global client count
-                if self.channel_id in proxy_server.client_managers:
-                    total = proxy_server.client_managers[self.channel_id].get_total_client_count()
-                    if total == 0:
-                        logger.info(f"Shutting down channel {self.channel_id} as no clients connected")
-                        proxy_server.stop_channel(self.channel_id)
-                    else:
-                        logger.info(f"Not shutting down channel {self.channel_id}, {total} clients still connected")
-
-            gevent.spawn(delayed_shutdown)
 
 def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None):
     """

apps/proxy/live_proxy/server.py

@@ -70,6 +70,7 @@ class ProxyServer:
         self.profile_managers = {}  # {channel_id: {profile_id: OutputProfileManager}}
         self.profile_buffers = {}   # {channel_id: {profile_id: StreamBuffer}}
         self._channel_names = {}
+        self._stopping_channels = set()  # channels with an active stop_channel call in progress
 
         # Generate a unique worker ID
         pid = os.getpid()
@@ -1267,6 +1268,10 @@ class ProxyServer:
 
     def stop_channel(self, channel_id):
         """Stop a channel with proper ownership handling"""
+        if channel_id in self._stopping_channels:
+            logger.debug(f"stop_channel already in progress for {channel_id}, ignoring duplicate call")
+            return
+        self._stopping_channels.add(channel_id)
         try:
             logger.info(f"Stopping channel {channel_id}")
 
@@ -1316,7 +1321,6 @@ class ProxyServer:
 
                 # Release ownership
                 self.release_ownership(channel_id)
-                logger.info(f"Released ownership of channel {channel_id}")
 
                 # Log channel stop event (after cleanup, before releasing ownership section ends)
                 try:
@@ -1404,6 +1408,8 @@ class ProxyServer:
         except Exception as e:
             logger.error(f"Error stopping channel {channel_id}: {e}")
             return False
+        finally:
+            self._stopping_channels.discard(channel_id)
 
     def check_inactive_channels(self):
         """Check for inactive channels (no clients) and stop them"""
@@ -1615,6 +1621,10 @@ class ProxyServer:
                             # Safety: if we have a stream_manager, we ARE the real owner
                             # but the Redis key may have expired. Try to re-acquire.
                             if channel_id in self.stream_managers:
+                                # Ownership was explicitly released by an active stop_channel call -
+                                # don't fight the shutdown by trying to re-acquire.
+                                if channel_id in self._stopping_channels:
+                                    continue
                                 logger.warning(
                                     f"Ownership gap for {channel_id}: this worker has stream_manager "
                                     f"but am_i_owner returned False. Attempting re-acquisition."