Merge branch 'dev' of https://github.com/Dispatcharr/Dispatcharr into custom-psycopg3-pool

CHANGELOG.md

@@ -43,6 +43,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now signals shutdown (`buffer.stopping`), runs model `release_stream()` and Redis key deletion, releases ownership, stops ffmpeg/output managers and local buffers, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~10s (or 2× `channel_shutdown_delay`, whichever is greater). Stream buffers ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at teardown start).
+- **Client reconnect during channel teardown could wedge the channel across uWSGI workers.** The disconnect path called `proxy_server.stop_channel()` directly, so other workers were not notified and reconnecting clients could attach while ownership was released on the owner worker. The cleanup thread then re-acquired expired ownership and looped on orphaned clients while the upstream connection stayed open. All lifecycle stops now go through coordinated Redis teardown (`channel_stopping` flag, metadata `stopping` state, and `CHANNEL_STOP` pubsub). New stream requests are rejected with `503 Retry-After` only during active teardown (not during the post-disconnect shutdown delay grace period); `initialize_channel()` and ownership re-acquisition refuse to proceed in those states; non-owner workers clean local resources only on pubsub; and orphaned-client sweeps force cleanup when teardown is already active. (Fixes #1342)
+- **Orphaned upstream threads kept writing Redis after teardown.** Partial cleanup could remove a channel from `stream_managers` while the `stream-{uuid}` OS thread kept running, so orphan metadata sweeps only deleted Redis keys and chunks/metadata were immediately recreated (bare `total_bytes` hash with TTL -1). Stream managers now stop upstream when ownership is lost, orphan cleanup stops local ffmpeg/stream threads even when registry entries are gone (`_live_stream_managers` tracks managers until the thread exits), and forced recovery stops processes before Redis deletion instead of dropping dict entries without calling `stop()`.
+- **Rapid channel switching could leave bare `buffer:index` keys in Redis.** `buffer.stop()` flushed a final partial chunk via `INCR` during teardown, creating a persistent index key (TTL -1) after chunk keys expired; under load overlapping disconnect and cleanup-thread stops could race on `_stop_local_stream_activity`, blocking on ffmpeg stderr join before `_clean_redis_keys` ever ran. Teardown no longer writes to Redis from `buffer.stop()`, disconnect uses coordinated stop only (no duplicate upstream stop), model `release_stream()` runs before Redis keys are scanned/deleted (so `profile_connections` counters are not left stuck), ownership is released after Redis cleanup but before the blocking local ffmpeg stop, and a `finally` block guarantees Redis cleanup if local teardown raises.
+- **Live proxy could leak geventpool DB checkouts outside HTTP requests.** With `django-db-geventpool`, `connection.close()` returns handles to the per-worker pool (`MAX_CONNS=8`); without it, greenlets and OS threads keep connections checked out until the pool blocks on `pool.get()`. Proxy paths that touch the ORM outside Django's request cycle now call `close_old_connections()` after work completes: `_clean_redis_keys()`, profile lookup in `_establish_transcode_connection()` before spawning ffmpeg, and fMP4 client disconnect cleanup when releasing streams (TS disconnect relies on `log_system_event()`).
+- **System events could block callers on Connect and plugin handlers.** `log_system_event()` ran Connect subscriptions and plugin `"events"` hooks synchronously after the DB write, so slow integrations could stall live-proxy and streaming paths even when inserting the event row was fast. Connect/plugin dispatch now runs on a separate gevent when the hub is available (synchronously in Celery workers without a hub). `log_system_event()` and `dispatch_event_system()` also call `close_old_connections()` in `finally` blocks so integration work does not leave pool slots checked out on the caller or dispatch greenlet.
+- **Plugins could leak geventpool DB checkouts after UI or Connect event runs.** Third-party plugins run inline on uWSGI greenlets (manual actions and Connect `"events"` hooks) with no guaranteed connection cleanup at the plugin boundary. `PluginManager.run_action()` and `stop_plugin()` now call `close_old_connections()` in a `finally` block so each action returns its pool slot whether the plugin succeeds or raises.
+- **VOD proxy could leak geventpool DB checkouts during playback and stats updates.** `stream_vod()` ran ORM lookups for content and M3U profiles, then returned a long-lived `StreamingHttpResponse` without releasing the checkout, so each movie/episode stream could hold a pool slot for its full duration. Background VOD stats refresh (`build_vod_stats_data()`, triggered on start/stop and by the admin stats API) also queried movie, episode, and profile rows from daemon threads with no cleanup. `stream_vod()` now calls `close_old_connections()` before handing off to the streaming generator, and `build_vod_stats_data()` releases its checkout in a `finally` block.
+- **Channel shutdown delay did not reset after a reconnect within the grace period.** `handle_client_disconnect()` used a fixed `gevent.sleep(shutdown_delay)` from the first last-client disconnect. If a client reconnected and disconnected again during the delay, an earlier disconnect handler could still stop the channel on the original timer instead of waiting the full delay from the latest disconnect. Shutdown now polls Redis (`last_client_disconnect` timestamp and client count) so concurrent disconnect handlers and multi-worker reconnects always honour the latest disconnect time.
+- **Zombie ffmpeg could survive owner-lock expiry and orphan Redis sweeps.** When the 30s owner lock lapsed under single-worker load, disconnect handling treated the worker as non-owner so coordinated stop never ran, while ffmpeg kept writing and the orphan sweeper only deleted Redis keys (recreating them immediately). Disconnect now re-acquires ownership when local upstream is still active, stops locally when the last client leaves without a lock, orphan cleanup stops local ffmpeg before Redis deletion via `_has_local_upstream_activity`, re-init stops lingering upstream before starting a duplicate thread, and `is_channel_teardown_active` includes channels mid-`stop_channel` on this worker so rapid reconnect gets 503 during teardown.
 - **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots.
 - **EPG auto-match reliability fixes.**
   - Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set.

Plugins.md

@@ -307,6 +307,18 @@ Plugins are server-side Python code running within the Django application. You c
 
 Prefer Celery tasks (`.delay()`) to keep `run` fast and non-blocking.
 
+### Database connections
+
+Dispatcharr uses `django-db-geventpool` with a bounded per-uWSGI-worker pool (`MAX_CONNS=8`). Each greenlet or OS thread that runs ORM code checks out a connection until Django closes it.
+
+`PluginManager.run_action()` and `stop_plugin()` always call `close_old_connections()` in a `finally` block after your plugin returns (success or error). That returns the current greenlet's checkout to the pool. **You do not need to call `close_old_connections()` yourself for normal inline ORM inside `run()` or `stop()`.**
+
+Still follow these rules:
+
+- **Heavy or long work:** dispatch a Celery task (`.delay()`) and return quickly from `run()`. Celery workers close connections after each task; blocking the uWSGI gevent hub with `time.sleep`, sync HTTP, or large CPU work can freeze the whole worker regardless of DB cleanup.
+- **Background threads or greenlets you spawn:** each thread/greenlet that uses the ORM must call `close_old_connections()` (or `connection.close()`) in its own `finally` block when done. The wrapper only covers the thread/greenlet that called `run_action()`.
+- **Connect event hooks:** actions with an `"events"` list are dispatched from `log_system_event()` on a separate gevent when uWSGI has an active hub (otherwise synchronously, e.g. Celery). Keep handlers short or defer heavy work to Celery.
+
 ### Important: Don’t Ask Users for URL/User/Password
 Dispatcharr plugins run **inside** the Dispatcharr backend process. That means they already have direct access to the app’s models, tasks, and internal utilities.  
 Plugins **should not** ask users for “Dispatcharr URL”, “Admin Username”, or “Admin Password” just to call the API. That is unnecessary and unsafe because:

apps/channels/models.py

@@ -626,11 +626,11 @@ class Channel(models.Model):
 
     def _release_stale_stream_assignment(self, redis_client, stream_id: int) -> None:
         """Release pool counters and remove stale channel/stream assignment keys."""
+        profile_id = None
         profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
         if profile_id_bytes:
             try:
                 profile_id = int(profile_id_bytes)
-                release_profile_slot(profile_id, redis_client)
             except (ValueError, TypeError):
                 logger.debug(
                     "Invalid profile ID for stale assignment on stream %s: %s",
@@ -638,6 +638,32 @@ class Channel(models.Model):
                     profile_id_bytes,
                 )
 
+        if profile_id is None:
+            metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+            meta_profile_id = redis_client.hget(
+                metadata_key, ChannelMetadataField.M3U_PROFILE
+            )
+            if meta_profile_id:
+                try:
+                    profile_id = int(meta_profile_id)
+                except (ValueError, TypeError):
+                    logger.debug(
+                        "Invalid profile ID in metadata for stale assignment on "
+                        "channel %s: %s",
+                        self.uuid,
+                        meta_profile_id,
+                    )
+
+        if profile_id is not None:
+            release_profile_slot(profile_id, redis_client)
+        else:
+            logger.warning(
+                "Channel %s: releasing stale stream %s assignment without profile "
+                "info - profile_connections may leak",
+                self.uuid,
+                stream_id,
+            )
+
         redis_client.delete(f"channel_stream:{self.id}")
         redis_client.delete(f"stream_profile:{stream_id}")
 

apps/channels/tests/test_ts_proxy_ghost_clients.py

@@ -247,6 +247,14 @@ class BasicStatsGhostClientTests(TestCase):
         redis.scard.return_value = len(client_ids)
         redis.smembers.return_value = client_ids
         redis.hget.return_value = None  # individual field lookups
+        redis.hmget.return_value = [
+            b'VLC/3.0',
+            b'127.0.0.1',
+            b'1773500000.0',
+            None,
+            b'mpegts',
+            None,
+        ]
 
         # Pipeline for remove_ghost_clients
         pipe = MagicMock()

apps/channels/tests/test_ts_proxy_keepalive.py

@@ -218,7 +218,7 @@ class DoStatsUpdateTests(TestCase):
         mock_redis.scan.return_value = (0, [])
 
         with patch("apps.proxy.live_proxy.client_manager.send_websocket_update") as mock_ws, \
-             patch("redis.Redis.from_url", return_value=mock_redis):
+             patch("core.utils.RedisClient.get_client", return_value=mock_redis):
             cm._do_stats_update()
 
         mock_ws.assert_called_once()
@@ -231,25 +231,25 @@ class DoStatsUpdateTests(TestCase):
         """Redis failure must be swallowed (logged), not propagated."""
         cm = self._make_client_manager()
 
-        with patch("redis.Redis.from_url", side_effect=Exception("Redis down")):
+        with patch("core.utils.RedisClient.get_client", side_effect=Exception("Redis down")):
             try:
                 cm._do_stats_update()
             except Exception as e:
                 self.fail(f"_do_stats_update raised an exception: {e}")
 
-    def test_do_stats_update_scans_channel_client_keys(self):
-        """Must scan for live:channel:*:clients pattern."""
+    def test_do_stats_update_scans_channel_metadata_keys(self):
+        """Must scan for live:channel:*:metadata pattern."""
         cm = self._make_client_manager()
 
         mock_redis = MagicMock()
         mock_redis.scan.return_value = (0, [])
 
         with patch("apps.proxy.live_proxy.client_manager.send_websocket_update"), \
-             patch("redis.Redis.from_url", return_value=mock_redis):
+             patch("core.utils.RedisClient.get_client", return_value=mock_redis):
             cm._do_stats_update()
 
         scan_call = mock_redis.scan.call_args
-        self.assertIn("live:channel:*:clients", str(scan_call))
+        self.assertIn("live:channel:*:metadata", str(scan_call))
 
 
 # ---------------------------------------------------------------------------

apps/channels/tests/test_ts_proxy_teardown.py

@@ -0,0 +1,807 @@
+"""Tests for multi-worker channel teardown coordination."""
+import time
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+
+from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
+from apps.proxy.live_proxy.input.buffer import StreamBuffer
+from apps.proxy.live_proxy.input.manager import StreamManager
+from apps.proxy.live_proxy.redis_keys import RedisKeys
+from apps.proxy.live_proxy.server import ProxyServer
+from apps.proxy.live_proxy.services.channel_service import ChannelService
+
+
+CHANNEL_ID = "00000000-0000-0000-0000-000000000099"
+
+
+def _configure_ownership_pipeline(
+    redis,
+    *,
+    stop_exists=0,
+    metadata_exists=1,
+    client_count=0,
+    owner=None,
+    disconnect=None,
+    state=None,
+):
+    pipe = MagicMock()
+    redis.pipeline.return_value = pipe
+    pipe.execute.return_value = (
+        stop_exists,
+        metadata_exists,
+        client_count,
+        owner,
+        disconnect,
+        state,
+    )
+    return pipe
+
+
+def _mock_proxy_server(redis_client=None):
+    server = MagicMock()
+    server.redis_client = redis_client or MagicMock()
+    server._stopping_channels = set()
+    return server
+
+
+class ChannelTeardownAvailabilityTests(TestCase):
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_teardown_active_when_stopping_key_exists(self, mock_get_instance):
+        redis = MagicMock()
+        redis.exists.side_effect = lambda key: key == RedisKeys.channel_stopping(CHANNEL_ID)
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID))
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_teardown_active_when_metadata_state_is_stopping(self, mock_get_instance):
+        redis = MagicMock()
+        redis.exists.return_value = False
+        redis.hget.return_value = ChannelState.STOPPING.encode()
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID))
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay")
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_shutdown_pending_within_delay_window(self, mock_get_instance, mock_delay):
+        mock_delay.return_value = 5
+        redis = MagicMock()
+        redis.exists.return_value = False
+        redis.get.return_value = str(time.time() - 2).encode()
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertTrue(ChannelService.is_shutdown_pending(CHANNEL_ID))
+        self.assertFalse(ChannelService.is_channel_unavailable_for_new_clients(CHANNEL_ID))
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay")
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_cancel_pending_shutdown_clears_disconnect_key(self, mock_get_instance, mock_delay):
+        mock_delay.return_value = 30
+        redis = MagicMock()
+        redis.exists.side_effect = lambda key: "last_client_disconnect" in key
+        redis.get.return_value = None
+        redis.hget.return_value = ChannelState.ACTIVE.encode()
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertTrue(ChannelService.cancel_pending_shutdown(CHANNEL_ID))
+        redis.delete.assert_any_call(RedisKeys.last_client_disconnect(CHANNEL_ID))
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay")
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_cancel_pending_shutdown_skips_during_active_stop(self, mock_get_instance, mock_delay):
+        mock_delay.return_value = 30
+        redis = MagicMock()
+        redis.exists.return_value = True
+        server = _mock_proxy_server(redis)
+        server._stopping_channels = {CHANNEL_ID}
+        mock_get_instance.return_value = server
+
+        self.assertFalse(ChannelService.cancel_pending_shutdown(CHANNEL_ID))
+        redis.delete.assert_not_called()
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay")
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_cancel_pending_shutdown_skips_real_teardown_without_grace(self, mock_get_instance, mock_delay):
+        mock_delay.return_value = 30
+        redis = MagicMock()
+        redis.exists.side_effect = lambda key: "stopping" in key
+        redis.get.return_value = None
+        redis.hget.return_value = ChannelState.STOPPING.encode()
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertFalse(ChannelService.cancel_pending_shutdown(CHANNEL_ID))
+        redis.hset.assert_not_called()
+        redis.delete.assert_not_called()
+
+    @patch("apps.proxy.live_proxy.services.channel_service.ConfigHelper.channel_shutdown_delay")
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_shutdown_pending_expired_after_delay(self, mock_get_instance, mock_delay):
+        mock_delay.return_value = 5
+        redis = MagicMock()
+        redis.exists.return_value = False
+        redis.get.return_value = str(time.time() - 10).encode()
+        mock_get_instance.return_value = _mock_proxy_server(redis)
+
+        self.assertFalse(ChannelService.is_shutdown_pending(CHANNEL_ID))
+
+
+class ClientManagerAddClientTests(TestCase):
+    @patch("apps.proxy.live_proxy.services.channel_service.ChannelService.cancel_pending_shutdown", return_value=False)
+    @patch("apps.proxy.live_proxy.client_manager.send_websocket_update")
+    def test_add_client_stores_ip_and_user_agent_in_redis(self, _mock_ws, _mock_cancel):
+        from apps.proxy.live_proxy.client_manager import ClientManager
+
+        redis = MagicMock()
+        cm = ClientManager(CHANNEL_ID, redis_client=redis, worker_id="worker-1")
+        cm.proxy_server = MagicMock()
+
+        result = cm.add_client(
+            "client-1",
+            "10.0.2.163",
+            user_agent="VLC/3.0.21",
+        )
+
+        self.assertEqual(result, 1)
+        mapping = redis.hset.call_args[1]["mapping"]
+        self.assertEqual(mapping["ip_address"], "10.0.2.163")
+        self.assertEqual(mapping["user_agent"], "VLC/3.0.21")
+
+
+class LocalStreamActivityTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server.profile_managers = {}
+        server.profile_buffers = {}
+        server._live_stream_managers = {}
+        server._stopping_channels = set()
+        server.redis_client = MagicMock()
+        server.stop_all_output_formats = MagicMock()
+        server.stop_all_output_profiles = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_join_stream_thread")
+    def test_stop_local_stream_activity_stops_live_manager(self, mock_join):
+        server = self._make_server()
+        manager = MagicMock()
+        server._live_stream_managers[CHANNEL_ID] = manager
+
+        server._stop_local_stream_activity(CHANNEL_ID)
+
+        manager.stop.assert_called_once()
+        mock_join.assert_called_once_with(CHANNEL_ID)
+        self.assertNotIn(CHANNEL_ID, server._live_stream_managers)
+
+
+class OrphanMetadataCleanupTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server.profile_managers = {}
+        server.profile_buffers = {}
+        server._live_stream_managers = {}
+        server._stopping_channels = set()
+        server.redis_client = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_orphan_metadata_stops_local_processes_before_redis(
+        self, mock_has_upstream, mock_stop_local, mock_clean_redis
+    ):
+        server = self._make_server()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        server.redis_client.keys.return_value = [metadata_key.encode()]
+        server.redis_client.hgetall.return_value = {
+            b"owner": b"",
+            b"state": b"unknown",
+        }
+        server.redis_client.exists.return_value = False
+        server.redis_client.scard.return_value = 0
+
+        server._check_orphaned_metadata()
+
+        mock_has_upstream.assert_called_with(CHANNEL_ID)
+        mock_stop_local.assert_called_once_with(CHANNEL_ID)
+        mock_clean_redis.assert_called_once_with(CHANNEL_ID)
+
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
+    def test_orphan_metadata_remote_channel_broadcasts_stop(
+        self, mock_has_upstream, mock_stop_local, mock_broadcast, mock_clean_redis
+    ):
+        server = self._make_server()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        server.redis_client.keys.return_value = [metadata_key.encode()]
+        server.redis_client.hgetall.return_value = {b"owner": b"", b"state": b"unknown"}
+        server.redis_client.exists.return_value = False
+        server.redis_client.scard.return_value = 0
+
+        server._check_orphaned_metadata()
+
+        mock_broadcast.assert_called_once_with(CHANNEL_ID)
+        mock_stop_local.assert_not_called()
+        mock_clean_redis.assert_called_once_with(CHANNEL_ID)
+
+
+class OrphanChannelCleanupTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server.profile_managers = {}
+        server.profile_buffers = {}
+        server._live_stream_managers = {}
+        server._stopping_channels = set()
+        server.redis_client = MagicMock()
+        server.get_channel_owner = MagicMock(return_value=None)
+        return server
+
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_orphan_channel_stops_local_before_redis(
+        self, mock_has_upstream, mock_stop_local, mock_clean_redis
+    ):
+        server = self._make_server()
+        metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
+        server.redis_client.keys.return_value = [metadata_key.encode()]
+        server.redis_client.scard.return_value = 0
+
+        server._check_orphaned_channels()
+
+        mock_stop_local.assert_called_once_with(CHANNEL_ID)
+        mock_clean_redis.assert_called_once_with(CHANNEL_ID)
+
+
+class StreamManagerOwnershipTests(TestCase):
+    def test_still_owner_false_when_different_worker(self):
+        buffer = MagicMock()
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            owner=b"worker-b",
+        )
+        manager = StreamManager(
+            CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
+        )
+
+        self.assertFalse(manager._still_owner())
+
+    def test_still_owner_true_when_owner_lock_expired_but_not_stopping(self):
+        buffer = MagicMock()
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            client_count=0,
+            owner=None,
+            state=ChannelState.CONNECTING.encode(),
+        )
+        manager = StreamManager(
+            CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
+        )
+
+        self.assertTrue(manager._still_owner())
+
+    def test_still_owner_false_when_channel_stopping_key_set(self):
+        buffer = MagicMock()
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            stop_exists=1,
+        )
+        manager = StreamManager(
+            CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
+        )
+
+        self.assertFalse(manager._still_owner())
+
+    def test_update_bytes_skipped_after_ownership_lost(self):
+        buffer = MagicMock()
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            owner=b"other-worker",
+        )
+        manager = StreamManager(
+            CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
+        )
+        manager.bytes_processed = 1000
+
+        manager._update_bytes_processed(500)
+
+        buffer.redis_client.hincrby.assert_not_called()
+
+
+class StreamBufferStopTests(TestCase):
+    def test_stop_discards_local_data_without_redis_writes(self):
+        redis = MagicMock()
+        buffer = StreamBuffer(channel_id=CHANNEL_ID, redis_client=redis)
+        buffer._write_buffer = bytearray(b"x" * 376)
+
+        buffer.stop()
+
+        self.assertTrue(buffer.stopping)
+        self.assertEqual(len(buffer._write_buffer), 0)
+        redis.incr.assert_not_called()
+        redis.setex.assert_not_called()
+        redis.delete.assert_not_called()
+
+
+class StopChannelTeardownTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server.profile_managers = {}
+        server.profile_buffers = {}
+        server._live_stream_managers = {}
+        server._stopping_channels = set()
+        server._stopping_since = {}
+        server.redis_client = MagicMock()
+        server.redis_client.exists.return_value = False
+        server.am_i_owner = MagicMock(return_value=False)
+        server._collect_channel_stop_event_data = MagicMock(return_value=None)
+        server.release_ownership = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    def test_stop_channel_cleans_redis_before_blocking_local_stop(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+        call_order = []
+
+        def stop_local(channel_id):
+            call_order.append("local")
+
+        def clean_redis(channel_id):
+            call_order.append("redis")
+
+        mock_stop_local.side_effect = stop_local
+        mock_clean_redis.side_effect = clean_redis
+
+        server.stop_channel(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["redis", "local"])
+        mock_spawn_event.assert_called_once_with(None)
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity", side_effect=RuntimeError("boom"))
+    def test_stop_channel_cleans_redis_in_finally_when_local_stop_fails(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+
+        result = server.stop_channel(CHANNEL_ID)
+
+        self.assertFalse(result)
+        mock_clean_redis.assert_called_once_with(CHANNEL_ID)
+        mock_spawn_event.assert_not_called()
+        self.assertNotIn(CHANNEL_ID, server._stopping_channels)
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    def test_stop_channel_owner_releases_after_redis_cleanup(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+        server.am_i_owner.return_value = True
+        stop_data = {"channel_id": CHANNEL_ID}
+        server._collect_channel_stop_event_data.return_value = stop_data
+        call_order = []
+
+        def stop_local(channel_id):
+            call_order.append("local")
+
+        def release(channel_id):
+            call_order.append("release")
+
+        def clean_redis(channel_id):
+            call_order.append("redis")
+
+        mock_stop_local.side_effect = stop_local
+        server.release_ownership.side_effect = release
+        mock_clean_redis.side_effect = clean_redis
+
+        server.stop_channel(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["redis", "release", "local"])
+        server._collect_channel_stop_event_data.assert_called_once_with(CHANNEL_ID)
+        mock_spawn_event.assert_called_once_with(stop_data)
+
+
+class CleanRedisKeysOrderTests(TestCase):
+    @patch("apps.proxy.live_proxy.server.Stream.objects.get")
+    @patch("apps.proxy.live_proxy.server.Channel.objects.get")
+    def test_clean_redis_keys_releases_profile_slot_before_live_keys_deleted(
+        self, mock_channel_get, mock_stream_get
+    ):
+        from apps.channels.models import Channel, Stream
+
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.redis_client = MagicMock()
+        call_order = []
+
+        channel = MagicMock()
+        channel.release_stream.return_value = True
+
+        def channel_get(uuid):
+            call_order.append("release")
+            return channel
+
+        mock_channel_get.side_effect = channel_get
+        mock_stream_get.side_effect = Stream.DoesNotExist
+
+        channel_key = f"live:channel:{CHANNEL_ID}:input:buffer:index".encode()
+
+        def scan(cursor, match=None, count=100):
+            call_order.append("redis")
+            if match == f"live:channel:{CHANNEL_ID}:*":
+                return (0, [channel_key])
+            return (0, [])
+
+        server.redis_client.scan.side_effect = scan
+
+        server._clean_redis_keys(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["release", "redis", "redis"])
+        channel.release_stream.assert_called_once()
+        server.redis_client.delete.assert_called_once_with(channel_key)
+
+
+class LocalUpstreamActivityTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server._live_stream_managers = {}
+        return server
+
+    def test_upstream_activity_excludes_reader_only_client_manager(self):
+        server = self._make_server()
+        server.client_managers[CHANNEL_ID] = MagicMock()
+        server.stream_buffers[CHANNEL_ID] = MagicMock()
+        self.assertFalse(server._has_local_upstream_activity(CHANNEL_ID))
+
+    def test_upstream_activity_from_live_registry(self):
+        server = self._make_server()
+        server._live_stream_managers[CHANNEL_ID] = MagicMock()
+        self.assertTrue(server._has_local_upstream_activity(CHANNEL_ID))
+
+
+class ClientDisconnectOwnershipTests(TestCase):
+    def test_last_client_triggers_stop_when_upstream_active_without_owner_lock(self):
+        from apps.proxy.live_proxy.client_manager import ClientManager
+
+        proxy_server = MagicMock()
+        proxy_server.worker_id = "testhost:1"
+        proxy_server.am_i_owner.return_value = False
+        proxy_server._has_local_upstream_activity.return_value = True
+        proxy_server.extend_ownership.return_value = False
+
+        redis = MagicMock()
+        redis.hget.return_value = b"viewer"
+        redis.scard.return_value = 0
+
+        manager = ClientManager(
+            channel_id=CHANNEL_ID,
+            redis_client=redis,
+            worker_id="testhost:1",
+        )
+        manager.proxy_server = proxy_server
+        manager.clients = {"client-1"}
+        manager.client_set_key = RedisKeys.clients(CHANNEL_ID)
+        manager._notify_owner_of_activity = MagicMock()
+        manager._trigger_stats_update = MagicMock()
+        manager.get_total_client_count = MagicMock(return_value=0)
+        proxy_server._spawn_on_hub = MagicMock()
+
+        manager.remove_client("client-1")
+
+        proxy_server._spawn_on_hub.assert_called_once_with(
+            proxy_server.handle_client_disconnect, CHANNEL_ID
+        )
+
+
+class TeardownActiveLocalStopTests(TestCase):
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_teardown_active_when_local_stop_in_progress(self, mock_get_instance):
+        server = MagicMock()
+        server._stopping_channels = {CHANNEL_ID}
+        server.redis_client = MagicMock()
+        server.redis_client.exists.return_value = False
+        mock_get_instance.return_value = server
+
+        self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID))
+
+
+class HandleClientDisconnectUpstreamFirstTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.client_managers = {CHANNEL_ID: MagicMock()}
+        server.stream_managers = {CHANNEL_ID: MagicMock()}
+        server._live_stream_managers = {}
+        server.profile_managers = {}
+        server.output_managers = {}
+        server.redis_client = MagicMock()
+        server.redis_client.scard.return_value = 0
+        server._stopping_channels = set()
+        return server
+
+    @patch.object(ProxyServer, "_coordinated_stop_channel")
+    @patch.object(ProxyServer, "_stop_upstream_before_redis_cleanup")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_last_client_uses_coordinated_stop_only(
+        self, _mock_upstream, mock_stop_upstream, mock_coordinated
+    ):
+        server = self._make_server()
+
+        with patch(
+            "apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay",
+            return_value=0,
+        ):
+            server.handle_client_disconnect(CHANNEL_ID)
+
+        mock_stop_upstream.assert_not_called()
+        mock_coordinated.assert_called_once_with(CHANNEL_ID)
+
+
+class ShutdownDelayWaitTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            with patch.object(ProxyServer, "_start_cleanup_thread"):
+                server = ProxyServer()
+        server.redis_client = MagicMock()
+        server.redis_client.scard.return_value = 0
+        return server
+
+    @patch("apps.proxy.live_proxy.server.gevent.sleep")
+    @patch("apps.proxy.live_proxy.server.time.time", return_value=1000.0)
+    @patch("apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay", return_value=30)
+    def test_aborts_when_disconnect_key_deleted(self, _mock_delay, _mock_time, mock_sleep):
+        server = self._make_server()
+        server.redis_client.get.side_effect = [b"1000.0", None]
+
+        result = server._wait_for_shutdown_delay(CHANNEL_ID)
+
+        self.assertFalse(result)
+        self.assertGreaterEqual(mock_sleep.call_count, 1)
+
+    @patch("apps.proxy.live_proxy.server.gevent.sleep")
+    @patch("apps.proxy.live_proxy.server.time.time", return_value=1000.0)
+    @patch("apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay", return_value=30)
+    def test_aborts_when_clients_reconnect(self, _mock_delay, _mock_time, _mock_sleep):
+        server = self._make_server()
+        server.redis_client.get.return_value = b"1000.0"
+        server.redis_client.scard.side_effect = [0, 1]
+
+        result = server._wait_for_shutdown_delay(CHANNEL_ID)
+
+        self.assertFalse(result)
+        server.redis_client.delete.assert_called_with(
+            RedisKeys.last_client_disconnect(CHANNEL_ID)
+        )
+
+    @patch("apps.proxy.live_proxy.server.gevent.sleep")
+    @patch("apps.proxy.live_proxy.server.time.time")
+    @patch("apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay", return_value=30)
+    def test_timer_resets_when_disconnect_timestamp_updated(
+        self, _mock_delay, mock_time, mock_sleep
+    ):
+        server = self._make_server()
+        disconnect_key = RedisKeys.last_client_disconnect(CHANNEL_ID)
+        current_time = [1000.0]
+        disconnect_timestamp = [1000.0]
+        poll_count = [0]
+
+        mock_time.side_effect = lambda: current_time[0]
+
+        def get_side_effect(key):
+            poll_count[0] += 1
+            if poll_count[0] >= 3:
+                disconnect_timestamp[0] = 1020.0
+            if key == disconnect_key:
+                return str(disconnect_timestamp[0]).encode()
+            return None
+
+        server.redis_client.get.side_effect = get_side_effect
+
+        def advance_sleep(duration):
+            current_time[0] += duration
+
+        mock_sleep.side_effect = advance_sleep
+
+        result = server._wait_for_shutdown_delay(CHANNEL_ID)
+
+        self.assertTrue(result)
+        self.assertGreaterEqual(current_time[0], 1050.0)
+
+    @patch.object(ProxyServer, "_coordinated_stop_channel")
+    @patch.object(ProxyServer, "_wait_for_shutdown_delay", return_value=True)
+    @patch("apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay", return_value=30)
+    def test_handle_client_disconnect_uses_polling_wait(
+        self, _mock_delay, mock_wait, mock_coordinated
+    ):
+        server = HandleClientDisconnectUpstreamFirstTests()._make_server()
+        server.redis_client.get.return_value = b"1700000000.0"
+
+        server.handle_client_disconnect(CHANNEL_ID)
+
+        mock_wait.assert_called_once_with(CHANNEL_ID)
+        mock_coordinated.assert_called_once_with(CHANNEL_ID)
+
+    @patch.object(ProxyServer, "_coordinated_stop_channel")
+    @patch.object(ProxyServer, "_wait_for_shutdown_delay", return_value=False)
+    @patch("apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay", return_value=30)
+    def test_handle_client_disconnect_skips_stop_when_wait_aborted(
+        self, _mock_delay, _mock_wait, mock_coordinated
+    ):
+        server = HandleClientDisconnectUpstreamFirstTests()._make_server()
+        server.redis_client.get.return_value = b"1700000000.0"
+
+        server.handle_client_disconnect(CHANNEL_ID)
+
+        mock_coordinated.assert_not_called()
+
+
+class InitWaitAbortTests(TestCase):
+    def _make_generator(self):
+        from apps.proxy.live_proxy.output.ts.generator import StreamGenerator
+
+        return StreamGenerator(
+            CHANNEL_ID,
+            "client-1",
+            "127.0.0.1",
+            "test-agent",
+            channel_initializing=True,
+        )
+
+    def test_abort_when_client_removed_locally(self):
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = set()
+        server.client_managers = {CHANNEL_ID: client_manager}
+        server.redis_client = MagicMock()
+
+        self.assertEqual(generator._init_wait_abort_reason(server, time.time()), "client_gone")
+
+    def test_abort_when_connect_stalled_without_buffer(self):
+        from apps.proxy.config import TSConfig as Config
+
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = {"client-1"}
+        server.client_managers = {CHANNEL_ID: client_manager}
+        buffer = MagicMock()
+        buffer.index = 0
+        server.stream_buffers = {CHANNEL_ID: buffer}
+        server.redis_client = MagicMock()
+        server.redis_client.hget.return_value = b"connecting"
+        server.redis_client.get.return_value = None
+
+        started = time.time() - (getattr(Config, "CONNECTION_TIMEOUT", 10) + 1)
+        self.assertEqual(generator._init_wait_abort_reason(server, started), "stalled")
+
+
+class UpstreamStopBroadcastTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server._live_stream_managers = {}
+        server.redis_client = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_local_upstream_stops_locally_without_broadcast(
+        self, _mock_has, mock_broadcast, mock_stop_local
+    ):
+        server = self._make_server()
+        server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
+        mock_stop_local.assert_called_once_with(CHANNEL_ID)
+        mock_broadcast.assert_not_called()
+
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
+    def test_orphan_cleanup_broadcasts_when_no_local_upstream(
+        self, _mock_has, mock_broadcast, mock_stop_local
+    ):
+        server = self._make_server()
+        server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
+        mock_broadcast.assert_called_once_with(CHANNEL_ID)
+        mock_stop_local.assert_not_called()
+
+
+class StreamManagerStillOwnerTests(TestCase):
+    def _make_manager(self, redis_client):
+        buffer = MagicMock()
+        buffer.redis_client = redis_client
+        buffer.channel_id = CHANNEL_ID
+        manager = StreamManager(
+            CHANNEL_ID,
+            "http://example/stream.ts",
+            buffer,
+            worker_id="testhost:1",
+        )
+        return manager
+
+    def test_stops_when_metadata_removed(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(redis, metadata_exists=0)
+        manager = self._make_manager(redis)
+        self.assertFalse(manager._still_owner())
+
+    def test_keeps_running_during_connecting_before_client_registered(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            state=ChannelState.CONNECTING.encode(),
+        )
+        manager = self._make_manager(redis)
+        self.assertTrue(manager._still_owner())
+
+    def test_stops_after_disconnect_when_shutdown_delay_is_zero(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            disconnect=b"1700000000.0",
+            state=ChannelState.ACTIVE.encode(),
+        )
+        manager = self._make_manager(redis)
+        with patch(
+            "apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
+            return_value=0,
+        ):
+            self.assertFalse(manager._still_owner())
+
+    def test_keeps_running_during_shutdown_delay(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            disconnect=str(time.time()).encode(),
+            state=ChannelState.ACTIVE.encode(),
+        )
+        manager = self._make_manager(redis)
+        with patch(
+            "apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
+            return_value=5,
+        ):
+            self.assertTrue(manager._still_owner())

apps/epg/tasks.py

@@ -561,9 +561,10 @@ def refresh_epg_data(source_id, force=False):
                 gc.collect()
                 return
 
-            # Build byte-offset index for preview lookups in the background so refresh isn't blocked by it
-            build_programme_index_task.delay(source.id)
+            # Build byte-offset index after programme data is committed so refresh
+            # does not compete for memory/IO during the programme swap.
             parse_programs_for_source(source)
+            build_programme_index_task.delay(source.id)
 
         elif source.source_type == 'schedules_direct':
             fetch_schedules_direct(source, force=force)
@@ -590,6 +591,7 @@ def refresh_epg_data(source_id, force=False):
         source = None
         # Force garbage collection before releasing the lock
         gc.collect()
+        connection.close()
         lock_renewer.stop()
         release_task_lock('refresh_epg_data', source_id)
 
@@ -1804,6 +1806,163 @@ def parse_programs_for_tvg_id(epg_id, force=False):
         release_task_lock('parse_epg_programs', epg_id)
 
 
+_EPG_PROGRAM_STAGING_TABLE = 'epg_program_staging'
+# Parse batches bound Python memory during XML iterparse; swap batches bound each
+# DELETE/INSERT statement inside the single atomic swap transaction.
+_EPG_PARSE_BATCH_SIZE = 2500
+_EPG_SWAP_BATCH_SIZE = 5000
+
+
+def _epg_program_staging_supported():
+    return connection.vendor == 'postgresql'
+
+
+def _prepare_epg_program_staging_table():
+    """Create/truncate a session-scoped temp table for streaming EPG programme inserts."""
+    if not _epg_program_staging_supported():
+        return False
+
+    with connection.cursor() as cursor:
+        cursor.execute(
+            f"""
+            CREATE TEMP TABLE IF NOT EXISTS {_EPG_PROGRAM_STAGING_TABLE} (
+                epg_id bigint NOT NULL,
+                start_time timestamptz NOT NULL,
+                end_time timestamptz NOT NULL,
+                title varchar(255) NOT NULL,
+                sub_title text,
+                description text,
+                tvg_id varchar(255),
+                custom_properties jsonb
+            ) ON COMMIT PRESERVE ROWS
+            """
+        )
+        cursor.execute(f"TRUNCATE {_EPG_PROGRAM_STAGING_TABLE}")
+    return True
+
+
+def _clear_epg_program_staging_table():
+    if not _epg_program_staging_supported():
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(f"TRUNCATE {_EPG_PROGRAM_STAGING_TABLE}")
+
+
+def _flush_epg_program_staging_batch(programs_batch):
+    """Insert a batch of unsaved ProgramData rows into the session staging table."""
+    if not programs_batch or not _epg_program_staging_supported():
+        return
+
+    values_sql = []
+    params = []
+    for program in programs_batch:
+        values_sql.append("(%s, %s, %s, %s, %s, %s, %s, %s)")
+        custom_properties = program.custom_properties
+        if custom_properties is not None and not isinstance(custom_properties, str):
+            custom_properties = json.dumps(custom_properties)
+        params.extend([
+            program.epg_id,
+            program.start_time,
+            program.end_time,
+            program.title,
+            program.sub_title,
+            program.description,
+            program.tvg_id,
+            custom_properties,
+        ])
+
+    with connection.cursor() as cursor:
+        cursor.execute(
+            f"""
+            INSERT INTO {_EPG_PROGRAM_STAGING_TABLE} (
+                epg_id, start_time, end_time, title, sub_title, description, tvg_id, custom_properties
+            ) VALUES {', '.join(values_sql)}
+            """,
+            params,
+        )
+
+
+def _swap_staged_epg_programs(mapped_epg_ids, epg_source, batch_size=_EPG_SWAP_BATCH_SIZE):
+    """
+    Atomically replace mapped programme rows with staged data.
+    Must be called inside transaction.atomic().
+
+    Staged rows are moved in batches (DELETE ... RETURNING + INSERT) so Postgres
+    does not need to materialize the entire catalogue in one statement.
+    """
+    with connection.cursor() as cursor:
+        cursor.execute("SET LOCAL statement_timeout = '10min'")
+
+    deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
+    logger.debug(f"Deleted {deleted_count} existing programs")
+
+    unmapped_epg_ids = list(
+        EPGData.objects.filter(epg_source=epg_source)
+        .exclude(id__in=mapped_epg_ids)
+        .values_list('id', flat=True)
+    )
+    if unmapped_epg_ids:
+        orphaned_count = ProgramData.objects.filter(epg_id__in=unmapped_epg_ids).delete()[0]
+        if orphaned_count > 0:
+            logger.info(
+                f"Cleaned up {orphaned_count} orphaned programs for "
+                f"{len(unmapped_epg_ids)} unmapped EPG entries"
+            )
+
+    if not _epg_program_staging_supported():
+        raise RuntimeError('_swap_staged_epg_programs requires PostgreSQL staging support')
+
+    program_table = ProgramData._meta.db_table
+    total_inserted = 0
+    while True:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                WITH moved AS (
+                    DELETE FROM {_EPG_PROGRAM_STAGING_TABLE}
+                    WHERE ctid IN (
+                        SELECT ctid FROM {_EPG_PROGRAM_STAGING_TABLE} LIMIT %s
+                    )
+                    RETURNING
+                        epg_id, start_time, end_time, title, sub_title,
+                        description, tvg_id, custom_properties
+                )
+                INSERT INTO {program_table} (
+                    epg_id, start_time, end_time, title, sub_title,
+                    description, tvg_id, custom_properties
+                )
+                SELECT
+                    epg_id, start_time, end_time, title, sub_title,
+                    description, tvg_id, custom_properties
+                FROM moved
+                """,
+                [batch_size],
+            )
+            moved_count = cursor.rowcount
+        if moved_count == 0:
+            break
+        total_inserted += moved_count
+
+    logger.debug(f"Inserted {total_inserted} staged programs in batches of {batch_size}")
+
+    return deleted_count
+
+
+def _swap_parsed_epg_programs(mapped_epg_ids, epg_source, programs_to_create, batch_size=_EPG_SWAP_BATCH_SIZE):
+    """SQLite/dev fallback: atomic delete + bulk insert from an in-memory batch list."""
+    with transaction.atomic():
+        deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
+        unmapped_epg_ids = list(
+            EPGData.objects.filter(epg_source=epg_source)
+            .exclude(id__in=mapped_epg_ids)
+            .values_list('id', flat=True)
+        )
+        if unmapped_epg_ids:
+            ProgramData.objects.filter(epg_id__in=unmapped_epg_ids).delete()
+        for i in range(0, len(programs_to_create), batch_size):
+            ProgramData.objects.bulk_create(programs_to_create[i:i + batch_size])
+    return deleted_count
+
 
 def parse_programs_for_source(epg_source, tvg_id=None):
     """
@@ -1910,106 +2069,164 @@ def parse_programs_for_source(epg_source, tvg_id=None):
                 send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided")
                 return False
 
-        # SINGLE PASS PARSING: Parse the XML file once and collect all programs in memory
-        # We parse FIRST, then do an atomic delete+insert to avoid race conditions
-        # where clients might see empty/partial EPG data during the transition
-        all_programs_to_create = []
-        programs_by_channel = {tvg_id: 0 for tvg_id in mapped_tvg_ids}  # Track count per channel
+        # Stream parsed rows into a session temp table, then swap in a short transaction.
+        # This bounds Python memory (batched staging inserts) and Postgres memory (no
+        # long-lived transaction spanning the entire XML parse).
+        programs_by_channel = {tvg_id: 0 for tvg_id in mapped_tvg_ids}
         total_programs = 0
         skipped_programs = 0
         last_progress_update = 0
+        parse_batch_size = _EPG_PARSE_BATCH_SIZE
+        swap_batch_size = _EPG_SWAP_BATCH_SIZE
+        programs_batch = []
+        deleted_count = 0
+        staging_prepared = False
+        use_staging = False
+        programs_accumulator = []
+
+        send_epg_update(epg_source.id, "parsing_programs", 10, message="Parsing programs...")
 
         try:
-            logger.debug(f"Opening file for single-pass parsing: {file_path}")
+            staging_prepared = _prepare_epg_program_staging_table()
+            use_staging = staging_prepared
+
+            logger.debug(f"Opening file for streaming parse: {file_path}")
             source_file = _open_xmltv_file(file_path)
+            try:
+                program_parser = etree.iterparse(
+                    source_file,
+                    events=('end',),
+                    tag='programme',
+                    remove_blank_text=True,
+                    recover=True,
+                )
 
-            # Stream parse the file using lxml's iterparse
-            program_parser = etree.iterparse(source_file, events=('end',), tag='programme', remove_blank_text=True, recover=True)
+                for _, elem in program_parser:
+                    channel_id = elem.get('channel')
 
-            for _, elem in program_parser:
-                channel_id = elem.get('channel')
+                    if channel_id not in mapped_tvg_ids:
+                        skipped_programs += 1
+                        clear_element(elem)
+                        continue
 
-                # Skip programmes for unmapped channels immediately
-                if channel_id not in mapped_tvg_ids:
-                    skipped_programs += 1
-                    # Clear element to free memory
-                    clear_element(elem)
-                    continue
+                    try:
+                        start_time = parse_xmltv_time(elem.get('start'))
+                        end_time = parse_xmltv_time(elem.get('stop'))
+                        title = None
+                        desc = None
+                        sub_title = None
 
-                # This programme is for a mapped channel - process it
-                try:
-                    start_time = parse_xmltv_time(elem.get('start'))
-                    end_time = parse_xmltv_time(elem.get('stop'))
-                    title = None
-                    desc = None
-                    sub_title = None
+                        for child in elem:
+                            if child.tag == 'title':
+                                title = child.text or 'No Title'
+                            elif child.tag == 'desc':
+                                desc = child.text or ''
+                            elif child.tag == 'sub-title':
+                                sub_title = child.text or ''
 
-                    # Efficiently process child elements
-                    for child in elem:
-                        if child.tag == 'title':
-                            title = child.text or 'No Title'
-                        elif child.tag == 'desc':
-                            desc = child.text or ''
-                        elif child.tag == 'sub-title':
-                            sub_title = child.text or ''
+                        if not title:
+                            title = 'No Title'
 
-                    if not title:
-                        title = 'No Title'
+                        custom_props = extract_custom_properties(elem)
+                        custom_properties_json = custom_props if custom_props else None
 
-                    # Extract custom properties
-                    custom_props = extract_custom_properties(elem)
-                    custom_properties_json = custom_props if custom_props else None
+                        if desc:
+                            has_season = (custom_properties_json or {}).get('season') is not None
+                            has_episode = (custom_properties_json or {}).get('episode') is not None
+                            if not has_season or not has_episode:
+                                d_season, d_episode, cleaned_desc = extract_season_episode_from_description(desc)
+                                if d_season is not None and d_episode is not None:
+                                    if custom_properties_json is None:
+                                        custom_properties_json = {}
+                                    if not has_season:
+                                        custom_properties_json['season'] = d_season
+                                    if not has_episode:
+                                        custom_properties_json['episode'] = d_episode
+                                    custom_properties_json['season_episode_source'] = 'description'
+                                    desc = cleaned_desc
 
-                    # Fallback: extract S/E from description when episode-num
-                    # elements didn't provide them
-                    if desc:
-                        has_season = (custom_properties_json or {}).get('season') is not None
-                        has_episode = (custom_properties_json or {}).get('episode') is not None
-                        if not has_season or not has_episode:
-                            d_season, d_episode, cleaned_desc = extract_season_episode_from_description(desc)
-                            if d_season is not None and d_episode is not None:
-                                if custom_properties_json is None:
-                                    custom_properties_json = {}
-                                if not has_season:
-                                    custom_properties_json['season'] = d_season
-                                if not has_episode:
-                                    custom_properties_json['episode'] = d_episode
-                                custom_properties_json['season_episode_source'] = 'description'
-                                desc = cleaned_desc
+                        epg_id = tvg_id_to_epg_id[channel_id]
+                        programs_batch.append(ProgramData(
+                            epg_id=epg_id,
+                            start_time=start_time,
+                            end_time=end_time,
+                            title=title[:255],
+                            description=desc,
+                            sub_title=sub_title,
+                            tvg_id=channel_id,
+                            custom_properties=custom_properties_json,
+                        ))
+                        total_programs += 1
+                        programs_by_channel[channel_id] += 1
+                        clear_element(elem)
 
-                    epg_id = tvg_id_to_epg_id[channel_id]
-                    all_programs_to_create.append(ProgramData(
-                        epg_id=epg_id,
-                        start_time=start_time,
-                        end_time=end_time,
-                        title=title[:255],
-                        description=desc,
-                        sub_title=sub_title,
-                        tvg_id=channel_id,
-                        custom_properties=custom_properties_json
-                    ))
-                    total_programs += 1
-                    programs_by_channel[channel_id] += 1
+                        if len(programs_batch) >= parse_batch_size:
+                            if use_staging:
+                                _flush_epg_program_staging_batch(programs_batch)
+                                programs_batch = []
+                            else:
+                                programs_accumulator.extend(programs_batch)
+                                programs_batch = []
 
-                    # Clear the element to free memory
-                    clear_element(elem)
+                        if total_programs - last_progress_update >= 5000:
+                            last_progress_update = total_programs
+                            progress = min(
+                                85,
+                                10 + int((total_programs / max(total_programs + 10000, 1)) * 75),
+                            )
+                            send_epg_update(
+                                epg_source.id,
+                                "parsing_programs",
+                                progress,
+                                processed=total_programs,
+                                channels=mapped_count,
+                                message=f"Staging programs... {total_programs:,}",
+                            )
 
-                    # Send progress update (estimate based on programs processed)
-                    if total_programs - last_progress_update >= 5000:
-                        last_progress_update = total_programs
-                        # Cap at 70% during parsing phase (save 30% for DB operations)
-                        progress = min(70, 10 + int((total_programs / max(total_programs + 10000, 1)) * 60))
-                        send_epg_update(epg_source.id, "parsing_programs", progress,
-                                      processed=total_programs, channels=mapped_count)
+                        if total_programs % 5000 == 0:
+                            gc.collect()
 
-                    # Periodic garbage collection during parsing
-                    if total_programs % 5000 == 0:
-                        gc.collect()
+                    except Exception as e:
+                        logger.error(f"Error processing program for {channel_id}: {e}", exc_info=True)
+                        clear_element(elem)
+                        continue
 
-                except Exception as e:
-                    logger.error(f"Error processing program for {channel_id}: {e}", exc_info=True)
-                    clear_element(elem)
-                    continue
+                if programs_batch:
+                    if use_staging:
+                        _flush_epg_program_staging_batch(programs_batch)
+                    else:
+                        programs_accumulator.extend(programs_batch)
+                    programs_batch = []
+            finally:
+                if source_file:
+                    source_file.close()
+                    source_file = None
+
+            try:
+                send_epg_update(epg_source.id, "parsing_programs", 90, message="Updating database...")
+                if use_staging:
+                    with transaction.atomic():
+                        deleted_count = _swap_staged_epg_programs(
+                            mapped_epg_ids, epg_source, batch_size=swap_batch_size
+                        )
+                else:
+                    deleted_count = _swap_parsed_epg_programs(
+                        mapped_epg_ids, epg_source, programs_accumulator, batch_size=swap_batch_size
+                    )
+                    programs_accumulator = []
+
+                logger.info(
+                    f"Atomic update complete: deleted {deleted_count}, inserted {total_programs} programs"
+                )
+            except Exception as db_error:
+                logger.error(f"Database error during atomic update: {db_error}", exc_info=True)
+                epg_source.status = EPGSource.STATUS_ERROR
+                epg_source.last_message = f"Database error: {str(db_error)}"
+                epg_source.save(update_fields=['status', 'last_message'])
+                send_epg_update(
+                    epg_source.id, "parsing_programs", 100, status="error", message=str(db_error)
+                )
+                return False
 
         except etree.XMLSyntaxError as xml_error:
             logger.error(f"XML syntax error parsing program data: {xml_error}")
@@ -2018,63 +2235,23 @@ def parse_programs_for_source(epg_source, tvg_id=None):
             epg_source.save(update_fields=['status', 'last_message'])
             send_epg_update(epg_source.id, "parsing_programs", 100, status="error", message=str(xml_error))
             return False
-        except Exception as e:
-            logger.error(f"Error parsing XML for programs: {e}", exc_info=True)
-            raise
-        finally:
-            if source_file:
-                source_file.close()
-                source_file = None
-
-        # Now perform atomic delete + bulk insert
-        # This ensures clients never see empty/partial EPG data
-        logger.info(f"Parsed {total_programs} programs, performing atomic database update...")
-        send_epg_update(epg_source.id, "parsing_programs", 75, message="Updating database...")
-
-        batch_size = 1000
-        try:
-            with transaction.atomic():
-                # Kill any individual statement that hangs longer than 10 minutes.
-                # SET LOCAL automatically resets when this transaction ends (commit or rollback).
-                with connection.cursor() as cursor:
-                    cursor.execute("SET LOCAL statement_timeout = '10min'")
-                # Delete existing programs for mapped EPGs
-                deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
-                logger.debug(f"Deleted {deleted_count} existing programs")
-
-                # Clean up orphaned programs for unmapped EPG entries
-                unmapped_epg_ids = list(EPGData.objects.filter(
-                    epg_source=epg_source
-                ).exclude(id__in=mapped_epg_ids).values_list('id', flat=True))
-
-                if unmapped_epg_ids:
-                    orphaned_count = ProgramData.objects.filter(epg_id__in=unmapped_epg_ids).delete()[0]
-                    if orphaned_count > 0:
-                        logger.info(f"Cleaned up {orphaned_count} orphaned programs for {len(unmapped_epg_ids)} unmapped EPG entries")
-
-                # Bulk insert all new programs in batches within the same transaction
-                for i in range(0, len(all_programs_to_create), batch_size):
-                    batch = all_programs_to_create[i:i + batch_size]
-                    ProgramData.objects.bulk_create(batch)
-
-                    # Update progress during insertion
-                    progress = 75 + int((i / len(all_programs_to_create)) * 20) if all_programs_to_create else 95
-                    if i % (batch_size * 5) == 0:
-                        send_epg_update(epg_source.id, "parsing_programs", min(95, progress),
-                                      message=f"Inserting programs... {i}/{len(all_programs_to_create)}")
-
-            logger.info(f"Atomic update complete: deleted {deleted_count}, inserted {total_programs} programs")
-
-        except Exception as db_error:
-            logger.error(f"Database error during atomic update: {db_error}", exc_info=True)
+        except Exception as parse_error:
+            logger.error(f"Error parsing programs from XML: {parse_error}", exc_info=True)
             epg_source.status = EPGSource.STATUS_ERROR
-            epg_source.last_message = f"Database error: {str(db_error)}"
+            epg_source.last_message = f"Error parsing programs: {str(parse_error)}"
             epg_source.save(update_fields=['status', 'last_message'])
-            send_epg_update(epg_source.id, "parsing_programs", 100, status="error", message=str(db_error))
+            send_epg_update(
+                epg_source.id, "parsing_programs", 100, status="error", message=str(parse_error)
+            )
             return False
         finally:
-            # Clear the large list to free memory
-            all_programs_to_create = None
+            programs_batch = None
+            programs_accumulator = None
+            if staging_prepared:
+                try:
+                    _clear_epg_program_staging_table()
+                except Exception:
+                    pass
             gc.collect()
 
         # Count channels that actually got programs
@@ -2130,7 +2307,7 @@ def parse_programs_for_source(epg_source, tvg_id=None):
             source_file = None
 
         # Explicitly release any remaining large data structures
-        programs_to_create = None
+        programs_batch = None
         programs_by_channel = None
         mapped_epg_ids = None
         mapped_tvg_ids = None

apps/epg/tests/test_parse_programs_for_source.py

@@ -0,0 +1,238 @@
+import os
+import tempfile
+from datetime import timedelta
+from unittest.mock import patch
+
+from django.db import connection, transaction
+from django.test import TestCase
+from django.utils import timezone
+
+from apps.channels.models import Channel
+from apps.epg.models import EPGSource, EPGData, ProgramData
+from apps.epg.tasks import (
+    parse_programs_for_source,
+    _flush_epg_program_staging_batch,
+    _swap_staged_epg_programs,
+    _EPG_PARSE_BATCH_SIZE,
+)
+
+
+def _programme_xml(channel_id, title, start, stop):
+    return (
+        f'  <programme start="{start}" stop="{stop}" channel="{channel_id}">\n'
+        f'    <title>{title}</title>\n'
+        f'  </programme>\n'
+    )
+
+
+def _xmltv_file(programmes):
+    body = (
+        '<?xml version="1.0" encoding="UTF-8"?>\n'
+        '<tv generator-info-name="test">\n'
+        f'{programmes}'
+        '</tv>\n'
+    )
+    handle = tempfile.NamedTemporaryFile(
+        mode='w',
+        suffix='.xml',
+        delete=False,
+        encoding='utf-8',
+    )
+    handle.write(body)
+    handle.close()
+    return handle.name
+
+
+class ParseProgramsForSourceTests(TestCase):
+    def setUp(self):
+        self.source = EPGSource.objects.create(
+            name='XMLTV Parse Test',
+            source_type='xmltv',
+        )
+        self.mapped_epg = EPGData.objects.create(
+            epg_source=self.source,
+            tvg_id='mapped.channel',
+            name='Mapped Channel',
+        )
+        self.unmapped_epg = EPGData.objects.create(
+            epg_source=self.source,
+            tvg_id='unmapped.channel',
+            name='Unmapped Channel',
+        )
+        Channel.objects.create(
+            channel_number=1,
+            name='Mapped Channel',
+            epg_data=self.mapped_epg,
+        )
+        self.base_time = timezone.now().replace(minute=0, second=0, microsecond=0)
+        self.start = self.base_time.strftime('%Y%m%d%H%M%S +0000')
+        self.stop = (self.base_time + timedelta(hours=1)).strftime('%Y%m%d%H%M%S +0000')
+
+    def tearDown(self):
+        if getattr(self, 'xml_path', None) and os.path.exists(self.xml_path):
+            os.unlink(self.xml_path)
+
+    def _configure_source_file(self, programmes):
+        self.xml_path = _xmltv_file(programmes)
+        self.source.file_path = self.xml_path
+        self.source.save(update_fields=['file_path'])
+
+    @patch('apps.epg.tasks.log_system_event')
+    @patch('apps.epg.tasks.send_epg_update')
+    def test_replaces_programs_for_mapped_channels(self, _send_update, _log_event):
+        old_start = self.base_time - timedelta(days=1)
+        ProgramData.objects.create(
+            epg=self.mapped_epg,
+            start_time=old_start,
+            end_time=old_start + timedelta(hours=1),
+            title='Old Programme',
+            tvg_id=self.mapped_epg.tvg_id,
+        )
+        orphan_start = self.base_time - timedelta(days=1)
+        ProgramData.objects.create(
+            epg=self.unmapped_epg,
+            start_time=orphan_start,
+            end_time=orphan_start + timedelta(hours=1),
+            title='Orphan Programme',
+            tvg_id=self.unmapped_epg.tvg_id,
+        )
+
+        programmes = (
+            _programme_xml('mapped.channel', 'New Show', self.start, self.stop)
+            + _programme_xml('unmapped.channel', 'Skipped Show', self.start, self.stop)
+        )
+        self._configure_source_file(programmes)
+
+        result = parse_programs_for_source(self.source)
+
+        self.assertTrue(result)
+        mapped_programs = ProgramData.objects.filter(epg=self.mapped_epg)
+        self.assertEqual(mapped_programs.count(), 1)
+        self.assertEqual(mapped_programs.get().title, 'New Show')
+        self.assertFalse(ProgramData.objects.filter(epg=self.unmapped_epg).exists())
+
+    @patch('apps.epg.tasks.log_system_event')
+    @patch('apps.epg.tasks.send_epg_update')
+    def test_atomic_failure_rolls_back_and_preserves_existing_programs(self, _send_update, _log_event):
+        old_start = self.base_time - timedelta(days=1)
+        ProgramData.objects.create(
+            epg=self.mapped_epg,
+            start_time=old_start,
+            end_time=old_start + timedelta(hours=1),
+            title='Keep Me',
+            tvg_id=self.mapped_epg.tvg_id,
+        )
+
+        self._configure_source_file(
+            _programme_xml('mapped.channel', 'Replacement', self.start, self.stop)
+        )
+
+        swap_path = (
+            'apps.epg.tasks._swap_staged_epg_programs'
+            if connection.vendor == 'postgresql'
+            else 'apps.epg.tasks._swap_parsed_epg_programs'
+        )
+        with patch(swap_path, side_effect=RuntimeError('simulated insert failure')):
+            result = parse_programs_for_source(self.source)
+
+        self.assertFalse(result)
+        self.assertEqual(ProgramData.objects.filter(epg=self.mapped_epg).count(), 1)
+        self.assertEqual(
+            ProgramData.objects.get(epg=self.mapped_epg).title,
+            'Keep Me',
+        )
+
+    @patch('apps.epg.tasks.log_system_event')
+    @patch('apps.epg.tasks.send_epg_update')
+    def test_streams_batches_without_holding_full_program_list(self, _send_update, _log_event):
+        if connection.vendor != 'postgresql':
+            self.skipTest('PostgreSQL staging batches are required for this assertion')
+
+        programme_count = _EPG_PARSE_BATCH_SIZE * 2
+        programmes = ''.join(
+            _programme_xml(
+                'mapped.channel',
+                f'Show {idx}',
+                self.start,
+                self.stop,
+            )
+            for idx in range(programme_count)
+        )
+        self._configure_source_file(programmes)
+        flush_sizes = []
+        original_flush = _flush_epg_program_staging_batch
+
+        def tracking_flush(batch):
+            flush_sizes.append(len(batch))
+            return original_flush(batch)
+
+        with patch('apps.epg.tasks._flush_epg_program_staging_batch', side_effect=tracking_flush):
+            result = parse_programs_for_source(self.source)
+
+        self.assertTrue(result)
+        self.assertEqual(ProgramData.objects.filter(epg=self.mapped_epg).count(), programme_count)
+        self.assertEqual(sum(flush_sizes), programme_count)
+        self.assertTrue(all(size <= _EPG_PARSE_BATCH_SIZE for size in flush_sizes))
+        self.assertGreater(len(flush_sizes), 1)
+
+    @patch('apps.epg.tasks.log_system_event')
+    @patch('apps.epg.tasks.send_epg_update')
+    def test_live_programs_remain_until_swap_commits(self, _send_update, _log_event):
+        if connection.vendor != 'postgresql':
+            self.skipTest('PostgreSQL staging swap is required for this assertion')
+
+        old_start = self.base_time - timedelta(days=1)
+        ProgramData.objects.create(
+            epg=self.mapped_epg,
+            start_time=old_start,
+            end_time=old_start + timedelta(hours=1),
+            title='Old Programme',
+            tvg_id=self.mapped_epg.tvg_id,
+        )
+        self._configure_source_file(
+            _programme_xml('mapped.channel', 'New Show', self.start, self.stop)
+        )
+
+        observed_titles_at_swap = []
+
+        def swap_with_visibility_check(mapped_epg_ids, epg_source, *args, **kwargs):
+            observed_titles_at_swap.append(
+                ProgramData.objects.get(epg=self.mapped_epg).title
+            )
+            return _swap_staged_epg_programs(mapped_epg_ids, epg_source, *args, **kwargs)
+
+        with patch('apps.epg.tasks._swap_staged_epg_programs', side_effect=swap_with_visibility_check):
+            result = parse_programs_for_source(self.source)
+
+        self.assertTrue(result)
+        self.assertEqual(observed_titles_at_swap, ['Old Programme'])
+        self.assertEqual(ProgramData.objects.get(epg=self.mapped_epg).title, 'New Show')
+
+    @patch('apps.epg.tasks.log_system_event')
+    @patch('apps.epg.tasks.send_epg_update')
+    def test_swap_delete_is_rolled_back_when_insert_fails(self, _send_update, _log_event):
+        if connection.vendor != 'postgresql':
+            self.skipTest('PostgreSQL staging swap is required for this assertion')
+
+        old_start = self.base_time - timedelta(days=1)
+        ProgramData.objects.create(
+            epg=self.mapped_epg,
+            start_time=old_start,
+            end_time=old_start + timedelta(hours=1),
+            title='Keep Me',
+            tvg_id=self.mapped_epg.tvg_id,
+        )
+        self._configure_source_file(
+            _programme_xml('mapped.channel', 'Replacement', self.start, self.stop)
+        )
+
+        def failing_swap(mapped_epg_ids, epg_source, *args, **kwargs):
+            with transaction.atomic():
+                ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()
+                raise RuntimeError('simulated insert failure')
+
+        with patch('apps.epg.tasks._swap_staged_epg_programs', side_effect=failing_swap):
+            result = parse_programs_for_source(self.source)
+
+        self.assertFalse(result)
+        self.assertEqual(ProgramData.objects.get(epg=self.mapped_epg).title, 'Keep Me')

apps/plugins/loader.py

@@ -10,7 +10,7 @@ import types
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional
 
-from django.db import transaction
+from django.db import close_old_connections, transaction
 
 from .models import PluginConfig
 
@@ -492,79 +492,85 @@ class PluginManager:
         return cfg.settings
 
     def run_action(self, key: str, action_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-        lp = self.get_plugin(key)
-        if not lp or not lp.instance:
-            # Attempt a lightweight re-discovery in case the registry was rebuilt
-            self.discover_plugins(sync_db=False, force_reload=False, use_cache=False)
+        try:
             lp = self.get_plugin(key)
             if not lp or not lp.instance:
-                raise ValueError(f"Plugin '{key}' not found")
+                # Attempt a lightweight re-discovery in case the registry was rebuilt
+                self.discover_plugins(sync_db=False, force_reload=False, use_cache=False)
+                lp = self.get_plugin(key)
+                if not lp or not lp.instance:
+                    raise ValueError(f"Plugin '{key}' not found")
 
-        cfg = PluginConfig.objects.get(key=key)
-        if not cfg.enabled:
-            raise PermissionError(f"Plugin '{key}' is disabled")
-        params = params or {}
+            cfg = PluginConfig.objects.get(key=key)
+            if not cfg.enabled:
+                raise PermissionError(f"Plugin '{key}' is disabled")
+            params = params or {}
 
-        context = self._build_context(lp, cfg)
+            context = self._build_context(lp, cfg)
 
-        # Run either via Celery if plugin provides a delayed method, or inline
-        run_method = getattr(lp.instance, "run", None)
-        if not callable(run_method):
-            raise ValueError(f"Plugin '{key}' has no runnable 'run' method")
+            run_method = getattr(lp.instance, "run", None)
+            if not callable(run_method):
+                raise ValueError(f"Plugin '{key}' has no runnable 'run' method")
 
-        try:
-            result = run_method(action_id, params, context)
-        except Exception:
-            logger.exception(f"Plugin '{key}' action '{action_id}' failed")
-            raise
+            try:
+                result = run_method(action_id, params, context)
+            except Exception:
+                logger.exception(f"Plugin '{key}' action '{action_id}' failed")
+                raise
 
-        # Normalize return
-        if isinstance(result, dict):
-            return result
-        return {"status": "ok", "result": result}
+            if isinstance(result, dict):
+                return result
+            return {"status": "ok", "result": result}
+        finally:
+            # Return geventpool checkouts for this greenlet/thread after every action,
+            # including Connect event hooks and manual UI runs.
+            close_old_connections()
 
     def stop_plugin(self, key: str, reason: Optional[str] = None) -> bool:
-        lp = self.get_plugin(key)
-        if not lp or not lp.instance:
-            return False
         try:
-            cfg = PluginConfig.objects.get(key=key)
-        except PluginConfig.DoesNotExist:
-            return False
-        if not cfg.enabled:
-            return False
-
-        context = self._build_context(lp, cfg)
-        if reason:
-            context["reason"] = reason
-
-        stop_method = getattr(lp.instance, "stop", None)
-        if callable(stop_method):
+            lp = self.get_plugin(key)
+            if not lp or not lp.instance:
+                return False
             try:
-                stop_method(context)
-                return True
-            except TypeError:
+                cfg = PluginConfig.objects.get(key=key)
+            except PluginConfig.DoesNotExist:
+                return False
+            if not cfg.enabled:
+                return False
+
+            context = self._build_context(lp, cfg)
+            if reason:
+                context["reason"] = reason
+
+            stop_method = getattr(lp.instance, "stop", None)
+            if callable(stop_method):
                 try:
-                    stop_method()
+                    stop_method(context)
                     return True
+                except TypeError:
+                    try:
+                        stop_method()
+                        return True
+                    except Exception:
+                        logger.exception("Plugin '%s' stop() failed", key)
+                        return False
                 except Exception:
                     logger.exception("Plugin '%s' stop() failed", key)
                     return False
-            except Exception:
-                logger.exception("Plugin '%s' stop() failed", key)
-                return False
 
-        run_method = getattr(lp.instance, "run", None)
-        if callable(run_method):
-            actions = {a.get("id") for a in (lp.actions or []) if isinstance(a, dict)}
-            if "stop" in actions:
-                try:
-                    run_method("stop", {}, context)
-                    return True
-                except Exception:
-                    logger.exception("Plugin '%s' stop action failed", key)
-                    return False
-        return False
+            run_method = getattr(lp.instance, "run", None)
+            if callable(run_method):
+                actions = {a.get("id") for a in (lp.actions or []) if isinstance(a, dict)}
+                if "stop" in actions:
+                    try:
+                        run_method("stop", {}, context)
+                        return True
+                    except Exception:
+                        logger.exception("Plugin '%s' stop action failed", key)
+                        return False
+            return False
+        finally:
+            close_old_connections()
 
     def stop_all_plugins(self, reason: Optional[str] = None) -> int:
         stopped = 0

apps/plugins/tests/test_run_action_db_cleanup.py

@@ -0,0 +1,64 @@
+"""PluginManager must release geventpool checkouts after every run/stop."""
+
+from contextlib import contextmanager
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from apps.plugins.loader import LoadedPlugin, PluginManager
+
+
+class PluginRunActionDbCleanupTests(SimpleTestCase):
+    @contextmanager
+    def _manager_with_plugin(self, run_impl):
+        instance = MagicMock()
+        instance.run = run_impl
+        lp = LoadedPlugin(
+            key="test_plugin",
+            name="Test Plugin",
+            instance=instance,
+            actions=[{"id": "do_work"}],
+        )
+        pm = PluginManager()
+        cfg = MagicMock(enabled=True, settings={})
+        with patch.object(pm, "get_plugin", return_value=lp), patch(
+            "apps.plugins.loader.PluginConfig.objects.get", return_value=cfg
+        ):
+            yield pm
+
+    @patch("apps.plugins.loader.close_old_connections")
+    def test_run_action_closes_connections_on_success(self, mock_close):
+        with self._manager_with_plugin(lambda *_a, **_k: {"status": "ok"}) as pm:
+            result = pm.run_action("test_plugin", "do_work")
+
+        self.assertEqual(result, {"status": "ok"})
+        mock_close.assert_called_once()
+
+    @patch("apps.plugins.loader.close_old_connections")
+    def test_run_action_closes_connections_on_plugin_error(self, mock_close):
+        def _boom(*_a, **_k):
+            raise RuntimeError("plugin failed")
+
+        with self._manager_with_plugin(_boom) as pm:
+            with self.assertRaises(RuntimeError):
+                pm.run_action("test_plugin", "do_work")
+
+        mock_close.assert_called_once()
+
+    @patch("apps.plugins.loader.close_old_connections")
+    def test_stop_plugin_closes_connections(self, mock_close):
+        instance = MagicMock()
+        instance.stop = MagicMock()
+        lp = LoadedPlugin(
+            key="test_plugin",
+            name="Test Plugin",
+            instance=instance,
+        )
+        pm = PluginManager()
+        cfg = MagicMock(enabled=True, settings={})
+        with patch.object(pm, "get_plugin", return_value=lp), patch(
+            "apps.plugins.loader.PluginConfig.objects.get", return_value=cfg
+        ):
+            self.assertTrue(pm.stop_plugin("test_plugin", reason="shutdown"))
+
+        mock_close.assert_called_once()

apps/proxy/live_proxy/client_manager.py

@@ -102,6 +102,7 @@ class ClientManager:
                         break
 
                     # Send heartbeat for all local clients
+                    clients_to_remove = set()
                     with self.lock:
                         # Skip this cycle if we have no local clients
                         if not self.clients:
@@ -109,7 +110,6 @@ class ClientManager:
 
                         # IMPROVED GHOST DETECTION: Check for stale clients before sending heartbeats
                         current_time = time.time()
-                        clients_to_remove = set()
 
                         # First identify clients that should be removed
                         for client_id in self.clients:
@@ -132,12 +132,11 @@ class ClientManager:
                                     logger.debug(f"Client {client_id} inactive for {current_time - last_active_time:.1f}s, removing as ghost")
                                     clients_to_remove.add(client_id)
 
-                        # Remove ghost clients in a separate step
-                        for client_id in clients_to_remove:
-                            self.remove_client(client_id)
-
                         if clients_to_remove:
-                            logger.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}")
+                            logger.info(
+                                f"Identified {len(clients_to_remove)} ghost clients "
+                                f"for channel {self.channel_id}"
+                            )
 
                         # Now send heartbeats only for remaining clients
                         pipe = self.redis_client.pipeline()
@@ -172,6 +171,9 @@ class ClientManager:
                         if self.clients and not all(c in clients_to_remove for c in self.clients):
                             self._notify_owner_of_activity()
 
+                    for client_id in clients_to_remove:
+                        self.remove_client(client_id)
+
                 except Exception as e:
                     logger.error(f"Error in client heartbeat thread: {e}")
 
@@ -254,6 +256,14 @@ class ClientManager:
 
                 # Store in Redis
                 if self.redis_client:
+                    from .services.channel_service import ChannelService
+
+                    if ChannelService.cancel_pending_shutdown(self.channel_id):
+                        logger.info(
+                            f"Cancelled pending shutdown for channel {self.channel_id} "
+                            f"(client {client_id} reconnected)"
+                        )
+
                     self.redis_client.hset(client_key, mapping=client_data)
                     self.redis_client.expire(client_key, self.client_ttl)
 
@@ -300,11 +310,17 @@ class ClientManager:
                 return len(self.clients)
 
         except Exception as e:
-            logger.error(f"Error adding client {client_id}: {e}")
+            logger.error(f"Error adding client {client_id}: {e}", exc_info=True)
+            with self.lock:
+                self.clients.discard(client_id)
+            self._registered_clients.discard(client_id)
             return False
 
     def remove_client(self, client_id):
         """Remove a client from this channel and Redis"""
+        client_username = "unknown"
+        remaining = None
+
         with self.lock:
             if client_id in self.clients:
                 self.clients.remove(client_id)
@@ -315,61 +331,81 @@ class ClientManager:
             self.last_active_time = time.time()
 
             if self.redis_client:
-                # Get client data before removing the data
                 client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
                 client_username = self.redis_client.hget(client_key, "username") or "unknown"
                 if isinstance(client_username, bytes):
                     client_username = client_username.decode("utf-8")
 
-                # Remove from channel's client set
                 self.redis_client.srem(self.client_set_key, client_id)
-
-                client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
                 self.redis_client.delete(client_key)
-
-                # Check if this was the last client
                 remaining = self.redis_client.scard(self.client_set_key) or 0
-                if remaining == 0:
-                    logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
 
-                    # Trigger disconnect time tracking even if we're not the owner
-                    disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
-                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
+            local_count = len(self.clients)
 
-                self._notify_owner_of_activity()
+        if not self.redis_client:
+            return local_count
 
-                # Check if we're the owner - if so, handle locally; if not, publish event
-                am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
+        if remaining == 0:
+            logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
+            disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
+            ttl = max(int(ConfigHelper.channel_shutdown_delay() * 2), 60)
+            self.redis_client.setex(disconnect_key, ttl, str(time.time()))
 
-                if am_i_owner:
-                    # We're the owner - handle the disconnect directly
-                    logger.debug(f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)")
-                    if (remaining == 0
-                            or self.proxy_server.output_managers.get(self.channel_id)
-                            or self.proxy_server.profile_managers.get(self.channel_id)):
-                        import gevent
-                        gevent.spawn(self.proxy_server.handle_client_disconnect, self.channel_id)
-                else:
-                    # We're not the owner - publish event so owner can handle it
-                    logger.debug(f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} on channel {self.channel_id} from worker {self.worker_id}")
-                    event_data = json.dumps({
-                        "event": EventType.CLIENT_DISCONNECTED,
-                        "channel_id": self.channel_id,
-                        "client_id": client_id,
-                        "worker_id": self.worker_id or "unknown",
-                        "timestamp": time.time(),
-                        "remaining_clients": remaining,
-                        "username": client_username
-                    })
-                    self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
+        self._notify_owner_of_activity()
 
-                # Trigger channel stats update via WebSocket
-                self._trigger_stats_update()
+        am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
 
-            total_clients = self.get_total_client_count()
-            logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})")
+        # Owner lock TTL can expire while local ffmpeg is still running on this worker.
+        if (not am_i_owner and self.proxy_server
+                and self.proxy_server._has_local_upstream_activity(self.channel_id)):
+            if self.proxy_server.extend_ownership(self.channel_id):
+                am_i_owner = True
 
