From 33ada59dbaa3e1ba6146fb36e42391794a12655b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 15 Jul 2026 21:26:25 +0000 Subject: [PATCH] fix(timeshift): refine catch-up timestamp handling and enhance EOF probe logic Updated the catch-up timestamp normalization to reflect XC client specifications. Introduced a new mechanism to handle near-EOF duration probes, ensuring that playback statistics are preserved without reanchoring to the end of the file. Additionally, improved the handling of presentation lengths in the timeshift proxy to align with XC behavior. Updated tests to validate these changes and ensure robust playback functionality. --- apps/timeshift/helpers.py | 4 +- apps/timeshift/stats.py | 30 + apps/timeshift/tests/test_helpers.py | 2 +- apps/timeshift/tests/test_stats.py | 16 + apps/timeshift/tests/test_views.py | 888 +++++++++++++++++++++++++-- apps/timeshift/views.py | 701 +++++++++++++++++++-- 6 files changed, 1563 insertions(+), 78 deletions(-) diff --git a/apps/timeshift/helpers.py b/apps/timeshift/helpers.py index 7b8d5ac1..207c2bc8 100644 --- a/apps/timeshift/helpers.py +++ b/apps/timeshift/helpers.py @@ -18,7 +18,7 @@ DEFAULT_DURATION_MINUTES = 120 DURATION_BUFFER_MINUTES = 5 MAX_DURATION_MINUTES = 480 -# Wall-clock shapes seen from XC / iPlayTV / TiviMate clients. Compiled once. +# Wall-clock shapes seen from XC catch-up clients. Compiled once. _CATCHUP_WALL_CLOCK_RE = re.compile( r"^" r"(?P\d{4}-\d{2}-\d{2})" @@ -38,7 +38,7 @@ def normalize_catchup_timestamp_input(timestamp_str): """Map a client catch-up timestamp to an ISO-8601 string for ``fromisoformat``. Supported inputs: - - ``YYYY-MM-DD:HH-MM`` (iPlayTV/TiviMate colon-dash) + - ``YYYY-MM-DD:HH-MM`` (XC colon-dash) - ``YYYY-MM-DD_HH-MM`` (XC underscore) - ``YYYY-MM-DD:HH:MM[:SS]`` (XC colon time in catch-up URLs) - ``YYYY-MM-DD HH:MM[:SS]`` (EPG / SQL datetime) diff --git a/apps/timeshift/stats.py b/apps/timeshift/stats.py index c7580cba..de5cf857 100644 --- a/apps/timeshift/stats.py +++ b/apps/timeshift/stats.py @@ -15,6 +15,10 @@ from apps.timeshift.redis_keys import TimeshiftRedisKeys, parse_stats_channel_id from apps.timeshift.helpers import parse_catchup_timestamp logger = logging.getLogger(__name__) + +# Shared with views near-EOF classification (common ~1.88MB duration probes). +EOF_PROBE_TAIL_BYTES = 2_097_152 + _STREAM_STATS_TO_METADATA = { "video_codec": ChannelMetadataField.VIDEO_CODEC, "resolution": ChannelMetadataField.RESOLUTION, @@ -165,6 +169,32 @@ def resolve_stats_playback_fields( existing_programme_start is not None and existing_programme_start != timestamp_utc ) + + # Near-EOF duration probes must not reanchor stats to end-of-file. + if ( + range_start is not None + and representation_length is not None + and not programme_changed + ): + try: + start = int(range_start) + total = int(representation_length) + except (TypeError, ValueError): + start = None + total = None + if start is not None and total is not None and total > 0: + if start >= max(0, total - EOF_PROBE_TAIL_BYTES): + try: + keep_base = ( + float(existing_playback_base) + if existing_playback_base is not None + else None + ) + except (TypeError, ValueError): + keep_base = None + keep_anchor = existing_position_anchor or now + return keep_base, keep_anchor + byte_base = compute_playback_base_from_byte_range( range_start, representation_length, programme_duration_secs, ) diff --git a/apps/timeshift/tests/test_helpers.py b/apps/timeshift/tests/test_helpers.py index 068bf1d0..882e01c8 100644 --- a/apps/timeshift/tests/test_helpers.py +++ b/apps/timeshift/tests/test_helpers.py @@ -21,7 +21,7 @@ from apps.timeshift.helpers import ( def _make_creds(): # The builders consume resolved per-profile credentials, never an account - # object — get_transformed_credentials() produces these in the view. + # object - get_transformed_credentials() produces these in the view. return TimeshiftCredentials("http://example.test", "user", "pass") diff --git a/apps/timeshift/tests/test_stats.py b/apps/timeshift/tests/test_stats.py index de60355a..c410f66f 100644 --- a/apps/timeshift/tests/test_stats.py +++ b/apps/timeshift/tests/test_stats.py @@ -144,6 +144,22 @@ class ByteRangePlaybackTests(TestCase): self.assertIsNone(base) self.assertEqual(anchor, "2000.0") + def test_resolve_near_eof_probe_keeps_existing_position(self): + # Clients probe ~1.88MB from EOF for duration; must not flash to end. + total = 8_783_238_116 + base, anchor = resolve_stats_playback_fields( + timestamp_utc="2026-07-14:14-59", + existing_programme_start="2026-07-14:14-59", + existing_position_anchor="1000.0", + existing_playback_base="2100.0", + range_start=total - 1_880_000, + representation_length=total, + programme_duration_secs=3600, + now="2000.0", + ) + self.assertAlmostEqual(base, 2100.0) + self.assertEqual(anchor, "1000.0") + class TimeshiftStreamStatsTests(TestCase): def test_stream_stats_to_metadata_fields(self): diff --git a/apps/timeshift/tests/test_views.py b/apps/timeshift/tests/test_views.py index 9aab8306..a6ff5c66 100644 --- a/apps/timeshift/tests/test_views.py +++ b/apps/timeshift/tests/test_views.py @@ -69,7 +69,7 @@ class FindTsSyncTests(TestCase): self.assertEqual(_find_ts_sync(preamble + aligned), len(preamble)) def test_returns_minus_one_when_no_chain_exists(self): - # Three lone 0x47 bytes that are NOT spaced at 188 — must not be + # Three lone 0x47 bytes that are NOT spaced at 188 - must not be # mistaken for a sync chain. self.assertEqual(_find_ts_sync(b"\x47\x00\x00\x47\x00\x00\x47" * 50), -1) @@ -84,10 +84,11 @@ def _make_ts_payload(size=1024): return (packet * ((size // 188) + 1))[:size] -def _fake_upstream(status_code, *, content_type="video/mp2t", body=b""): +def _fake_upstream(status_code, *, content_type="video/mp2t", body=b"", url=None): resp = MagicMock() resp.status_code = status_code resp.headers = {"Content-Type": content_type} + resp.url = url or "http://cdn.example.test/timeshift.ts" resp.iter_content = MagicMock(return_value=iter([body] if body else [])) resp.close = MagicMock() # Simulate raw.read() for the TS sync peek in _stream_from_provider. @@ -141,12 +142,104 @@ class StreamFromProviderStatusMappingTests(TestCase): @patch.object(views, "_open_upstream") def test_upstream_403_short_circuits_loop(self, mocked_open): - # 403 is decisive (auth) — no retry of further candidates. + # 403 is decisive (auth) - no retry of further candidates. mocked_open.return_value = _fake_upstream(403) response = views._stream_from_provider(**self.kwargs) self.assertEqual(response.status_code, 403) self.assertEqual(mocked_open.call_count, 1) + @patch.object(views, "_open_upstream") + def test_stores_final_url_after_successful_open(self, mocked_open): + redis = _FakeRedis() + session_id = "sess-cdn-store" + cdn = "http://cdn.example.test/tok/archive.ts" + mocked_open.return_value = _fake_upstream( + 200, body=_make_ts_payload(), url=cdn, + ) + with patch.object(views, "_register_stats_client"), \ + patch.object(views, "_unregister_stats_client"): + response = views._stream_from_provider( + **self.kwargs, + redis_client=redis, + pool_session_id=session_id, + ) + self.assertEqual(response.status_code, 200) + self.assertEqual( + redis.hget(views._pool_key(session_id), "final_url"), cdn, + ) + + @patch.object(views, "_open_upstream") + def test_reuses_cached_final_url_without_portal(self, mocked_open): + cdn = "http://cdn.example.test/tok/archive.ts" + mocked_open.return_value = _fake_upstream( + 200, body=_make_ts_payload(), url=cdn, + ) + with patch.object(views, "_register_stats_client"), \ + patch.object(views, "_unregister_stats_client"): + response = views._stream_from_provider( + **self.kwargs, final_url=cdn, + ) + self.assertEqual(response.status_code, 200) + mocked_open.assert_called_once() + self.assertEqual(mocked_open.call_args.args[0], cdn) + self.assertFalse(mocked_open.call_args.kwargs.get("allow_redirects", True)) + + @patch.object(views, "_open_upstream") + def test_xc_scrub_rewrite_does_not_byte_map_stats_position(self, mocked_open): + # Injected CDN Range is for the provider only; stats must follow the XC + # URL timestamp, not archive_offset/programme_duration (false "26:00"). + cdn = "http://cdn.example.test/timeshift/u/p/60/2026-05-12:17-00/1.ts?token=x" + upstream = _fake_upstream(206, body=_make_ts_payload(), url=cdn) + upstream.headers["Content-Range"] = "bytes 500000000-999999999/1000000000" + upstream.headers["Content-Length"] = "500000000" + mocked_open.return_value = upstream + kwargs = dict( + self.kwargs, + final_url=cdn, + range_header="bytes=500000000-", + rewrite_plain_get=True, + presentation_remaining=500000000, + presentation_byte_base=500000000, + duration_minutes=120, + ) + with patch.object(views, "_register_stats_client") as register_mock, \ + patch.object(views, "_unregister_stats_client"): + response = views._stream_from_provider(**kwargs) + self.assertEqual(response.status_code, 200) + self.assertIsNone(register_mock.call_args.kwargs.get("range_start")) + + @patch.object(views, "_open_upstream") + def test_expired_final_url_falls_back_to_portal(self, mocked_open): + cdn = "http://cdn.example.test/expired.ts" + portal = self.kwargs["candidate_urls"][0] + redis = _FakeRedis() + session_id = "sess-cdn-expire" + redis.hset(views._pool_key(session_id), "final_url", cdn) + mocked_open.side_effect = [ + _fake_upstream(403, url=cdn), + _fake_upstream(200, body=_make_ts_payload(), url=cdn + "?fresh=1"), + ] + with patch.object(views, "_register_stats_client"), \ + patch.object(views, "_unregister_stats_client"): + response = views._stream_from_provider( + **self.kwargs, + final_url=cdn, + redis_client=redis, + pool_session_id=session_id, + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(mocked_open.call_count, 2) + self.assertEqual(mocked_open.call_args_list[0].args[0], cdn) + self.assertEqual(mocked_open.call_args_list[1].args[0], portal) + self.assertTrue( + mocked_open.call_args_list[1].kwargs.get("allow_redirects", True) + ) + # Fresh portal response re-stores CDN URL. + self.assertEqual( + redis.hget(views._pool_key(session_id), "final_url"), + cdn + "?fresh=1", + ) + @patch.object(views, "_open_upstream") def test_upstream_302_short_circuits_loop(self, mocked_open): # Any 3xx is decisive: for XC providers a 302 is the first sign of @@ -161,7 +254,7 @@ class StreamFromProviderStatusMappingTests(TestCase): def test_upstream_500_continues_to_next_candidate(self, mocked_open): # A 5xx is format-specific on many XC servers (PHP fatal with # display_errors off turns an "Undefined array key" warning into a - # hard 500), so the cascade must keep trying — the next timestamp + # hard 500), so the cascade must keep trying - the next timestamp # shape often succeeds. Regression: providers that 500 on the first # shape used to fail outright because the loop short-circuited. mocked_open.side_effect = [ @@ -258,7 +351,7 @@ class StreamFromProviderStatusMappingTests(TestCase): @patch.object(views, "_open_upstream") def test_cache_promotes_winning_index_to_first(self, mocked_open): """Once a candidate succeeds for an account, the next request reorders - the list so the cached winner is tried first — saving cascade + the list so the cached winner is tried first - saving cascade overhead on fast-forward.""" # locmem cache: isolates this test from the shared Redis-backed django # cache (which persists across runs and parallel test sessions). @@ -279,7 +372,7 @@ class StreamFromProviderStatusMappingTests(TestCase): self.assertEqual(mocked_open.call_count, 2) # Second request: cached winner (index 1) is tried first, succeeds - # immediately — no cascade. + # immediately - no cascade. mocked_open.reset_mock() mocked_open.side_effect = [_fake_upstream(200, body=_make_ts_payload())] with patch.object(views, "RedisClient"), \ @@ -414,7 +507,7 @@ class StreamFromProviderStatusMappingTests(TestCase): class RedactUrlTests(TestCase): - """`_redact_url` is the guard that keeps XC credentials out of logs — + """`_redact_url` is the guard that keeps XC credentials out of logs - both URL forms carry them (query params in format A, path segments in format B).""" @@ -685,7 +778,7 @@ def _fake_creds(acc, prof): class TimeshiftProxyTimestampWiringTests(TestCase): """`timeshift_proxy` must convert the client's UTC timestamp to the serving provider's zone for the upstream URL, while keeping the ORIGINAL - UTC timestamp for the EPG duration lookup — the only timezone conversion + UTC timestamp for the EPG duration lookup - the only timezone conversion in the chain.""" def setUp(self): @@ -722,7 +815,7 @@ class TimeshiftProxyTimestampWiringTests(TestCase): self.assertEqual(build_mock.call_args[0][2], "2026-06-08:19-00") def test_duration_lookup_keeps_original_utc_timestamp(self): - # The EPG is stored in UTC — the duration lookup must NOT receive the + # The EPG is stored in UTC - the duration lookup must NOT receive the # provider-converted value. _, _, _, duration_mock, _ = self._call("2026-06-08:17-00") self.assertEqual(duration_mock.call_args[0][1], "2026-06-08:17-00") @@ -770,7 +863,7 @@ class TimeshiftProxyTimestampWiringTests(TestCase): class TimeshiftProxyFailoverTests(TestCase): """When the first catch-up stream's provider cannot serve the archive, - the proxy must fail over to the channel's next catch-up stream — each + the proxy must fail over to the channel's next catch-up stream - each attempt with its own provider context.""" def setUp(self): @@ -882,7 +975,7 @@ class TimeshiftProxyFailoverTests(TestCase): class _ProxyLoopTestMixin: - """Shared driver for tests exercising the failover loop end to end — + """Shared driver for tests exercising the failover loop end to end - pool reservation, credential resolution and Redis are all controlled.""" def setUp(self): @@ -938,7 +1031,7 @@ class TimeshiftProxyFailoverHardeningTests(_ProxyLoopTestMixin, TestCase): def test_decisive_failure_skips_same_accounts_other_streams(self): # Account 1 carries two variants (e.g. FHD + HD). A decisive # (auth/ban-class) failure on the first must NOT retry account 1's - # second stream — that would hammer a banning provider — but a + # second stream - that would hammer a banning provider - but a # DIFFERENT account stays fair game. streams = [ _make_catchup_stream(account_id=1, stream_id="111"), @@ -956,7 +1049,7 @@ class TimeshiftProxyFailoverHardeningTests(_ProxyLoopTestMixin, TestCase): def test_soft_failure_still_tries_same_accounts_other_streams(self): # A soft failure (404: this stream's archive missing) is stream- - # specific — the same account's other variant may still have it. + # specific - the same account's other variant may still have it. streams = [ _make_catchup_stream(account_id=1, stream_id="111"), _make_catchup_stream(account_id=1, stream_id="112"), @@ -1057,7 +1150,7 @@ class StreamFromProviderDecisiveEdgeTests(TestCase): @patch.object(views, "_open_upstream") def test_406_is_decisive_and_marks_response(self, mocked_open): - # 406 = IP-wide block in the XC ban escalation — single attempt, + # 406 = IP-wide block in the XC ban escalation - single attempt, # generic 400 to the client, and the failover loop must see the # decisive marker so it skips this account's other streams. mocked_open.return_value = _fake_upstream(406) @@ -1087,7 +1180,7 @@ class StreamFromProviderDecisiveEdgeTests(TestCase): class CatchupStreamsDbTests(TestCase): """get_channel_catchup_streams: the function that defines the failover - order — channelstream order, catch-up streams only, active accounts only.""" + order - channelstream order, catch-up streams only, active accounts only.""" @classmethod def setUpTestData(cls): @@ -1134,7 +1227,7 @@ class CatchupStreamsDbTests(TestCase): class AuthHelpersDbTests(TestCase): """_authenticate_user (xc_password custom property) and - _user_can_access_channel (user_level gate) — exercised against real models + _user_can_access_channel (user_level gate) - exercised against real models instead of being mocked away.""" @classmethod @@ -1163,7 +1256,7 @@ class AuthHelpersDbTests(TestCase): def test_user_without_xc_password_rejected(self): # Accounts with no xc_password set (e.g. admins) must be denied even - # if the caller guesses any string — there is nothing to compare to. + # if the caller guesses any string - there is nothing to compare to. self.assertIsNone(views._authenticate_user("ts-test-noxc", "")) self.assertIsNone(views._authenticate_user("ts-test-noxc", "anything")) @@ -1269,7 +1362,7 @@ class TimeshiftSlotPoolTests(_ProxyLoopTestMixin, TestCase): def test_capacity_failure_is_not_decisive_for_the_account(self): # profile_full on account 1's first stream must NOT mark account 1 - # decisive — capacity is transient, unlike a ban-class status. + # decisive - capacity is transient, unlike a ban-class status. streams = [ _make_catchup_stream(account_id=1, stream_id="111", profile_id=31), _make_catchup_stream(account_id=1, stream_id="112", profile_id=31), @@ -1295,7 +1388,7 @@ class TimeshiftSlotPoolTests(_ProxyLoopTestMixin, TestCase): def test_exception_from_provider_releases_slot(self): # An unexpected exception between reserve and response construction - # must release the slot before propagating — otherwise the counter + # must release the slot before propagating - otherwise the counter # (no TTL) leaks until the next Redis flush. streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] with self.assertRaises(RuntimeError): @@ -1305,7 +1398,7 @@ class TimeshiftSlotPoolTests(_ProxyLoopTestMixin, TestCase): def test_exception_before_upstream_releases_slot(self): # Same guarantee for failures BEFORE the upstream call (URL building, - # credential resolution, user-agent lookup) — the guarded window + # credential resolution, user-agent lookup) - the guarded window # starts right after the reservation. streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] with self.assertRaises(RuntimeError): @@ -1590,6 +1683,16 @@ class TimeshiftSessionReuseTests(TestCase): with patch.object(views, "release_profile_slot"): views._release_pool_session(self.redis, self.SESSION, 31) + def test_store_pool_provider_user_agent_snapshots_resolved_value(self): + _seed_pool_session(self.redis, session_id=self.SESSION) + views._store_pool_provider_user_agent( + self.redis, self.SESSION, "provider-agent", + ) + self.assertEqual( + self.redis.hget(self._pool_key(), "provider_user_agent"), + "provider-agent", + ) + def test_wait_returns_none_without_blocking_when_pool_empty(self): start = time.monotonic() acquired = views._wait_for_idle_pool_session(self.redis, self.SESSION) @@ -1773,7 +1876,7 @@ class TimeshiftSessionReuseTests(TestCase): self.assertEqual(matched, "old") def test_fresh_session_id_does_not_adopt_idle_exact_media_pool(self): - """TiviMate FF race: redirect mints a new session, then reconnects to + """FF race: redirect mints a new session, then reconnects to the old programme timestamp before the real seek arrives. Must not fingerprint-adopt the idle pool from the previous session.""" old_session = "oldrewsession1" @@ -1987,8 +2090,381 @@ class TimeshiftSessionReuseTests(TestCase): ) return response, ok, attempt_mock + def test_session_scrub_reuses_final_url_and_injects_range(self): + # Client FF rebuilds /timeshift/...//... with the same + # session_id. That must Range-seek the open CDN archive, not portal. + _seed_pool_session(self.redis, session_id=self.SESSION) + cdn = "http://cdn.example/archive.ts?token=ok" + self.redis.hset(self._pool_key(), mapping={ + "final_url": cdn, + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=200) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": TEST_MEDIA_ID, + "provider_timestamp": "2026-06-08:19-00", + "provider_tz_name": "Europe/Brussels", + "final_url": cdn, + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + }, + profile=profile, + channel=self.channel, + media_id="8_2026-06-08-17-30", + safe_ts="2026-06-08-17-30", + timestamp="2026-06-08:17-30", + duration_minutes=40, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=None, + channel_logo_id=None, + user=self.user, + debug=False, + ) + kwargs = attempt_mock.call_args.kwargs + self.assertEqual(kwargs.get("final_url"), cdn) + self.assertTrue(kwargs.get("rewrite_plain_get")) + self.assertTrue(kwargs.get("range_header", "").startswith("bytes=")) + self.assertIsNotNone(kwargs.get("presentation_remaining")) + self.assertIsNotNone(kwargs.get("presentation_byte_base")) + # Archive CDN state must survive the media_id move. + self.assertEqual(self.redis.hget(self._pool_key(), "final_url"), cdn) + + def test_session_scrub_reuses_opaque_final_url(self): + """Opaque CDNs still scrub via Range on the cached URL (no portal hop).""" + _seed_pool_session(self.redis, session_id=self.SESSION) + opaque = "http://opaque-cdn.example/hash/token/t5/test-stream/serve" + self.redis.hset(self._pool_key(), mapping={ + "final_url": opaque, + "content_length": "722718720", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=200) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": TEST_MEDIA_ID, + "provider_timestamp": "2026-06-08:19-00", + "provider_tz_name": "Europe/Brussels", + "final_url": opaque, + "content_length": "722718720", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + }, + profile=profile, + channel=self.channel, + media_id="8_2026-06-08-17-11", + safe_ts="2026-06-08-17-11", + timestamp="2026-06-08:17-11", + duration_minutes=35, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=None, + channel_logo_id=None, + user=self.user, + debug=False, + ) + kwargs = attempt_mock.call_args.kwargs + self.assertEqual(kwargs.get("final_url"), opaque) + self.assertTrue(kwargs.get("rewrite_plain_get")) + self.assertTrue(str(kwargs.get("range_header") or "").startswith("bytes=")) + + def test_session_range_after_scrub_maps_through_presentation_base(self): + # After a scrub rewrite, near-EOF probes are relative to the presented + # window, not the full CDN file. + _seed_pool_session(self.redis, session_id=self.SESSION) + cdn = "http://cdn.example/archive.ts?token=ok" + base = 314_934_968 + remaining = 419_913_672 + self.redis.hset(self._pool_key(), mapping={ + "final_url": cdn, + "content_length": str(base + remaining), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + "presentation_length": str(remaining), + "presentation_byte_base": str(base), + "media_id": "8_2026-06-08-17-30", + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=206) + client_range = f"bytes={remaining - 112_800}-" + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": "8_2026-06-08-17-30", + "provider_timestamp": "2026-06-08:19-30", + "provider_tz_name": "Europe/Brussels", + "final_url": cdn, + "content_length": str(base + remaining), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + "presentation_length": str(remaining), + "presentation_byte_base": str(base), + }, + profile=profile, + channel=self.channel, + media_id="8_2026-06-08-17-30", + safe_ts="2026-06-08-17-30", + timestamp="2026-06-08:17-30", + duration_minutes=40, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=client_range, + channel_logo_id=None, + user=self.user, + debug=False, + ) + kwargs = attempt_mock.call_args.kwargs + self.assertEqual( + kwargs.get("range_header"), + f"bytes={base + remaining - 112_800}-", + ) + self.assertTrue(kwargs.get("relative_presentation_range")) + self.assertFalse(kwargs.get("rewrite_plain_get")) + self.assertEqual(kwargs.get("final_url"), cdn) + + def test_return_to_archive_start_resets_presentation_base(self): + """Scrubbing back to the archive open must clear the prior scrub window.""" + _seed_pool_session(self.redis, session_id=self.SESSION) + cdn = "http://cdn.example/archive.ts?token=ok" + archive_total = 870_621_184 + stale_base = 348_248_380 + self.redis.hset(self._pool_key(), mapping={ + "final_url": cdn, + "content_length": str(archive_total), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + "presentation_length": str(archive_total - stale_base), + "presentation_byte_base": str(stale_base), + "media_id": "8_2026-06-08-17-14", + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=200) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": "8_2026-06-08-17-14", + "provider_timestamp": "2026-06-08:19-14", + "provider_tz_name": "Europe/Brussels", + "final_url": cdn, + "content_length": str(archive_total), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + "presentation_length": str(archive_total - stale_base), + "presentation_byte_base": str(stale_base), + }, + profile=profile, + channel=self.channel, + media_id=TEST_MEDIA_ID, + safe_ts="2026-06-08-17-00", + timestamp="2026-06-08:17-00", + duration_minutes=35, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=None, + channel_logo_id=None, + user=self.user, + debug=False, + ) + kwargs = attempt_mock.call_args.kwargs + self.assertEqual(kwargs.get("presentation_byte_base"), 0) + self.assertEqual(kwargs.get("presentation_remaining"), archive_total) + self.assertIsNone(kwargs.get("range_header")) + self.assertFalse(kwargs.get("rewrite_plain_get")) + self.assertFalse(kwargs.get("relative_presentation_range")) + + def test_return_to_archive_start_range_skips_stale_presentation_map(self): + """Ranges after return-to-start are archive-absolute, not scrub-relative.""" + _seed_pool_session(self.redis, session_id=self.SESSION) + cdn = "http://cdn.example/archive.ts?token=ok" + archive_total = 870_621_184 + stale_base = 348_248_380 + client_range = f"bytes={archive_total - 112_800}-" + self.redis.hset(self._pool_key(), mapping={ + "final_url": cdn, + "content_length": str(archive_total), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + "presentation_length": str(archive_total - stale_base), + "presentation_byte_base": str(stale_base), + "media_id": TEST_MEDIA_ID, + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=206) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": TEST_MEDIA_ID, + "provider_timestamp": "2026-06-08:19-00", + "provider_tz_name": "Europe/Brussels", + "final_url": cdn, + "content_length": str(archive_total), + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "2100", + "presentation_length": str(archive_total - stale_base), + "presentation_byte_base": str(stale_base), + }, + profile=profile, + channel=self.channel, + media_id=TEST_MEDIA_ID, + safe_ts="2026-06-08-17-00", + timestamp="2026-06-08:17-00", + duration_minutes=35, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=client_range, + channel_logo_id=None, + user=self.user, + debug=False, + ) + kwargs = attempt_mock.call_args.kwargs + self.assertEqual(kwargs.get("range_header"), client_range) + self.assertEqual(kwargs.get("presentation_byte_base"), 0) + self.assertFalse(kwargs.get("relative_presentation_range")) + + def test_session_scrub_outside_window_forces_portal(self): + _seed_pool_session(self.redis, session_id=self.SESSION) + stale_cdn = "http://cdn.example/old.ts?token=stale" + self.redis.hset(self._pool_key(), mapping={ + "final_url": stale_cdn, + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "1800", # 30 min window + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=200) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": TEST_MEDIA_ID, + "provider_timestamp": "2026-06-08:19-00", + "provider_tz_name": "Europe/Brussels", + "final_url": stale_cdn, + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "1800", + }, + profile=profile, + channel=self.channel, + # 40 minutes after anchor, outside the 30 min archive window. + media_id="8_2026-06-08-17-40", + safe_ts="2026-06-08-17-40", + timestamp="2026-06-08:17-40", + duration_minutes=40, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=None, + channel_logo_id=None, + user=self.user, + debug=False, + ) + self.assertIsNone(attempt_mock.call_args.kwargs.get("final_url")) + self.assertIsNone(self.redis.hget(self._pool_key(), "final_url")) + + def test_same_programme_reuse_keeps_final_url(self): + _seed_pool_session(self.redis, session_id=self.SESSION) + cdn = "http://cdn.example/same.ts?token=ok" + self.redis.hset(self._pool_key(), mapping={ + "final_url": cdn, + "content_length": "1000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + }) + profile = MagicMock(id=31, custom_properties={}) + account = MagicMock(id=1) + ok = MagicMock(status_code=200) + with patch.object(views.M3UAccount.objects, "get", return_value=account), \ + patch.object(views, "_attempt_timeshift_stream", + return_value=ok) as attempt_mock: + views._stream_reused_session( + self.redis, + session_id=self.SESSION, + descriptor={ + "account_id": "1", + "stream_id": "111", + "media_id": TEST_MEDIA_ID, + "provider_timestamp": "2026-06-08:19-00", + "provider_tz_name": "Europe/Brussels", + "final_url": cdn, + "content_length": "1000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + }, + profile=profile, + channel=self.channel, + media_id=TEST_MEDIA_ID, + safe_ts="2026-06-08-17-00", + timestamp="2026-06-08:17-00", + duration_minutes=40, + client_id=self.SESSION, + client_ip="1.2.3.4", + client_user_agent="test-agent", + range_header=None, + channel_logo_id=None, + user=self.user, + debug=False, + ) + self.assertEqual(attempt_mock.call_args.kwargs.get("final_url"), cdn) + self.assertFalse(attempt_mock.call_args.kwargs.get("rewrite_plain_get")) + def test_reused_session_serves_requested_timestamp_not_stored_anchor(self): - # Field bug: rewind to X, play, then FF — TiviMate keeps the + # Field bug: rewind to X, play, then FF. Some clients keep the # ?session_id= query when rebuilding the seek URL with a new start, # and the reused session replayed the STORED position, snapping # playback back to the rewind anchor X. The reuse path must always @@ -1998,7 +2474,7 @@ class TimeshiftSessionReuseTests(TestCase): "2026-06-08:17-30", "8_2026-06-08-17-30", ) self.assertIs(response, ok) - # June -> CEST: 17:30 UTC reaches the provider as 19:30 local — never + # June -> CEST: 17:30 UTC reaches the provider as 19:30 local, never # the session's stored 19:00 anchor. self.assertEqual( attempt_mock.call_args.kwargs["provider_timestamp"], @@ -2023,22 +2499,27 @@ class TimeshiftSessionReuseTests(TestCase): ) def test_position_move_drops_previous_byte_state(self): - # content_length / serving_range describe ONE provider file; carrying - # them across a seek would feed the near-EOF/displacement heuristics - # another programme's size. A same-position update must keep them. + # content_length / serving_range / final_url describe ONE provider file; + # carrying them across a seek would feed near-EOF heuristics another + # programme's size and hit the wrong CDN token. Same-position keeps them. _seed_pool_session(self.redis, session_id=self.SESSION) self.redis.hset(self._pool_key(), mapping={ - "content_length": "2000000000", "serving_range": "start", + "content_length": "2000000000", + "serving_range": "start", + "final_url": "http://cdn.example/old.ts", }) self._call_reused_session( "2026-06-08:17-30", "8_2026-06-08-17-30", ) self.assertIsNone(self.redis.hget(self._pool_key(), "content_length")) self.assertIsNone(self.redis.hget(self._pool_key(), "serving_range")) + self.assertIsNone(self.redis.hget(self._pool_key(), "final_url")) # Same-position call (media unchanged): byte state survives. self.redis.hset(self._pool_key(), mapping={ - "content_length": "1000000", "serving_range": "range", + "content_length": "1000000", + "serving_range": "range", + "final_url": "http://cdn.example/same.ts", }) self._call_reused_session( "2026-06-08:17-30", "8_2026-06-08-17-30", @@ -2046,6 +2527,10 @@ class TimeshiftSessionReuseTests(TestCase): self.assertEqual( self.redis.hget(self._pool_key(), "content_length"), "1000000", ) + self.assertEqual( + self.redis.hget(self._pool_key(), "final_url"), + "http://cdn.example/same.ts", + ) def test_position_update_never_resurrects_vanished_entry(self): # If the pool key expired/vanished mid-request, writing to it would @@ -2458,7 +2943,7 @@ class FakeRedisScanTests(TestCase): class TimeshiftRangeClassificationTests(TestCase): - """Startup probes must not be treated as scrubs.""" + """Range classification and presentation helpers.""" def test_full_file_request_is_not_displacing(self): self.assertFalse(views._should_displace_busy_playback(None)) @@ -2469,6 +2954,7 @@ class TimeshiftRangeClassificationTests(TestCase): ) def test_bytes_zero_does_not_displace_active_start_stream(self): + # Scrub rule still false; residual same-session path preempts at serve time. self.assertFalse( views._should_displace_busy_playback("bytes=0-", busy_serving_range="start") ) @@ -2480,6 +2966,83 @@ class TimeshiftRangeClassificationTests(TestCase): self.assertTrue(views._is_near_eof_probe("bytes=2527702896-")) self.assertFalse(views._should_displace_busy_playback("bytes=2527702896-")) + def test_cap_open_ended_range_limits_probe_span(self): + self.assertEqual( + views._cap_open_ended_range("bytes=1000-", 100), + "bytes=1000-1099", + ) + self.assertEqual( + views._cap_open_ended_range("bytes=1000-2000", 100), + "bytes=1000-2000", + ) + + def test_resolve_session_archive_scrub_maps_ff_offset(self): + scrub = views._resolve_session_archive_scrub( + { + "final_url": "http://cdn/x", + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "3600", + }, + "2026-06-08:17-30", + ) + self.assertEqual(scrub["kind"], "scrub") + # 30 minutes into 60 minutes ≈ half file, aligned to 188-byte TS packets. + self.assertEqual(scrub["byte_offset"] % 188, 0) + self.assertAlmostEqual(scrub["byte_offset"] / 1800000000, 0.5, places=2) + self.assertEqual(scrub["remaining"], 1800000000 - scrub["byte_offset"]) + + def test_resolve_session_archive_scrub_rejects_outside_window(self): + self.assertIsNone( + views._resolve_session_archive_scrub( + { + "final_url": "http://cdn/x", + "content_length": "1800000000", + "archive_anchor_ts": "2026-06-08:17-00", + "archive_duration_secs": "1800", + }, + "2026-06-08:17-40", + ) + ) + + def test_map_client_range_through_presentation(self): + self.assertEqual( + views._map_client_range_through_presentation( + "bytes=419800872-", 314934968, + ), + "bytes=734735840-", + ) + self.assertEqual( + views._map_client_range_through_presentation( + "bytes=100-200", 1000, + ), + "bytes=1100-1200", + ) + + def test_presentation_relative_content_range(self): + self.assertEqual( + views._presentation_relative_content_range( + "bytes 314934968-734848639/734848640", + presentation_byte_base=314934968, + presentation_length=419913672, + ), + "bytes 0-419913671/419913672", + ) + + def test_near_eof_probe_uses_presentation_length_not_full_archive(self): + # After scrub rewrite CL≈420MB; Range near that end must not displace + # (XC parallel probe) even though the offset is mid-file on the CDN archive. + self.assertTrue( + views._is_near_eof_probe( + "bytes=419800872-", content_length="419913672", + ) + ) + self.assertFalse( + views._should_displace_busy_playback( + "bytes=419800872-", content_length="419913672", + ) + ) + def test_near_eof_probe_uses_cached_content_length(self): # 5 MB into a 10 MB file is a scrub, not a tail probe. self.assertFalse( @@ -2488,10 +3051,21 @@ class TimeshiftRangeClassificationTests(TestCase): self.assertTrue( views._should_displace_busy_playback("bytes=5000000-", content_length="10000000") ) - # Within 512 KB of EOF is a tail probe once length is known. + # Within 2 MiB of EOF (incl. common 1.88MB / 10000-packet probes). self.assertTrue( views._is_near_eof_probe("bytes=9990000-", content_length="10000000") ) + total = 8_783_238_116 + self.assertTrue( + views._is_near_eof_probe( + f"bytes={total - 1_880_000}-", content_length=str(total), + ) + ) + self.assertFalse( + views._should_displace_busy_playback( + f"bytes={total - 1_880_000}-", content_length=str(total), + ) + ) def test_midfile_seek_is_displacing(self): self.assertTrue(views._should_displace_busy_playback("bytes=5000000-")) @@ -2545,6 +3119,31 @@ class TimeshiftRangeClassificationTests(TestCase): ), ) + def test_active_playback_seek_always_preempts(self): + # Heartbeats keep last_activity fresh; seek during play must still preempt. + redis = _FakeRedis() + _seed_pool_session( + redis, session_id=TEST_SESSION_ID, + media_id="8_2026-06-08-17-00", + ) + redis.hset( + views._pool_key(TEST_SESSION_ID), + "last_activity", + str(views.time.time()), + ) + user = MagicMock(id=5) + with patch.object( + views, "_session_has_active_timeshift_stream", return_value=True, + ): + self.assertTrue( + views._should_preempt_for_programme_change( + redis, TEST_SESSION_ID, + "8_2026-06-08-17-00", + "8_2026-06-08-17-30", + user=user, + ), + ) + class TimeshiftStatsClientTests(TestCase): def setUp(self): @@ -3945,13 +4544,10 @@ class TimeshiftScrubPreemptTests(TestCase): request, "u", "p", "8", "2026-06-08:17-00", "8.ts", ) self.assertIs(response, ok) - reacquire_mock.assert_called_once_with( - self.redis, existing, user_id=5, user=self.user, - wait_seconds=views._POOL_SCRUB_WAIT_SECONDS, - ) + reacquire_mock.assert_called_once() def test_plain_reconnect_preempts_busy_pool_without_range(self): - """TiviMate FF reconnects with plain GET; match provider byte-0 restart.""" + """Plain GET reconnect should match provider byte-0 restart.""" _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) request = self.factory.get(_proxy_url(TEST_SESSION_ID)) streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] @@ -4115,6 +4711,7 @@ class TimeshiftScrubPreemptTests(TestCase): attempt_mock.assert_not_called() def test_eof_probe_deferred_without_preempt(self): + """Near-EOF probes without a cached CDN URL still get 503 (no preempt).""" _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) request = self.factory.get( _proxy_url(TEST_SESSION_ID), @@ -4133,7 +4730,9 @@ class TimeshiftScrubPreemptTests(TestCase): patch.object(views, "release_profile_slot"), \ patch.object(views, "get_transformed_credentials", side_effect=_fake_creds), \ patch.object(views, "get_user_active_connections", return_value=[]), \ + patch.object(views, "_try_reacquire_idle_pool") as reacquire_mock, \ patch.object(views, "_preempt_playback_streams") as preempt_mock, \ + patch.object(views, "_open_upstream") as open_mock, \ patch.object(views, "_attempt_timeshift_stream") as attempt_mock: redis_cls.get_client.return_value = self.redis channel_cls.objects.get.return_value = MagicMock( @@ -4143,8 +4742,221 @@ class TimeshiftScrubPreemptTests(TestCase): request, "u", "p", "8", "2026-06-08:17-00", "8.ts", ) self.assertEqual(response.status_code, 503) + reacquire_mock.assert_not_called() + preempt_mock.assert_not_called() + open_mock.assert_not_called() + attempt_mock.assert_not_called() + + def test_eof_probe_busy_session_serves_cached_cdn_without_preempt(self): + """Busy near-EOF duration probes answer from cached CDN, no slot churn.""" + presentation_base = 500_000_000 + presentation_length = 615_251_824 + client_start = 615_139_024 + cdn_start = presentation_base + client_start + body = b"probe-tail" + cdn_end = cdn_start + len(body) - 1 + archive_total = 2_527_702_896 + + _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) + pool_key = TimeshiftRedisKeys.pool(TEST_SESSION_ID) + self.redis.hset(pool_key, mapping={ + "final_url": "http://cdn.example.test/archive.ts", + "content_length": str(archive_total), + "presentation_byte_base": str(presentation_base), + "presentation_length": str(presentation_length), + "provider_user_agent": "provider-agent", + "busy": "1", + }) + + request = self.factory.get( + _proxy_url(TEST_SESSION_ID), + HTTP_RANGE=f"bytes={client_start}-", + ) + streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] + upstream = _fake_upstream(206, body=body) + upstream.headers["Content-Range"] = ( + f"bytes {cdn_start}-{cdn_end}/{archive_total}" + ) + upstream.headers["Content-Length"] = str(len(body)) + upstream.raw.read = MagicMock(side_effect=[body, b""]) + with patch.object(views, "_authenticate_user", return_value=MagicMock(id=5)), \ + 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=streams), \ + patch.object(views, "get_programme_duration", return_value=40), \ + patch.object(views, "check_user_stream_limits", return_value=True), \ + patch.object(views, "RedisClient") as redis_cls, \ + patch.object(views, "reserve_profile_slot") as reserve_mock, \ + patch.object(views, "release_profile_slot") as release_mock, \ + patch.object(views, "get_transformed_credentials", side_effect=_fake_creds), \ + patch.object(views, "get_user_active_connections", return_value=[]), \ + patch.object(views, "_try_reacquire_idle_pool") as reacquire_mock, \ + patch.object(views, "_preempt_playback_streams") as preempt_mock, \ + patch.object(views, "_register_stats_client") as stats_mock, \ + patch.object(views, "_attempt_timeshift_stream") as attempt_mock, \ + patch.object(views.M3UAccount.objects, "get") as account_get_mock, \ + patch.object(views, "_open_upstream", return_value=upstream) as open_mock: + redis_cls.get_client.return_value = self.redis + channel_cls.objects.get.return_value = MagicMock( + id=8, name="Test", logo_id=None, + ) + response = views.timeshift_proxy( + request, "u", "p", "8", "2026-06-08:17-00", "8.ts", + ) + + self.assertEqual(response.status_code, 206) + self.assertEqual( + response["Content-Range"], + f"bytes {client_start}-{client_start + len(body) - 1}/{presentation_length}", + ) + self.assertEqual(response["Content-Length"], str(len(body))) + open_mock.assert_called_once() + open_args, open_kwargs = open_mock.call_args + self.assertEqual(open_args[0], "http://cdn.example.test/archive.ts") + self.assertEqual(open_args[1], "provider-agent") + self.assertEqual( + open_args[2], + f"bytes={cdn_start}-{cdn_start + views._EOF_PROBE_TAIL_BYTES - 1}", + ) + self.assertFalse(open_kwargs.get("allow_redirects", True)) + account_get_mock.assert_not_called() + reacquire_mock.assert_not_called() preempt_mock.assert_not_called() attempt_mock.assert_not_called() + stats_mock.assert_not_called() + reserve_mock.assert_not_called() + release_mock.assert_not_called() + self.assertEqual(self.redis.hget(pool_key, "busy"), "1") + # Drain the short probe body so the generator finishes cleanly. + self.assertEqual(b"".join(response.streaming_content), body) + upstream.close.assert_called() + + def test_eof_probe_stale_presentation_base_uses_absolute_range(self): + """After return-to-start, archive-absolute EOF probes must not remap past EOF.""" + stale_base = 348_248_380 + archive_total = 870_621_184 + client_start = archive_total - 112_800 + body = b"probe-tail" + + _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) + pool_key = TimeshiftRedisKeys.pool(TEST_SESSION_ID) + self.redis.hset(pool_key, mapping={ + "final_url": "http://cdn.example.test/archive.ts", + "content_length": str(archive_total), + # Stale scrub window left behind after return-to-start. + "presentation_byte_base": str(stale_base), + "presentation_length": str(archive_total), + "provider_user_agent": "provider-agent", + "busy": "1", + }) + + request = self.factory.get( + _proxy_url(TEST_SESSION_ID), + HTTP_RANGE=f"bytes={client_start}-", + ) + streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] + upstream = _fake_upstream(206, body=body) + upstream.headers["Content-Range"] = ( + f"bytes {client_start}-{client_start + len(body) - 1}/{archive_total}" + ) + upstream.headers["Content-Length"] = str(len(body)) + upstream.raw.read = MagicMock(side_effect=[body, b""]) + + with patch.object(views, "_authenticate_user", return_value=MagicMock(id=5)), \ + 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=streams), \ + patch.object(views, "get_programme_duration", return_value=40), \ + patch.object(views, "check_user_stream_limits", return_value=True), \ + patch.object(views, "RedisClient") as redis_cls, \ + patch.object(views, "reserve_profile_slot"), \ + patch.object(views, "release_profile_slot"), \ + patch.object(views, "get_transformed_credentials", side_effect=_fake_creds), \ + patch.object(views, "get_user_active_connections", return_value=[]), \ + patch.object(views, "_try_reacquire_idle_pool") as reacquire_mock, \ + patch.object(views, "_preempt_playback_streams") as preempt_mock, \ + patch.object(views, "_attempt_timeshift_stream") as attempt_mock, \ + patch.object(views.M3UAccount.objects, "get") as account_get_mock, \ + patch.object(views, "_open_upstream", return_value=upstream) as open_mock: + redis_cls.get_client.return_value = self.redis + channel_cls.objects.get.return_value = MagicMock( + id=8, name="Test", logo_id=None, + ) + response = views.timeshift_proxy( + request, "u", "p", "8", "2026-06-08:17-00", "8.ts", + ) + + self.assertEqual(response.status_code, 206) + open_args, _open_kwargs = open_mock.call_args + # Must NOT be stale_base + client_start (that is past archive EOF). + self.assertEqual(open_args[1], "provider-agent") + self.assertEqual( + open_args[2], + f"bytes={client_start}-{client_start + views._EOF_PROBE_TAIL_BYTES - 1}", + ) + account_get_mock.assert_not_called() + reacquire_mock.assert_not_called() + preempt_mock.assert_not_called() + attempt_mock.assert_not_called() + self.assertEqual(b"".join(response.streaming_content), body) + + def test_eof_probe_cdn_416_returns_416_not_503(self): + """Unsatisfiable EOF probes must not fall through to busy-slot 503.""" + archive_total = 870_621_184 + client_start = archive_total - 112_800 + + _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) + pool_key = TimeshiftRedisKeys.pool(TEST_SESSION_ID) + self.redis.hset(pool_key, mapping={ + "final_url": "http://cdn.example.test/archive.ts", + "content_length": str(archive_total), + "presentation_byte_base": "0", + "presentation_length": str(archive_total), + "provider_user_agent": "provider-agent", + "busy": "1", + }) + + request = self.factory.get( + _proxy_url(TEST_SESSION_ID), + HTTP_RANGE=f"bytes={client_start}-", + ) + streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)] + upstream = MagicMock() + upstream.status_code = 416 + upstream.headers = {} + upstream.close = MagicMock() + + with patch.object(views, "_authenticate_user", return_value=MagicMock(id=5)), \ + 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=streams), \ + patch.object(views, "get_programme_duration", return_value=40), \ + patch.object(views, "check_user_stream_limits", return_value=True), \ + patch.object(views, "RedisClient") as redis_cls, \ + patch.object(views, "reserve_profile_slot"), \ + patch.object(views, "release_profile_slot"), \ + patch.object(views, "get_transformed_credentials", side_effect=_fake_creds), \ + patch.object(views, "get_user_active_connections", return_value=[]), \ + patch.object(views, "_attempt_timeshift_stream") as attempt_mock, \ + patch.object(views.M3UAccount.objects, "get") as account_get_mock, \ + patch.object(views, "_open_upstream", return_value=upstream) as open_mock: + redis_cls.get_client.return_value = self.redis + channel_cls.objects.get.return_value = MagicMock( + id=8, name="Test", logo_id=None, + ) + response = views.timeshift_proxy( + request, "u", "p", "8", "2026-06-08:17-00", "8.ts", + ) + + self.assertEqual(response.status_code, 416) + self.assertEqual(response["Content-Range"], f"bytes */{archive_total}") + self.assertEqual(open_mock.call_args.args[1], "provider-agent") + account_get_mock.assert_not_called() + attempt_mock.assert_not_called() + upstream.close.assert_called() def test_scrub_opens_failover_when_pool_still_busy(self): _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) @@ -4394,7 +5206,7 @@ class CatchupProxyTests(TestCase): class TimeshiftProviderFaithfulPlainGetTests(_ProxyLoopTestMixin, TestCase): - """Contract tests for TiviMate-style plain GET reconnect (no Range header).""" + """Contract tests for plain GET reconnect (no Range header).""" def setUp(self): self.redis = _FakeRedis() @@ -4417,7 +5229,7 @@ class TimeshiftProviderFaithfulPlainGetTests(_ProxyLoopTestMixin, TestCase): stats_channel_id=self.stats_channel_id, client_id=self.client_id, client_ip="1.2.3.4", - client_user_agent="tivimate", + client_user_agent="test-agent", user=self.user, channel_display_name="A&E", timestamp_utc="2026-06-08:17-00", @@ -4453,7 +5265,7 @@ class TimeshiftProviderFaithfulPlainGetTests(_ProxyLoopTestMixin, TestCase): stats_channel_id=self.stats_channel_id, client_id=self.client_id, client_ip="1.2.3.4", - client_user_agent="tivimate", + client_user_agent="test-agent", user=self.user, channel_display_name="A&E", timestamp_utc="2026-06-08:17-00", diff --git a/apps/timeshift/views.py b/apps/timeshift/views.py index 71717e36..5b8a9953 100644 --- a/apps/timeshift/views.py +++ b/apps/timeshift/views.py @@ -62,7 +62,11 @@ from .helpers import ( parse_catchup_timestamp, ) from .sessions import catchup_session_exists, delete_catchup_session, resolve_catchup_playback -from .stats import resolve_stats_playback_fields, seed_stream_stats_metadata +from .stats import ( + EOF_PROBE_TAIL_BYTES, + resolve_stats_playback_fields, + seed_stream_stats_metadata, +) logger = logging.getLogger(__name__) @@ -98,7 +102,7 @@ def _finalize_timeshift_response(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. - URL shape (iPlayTV / TiviMate): + URL shape (XC catch-up clients): ``stream_id``: EPG channel number (ignored here). ``duration``: Dispatcharr ``Channel.id`` (XC API exposes channel.id as stream_id). ``timestamp``: UTC programme start (``YYYY-MM-DD:HH-MM`` or XC colon form @@ -151,7 +155,7 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration) "redirect round trip.\n\n" "**Plain GET (no ``Range``):** upstream archive is streamed from byte " "0 with ``Content-Length`` when the provider reports file size " - "(provider-faithful behaviour for clients such as TiviMate)." + "(provider-faithful behaviour for XC-style IPTV clients)." ), parameters=[ OpenApiParameter( @@ -377,15 +381,22 @@ def _serve_catchup(request, user, channel, timestamp): pool_content_length = pool["content_length"] if pool else None busy_serving_range = pool["serving_range"] if pool else None pool_media_id = None + pool_presentation_length = None if pool and pool.get("entry"): pool_media_id = pool["entry"].get("media_id") + pool_presentation_length = pool["entry"].get("presentation_length") elif session_entry: pool_media_id = session_entry.get("media_id") + pool_presentation_length = session_entry.get("presentation_length") + # After a scrub rewrite the client sees a shorter file; EOF probes and + # Range seeks are relative to that presentation, not the full CDN archive. + probe_length = pool_presentation_length or pool_content_length + scrub_displacement = ( pool_exists and _should_displace_busy_pool( range_header, - pool_content_length, + probe_length, busy_serving_range, pool_media_id=pool_media_id, media_id=media_id, @@ -407,6 +418,7 @@ def _serve_catchup(request, user, channel, timestamp): ) elif _should_preempt_for_programme_change( redis_client, effective_session_id, pool_media_id, media_id, + user=user, ): acquired = _try_reacquire_idle_pool( redis_client, effective_session_id, @@ -482,6 +494,21 @@ def _serve_catchup(request, user, channel, timestamp): pass if pool_exists and pool_busy and acquired is None: + probe_entry = None + if pool and pool.get("entry"): + probe_entry = pool["entry"] + elif session_entry: + probe_entry = session_entry + probe_response = _try_serve_busy_eof_probe( + redis_client=redis_client, + session_id=effective_session_id, + entry=probe_entry, + range_header=range_header, + probe_length=probe_length, + debug=debug, + ) + if probe_response is not None: + return probe_response logger.debug( "Timeshift: deferring busy session %s range=%s media=%s pool_media=%s", effective_session_id, range_header or "(none)", media_id, pool_media_id, @@ -675,7 +702,8 @@ _POOL_WAIT_SECONDS = 1.0 # Brief wait after scrub preempt before failover. _POOL_SCRUB_WAIT_SECONDS = 2.0 _POOL_POLL_INTERVAL = 0.05 -_EOF_PROBE_TAIL_BYTES = 512_000 +# Near-EOF duration probes (~10000 MPEG-TS packets / 1.88MB). Shared with stats. +_EOF_PROBE_TAIL_BYTES = EOF_PROBE_TAIL_BYTES _EOF_PROBE_UNKNOWN_LENGTH_MIN = 100_000_000 @@ -1135,7 +1163,7 @@ def _build_downstream_length_headers( headers["Content-Length"] = str(up_end - up_start + 1) return headers - # Plain GET streaming 200: match provider/CDN Content-Length (TiviMate uses + # Plain GET streaming 200: match provider/CDN Content-Length (clients use # it for archive duration; omitting it breaks FF reconnect playback). if streaming and status_code == 200 and not range_header: if representation_length is not None: @@ -1187,18 +1215,26 @@ def _should_displace_busy_pool( def _should_preempt_for_programme_change( - redis_client, session_id, pool_media_id, media_id, + redis_client, session_id, pool_media_id, media_id, *, user=None, ): - """True when the viewer moved to a different programme on the same session.""" + """True when the viewer moved to a different start on the same session. + + XC clients often rebuild the catch-up URL with a new ``start`` while keeping + ``session_id``. During active playback that must always preempt the + in-flight stream (heartbeats keep ``last_activity`` fresh, so the old ~2s + startup-probe window was returning 503 on every seek). + """ if pool_media_id is None or str(pool_media_id) == str(media_id): return False + if user is not None and _session_has_active_timeshift_stream(user, session_id): + return True try: last_activity = float( _get_pool_entry(redis_client, session_id).get("last_activity") or 0 ) except (TypeError, ValueError): last_activity = 0 - # Ignore parallel startup probes within ~2s. + # Parallel startup probes with no live stream yet: ignore brief media churn. return (time.time() - last_activity) >= _STATS_GRACE_MIN_ELAPSED_SECONDS @@ -1224,9 +1260,9 @@ def _should_preempt_plain_reconnect( ): """True when a plain GET should restart playback like the provider does. - TiviMate fast-forward closes the HTTP connection and reopens the same - programme URL without a Range header. Providers answer that with a full - 200 from byte 0, not an internal byte-range resume. + IPTV clients often close the HTTP connection and reopen the same programme + URL without a Range header. Providers answer + that with a full 200 from byte 0, not an internal byte-range resume. """ if range_header: return False @@ -1254,6 +1290,189 @@ def _should_displace_busy_playback( return True +def _cap_open_ended_range(range_header, max_span_bytes): + """Limit an open-ended ``bytes=START-`` Range to at most ``max_span_bytes``.""" + parsed = _parse_client_range(range_header) + if parsed is None: + return range_header + start, end = parsed + if end is not None or max_span_bytes is None or max_span_bytes <= 0: + return range_header + return f"bytes={start}-{start + int(max_span_bytes) - 1}" + + +def _try_serve_busy_eof_probe( + *, + redis_client, + session_id, + entry, + range_header, + probe_length, + debug=False, +): + """Serve a near-EOF duration probe via cached CDN without preempting playback. + + Returns a response, or ``None`` to fall through to busy ``503``. + No pool/stats/profile side effects. Caps open-ended Ranges so a probe + cannot pull the remainder of a multi-GB archive. + """ + if not entry or not range_header: + return None + if not _is_near_eof_probe(range_header, probe_length): + return None + + final_url = entry.get("final_url") + if isinstance(final_url, bytes): + final_url = final_url.decode() + if not final_url: + return None + + content_length = _pool_int_field(entry.get("content_length")) + presentation_base = _pool_int_field(entry.get("presentation_byte_base")) + presentation_length = _pool_int_field(entry.get("presentation_length")) + effective_range = range_header + relative_presentation = False + if presentation_base and presentation_base > 0: + mapped_range = _map_client_range_through_presentation( + range_header, presentation_base, + ) + mapped_start = _parse_range_start(mapped_range) + client_start = _parse_range_start(range_header) + # Stale scrub base: client Range is already archive-absolute. + if ( + content_length is not None + and mapped_start is not None + and mapped_start >= content_length + and client_start is not None + and client_start < content_length + ): + effective_range = range_header + relative_presentation = False + presentation_length = content_length + else: + effective_range = mapped_range + relative_presentation = True + + effective_range = _cap_open_ended_range(effective_range, _EOF_PROBE_TAIL_BYTES) + + user_agent = entry.get("provider_user_agent") or "" + if isinstance(user_agent, bytes): + user_agent = user_agent.decode() + + try: + upstream = _open_upstream( + final_url, user_agent, effective_range, allow_redirects=False, + ) + except Exception as exc: + logger.debug( + "Timeshift EOF probe CDN open failed session=%s: %s", + session_id, exc, + ) + return None + + if upstream.status_code == 416: + try: + upstream.close() + except Exception: + pass + total = ( + presentation_length + if relative_presentation and presentation_length + else content_length + ) + if total is None: + total = _pool_int_field(probe_length) + response = HttpResponse(status=416) + response["Accept-Ranges"] = "bytes" + if total is not None: + response["Content-Range"] = f"bytes */{int(total)}" + if debug: + logger.debug( + "Timeshift EOF probe 416: session=%s client_range=%s cdn_range=%s", + session_id, range_header, effective_range, + ) + return _finalize_timeshift_response(response) + + if upstream.status_code not in (200, 206): + try: + upstream.close() + except Exception: + pass + logger.debug( + "Timeshift EOF probe CDN rejected session=%s status=%s", + session_id, getattr(upstream, "status_code", None), + ) + return None + + content_type = upstream.headers.get("Content-Type", "video/mp2t") + status = upstream.status_code + content_range = upstream.headers.get("Content-Range", "") or None + if relative_presentation and content_range and presentation_length is not None: + content_range = _presentation_relative_content_range( + content_range, + presentation_byte_base=presentation_base, + presentation_length=presentation_length, + ) + + client_headers = _build_downstream_length_headers( + range_header=None if relative_presentation else range_header, + status_code=status, + representation_length=( + presentation_length + if relative_presentation and presentation_length is not None + else _extract_representation_length(upstream) + ), + upstream_content_range=content_range, + upstream_content_length=upstream.headers.get("Content-Length"), + streaming=True, + ) + + chunk_size = max(ConfigHelper.chunk_size(), 262144) + closed = {"done": False} + + def _finish(): + if closed["done"]: + return + closed["done"] = True + try: + upstream.close() + except Exception: + pass + + def _generator(): + try: + while True: + try: + chunk = upstream.raw.read(chunk_size) + except Exception: + break + if not chunk: + break + yield chunk + except GeneratorExit: + pass + finally: + _finish() + + if debug: + logger.debug( + "Timeshift EOF probe pass-through: session=%s client_range=%s " + "cdn_range=%s status=%d", + session_id, range_header, effective_range, status, + ) + + stream_iter = _SlotReleasingStream(_generator(), _finish) + response = StreamingHttpResponse( + stream_iter, + content_type=content_type, + status=status, + ) + response["X-Accel-Buffering"] = "no" + for header_name, header_value in client_headers.items(): + response[header_name] = header_value + return _finalize_timeshift_response(response) + + def _score_pool_fingerprint(entry, client_ip, client_user_agent): """Score IP/UA overlap for fingerprint adoption (user and media pre-filtered).""" score = 0 @@ -1402,8 +1621,15 @@ def _store_pool_serving_range(redis_client, session_id, range_header): logger.debug("Timeshift pool serving_range store failed: %s", exc) -def _update_pool_position(redis_client, session_id, *, media_id, provider_timestamp): - """Move a reused session's descriptor to the position actually served.""" +def _update_pool_position( + redis_client, session_id, *, media_id, provider_timestamp, keep_archive=False, +): + """Move a reused session's descriptor to the position actually served. + + XC FF/RW rebuilds the catch-up URL with a new *start* timestamp (same + session). That changes ``media_id`` but is still the same opened CDN + archive; keep ``final_url`` / sizes when *keep_archive* is True. + """ if redis_client is None or not session_id: return key = _pool_key(session_id) @@ -1419,8 +1645,18 @@ def _update_pool_position(redis_client, session_id, *, media_id, provider_timest "media_id": str(media_id), "provider_timestamp": str(provider_timestamp), }) - if position_changed: - redis_client.hdel(key, "content_length", "serving_range") + if position_changed and not keep_archive: + # Left the opened archive window; drop file-specific state. + redis_client.hdel( + key, + "content_length", + "serving_range", + "final_url", + "archive_anchor_ts", + "archive_duration_secs", + "presentation_length", + "presentation_byte_base", + ) except Exception as exc: logger.debug("Timeshift pool position update failed: %s", exc) @@ -1439,6 +1675,178 @@ def _store_pool_content_length(redis_client, session_id, upstream_response): logger.debug("Timeshift pool content_length store failed: %s", exc) +def _store_pool_final_url(redis_client, session_id, final_url): + """Persist the post-redirect CDN URL for VOD-style reconnect reuse.""" + if redis_client is None or not session_id or not final_url: + return + try: + redis_client.hset(_pool_key(session_id), "final_url", str(final_url)) + except Exception as exc: + logger.debug("Timeshift pool final_url store failed: %s", exc) + + +def _store_pool_provider_user_agent(redis_client, session_id, user_agent): + """Snapshot the resolved account User-Agent for this playback session.""" + if redis_client is None or not session_id: + return + key = _pool_key(session_id) + try: + if redis_client.exists(key): + redis_client.hset(key, "provider_user_agent", str(user_agent or "")) + except Exception as exc: + logger.debug("Timeshift pool provider User-Agent store failed: %s", exc) + + +def _clear_pool_final_url(redis_client, session_id): + if redis_client is None or not session_id: + return + try: + redis_client.hdel(_pool_key(session_id), "final_url") + except Exception as exc: + logger.debug("Timeshift pool final_url clear failed: %s", exc) + + +def _store_pool_presentation_window( + redis_client, session_id, length, *, byte_base=0, +): + """Client-visible size/origin after a scrub rewrite (relative Ranges map here).""" + if redis_client is None or not session_id or length is None: + return + try: + redis_client.hset( + _pool_key(session_id), + mapping={ + "presentation_length": str(int(length)), + "presentation_byte_base": str(int(byte_base or 0)), + }, + ) + except Exception as exc: + logger.debug("Timeshift pool presentation window store failed: %s", exc) + + +def _pool_int_field(value): + if value is None or value == "": + return None + try: + if isinstance(value, bytes): + value = value.decode() + return int(value) + except (TypeError, ValueError): + return None + + +def _map_client_range_through_presentation(range_header, presentation_byte_base): + """Translate presentation-relative Range into absolute CDN archive Range.""" + if presentation_byte_base is None or not range_header: + return range_header + parsed = _parse_client_range(range_header) + if parsed is None: + return range_header + start, end = parsed + base = int(presentation_byte_base) + abs_start = base + start + if end is None: + return f"bytes={abs_start}-" + return f"bytes={abs_start}-{base + end}" + + +def _presentation_relative_content_range( + upstream_content_range, *, presentation_byte_base, presentation_length, +): + """Rewrite absolute CDN Content-Range into presentation-relative coords.""" + if ( + not upstream_content_range + or presentation_byte_base is None + or presentation_length is None + ): + return upstream_content_range + parsed = _parse_content_range_header(upstream_content_range) + if not parsed or parsed.get("end") is None: + return upstream_content_range + base = int(presentation_byte_base) + rel_start = parsed["start"] - base + rel_end = parsed["end"] - base + if rel_start < 0 or rel_end < rel_start: + return upstream_content_range + return f"bytes {rel_start}-{rel_end}/{int(presentation_length)}" + + +def _ensure_pool_archive_anchor( + redis_client, session_id, *, timestamp, duration_minutes, force=False, +): + """Remember the CDN archive window opened for this session (first portal hit).""" + if redis_client is None or not session_id or not timestamp: + return + key = _pool_key(session_id) + try: + if not force and redis_client.hget(key, "archive_anchor_ts"): + return + duration_secs = max(int(float(duration_minutes or 0) * 60), 60) + redis_client.hset(key, mapping={ + "archive_anchor_ts": str(timestamp), + "archive_duration_secs": str(duration_secs), + }) + except Exception as exc: + logger.debug("Timeshift pool archive anchor store failed: %s", exc) + + +def _resolve_session_archive_scrub(descriptor, requested_timestamp): + """Map an in-session timestamp rebuild onto the open CDN archive. + + XC clients often rebuild ``/timeshift/...//...`` while keeping + ``session_id`` instead of Range-seeking. Same session + in-window start + maps to a byte offset into the already-open CDN file (no portal). + + Returns ``None`` when the request is outside the opened window (new portal + required), or a dict with ``kind`` ``same`` / ``scrub``, ``byte_offset``, + and ``remaining``. + """ + if not descriptor or not requested_timestamp: + return None + final_url = descriptor.get("final_url") or "" + if isinstance(final_url, bytes): + final_url = final_url.decode() + if not final_url: + return None + try: + content_length = int(descriptor.get("content_length")) + duration_secs = float(descriptor.get("archive_duration_secs")) + except (TypeError, ValueError): + return None + if content_length <= 0 or duration_secs <= 0: + return None + anchor_raw = descriptor.get("archive_anchor_ts") + if isinstance(anchor_raw, bytes): + anchor_raw = anchor_raw.decode() + requested_dt = parse_catchup_timestamp(requested_timestamp) + anchor_dt = parse_catchup_timestamp(anchor_raw) if anchor_raw else None + if requested_dt is None or anchor_dt is None: + return None + offset_secs = (requested_dt - anchor_dt).total_seconds() + if abs(offset_secs) < 1.0: + return { + "kind": "same", + "byte_offset": 0, + "remaining": content_length, + } + # Reverse seek before the opened CDN file: content is not on this URL. + # Caller must portal/re-anchor (cannot Range-seek earlier than the open). + if offset_secs < 0: + return None + if offset_secs >= duration_secs: + return None + byte_offset = int((offset_secs / duration_secs) * content_length) + byte_offset = (byte_offset // 188) * 188 + remaining = content_length - byte_offset + if remaining <= 0: + return None + return { + "kind": "scrub", + "byte_offset": byte_offset, + "remaining": remaining, + } + + def _pool_lock(redis_client, session_id): return redis_client.lock( TimeshiftRedisKeys.pool_lock(session_id), @@ -1995,6 +2403,11 @@ def _attempt_timeshift_stream( pool_session_id=None, stats_stream_id=None, stream_stats=None, + final_url=None, + rewrite_plain_get=False, + presentation_remaining=None, + presentation_byte_base=None, + relative_presentation_range=False, ): """Build the provider URL set for one (account, profile, stream) and stream it.""" server_url, xc_username, xc_password = get_transformed_credentials( @@ -2009,6 +2422,9 @@ def _attempt_timeshift_stream( user_agent = m3u_account.get_user_agent().user_agent except AttributeError: user_agent = "" + _store_pool_provider_user_agent( + redis_client, pool_session_id, user_agent, + ) virtual_channel_id = make_virtual_channel_id( channel.id, safe_ts, stream_id_value, @@ -2018,10 +2434,12 @@ def _attempt_timeshift_stream( if debug: logger.debug( "Timeshift attempt: channel=%s ts=%s (provider tz=%s -> %s) " - "account=%s profile=%s provider_sid=%s vid=%s client=%s range=%s", + "account=%s profile=%s provider_sid=%s vid=%s client=%s range=%s " + "cdn=%s", channel.name, timestamp, provider_tz_name, provider_timestamp, m3u_account.id, profile.id, stream_id_value, virtual_channel_id, client_id, range_header or "(none)", + "cached" if final_url else "portal", ) return _stream_from_provider( @@ -2048,6 +2466,11 @@ def _attempt_timeshift_stream( stats_stream_id=stats_stream_id, stream_stats=stream_stats, duration_minutes=duration_minutes, + final_url=final_url, + rewrite_plain_get=rewrite_plain_get, + presentation_remaining=presentation_remaining, + presentation_byte_base=presentation_byte_base, + relative_presentation_range=relative_presentation_range, ) @@ -2087,11 +2510,99 @@ def _stream_reused_session( 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 - # matching and same-channel displacement keep working after the seek. + prior_media_id = descriptor.get("media_id") + if isinstance(prior_media_id, bytes): + prior_media_id = prior_media_id.decode() + media_changed = str(prior_media_id or "") != str(media_id) + + scrub_info = _resolve_session_archive_scrub(descriptor, timestamp) + raw_final_url = descriptor.get("final_url") + if isinstance(raw_final_url, bytes): + raw_final_url = raw_final_url.decode() + rewrite_plain_get = False + presentation_remaining = None + presentation_byte_base = None + relative_presentation_range = False + effective_range = range_header + prior_presentation_base = _pool_int_field( + descriptor.get("presentation_byte_base"), + ) + prior_presentation_length = _pool_int_field( + descriptor.get("presentation_length"), + ) + + if scrub_info is not None: + # Same opened CDN archive: FF within the file opened for this session. + keep_archive = True + final_url = raw_final_url or None + if scrub_info["kind"] == "scrub" and not range_header: + effective_range = f"bytes={scrub_info['byte_offset']}-" + rewrite_plain_get = True + presentation_remaining = scrub_info["remaining"] + presentation_byte_base = scrub_info["byte_offset"] + if debug: + logger.debug( + "Timeshift session scrub: session=%s offset=%d remaining=%d " + "(%s -> %s)", + session_id, scrub_info["byte_offset"], scrub_info["remaining"], + prior_media_id, media_id, + ) + elif scrub_info["kind"] == "same": + # Back at archive open: reset scrub window so Ranges stay absolute. + presentation_remaining = scrub_info["remaining"] + presentation_byte_base = 0 + if range_header: + effective_range = range_header + relative_presentation_range = False + elif range_header and prior_presentation_base: + # Post-scrub client Ranges are relative to the shorter presented file. + effective_range = _map_client_range_through_presentation( + range_header, prior_presentation_base, + ) + relative_presentation_range = True + presentation_byte_base = prior_presentation_base + presentation_remaining = prior_presentation_length + if debug: + logger.debug( + "Timeshift presentation range map: session=%s %s -> %s " + "(base=%d)", + session_id, range_header, effective_range, + prior_presentation_base, + ) + elif media_changed: + # Outside the opened archive: mint a fresh portal/CDN URL. Retargeting + # an old CDN token onto a new start path often still serves the prior + # programme (token is bound to the archive that minted it). + keep_archive = False + final_url = None + if debug: + logger.debug( + "Timeshift session re-anchor: session=%s %s -> %s " + "(outside open archive window, portal)", + session_id, prior_media_id, media_id, + ) + else: + keep_archive = True + final_url = raw_final_url or None + if range_header and prior_presentation_base: + effective_range = _map_client_range_through_presentation( + range_header, prior_presentation_base, + ) + relative_presentation_range = True + presentation_byte_base = prior_presentation_base + presentation_remaining = prior_presentation_length + if debug: + logger.debug( + "Timeshift presentation range map: session=%s %s -> %s " + "(base=%d)", + session_id, range_header, effective_range, + prior_presentation_base, + ) + _update_pool_position( redis_client, session_id, media_id=media_id, provider_timestamp=provider_timestamp, + keep_archive=keep_archive, ) release_cb = _make_release_once(redis_client, session_id, profile.id) @@ -2104,6 +2615,7 @@ def _stream_reused_session( stats_stream_id = int(raw_stream_id) except (TypeError, ValueError): stats_stream_id = None + try: response = _attempt_timeshift_stream( m3u_account=m3u_account, @@ -2118,7 +2630,7 @@ def _stream_reused_session( client_id=client_id, client_ip=client_ip, client_user_agent=client_user_agent, - range_header=range_header, + range_header=effective_range, channel_logo_id=channel_logo_id, user=user, redis_client=redis_client, @@ -2126,6 +2638,11 @@ def _stream_reused_session( release_cb=release_cb, pool_session_id=session_id, stats_stream_id=stats_stream_id, + final_url=final_url, + rewrite_plain_get=rewrite_plain_get, + presentation_remaining=presentation_remaining, + presentation_byte_base=presentation_byte_base, + relative_presentation_range=relative_presentation_range, ) except Exception: _discard_pool_session(redis_client, session_id, profile.id) @@ -2328,8 +2845,13 @@ def _unregister_stats_client(redis_client, stats_channel_id, client_id): logger.warning("Timeshift stats unregister failed: %s", exc) -def _open_upstream(url, user_agent, range_header): - """Open upstream HTTP; redirects are followed (XC load-balancer nodes).""" +def _open_upstream(url, user_agent, range_header, *, allow_redirects=True): + """Open upstream HTTP. + + Portal URLs need ``allow_redirects=True`` (XC → CDN). Cached CDN + ``final_url`` values use ``allow_redirects=False`` so reconnects do not + mint a new provider timeshift lock/token. + """ # identity: raw peek bytes are not gzip-transparent. headers = {"Accept-Encoding": "identity"} if user_agent: @@ -2344,6 +2866,7 @@ def _open_upstream(url, user_agent, range_header): ConfigHelper.connection_timeout(), ConfigHelper.chunk_timeout(), ), + allow_redirects=allow_redirects, ) @@ -2402,9 +2925,24 @@ def _stream_from_provider( stats_stream_id=None, stream_stats=None, duration_minutes=None, + final_url=None, + rewrite_plain_get=False, + presentation_remaining=None, + presentation_byte_base=None, + relative_presentation_range=False, ): """Try each upstream URL until one returns streamable MPEG-TS. + When ``final_url`` is set (cached post-redirect CDN URL from a prior + request in this session), try it first with redirects disabled (same + pattern as VOD) so Range reconnects do not mint a new portal token. + + ``rewrite_plain_get`` maps an injected CDN Range (XC start-URL scrub) back + to a plain ``200`` + remaining ``Content-Length`` so clients that rebuild + the XC start URL (instead of sending ``Range``) still get provider-like + headers. Subsequent client Ranges are relative to that window and must be + remapped via ``relative_presentation_range``. + Sets ``timeshift_decisive`` on auth/ban-class failures (401/403/406) so the failover loop skips the rest of that account's streams. ``release_cb`` frees the provider slot when the streaming response is closed. @@ -2427,16 +2965,33 @@ def _stream_from_provider( ordered_urls = list(candidate_urls) original_indices = list(range(len(candidate_urls))) + # (url, allow_redirects, cached_final, format_index_or_None) + attempts = [] + if final_url: + attempts.append((final_url, False, True, None)) + for url, orig_idx in zip(ordered_urls, original_indices): + attempts.append((url, True, False, orig_idx)) + # Peek for MPEG-TS sync; some providers return HTTP 200 with PHP/HTML errors. upstream = None last_status = None - last_url = ordered_urls[0] + last_url = attempts[0][0] if attempts else "" winning_index = None + used_cached_final = False decisive_failure = False - for url, orig_idx in zip(ordered_urls, original_indices): + for url, follow_redirects, cached_final, orig_idx in attempts: try: - response = _open_upstream(url, user_agent, range_header) + response = _open_upstream( + url, user_agent, range_header, allow_redirects=follow_redirects, + ) except requests.exceptions.RequestException as exc: + if cached_final: + logger.warning( + "Timeshift cached CDN unreachable (%s): %s; falling back to portal", + _redact_url(url), type(exc).__name__, + ) + _clear_pool_final_url(redis_client, pool_session_id) + continue logger.error( "Timeshift provider unreachable (%s): %s", _redact_url(url), type(exc).__name__, @@ -2445,11 +3000,12 @@ def _stream_from_provider( HttpResponseBadRequest("Provider connection error") ) last_status = response.status_code - last_url = url + last_url = getattr(response, "url", None) or url + cascade_label = "cdn" if cached_final else (orig_idx if orig_idx is not None else "?") if debug: logger.debug( - "Timeshift cascade[%d]: status=%d type=%s url=%s", - orig_idx, response.status_code, + "Timeshift cascade[%s]: status=%d type=%s url=%s", + cascade_label, response.status_code, response.headers.get("Content-Type", "?"), _redact_url(url), ) @@ -2470,12 +3026,14 @@ def _stream_from_provider( response._peek_data = peek upstream = response winning_index = orig_idx + used_cached_final = cached_final break sync_offset = find_ts_sync(peek) if peek else -1 if sync_offset >= 0: response._peek_data = peek[sync_offset:] upstream = response winning_index = orig_idx + used_cached_final = cached_final break snippet = peek[:200].decode("utf-8", errors="replace") if peek else "(empty)" logger.warning( @@ -2487,9 +3045,19 @@ def _stream_from_provider( _redact_url(url), ) response.close() + if cached_final: + _clear_pool_final_url(redis_client, pool_session_id) last_status = 404 # Treat as soft rejection for cascade continue response.close() + if cached_final: + # Expired/rotated CDN token; clear and retry portal shapes. + logger.info( + "Timeshift cached CDN returned %d, clearing final_url for session %s", + response.status_code, pool_session_id, + ) + _clear_pool_final_url(redis_client, pool_session_id) + continue # Auth/ban-class statuses stop trying more shapes on this account; 5xx does not. code = response.status_code if code in (401, 403, 406) or 300 <= code < 400: @@ -2518,6 +3086,20 @@ def _stream_from_provider( _store_pool_content_length(redis_client, pool_session_id, upstream) _store_pool_serving_range(redis_client, pool_session_id, range_header) + # Capture post-redirect CDN URL for reconnects (VOD final_url pattern). + resolved_url = getattr(upstream, "url", None) or last_url + if resolved_url: + _store_pool_final_url(redis_client, pool_session_id, resolved_url) + # Portal opens define (or redefine) the session archive window; CDN scrubs reuse it. + # force=False: after keep_archive=False clear, these keys are empty and get set; + # after CDN→portal fallback mid-session, keep the original anchor. + _ensure_pool_archive_anchor( + redis_client, + pool_session_id, + timestamp=timestamp_utc, + duration_minutes=duration_minutes, + force=False, + ) representation_length = _extract_representation_length(upstream) if representation_length is None and redis_client and pool_session_id: @@ -2532,19 +3114,64 @@ def _stream_from_provider( except (TypeError, ValueError): representation_length = None - client_length_headers = _build_downstream_length_headers( - range_header=range_header, - status_code=status, - representation_length=representation_length, - upstream_content_range=content_range or None, - upstream_content_length=upstream.headers.get("Content-Length"), - streaming=True, - ) + if rewrite_plain_get and status == 206: + # Client sent a plain GET with a new start; we injected CDN Range. + # Present a provider-like 200 with remaining Content-Length. + remaining = presentation_remaining + if remaining is None and representation_length is not None: + start = _parse_range_start(range_header) or 0 + remaining = max(int(representation_length) - int(start), 0) + status = 200 + content_range = "" + client_length_headers = {"Accept-Ranges": "bytes"} + if remaining is not None: + client_length_headers["Content-Length"] = str(remaining) + byte_base = presentation_byte_base + if byte_base is None: + byte_base = _parse_range_start(range_header) or 0 + _store_pool_presentation_window( + redis_client, pool_session_id, remaining, byte_base=byte_base, + ) + else: + outbound_content_range = content_range or None + if relative_presentation_range and outbound_content_range: + outbound_content_range = _presentation_relative_content_range( + outbound_content_range, + presentation_byte_base=presentation_byte_base, + presentation_length=presentation_remaining, + ) + client_length_headers = _build_downstream_length_headers( + # Absolute CDN Range must not leak into Content-Range synthesis; + # the client still thinks this file starts at presentation byte 0. + range_header=None if relative_presentation_range else range_header, + status_code=status, + representation_length=( + presentation_remaining + if relative_presentation_range and presentation_remaining is not None + else representation_length + ), + upstream_content_range=outbound_content_range, + upstream_content_length=upstream.headers.get("Content-Length"), + streaming=True, + ) + if presentation_remaining is not None and presentation_byte_base is not None: + _store_pool_presentation_window( + redis_client, pool_session_id, presentation_remaining, + byte_base=presentation_byte_base, + ) + elif representation_length is not None and not range_header: + _store_pool_presentation_window( + redis_client, pool_session_id, representation_length, byte_base=0, + ) programme_duration_secs = None if duration_minutes: programme_duration_secs = float(duration_minutes) * 60.0 + # XC start-URL scrub injects a CDN Range for the provider only. Stats must + # use the URL timestamp vs EPG; mapping archive bytes onto programme + # duration falsely parks the card at an unrelated offset. + stats_range_start = None if rewrite_plain_get else _parse_range_start(range_header) _register_stats_client( redis_client, stats_channel_id, @@ -2562,7 +3189,7 @@ def _stream_from_provider( channel_uuid=channel_uuid, stats_stream_id=stats_stream_id, stream_stats=stream_stats, - range_start=_parse_range_start(range_header), + range_start=stats_range_start, representation_length=representation_length, programme_duration_secs=programme_duration_secs, emit_stats_update=True,