Page links played here only also go through a reader of their own: Bilibili's cuts never end playback

Alone, or with Everyone fetches it, a page link with plain-file tracks
plays through SharedLinks.local (a SharedSource without the upload): it
resumes after a cut, backs off on 412/429/5xx and resolves the page again
on 403/410/412, where ffmpeg's own reconnect gave up for good. Pieces
tracks and a reader that cannot open keep yt-dlp's URLs as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/desktop/lib/src/session/media_pipeline.dart

@@ -233,9 +233,24 @@ class MediaPipeline {
       throw ResolveFailure(probe.ytDlpError ?? (probe.info == null ? 'Not a video page yt-dlp knows' : 'no format a player opens'),
           cookies: probe.cookies);
     }
+    if (headers == null || headers.isEmpty) {
+      // Played here only: through a reader of its own too, which resumes
+      // after a cut and gets a fresh address (a CDN cuts mpv's reconnects
+      // off). Pieces tracks, or a reader that cannot open: the site as is.
+      final y = await l.ytDlp;
+      final local =
+          await shared.local(url, probe, () => LinkRelayMedia(Uri.parse(url), ytDlp: y, probe: probe), fresh: fresh);
+      if (local != null) {
+        return _Target(local.source.playerUrl, const {}, subtitles: local.subtitles, shared: true, local: true);
+      }
+    }
     return _Target(play.mpvUrl, {...play.headers, ...?headers}, subtitles: play.subtitles);
   }
 