-        return len(self.clients)
+        schedule_disconnect = False
+        if am_i_owner:
+            logger.debug(
+                f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)"
+            )
+            if (remaining == 0
+                    or self.proxy_server.output_managers.get(self.channel_id)
+                    or self.proxy_server.profile_managers.get(self.channel_id)):
+                schedule_disconnect = True
+        elif (remaining == 0 and self.proxy_server
+              and self.proxy_server._has_local_upstream_activity(self.channel_id)):
+            logger.warning(
+                f"Last client removed while local upstream still active for "
+                f"{self.channel_id} without owner lock - stopping channel"
+            )
+            schedule_disconnect = True
+        else:
+            logger.debug(
+                f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} "
+                f"on channel {self.channel_id} from worker {self.worker_id}"
+            )
+            event_data = json.dumps({
+                "event": EventType.CLIENT_DISCONNECTED,
+                "channel_id": self.channel_id,
+                "client_id": client_id,
+                "worker_id": self.worker_id or "unknown",
+                "timestamp": time.time(),
+                "remaining_clients": remaining,
+                "username": client_username
+            })
+            self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
+
+        if schedule_disconnect and self.proxy_server:
+            self.proxy_server._spawn_on_hub(
+                self.proxy_server.handle_client_disconnect, self.channel_id
+            )
+
+        self._trigger_stats_update()
+
+        total_clients = self.get_total_client_count()
+        logger.info(
+            f"Client disconnected: {client_id} (local: {local_count}, total: {total_clients})"
+        )
+
+        return local_count
 
     def get_client_count(self):
         """Get local client count"""
