diff --git a/CHANGELOG.md b/CHANGELOG.md index d633edca..325ed735 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Strictly-UTC API surface, automatic per-provider timezone.** `server_info.timezone` is always `UTC` and the XC EPG `start`/`end` strings are emitted in UTC; the proxy converts the requested instant to the serving provider's own reported timezone (`server_info.timezone` captured on account refresh) at request time. No timezone to configure. - **Multi-provider failover.** Catch-up walks the channel's catch-up streams in order (like live playback): if one provider cannot serve the archive the next is tried, each attempt with its own account context (credentials, reported timezone, user-agent). Accounts that fail with an auth/ban-class status are not retried via their other streams. - **Provider pool accounting.** Catch-up reserves a provider profile slot (`connection_pool`) before connecting upstream — same contract as live and VOD — and releases it when the session ends. When the default profile is at capacity it walks the account's alternate profiles with their own resolved credentials, and answers `503` when every profile is full. - - **Per-client session pool.** The first request without `session_id` receives a `301` redirect to the same URL with a stable `?session_id=` query parameter; that ID becomes the client identity for stats, stop keys, and pool coordination. Each viewer gets their own Redis-backed pool entry even when watching the same programme; idle sessions can be reclaimed via fingerprint matching (same user and programme, IP + user-agent score). Real scrubs within a session stop the in-flight stream through the standard stop-key mechanism; parallel startup probes (full-file, open-ended, and EOF tail `Range` requests) are deferred with `503` instead of opening extra upstream connections. Parallel HTTP probes from the same `session_id` for the same programme do not count as extra streams toward the user limit; distinct sessions still each consume a slot. + - **Per-client session pool.** The first request without `session_id` and no fingerprint match receives a `301` redirect to the same URL with a stable `?session_id=` query parameter; that ID becomes the client identity for stats, stop keys, and pool coordination. Reconnects that omit `session_id` but match an existing pool entry (same user, IP, and user-agent) are served immediately without a redirect, matching IPTV client fast-forward behaviour. Each viewer gets their own Redis-backed pool entry even when watching the same programme; idle sessions can be reclaimed via fingerprint matching. Plain GET restarts stream from byte 0 with upstream `Content-Length` when known (provider-faithful). Real scrubs within a session stop the in-flight stream through the standard stop-key mechanism; parallel startup probes (full-file, open-ended, and EOF tail `Range` requests) are deferred with `503` instead of opening extra upstream connections. Parallel HTTP probes from the same `session_id` for the same programme do not count as extra streams toward the user limit; distinct sessions still each consume a slot. - **`xmltv_prev_days_override` under `Settings → EPG`** (default `0` = auto-detect). Verbose timeshift logging follows the standard logger DEBUG level. - **EPG XMLTV `prev_days` auto-detection** from the provider's largest `tv_archive_duration` (capped at 30 days), overridable via setting or `?prev_days=` URL parameter. - **Byte path streams directly via `iter_content` + generator yield** (same pattern as the VOD proxy), with an upfront MPEG-TS sync peek that strips any PHP-warning preamble before bytes reach strict demuxers. Throughput stays comfortably above the typical 5 Mbps FHD bitrate so clients don't buffer. diff --git a/apps/timeshift/stats.py b/apps/timeshift/stats.py index e3177add..c7580cba 100644 --- a/apps/timeshift/stats.py +++ b/apps/timeshift/stats.py @@ -178,14 +178,9 @@ def resolve_stats_playback_fields( if range_start == 0: return None, now - if existing_programme_start == timestamp_utc and existing_position_anchor: - existing_base = None - if existing_playback_base is not None: - try: - existing_base = float(existing_playback_base) - except (TypeError, ValueError): - existing_base = None - return existing_base, existing_position_anchor + # Plain GET (no Range): providers restart from byte 0 on reconnect. + if range_start is None: + return None, now return None, now diff --git a/apps/timeshift/tests/test_stats.py b/apps/timeshift/tests/test_stats.py index 5a5f9e0b..de60355a 100644 --- a/apps/timeshift/tests/test_stats.py +++ b/apps/timeshift/tests/test_stats.py @@ -130,7 +130,7 @@ class ByteRangePlaybackTests(TestCase): self.assertAlmostEqual(base, 2371.0, delta=2.0) self.assertEqual(anchor, "2000.0") - def test_resolve_same_programme_preserves_without_range(self): + def test_resolve_plain_get_reanchors_on_same_programme(self): base, anchor = resolve_stats_playback_fields( timestamp_utc="2026-06-08:17-00", existing_programme_start="2026-06-08:17-00", @@ -141,8 +141,8 @@ class ByteRangePlaybackTests(TestCase): programme_duration_secs=3900, now="2000.0", ) - self.assertAlmostEqual(base, 900.0) - self.assertEqual(anchor, "1000.0") + self.assertIsNone(base) + self.assertEqual(anchor, "2000.0") class TimeshiftStreamStatsTests(TestCase): diff --git a/apps/timeshift/tests/test_views.py b/apps/timeshift/tests/test_views.py index 1d9e3f40..73e55bc4 100644 --- a/apps/timeshift/tests/test_views.py +++ b/apps/timeshift/tests/test_views.py @@ -2154,7 +2154,7 @@ class TimeshiftSessionReuseTests(TestCase): class TimeshiftSessionRedirectTests(TestCase): - """First request must establish a session via 301 redirect (VOD-style).""" + """Session mint (301) and inline pool adopt when session_id is omitted.""" def setUp(self): self.factory = RequestFactory() @@ -2178,7 +2178,7 @@ class TimeshiftSessionRedirectTests(TestCase): self.assertEqual(response.status_code, 301) self.assertIn("session_id=", response["Location"]) - def test_missing_session_id_redirects_to_existing_busy_pool(self): + def test_missing_session_id_serves_existing_busy_pool_without_redirect(self): existing = "existingbusy1" redis = _FakeRedis() _seed_pool_session( @@ -2194,6 +2194,9 @@ class TimeshiftSessionRedirectTests(TestCase): HTTP_USER_AGENT="vlc-test", REMOTE_ADDR="127.0.0.1", ) + ok = MagicMock(status_code=200) + profile = MagicMock(id=31) + descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"} with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \ patch.object(views, "network_access_allowed", return_value=True), \ patch.object(views, "Channel") as channel_cls, \ @@ -2202,16 +2205,23 @@ class TimeshiftSessionRedirectTests(TestCase): return_value=[_make_catchup_stream()]), \ patch.object(views, "get_programme_duration", return_value=40), \ patch.object(views, "parse_catchup_timestamp", return_value=True), \ - patch.object(views, "RedisClient") as redis_cls: + patch.object(views, "check_user_stream_limits", return_value=True), \ + patch.object(views, "RedisClient") as redis_cls, \ + patch.object( + views, "_try_reacquire_idle_pool", + return_value=(descriptor, profile), + ) as reacquire_mock, \ + patch.object(views, "_stream_reused_session", return_value=ok) as reuse_mock: redis_cls.get_client.return_value = redis channel_cls.objects.get.return_value = MagicMock(id=8) response = views.timeshift_proxy( request, "u", "p", "8", "2026-06-08:17-00", "8.ts", ) - self.assertEqual(response.status_code, 301) - self.assertIn(f"session_id={existing}", response["Location"]) + self.assertIs(response, ok) + reacquire_mock.assert_called_once() + reuse_mock.assert_called_once() - def test_missing_session_id_redirects_to_busy_pool_on_channel_hop(self): + def test_missing_session_id_serves_busy_pool_on_channel_hop_without_redirect(self): existing = "channelhop1" redis = _FakeRedis() _seed_pool_session( @@ -2228,6 +2238,9 @@ class TimeshiftSessionRedirectTests(TestCase): HTTP_USER_AGENT="vlc-test", REMOTE_ADDR="127.0.0.1", ) + ok = MagicMock(status_code=200) + profile = MagicMock(id=31) + descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"} with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \ patch.object(views, "network_access_allowed", return_value=True), \ patch.object(views, "Channel") as channel_cls, \ @@ -2236,14 +2249,21 @@ class TimeshiftSessionRedirectTests(TestCase): return_value=[_make_catchup_stream()]), \ patch.object(views, "get_programme_duration", return_value=40), \ patch.object(views, "parse_catchup_timestamp", return_value=True), \ - patch.object(views, "RedisClient") as redis_cls: + patch.object(views, "check_user_stream_limits", return_value=True), \ + patch.object(views, "RedisClient") as redis_cls, \ + patch.object( + views, "_try_reacquire_idle_pool", + return_value=(descriptor, profile), + ) as reacquire_mock, \ + patch.object(views, "_stream_reused_session", return_value=ok) as reuse_mock: redis_cls.get_client.return_value = redis channel_cls.objects.get.return_value = MagicMock(id=8) response = views.timeshift_proxy( request, "u", "p", "8", "2026-06-08:17-30", "8.ts", ) - self.assertEqual(response.status_code, 301) - self.assertIn(f"session_id={existing}", response["Location"]) + self.assertIs(response, ok) + reacquire_mock.assert_called_once() + reuse_mock.assert_called_once() def test_redirect_preserves_existing_query_params(self): request = self.factory.get( @@ -2692,12 +2712,13 @@ class TimeshiftStatsClientTests(TestCase): self.assertNotIn(old_gen, self.redis.store) - def test_register_stats_preserves_position_anchor_on_same_programme(self): + def test_register_stats_reanchors_position_on_plain_get_reconnect(self): from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, TimeshiftRedisKeys client_key = RedisKeys.client_metadata(self.stats_channel_id, self.client_id) self.redis.hset(client_key, "programme_start", "2026-06-08:17-00") self.redis.hset(client_key, "position_anchor_at", "1000.0") + self.redis.hset(client_key, "playback_base_secs", "900.0") views._register_stats_client( self.redis, @@ -2713,7 +2734,8 @@ class TimeshiftStatsClientTests(TestCase): channel_id=8, channel_uuid="00000000-0000-0000-0000-000000000008", ) - self.assertEqual(self.redis.hget(client_key, "position_anchor_at"), "1000.0") + self.assertNotEqual(self.redis.hget(client_key, "position_anchor_at"), "1000.0") + self.assertIsNone(self.redis.hget(client_key, "playback_base_secs")) def test_register_stats_byte_range_seek_sets_playback_base(self): from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, TimeshiftRedisKeys @@ -3922,6 +3944,45 @@ class TimeshiftScrubPreemptTests(TestCase): wait_seconds=views._POOL_SCRUB_WAIT_SECONDS, ) + def test_plain_reconnect_preempts_busy_pool_without_range(self): + """TiviMate FF reconnects with plain GET; 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)] + ok = MagicMock(status_code=200) + profile = MagicMock(id=31) + descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"} + 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", return_value=(True, 1, None)), \ + 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", + return_value=(descriptor, profile), + ) as reacquire_mock, \ + patch.object(views, "_stream_reused_session", return_value=ok) as reuse_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( + id=8, name="Test", logo_id=None, + ) + response = views.timeshift_proxy( + request, "u", "p", "8", "2026-06-08:17-00", "8.ts", + ) + self.assertIs(response, ok) + reacquire_mock.assert_called_once() + reuse_mock.assert_called_once() + self.assertIsNone(reuse_mock.call_args.kwargs["range_header"]) + attempt_mock.assert_not_called() + def test_busy_pool_reuses_slot_after_scrub_preempt(self): _seed_pool_session(self.redis, session_id=TEST_SESSION_ID) request = self.factory.get( @@ -4326,6 +4387,87 @@ class CatchupProxyTests(TestCase): serve.assert_called_once() +class TimeshiftProviderFaithfulPlainGetTests(_ProxyLoopTestMixin, TestCase): + """Contract tests for TiviMate-style plain GET reconnect (no Range header).""" + + def setUp(self): + self.redis = _FakeRedis() + self.virtual_channel_id = f"{TEST_MEDIA_ID}_111" + self.stats_channel_id = views.stats_channel_id(8, TEST_SESSION_ID) + self.client_id = TEST_SESSION_ID + self.user = MagicMock(id=5, username="viewer") + + @patch.object(views, "_trigger_timeshift_stats_update") + @patch.object(views, "_open_upstream") + def test_plain_get_does_not_send_range_to_upstream(self, mocked_open, _trigger_mock): + ts = _make_ts_payload(188 * 7) + mocked_open.return_value = _fake_upstream(200, body=ts) + + views._stream_from_provider( + candidate_urls=["http://example.test/timeshift.ts"], + user_agent="provider-agent", + range_header=None, + virtual_channel_id=self.virtual_channel_id, + stats_channel_id=self.stats_channel_id, + client_id=self.client_id, + client_ip="1.2.3.4", + client_user_agent="tivimate", + user=self.user, + channel_display_name="A&E", + timestamp_utc="2026-06-08:17-00", + channel_logo_id=None, + m3u_profile_id=31, + debug=False, + account_id=None, + redis_client=self.redis, + pool_session_id=self.client_id, + channel_id=8, + channel_uuid="00000000-0000-0000-0000-000000000008", + duration_minutes=40, + ) + + mocked_open.assert_called_once() + self.assertIsNone(mocked_open.call_args.args[2]) + + @patch.object(views, "_trigger_timeshift_stats_update") + @patch.object(views, "_open_upstream") + def test_plain_get_passes_through_provider_200_without_content_range( + self, mocked_open, _trigger_mock, + ): + ts = _make_ts_payload(188 * 7) + upstream = _fake_upstream(200, body=ts) + upstream.headers["Content-Length"] = str(len(ts)) + mocked_open.return_value = upstream + + response = views._stream_from_provider( + candidate_urls=["http://example.test/timeshift.ts"], + user_agent="provider-agent", + range_header=None, + virtual_channel_id=self.virtual_channel_id, + stats_channel_id=self.stats_channel_id, + client_id=self.client_id, + client_ip="1.2.3.4", + client_user_agent="tivimate", + user=self.user, + channel_display_name="A&E", + timestamp_utc="2026-06-08:17-00", + channel_logo_id=None, + m3u_profile_id=31, + debug=False, + account_id=None, + redis_client=self.redis, + pool_session_id=self.client_id, + channel_id=8, + channel_uuid="00000000-0000-0000-0000-000000000008", + duration_minutes=40, + ) + + self.assertEqual(response.status_code, 200) + self.assertNotIn("Content-Range", response) + self.assertEqual(response["Accept-Ranges"], "bytes") + self.assertEqual(response["Content-Length"], str(len(ts))) + + class TimeshiftDownstreamLengthHeaderTests(TestCase): def test_open_ended_range_passthrough_upstream_headers(self): headers = views._build_downstream_length_headers( @@ -4372,7 +4514,7 @@ class TimeshiftDownstreamLengthHeaderTests(TestCase): self.assertEqual(headers["Content-Length"], "10000") self.assertNotIn("Content-Range", headers) - def test_streaming_full_file_omits_content_length(self): + def test_streaming_plain_get_includes_content_length(self): headers = views._build_downstream_length_headers( range_header=None, status_code=200, @@ -4381,8 +4523,9 @@ class TimeshiftDownstreamLengthHeaderTests(TestCase): upstream_content_length="10000", streaming=True, ) - self.assertNotIn("Content-Length", headers) + self.assertEqual(headers["Content-Length"], "10000") self.assertEqual(headers["Accept-Ranges"], "bytes") + self.assertNotIn("Content-Range", headers) def test_passthrough_includes_accept_ranges(self): response = views._passthrough_response(416, "bytes */10000") diff --git a/apps/timeshift/views.py b/apps/timeshift/views.py index 9c8b64e0..71717e36 100644 --- a/apps/timeshift/views.py +++ b/apps/timeshift/views.py @@ -103,6 +103,12 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration) ``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 ``YYYY-MM-DD:HH:MM:SS``). + + Session handling (``?session_id=``): + First request with no ``session_id`` and no matching pool entry → ``301`` with a + minted ``session_id``. Reconnects that omit ``session_id`` but fingerprint-match + an in-flight or idle pool entry for the same viewer are served immediately + (no redirect). Reuse ``session_id`` for all range/seek requests in a programme. """ raw_id = duration[:-3] if duration.endswith(".ts") else duration @@ -137,8 +143,15 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration) "time) plus ``Authorization: Bearer``, ``X-API-Key``, or " "``?token=``. Optionally include a client ``session_id`` for " "provider pooling.\n\n" - "If ``session_id`` is omitted on a direct-auth request, the server " - "responds with **301** adding a server-minted ``session_id``." + "**``session_id`` without a matching pool:** the server responds with " + "**301** and a minted ``session_id`` (direct-auth / first play only).\n\n" + "**``session_id`` omitted but pool match:** when the same viewer " + "reconnects (e.g. IPTV fast-forward) and fingerprint-matches an " + "existing pool entry, the stream is served on that request with no " + "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)." ), parameters=[ OpenApiParameter( @@ -147,10 +160,11 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration) location=OpenApiParameter.QUERY, required=False, description=( - "Playback session from ``POST /api/catchup/sessions/``. " - "When present, JWT is optional. The session authenticates " - "the player. Reuse for all range/seek requests during the " - "same programme." + "Playback session from ``POST /api/catchup/sessions/`` or a " + "prior **301** redirect. When present, JWT is optional. Reuse " + "for all range/seek requests during the same programme. " + "May be omitted on reconnect when the server adopts a " + "fingerprint-matched pool entry." ), ), OpenApiParameter( @@ -176,11 +190,18 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration) ), ], responses={ - 200: {"description": "MPEG-TS stream (may be partial content for Range requests)."}, + 200: { + "description": ( + "MPEG-TS stream. Partial content (``206``) when the client " + "sent ``Range``. Full-file ``200`` on plain GET includes " + "``Content-Length`` when the upstream archive size is known." + ), + }, 301: { "description": ( "Redirect to the same URL with a server-minted ``session_id`` " - "when the client did not supply one (direct-auth flow only)." + "when the client did not supply one and no fingerprint-matched " + "pool entry exists (first play / direct-auth only)." ), }, 401: {"description": "Missing or expired authentication / session."}, @@ -278,15 +299,15 @@ def _serve_catchup(request, user, channel, timestamp): include_busy=True, ) if matched: - logger.debug( - "Timeshift session redirect: reusing %s for %s", - matched, request.path, - ) - return _finalize_timeshift_response( - _redirect_with_session(request, matched), - ) - logger.debug("Timeshift session redirect: %s (new session)", request.path) - return _finalize_timeshift_response(_redirect_with_new_session(request)) + if debug: + logger.debug( + "Timeshift session adopt: reusing %s for %s", + matched, request.path, + ) + session_id = matched + else: + logger.debug("Timeshift session redirect: %s (new session)", request.path) + return _finalize_timeshift_response(_redirect_with_new_session(request)) session_entry = _get_pool_entry(redis_client, session_id) if session_entry and not _pool_entry_owned_by_user(session_entry, user.id): @@ -391,6 +412,17 @@ def _serve_catchup(request, user, channel, timestamp): redis_client, effective_session_id, user_id=user.id, user=user, ) + elif _should_preempt_plain_reconnect( + range_header, + pool_exists=pool_exists, + pool_busy=pool_busy, + pool_media_id=pool_media_id, + media_id=media_id, + ): + acquired = _try_reacquire_idle_pool( + redis_client, effective_session_id, + user_id=user.id, user=user, + ) elif scrub_displacement: acquired = _try_reacquire_idle_pool( redis_client, effective_session_id, @@ -634,7 +666,8 @@ def _user_can_access_channel(user, channel): ) -# Per-client pool (session_id from 301 redirect). Fingerprint-match when ?session_id= is dropped. +# Per-client pool (session_id from 301 redirect or inline adopt when ?session_id= +# is dropped on reconnect). Fingerprint-match when ?session_id= is absent. _POOL_ENTRY_TTL = 10 * 60 # Refreshed on playback GET and active-stream heartbeats. _STREAM_READ_INACTIVITY_SECONDS = CLIENT_TTL_SECONDS # Shorter than pool TTL; data should flow. _POOL_IDLE_TTL = CLIENT_TTL_SECONDS # Must match CLIENT_TTL_SECONDS (stats + fingerprint agree). @@ -1102,7 +1135,16 @@ def _build_downstream_length_headers( headers["Content-Length"] = str(up_end - up_start + 1) return headers - # Omit Content-Length on streaming full-file responses (seeks may preempt). + # Plain GET streaming 200: match provider/CDN Content-Length (TiviMate uses + # 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: + headers["Content-Length"] = str(representation_length) + elif upstream_content_length: + headers["Content-Length"] = str(upstream_content_length) + return headers + + # Omit Content-Length on other streaming full-file responses (seeks may preempt). if not streaming: if representation_length is not None: headers["Content-Length"] = str(representation_length) @@ -1177,6 +1219,24 @@ def _try_reacquire_idle_pool( ) +def _should_preempt_plain_reconnect( + range_header, *, pool_exists, pool_busy, pool_media_id, media_id, +): + """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. + """ + if range_header: + return False + if not pool_exists or not pool_busy: + return False + if pool_media_id is not None and str(pool_media_id) != str(media_id): + return False + return True + + def _should_displace_busy_playback( range_header, content_length=None, busy_serving_range=None, ):