diff --git a/lib/media/live_tv_support.dart b/lib/media/live_tv_support.dart index 3bfbb7c16..4386261a1 100644 --- a/lib/media/live_tv_support.dart +++ b/lib/media/live_tv_support.dart @@ -23,6 +23,27 @@ class LiveProgramInfo { static const none = LiveProgramInfo(); } +/// What a live heartbeat learned from the server. Both windows are +/// `TranscodeSession` snapshots (epoch origin plus min/max offsets), but they +/// describe different things: +/// +/// - [captureBuffer]: the tuner's seekable history — the coordinate system +/// for time-shift offsets and the timeline's range. +/// - [playbackStream]: the transcode currently feeding the player. Its +/// `startedAt` is the epoch of stream position zero, i.e. the exact clock +/// anchor for `epoch = startedAt + player position`. Plex's own client +/// derives the playhead from this object, not from wall clock (#2100). +/// +/// Either may be null when the backend does not report it. +class LiveTimelineUpdate { + final CaptureBuffer? captureBuffer; + final CaptureBuffer? playbackStream; + + const LiveTimelineUpdate({this.captureBuffer, this.playbackStream}); + + bool get isEmpty => captureBuffer == null && playbackStream == null; +} + /// One live-TV playback session, produced by [LiveTvSupport.startPlayback]. /// /// This is the backend-neutral handle the player drives; the @@ -80,9 +101,9 @@ abstract class LiveTvPlaybackSession { /// Send a playback heartbeat (`'playing'` / `'paused'` / `'stopped'`). /// [positionMs] is elapsed playback time; [durationMs] the program - /// duration when known. Returns an updated capture buffer when the backend - /// supplies one, null otherwise. - Future reportTimeline({required String state, required int positionMs, required int durationMs}); + /// duration when known. Returns what the backend reported back (capture + /// window, playback-stream origin), null when it reports nothing. + Future reportTimeline({required String state, required int positionMs, required int durationMs}); /// Re-establish playback after stream death. Plex re-tunes (the previous /// capture session expires while the player exhausts its reconnect diff --git a/lib/screens/video_player/live_timeline_report.dart b/lib/screens/video_player/live_timeline_report.dart index cc8e9c370..7580cb022 100644 --- a/lib/screens/video_player/live_timeline_report.dart +++ b/lib/screens/video_player/live_timeline_report.dart @@ -1,7 +1,6 @@ import '../../media/live_tv_support.dart'; -import '../../models/livetv_capture_buffer.dart'; -/// Sends one live-TV timeline report and commits its capture window only while +/// Sends one live-TV timeline report and commits what it learned only while /// the dispatching session and scheduling generation still own the screen. Future runLiveTimelineReport({ required LiveTvPlaybackSession requestSession, @@ -11,19 +10,19 @@ Future runLiveTimelineReport({ required LiveTvPlaybackSession? Function() currentSession, required int Function() currentGeneration, required bool Function() isMounted, - required void Function(CaptureBuffer buffer) commit, + required void Function(LiveTimelineUpdate update) commit, }) async { - final updatedBuffer = await requestSession.reportTimeline( + final update = await requestSession.reportTimeline( state: state, positionMs: positionMs, durationMs: requestSession.program.durationMs ?? 0, ); - if (updatedBuffer == null || + if (update == null || state == 'stopped' || !isMounted() || currentGeneration() != requestGeneration || !identical(currentSession(), requestSession)) { return; } - commit(updatedBuffer); + commit(update); } diff --git a/lib/screens/video_player/live_tv_session_state.dart b/lib/screens/video_player/live_tv_session_state.dart index 3fe71c054..1f71a72f1 100644 --- a/lib/screens/video_player/live_tv_session_state.dart +++ b/lib/screens/video_player/live_tv_session_state.dart @@ -55,9 +55,19 @@ class LiveTvSessionState { /// [remapSubtitleSelection]). MediaSubtitleTrack? selectedSubtitle; + /// Epoch of player position zero for the stream MPV is playing. Seeded + /// provisionally at open ([markStreamRestartedAtLiveEdge], the offset of a + /// time-shift open), calibrated against the first rendered position, then + /// replaced by the server's own origin for the playback transcode as soon + /// as a heartbeat reports it ([adoptPlaybackStreamOrigin]). double streamStartEpoch = 0; bool atLiveEdge = true; + /// Bumped on every stream open. A heartbeat snapshots it when dispatched so + /// a response describing a stream that has since been replaced cannot + /// re-anchor the clock of its replacement. + int streamGeneration = 0; + int _nextClockGeneration = 0; int? _latestClockGeneration; int? activeClockSourceId; @@ -192,6 +202,21 @@ class LiveTvSessionState { return (streamStartEpoch + position.inMilliseconds / 1000.0).round(); } + /// Re-anchor the clock on the playback transcode's server-reported origin: + /// its `timeStamp` is the epoch of stream position zero (its first segment's + /// program date-time is `timeStamp + minOffsetAvailable`), which no client + /// side guess — wall clock at open, the requested offset — can match once + /// tuner latency and keyframe snapping are in play (#2100). + /// + /// Skipped while an open is still calibrating: the heartbeat may describe + /// the transcode being replaced. Returns whether the anchor moved. + bool adoptPlaybackStreamOrigin(CaptureBuffer playbackStream, {required int generation}) { + if (generation != streamGeneration || pendingStreamEpoch != null) return false; + if (streamStartEpoch == playbackStream.startedAt) return false; + streamStartEpoch = playbackStream.startedAt; + return true; + } + /// Make [newSession] current and seed the seekable window from its tune /// snapshot. Every flow that produces a session (start, retry, channel /// zap) adopts it here, so a field can't be forgotten in one copy. @@ -218,11 +243,20 @@ class LiveTvSessionState { } /// The stream just (re)started at the live edge — align the epoch - /// bookkeeping every restart flow shares (retry, channel zap). - void markStreamRestartedAtLiveEdge() { + /// bookkeeping every restart flow shares (start, retry, channel zap, + /// subtitle switch). + /// + /// [buffer] is the freshest capture window known for the stream being + /// opened (the tune snapshot for start/retry/zap). An offset-less open + /// starts at the capture's live edge, so its end is the provisional anchor + /// until the first heartbeat reports the transcode's real origin — wall + /// clock is not: the edge already trails real time by the tuner's ingest + /// latency, and skipping back from a wall-clock anchor landed on or after + /// the frame being shown (#2100). + void markStreamRestartedAtLiveEdge(CaptureBuffer? buffer) { final now = DateTime.now(); playbackStartTime = now; - streamStartEpoch = now.millisecondsSinceEpoch / 1000.0; + streamStartEpoch = buffer == null ? now.millisecondsSinceEpoch / 1000.0 : buffer.startedAt + buffer.seekEndSeconds; atLiveEdge = true; } } diff --git a/lib/screens/video_player/parts/live_tv.dart b/lib/screens/video_player/parts/live_tv.dart index eaf1f3191..a291d7b84 100644 --- a/lib/screens/video_player/parts/live_tv.dart +++ b/lib/screens/video_player/parts/live_tv.dart @@ -84,6 +84,7 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { final requestSession = _live.session; if (requestSession == null) return; final requestGeneration = _live.timelineGeneration; + final requestStreamGeneration = _live.streamGeneration; // For live TV, player position/duration are unreliable (often 0). Use // elapsed wall-clock as the position and the program duration from tune // metadata; the per-backend session owns the wire mapping. @@ -100,12 +101,20 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { currentSession: () => _live.session, currentGeneration: () => _live.timelineGeneration, isMounted: () => mounted, - commit: (updatedBuffer) { + commit: (update) { _setPlayerState(() { - _live.captureBuffer = updatedBuffer; - _live.atLiveEdge = - (_currentPositionEpoch >= - updatedBuffer.seekableEndEpoch - VideoPlayerScreenState._liveEdgeThresholdSeconds); + final playbackStream = update.playbackStream; + if (playbackStream != null && + _live.adoptPlaybackStreamOrigin(playbackStream, generation: requestStreamGeneration)) { + appLogger.d('Live clock re-anchored on playback transcode origin ${playbackStream.startedAt}'); + } + final buffer = update.captureBuffer; + if (buffer != null) _live.captureBuffer = buffer; + final window = _live.captureBuffer; + if (window != null) { + _live.atLiveEdge = + (_currentPositionEpoch >= window.seekableEndEpoch - VideoPlayerScreenState._liveEdgeThresholdSeconds); + } }); }, ); @@ -174,11 +183,11 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { // is re-mapped onto the recovered session's track list. Recovering the // video outranks keeping subtitles — a failed burn re-apply drops them. MediaSubtitleTrack? recoveredSubtitle; - var recoveredHasCaptureBuffer = false; + CaptureBuffer? recoveredCaptureBuffer; final result = await runLiveStreamRetry( recover: () => session.recover(directStream: ds, directStreamAudio: dsa), lookupStreamUrl: (recovered) async { - recoveredHasCaptureBuffer = recovered.captureBuffer != null; + recoveredCaptureBuffer = recovered.captureBuffer; recoveredSubtitle = LiveTvSessionState.remapSubtitleSelection(recovered.subtitleTracks, _live.selectedSubtitle); if (recoveredSubtitle != null) { final url = await recovered.streamUrlAt(subtitleTrack: recoveredSubtitle); @@ -190,8 +199,8 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { }, applyPlayerOptions: () => _setLiveStreamOptions(currentPlayer), open: (streamUrl) async { - _live.markStreamRestartedAtLiveEdge(); - final targetEpoch = recoveredHasCaptureBuffer ? _live.streamStartEpoch.round() : null; + _live.markStreamRestartedAtLiveEdge(recoveredCaptureBuffer); + final targetEpoch = recoveredCaptureBuffer == null ? null : _live.streamStartEpoch.round(); await _openLiveStream(currentPlayer, streamUrl, targetEpoch: targetEpoch, applyOptions: false); }, isCurrent: isCurrent, @@ -236,6 +245,7 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { bool? play, bool applyOptions = true, }) async { + _live.streamGeneration++; final clockGeneration = targetEpoch != null && player is PlayerNative ? _live.beginClockOpen(targetEpoch) : null; final clockResult = clockGeneration == null ? null : _live.clockOpenResult(clockGeneration); try { @@ -363,7 +373,7 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { _live.selectedSubtitle = previous; return PlaybackSourceChangeOutcome.failed; } - _live.markStreamRestartedAtLiveEdge(); + _live.markStreamRestartedAtLiveEdge(_live.captureBuffer); await _openLiveStream( currentPlayer, streamUrl, @@ -471,7 +481,7 @@ extension _VideoPlayerLiveTvMethods on VideoPlayerScreenState { _setPlayerState(() { _firstFrame.reset(); }); - _live.markStreamRestartedAtLiveEdge(); + _live.markStreamRestartedAtLiveEdge(session.captureBuffer); final targetEpoch = session.captureBuffer == null ? null : _live.streamStartEpoch.round(); replacementOpenStarted = true; await _openLiveStream(currentPlayer, streamUrl, targetEpoch: targetEpoch); diff --git a/lib/screens/video_player/parts/playback_start.dart b/lib/screens/video_player/parts/playback_start.dart index e953d4ea1..2c612244b 100644 --- a/lib/screens/video_player/parts/playback_start.dart +++ b/lib/screens/video_player/parts/playback_start.dart @@ -74,7 +74,7 @@ extension _VideoPlayerPlaybackStartMethods on VideoPlayerScreenState { _live.atLiveEdge = false; _live.playbackStartTime = DateTime.now(); } else { - _live.markStreamRestartedAtLiveEdge(); + _live.markStreamRestartedAtLiveEdge(captureBuffer); targetEpoch = captureBuffer == null ? null : _live.streamStartEpoch.round(); } diff --git a/lib/screens/video_player_screen.dart b/lib/screens/video_player_screen.dart index e566c292d..71c94ad5a 100644 --- a/lib/screens/video_player_screen.dart +++ b/lib/screens/video_player_screen.dart @@ -23,6 +23,7 @@ import '../media/media_item_types.dart'; import '../media/media_server_client.dart'; import '../media/episode_collection.dart'; import '../media/live_tv_support.dart'; +import '../models/livetv_capture_buffer.dart'; import '../models/livetv_channel.dart'; import '../services/live_seek_accumulator.dart'; import '../services/plex_client.dart'; @@ -416,7 +417,11 @@ class VideoPlayerScreen extends StatefulWidget { } class VideoPlayerScreenState extends State with WidgetsBindingObserver, MountedSetStateMixin { - static const int _liveEdgeThresholdSeconds = 5; + /// How close to the capture buffer's end counts as "live". A live-edge + /// transcode starts behind the buffer's edge by tuner ingest and encoder + /// start-up latency (10–20 s observed), so a tighter threshold would flag + /// a freshly tuned stream as time-shifted. Matches Plex's own client. + static const int _liveEdgeThresholdSeconds = 15; // Track the currently active route target to guard duplicate navigation and // project the server-qualified media key to housekeeping consumers. diff --git a/lib/services/jellyfin_client/parts/live_tv.dart b/lib/services/jellyfin_client/parts/live_tv.dart index 83d29c0a3..741902ead 100644 --- a/lib/services/jellyfin_client/parts/live_tv.dart +++ b/lib/services/jellyfin_client/parts/live_tv.dart @@ -414,7 +414,7 @@ class _JellyfinLiveTvPlaybackSession implements LiveTvPlaybackSession { offsetSeconds == null && subtitleTrack == null ? _url : null; @override - Future reportTimeline({ + Future reportTimeline({ required String state, required int positionMs, required int durationMs, diff --git a/lib/services/plex_client/parts/live_tv.dart b/lib/services/plex_client/parts/live_tv.dart index bef2eec87..5e651dadf 100644 --- a/lib/services/plex_client/parts/live_tv.dart +++ b/lib/services/plex_client/parts/live_tv.dart @@ -77,10 +77,11 @@ mixin _PlexLiveTvClientMethods on _PlexClientInternals implements LiveTvSupport, /// Send a live TV timeline heartbeat to keep the transcode session alive. /// - /// Returns an updated [CaptureBuffer] if the response contains a - /// `TranscodeSession` with seek-range data (used to expand the seekable - /// window over time). - Future _updateLiveTimeline({ + /// The response carries up to two `TranscodeSession`s: the tuner's capture + /// buffer under `CaptureBuffer`, and the playback transcode at the top + /// level (only once the stream has started). Returns both; null when the + /// response carries neither. + Future _updateLiveTimeline({ required String ratingKey, required String sessionPath, required String sessionIdentifier, @@ -113,13 +114,13 @@ mixin _PlexLiveTvClientMethods on _PlexClientInternals implements LiveTvSupport, return null; } - // Parse updated capture buffer from TranscodeSession in the response + // Parse the capture window and the playback transcode from the response. try { final data = response.data; if (data is! Map) return null; final container = data['MediaContainer'] as Map? ?? data; - // Try CaptureBuffer wrapper first, then TranscodeSession directly + CaptureBuffer? capture; final captureBufferWrapper = container['CaptureBuffer']; if (captureBufferWrapper != null) { final cbMap = captureBufferWrapper is List @@ -128,16 +129,24 @@ mixin _PlexLiveTvClientMethods on _PlexClientInternals implements LiveTvSupport, if (cbMap != null) { final ts = cbMap['TranscodeSession']; final tsMap = ts is List ? ts.firstOrNull as Map? : ts as Map?; - if (tsMap != null) return CaptureBuffer.fromTranscodeSession(tsMap); + if (tsMap != null) capture = CaptureBuffer.fromTranscodeSession(tsMap); } } + CaptureBuffer? topLevel; final transcodeSessions = container['TranscodeSession']; if (transcodeSessions is List && transcodeSessions.isNotEmpty) { - return CaptureBuffer.fromTranscodeSession(transcodeSessions.first as Map); + topLevel = CaptureBuffer.fromTranscodeSession(transcodeSessions.first as Map); } else if (transcodeSessions is Map) { - return CaptureBuffer.fromTranscodeSession(transcodeSessions); + topLevel = CaptureBuffer.fromTranscodeSession(transcodeSessions); } + + // Without the wrapper the lone top-level session is the capture buffer + // (the tune-response shape); with it, the top-level one is playback. + final update = capture == null + ? LiveTimelineUpdate(captureBuffer: topLevel) + : LiveTimelineUpdate(captureBuffer: capture, playbackStream: topLevel); + return update.isEmpty ? null : update; } catch (e) { // Parsing failure is non-fatal — just no updated seek range } @@ -1012,7 +1021,11 @@ class _PlexLiveTvPlaybackSession implements LiveTvPlaybackSession { } @override - Future reportTimeline({required String state, required int positionMs, required int durationMs}) { + Future reportTimeline({ + required String state, + required int positionMs, + required int durationMs, + }) { // Plex rejects timeline pings where time > duration; grow duration to // match — otherwise Tunarr-style short synthetic programs 400 mid-stream. final duration = durationMs >= positionMs ? durationMs : positionMs; diff --git a/test/screens/video_player/live_timeline_report_test.dart b/test/screens/video_player/live_timeline_report_test.dart index 4006a29a2..a4441ceca 100644 --- a/test/screens/video_player/live_timeline_report_test.dart +++ b/test/screens/video_player/live_timeline_report_test.dart @@ -168,7 +168,7 @@ Future _run( currentSession: currentSession, currentGeneration: currentGeneration, isMounted: isMounted ?? () => true, - commit: commit, + commit: (update) => commit(update.captureBuffer!), ); } @@ -180,9 +180,10 @@ class _FakeSession implements LiveTvPlaybackSession { @override final CaptureBuffer captureBuffer; final List states = []; - final List> _reports = []; + final List> _reports = []; - void complete(int index, CaptureBuffer? buffer) => _reports[index].complete(buffer); + void complete(int index, CaptureBuffer? buffer) => + _reports[index].complete(buffer == null ? null : LiveTimelineUpdate(captureBuffer: buffer)); @override LiveTvBackgroundPolicy get backgroundPolicy => LiveTvBackgroundPolicy.retainSession; @@ -197,9 +198,13 @@ class _FakeSession implements LiveTvPlaybackSession { Future recover({required bool directStream, required bool directStreamAudio}) async => this; @override - Future reportTimeline({required String state, required int positionMs, required int durationMs}) { + Future reportTimeline({ + required String state, + required int positionMs, + required int durationMs, + }) { states.add(state); - final completer = Completer(); + final completer = Completer(); _reports.add(completer); return completer.future; } diff --git a/test/screens/video_player/live_tv_session_state_test.dart b/test/screens/video_player/live_tv_session_state_test.dart index 48cd91352..aebc19a77 100644 --- a/test/screens/video_player/live_tv_session_state_test.dart +++ b/test/screens/video_player/live_tv_session_state_test.dart @@ -167,6 +167,55 @@ void main() { }); }); }); + + group('LiveTvSessionState stream anchor', () { + const capture = CaptureBuffer(startedAt: 1000, seekStartSeconds: 0, seekEndSeconds: 40); + + test('a live-edge open anchors on the capture edge, not wall clock', () { + final state = LiveTvSessionState(null); + + state.markStreamRestartedAtLiveEdge(capture); + + expect(state.streamStartEpoch, 1040); + expect(state.atLiveEdge, isTrue); + expect(state.epochForPosition(const Duration(seconds: 10)), 1050); + }); + + test('a heartbeat re-anchors the clock on the playback transcode origin', () { + final state = LiveTvSessionState(null)..markStreamRestartedAtLiveEdge(capture); + final playback = CaptureBuffer(startedAt: 1027.5, seekStartSeconds: 0.033, seekEndSeconds: 12); + + expect(state.adoptPlaybackStreamOrigin(playback, generation: state.streamGeneration), isTrue); + + // A 10 s skip back from 20 s in now targets 1017, not 1050. + expect(state.epochForPosition(const Duration(seconds: 20)), 1048); + expect(state.adoptPlaybackStreamOrigin(playback, generation: state.streamGeneration), isFalse); + }); + + test('a heartbeat dispatched before a re-open cannot anchor the replacement stream', () async { + final state = LiveTvSessionState(null)..streamStartEpoch = 1040; + final staleGeneration = state.streamGeneration; + state.streamGeneration++; + final generation = state.beginClockOpen(990); + final result = state.clockOpenResult(generation); + state.bindClockSource(const PlayerSourceStarted(3)); + state.calibrateClockSource(const PlayerSourceReady(sourceId: 3, position: Duration.zero)); + expect(await result, isTrue); + + final old = CaptureBuffer(startedAt: 1027.5, seekStartSeconds: 0, seekEndSeconds: 60); + expect(state.adoptPlaybackStreamOrigin(old, generation: staleGeneration), isFalse); + expect(state.epochForPosition(Duration.zero), 990); + }); + + test('a heartbeat cannot re-anchor while an open is still calibrating', () { + final state = LiveTvSessionState(null)..streamStartEpoch = 1040; + state.beginClockOpen(990); + + final old = CaptureBuffer(startedAt: 1027.5, seekStartSeconds: 0, seekEndSeconds: 60); + expect(state.adoptPlaybackStreamOrigin(old, generation: state.streamGeneration), isFalse); + expect(state.epochForPosition(const Duration(seconds: 5)), 990); + }); + }); } class _FakeSession implements LiveTvPlaybackSession { @@ -186,8 +235,11 @@ class _FakeSession implements LiveTvPlaybackSession { List get subtitleTracks => const []; @override - Future reportTimeline({required String state, required int positionMs, required int durationMs}) => - Future.value(null); + Future reportTimeline({ + required String state, + required int positionMs, + required int durationMs, + }) => Future.value(null); @override Future recover({required bool directStream, required bool directStreamAudio}) => diff --git a/test/services/live_tv_playback_session_test.dart b/test/services/live_tv_playback_session_test.dart index 1cd81af7d..0dee4944b 100644 --- a/test/services/live_tv_playback_session_test.dart +++ b/test/services/live_tv_playback_session_test.dart @@ -294,10 +294,19 @@ void main() { } if (request.url.path == '/:/timeline') { timelineQuery = request.url.queryParameters; + // Once the stream plays, the top-level session is the playback + // transcode and the capture buffer sits under its wrapper. return jsonResponse({ 'MediaContainer': { 'TranscodeSession': [ - {'timeStamp': '1700000100', 'minOffsetAvailable': '0', 'maxOffsetAvailable': '300'}, + {'timeStamp': '1700000230.5', 'minOffsetAvailable': '0.033', 'maxOffsetAvailable': '12'}, + ], + 'CaptureBuffer': [ + { + 'TranscodeSession': [ + {'timeStamp': '1700000100', 'minOffsetAvailable': '0', 'maxOffsetAvailable': '300'}, + ], + }, ], }, }); @@ -316,8 +325,36 @@ void main() { expect(timelineQuery!['state'], 'playing'); expect(timelineQuery!['time'], '2000000'); expect(timelineQuery!['duration'], '2000000'); - expect(updated, isNotNull); - expect(updated!.seekableDurationSeconds, 300); + expect(updated!.captureBuffer!.seekableDurationSeconds, 300); + expect(updated.captureBuffer!.startedAt, 1700000100); + // The playback transcode's origin is the exact clock anchor (#2100); + // it must not be mistaken for the capture window. + expect(updated.playbackStream!.startedAt, 1700000230.5); + }); + + test('reportTimeline reads a lone top-level session as the capture buffer', () async { + final client = makeClient((request) async { + if (request.url.path.endsWith('/tune')) { + return jsonResponse(tuneResponse()); + } + if (request.url.path == '/:/timeline') { + return jsonResponse({ + 'MediaContainer': { + 'TranscodeSession': [ + {'timeStamp': '1700000100', 'minOffsetAvailable': '0', 'maxOffsetAvailable': '300'}, + ], + }, + }); + } + return jsonResponse(const {}); + }); + addTearDown(client.close); + + final session = (await client.liveTv.startPlayback('ch-1', dvrKey: 'dvr-1'))!; + final updated = await session.reportTimeline(state: 'playing', positionMs: 1000, durationMs: 1800000); + + expect(updated!.captureBuffer!.seekableDurationSeconds, 300); + expect(updated.playbackStream, isNull); }); test('reportTimeline does not fail over because it keeps the active live session alive', () async {