@@ -442,3 +478,19 @@ class ClientManager:
             )
 
         return stale_ids
+
+    @staticmethod
+    def clear_all_clients(redis_client, channel_id):
+        """Remove every client entry from Redis for a channel."""
+        client_set_key = RedisKeys.clients(channel_id)
+        client_ids = redis_client.smembers(client_set_key)
+        if not client_ids:
+            return 0
+
+        pipe = redis_client.pipeline(transaction=False)
+        for cid in client_ids:
+            cid_str = cid.decode() if isinstance(cid, bytes) else cid
+            pipe.delete(RedisKeys.client_metadata(channel_id, cid_str))
+        pipe.delete(client_set_key)
+        pipe.execute()
+        return len(client_ids)

apps/proxy/live_proxy/input/buffer.py

@@ -64,7 +64,7 @@ class StreamBuffer:
 
     def add_chunk(self, chunk):
         """Add data with optimized Redis storage and TS packet alignment"""
-        if not chunk:
+        if not chunk or self.stopping:
             return False
 
         try:
@@ -309,38 +309,16 @@ class StreamBuffer:
         self.fill_timers.clear()
 
         try:
-            # Flush any remaining data in the write buffer
-            if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
-                # Ensure remaining data is aligned to TS packets
-                complete_size = (len(self._write_buffer) // 188) * 188
-
-                if complete_size > 0:
-                    final_chunk = self._write_buffer[:complete_size]
-
-                    # Write final chunk to Redis
-                    with self.lock:
-                        if self.redis_client:
-                            try:
-                                chunk_index = self.redis_client.incr(self.buffer_index_key)
-                                chunk_key = f"{self.buffer_prefix}{chunk_index}"
-                                self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(final_chunk))
-                                self.index = chunk_index
-                                logger.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis")
-                            except Exception as e:
-                                logger.error(f"Error flushing final chunk: {e}")
-
-                # Clear buffers
-                self._write_buffer = bytearray()
-                if hasattr(self, '_partial_packet'):
-                    self._partial_packet = bytearray()
-
-            # Clean up the chunk timestamps sorted set
-            if self.redis_client and self.chunk_timestamps_key:
-                try:
-                    self.redis_client.delete(self.chunk_timestamps_key)
-                except Exception as e:
-                    logger.error(f"Error deleting chunk timestamps key: {e}")
-
+            with self.lock:
+                if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
+                    discarded = len(self._write_buffer)
+                    self._write_buffer = bytearray()
+                    if hasattr(self, '_partial_packet'):
+                        self._partial_packet = bytearray()
+                    logger.debug(
+                        f"Discarded {discarded} bytes from local write buffer "
+                        f"for channel {self.channel_id}"
+                    )
         except Exception as e:
             logger.error(f"Error during buffer stop: {e}")
 

