fix(timeshift): ensure database connections are closed before returning responses

Added a helper function to close old database connections in the `timeshift_proxy` view before returning any HTTP responses. This change prevents potential database connection leaks and ensures that connections are properly managed during the request lifecycle. Updated tests to verify that connections are closed appropriately in various response scenarios.

Also stores provider_tz_name with session pool in redis to avoid an extra database call.

apps/timeshift/tests/test_views.py

@@ -29,6 +29,7 @@ def _seed_pool_session(
     user_id=5,
     client_ip="1.2.3.4",
     client_user_agent="test-agent",
+    provider_tz_name="Europe/Brussels",
 ):
     views._create_pool_session(
         redis,
@@ -41,6 +42,7 @@ def _seed_pool_session(
         profile_id=31,
         stream_id="111",
         provider_timestamp="2026-06-08:19-00",
+        provider_tz_name=provider_tz_name,
     )
     if serving_range is not None:
         redis.hset(f"timeshift_pool:{session_id}", "serving_range", serving_range)
@@ -183,6 +185,37 @@ class StreamFromProviderStatusMappingTests(TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(mocked_open.call_count, 1)
 
+    @patch.object(views, "close_old_connections")
+    @patch.object(views, "_open_upstream")
+    def test_streaming_response_closes_db_before_return(
+        self, mocked_open, mock_close,
+    ):
+        mocked_open.side_effect = [_fake_upstream(200, body=_make_ts_payload())]
+        with patch.object(views, "RedisClient"), \
+             patch.object(views, "_register_stats_client"), \
+             patch.object(views, "_unregister_stats_client"):
+            response = views._stream_from_provider(**self.kwargs)
+        self.assertEqual(response.status_code, 200)
+        mock_close.assert_called_once()
+
+    @patch.object(views, "close_old_connections")
+    @patch.object(views, "_open_upstream")
+    def test_upstream_failure_closes_db_before_return(self, mocked_open, mock_close):
+        mocked_open.return_value = _fake_upstream(404)
+        response = views._stream_from_provider(**self.kwargs)
+        self.assertEqual(response.status_code, 404)
+        mock_close.assert_called_once()
+
+    @patch.object(views, "close_old_connections")
+    @patch.object(views, "_open_upstream")
+    def test_passthrough_416_closes_db_before_return(self, mocked_open, mock_close):
+        upstream = _fake_upstream(416)
+        upstream.headers["Content-Range"] = "bytes */1000"
+        mocked_open.return_value = upstream
+        response = views._stream_from_provider(**self.kwargs)
+        self.assertEqual(response.status_code, 416)
+        mock_close.assert_called_once()
+
     @patch.object(views, "_open_upstream")
     def test_second_candidate_succeeds_after_404(self, mocked_open):
         # Primary 404 → second candidate 200 → streams successfully.
@@ -1617,11 +1650,7 @@ class TimeshiftSessionReuseTests(TestCase):
         """Drive _stream_reused_session against a session anchored at 17-00
         (provider position 19-00) with a NEW requested timestamp."""
         profile = MagicMock(id=31, custom_properties={})
-        tz_profile = MagicMock(
-            custom_properties={"server_info": {"timezone": "Europe/Brussels"}}
-        )
         account = MagicMock(id=1)
-        account.profiles.filter.return_value.first.return_value = tz_profile
         ok = MagicMock(status_code=200)
         with patch.object(views.M3UAccount.objects, "get", return_value=account), \
              patch.object(views, "_attempt_timeshift_stream",
@@ -1633,6 +1662,7 @@ class TimeshiftSessionReuseTests(TestCase):
                     "account_id": "1",
                     "stream_id": "111",
                     "provider_timestamp": "2026-06-08:19-00",
+                    "provider_tz_name": "Europe/Brussels",
                 },
                 profile=profile,
                 channel=self.channel,
@@ -1720,14 +1750,14 @@ class TimeshiftSessionReuseTests(TestCase):
         self.assertFalse(self.redis.exists(self._pool_key()))
 
     def test_reused_session_tz_falls_back_to_reserved_profile(self):
-        # No active default profile -> the reserved profile's server_info is
-        # the only tz source left; conversion must still apply.
-        _seed_pool_session(self.redis, session_id=self.SESSION)
+        # Legacy pool entries without provider_tz_name fall back to the
+        # reserved profile's server_info for conversion.
+        _seed_pool_session(self.redis, session_id=self.SESSION, provider_tz_name=None)
+        self.redis.hset(self._pool_key(), "provider_tz_name", "")
         profile = MagicMock(id=31, custom_properties={
             "server_info": {"timezone": "Europe/Brussels"},
         })
         account = MagicMock(id=1)
-        account.profiles.filter.return_value.first.return_value = None
         ok = MagicMock(status_code=200)
         with patch.object(views.M3UAccount.objects, "get", return_value=account), \
              patch.object(views, "_attempt_timeshift_stream",
@@ -1863,6 +1893,26 @@ class TimeshiftSessionRedirectTests(TestCase):
         self.assertIn("foo=bar", location)
         self.assertIn("baz=1", location)
 
+    @patch.object(views, "close_old_connections")
+    def test_redirect_closes_db_after_orm(self, mock_close):
+        request = self.factory.get(_proxy_url(session_id=None))
+        with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \
+             patch.object(views, "network_access_allowed", return_value=True), \
+             patch.object(views, "Channel") as channel_cls, \
+             patch.object(views, "_user_can_access_channel", return_value=True), \
+             patch.object(views, "get_channel_catchup_streams",
+                          return_value=[_make_catchup_stream()]), \
+             patch.object(views, "get_programme_duration", return_value=40), \
+             patch.object(views, "parse_catchup_timestamp", return_value=True), \
+             patch.object(views, "RedisClient") as redis_cls:
+            redis_cls.get_client.return_value = _FakeRedis()
+            channel_cls.objects.get.return_value = MagicMock(id=8)
+            response = views.timeshift_proxy(
+                request, "u", "p", "8", "2026-06-08:17-00", "8.ts",
+            )
+        self.assertEqual(response.status_code, 301)
+        mock_close.assert_called_once()
+
 
 class TimeshiftStreamLimitExemptionTests(TestCase):
     """Timeshift stream-limit bypass requires the same client session."""
@@ -2171,6 +2221,25 @@ class TimeshiftScrubPreemptTests(TestCase):
         self.assertFalse(second)
         self.assertTrue(self.redis.exists(f"timeshift_pool:{TEST_SESSION_ID}"))
 
+    def test_create_pool_session_stores_provider_tz_name(self):
+        views._create_pool_session(
+            self.redis,
+            session_id=TEST_SESSION_ID,
+            media_id=TEST_MEDIA_ID,
+            user_id=5,
+            client_ip="1.2.3.4",
+            client_user_agent="test-agent",
+            account_id=1,
+            profile_id=31,
+            stream_id="111",
+            provider_timestamp="2026-06-08:19-00",
+            provider_tz_name="Europe/Brussels",
+        )
+        self.assertEqual(
+            self.redis.hget(f"timeshift_pool:{TEST_SESSION_ID}", "provider_tz_name"),
+            "Europe/Brussels",
+        )
+
     def test_scrub_reuses_idle_pool_without_opening_failover(self):
         _seed_pool_session(self.redis, session_id=TEST_SESSION_ID)
         with patch.object(views, "release_profile_slot"):

apps/timeshift/views.py

@@ -9,6 +9,7 @@ from urllib.parse import urlencode
 
 import requests
 from django.core.cache import cache
+from django.db import close_old_connections
 from django.http import (
     Http404,
     HttpResponse,
@@ -50,6 +51,12 @@ CLIENT_TTL_SECONDS = 60
 _MATCH_SCORE_THRESHOLD = 8  # client_ip (5) + client_user_agent (3)
 
 
+def _finalize_timeshift_response(response):
+    """Release ORM database connections before returning to the client."""
+    close_old_connections()
+    return response
+
+
 def timeshift_proxy(request, username, password, stream_id, timestamp, duration):  # noqa: ARG001 stream_id
     """Proxy an XC catch-up request to the provider with multi-stream failover.
 
@@ -63,26 +70,29 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
 
     user = _authenticate_user(username, password)
     if user is None:
-        return HttpResponseForbidden("Invalid credentials")
+        return _finalize_timeshift_response(HttpResponseForbidden("Invalid credentials"))
 
     if not network_access_allowed(request, "XC_API", user):
-        return HttpResponseForbidden("Access denied")
+        return _finalize_timeshift_response(HttpResponseForbidden("Access denied"))
 
     try:
         channel = Channel.objects.get(id=int(raw_id))
     except (Channel.DoesNotExist, ValueError, TypeError):
+        close_old_connections()
         raise Http404("Channel not found") from None
 
     if not _user_can_access_channel(user, channel):
-        return HttpResponseForbidden("Access denied")
+        return _finalize_timeshift_response(HttpResponseForbidden("Access denied"))
 
     # Shape helpers pass through on parse failure; reject bad input before upstream.
     if parse_catchup_timestamp(timestamp) is None:
-        return HttpResponseBadRequest("Invalid timestamp")
+        return _finalize_timeshift_response(HttpResponseBadRequest("Invalid timestamp"))
 
     catchup_streams = get_channel_catchup_streams(channel)
     if not catchup_streams:
-        return HttpResponseBadRequest("Timeshift not supported for this channel")
+        return _finalize_timeshift_response(
+            HttpResponseBadRequest("Timeshift not supported for this channel")
+        )
 
     debug = logger.isEnabledFor(logging.DEBUG)
 
@@ -105,14 +115,14 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
     session_id = request.GET.get("session_id")
     if not session_id:
         logger.debug("Timeshift session redirect: %s (new session)", request.path)
-        return _redirect_with_new_session(request)
+        return _finalize_timeshift_response(_redirect_with_new_session(request))
 
     session_entry = _get_pool_entry(redis_client, session_id)
     if session_entry and not _pool_entry_owned_by_user(session_entry, user.id):
         logger.info(
             "Timeshift: rejecting foreign session_id for user %s", user.id,
         )
-        return _redirect_with_new_session(request)
+        return _finalize_timeshift_response(_redirect_with_new_session(request))
 
     # Stable client identity for stats, stop keys, and the provider pool.
     effective_session_id = session_id
@@ -158,7 +168,7 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
     )
 
     if not check_user_stream_limits(user, client_id, media_id=media_id):
-        return HttpResponseForbidden("Stream limit exceeded")
+        return _finalize_timeshift_response(HttpResponseForbidden("Stream limit exceeded"))
 
     if effective_session_id == session_id:
         pool = _snapshot_from_entry(session_entry)
@@ -211,9 +221,9 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
         )
         if reuse_response is not None:
             if reuse_response.status_code < 400:
-                return reuse_response
+                return _finalize_timeshift_response(reuse_response)
             if getattr(reuse_response, "timeshift_passthrough", False) is True:
-                return reuse_response
+                return _finalize_timeshift_response(reuse_response)
             last_response = reuse_response
             if getattr(reuse_response, "timeshift_decisive", False):
                 try:
@@ -228,14 +238,14 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
             "Timeshift: deferring non-displacing request for session %s range=%s",
             effective_session_id, range_header or "(none)",
         )
-        return HttpResponse("Stream slot busy", status=503)
+        return _finalize_timeshift_response(HttpResponse("Stream slot busy", status=503))
 
     if pool_exists and pool_busy:
         logger.warning(
             "Timeshift: session %s did not become idle in time",
             effective_session_id,
         )
-        return HttpResponse("Stream slot busy", status=503)
+        return _finalize_timeshift_response(HttpResponse("Stream slot busy", status=503))
 
     capacity_blocked = False
     for catchup_stream in catchup_streams:
@@ -301,6 +311,7 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
             profile_id=reserved_profile.id,
             stream_id=stream_id_value,
             provider_timestamp=provider_timestamp,
+            provider_tz_name=provider_tz_name,
         ):
             try:
                 release_profile_slot(reserved_profile.id, redis_client)
@@ -313,7 +324,9 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
                 "Timeshift: pool entry already exists for session %s, deferring",
                 effective_session_id,
             )
-            return HttpResponse("Stream slot busy", status=503)
+            return _finalize_timeshift_response(
+                HttpResponse("Stream slot busy", status=503)
+            )
         release_cb = _make_release_once(
             redis_client, effective_session_id, reserved_profile.id
         )
@@ -341,17 +354,18 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
             )
         except Exception:
             _discard_pool_session(redis_client, effective_session_id, reserved_profile.id)
+            close_old_connections()
             raise
         if response.status_code < 400:
             # Streaming: the generator's close path frees the slot via release_cb.
-            return response
+            return _finalize_timeshift_response(response)
 
         if getattr(response, "timeshift_passthrough", False) is True:
             # Terminal range answer (e.g. 416 past EOF): the upstream session is
             # healthy, so free the slot but keep the entry idle for the next
             # probe, and return verbatim without failing over to other accounts.
             release_cb()
-            return response
+            return _finalize_timeshift_response(response)
 
         # Real failure: drop this session entirely and fail over.
         _discard_pool_session(redis_client, effective_session_id, reserved_profile.id)
@@ -368,10 +382,14 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
         )
 
     if last_response is not None:
-        return last_response
+        return _finalize_timeshift_response(last_response)
     if capacity_blocked:
-        return HttpResponse("No available stream slot", status=503)
-    return HttpResponseBadRequest("Cannot build timeshift URL")
+        return _finalize_timeshift_response(
+            HttpResponse("No available stream slot", status=503)
+        )
+    return _finalize_timeshift_response(
+        HttpResponseBadRequest("Cannot build timeshift URL")
+    )
 
 
 def _authenticate_user(username, password):
@@ -714,6 +732,7 @@ def _create_pool_session(
     profile_id,
     stream_id,
     provider_timestamp,
+    provider_tz_name=None,
 ):
     """Register an already-reserved slot for this client session."""
     if redis_client is None or not session_id:
@@ -733,6 +752,7 @@ def _create_pool_session(
                 "profile_id": str(profile_id),
                 "stream_id": str(stream_id),
                 "provider_timestamp": str(provider_timestamp),
+                "provider_tz_name": str(provider_tz_name or ""),
                 "busy": "1",
                 "last_activity": now,
             })
@@ -947,18 +967,17 @@ def _stream_reused_session(
     # for. Clients (TiviMate) keep the ?session_id= query when they rebuild
     # the seek URL with a new start, so a reused session must serve the
     # REQUESTED position, never replay the stored one — otherwise every
-    # timestamp-jump FF/RW snaps back to the session's original anchor. The
-    # provider zone is a property of the account (server_info reported on
-    # auth, stored on the default profile), so recomputing costs one trivial
-    # conversion.
-    provider_tz_name = None
-    tz_profile = (
-        m3u_account.profiles.filter(is_active=True, is_default=True).first()
-        or profile
-    )
-    server_info = (tz_profile.custom_properties or {}).get("server_info") or {}
-    if isinstance(server_info, dict):
-        provider_tz_name = server_info.get("timezone")
+    # timestamp-jump FF/RW snaps back to the session's original anchor.
+    # Provider zone is cached on the pool entry at session creation (account
+    # property from server_info); fall back to the reserved profile for legacy
+    # entries created before that field existed.
+    provider_tz_name = descriptor.get("provider_tz_name") or None
+    if provider_tz_name:
+        provider_tz_name = str(provider_tz_name)
+    if not provider_tz_name:
+        server_info = (profile.custom_properties or {}).get("server_info") or {}
+        if isinstance(server_info, dict):
+            provider_tz_name = server_info.get("timezone")
     provider_timestamp = convert_timestamp_to_provider_tz(timestamp, provider_tz_name)
 
     # Move the descriptor to the position actually served so fingerprint
@@ -1216,7 +1235,9 @@ def _stream_from_provider(
                 "Timeshift provider unreachable (%s): %s",
                 _redact_url(url), type(exc).__name__,
             )
-            return HttpResponseBadRequest("Provider connection error")
+            return _finalize_timeshift_response(
+                HttpResponseBadRequest("Provider connection error")
+            )
         last_status = response.status_code
         last_url = url
         if debug:
@@ -1233,7 +1254,7 @@ def _stream_from_provider(
             # and only multiplies upstream connections.
             content_range = response.headers.get("Content-Range")
             response.close()
-            return _passthrough_response(416, content_range)
+            return _finalize_timeshift_response(_passthrough_response(416, content_range))
         if response.status_code in (200, 206):
             peek = response.raw.read(1024)
             sync_offset = find_ts_sync(peek) if peek else -1
@@ -1286,7 +1307,7 @@ def _stream_from_provider(
         else:
             failure = HttpResponseBadRequest("Provider error")
         failure.timeshift_decisive = decisive_failure
-        return failure
+        return _finalize_timeshift_response(failure)
 
     content_type = upstream.headers.get("Content-Type", "video/mp2t")
     content_range = upstream.headers.get("Content-Range", "")
@@ -1400,7 +1421,7 @@ def _stream_from_provider(
     if content_range:
         response["Content-Range"] = content_range
     response["Accept-Ranges"] = "bytes"
-    return response
+    return _finalize_timeshift_response(response)
 
 
 def _redact_url(url):