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,