apps/proxy/live_proxy/input/manager.py

@@ -67,6 +67,7 @@ class StreamManager:
         # Add to your __init__ method
         self._buffer_check_timers = []
         self.stopping = False
+        self.stop_requested = False
 
         # Add tracking for tried streams and current stream
         self.current_stream_id = stream_id
@@ -112,6 +113,11 @@ class StreamManager:
         # Add tracking for data throughput
         self.bytes_processed = 0
         self.last_bytes_update = time.time()
+
+        # Cached result of the pipelined Redis ownership audit (hot read path).
+        self._ownership_cache_valid_until = 0.0
+        self._ownership_cached = True
+        self._OWNERSHIP_CHECK_INTERVAL = 1.0
         self.bytes_update_interval = 5  # Update Redis every 5 seconds
 
         # Add stderr reader thread property
@@ -182,6 +188,153 @@ class StreamManager:
         logger.warning(f"Timeout waiting for existing processes to close for channel {self.channel_id} after {timeout}s")
         return False
 
+    def _invalidate_ownership_cache(self):
+        self._ownership_cache_valid_until = 0.0
+
+    @staticmethod
+    def _decode_redis_value(value):
+        if value is None:
+            return None
+        if isinstance(value, bytes):
+            return value.decode()
+        return value
+
+    def _disconnect_shutdown_ready(self, disconnect_value):
+        """True when last-client disconnect has passed the configured shutdown delay."""
+        if not disconnect_value:
+            return False
+
+        shutdown_delay = ConfigHelper.channel_shutdown_delay()
+        if shutdown_delay <= 0:
+            return True
+
+        disconnect_value = self._decode_redis_value(disconnect_value)
+        try:
+            disconnect_time = float(disconnect_value)
+        except (ValueError, TypeError):
+            return False
+        return (time.time() - disconnect_time) >= shutdown_delay
+
+    def _evaluate_ownership_from_redis(self, redis_client):
+        """Single pipelined Redis round-trip for the full ownership audit."""
+        stop_key = RedisKeys.channel_stopping(self.channel_id)
+        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+        clients_key = RedisKeys.clients(self.channel_id)
+        owner_key = RedisKeys.channel_owner(self.channel_id)
+        disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
+
+        pipe = redis_client.pipeline(transaction=False)
+        pipe.exists(stop_key)
+        pipe.exists(metadata_key)
+        pipe.scard(clients_key)
+        pipe.get(owner_key)
+        pipe.get(disconnect_key)
+        pipe.hget(metadata_key, ChannelMetadataField.STATE)
+        (
+            stop_exists,
+            metadata_exists,
+            client_count,
+            current_owner,
+            disconnect_value,
+            state_raw,
+        ) = pipe.execute()
+
+        if stop_exists:
+            return False
+
+        if not metadata_exists:
+            logger.warning(
+                f"Channel {self.channel_id} metadata removed from Redis - stopping upstream"
+            )
+            return False
+
+        client_count = client_count or 0
+        state = self._decode_redis_value(state_raw)
+        current_owner = self._decode_redis_value(current_owner)
+
+        if current_owner and current_owner != self.worker_id:
+            return False
+
+        if not current_owner:
+            if client_count == 0 and state not in ChannelState.PRE_ACTIVE:
+                logger.warning(
+                    f"Channel {self.channel_id} has no owner and no clients - stopping upstream"
+                )
+                return False
+            return True
+
+        if client_count == 0 and self._disconnect_shutdown_ready(disconnect_value):
+            logger.info(
+                f"Channel {self.channel_id} disconnect shutdown ready - stopping upstream"
+            )
+            return False
+
+        return True
+
+    def _still_owner(self, *, force=False):
+        """Return True while this worker should keep the upstream connection open."""
+        if self.stopping or self.stop_requested:
+            return False
+
+        if not self.worker_id:
+            return True
+
+        redis_client = getattr(self.buffer, 'redis_client', None)
+        if not redis_client:
+            return True
+
+        now = time.time()
+        if not force and now < self._ownership_cache_valid_until:
+            return self._ownership_cached
+
+        try:
+            result = self._evaluate_ownership_from_redis(redis_client)
+            self._ownership_cached = result
+            self._ownership_cache_valid_until = now + self._OWNERSHIP_CHECK_INTERVAL
+            return result
+        except Exception as e:
+            logger.debug(f"Ownership check failed for channel {self.channel_id}: {e}")
+            return True
+
+    def _upstream_may_continue(self):
+        """
+        Per-chunk gate for the hot read path.
+
+        Local stop flags are checked every chunk. The coordinated-teardown Redis
+        flag is checked every chunk (one EXISTS). The full ownership audit is
+        pipelined and cached for ~1s — enough for loop boundaries while avoiding
+        6+ Redis round-trips per chunk during steady streaming.
+        """
+        if self.stopping or self.stop_requested or not self.running:
+            return False
+        if self.buffer is not None and self.buffer.stopping:
+            return False
+
+        redis_client = getattr(self.buffer, 'redis_client', None)
+        if redis_client and self.worker_id:
+            try:
+                if redis_client.exists(RedisKeys.channel_stopping(self.channel_id)):
+                    self._ownership_cached = False
+                    self._ownership_cache_valid_until = 0.0
+                    return False
+            except Exception as e:
+                logger.debug(
+                    f"Channel stopping check failed for {self.channel_id}: {e}"
+                )
+
+        return self._still_owner()
+
+    def _ensure_owner_or_stop(self):
+        if self._still_owner(force=True):
+            return True
+
+        logger.warning(
+            f"Stream manager for channel {self.channel_id} lost ownership "
+            f"(worker {self.worker_id}) - stopping upstream"
+        )
+        self.stop()
+        return False
+
     def run(self):
         """Main execution loop using HTTP streaming with improved connection handling and stream switching"""
         # Add a stop flag to the class properties
