diff --git a/apps/timeshift/tests/test_views.py b/apps/timeshift/tests/test_views.py index ebe3069e..5b91d52e 100644 --- a/apps/timeshift/tests/test_views.py +++ b/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"): diff --git a/apps/timeshift/views.py b/apps/timeshift/views.py index 77454664..b34cf379 100644 --- a/apps/timeshift/views.py +++ b/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):