+  /// The [SharedLinks.local] readers the player and the preload use stay;
+  /// the rest close.
+  void _trimLocal() => shared.trimLocal({_openUrl, _pre?.url});
+
   /// The page's subtitles on [p] once its media is in (not selected: the
   /// usual choice applies).
   Future<void> _addSubtitles(HostedPlayer p, List<YtDlpSubtitle> subs) async {
@@ -303,6 +318,7 @@ class MediaPipeline {
     _openHeaders = headers;
     _openResolve = resolve;
     _opening = true;
+    _trimLocal();
     try {
       trace.mark(mediaId, SwitchTrace.open);
       final promoted = await _openInPlayer(url, headers,
@@ -360,6 +376,7 @@ class MediaPipeline {
     }
     if (headers != null && headers.isNotEmpty) return 'Jellyfin, own sign-in';
     if (shared.has(url)) return 'link passed on, read once here';
+    if (shared.hasLocal(url)) return 'link, read here with resume';
     return 'link, fetched here';
   }
 
@@ -465,6 +482,7 @@ class MediaPipeline {
     _openUrl = null;
     _openHeaders = null;
     await playback.stop();
+    _trimLocal();
   }
 
   /// The same media came back as [r.mediaId] (a restarted server put back
@@ -546,6 +564,7 @@ class MediaPipeline {
     _openHeaders = null;
     _recovery = null;
     if (ready) await playback.stop();
+    _trimLocal();
   }
 
   /// The next state's media is new to this player: it opens again from
@@ -794,9 +813,10 @@ class MediaPipeline {
     _pre = null;
     if (had != null) unawaited(_disposePrepared(had));
     final pre = _pre = _Prepared(url, headers ?? const {});
+    _trimLocal();
     try {
       final target = await _resolve(url, headers, fresh: fresh);
-      pre.shared = target.shared;
+      pre.shared = target.shared && !target.local;
       if (!identical(_pre, pre)) {
         pre.fail();
         return false;
@@ -857,6 +877,7 @@ class MediaPipeline {
     await pre.logSub?.cancel();
     final p = pre.player;
     if (p != null) await host.dispose(p);
+    if (!identical(_pre, pre)) _trimLocal();
   }
 
   /// [pre] becomes the session player; its buffer comes along.
@@ -935,7 +956,10 @@ class _Target {
 
   /// A link this member passes on, read through its shared reader.
   final bool shared;
-  const _Target(this.url, this.headers, {this.subtitles = const [], this.shared = false});
+
+  /// Read through a [SharedLinks.local] reader (played here only).
+  final bool local;
+  const _Target(this.url, this.headers, {this.subtitles = const [], this.shared = false, this.local = false});
 }
 
 /// A page could not be resolved: why ([message], yt-dlp's words) and which

apps/desktop/lib/src/session/shared_links.dart

@@ -20,6 +20,8 @@ class SharedLink {
 /// Each one's player opens [SharedSource.playerUrl] and its relay serves
 /// [SharedSource.upload]. A page with a pieces track (HLS/DASH) is not
 /// shared ([fits]): its player and relay keep their own fetches.
+/// Page links played here only ([local]) get a reader of their own too,
+/// for its resume after cuts.
 class SharedLinks {
   SharedLink? _current;
   ({String queueId, SharedLink link})? _warm;
@@ -146,6 +148,57 @@ class SharedLinks {
   /// [url] is shared here (the current media's or the warm one).
   bool has(String url) => _current?.url == url || _warm?.link.url == url;
 
+  // Page links this app plays without passing them on (alone, or
+  // everyone fetches it), by page URL: the session player's and the
+  // preload's.
+  final _local = <String, Future<SharedLink?>>{};
+
+  /// A page link played here only: read through its own [SharedSource]
+  /// over [media] too, so the player gets the reader's resume after a cut,
+  /// back-off and fresh address (403/410/412) instead of ffmpeg's
+  /// reconnect, which a CDN like Bilibili cuts off for good. The one [url]
+  /// has is kept unless [fresh]. Null when the page does not fit ([fits]:
+  /// a pieces track) or the reader cannot open it: the player reads the
+  /// site itself. [trimLocal] closes the ones no longer needed.
+  Future<SharedLink?> local(String url, LinkProbe probe, RelayMedia Function() media,
+      {bool fresh = false, Duration timeout = const Duration(seconds: 15)}) {
+    final had = _local[url];
+    if (had != null && !fresh) return had;
+    if (had != null) _local.remove(url)!.then((l) => l?.source.close());
+    if (!fits(probe)) return Future.value();
+    Future<SharedLink?> make() async {
+      final s = SharedSource(media());
+      try {
+        // The sizes too: a format URL the site refuses (403) fails here,
+        // and the player then gets the site's own answer.
+        await s.start().timeout(timeout);
+        return SharedLink(url, s, subtitles: _subs(probe));
+      } catch (e) {
+        _log.info('link ${logUrl(url)} read by the player itself: $e');
+        unawaited(s.close());
+        return null;
+      }
+    }
+
+    final f = _local[url] = make();
+    unawaited(f.then((l) {
+      if (l == null && identical(_local[url], f)) _local.remove(url);
+    }));
+    return f;
+  }
+
+  /// [url] is read here through a [local] reader.
+  bool hasLocal(String url) => _local.containsKey(url);
+
+  /// Closes the [local] readers of every URL but [keep] (the player's and
+  /// the preload's).
+  void trimLocal(Set<String?> keep) {
+    for (final url in [..._local.keys]) {
+      if (keep.contains(url)) continue;
+      unawaited(_local.remove(url)!.then((l) => l?.source.close()));
+    }
+  }
+
   Future<void> closeCurrent() async {
     final c = _current;
     _current = null;
@@ -156,6 +209,7 @@ class SharedLinks {
   Future<void> closeAll() async {
     _dropPending();
     dropWarm();
+    trimLocal(const {});
     await closeCurrent();
   }
 }

apps/desktop/test/ytdl_headers_e2e_test.dart

@@ -34,6 +34,14 @@ class _Cdn {
   final Map<String, String> _files;
   int ok = 0, forbidden = 0;
   final cookies = <String?>[];
+  final agents = <String?>[];
+
+  /// Like Bilibili cutting a long read: a response stops after this many
+  /// bytes (the connection dropped), and the next request after a cut is
+  /// refused with 412 (where ffmpeg's reconnect gives up). Null: whole.
+  int? cutAfter;
+  bool _cut = false;
+  int cuts = 0, refused = 0;
 
   String url(String name) => 'http://127.0.0.1:${_server.port}/$name';
 
@@ -61,12 +69,21 @@ class _Cdn {
       await res.close();
       return;
     }
+    if (_cut) {
+      _cut = false;
+      refused++;
+      res.statusCode = HttpStatus.preconditionFailed;
+      await res.close();
+      return;
+    }
     ok++;
     cookies.add(req.headers.value('cookie'));
+    agents.add(req.headers.value('user-agent'));
     final bytes = await File(file).readAsBytes();
     var start = 0, end = bytes.length - 1;
     final range = RegExp(r'bytes=(\d*)-(\d*)').firstMatch(req.headers.value('range') ?? '');
     res.headers.set('accept-ranges', 'bytes');
+    res.headers.contentType = ContentType('video', 'mp4');
     if (range != null) {
       start = int.tryParse(range[1]!) ?? 0;
       end = int.tryParse(range[2]!) ?? end;
@@ -74,6 +91,23 @@ class _Cdn {
       res.headers.set('content-range', 'bytes $start-$end/${bytes.length}');
     }
     res.contentLength = end - start + 1;
+    final cut = cutAfter;
+    if (cut != null && end - start + 1 > cut) {
+      // Headers by hand on the raw socket, part of the body, then dropped.
+      final socket = await res.detachSocket(writeHeaders: false);
+      final status = range == null ? '200 OK' : '206 Partial Content';
+      socket.add(utf8.encode('HTTP/1.1 $status\r\naccept-ranges: bytes\r\ncontent-type: video/mp4\r\n'
+          '${range == null ? '' : 'content-range: bytes $start-$end/${bytes.length}\r\n'}'
+          'content-length: ${end - start + 1}\r\nconnection: close\r\n\r\n'));
+      socket.add(bytes.sublist(start, start + cut));
+      _cut = true;
+      cuts++;
+      try {
+        await socket.flush();
+      } catch (_) {}
+      socket.destroy();
+      return;
+    }
     res.add(bytes.sublist(start, end + 1));
     await res.close();
   }
@@ -88,16 +122,19 @@ Future<AppYtDlp> _fakeYtDlp(_Cdn cdn, {required bool headers, bool fail = false}
   if (dir.existsSync()) dir.deleteSync(recursive: true);
   dir.createSync(recursive: true);
   final h = headers ? {'Referer': _referer, 'User-Agent': 'fake-agent'} : <String, String>{};
+  final formats = [
+    {'format_id': 'v', 'url': cdn.url('v.m4s'), 'protocol': 'https', 'ext': 'mp4',
+      'vcodec': 'avc1.64001F', 'acodec': 'none', 'width': 320, 'height': 180, 'http_headers': h,
+      'cookies': 'SESSDATA=abc; Domain=.127.0.0.1; Path=/'},
+    {'format_id': 'a', 'url': cdn.url('a.m4s'), 'protocol': 'https', 'ext': 'm4a',
+      'vcodec': 'none', 'acodec': 'mp4a.40.2', 'http_headers': h,
+      'cookies': 'SESSDATA=abc; Domain=.127.0.0.1; Path=/'},
+  ];
   final json = jsonEncode({
     'id': 'BV1', 'title': 'Fake page', 'duration': 30, 'extractor': 'fake',
     'format_id': 'v+a', 'ext': 'mp4',
-    'requested_formats': [
-      {'format_id': 'v', 'url': cdn.url('v.m4s'), 'protocol': 'https', 'ext': 'mp4',
-        'vcodec': 'avc1.64001F', 'acodec': 'none', 'width': 320, 'height': 180, 'http_headers': h,
-        'cookies': 'SESSDATA=abc; Domain=.127.0.0.1; Path=/'},
-      {'format_id': 'a', 'url': cdn.url('a.m4s'), 'protocol': 'https', 'ext': 'm4a',
-        'vcodec': 'none', 'acodec': 'mp4a.40.2', 'http_headers': h},
-    ],
+    'formats': formats,
+    'requested_formats': formats,
     'requested_subtitles': {
       'en': {'url': cdn.url('en.vtt'), 'ext': 'vtt', 'name': 'English'},
       'danmaku': {'url': cdn.url('danmaku.xml'), 'ext': 'xml'},
@@ -124,8 +161,9 @@ void main() {
   tearDownAll(() => server.close(force: true));
 
   // A page "everyone fetches" opens through the app's own resolve: its
-  // stream URLs (an EDL of video + audio) with the CDN's headers; mpv's
-  // yt-dlp hook is off.
+  // stream URLs (video + audio) read by the app's own reader
+  // (SharedLinks.local) with the CDN's headers and cookies, mpv on its
+  // loopback EDL; mpv's yt-dlp hook is off.
   test(
     "a page plays from yt-dlp's stream URLs with each CDN's Referer and the cookies; subtitles but danmaku",
     timeout: const Timeout(Duration(minutes: 1)),
@@ -146,6 +184,8 @@ void main() {
         expect(cdn.forbidden, 0, reason: 'every request had the Referer');
         expect(cdn.ok, greaterThanOrEqualTo(2));
         expect(cdn.cookies, everyElement('SESSDATA=abc'));
+        expect(m.shared.hasLocal(cdn.page('BV1fake')), isTrue);
+        expect(cdn.agents, everyElement(startsWith('fake-agent')), reason: 'the reader sends the format\'s headers');
         expect(cdn.subtitles, 1, reason: 'the page\'s subtitles are added, danmaku is not');
         await _until(() => p.tracks.peek().any((t) => t.isExternal), 'the subtitle track listed');
       } finally {
@@ -185,6 +225,43 @@ void main() {
     skip: Platform.isLinux ? false : 'fake yt-dlp is a shell script',
   );
 
+  // Bilibili cuts a long read and refuses the reconnect right after it
+  // (412): ffmpeg's own reconnect ends playback there ("Stream ends
+  // prematurely", "Operation not permitted"); the reader resumes after a
+  // back-off and the player never sees the cut.
+  test(
+    'a CDN that cuts reads and refuses the reconnect (412): played to the end without an error',
+    timeout: const Timeout(Duration(minutes: 1)),
+    () async {
+      final m = MediaPipeline(host: MediaKitHost()..init(), links: await _fakeYtDlp(cdn, headers: true));
+      final p = m.playback;
+      final errors = <String>[];
+      p.onLoadError = errors.add;
+      m.onError = (raw, {cookies}) => errors.add(raw);
+      cdn.cutAfter = 48 * 1024;
+      final cutsBefore = cdn.cuts;
+      try {
+        await m.start();
+        final page = cdn.page('BV4fake');
+        expect(await m.open(page), isTrue);
+        await _until(() => p.duration.peek() > Duration.zero, 'opened', secs: 20);
+        await p.player.setProperty('speed', '8');
+        await p.player.play();
+        await _until(() => p.positionSecs > 25 || p.player.state.completed, 'played to the end', secs: 30);
+        // ignore: avoid_print
+        print('cuts: ${cdn.cuts - cutsBefore}, refused: ${cdn.refused}, errors: $errors');
+        expect(cdn.cuts - cutsBefore, greaterThan(2), reason: 'the CDN did cut');
+        expect(cdn.refused, greaterThan(0));
+        expect(errors, isEmpty);
+        expect(m.shared.hasLocal(page), isTrue);
+      } finally {
+        cdn.cutAfter = null;
+        await m.dispose();
+      }
+    },
+    skip: Platform.isLinux ? false : 'fake yt-dlp is a shell script',
+  );
+
   // D11: yt-dlp finds no video; the error names the cookies it used.
   test('a page yt-dlp finds no video on: the error says which cookies were used', () async {
     final m = MediaPipeline(host: MediaKitHost()..init(), links: await _fakeYtDlp(cdn, headers: true, fail: true));
@@ -203,8 +280,8 @@ void main() {
   }, skip: Platform.isLinux ? false : 'fake yt-dlp is a shell script');
 }
 
-Future<void> _until(bool Function() cond, String what) async {
-  final end = DateTime.now().add(const Duration(seconds: 10));
+Future<void> _until(bool Function() cond, String what, {int secs = 10}) async {
+  final end = DateTime.now().add(Duration(seconds: secs));
   while (!cond()) {
     if (DateTime.now().isAfter(end)) fail('timed out waiting for $what');
     await Future.delayed(const Duration(milliseconds: 100));