@@ -203,6 +356,8 @@ class StreamManager:
             # Main stream switching loop - we'll try different streams if needed
             while self.running and stream_switch_attempts <= max_stream_switches:
                 close_old_connections()
+                if not self._ensure_owner_or_stop():
+                    break
                 # Check for stuck switching state
                 if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout:
                     logger.warning(f"URL switching state appears stuck for channel {self.channel_id} "
@@ -255,6 +410,8 @@ class StreamManager:
                     continue
                 # Connection retry loop for current URL
                 while self.running and self.retry_count < self.max_retries and not url_failed and not self.needs_stream_switch:
+                    if not self._ensure_owner_or_stop():
+                        break
 
                     logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url} for channel {self.channel_id}")
 
@@ -382,6 +539,12 @@ class StreamManager:
         except Exception as e:
             logger.error(f"Stream error: {e}", exc_info=True)
         finally:
+            try:
+                from ..server import ProxyServer
+                ProxyServer.get_instance()._live_stream_managers.pop(self.channel_id, None)
+            except Exception:
+                pass
+
             # Enhanced cleanup in the finally block
             self.connected = False
 
@@ -410,7 +573,9 @@ class StreamManager:
                 try:
                     metadata_key = RedisKeys.channel_metadata(self.channel_id)
                     owner_key = RedisKeys.channel_owner(self.channel_id)
