From e91539d1130d4cdca1abfb46b5a1c92f26f593fe Mon Sep 17 00:00:00 2001 From: foxyz Date: Sun, 18 Oct 2026 01:27:34 +0300 Subject: [PATCH] Relay: viewers keep the file size (headers within 4 s, connection: close), retries back off, 412 re-resolves Two server behaviours left ffmpeg (8.1/9.0) without the file size, after which every seek failed with "size -38" until a reload: - dart:io holds response headers until body bytes exist: a stalled source meant no headers within mpv's 5 s timeout. Relay byte responses are now written on the raw connection, headers with the first bytes or after 4 s, and a swap ends waiting reads with a plain close. - every response closed its socket without saying connection: close, so ffmpeg 8.x reused a dead connection for its next seek (pol's X clips). The server now says it whenever the request asks. A failed pull waits 250 ms doubling to 8 s instead of re-pulling at once (the ~130 ms storm that drew Bilibili's 412/403); 412 counts as an expired URL and re-resolves the page, at most once per 60 s. The app reloads a player once per media when ffmpeg reports the lost size anyway. P37: a read that waits past ffmpeg's retry budget still ends the file early. Co-Authored-By: Claude Opus 5.5 --- .../lib/src/platform/ports/player_host.dart | 7 + .../lib/src/playback/playback_controller.dart | 16 +- .../lib/src/session/session_coordinator.dart | 14 ++ apps/desktop/test/stream_lost_test.dart | 20 +++ docs/ARCHITECTURE.md | 40 ++++- docs/DEVELOPMENT.md | 11 +- docs/PROTOCOL.md | 8 +- docs/ROADMAP.md | 8 + packages/core/lib/src/constants.dart | 7 + packages/core/lib/src/link_probe.dart | 36 ++-- packages/core/test/page_relay_test.dart | 107 ++++++++++++ packages/server/lib/src/media_endpoint.dart | 20 +++ packages/server/lib/src/piece_buffer.dart | 12 +- packages/server/lib/src/relay_buffer.dart | 53 +++++- packages/server/lib/src/relay_routes.dart | 121 ++++++++++---- packages/server/lib/src/sync_server.dart | 1 + packages/server/test/relay_test.dart | 157 ++++++++++++++++++ 17 files changed, 575 insertions(+), 63 deletions(-) create mode 100644 apps/desktop/test/stream_lost_test.dart create mode 100644 packages/core/test/page_relay_test.dart diff --git a/apps/desktop/lib/src/platform/ports/player_host.dart b/apps/desktop/lib/src/platform/ports/player_host.dart index 3be60738..74570364 100644 --- a/apps/desktop/lib/src/platform/ports/player_host.dart +++ b/apps/desktop/lib/src/platform/ports/player_host.dart @@ -184,6 +184,13 @@ abstract interface class PlayerHost { int get live; } +/// [log] says ffmpeg's HTTP stream lost the file's size and so every +/// seek on it fails ("Seek failed (to 3417735, size -38)": -38 is ENOSYS, +/// what ffmpeg answers for the size once a reconnect's or a seek's request +/// failed). Only a fresh open gets it back. +bool mpvStreamLost(PlayerLogLine log) => + log.prefix == 'ffmpeg' && log.text.contains('Seek failed') && log.text.contains('size -38)'); + /// [log] says the media could not be opened: before a [duration] is known, /// mpv's fatal "nothing to play" or a demuxer error. bool mpvLoadFailed(PlayerLogLine log, Duration duration) { diff --git a/apps/desktop/lib/src/playback/playback_controller.dart b/apps/desktop/lib/src/playback/playback_controller.dart index 550163a1..72c2cae0 100644 --- a/apps/desktop/lib/src/playback/playback_controller.dart +++ b/apps/desktop/lib/src/playback/playback_controller.dart @@ -63,6 +63,10 @@ class PlaybackController { /// e.g. a CDN's 403), at most once per [beginOpen]. void Function(String error)? onLoadError; + /// The current media's HTTP stream lost its size (see [mpvStreamLost]): + /// at most once per player and media; the coordinator opens it again. + void Function()? onStreamLost; + /// Seeks from the controls (seek bar, the video widget's own) go here /// instead of to mpv. void Function(Duration position)? onUserSeek; @@ -71,6 +75,7 @@ class PlaybackController { // load error was reported. String? _httpError; bool _loadErrorSent = false; + bool _streamLostSent = false; /// The current player. Only while [alive]. HostedPlayer get player => _player!; @@ -144,6 +149,7 @@ class PlaybackController { _chapterJson = ''; _httpError = null; _loadErrorSent = false; + _streamLostSent = false; _shaders = ''; _loaded = loaded && p.state.duration > Duration.zero; _opened = loaded; @@ -196,6 +202,7 @@ class PlaybackController { _chapterJson = ''; _httpError = null; _loadErrorSent = false; + _streamLostSent = false; _loaded = false; chapters.value = const []; tracks.value = const []; @@ -286,7 +293,14 @@ class PlaybackController { // error while shaders are on is a shader that failed to load or compile // (mpv then renders without it). void _onLog(PlayerLogLine log) { - if (log.prefix == 'ffmpeg') return; + if (log.prefix == 'ffmpeg') { + if (!_streamLostSent && mpvStreamLost(log)) { + _streamLostSent = true; + _log.warning('the media stream lost its size, every seek fails: [ffmpeg] ${log.text.trim()}'); + onStreamLost?.call(); + } + return; + } if (!_loadErrorSent && mpvLoadFailed(log, duration.peek())) { _loadErrorSent = true; final http = _httpError; diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index 8a073f8d..36d6672d 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -426,6 +426,7 @@ class SessionCoordinator { if (_playerReady) return; try { playback.onLoadError = _setPlayerError; + playback.onStreamLost = _streamLost; playback.onUserSeek = (d) => _userSeek(d.inMicroseconds / 1e6); _bufferSub ??= playback.buffered.subscribe((_) => _onBuffered()); await media.start(); @@ -479,6 +480,19 @@ class SessionCoordinator { } bool _reloadingVideo = false; + // The media whose lost stream was reloaded already (once per media). + int _streamReloadedFor = -1; + + /// The player's HTTP stream lost the file's size (an old server, a + /// request that timed out): every seek on it would fail until a fresh + /// open. "Reload video" once per media, as the user would. + void _streamLost() { + if (_streamReloadedFor == _mediaId || _mediaId < 0) return; + _streamReloadedFor = _mediaId; + _log.info('media $_mediaId: the stream lost its size; reloading the video once'); + unawaited(reloadVideo()); + } + void Function()? _bufferSub; /// "Reload video" (button, Ctrl+R, status menu): fresh diff --git a/apps/desktop/test/stream_lost_test.dart b/apps/desktop/test/stream_lost_test.dart new file mode 100644 index 00000000..0c552180 --- /dev/null +++ b/apps/desktop/test/stream_lost_test.dart @@ -0,0 +1,20 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:starsync/src/platform/ports/player_host.dart'; + +// ffmpeg's lost stream (its HTTP context lost the file's size, every seek +// fails until a fresh open): the lines the coordinator reloads the video +// on, once per media. +void main() { + test('a seek that failed for want of the size is a lost stream', () { + expect(mpvStreamLost(const PlayerLogLine('ffmpeg', 'error', 'Seek failed (to 3417735, size -38)')), isTrue); + expect(mpvStreamLost(const PlayerLogLine('ffmpeg', 'warn', 'Seek failed (to 11997, size -38)\n')), isTrue); + }); + + test('anything else is not', () { + // A seek that failed with the size known, a reconnect, another prefix. + expect(mpvStreamLost(const PlayerLogLine('ffmpeg', 'error', 'Seek failed (to 20851, size 1023081)')), isFalse); + expect(mpvStreamLost(const PlayerLogLine('ffmpeg', 'warn', 'http: Will reconnect at 3407872 in 0 second(s)')), + isFalse); + expect(mpvStreamLost(const PlayerLogLine('lavf', 'warn', 'Seek failed (to 1, size -38)')), isFalse); + }); +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7565262c..48aeb92e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -561,6 +561,12 @@ opening/initializing the selected video_out") is opened again once, 500 ms later (`render: the video output was not up for the open`); the player error it raised clears when the media loads. +A player whose HTTP stream lost the file's size (ffmpeg's "Seek failed +(to N, size -38)": a request of its that got no headers in time, from a +server without the relay's header rule, or the connection reuse above) +fails every seek until a fresh open, so the coordinator runs "Reload +video" once per media for it (`mpvStreamLost`, `onStreamLost`). + Render mode: media_kit's Linux/Windows video output creates mpv's render context (OpenGL, else its software renderer) when the `VideoController` is created, and nothing in its API says which. The host therefore requests @@ -899,7 +905,7 @@ How bytes move: kept; segments are evicted by distance (behind costs 4x), sizes estimated from the bytes per second of the pieces in so far (before that 4 Mbit/s video, 160 kbit/s audio, which also splits the budget). - A failed pull is retried after 1 s. Measured (mpv 0.41 through the + A failed pull is retried with backoff, as for bytes. Measured (mpv 0.41 through the dedicated server, loopback, ~30 MB/s upstream): Mux's TS HLS (1080p, one muxed playlist) opened in 0.8 s, seeks 0.07 to 1.1 s; Akamai's DASH (1080p + aac, two playlists, EDL) 1.3 s, seeks 0.03 to 1.4 s; an X @@ -908,7 +914,11 @@ How bytes move: opened in 0.02 s from the buffer. - Filling a byte buffer: a viewer blocked on a slice first (oldest wait first, handing the source over at slice boundaries), then the pinned - slices, then read-ahead from the slowest viewer. A viewer reads the + slices, then read-ahead from the slowest viewer. A failed pull (the + source's upstream dropping, a CDN refusing) is followed by the next + one after 250 ms, doubling up to 8 s (`relayRetryDelay`), back to none + once a slice or a pull completes: a failing CDN is not asked again + every round trip (that drew Bilibili's 412/403). A viewer reads the slice being filled as its bytes arrive, not once it is whole (mpv's demuxer, blocked in a read, takes a seek only once that read returns). Nothing is fetched while nobody watches through the relay (a Jellyfin @@ -969,7 +979,23 @@ How bytes move: its source before opening its own player, and the source reads its tracks' sizes at once. - Source away (D6): viewers keep playing from the buffer; a request - waits up to 60 s for missing bytes. The source member going `gone` + waits up to 60 s for missing bytes. +- A viewer's byte answer is written on the raw connection (relic + `Hijack`, `_rangeResponse`): its headers go out with the first bytes, + or after 4 s without them (`relayHeaderWaitMs`). ffmpeg gives up on a + request whose headers do not come within mpv's 5 s `network-timeout` + and then has lost the file's size for good (8.x and 9.0 do not restore + it): every later seek fails with "Seek failed (to N, size -38)". With + the headers in time, a stalled read only times out and is reconnected + (`reconnect_*`, see "Media pipeline"), and plays on once the bytes come. + The answer always says `connection: close` and closes after its body: + ffmpeg asks for close, and ffmpeg 8.x, not told so, sends its next seek + on the connection after reading a range to its end (an mp4 with a box + after `mdat`, like X's clips: the 58-byte tail, then back to the + start) and fails it the same way. Other answers say so too when asked + (`closeWhenAsked`, every server route). A swap that ends the relay + closes its viewers' answers plainly (`RelayEnded`, not logged as a + broken read). The source member going `gone` does not end the relay: it lives through the source's grace (`media.relay.sourceGone`, `ServerHub.refreshRelaySource`). The source (`RelaySource`, kept by the media pipeline's `RelayFeeder` through the @@ -1150,9 +1176,11 @@ all three link modes. Nothing is bundled or installed by hand. track's playlist must be readable, else the swap is refused ("Live streams can't be passed on", DRM, or nothing a relay can read) with a hint to use "Everyone fetches it". The swap names the tracks; its relay - source reads them with yt-dlp's headers (`PageRelayMedia`). On 403/410 - (expired format URLs, 6 h on YouTube; signed segment URLs) from any - track it resolves the page again once for all tracks, finds the same + source reads them with yt-dlp's headers (`PageRelayMedia`). On + 403/410/412 (expired format URLs, 6 h on YouTube; signed segment URLs; + Bilibili's 412) from any track it resolves the page again once for all + tracks (not again within 60 s: a further refusal fails the read and the + server's backoff asks later), finds the same format ids with the same sizes and piece counts and roles (playlists re-read and re-mapped by index; else "the video changed on the page") and goes on from where it stopped. Format URLs are bound to the diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index da305e57..57f5e39b 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -227,11 +227,12 @@ Traps: so shell tests run the real player layout without it; otherwise test pure logic (`StateDiff`, coordinator without widgets). - A `SyncServer` started inside a test: relic reports an error writing a - response (a relay response cut short when its relay is dropped at a - swap or stop) to the zone the server started in, which fails the - test; outside tests it is only logged. Start it in `runZonedGuarded` - and fail on anything else (`queue_switch_e2e`, `relay_e2e`): a viewer's - read is still in flight at the `stop` in tearDown. + response (one cut short at `stop`: `/m`, `/j`; the relay's byte + answers are written on the raw connection and end on their own) to + the zone the server started in, which fails the test; outside tests it + is only logged. Start it in `runZonedGuarded` and fail on anything + else (`queue_switch_e2e`, `relay_e2e`): a viewer's read can still be + in flight at the `stop` in tearDown. - The Link page probes a filled-in link (network, yt-dlp), which never answers on the widget-test clock: pass `AppShell(linkProbe:)` / `LibraryTab(probe:)` a fake, or `pumpAndSettle` never returns. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index e3f1f9b5..34ff435f 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -543,8 +543,12 @@ anything else means one file. one-file URL of a media with `tracks`) or piece (or `p/` of a byte track), 410 unknown session, 503 + `retry-after` the source never connected or the server could not read the link (60 s), or a piece did - not arrive within 60 s. A byte request whose bytes do not arrive within - 60 s ends early. Bytes count toward `bytesServed`. + not arrive within 60 s. A byte request's headers come with its first + bytes, or after 4 s without them (under mpv's 5 s timeout); its body + then waits for missing bytes, up to 60 s for each, else the connection + is reset. Every byte answer says `connection: close` and closes after + its body; a swap that ends the relay closes it early without a reset. + Bytes count toward `bytesServed`. - `GET /r///source` (WebSocket): the source's relay channel, dialled by the source (works behind NAT); `?q=` for the warm relay of that item (or the media's relay started from it). Only that diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 15d2da0a..f7006fa5 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -58,6 +58,14 @@ Problems are numbered; delete a line when its fix ships. - P36 After a reconnect the app re-dials the warm relay of an Up next item that was already played ("relay channel refused: not the source of Up next item q8"): harmless, the feeder should end with its item. +- P37 A relay viewer whose read waits on bytes the source does not send + for longer than ffmpeg's reconnect budget (4 tries per read, about + 40 s with the 4 s header rule) gets an I/O error that mpv takes as the + end of the file. Measured with mpv 0.41 / ffmpeg 9.0.2 and the source + away about a minute across a seek: it played on; one read stalled past + the budget ends the file. Options: a longer `reconnect_delay_max` for + relay URLs only, or the end-of-file path reopening a relay viewer + that stopped short while the server plays. **Restructure** (in order; reviewed against the code). Truth lives in three places: the server (members, presence, gate, transport, queue incl. diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index a1aff6d4..70a0d31b 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -31,6 +31,13 @@ abstract class RelayConstants { /// A viewer request waits this long for its bytes (source away) before /// the server gives up on it. static const int relayWaitSecs = 60; + /// Failed pulls in a row back off: 250 ms doubling up to this, back to + /// none once bytes come again. + static const int relayRetryMaxMs = 8000; + /// A relay viewer's response headers wait at most this long for its + /// first bytes, under mpv's 5 s `network-timeout`: a header that never + /// comes in time costs ffmpeg the file's size for good. + static const int relayHeaderWaitMs = 4000; /// The warm relay (Up next's first relayed item, before it goes up) /// keeps this much ahead of and behind its viewers' preloads, besides /// the pins: about what a preload player reads ahead (48 MiB), never the diff --git a/packages/core/lib/src/link_probe.dart b/packages/core/lib/src/link_probe.dart index 402e2206..6533baa7 100644 --- a/packages/core/lib/src/link_probe.dart +++ b/packages/core/lib/src/link_probe.dart @@ -282,9 +282,12 @@ Future _readText(HttpClientResponse res) async { /// audio, or segmented playlists). A byte track is range-read with /// yt-dlp's headers in GETs of the format's `http_chunk_size`; a pieces /// track's playlist (or DASH fragments) is read on first use, then each -/// piece is one GET. Format URLs expire: on 403/410 from any track the page -/// is resolved again once for all tracks (same format ids, same sizes and -/// piece counts) and the read goes on from where it stopped. +/// piece is one GET. Format URLs expire: on 403/410/412 from any track the +/// page is resolved again once for all tracks (same format ids, same sizes +/// and piece counts) and the read goes on from where it stopped. Another +/// expiry within [refreshGap] of that resolve fails the read instead (the +/// server backs off and asks again), so a site refusing the fresh URLs too +/// is not resolved again and again. class PageRelayMedia implements RelayMedia { final Uri page; final YtDlp ytDlp; @@ -296,7 +299,10 @@ class PageRelayMedia implements RelayMedia { final _http = HttpClient(); var _gen = 0; Future? _refreshing; - PageRelayMedia(this.page, this.ytDlp, {List? tracks}) : _picked = tracks; + DateTime? _refreshedAt; + final Duration refreshGap; + PageRelayMedia(this.page, this.ytDlp, {List? tracks, this.refreshGap = const Duration(seconds: 60)}) + : _picked = tracks; Future _info() async { final info = await ytDlp.resolve(page.toString()); @@ -335,14 +341,22 @@ class PageRelayMedia implements RelayMedia { return fetchHlsPlaylist(_http, f.url, f.headers); } - static bool _expired(Object e) => e is UpstreamHttpException && (e.status == 403 || e.status == 410); + /// A format URL the site no longer takes: 403, 410, or Bilibili's 412. + static bool _expired(Object e) => + e is UpstreamHttpException && (e.status == 403 || e.status == 410 || e.status == 412); /// Resolves the page again: the same format ids with the same sizes and /// piece shapes, else the video changed. - Future _refresh(int gen) { + Future _refresh(int gen, Object cause) { if (gen != _gen) return Future.value(); + final last = _refreshedAt; + if (_refreshing == null && last != null && DateTime.now().difference(last) < refreshGap) { + // Resolved a moment ago and refused again: not once more. + return Future.error(cause); + } return _refreshing ??= () async { - _log.info('relay of ${logUrl(page)}: a format URL expired (403/410), resolving the page again'); + _refreshedAt = DateTime.now(); + _log.info('relay of ${logUrl(page)}: a format URL expired (HTTP ${cause is UpstreamHttpException ? cause.status : '?'}), resolving the page again'); try { const changed = LinkProbeException('the video changed on the page'); final info = await _info(); @@ -384,7 +398,7 @@ class PageRelayMedia implements RelayMedia { return _sizes[i] = await _open(i).size(); } catch (e) { if (!_expired(e)) rethrow; - await _refresh(gen); + await _refresh(gen, e); return _sizes[i] = await _open(i).size(); } } @@ -397,7 +411,7 @@ class PageRelayMedia implements RelayMedia { return _pieces[i] = await _loadPieces(_picked![i].format); } catch (e) { if (!_expired(e)) rethrow; - await _refresh(gen); + await _refresh(gen, e); return _pieces[i] ??= await _loadPieces(_picked![i].format); } } @@ -416,7 +430,7 @@ class PageRelayMedia implements RelayMedia { } catch (e) { if (retried || !_expired(e)) rethrow; retried = true; - await _refresh(gen); + await _refresh(gen, e); } } } @@ -435,7 +449,7 @@ class PageRelayMedia implements RelayMedia { } catch (e) { if (retried || !_expired(e)) rethrow; retried = true; - await _refresh(gen); + await _refresh(gen, e); } } } diff --git a/packages/core/test/page_relay_test.dart b/packages/core/test/page_relay_test.dart new file mode 100644 index 00000000..2db61a26 --- /dev/null +++ b/packages/core/test/page_relay_test.dart @@ -0,0 +1,107 @@ +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:sync_relay_core/core.dart'; +import 'package:test/test.dart'; + +// A page relayed by its formats: an expired format URL (403/410/412) is +// resolved again once, and not again and again when the site keeps +// refusing. + +/// yt-dlp that resolves the page to the origin's current format URL. +class _CountingYtDlp extends YtDlp { + final Uri Function() url; + var resolves = 0; + _CountingYtDlp(super.dir, this.url); + + @override + Future resolve(String page, {Duration timeout = const Duration(seconds: 25)}) async { + resolves++; + final f = YtDlpFormat(id: 'v', url: url()); + return YtDlpInfo(title: 'page', tracks: [(kind: '', format: f)], formats: {'v': f}); + } +} + +void main() { + late HttpServer origin; + late Directory tmp; + // Only `/v.mp4` of the current generation is served; older ones get + // [refusal] (a format URL that expired). + var gen = 1; + var refusal = 412; + final bytes = Uint8List.fromList(List.generate(100000, (i) => i % 251)); + + setUp(() async { + gen = 1; + refusal = 412; + tmp = await Directory.systemTemp.createTemp('page-relay'); + origin = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + origin.listen((req) async { + if (req.uri.path != '/v$gen.mp4') { + req.response.statusCode = refusal; + await req.response.close(); + return; + } + final m = RegExp(r'bytes=(\d+)-(\d+)').firstMatch(req.headers.value('range') ?? '')!; + final start = int.parse(m.group(1)!), end = int.parse(m.group(2)!); + req.response + ..statusCode = 206 + ..headers.contentType = ContentType('video', 'mp4') + ..headers.set('content-range', 'bytes $start-$end/${bytes.length}') + ..add(bytes.sublist(start, end + 1)); + await req.response.close(); + }); + }); + + tearDown(() async { + await origin.close(force: true); + await tmp.delete(recursive: true); + }); + + Uri current() => Uri.parse('http://127.0.0.1:${origin.port}/v$gen.mp4'); + + Future> read(RelayTrack t, int start, int end) async => + [for (final c in await t.bytes!.read(start, end).toList()) ...c]; + + for (final status in [403, 410, 412]) { + test('HTTP $status: the page is resolved again once and the read goes on', () async { + refusal = status; + final y = _CountingYtDlp(tmp.path, current); + final media = PageRelayMedia(Uri.parse('https://example.com/watch'), y); + addTearDown(media.close); + final t = (await media.open()).single; + expect(await t.bytes!.size(), bytes.length); + expect(await read(t, 0, 999), bytes.sublist(0, 1000)); + gen = 2; // the first URL expired + expect(await read(t, 5000, 5999), bytes.sublist(5000, 6000)); + expect(y.resolves, 2); + }); + } + + test('refused again right after a resolve: the read fails, no resolve storm', () async { + final y = _CountingYtDlp(tmp.path, current); + final media = PageRelayMedia(Uri.parse('https://example.com/watch'), y); + addTearDown(media.close); + final t = (await media.open()).single; + await t.bytes!.size(); + gen = 2; + expect(await read(t, 0, 9), bytes.sublist(0, 10)); + gen = 3; // the fresh URL is refused too + for (var i = 0; i < 5; i++) { + await expectLater(read(t, 0, 9), throwsA(isA().having((e) => e.status, 'status', 412))); + } + expect(y.resolves, 2); + }); + + test('another expiry after the gap resolves again', () async { + final y = _CountingYtDlp(tmp.path, current); + final media = PageRelayMedia(Uri.parse('https://example.com/watch'), y, refreshGap: Duration.zero); + addTearDown(media.close); + final t = (await media.open()).single; + await t.bytes!.size(); + for (gen = 2; gen <= 4; gen++) { + expect(await read(t, 100, 199), bytes.sublist(100, 200)); + } + expect(y.resolves, 4); + }); +} diff --git a/packages/server/lib/src/media_endpoint.dart b/packages/server/lib/src/media_endpoint.dart index c460b5d0..a587cb49 100644 --- a/packages/server/lib/src/media_endpoint.dart +++ b/packages/server/lib/src/media_endpoint.dart @@ -249,3 +249,23 @@ MimeType _mimeOf(String? contentType) { return MimeType.octetStream; } } + +/// Says `connection: close` on a response to a request that asked for it +/// (ffmpeg always does). dart:io closes such a connection after the +/// response but, behind relic, without saying so; ffmpeg (8.x) then sends +/// its next seek on that dead connection after reading a range to its end +/// (an mp4's tail) and the seek fails for good ("size -38"). +Handler closeWhenAsked(Handler next) => (req) async { + final res = await next(req); + if (res is! Response) return res; + bool asked; + try { + asked = req.headers.connection?.isClose ?? false; + } catch (_) { + asked = false; + } + if (!asked) return res; + return res.copyWith( + headers: res.headers.transform( + (h) => h.connection = const ConnectionHeader.directives([ConnectionHeaderType.close]))); + }; diff --git a/packages/server/lib/src/piece_buffer.dart b/packages/server/lib/src/piece_buffer.dart index 22737638..907b389b 100644 --- a/packages/server/lib/src/piece_buffer.dart +++ b/packages/server/lib/src/piece_buffer.dart @@ -81,6 +81,7 @@ class PieceBuffer implements TrackBuffer { /// null before a minute of it has arrived. double? get bytesPerSec => _knownSecs >= 60 ? _knownBytes / _knownSecs : null; DateTime? _retryAt; + var _failures = 0; // failed pulls in a row (backoff) var _disposed = false; Completer _kick = Completer(); final Completer _ready = Completer(); @@ -167,6 +168,7 @@ class PieceBuffer implements TrackBuffer { final pl = _pull; if (pl == null || pl.id != id || _disposed) return; _pull = null; + _failures = 0; final data = pl.buf.takeBytes(); final i = pl.piece; _pieces[i] = data; @@ -182,12 +184,18 @@ class PieceBuffer implements TrackBuffer { _wake(); } + static final _pullFailThrottle = LogThrottle(burst: 3); + @override void feedFail(int id, String reason) { if (_pull?.id != id) return; - _log.warning('relay piece pull $id failed: $reason'); _pull = null; - _retryAt = DateTime.now().add(const Duration(seconds: 1)); + final delay = relayRetryDelay(++_failures); + _retryAt = DateTime.now().add(delay); + final pass = _pullFailThrottle.allow('${identityHashCode(this)}'); + if (pass != null) { + _log.warning('${pass}relay piece pull $id failed ($_failures in a row, next in ${delay.inMilliseconds} ms): $reason'); + } _wake(); } diff --git a/packages/server/lib/src/relay_buffer.dart b/packages/server/lib/src/relay_buffer.dart index 0b99689c..5b9414dd 100644 --- a/packages/server/lib/src/relay_buffer.dart +++ b/packages/server/lib/src/relay_buffer.dart @@ -9,6 +9,21 @@ import 'relay_stats.dart'; final _log = createLogger('relay'); +/// A viewer's read ended because its relay went (a swap, server stop): +/// not a failure, the response just ends. +class RelayEnded implements Exception { + const RelayEnded(); + @override + String toString() => 'relay ended'; +} + +/// The wait before the next pull after [failures] failed ones in a row: +/// 250 ms doubling to [RelayConstants.relayRetryMaxMs] (a source whose +/// upstream keeps failing must not be asked again and again at once). +Duration relayRetryDelay(int failures) => failures <= 0 + ? Duration.zero + : Duration(milliseconds: min(RelayConstants.relayRetryMaxMs, 250 * (1 << min(failures - 1, 10)))); + /// One track's buffer on the server: a byte stream ([RelayBuffer]) or /// pieces ([PieceBuffer]). abstract interface class TrackBuffer { @@ -209,6 +224,9 @@ class RelayBuffer implements TrackBuffer { var _reqSeq = 0; var _disposed = false; + // Failed pulls in a row, and when the next pull may start (backoff). + var _failures = 0; + DateTime? _retryAt; Completer _kick = Completer(); final Completer _ready = Completer(); @@ -292,7 +310,11 @@ class RelayBuffer implements TrackBuffer { off += take; pl.pos += take; final done = pl.pos == _sliceEnd(s) + 1; - if (done) _slices[s] = pl.takeSlice(); + if (done) { + _slices[s] = pl.takeSlice(); + // A whole slice came: the source works again. + _failures = 0; + } // Whole, or more of it: whoever waits on it reads on. for (final c in _waiters.remove(s) ?? const >[]) { c.complete(); @@ -309,17 +331,27 @@ class RelayBuffer implements TrackBuffer { final pl = _pull; if (pl != null && pl.id == id) { _pull = null; + _failures = 0; _wake(); } } @override void feedFail(int id, String reason) { - if (_pull?.id != id) return; - _log.warning('relay pull $id failed: $reason'); - feedEnd(id); + final pl = _pull; + if (pl == null || pl.id != id) return; + _pull = null; + final delay = relayRetryDelay(++_failures); + _retryAt = DateTime.now().add(delay); + final pass = _pullFailThrottle.allow('${identityHashCode(this)}'); + if (pass != null) { + _log.warning('${pass}relay pull $id failed ($_failures in a row, next in ${delay.inMilliseconds} ms): $reason'); + } + _wake(); } + static final _pullFailThrottle = LogThrottle(burst: 3); + // ------------------------------------------------------------- scheduling bool _shouldStop(_Pull pl, int done) { @@ -412,6 +444,12 @@ class RelayBuffer implements TrackBuffer { await _sleep(); continue; } + final retry = _retryAt; + if (retry != null && DateTime.now().isBefore(retry)) { + await _sleep(retry.difference(DateTime.now())); + continue; + } + _retryAt = null; _evict(); final s = _nextMissing(); if (s == null) { @@ -431,10 +469,10 @@ class RelayBuffer implements TrackBuffer { } } - Future _sleep() async { + Future _sleep([Duration limit = const Duration(seconds: 5)]) async { _kick = Completer(); // Also wake periodically: viewers age out of [_active]. - await _kick.future.timeout(const Duration(seconds: 5), onTimeout: () {}); + await _kick.future.timeout(limit, onTimeout: () {}); } /// The bytes of slice [s] from [from] (offset in the slice) as soon as @@ -499,7 +537,7 @@ class RelayBuffer implements TrackBuffer { /// Streams [start]..[end] to [viewer]. Cancelling the subscription (the /// viewer went away) ends it; a slice that never arrives ends it with an - /// error after [waitLimit]. + /// error after [waitLimit]; the buffer's [dispose] with [RelayEnded]. Stream read(String viewer, int start, int end, {void Function(int n)? onSent}) async* { final v = _viewers.putIfAbsent(viewer, _Viewer.new); final tail = start >= size - pin; @@ -520,6 +558,7 @@ class RelayBuffer implements TrackBuffer { while (off <= end) { final s = off ~/ slice; final data = await _waitBytes(v, s, off - s * slice, me, () => tail || v.newest == me); + if (_disposed) throw const RelayEnded(); if (data == null) throw StateError('relay: bytes at $off did not arrive'); final n = min(data.length, end - off + 1); if (n <= 0) continue; diff --git a/packages/server/lib/src/relay_routes.dart b/packages/server/lib/src/relay_routes.dart index eee44ef0..c9d9eebd 100644 --- a/packages/server/lib/src/relay_routes.dart +++ b/packages/server/lib/src/relay_routes.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:io' show HttpDate, IOSink, Socket; import 'dart:math'; import 'dart:typed_data'; @@ -491,7 +492,7 @@ void mountRelayRoutes( }); }); - Future viewer(Request req, {required bool tracked, bool piece = false}) async { + Future viewer(Request req, {required bool tracked, bool piece = false}) async { final stopwatch = Stopwatch()..start(); final sid = req.pathParameters.raw[#sid] ?? ''; final token = req.pathParameters.raw[#token] ?? ''; @@ -506,17 +507,18 @@ void mountRelayRoutes( } // Every outcome goes to the log (throttled per member there); a 503 // after a long wait is a source that never came up. + void note(int status, {int bytes = 0}) => onLog?.call( + token, + MediaLog( + method: req.method.name, + path: '/r/$sid/$key/${track == null ? '' : '$track/'}${pieceIndex == null ? '' : 'p/$pieceIndex'}', + rangeHeader: rangeHeader, + status: status, + bytesWritten: bytes, + durationMs: stopwatch.elapsedMilliseconds, + )); Response logged(Response res, {int bytes = 0}) { - onLog?.call( - token, - MediaLog( - method: req.method.name, - path: '/r/$sid/$key/${track == null ? '' : '$track/'}${pieceIndex == null ? '' : 'p/$pieceIndex'}', - rangeHeader: rangeHeader, - status: res.statusCode, - bytesWritten: bytes, - durationMs: stopwatch.elapsedMilliseconds, - )); + note(res.statusCode, bytes: bytes); return res; } @@ -571,24 +573,17 @@ void mountRelayRoutes( return logged(Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers)); } final end = res.end ?? size - 1; - final Stream body = buffer - .read(token, res.start, end, onSent: (n) => session.addBytes(token, n)) - .transform(StreamTransformer.fromHandlers(handleError: (e, st, sink) { - _readFails(token, '${r.name} track ${track ?? 0} from ${res.start}: $e'); - sink.addError(e, st); - })); - MimeType mime; - try { - mime = MimeType.parse(res.headers['content-type'] ?? ''); - } catch (_) { - mime = MimeType.octetStream; - } - return logged( - Response(res.status, - body: Body.fromDataStream(body, mimeType: mime, contentLength: end - res.start + 1), - headers: headers), - bytes: end - res.start + 1); - } + final length = end - res.start + 1; + note(res.status, bytes: length); + final what = '${r.name} track ${track ?? 0} from ${res.start}'; + return _rangeResponse(res.status, res.headers, length, + req.method == Method.head + ? null + : buffer.read(token, res.start, end, onSent: (n) => session.addBytes(token, n)), + onBroke: (e) => e is RelayEnded + ? _log.fine('relay ended under a viewer\'s read: $what') + : _readFails(token, '$what: $e')); + } // `:seg` is the name of a single file, the track of several (the router // wants one parameter name per level). @@ -597,6 +592,74 @@ void mountRelayRoutes( app.get('/r/:sid/:token/:mediaId/:seg/p/:piece', (final req) => viewer(req, tracked: true, piece: true)); } +/// A relayed byte range written on the raw connection: the status line +/// and headers go out with the first body bytes, or after [headerWait] +/// without them (dart:io would hold them until body bytes come, and an +/// ffmpeg whose request gets no headers within its 5 s timeout has lost +/// the file's size: every later seek fails with "size -38"). Then the +/// body as the buffer has it; a read waiting on missing bytes keeps the +/// connection open meanwhile. Always `connection: close` and closed after +/// the body: ffmpeg asks for close, and one that is not told so reuses the +/// connection the server closed for its next seek and fails it the same +/// way. A relay that ends ([RelayEnded]) closes the connection plainly; +/// any other failure ([onBroke]) resets it. The client going away cancels +/// the read. +Hijack _rangeResponse(int status, Map headers, int length, Stream? body, + {required void Function(Object e) onBroke, Duration headerWait = const Duration(milliseconds: RelayConstants.relayHeaderWaitMs)}) { + return Hijack((ch) async { + final out = ch.sink; + final head = StringBuffer('HTTP/1.1 $status ${status == 206 ? 'Partial Content' : 'OK'}\r\n') + ..write('date: ${HttpDate.format(DateTime.now())}\r\n'); + headers.forEach((k, v) { + if (k != 'content-length' && k != 'connection') head.write('$k: $v\r\n'); + }); + head.write('content-length: $length\r\nconnection: close\r\n\r\n'); + final it = body == null ? null : StreamIterator(body); + // A read cancelled while it waits ends with what it was waiting for + // (its relay ending, say): moot then. + Future stop() async => it?.cancel().catchError((_) {}); + // The client closing (a seek elsewhere, the player gone) ends the read. + final input = ch.stream.listen(null, + onDone: () => unawaited(stop()), onError: (_) => unawaited(stop()), cancelOnError: true); + var reset = false; + // The first bytes go out with the headers when they come in time. + Future? first = it?.moveNext(); + if (first != null) { + await Future.any([first.then((_) {}, onError: (_) {}), Future.delayed(headerWait)]); + } + try { + out.add(latin1.encode('$head')); + if (out is IOSink) await out.flush(); + while (it != null) { + final bool more; + try { + final next = first ?? it.moveNext(); + first = null; + more = await next; + } catch (e) { + onBroke(e); + reset = e is! RelayEnded; + break; + } + if (!more) break; + out.add(it.current); + if (out is IOSink) await out.flush(); + } + } catch (_) { + // The client went away mid-write: nothing to say. + reset = true; + } finally { + await input.cancel().catchError((_) {}); + await stop(); + if (reset && out is Socket) { + out.destroy(); + } else { + await out.close().catchError((_) {}); + } + } + }); +} + Response _plain(int status) => Response(status, body: Body.fromString('status $status')); final _readFailThrottle = LogThrottle(burst: 3); diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index 9a3bcb5d..c4d086e9 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -121,6 +121,7 @@ class SyncServer { })), ); }); + _app.use('/', closeWhenAsked); final requests = MediaRequestLog(); void logRequest(String token, MediaLog log) => requests(hub.who(token), log); mountSessionMediaRoute(_app, sessions, onLog: logRequest); diff --git a/packages/server/test/relay_test.dart b/packages/server/test/relay_test.dart index 9a8cac3c..d07e1648 100644 --- a/packages/server/test/relay_test.dart +++ b/packages/server/test/relay_test.dart @@ -393,6 +393,64 @@ void main() { expect(carol.welcome.mediaUrl, mpvEdl(['$carolBase/0/$name.m3u8', '$carolBase/1/$name'])); }); + test('a slow source: the headers still come in time, the bytes once they arrive', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + final gated = _GatedBytes(FileRelayBytes(file)); + await startSource(alice, from: RelayMedia.single(gated)); + await _until(() => server.hub.relay!.live, 'live'); + final url = Uri.parse(viewerUrl(bob, mediaIdOf(alice))); + final start = 2 * RelayBuffer.slice + 10; + final sw = Stopwatch()..start(); + final res = await _RawResponse.get(url, range: 'bytes=$start-'); + // No bytes: the headers go out on their own before mpv's 5 s timeout. + final head = await res.head.timeout(const Duration(seconds: 6)); + expect(sw.elapsedMilliseconds, lessThan(5000)); + expect(head, startsWith('HTTP/1.1 206')); + expect(head, contains('content-range: bytes $start-${bytes.length - 1}/${bytes.length}')); + expect(head, contains('content-length: ${bytes.length - start}')); + expect(head, contains('connection: close')); + expect(res.body, isEmpty); + gated.open(); + expect(await res.done.timeout(const Duration(seconds: 10)), isTrue, reason: 'closed plainly'); + expect(res.body, bytes.sublist(start)); + }); + + test('a range read to its end says connection: close (ffmpeg 8 reuses it otherwise)', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + await startSource(alice); + final url = Uri.parse(viewerUrl(bob, mediaIdOf(alice))); + // The 58-byte tail box of an mp4 whose moov comes first, as ffmpeg asks. + final res = await _RawResponse.get(url, range: 'bytes=${bytes.length - 58}-'); + expect(await res.head, contains('connection: close')); + expect(await res.done.timeout(const Duration(seconds: 5)), isTrue); + expect(res.body, bytes.sublist(bytes.length - 58)); + // Any other answer to a request that asks for close says so too. + final health = await _RawResponse.get(Uri.parse('http://127.0.0.1:${server.boundPort}/health')); + expect(await health.head, contains('connection: close')); + await health.done; + }); + + test('a swap ends a viewer\'s waiting read plainly', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + await startSource(alice, from: RelayMedia.single(_GatedBytes(FileRelayBytes(file)))); + await _until(() => server.hub.relay!.live, 'live'); + final res = await _RawResponse.get(Uri.parse(viewerUrl(bob, mediaIdOf(alice))), range: 'bytes=1000-'); + await res.head.timeout(const Duration(seconds: 6)); + alice.send({ + 't': 'swap', + 'media': {'name': 'x', 'size': 0, 'duration': 0, 'url': 'http://example.invalid/x.mkv'}, + }); + // Closed, not reset. + expect(await res.done.timeout(const Duration(seconds: 5)), isTrue); + expect(res.body, isEmpty); + }); + test('Jellyfin swap can carry a relay', () async { final alice = await join('alice'); final before = alice.raw.length; @@ -704,6 +762,28 @@ void main() { await old.cancel(); }); + test('a source that keeps failing is asked again with backoff, at once again after bytes', () async { + final failing = _FailingBytes(FileRelayBytes(file)); + final b = RelayBuffer(ahead: 2 * RelayBuffer.slice, back: RelayBuffer.slice, pin: RelayBuffer.slice); + final f = LocalRelayFeed(RelayTrack('', failing), b); + b.attach(f, bytes.length); + addTearDown(() async { + b.dispose(); + await f.close(); + }); + final reader = b.read('a', 0, 99).toList(); + await Future.delayed(const Duration(milliseconds: 2500)); + // 0, 0.25, 0.75, 1.75 s: not one pull per round trip. + expect(failing.reads, inInclusiveRange(3, 5)); + failing.failing = false; + final got = await reader.timeout(const Duration(seconds: 10)); + expect([for (final c in got) ...c], bytes.sublist(0, 100)); + expect(relayRetryDelay(0), Duration.zero); + expect(relayRetryDelay(1), const Duration(milliseconds: 250)); + expect(relayRetryDelay(3), const Duration(seconds: 1)); + expect(relayRetryDelay(50), const Duration(milliseconds: RelayConstants.relayRetryMaxMs)); + }); + test('dispose frees everything and ends waiting readers', () async { await read('a', 0, 10); buf.dispose(); @@ -762,3 +842,80 @@ class _FakeYtDlp extends YtDlp { Future resolve(String url, {Duration timeout = const Duration(seconds: 25)}) async => YtDlpInfo(title: 'page', tracks: tracks, formats: {for (final t in tracks) t.format.id: t.format}); } + +/// Bytes that come only once [open] is called (a source whose upstream +/// stalls). +class _GatedBytes implements RelayBytes { + final RelayBytes inner; + final _gate = Completer(); + _GatedBytes(this.inner); + void open() => _gate.complete(); + @override + Future size() => inner.size(); + @override + Stream> read(int start, int end) async* { + await _gate.future; + yield* inner.read(start, end); + } + + @override + Future close() => inner.close(); +} + +/// Reads that fail at once while [failing] (an upstream dropping every +/// connection); [reads] counts them. +class _FailingBytes implements RelayBytes { + final RelayBytes inner; + var failing = true; + var reads = 0; + _FailingBytes(this.inner); + @override + Future size() => inner.size(); + @override + Stream> read(int start, int end) { + reads++; + if (failing) return Stream.error(const HttpException('Connection closed while receiving data')); + return inner.read(start, end); + } + + @override + Future close() => inner.close(); +} + +/// One HTTP/1.1 GET on its own socket, as ffmpeg sends it +/// (`Connection: close`): the header block as soon as it comes, the body +/// as it comes, and how the connection ended ([done]: true when the +/// server closed it plainly, false on a reset). +class _RawResponse { + final _head = Completer(); + Future get head => _head.future; + final body = []; + final _done = Completer(); + Future get done => _done.future; + + static Future<_RawResponse> get(Uri url, {String? range}) async { + final r = _RawResponse(); + final s = await Socket.connect(url.host, url.port); + final buf = []; + s.listen((d) { + if (r._head.isCompleted) { + r.body.addAll(d); + return; + } + buf.addAll(d); + final text = latin1.decode(buf); + final at = text.indexOf('\r\n\r\n'); + if (at < 0) return; + r._head.complete(text.substring(0, at).toLowerCase().replaceFirst('http/1.1', 'HTTP/1.1')); + r.body.addAll(buf.sublist(at + 4)); + }, onError: (_) { + if (!r._done.isCompleted) r._done.complete(false); + }, onDone: () { + if (!r._done.isCompleted) r._done.complete(true); + s.destroy(); + }); + s.write('GET ${url.path} HTTP/1.1\r\nHost: ${url.host}\r\n${range == null ? '' : 'Range: $range\r\n'}' + 'Connection: close\r\n\r\n'); + return r; + } +}