-                    current_owner = self.buffer.redis_client.get(owner_key)
+                    current_owner = self._decode_redis_value(
+                        self.buffer.redis_client.get(owner_key)
+                    )
 
                     is_owner = (
                         current_owner
@@ -421,12 +586,10 @@ class StreamManager:
 
                     should_update = is_owner
                     if not should_update and no_owner:
-                        current_state_bytes = self.buffer.redis_client.hget(
-                            metadata_key, ChannelMetadataField.STATE
-                        )
-                        current_state = (
-                            current_state_bytes
-                            if current_state_bytes else None
+                        current_state = self._decode_redis_value(
+                            self.buffer.redis_client.hget(
+                                metadata_key, ChannelMetadataField.STATE
+                            )
                         )
                         should_update = current_state in ChannelState.PRE_ACTIVE
                         if not should_update and current_state:
@@ -488,44 +651,48 @@ class StreamManager:
                 logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}")
                 self._close_connection()
 
-            channel = get_stream_object(self.channel_id)
+            try:
+                channel = get_stream_object(self.channel_id)
 
-            # Use FFmpeg specifically for HLS streams
-            if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg:
-                from core.models import StreamProfile
-                try:
-                    stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True)
-                    logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)")
-                except StreamProfile.DoesNotExist:
-                    # Fall back to channel's profile if FFmpeg not found
+                # Use FFmpeg specifically for HLS streams
+                if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg:
+                    from core.models import StreamProfile
+                    try:
+                        stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True)
+                        logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)")
+                    except StreamProfile.DoesNotExist:
+                        # Fall back to channel's profile if FFmpeg not found
+                        stream_profile = channel.get_stream_profile()
+                        logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}")
+                else:
                     stream_profile = channel.get_stream_profile()
-                    logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}")
-            else:
-                stream_profile = channel.get_stream_profile()
 
-            # Build and start transcode command
-            self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent)
+                # Build and start transcode command
+                self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent)
 
-            # Store stream command for efficient log parser routing
-            self.stream_command = stream_profile.command
-            # Map actual commands to parser types for direct routing
-            command_to_parser = {
-                'ffmpeg': 'ffmpeg',
-                'cvlc': 'vlc',
-                'vlc': 'vlc',
-                'streamlink': 'streamlink'
-            }
-            self.parser_type = command_to_parser.get(self.stream_command.lower())
-            if self.parser_type:
-                logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})")
-            else:
-                logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing")
+                # Store stream command for efficient log parser routing
+                self.stream_command = stream_profile.command
+                # Map actual commands to parser types for direct routing
+                command_to_parser = {
+                    'ffmpeg': 'ffmpeg',
+                    'cvlc': 'vlc',
+                    'vlc': 'vlc',
+                    'streamlink': 'streamlink'
+                }
+                self.parser_type = command_to_parser.get(self.stream_command.lower())
+                if self.parser_type:
+                    logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})")
+                else:
+                    logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing")
 
-            # For UDP streams, remove any user_agent parameters from the command
-            if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP:
-                # Filter out any arguments that contain the user_agent value or related headers
-                self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()]
-                logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}")
+                # For UDP streams, remove any user_agent parameters from the command
+                if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP:
+                    # Filter out any arguments that contain the user_agent value or related headers
+                    self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()]
+                    logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}")
+            finally:
+                # Release the pool slot before posix_spawn or before returning on profile errors.
+                close_old_connections()
 
             logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}")
 
@@ -1020,6 +1187,9 @@ class StreamManager:
 
     def _update_bytes_processed(self, chunk_size):
         """Update the total bytes processed in Redis metadata"""
+        if not self._upstream_may_continue():
+            return
+
         try:
             # Update local counter
             self.bytes_processed += chunk_size
@@ -1092,14 +1262,10 @@ class StreamManager:
         """Stop the stream manager and cancel all timers"""
         logger.info(f"Stopping stream manager for channel {self.channel_id}")
 
-        # Add at the beginning of your stop method
         self.stopping = True
-
-        # Release stream resources if we're the owner
-        if self.current_stream_id and hasattr(self, 'worker_id') and self.worker_id:
-            if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
-                owner_key = RedisKeys.channel_owner(self.channel_id)
-                current_owner = self.buffer.redis_client.get(owner_key)
+        self._invalidate_ownership_cache()
+        if self.buffer is not None:
+            self.buffer.stopping = True
 
         # Cancel all buffer check timers
         for timer in list(self._buffer_check_timers):
@@ -1489,7 +1655,8 @@ class StreamManager:
             if hasattr(self, 'stderr_reader_thread') and self.stderr_reader_thread and self.stderr_reader_thread.is_alive():
                 try:
                     logger.debug(f"Waiting for stderr reader thread to terminate for channel {self.channel_id}")
-                    self.stderr_reader_thread.join(timeout=2.0)
+                    stderr_join_timeout = 0.25 if self.stopping else 2.0
+                    self.stderr_reader_thread.join(timeout=stderr_join_timeout)
                     if self.stderr_reader_thread.is_alive():
                         logger.warning(f"Stderr reader thread did not terminate within timeout for channel {self.channel_id}")
                 except Exception as e:
@@ -1582,6 +1749,10 @@ class StreamManager:
                 self.connected = False
                 return False
 
+            if not self._upstream_may_continue():
+                self.stop()
+                return False
+
             # Track chunk size before adding to buffer
             chunk_size = len(chunk)
             self._update_bytes_processed(chunk_size)
@@ -1589,7 +1760,6 @@ class StreamManager:
             # Add directly to buffer without TS-specific processing
             success = self.buffer.add_chunk(chunk)
 
-            # Update last data timestamp in Redis if successful
             if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
                 last_data_key = RedisKeys.last_data(self.buffer.channel_id)
                 self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60)

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

@@ -11,6 +11,7 @@ import time
 import gevent
 from apps.channels.models import Channel, Stream
 from core.utils import log_system_event
+from django.db import close_old_connections
 from ...server import ProxyServer
 from ...redis_keys import RedisKeys
 from ...constants import ChannelMetadataField
@@ -331,46 +332,53 @@ class FMP4StreamGenerator:
     # ------------------------------------------------------------------
 
     def _cleanup(self):
-        elapsed = time.time() - self.stream_start_time
-        proxy_server = ProxyServer.get_instance()
+        try:
+            elapsed = time.time() - self.stream_start_time
+            proxy_server = ProxyServer.get_instance()
 
-        # Release stream allocation if last client (mirrors StreamGenerator)
-        if proxy_server.redis_client:
-            try:
-                meta_key = RedisKeys.channel_metadata(self.channel_id)
-                stream_id_bytes = proxy_server.redis_client.hget(
-                    meta_key, ChannelMetadataField.STREAM_ID
-                )
-                if stream_id_bytes:
-                    if self.channel_id in proxy_server.client_managers:
-                        client_count = proxy_server.client_managers[
-                            self.channel_id
-                        ].get_total_client_count()
-                        if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
-                            try:
+            # Release stream allocation if last client (mirrors StreamGenerator)
+            if proxy_server.redis_client:
+                try:
+                    meta_key = RedisKeys.channel_metadata(self.channel_id)
+                    stream_id_bytes = proxy_server.redis_client.hget(
+                        meta_key, ChannelMetadataField.STREAM_ID
+                    )
+                    if stream_id_bytes:
+                        if self.channel_id in proxy_server.client_managers:
+                            client_count = proxy_server.client_managers[
+                                self.channel_id
+                            ].get_total_client_count()
+                            if (
+                                client_count <= 1
+                                and proxy_server.am_i_owner(self.channel_id)
+                                and ConfigHelper.channel_shutdown_delay() <= 0
+                            ):
                                 try:
-                                    obj = Channel.objects.get(uuid=self.channel_id)
-                                except (Channel.DoesNotExist, Exception):
-                                    obj = Stream.objects.get(stream_hash=self.channel_id)
-                                obj.release_stream()
-                            except Exception as e:
-                                logger.error(
-                                    f"[{self.client_id}] Error releasing stream: {e}"
-                                )
-            except Exception as e:
-                logger.error(f"[{self.client_id}] Error in stream release check: {e}")
+                                    try:
+                                        obj = Channel.objects.get(uuid=self.channel_id)
+                                    except (Channel.DoesNotExist, Exception):
+                                        obj = Stream.objects.get(stream_hash=self.channel_id)
+                                    obj.release_stream()
+                                except Exception as e:
+                                    logger.error(
+                                        f"[{self.client_id}] Error releasing stream: {e}"
+                                    )
+                except Exception as e:
+                    logger.error(f"[{self.client_id}] Error in stream release check: {e}")
 
-        # Remove from MAIN ClientManager - this is what triggers handle_client_disconnect
-        # and the zero-clients → stop_channel path, same as TS clients.
-        local_clients = 0
-        total_clients = 0
-        if self.channel_id in proxy_server.client_managers:
-            client_manager = proxy_server.client_managers[self.channel_id]
-            local_clients = client_manager.remove_client(self.client_id)
-            total_clients = client_manager.get_total_client_count()
+            # Remove from MAIN ClientManager - this is what triggers handle_client_disconnect
+            # and the zero-clients → stop_channel path, same as TS clients.
+            local_clients = 0
+            total_clients = 0
+            if self.channel_id in proxy_server.client_managers:
+                client_manager = proxy_server.client_managers[self.channel_id]
+                local_clients = client_manager.remove_client(self.client_id)
+                total_clients = client_manager.get_total_client_count()
 
-        logger.info(
-            f"[{self.client_id}] fMP4 client disconnected after {elapsed:.2f}s "
-            f"({self.bytes_sent / 1024:.1f} KB sent, "
-            f"local: {local_clients}, total: {total_clients})"
-        )
+            logger.info(
+                f"[{self.client_id}] fMP4 client disconnected after {elapsed:.2f}s "
+                f"({self.bytes_sent / 1024:.1f} KB sent, "
+                f"local: {local_clients}, total: {total_clients})"
+            )
+        finally:
+            close_old_connections()

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

@@ -139,12 +139,75 @@ class StreamGenerator:
         finally:
             self._cleanup()
 
+    def _init_wait_abort_reason(self, proxy_server, initialization_start):
+        """Return a reason string when init wait should end early, else None."""
+        from ...services.channel_service import ChannelService
+
+        if ChannelService.is_channel_teardown_active(self.channel_id):
+            return "stopping"
+
+        client_mgr = proxy_server.client_managers.get(self.channel_id)
+        if client_mgr is not None:
+            if self.client_id not in client_mgr.clients:
+                return "client_gone"
+        elif proxy_server.redis_client:
+            client_key = RedisKeys.client_metadata(self.channel_id, self.client_id)
+            if not proxy_server.redis_client.exists(client_key):
+                total = proxy_server.redis_client.scard(RedisKeys.clients(self.channel_id)) or 0
+                if total == 0:
+                    return "client_gone"
+
+        elapsed = time.time() - initialization_start
+        connection_timeout = getattr(Config, 'CONNECTION_TIMEOUT', 10)
+        if elapsed < connection_timeout or not proxy_server.redis_client:
+            return None
+
+        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+        state_raw = proxy_server.redis_client.hget(metadata_key, 'state')
+        if not state_raw:
+            return None
+
+        state = state_raw.decode() if isinstance(state_raw, bytes) else state_raw
+        if state not in ('connecting', 'initializing', 'buffering'):
+            return None
+
+        buffer = proxy_server.stream_buffers.get(self.channel_id)
+        if buffer is not None and buffer.index > 0:
+            return None
+
+        if proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)):
+            return None
+
+        return "stalled"
+
     def _wait_for_initialization(self):
         initialization_start = time.time()
         max_init_wait = ConfigHelper.client_wait_timeout()
         proxy_server = ProxyServer.get_instance()
 
         while time.time() - initialization_start < max_init_wait:
+            abort_reason = self._init_wait_abort_reason(proxy_server, initialization_start)
+            if abort_reason == "stopping":
+                logger.error(
+                    f"[{self.client_id}] Channel {self.channel_id} teardown active during initialization"
+                )
+                yield create_ts_packet('error', "Error: Channel is stopping")
+                return False
+            if abort_reason == "client_gone":
+                logger.info(
+                    f"[{self.client_id}] Client disconnected during initialization wait, aborting"
+                )
+                yield create_ts_packet('error', "Error: Client disconnected")
+                return False
+            if abort_reason == "stalled":
+                logger.warning(
+                    f"[{self.client_id}] Channel {self.channel_id} stalled in connecting state "
+                    f"with no buffer data after "
+                    f"{getattr(Config, 'CONNECTION_TIMEOUT', 10)}s, aborting init wait"
+                )
+                yield create_ts_packet('error', "Error: Connection stalled")
+                return False
+
             if proxy_server.redis_client:
                 metadata_key = RedisKeys.channel_metadata(self.channel_id)
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
@@ -537,35 +600,32 @@ class StreamGenerator:
         stream_released = False
         if proxy_server.redis_client:
             try:
-                metadata_key = RedisKeys.channel_metadata(self.channel_id)
-                metadata = proxy_server.redis_client.hgetall(metadata_key)
-                if metadata:
-                    stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
-                    if stream_id_bytes:
-                        # Check if we're the last client
-                        if self.channel_id in proxy_server.client_managers:
-                            client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
-                            # Only the last client or owner should release the stream
-                            if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
-                                try:
-                                    # Try Channel first (normal flow), fall back to Stream (preview flow)
-                                    try:
-                                        obj = Channel.objects.get(uuid=self.channel_id)
-                                    except (Channel.DoesNotExist, Exception):
-                                        obj = Stream.objects.get(stream_hash=self.channel_id)
-                                    stream_released = obj.release_stream()
-                                    if stream_released:
-                                        logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
-                                    else:
-                                        logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
-                                except Exception as e:
-                                    logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
+                if self.channel_id in proxy_server.client_managers:
+                    client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
+                    # Pool slots are global; the last client on any worker must release.
+                    # During shutdown_delay, keep the slot until coordinated stop runs.
+                    if client_count <= 1 and ConfigHelper.channel_shutdown_delay() <= 0:
+                        try:
+                            try:
+                                obj = Channel.objects.get(uuid=self.channel_id)
+                            except (Channel.DoesNotExist, Exception):
+                                obj = Stream.objects.get(stream_hash=self.channel_id)
+                            stream_released = obj.release_stream()
+                            if stream_released:
+                                logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
+                            else:
+                                logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
+                        except Exception as e:
+                            logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
             except Exception as e:
                 logger.error(f"[{self.client_id}] Error checking stream data for release: {e}")
 
         if self.channel_id in proxy_server.client_managers:
             client_manager = proxy_server.client_managers[self.channel_id]
-            local_clients = client_manager.remove_client(self.client_id)
+            if self.client_id in client_manager.clients:
+                local_clients = client_manager.remove_client(self.client_id)
+            else:
+                local_clients = client_manager.get_client_count()
             total_clients = client_manager.get_total_client_count()
             logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})")
 

apps/proxy/live_proxy/services/channel_service.py

@@ -10,6 +10,7 @@ from apps.channels.models import Channel, Stream
 from ..server import ProxyServer
 from ..redis_keys import RedisKeys
 from ..constants import EventType, ChannelState, ChannelMetadataField
+from ..config_helper import ConfigHelper
 from ..url_utils import get_stream_info_for_switch
 from core.utils import log_system_event
 from .log_parsers import LogParserFactory
@@ -19,6 +20,178 @@ logger = logging.getLogger("live_proxy")
 class ChannelService:
     """Service class for channel operations"""
 
+    @staticmethod
+    def mark_channel_stopping(channel_id, broadcast=False):
+        """Mark a channel as stopping in Redis so all uWSGI workers converge on teardown."""
+        proxy_server = ProxyServer.get_instance()
+        if not proxy_server.redis_client:
+            return False
+
+        try:
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            if proxy_server.redis_client.exists(metadata_key):
+                proxy_server.redis_client.hset(metadata_key, mapping={
+                    ChannelMetadataField.STATE: ChannelState.STOPPING,
+                    ChannelMetadataField.STATE_CHANGED_AT: str(time.time()),
+                })
+
+            stop_key = RedisKeys.channel_stopping(channel_id)
+            proxy_server.redis_client.setex(stop_key, 60, "true")
+
+            if broadcast:
+                ChannelService._publish_channel_stop_event(channel_id)
+            return True
+        except Exception as e:
+            logger.error(f"Error marking channel {channel_id} as stopping: {e}")
+            return False
+
+    @staticmethod
+    def is_shutdown_pending(channel_id):
+        """True while the post-disconnect shutdown delay has started but stop has not run."""
+        proxy_server = ProxyServer.get_instance()
+        if not proxy_server.redis_client:
+            return False
+
+        delay = ConfigHelper.channel_shutdown_delay()
+        if delay <= 0:
+            return False
+
+        disconnect_value = proxy_server.redis_client.get(
+            RedisKeys.last_client_disconnect(channel_id)
+        )
+        if not disconnect_value:
+            return False
+
+        try:
+            disconnect_time = float(disconnect_value)
+        except (ValueError, TypeError):
+            return False
+
+        return (time.time() - disconnect_time) < delay
+
+    @staticmethod
+    def is_channel_teardown_active(channel_id):
+        """True when a coordinated channel stop is in progress (visible to all workers)."""
+        proxy_server = ProxyServer.get_instance()
+        if channel_id in proxy_server._stopping_channels:
+            return True
+
+        if not proxy_server.redis_client:
+            return False
+
+        if proxy_server.redis_client.exists(RedisKeys.channel_stopping(channel_id)):
+            return True
+
+        metadata_key = RedisKeys.channel_metadata(channel_id)
+        state = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STATE)
+        if state:
+            state_str = state.decode() if isinstance(state, bytes) else state
+            if state_str == ChannelState.STOPPING:
+                return True
+
+        return False
+
+    @staticmethod
+    def is_channel_unavailable_for_new_clients(channel_id):
+        """Reject new stream requests only while coordinated teardown is active."""
+        return ChannelService.is_channel_teardown_active(channel_id)
+
+    @staticmethod
+    def cancel_pending_shutdown(channel_id):
+        """
+        Abort the post-disconnect grace timer when a client reconnects.
+
+        Clears the disconnect timestamp and any leaked stopping markers. When
+        upstream is still active but the last client released its profile slot
+        during the grace window, re-reserve the slot from Redis metadata.
+
+        Does not run during coordinated stop_channel() — clearing teardown
+        markers mid-stop would leave clients attached to upstream that is
+        about to be torn down.
+        """
+        from django.db import close_old_connections
+
+        proxy_server = ProxyServer.get_instance()
+        if not proxy_server.redis_client:
+            return False
+
+        if channel_id in proxy_server._stopping_channels:
+            return False
+
+        disconnect_key = RedisKeys.last_client_disconnect(channel_id)
+        had_pending = bool(proxy_server.redis_client.exists(disconnect_key))
+        in_grace = had_pending or ChannelService.is_shutdown_pending(channel_id)
+
+        if not in_grace:
+            return False
+
+        try:
+            proxy_server.redis_client.delete(disconnect_key)
+
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            state = proxy_server.redis_client.hget(
+                metadata_key, ChannelMetadataField.STATE
+            )
+            if state:
+                state_str = state.decode() if isinstance(state, bytes) else state
+                if state_str == ChannelState.STOPPING:
+                    proxy_server.redis_client.hset(metadata_key, mapping={
+                        ChannelMetadataField.STATE: ChannelState.ACTIVE,
+                        ChannelMetadataField.STATE_CHANGED_AT: str(time.time()),
+                    })
+
+            stop_key = RedisKeys.channel_stopping(channel_id)
+            if proxy_server.redis_client.exists(stop_key):
+                proxy_server.redis_client.delete(stop_key)
+
+            if ChannelService._channel_proxy_is_active(
+                proxy_server.redis_client, channel_id
+            ):
+                from apps.channels.models import Channel
+
+                channel = Channel.objects.filter(uuid=channel_id).first()
+                if channel and not proxy_server.redis_client.get(
+                    f"channel_stream:{channel.id}"
+                ):
+                    sid, pid, error, slot_reserved = channel.get_stream()
+                    if error:
+                        logger.warning(
+                            f"Could not re-reserve stream for {channel_id} "
+                            f"after shutdown cancel: {error}"
+                        )
+                    elif slot_reserved and sid and pid:
+                        proxy_server.redis_client.hset(metadata_key, mapping={
+                            ChannelMetadataField.STREAM_ID: str(sid),
+                            ChannelMetadataField.M3U_PROFILE: str(pid),
+                        })
+                        logger.info(
+                            f"Re-reserved profile slot for {channel_id} "
+                            f"(stream={sid}, profile={pid})"
+                        )
+        finally:
+            close_old_connections()
+
+        return True
+
+    @staticmethod
+    def _channel_proxy_is_active(redis_client, channel_id):
+        """True when live proxy metadata shows this channel is still running."""
+        metadata_key = RedisKeys.channel_metadata(channel_id)
+        if not redis_client.exists(metadata_key):
+            return False
+        state = redis_client.hget(metadata_key, ChannelMetadataField.STATE)
+        if state is None:
+            return False
+        if isinstance(state, bytes):
+            state = state.decode()
+        return state in (
+            ChannelState.ACTIVE,
+            ChannelState.WAITING_FOR_CLIENTS,
+            ChannelState.BUFFERING,
+            ChannelState.INITIALIZING,
+            ChannelState.CONNECTING,
+        )
+
     @staticmethod
     def initialize_channel(channel_id, stream_url, user_agent, transcode=False, stream_profile_value=None, stream_id=None, m3u_profile_id=None, channel_name=None, stream_name=None):
         """
@@ -39,6 +212,7 @@ class ChannelService:
             bool: Success status
         """
         proxy_server = ProxyServer.get_instance()
+
         if stream_id and proxy_server.redis_client:
             metadata_key = RedisKeys.channel_metadata(channel_id)
             # Check if metadata already exists
@@ -263,58 +437,23 @@ class ChannelService:
             try:
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
                 if metadata and 'state' in metadata:
-                    state = metadata['state']
-                    channel_info = {"state": state}
-
-                    # Immediately mark as stopping in metadata so clients detect it faster
-                    proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.STOPPING)
-                    proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE_CHANGED_AT, str(time.time()))
+                    channel_info = {"state": metadata['state']}
             except Exception as e:
                 logger.error(f"Error fetching channel state: {e}")
 
-        # Set stopping flag with higher TTL to ensure it persists
+        # Mark stopping in Redis and notify all workers before local teardown.
+        # stop_channel() releases profile slots via _clean_redis_keys() before Redis deletion.
         if proxy_server.redis_client:
-            stop_key = RedisKeys.channel_stopping(channel_id)
-            proxy_server.redis_client.setex(stop_key, 60, "true")  # Higher TTL of 60 seconds
-            logger.info(f"Set channel stopping flag with 60s TTL for channel {channel_id}")
-
-        # Broadcast stop event to all workers via PubSub
-        if proxy_server.redis_client:
-            ChannelService._publish_channel_stop_event(channel_id)
-
-            # Also stop locally to ensure this worker cleans up right away
-            local_result = proxy_server.stop_channel(channel_id)
-        else:
-            # No Redis, just stop locally
-            local_result = proxy_server.stop_channel(channel_id)
-
-        # Release the channel in the channel model if applicable
-        try:
-            channel = Channel.objects.get(uuid=channel_id)
-            model_released = channel.release_stream()
-            if model_released:
-                logger.info(f"Released channel {channel_id} stream allocation")
-            else:
-                logger.warning(f"Channel {channel_id}: release_stream found no keys to clean")
-        except (Channel.DoesNotExist, Exception):
-            logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash")
-            try:
-                stream = Stream.objects.get(stream_hash=channel_id)
-                model_released = stream.release_stream()
-                if model_released:
-                    logger.info(f"Released stream {channel_id} stream allocation")
-                else:
-                    logger.warning(f"Stream {channel_id}: release_stream found no keys to clean")
-            except (Stream.DoesNotExist, Exception) as e:
-                logger.error(f"No Channel or Stream found for {channel_id}: {e}")
-                model_released = False
+            ChannelService.mark_channel_stopping(channel_id, broadcast=True)
+            logger.info(f"Marked channel {channel_id} stopping and broadcast stop to all workers")
+        local_result = proxy_server.stop_channel(channel_id)
 
         return {
             'status': 'success',
             'message': 'Channel stop request sent',
             'channel_id': channel_id,
             'previous_state': channel_info,
-            'model_released': model_released,
+            'model_released': bool(local_result),
             'local_stop_result': local_result
         }
 

apps/proxy/live_proxy/views.py

@@ -43,6 +43,15 @@ from apps.proxy.utils import check_user_stream_limits
 logger = get_logger()
 
 
+def _channel_stopping_response():
+    response = JsonResponse(
+        {"error": "Channel is stopping, retry shortly"},
+        status=503,
+    )
+    response["Retry-After"] = "1"
+    return response
+
+
 def _resolve_output_format(user, force=None, request=None):
     """Return the output format string to use for this client."""
     _FORMAT_ALIASES = {
@@ -123,6 +132,12 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                     status=429
                 )
 
+        if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+            logger.info(
+                f"[{client_id}] Channel {channel_id} unavailable. Teardown or pending shutdown"
+            )
+            return _channel_stopping_response()
+
         # Check if we need to reinitialize the channel
         needs_initialization = True
         channel_state = None
@@ -144,7 +159,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                         ChannelState.BUFFERING,
                         ChannelState.INITIALIZING,
                         ChannelState.CONNECTING,
-                        ChannelState.STOPPING,
                     ]:
                         needs_initialization = False
                         logger.debug(
@@ -160,6 +174,11 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                             logger.debug(
                                 f"[{client_id}] Channel {channel_id} is still initializing, client will wait"
                             )
+                    elif channel_state == ChannelState.STOPPING:
+                        logger.info(
+                            f"[{client_id}] Channel {channel_id} is stopping, rejecting request"
+                        )
+                        return _channel_stopping_response()
                     # Terminal states - channel needs cleanup before reinitialization
                     elif channel_state in [
                         ChannelState.ERROR,
@@ -192,6 +211,12 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
 
         # Start initialization if needed
         if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
+            if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+                logger.info(
+                    f"[{client_id}] Channel {channel_id} became unavailable before init, rejecting"
+                )
+                return _channel_stopping_response()
+
             logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
             # Force cleanup of any previous instance if in terminal state
             if channel_state in [
@@ -414,6 +439,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                     )  # 502 Bad Gateway
 
             # Initialize channel with the stream's user agent (not the client's)
+            if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+                if connection_allocated:
+                    if not channel.release_stream():
+                        logger.warning(f"[{client_id}] Failed to release stream before teardown reject")
+                    connection_allocated = False
+                logger.info(
+                    f"[{client_id}] Channel {channel_id} unavailable before init call, rejecting"
+                )
+                return _channel_stopping_response()
+
             success = ChannelService.initialize_channel(
                 channel_id,
                 stream_url,
@@ -522,6 +557,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                 f"[{client_id}] Successfully initialized channel {channel_id} locally"
             )
 
+        if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+            if _client_pre_registered:
+                mgr = proxy_server.client_managers.get(channel_id)
+                if mgr:
+                    mgr.remove_client(client_id)
+            logger.info(
+                f"[{client_id}] Channel {channel_id} became unavailable during setup, rejecting"
+            )
+            return _channel_stopping_response()
+
         # Register client
         output_profile = _resolve_output_profile(request, user)
         if output_profile:

apps/proxy/vod_proxy/tests/test_vod_db_cleanup.py

@@ -0,0 +1,146 @@
+"""VOD proxy must release geventpool checkouts after ORM on stream and stats paths."""
+
+from unittest.mock import MagicMock, patch
+
+from django.http import StreamingHttpResponse
+from django.test import RequestFactory, SimpleTestCase
+
+
+class StreamVodDbCleanupTests(SimpleTestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+
+    @patch("apps.proxy.vod_proxy.views.close_old_connections")
+    @patch("apps.proxy.vod_proxy.views.MultiWorkerVODConnectionManager")
+    @patch("apps.proxy.vod_proxy.views._transform_url", return_value="http://example.com/movie.mp4")
+    @patch("apps.proxy.vod_proxy.views._get_m3u_profile")
+    @patch("apps.proxy.vod_proxy.views._get_stream_url_from_relation", return_value="http://upstream/movie.mp4")
+    @patch("apps.proxy.vod_proxy.views._get_content_and_relation")
+    @patch("apps.proxy.vod_proxy.views.network_access_allowed", return_value=True)
+    def test_stream_vod_closes_db_before_streaming_response(
+        self,
+        _network_ok,
+        mock_content,
+        _stream_url,
+        mock_profile,
+        _transform,
+        mock_manager_cls,
+        mock_close,
+    ):
+        movie = MagicMock()
+        movie.name = "Test Movie"
+        relation = MagicMock()
+        relation.m3u_account.name = "Provider"
+        mock_content.return_value = (movie, relation)
+
+        profile = MagicMock()
+        profile.id = 1
+        profile.max_streams = 5
+        mock_profile.return_value = (profile, 0)
+
+        mock_manager = MagicMock()
+        mock_manager.stream_content_with_session.return_value = StreamingHttpResponse(
+            streaming_content=iter([b"data"]),
+            content_type="video/mp4",
+        )
+        mock_manager_cls.get_instance.return_value = mock_manager
+
+        request = self.factory.get(
+            "/proxy/vod/movie/uuid/session123/",
+            HTTP_USER_AGENT="test-agent",
+        )
+        request.user = MagicMock(is_authenticated=False)
+
+        from apps.proxy.vod_proxy.views import stream_vod
+
+        response = stream_vod(
+            request,
+            content_type="movie",
+            content_id="uuid",
+            session_id="session123",
+        )
+
+        self.assertIsInstance(response, StreamingHttpResponse)
+        mock_close.assert_called_once()
+        mock_manager.stream_content_with_session.assert_called_once()
+
+
+class BuildVodStatsDbCleanupTests(SimpleTestCase):
+    @patch("apps.proxy.vod_proxy.views.close_old_connections")
+    @patch("apps.proxy.vod_proxy.views.Movie")
+    def test_build_vod_stats_data_closes_db(self, mock_movie, mock_close):
+        redis_client = MagicMock()
+        redis_client.scan.side_effect = [
+            (0, ["vod_persistent_connection:s1"]),
+        ]
+        redis_client.hgetall.return_value = {
+            "content_obj_type": "movie",
+            "content_uuid": "movie-uuid",
+            "content_name": "Test Movie",
+            "m3u_profile_id": "1",
+            "client_ip": "127.0.0.1",
+            "client_user_agent": "agent",
+            "connected_at": "1000.0",
+            "last_activity": "1001.0",
+            "active_streams": "1",
+        }
+
+        movie_obj = MagicMock(
+            name="Test Movie",
+            logo=None,
+            year=2020,
+            rating=7.5,
+            genre="Action",
+            description="Desc",
+            tmdb_id="1",
+            imdb_id="tt1",
+        )
+        mock_movie.objects.select_related.return_value.get.return_value = movie_obj
+
+        with patch("apps.m3u.models.M3UAccountProfile") as mock_profile_model:
+            mock_profile_model.objects.select_related.return_value.get.return_value = MagicMock(
+                name="Profile 1",
+                m3u_account=MagicMock(name="Account", id=1),
+            )
+
+            from apps.proxy.vod_proxy.views import build_vod_stats_data
+
+            stats = build_vod_stats_data(redis_client)
+
+        self.assertEqual(stats["total_connections"], 1)
+        mock_close.assert_called_once()
+
+    @patch("apps.proxy.vod_proxy.views.close_old_connections")
+    def test_build_vod_stats_data_closes_db_on_error(self, mock_close):
+        redis_client = MagicMock()
+        redis_client.scan.side_effect = RuntimeError("redis down")
+
+        from apps.proxy.vod_proxy.views import build_vod_stats_data
+
+        stats = build_vod_stats_data(redis_client)
+
+        self.assertEqual(stats["total_connections"], 0)
+        mock_close.assert_called_once()
+
+
+class VodStatsUpdateDbCleanupTests(SimpleTestCase):
+    @patch("core.utils.send_websocket_update")
+    @patch("apps.proxy.vod_proxy.views.build_vod_stats_data")
+    def test_do_vod_stats_update_uses_build_vod_stats_data(self, mock_build, mock_ws):
+        mock_build.return_value = {
+            "vod_connections": [],
+            "total_connections": 0,
+            "timestamp": 0,
+        }
+
+        from apps.proxy.vod_proxy.multi_worker_connection_manager import (
+            MultiWorkerVODConnectionManager,
+        )
+
+        manager = MultiWorkerVODConnectionManager.__new__(MultiWorkerVODConnectionManager)
+        manager.redis_client = MagicMock()
+
+        manager._do_vod_stats_update()
+
+        mock_build.assert_called_once_with(manager.redis_client)
+        mock_ws.assert_called_once()

apps/proxy/vod_proxy/views.py

@@ -8,6 +8,7 @@ import random
 import logging
 import requests
 from urllib.parse import urlencode
+from django.db import close_old_connections
 from django.http import  JsonResponse, Http404, HttpResponse
 from django.shortcuts import get_object_or_404
 from django.views.decorators.csrf import csrf_exempt
@@ -601,6 +602,9 @@ def stream_vod(request, content_type, content_id, session_id=None, profile_id=No
         # Get connection manager (Redis-backed for multi-worker support)
         connection_manager = MultiWorkerVODConnectionManager.get_instance()
 
+        # Release ORM checkout before returning a long-lived StreamingHttpResponse.
+        close_old_connections()
+
         # Stream the content with session-based connection reuse
         logger.info("[VOD-STREAM] Calling connection manager to stream content")
         response = connection_manager.stream_content_with_session(
@@ -1063,6 +1067,8 @@ def build_vod_stats_data(redis_client):
     except Exception as e:
         logger.error(f"Error building VOD stats: {e}")
         return {'vod_connections': [], 'total_connections': 0, 'timestamp': time.time()}
+    finally:
+        close_old_connections()
 
 
 @api_view(["GET"])

core/utils.py

@@ -623,6 +623,8 @@ def validate_flexible_url(value):
         raise ValidationError("Enter a valid URL.")
 
 def dispatch_event_system(event_type, channel_id=None, channel_name=None, **details):
+    from django.db import close_old_connections
+
     try:
         from apps.connect.utils import trigger_event
         from apps.channels.models import Channel, Stream
@@ -696,9 +698,48 @@ def dispatch_event_system(event_type, channel_id=None, channel_name=None, **deta
 
         trigger_event(event_type, payload)
 
-    except Exception as e:
+    except Exception:
         # Don't fail main path if connect dispatch fails
         pass
+    finally:
+        close_old_connections()
+
+
+def _dispatch_system_event_integrations(
+    event_type, channel_id=None, channel_name=None, **details
+):
+    """
+    Run Connect subscriptions and plugin event hooks without blocking the caller.
+
+    On gevent uWSGI workers, dispatch runs in a spawned greenlet so slow webhooks,
+    scripts, or plugin handlers cannot stall live-proxy teardown or streaming paths.
+    Celery prefork workers (gevent patched but no hub) run synchronously instead.
+    """
+
+    def _run():
+        try:
+            dispatch_event_system(
+                event_type,
+                channel_id=channel_id,
+                channel_name=channel_name,
+                **details,
+            )
+        except Exception as e:
+            logger.error(
+                "Failed to dispatch Connect/plugin handlers for event %s: %s",
+                event_type,
+                e,
+            )
+
+    if _should_use_sync_websocket_send():
+        _run()
+    elif _is_gevent_monkey_patched():
+        import gevent
+
+        gevent.spawn(_run)
+    else:
+        _run()
+
 
 def log_system_event(event_type, channel_id=None, channel_name=None, **details):
     """
@@ -715,6 +756,7 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details):
                         stream_url='http://...', user='admin')
     """
     from core.models import SystemEvent, CoreSettings
+    from django.db import close_old_connections
 
     try:
         # Create the event
@@ -725,8 +767,13 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details):
             details=details
         )
 
-        # Trigger connect integrations for specific events
-        dispatch_event_system(event_type, channel_id=channel_id, channel_name=channel_name, **details)
+        # Connect integrations and plugin event hooks (non-blocking on gevent uWSGI)
+        _dispatch_system_event_integrations(
+            event_type,
+            channel_id=channel_id,
+            channel_name=channel_name,
+            **details,
+        )
 
         # Get max events from settings (default 100)
         try:
@@ -750,6 +797,10 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details):
     except Exception as e:
         # Don't let event logging break the main application
         logger.error(f"Failed to log system event {event_type}: {e}")
+    finally:
+        # geventpool keeps checked-out connections until close(); release promptly
+        # when logging from proxy greenlets/threads outside a normal request cycle.
+        close_old_connections()
 
 
 def _send_async(channel_layer, group, message):

dispatcharr/celery.py

@@ -94,6 +94,7 @@ def cleanup_task_memory(**kwargs):
         'apps.epg.tasks.refresh_all_epg_data',
         'apps.epg.tasks.parse_programs_for_source',
         'apps.epg.tasks.parse_programs_for_tvg_id',
+        'apps.epg.tasks.build_programme_index_task',
         'apps.channels.tasks.match_epg_channels',
         'apps.channels.tasks.match_selected_channels_epg',
         'apps.channels.tasks.match_single_channel_epg',

tests/test_log_system_event.py

@@ -0,0 +1,55 @@
+"""Tests for log_system_event Connect/plugin dispatch and DB cleanup."""
+
+from unittest.mock import patch
+
+from django.test import SimpleTestCase
+
+
+class LogSystemEventDispatchTests(SimpleTestCase):
+    @patch("django.db.close_old_connections")
+    @patch("core.utils._dispatch_system_event_integrations")
+    @patch("core.models.SystemEvent.objects")
+    @patch("core.models.CoreSettings.objects")
+    def test_log_system_event_dispatches_integrations_and_closes_db(
+        self, mock_core_settings, mock_system_event, mock_dispatch, mock_close
+    ):
+        mock_system_event.count.return_value = 1
+        mock_core_settings.filter.return_value.first.return_value = None
+
+        from core.utils import log_system_event
+
+        log_system_event("channel_start", channel_id="abc", channel_name="Test")
+
+        mock_dispatch.assert_called_once_with(
+            "channel_start",
+            channel_id="abc",
+            channel_name="Test",
+        )
+        mock_close.assert_called_once()
+
+    @patch("django.db.close_old_connections")
+    @patch("apps.connect.utils.trigger_event")
+    def test_integration_dispatch_closes_db_on_sync_path(
+        self, mock_trigger, mock_close
+    ):
+        from core.utils import _dispatch_system_event_integrations
+
+        with patch("core.utils._should_use_sync_websocket_send", return_value=True):
+            _dispatch_system_event_integrations("client_connect", channel_id="abc")
+
+        mock_trigger.assert_called_once()
+        mock_close.assert_called_once()
+
+    @patch("core.utils.dispatch_event_system")
+    def test_integration_dispatch_spawns_on_gevent_uwsgi(
+        self, mock_dispatch
+    ):
+        from core.utils import _dispatch_system_event_integrations
+
+        with patch("core.utils._should_use_sync_websocket_send", return_value=False), patch(
+            "core.utils._is_gevent_monkey_patched", return_value=True
+        ), patch("gevent.spawn") as mock_spawn:
+            _dispatch_system_event_integrations("channel_stop", channel_id="abc")
+
+        mock_spawn.assert_called_once()
+        mock_dispatch.assert_not_called()