diff --git a/README.md b/README.md index c2078e0..9af6bd1 100644 --- a/README.md +++ b/README.md @@ -167,9 +167,12 @@ or an episode on the Schedule for **Add to queue** / **Play next**; the Jellyfin details panel has a queue button (right-click it to play next), and the card for a dropped or pasted video or link has both. A link is checked first, like Put it up. Every app keeps the next item ready, -its first seconds already loaded, so skipping to it starts at once -(not yet for a link passed on, someone else's file or one the server -fetches: those load when they go up). Everyone shares one Up next and anyone can change it; what +its first seconds already loaded, so skipping to it starts at once. +For a link passed on, a member's file or a link the server fetches, +the server starts getting the next item while the current one plays +(from the app of whoever added it, or by fetching the link itself), so +everyone has its start before it goes up; whoever added it keeps the +app open for that. Everyone shares one Up next and anyone can change it; what you added stays when you leave. In the player, Up next sits in the top bar next to the chat button: the diff --git a/apps/desktop/lib/src/queue/queue_controller.dart b/apps/desktop/lib/src/queue/queue_controller.dart index 8aab2f3..64417d8 100644 --- a/apps/desktop/lib/src/queue/queue_controller.dart +++ b/apps/desktop/lib/src/queue/queue_controller.dart @@ -525,28 +525,35 @@ class QueueController { if (head != null) { final performer = queuePerformer(head, r.members).id == _me; final local = _local[head.key]?.pick; + // The server's warm relay of the item, once bytes can flow: what + // everyone without their own copy preloads. + final relay = head.media.relay; + final warmUrl = relay != null && relay.live ? coord.warmRelayUrl(head, relay) : null; switch (head.how) { case QueueHow.jellyfin: final ref = head.media.jellyfin; if (ref == null) return null; if (_ownAccount(ref.serverId)) return _WarmTarget(head.id, ref: ref); - // No account here: the server's login passes it on (`/j`). - final relay = r.relayNext ? coord.jellyfinRelayUrl(ref) : null; - return relay == null ? null : _WarmTarget(head.id, open: relay); + // No account here: the server's login passes it on (`/j`), else + // the warm relay a member with an account feeds. + final open = r.relayNext ? coord.jellyfinRelayUrl(ref) : warmUrl; + return open == null ? null : _WarmTarget(head.id, open: open); case QueueHow.direct: final url = head.media.url; if (url == null) return null; return _WarmTarget(head.id, probe: performer ? url : null, open: url); case QueueHow.through: - // Only its adder has the link; the others' relay starts with the swap. - if (!performer) return null; + // Only its adder has the link (and feeds the warm relay). + if (!performer) return warmUrl == null ? null : _WarmTarget(head.id, open: warmUrl); final url = local is LinkMedia ? local.url : null; return _WarmTarget(head.id, probe: url, open: url); case QueueHow.file: - if (!performer) return null; + if (!performer) return warmUrl == null ? null : _WarmTarget(head.id, open: warmUrl); // The host's own file plays from its `/m` URL once it is up. return _WarmTarget(head.id, open: local is FileMedia && coord.hostFile == null ? local.path : null); case QueueHow.server: + // The server fetches it into its warm relay; everyone preloads that. + if (warmUrl != null) return _WarmTarget(head.id, open: warmUrl); return performer ? _WarmTarget(head.id) : null; } } @@ -564,6 +571,7 @@ class QueueController { /// playing has started. Done as soon as it is first, again before that /// runs out, dropped when it is no longer first. void _warm(StateBroadcast? r) { + _feedWarm(r); final t = _warmTarget(r); final key = t?.key ?? ''; if (key != _warmKey) { @@ -581,6 +589,59 @@ class QueueController { _syncPreload(r); } + // The warm relay this app feeds (`|`), started once. + String _fedWarm = ''; + + /// Feeds the server's warm relay of Up next's first item from here when + /// it names this member as its source (a link passed on, a file), or a + /// Jellyfin item this app puts up with its own account when the server + /// asks for a source; see [SessionCoordinator.feedWarm]. The server + /// drops the relay (and with it the source) once the item is no longer + /// first. + void _feedWarm(StateBroadcast? r) { + final head = r?.queue.firstOrNull; + final relay = head?.media.relay; + if (r == null || head == null || relay == null || relay.fetch) return; + final key = '${head.id}|${coord.sync.token}'; + if (key == _fedWarm) return; + switch (head.how) { + case QueueHow.through: + case QueueHow.file: + if (relay.source != _me) return; + final pick = _local[head.key]?.pick; + if (pick is FileMedia) { + _fedWarm = key; + coord.feedWarm(head.id, RelayMedia.single(FileRelayBytes(File(pick.path)))); + } else if (pick is LinkMedia) { + _fedWarm = key; + unawaited(() async { + try { + // The probe [_resolveWarm] keeps: no second yt-dlp run. + final probe = await coord.ytDlp.probe(pick.url, keep: _warmKeep).timeout(const Duration(seconds: 60)); + final y = await coord.ytDlp.ytDlp; + if (_fedWarm != key) return; + coord.feedWarm(head.id, LinkRelayMedia(Uri.parse(pick.url), ytDlp: y, probe: probe)); + } catch (e) { + debugPrint('queue: feeding ${head.id}: $e'); + } + }()); + } + case QueueHow.jellyfin: + final ref = head.media.jellyfin; + if (ref == null || (relay.source.isNotEmpty && relay.source != _me)) return; + if (queuePerformer(head, r.members).id != _me) return; + // Resolved with this app's account by [_resolveWarm]. + final w = coord.warmJellyfin; + if (w == null || !sameJellyfinItem(w.ref, ref)) return; + _fedWarm = key; + coord.feedWarm( + head.id, RelayMedia.single(HttpRelayBytes(Uri.parse(w.plan.url), headers: w.plan.headers))); + case QueueHow.direct: + case QueueHow.server: + return; + } + } + void _dropWarm(StateBroadcast? r) { _warmTimer?.cancel(); _warmTimer = null; @@ -613,6 +674,7 @@ class QueueController { if (plan is OwnPlay) { coord.warmJellyfin = (ref: ref, plan: plan, until: DateTime.now().add(_warmKeep)); open = (url: plan.url, headers: plan.headers); + _feedWarm(coord.serverState.peek()); } } else if (url != null && ok) { open = (url: url, headers: null); diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index 9f02cce..1397cc4 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -680,13 +680,62 @@ class SessionCoordinator { /// This viewer's `/r/` URL; with tracks, a pieces track's playlist, or /// an mpv EDL of every track when there are several (a page's separate - /// video and audio). - String _relayMediaUrlFor(String name, int mediaId, [List tracks = const []]) => - relayViewerUrl('http://$_wsHost:$_wsPort/r/default/${sync.token}/$mediaId', name, tracks: tracks); + /// video and audio). A relay started in Up next ([queueId]) keeps its + /// queue id in the URL, so the preload of its warm relay matches. + String _relayMediaUrlFor(String name, int mediaId, + [List tracks = const [], String queueId = '']) => + relayViewerUrl('http://$_wsHost:$_wsPort/r/default/${sync.token}/${queueId.isEmpty ? mediaId : queueId}', name, + tracks: tracks); + + /// This viewer's URL of the warm relay of Up next's first [item] + /// (`queue[0].media.relay`, see ARCHITECTURE "Getting ready"): the same + /// URL the media has once the item goes up, so the preload is promoted. + String warmRelayUrl(QueueItem item, MediaRelay relay) => + _relayMediaUrlFor(item.media.name.isEmpty ? item.title : item.media.name, -1, relay.tracks, item.id); + + // This member feeds the warm relay of Up next item `queueId` (its + // link, file or Jellyfin stream) before the item goes up; the swap + // that puts it up takes the source over. + ({String queueId, RelaySource src})? _warmSource; + + /// Feeds the warm relay of Up next item [queueId] from here with + /// [media]; one already feeding it is kept. The server pulls only while + /// someone preloads it, and closes the channel once the item is no + /// longer first (the source then stops). + void feedWarm(String queueId, RelayMedia media) { + if (_warmSource?.queueId == queueId) { + unawaited(media.close()); + return; + } + stopWarmSource(); + final src = RelaySource( + Uri.parse('ws://$_wsHost:$_wsPort/r/default/${sync.token}/source?q=${Uri.encodeQueryComponent(queueId)}'), + media); + _warmSource = (queueId: queueId, src: src); + unawaited(() async { + try { + await src.start(); + } catch (e) { + debugPrint('warm relay source of $queueId failed: $e'); + if (identical(_warmSource?.src, src)) _warmSource = null; + return; + } + await src.done; + if (identical(_warmSource?.src, src)) _warmSource = null; + }()); + } + + void stopWarmSource() { + final w = _warmSource; + _warmSource = null; + if (w != null) unawaited(w.src.close()); + } /// Serves [media] for media [mediaId] on this member's relay channel. - /// The server only pulls while someone watches through it. - void _startRelaySource(int mediaId, RelayMedia media, {bool announce = false}) { + /// The server only pulls while someone watches through it. When the + /// media came from Up next item [queueId] whose warm relay this member + /// feeds, that source carries on instead. + void _startRelaySource(int mediaId, RelayMedia media, {bool announce = false, String queueId = ''}) { // Back after a reconnect on the same media: it is still serving (and // redials by itself). if (_relaySource != null && _relaySourceMediaId == mediaId) { @@ -694,6 +743,24 @@ class SessionCoordinator { return; } _stopRelaySource(); + final warm = _warmSource; + if (queueId.isNotEmpty && warm != null && warm.queueId == queueId) { + _warmSource = null; + unawaited(media.close()); + _relaySource = warm.src; + _relaySourceMediaId = mediaId; + relaying.value = true; + if (announce) { + _pushToast('Passing the video on through the server. Keep the app open while it plays.'); + } + unawaited(warm.src.done.then((_) { + if (identical(_relaySource, warm.src)) { + _relaySource = null; + relaying.value = false; + } + })); + return; + } final src = RelaySource( Uri.parse('ws://$_wsHost:$_wsPort/r/default/${sync.token}/source'), media); _relaySource = src; @@ -811,15 +878,17 @@ class SessionCoordinator { if (ref != null) { await _playJellyfin(ref, r, origin: origin is _JellyfinOrigin ? origin : null, - memberRelayUrl: relay != null && !mine ? _relayMediaUrlFor(name, r.mediaId) : null); + memberRelayUrl: + relay != null && !mine ? _relayMediaUrlFor(name, r.mediaId, const [], relay.queueId) : null); } else if (origin is _FileOrigin) { await _openMedia(origin.path); - _startRelaySource(r.mediaId, RelayMedia.single(FileRelayBytes(File(origin.path))), announce: true); + _startRelaySource(r.mediaId, RelayMedia.single(FileRelayBytes(File(origin.path))), + announce: true, queueId: relay?.queueId ?? ''); } else if (origin is _LinkOrigin) { await _openMedia(origin.url); _startRelaySource( r.mediaId, LinkRelayMedia(Uri.parse(origin.url), ytDlp: origin.ytDlp, probe: origin.probe), - announce: true); + announce: true, queueId: relay?.queueId ?? ''); } else if (relay != null && !mine) { if (relay.fetch && !relay.live) { // The server is still getting it (yt-dlp takes seconds; mpv @@ -828,7 +897,7 @@ class SessionCoordinator { _awaitLiveFor = r.mediaId; health.value = const HealthStatus(HealthSeverity.info, 'the server is getting the video…'); } else { - await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks)); + await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks, relay.queueId)); } } else { var url = stateMediaUrl(r); @@ -848,8 +917,9 @@ class SessionCoordinator { if (_mediaId == r.mediaId) settledMediaId.value = r.mediaId; } else if (_awaitLiveFor == r.mediaId && (stateMediaRelay(r)?.live ?? false)) { _awaitLiveFor = -1; + final relay = stateMediaRelay(r)!; await _openMedia( - _relayMediaUrlFor((r.media?['name'] as String?) ?? '', r.mediaId, stateMediaRelay(r)!.tracks)); + _relayMediaUrlFor((r.media?['name'] as String?) ?? '', r.mediaId, relay.tracks, relay.queueId)); } _jellyfinTransport(r); _relayStatus(r); @@ -1469,11 +1539,14 @@ class SessionCoordinator { final relayProblem = probe.relayProblem; if (relayProblem != null) return relayProblem; final y = await ytDlp.ytDlp; - final check = LinkRelayMedia(uri, ytDlp: y, probe: probe); + // Up next's item this app already feeds (its warm relay): read. + final warm = _warmSource; + final fed = queueId != null && warm != null && warm.queueId == queueId && warm.src.tracks.isNotEmpty; + final check = fed ? null : LinkRelayMedia(uri, ytDlp: y, probe: probe); var size = 0; final List specs; try { - final tracks = await check.open().timeout(const Duration(seconds: 15)); + final tracks = fed ? warm.src.tracks : await check!.open().timeout(const Duration(seconds: 15)); for (final t in tracks) { // A playlist is read (DRM and live refused) but has no size. final pieces = t.pieces; @@ -1487,7 +1560,7 @@ class SessionCoordinator { } catch (e) { return 'cannot pass this link on: $e'; } finally { - await check.close(); + await check?.close(); } _pendingOrigin = _LinkOrigin(url, probe, y); _swap(HelloMedia( @@ -1717,7 +1790,8 @@ class SessionCoordinator { // its own sign-in, else the pasted link (played directly here too). if (origin != null) { if (plan is OwnPlay) { - _startRelaySource(r.mediaId, RelayMedia.single(HttpRelayBytes(Uri.parse(plan.url), headers: plan.headers))); + _startRelaySource(r.mediaId, RelayMedia.single(HttpRelayBytes(Uri.parse(plan.url), headers: plan.headers)), + queueId: stateMediaRelay(r)?.queueId ?? ''); } else if (origin.link != null) { plan = RelayPlay(origin.link!); _startRelaySource(r.mediaId, RelayMedia.single(HttpRelayBytes(Uri.parse(origin.link!)))); @@ -2100,6 +2174,7 @@ class SessionCoordinator { reconnecting.value = false; await _stopJellyfin(); _stopRelaySource(); + stopWarmSource(); _pendingOrigin = null; _pendingArt = null; _activeOrigin = null; diff --git a/apps/desktop/test/queue_switch_e2e_test.dart b/apps/desktop/test/queue_switch_e2e_test.dart index eb49517..3460fec 100644 --- a/apps/desktop/test/queue_switch_e2e_test.dart +++ b/apps/desktop/test/queue_switch_e2e_test.dart @@ -18,7 +18,9 @@ import 'package:watch_together/src/session/switch_trace.dart'; // How long an Up next switch takes, end to end, over real mpv and a real // sync server: two Jellyfin account holders and a guest (through the // server's /j login) watch an episode, then ⏭ puts up the next episode, -// then a direct link. The fake Jellyfin and the link's host answer with a +// then a direct link, then a link passed on "through" alice, then one the +// server fetches (both relayed: the guests preload them from the server's +// warm relay of the item). The fake Jellyfin and the link's host answer with a // delay and a capped rate, like a server elsewhere. Prints each client's // phases (SwitchTrace) and the wall time from ⏭ to the server playing. // @@ -206,7 +208,7 @@ void main() { final link = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); link.listen((req) => _serveFile(req, file)); final linkUrl = 'http://127.0.0.1:${link.port}/clip.mkv'; - final server = SyncServer(AppConfig( + final server = SyncServer(ytDlp: YtDlp('$_tmp/queue_switch_server_ytdlp', enabled: false), AppConfig( port: 0, secret: 'q', name: 'q', @@ -219,7 +221,14 @@ void main() { deviceId: 'host-1', version: '9.9.9'), )); - await server.start(); + // Relic logs errors writing a response (a relay response cut short + // when its relay is dropped at a swap or stop) as it does outside + // tests; anything else fails the test. + final serverErrors = []; + await runZonedGuarded(server.start, (e, _) { + if ('$e'.contains('Content size below specified contentLength') || '$e'.contains('did not arrive')) return; + serverErrors.add(e); + }); final ytDlp = AppYtDlp(YtDlp('$_tmp/queue_switch_ytdlp', enabled: false)); final accA = await _account(jelly.base, 'ATOKEN', 'ua', 'A-PC'); final accC = await _account(jelly.base, 'CTOKEN', 'uc', 'C-PC'); @@ -297,8 +306,30 @@ void main() { } await switchOnce('direct link'); expect(stateMediaUrl(a.serverState.peek()!), linkUrl); + + // Relayed items: alice passes the link on, then the server fetches + // it. Each is fed into the server's warm relay while it waits, so + // bob and carol (and for the fetched one alice) preload from it, + // and the swap promotes that relay with its buffer. + for (final (what, mode) in [('link through alice', LinkRelay.me), ('server-fetched link', LinkRelay.server)]) { + expect(await a.queue.add(LinkMedia(linkUrl, relay: mode)), isNull); + await _poll(() => a.serverState.peek()?.queue.isNotEmpty ?? false, '$what queued'); + final id = a.serverState.peek()!.queue.first.id; + await _poll(() => a.serverState.peek()?.queue.firstOrNull?.media.relay?.live ?? false, '$what warm relay live'); + await switchOnce(what); + final relay = stateMediaRelay(a.serverState.peek()!); + expect(relay?.queueId, id, reason: 'the warm relay became the media\'s'); + expect(relay?.fetch, mode == LinkRelay.server); + if (preload) { + for (final x in all.values) { + expect(x.switchTrace.of(x.serverState.peek()!.mediaId), contains(SwitchTrace.preloaded), + reason: '$what preloaded'); + } + } + } // ignore: avoid_print print(report.join('\n')); + expect(serverErrors, isEmpty); // Nothing queued: no preload. Queued and taken off again: dropped. if (preload) { diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 226b4f3..8718147 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -595,6 +595,32 @@ How bytes move: - Cleanup: the buffer is memory only. Any swap disposes it (its URLs then 404, the source is closed with 4001 and stops), and so does server stop. A crash frees it with the process. +- Warm relay (`ServerHub.syncWarm`, run on every change; wire in + [PROTOCOL.md](PROTOCOL.md) "Relay"): a second, small `ActiveRelay` for + Up next's first item while it waits, keyed by its queue id, so viewers + preload it (see "Getting ready" under "Up next"). A server-fetched + link starts its fetch then; a link passed on or a file is fed by its + adder's app (a second channel, `/r/.../source?q=`), only + while someone else is in the server and not for the host's own file; + a Jellyfin item someone present cannot play (and the server's login + does not cover) by the first member with an account that dials in. + Its budget is 32 MiB ahead and 8 MiB behind its viewers + (`relayWarmAheadBytes`, split by track size with the usual 8 MiB + floor) plus the pins: about what a preload player reads, so the + server holds at most that much on top of the current relay (there is + no global relay budget; each relay is bounded by its own). Like any + buffer it pulls nothing while nobody preloads. The swap that puts the + item up (`swap.queueId`, from its source, or anyone for a fetch, with + the same tracks) promotes it (`ServerHub.takeWarm`): the same buffer, + feed and source channel become the media's relay with the full + budget, `media.relay.queueId` names the item and the media keeps the + item's name, so its `/r/...//...` URLs (and the preloads of + them) stay valid. Any other swap of it, the item no longer first (a + removal, a reorder, Play now of another), its adder leaving, or server + stop drops it (its URLs 404, its source gets 4001); a restart starts + with none. A manual swap to something else keeps it (the item is + still first). A failed warm fetch is not promoted: the swap fetches + again and reports the reason. - Source away: viewers keep playing from the buffer; a request waits up to 60 s for missing bytes. The source reconnects with backoff (same token after a resume); a different size (or piece count and roles) @@ -917,10 +943,17 @@ every `state` and feed), the pure rules in `queue_logic.dart`, the UI in "Jellyfin relay"). A link everyone opens is that link; the app that puts it up also probes it (`AppYtDlp.probe` with `keep`: kept 35 minutes instead of 10, a refresh replaces it only once it is in; swaps - read it from there). A link passed on "through me" and a file only on + read it from there). A link passed on "through me" and a file on their adder's app (the link, the path; not the host's own file, which - plays from `/m`). A server-fetched link and others' relayed items have - nothing to get ready (the relay starts with the swap; see ROADMAP). It + plays from `/m`), which also feeds the server's warm relay of it + (`QueueController._feedWarm`, `SessionCoordinator.feedWarm`; a + Jellyfin item's from the member who puts it up, with its resolved + stream, when the server asks for one); everyone else, and everyone + for a server-fetched link, preloads that warm relay's URL + (`SessionCoordinator.warmRelayUrl`) once `queue[0].media.relay.live` + says bytes can flow (see "Relay"). At the swap the adder's app skips + re-reading the link's sizes (its warm source has them) and that + source carries on as the media's. It is resolved as soon as it is first there, again every 30 minutes, and dropped when it stops being first. **Preload**: once resolved, the stream is opened in a second, hidden @@ -952,6 +985,9 @@ every `state` and feed), the pure rules in `queue_logic.dart`, the UI in account holders and a `/j` guest): ⏭ to the server playing 59 to 86 ms for the next episode and 35 ms for a direct link, against about 1.5 s cold; at 300 ms and 12 Mbit/s 52 and 103 ms against about 2.9 s. + Relayed, from the warm relay (same setup, loopback server): a link + passed on 21 to 22 ms, a server-fetched link 35 to 37 ms, against + about 1.6 s cold. `SwitchTrace` (env `SYNC_RELAY_TRACE_SWITCH=1` prints it) times each phase of a media change: state, resolved, open, player (a fresh player is up; see "Player lifecycle" under "Media"), opened or diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 68ab928..8597946 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -56,7 +56,8 @@ and `e2e_check` (simulated viewers against a live server). `seek_drag_e2e`, `end_of_file_e2e`, `readahead_e2e`, `jellyfin_e2e` (a local fake Jellyfin over HTTP; tests never talk to a real Jellyfin), `queue_switch_e2e` (times Up next switches against a delayed, rate-capped - fake Jellyfin and link host; `WT_SWITCH_PRELOAD=0` for the cold path, + fake Jellyfin and link host: an episode, a direct link, a link passed + on and a server-fetched one; `WT_SWITCH_PRELOAD=0` for the cold path, `WT_SWITCH_DELAY_MS` / `WT_SWITCH_RATE_KBS` for the link). The sync and end-of-file ones expect `~/Videos/chiyo.mp4`; others generate media with ffmpeg into `WT_TEST_TMP`. Headless mpv (no video output attached) won't @@ -77,6 +78,11 @@ Traps: the `flutter_tester` by PID). `AppShell(videoBuilder:)` takes a stand-in, 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`). - Never depend on a server-side WebSocket `onDone`; assert presence with deadline polling. - `sub.asFuture()` can resolve before `onDone`; poll with a deadline. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index ab695f0..34a1caf 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,4 +1,4 @@ -# Protocol (proto = 13) +# Protocol (proto = 14) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: @@ -31,7 +31,7 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). ## Client to server ``` -{"t":"hello","proto":13,"secret":"","user":{"name":"","avatar":""}, +{"t":"hello","proto":14,"secret":"","user":{"name":"","avatar":""}, "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"...", "resume":""} {"t":"ready","ready":true,"source":"manual"} source: manual | auto @@ -275,6 +275,9 @@ swap media.relay = {} the swapper's machine serves the bytes state media.relay = {"source":"m2","live":true} or {"fetch":true,"live":true} media.relay = {"fetch":true,"live":true,"tracks":[{"kind":"video"},{"kind":"audio"}]} media.relay = {"source":"m2","live":true,"rate":4100000,"need":2300000} + media.relay = {"source":"m2","live":true,"queueId":"q3"} started in Up next (proto 14) + queue[0].media.relay = {"source":"m2","live":true,"tracks":[...]} its warm relay + queue[0].media.relay = {"fetch":true,"live":true} ``` Tracks: a relayed media is one file, or tracks that viewers play as one @@ -313,37 +316,57 @@ anything else means one file. its segments in so far, after a minute of them). Both two significant digits, updated once a second. `rate` under `need` means the source cannot keep up; clients tell the source once per media. +- Warm relay (proto 14): while Up next's first item is relayed (`how` + `server`, `through` or `file`, or `jellyfin` when someone present has + no access to that server and `state.relayNext` is false), the server + holds a small relay for it and shows it as that item's + `media.relay`: `source` (the adder; for `jellyfin` empty until the + first member with `jf` true for that server dials in), or `fetch`, and + `live` with `tracks` once bytes can flow. Its URLs use the item's + queue id where a media id would be. Not for the host's own file, nor + for an adder alone in the server. A swap with `queueId` of that item, + from its source (or, `fetch`, from anyone) with the same `tracks`, + makes it the media's relay: `media.relay.queueId` is the item's id, + `media.name` the item's `media.name`, the buffer and source carry on + and the queue-id URLs keep working. Any other swap of the item, or the + item no longer first, drops it (its URLs 404, its source is closed + with 4001). - `welcome.mediaUrl` for relayed media without `jellyfin`: this member's `/r/` URL; absent for the source itself (it plays its own copy) and for a fetched link the server has not read yet (clients open it on `live`). Clients build it after a swap like the `/m` one: - `http:///r/default///` (name with `/` and - `\` made `_`, then URL-encoded; the server ignores it). With `tracks`, - each track's URL is `http:///r/default////` + `http:///r/default///` (`` the + `mediaId`, or `media.relay.queueId` when set; name with `/` and `\` + made `_`, then URL-encoded; the server ignores it). With `tracks`, + each track's URL is `http:///r/default////` (a pieces track: `.m3u8`); one track is opened as is, several as an mpv EDL of them all, each framed by its byte length: `edl://!no_clip;%%;!new_stream;!no_clip;%%`. -- `GET /r////`: the media (one file), Range - table as for `/m`. `GET /r/////`: track +- `GET /r////`: the media (one file), Range + table as for `/m`; `` a `mediaId`, or a queue id (the warm relay, + or the media's relay started from it). `GET /r/////`: track `i` (from 0) of a media with `tracks` (track 0 of a one-file media too); for a pieces track (any name), its media playlist (`application/vnd.apple.mpegurl`): `#EXT-X-PLAYLIST-TYPE:VOD`, `EXT-X-MEDIA-SEQUENCE`, `EXTINF` durations, `EXT-X-DISCONTINUITY`, `EXT-X-MAP` and `EXT-X-KEY:METHOD=AES-128` (with `IV` when given) as upstream, `EXT-X-ENDLIST`; every URI is `p/`. `GET - /r/////p/`: piece `j` of pieces track `i`, + /r/////p/`: piece `j` of pieces track `i`, whole, once it is in (a Range inside the piece answers 206). Errors: 401 - bad token, 404 not the current relayed media (a player still on an - older `mediaId` gets 404, never the new bytes), no such track (the + bad token, 404 not the current relayed media nor the warm one (a player + still on an older `mediaId` gets 404, never the new bytes), no such track (the 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`. - `GET /r///source` (WebSocket): the source's relay channel, - dialled by the source (works behind NAT). Only the current source's - token is accepted; anyone else, or the source after the next swap, is - closed with code 4001 (the source stops instead of reconnecting). + 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 + relay's source token is accepted; anyone else, or the source once its + relay is dropped, is closed with code 4001 (the source stops instead + of reconnecting). A warm relay's tracks are those of its source's + first `info`. Frames: ``` @@ -442,4 +465,5 @@ state "queue":[,...], "queuePlay":"q3" leave out `url`: the server uses the one it kept. A swap without `queueId` leaves the queue as it is. - `state.queue` (omitted when empty) and `state.queuePlay` (omitted when - none) ride in every state, `welcome.state` included. + none) ride in every state, `welcome.state` included. The first item's + `media.relay` shows its warm relay (see "Relay"). diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index d6624a0..a9b643f 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -56,25 +56,19 @@ off by deleting it in the same commit as the code. Wire changes also edit over instead of the buffer running dry. - "Bob can't open this link — relay it?": offer the relay to whoever put a direct link up when a viewer's player fails to open it. - - Preload relayed Up next items (see ARCHITECTURE "Getting ready"): - today a link passed on "through me", another member's file, a - server-fetched link and the host's own file are opened cold by - everyone but their source when they go up (their relay or `/m` - starts with the swap), so the gate waits for the slowest open. Plan - (wire change): the server holds a second, small relay for Up next's - first item, keyed by its queue id. Its adder's app dials a second - source for it as soon as it is first (`/r/.../source` naming the - queue id); a server-fetched item starts its fetch then; - `state.queue[0].relay` (`live`, `tracks`) tells viewers when and - how to open `/r///q//` in their preload - player. The buffer keeps only the head and tail pins plus about - 32 MiB ahead of the viewers until the swap, pulls nothing while - nobody preloads, and is dropped when the item stops being first. A - swap with `queueId` turns it into the media's relay (same buffer; - the `q/` URL keeps working for that media, so the preloads match - and are promoted). The host's file: the hosting app hands the - server the first item's path early and `/m` serves it under a - `next/` name until the swap. + - The host's own file as Up next's first item: it plays from `/m` + once up, so others open it cold (the warm relay covers every other + relayed item, see ARCHITECTURE "Relay"). The hosting app could hand + the server the item's path early and `/m` serve it under a `next/` + name until the swap. + - Research (before building anything): can stream URLs a member's + app resolved with yt-dlp (its cookies, its address) be fetched by + the server from another address? Per site: YouTube (its + `googlevideo` URLs carry the resolving IP), Bilibili, X, generic + HLS/DASH, others; and YouTube resolved by the server itself from a + VPS, with and without cookies. Decides whether a server-driven + fetch of member-resolved links (the member need not upload or stay) + is worth building. 8. **Jellyfin** (epic, below). @@ -98,9 +92,7 @@ off by deleting it in the same commit as the code. Wire changes also edit - HTTP: move the app's clients (Jellyfin, AniList, AnimeFlow, updater) to dio from `fox_utils` when touching them, for retries/interceptors. - TLS: document a reverse proxy in front of the server. - - Up next follow-ups (see ARCHITECTURE "Up next"): getting the first - item ready only resolves its link; fetching its first seconds early - through the relay would make relayed starts instant too. A member + - Up next follow-ups (see ARCHITECTURE "Up next"): a member whose reconnect lands as a new member loses its "through me" links and files in the queue (skipped as gone); the key could follow the member. A whole season or a pick list at once ("Add the rest of the diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 8efe4d5..e000a85 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -1,5 +1,5 @@ abstract class RelayConstants { - static const int protoVersion = 13; + static const int protoVersion = 14; // Relay (`/r/...`): a member's bytes through the server's buffer. static const int relaySliceBytes = 1 << 20; @@ -24,6 +24,12 @@ 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; + /// 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 + /// whole media. + static const int relayWarmAheadBytes = 32 << 20; + static const int relayWarmBackBytes = 8 << 20; static const int seekDebounceMs = 300; static const double driftSeekThresholdSecs = 5.0; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index 272d129..3aacc7b 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -182,7 +182,10 @@ class RelayTrackSpec { /// them, the server fills them in for [fetch] once it has read the link. /// [rate] and [need] are the server's measures, 0 while unknown: bytes per /// second the source (or the server's own fetch) delivers while the -/// buffer asks for bytes, and the media's average bitrate. +/// buffer asks for bytes, and the media's average bitrate. [queueId]: +/// the Up next item this relay was started for (its warm relay) before +/// it went up; viewers open its `/r/...//...` URLs, so their +/// preloads of it carry on. class MediaRelay { final String source; final bool fetch; @@ -190,8 +193,15 @@ class MediaRelay { final List tracks; final int rate; final int need; + final String queueId; const MediaRelay( - {this.source = '', this.fetch = false, this.live = false, this.tracks = const [], this.rate = 0, this.need = 0}); + {this.source = '', + this.fetch = false, + this.live = false, + this.tracks = const [], + this.rate = 0, + this.need = 0, + this.queueId = ''}); /// The source is measured slower than the media plays (both known): /// below 85 % of [need], so a rate close to it does not flicker. @@ -210,7 +220,8 @@ class MediaRelay { live: live ?? this.live, tracks: tracks ?? this.tracks, rate: rate ?? this.rate, - need: need ?? this.need); + need: need ?? this.need, + queueId: queueId); Map toJson() => { if (source.isNotEmpty) 'source': source, @@ -219,6 +230,7 @@ class MediaRelay { if (validTracks(tracks)) 'tracks': [for (final t in tracks) t.toJson()], if (rate > 0) 'rate': rate, if (need > 0) 'need': need, + if (queueId.isNotEmpty) 'queueId': queueId, }; /// Unknown kinds, a single byte track or too many make it one file. @@ -232,7 +244,8 @@ class MediaRelay { live: json['live'] == true, tracks: ok ? List.unmodifiable(specs.cast()) : const [], rate: _byteRate(json['rate']), - need: _byteRate(json['need'])); + need: _byteRate(json['need']), + queueId: json['queueId'] is String ? json['queueId'] as String : ''); } /// A bytes-per-second field: a finite non-negative number, else 0. diff --git a/packages/core/lib/src/relay.dart b/packages/core/lib/src/relay.dart index 3f2f2da..8e7105e 100644 --- a/packages/core/lib/src/relay.dart +++ b/packages/core/lib/src/relay.dart @@ -348,6 +348,9 @@ class RelaySource { var _tracks = const []; + /// The media's tracks once [start] has opened it (empty before). + List get tracks => _tracks; + /// Opens the media (a page is resolved) and starts serving; throws /// when it cannot be read. Future start() async { diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index 9654394..f49cbab 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -535,7 +535,9 @@ class SyncEngine { /// [relay]: whether the server can relay the current Jellyfin media; the /// engine does not know the server's credential. - StateBroadcast broadcast({bool relay = false, bool relayNext = false}) { + /// [warm]: the server's warm relay for Up next's first item, shown as + /// that item's `media.relay`. + StateBroadcast broadcast({bool relay = false, bool relayNext = false, ({String id, MediaRelay relay})? warm}) { final bySeek = _seekEventId == eventId; return StateBroadcast( eventId: eventId, @@ -555,7 +557,10 @@ class SyncEngine { seekSeq: bySeek ? _seekSeq : null, relay: relay, relayNext: relayNext, - queue: queue.items, + queue: [ + for (final q in queue.items) + warm != null && q.id == warm.id ? q.copyWith(media: q.media.copyWith(relay: warm.relay)) : q, + ], queuePlay: queue.play, ); } diff --git a/packages/server/lib/src/relay_routes.dart b/packages/server/lib/src/relay_routes.dart index 962ab73..1554668 100644 --- a/packages/server/lib/src/relay_routes.dart +++ b/packages/server/lib/src/relay_routes.dart @@ -32,22 +32,39 @@ List<(int, int)> relayBudgets(List sizes, /// video, 160 kbit/s for audio. double _piecesRate(String kind) => kind == 'audio' ? 20000 : 500000; -/// One track as the source (or the fetcher) described it: a byte size, -/// or a piece list. -typedef _TrackInfo = ({int? size, RelayPieceList? pieces}); +/// One track as the source (or the fetcher) described it: its kind, a +/// byte size or a piece list. +typedef _TrackInfo = ({String kind, int? size, RelayPieceList? pieces}); -/// The relay of the current media: one buffer per track and where the -/// bytes come from. One per relayed swap; [dispose] on the next swap or -/// server stop (the buffers are memory only, so that is the whole cleanup). +/// A relayed media on the server: one buffer per track and where the +/// bytes come from. The relay of the current media (one per relayed swap; +/// [dispose] on the next swap or server stop: the buffers are memory +/// only, so that is the whole cleanup), or the warm relay of Up next's +/// first item ([warm], keyed by [queueId]), which a swap of that item +/// [promote]s to the current one, buffer and all. class ActiveRelay { - final int mediaId; + /// The media it relays; -1 while it is warm. + int mediaId; final String name; /// Token of the member whose machine serves the bytes; null when the - /// server fetches the link itself. - final String? sourceToken; + /// server fetches the link itself, or (a warm Jellyfin item) until a + /// member with an account dials in. + String? sourceToken; + /// That member's id (kept after it leaves); empty with [fetch]. + String sourceId; + /// The server reads the link itself. + final bool fetch; + /// The Up next item it was started for (the warm relay); kept once + /// promoted, so the `/r/...//...` URLs keep working. + final String? queueId; + bool _warm; final void Function() onLiveChanged; + /// The server's own fetch could not read the link (with the reason). + void Function(String reason)? onFail; + bool failed = false; /// One per track, in track order ([RelayBuffer] for bytes, [PieceBuffer] - /// for pieces); empty until the server has read a fetched link. + /// for pieces); empty until the tracks are known (a fetched link read, + /// a warm source's first `info`). List buffers = const []; /// `media.relay.tracks`: empty for one file. List tracks = const []; @@ -57,11 +74,17 @@ class ActiveRelay { var _local = const []; RelicWebSocket? _channel; var _channelFeeds = const []; + List<_TrackInfo>? _infos; /// Shared by the tracks: time any of them waits on the source. final clock = FeedClock(); final _rate = SourceRate(); - ActiveRelay._(this.mediaId, this.name, this.sourceToken, this.onLiveChanged); + ActiveRelay._(this.mediaId, this.name, this.sourceToken, this.onLiveChanged, + {this.sourceId = '', this.fetch = false, this.queueId, bool warm = false}) + : _warm = warm; + + /// Up next's first item, before it goes up. + bool get warm => _warm; /// Takes the bytes the source sent so far, for [rate]; the server calls /// it once a second. @@ -71,6 +94,9 @@ class ActiveRelay { /// (`media.relay.rate`); 0 until measured. int get rate => _rate.rate; + /// Bytes the buffers hold now. + int get cachedBytes => buffers.fold(0, (a, b) => a + b.cachedBytes); + /// The media's average bytes per second for a [duration] in seconds /// (`media.relay.need`): byte tracks by size, pieces tracks from the /// pieces in so far; 0 while any track is unknown. @@ -106,6 +132,12 @@ class ActiveRelay { if (!_known.isCompleted) _known.complete(); } + /// The tracks a warm source described in its first `info`. + static List _specsOf(List<_TrackInfo> infos) => + infos.length == 1 && infos.single.pieces == null + ? const [] + : [for (final t in infos) RelayTrackSpec(t.kind.isEmpty ? 'video' : t.kind, pieces: t.pieces != null)]; + /// The info of every track matches [tracks] (bytes for a byte track, /// pieces for a pieces one). bool _matches(List<_TrackInfo> infos) { @@ -118,10 +150,8 @@ class ActiveRelay { /// Attaches every buffer to its feed. void _attachAll(List feeds, List<_TrackInfo> infos) { - _budget([ - for (var i = 0; i < infos.length; i++) - infos[i].size ?? (_piecesRate(tracks[i].kind) * infos[i].pieces!.duration).round(), - ]); + _infos = infos; + _budget(); for (var i = 0; i < buffers.length; i++) { final b = buffers[i]; if (b is PieceBuffer) { @@ -132,8 +162,18 @@ class ActiveRelay { } } - void _budget(List sizes) { - final b = relayBudgets(sizes); + /// The tracks share the budget by size: a warm relay's small one + /// ([RelayConstants.relayWarmAheadBytes]), else the full one. + void _budget() { + final infos = _infos; + if (infos == null) return; + final sizes = [ + for (var i = 0; i < infos.length; i++) + infos[i].size ?? (_piecesRate(infos[i].kind) * infos[i].pieces!.duration).round(), + ]; + final b = _warm + ? relayBudgets(sizes, ahead: RelayConstants.relayWarmAheadBytes, back: RelayConstants.relayWarmBackBytes) + : relayBudgets(sizes); for (var i = 0; i < buffers.length; i++) { buffers[i] ..ahead = b[i].$1 @@ -144,22 +184,33 @@ class ActiveRelay { /// Bytes from member [sourceToken] over its relay channel, as the /// [tracks] its swap named (empty: one file). factory ActiveRelay.fromMember(int mediaId, String name, String sourceToken, - {List tracks = const [], required void Function() onLiveChanged}) => - ActiveRelay._(mediaId, name, sourceToken, onLiveChanged).._setTracks(tracks); + {List tracks = const [], String sourceId = '', required void Function() onLiveChanged}) => + ActiveRelay._(mediaId, name, sourceToken, onLiveChanged, sourceId: sourceId).._setTracks(tracks); + + /// The warm relay of Up next's item [queueId] fed by member + /// [sourceToken] (null: the first qualified member to dial in); its + /// tracks are what the source's first `info` says. + factory ActiveRelay.warmFromMember(String queueId, String name, String? sourceToken, + {String sourceId = '', required void Function() onLiveChanged}) => + ActiveRelay._(-1, name, sourceToken, onLiveChanged, sourceId: sourceId, queueId: queueId, warm: true); /// The server reads [url] itself: a media file as is, a web page /// through [ytDlp] (one file, video + audio tracks, or segmented - /// playlists). [onFail] gets the reason when it cannot. - static ActiveRelay fetch(int mediaId, String name, Uri url, - {YtDlp? ytDlp, required void Function() onLiveChanged, void Function(String reason)? onFail}) { - final r = ActiveRelay._(mediaId, name, null, onLiveChanged); + /// playlists). [onFail] gets the reason when it cannot. With [queueId] + /// it is that Up next item's warm relay. + static ActiveRelay fetchLink(int mediaId, String name, Uri url, + {YtDlp? ytDlp, String? queueId, required void Function() onLiveChanged, void Function(String reason)? onFail}) { + final r = ActiveRelay._(mediaId, name, null, onLiveChanged, fetch: true, queueId: queueId, warm: queueId != null) + ..onFail = onFail; final media = r._media = LinkRelayMedia(url, ytDlp: ytDlp); unawaited(() async { try { final tracks = await media.open(); final infos = <_TrackInfo>[ for (final t in tracks) - t.pieces != null ? (size: null, pieces: await t.pieces!.list()) : (size: await t.bytes!.size(), pieces: null), + t.pieces != null + ? (kind: t.kind, size: null, pieces: await t.pieces!.list()) + : (kind: t.kind, size: await t.bytes!.size(), pieces: null), ]; if (r._disposed) return; r._setTracks(relayTrackSpecs(tracks)); @@ -168,17 +219,34 @@ class ActiveRelay { onLiveChanged(); } catch (e) { _log.warning('relay fetch: cannot read the link: $e'); - if (!r._disposed) onFail?.call('$e'); + if (!r._disposed) { + r.failed = true; + r.onFail?.call('$e'); + } } }()); return r; } + /// A member relay whose source has not described its tracks yet takes + /// [tracks] from the swap that puts it up. + void ensureTracks(List tracks) { + if (!fetch && !_known.isCompleted) _setTracks(tracks); + } + + /// The warm relay becomes media [mediaId]'s: its buffer and feed carry + /// on, with the full budget. + void promote(int mediaId) { + this.mediaId = mediaId; + _warm = false; + _budget(); + } + var _disposed = false; bool get live => buffers.isNotEmpty && buffers.every((b) => b.live); /// The tracks are known (a fetched link has been read). - bool get known => _known.isCompleted && !_disposed; + bool get known => _known.isCompleted && !_disposed && buffers.isNotEmpty; /// Waits until the tracks are known, up to [RelayConstants.relayWaitSecs]. Future whenKnown() async { @@ -214,6 +282,13 @@ class ActiveRelay { if (ch != null) unawaited(_closeNotSource(ch)); } + /// The source dialled in with its tracks: a warm relay learns them + /// here; false when they are not this relay's. + bool _channelInfo(List<_TrackInfo> infos) { + if (!_known.isCompleted) _setTracks(_specsOf(infos)); + return _matches(infos); + } + void _attachChannel(RelicWebSocket ws, List<_TrackInfo> infos) { final old = _channel; if (old != null && !identical(old, ws)) unawaited(old.tryClose()); @@ -243,19 +318,21 @@ class ActiveRelay { /// malformed. List<_TrackInfo>? _infoTracks(Map j) { final size = j['size'], tracks = j['tracks']; - if (size is int) return size > 0 ? [(size: size, pieces: null)] : null; + if (size is int) return size > 0 ? [(kind: '', size: size, pieces: null)] : null; if (tracks is! List || tracks.isEmpty || tracks.length > MediaRelay.maxTracks) return null; final out = <_TrackInfo>[]; for (final t in tracks) { if (t is! Map) return null; + final kind = t['kind']; + if (!MediaRelay.trackKinds.contains(kind)) return null; if (t['as'] == 'pieces') { final list = RelayPieceList.fromJson(t); if (list == null) return null; - out.add((size: null, pieces: list)); + out.add((kind: kind as String, size: null, pieces: list)); } else { final n = t['size']; if (n is! int || n <= 0) return null; - out.add((size: n, pieces: null)); + out.add((kind: kind as String, size: n, pieces: null)); } } return out; @@ -270,32 +347,32 @@ Future _closeNotSource(RelicWebSocket ws) async { } /// `GET /r///source` (WebSocket): the source member's relay -/// channel. `GET /r////`: a viewer reads the -/// relayed file, Range as for `/m`; `...///` one -/// track of a media with tracks (a pieces track's media playlist), and -/// `...///p/` piece `i` of a pieces track. [relay] -/// gives the current relay. +/// channel; `?q=` for the warm relay of that Up next item. +/// `GET /r////`: a viewer reads the relayed file, +/// Range as for `/m`; `...///` one track of a media +/// with tracks (a pieces track's media playlist), and +/// `...///p/` piece `i` of a pieces track. `` is the +/// media id, or the queue id of a warm (or promoted) relay. [relayAt] +/// finds the relay of a key, [sourceOf] the relay a token may feed. void mountRelayRoutes( RelicApp app, - SessionMap sessions, - ActiveRelay? Function() relay, { + SessionMap sessions, { + required ActiveRelay? Function(String key) relayAt, + required ActiveRelay? Function(String token, String? queueId) sourceOf, void Function(String token, MediaLog log)? onLog, }) { app.get('/r/:sid/:token/source', (final req) { final sid = req.pathParameters.raw[#sid] ?? ''; final token = req.pathParameters.raw[#token] ?? ''; + final queueId = req.url.queryParameters['q']; final session = sessions.get(sid); - final r = relay(); - final ok = session != null && - session.gate(token) == MediaGate.ok && - r != null && - r.sourceToken == token; + final found = session != null && session.gate(token) == MediaGate.ok ? sourceOf(token, queueId) : null; return WebSocketUpgrade((final ws) async { - if (!ok) { + final r = found; + if (r == null) { await _closeNotSource(ws); return; } - final r = relay()!; ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs); var attached = false; try { @@ -313,7 +390,7 @@ void mountRelayRoutes( case 'info': if (attached) continue; final infos = _infoTracks(j); - if (infos == null || !r._matches(infos)) { + if (infos == null || !r._channelInfo(infos)) { // Not the media this swap named: the source stops. _log.warning('relay channel: info does not match the media (${r.name})'); await _closeNotSource(ws); @@ -359,7 +436,7 @@ void mountRelayRoutes( final stopwatch = Stopwatch()..start(); final sid = req.pathParameters.raw[#sid] ?? ''; final token = req.pathParameters.raw[#token] ?? ''; - final mediaId = int.tryParse(req.pathParameters.raw[#mediaId] ?? ''); + final key = req.pathParameters.raw[#mediaId] ?? ''; final track = tracked ? int.tryParse(req.pathParameters.raw[#seg] ?? '') ?? -1 : null; final session = sessions.get(sid); if (session == null) return _plain(410); @@ -369,9 +446,9 @@ void mountRelayRoutes( default: return _plain(401); } - final r = relay(); // A player still on an older media must not get the new one's bytes. - if (r == null || r.mediaId != mediaId) return _plain(404); + final r = relayAt(key); + if (r == null) return _plain(404); final notConnected = Response(503, body: Body.fromString('relay source not connected'), headers: Headers.build((h) => h['retry-after'] = ['2'])); @@ -421,7 +498,7 @@ void mountRelayRoutes( token, MediaLog( method: req.method.name, - path: '/r/$sid/$shortToken/$mediaId/${track == null ? '' : '$track/'}${r.name}', + path: '/r/$sid/$shortToken/$key/${track == null ? '' : '$track/'}${r.name}', rangeHeader: rangeHeader, status: res.status, bytesWritten: end - res.start + 1, diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 1eb437c..11a9e20 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -61,7 +61,8 @@ class ServerHub { : '$head: $why'; } - StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable, relayNext: relayNextAvailable); + StateBroadcast stateBroadcast() => + engine.broadcast(relay: relayAvailable, relayNext: relayNextAvailable, warm: _warmView()); /// The relay of the current media (`/r/...`), if it is relayed. ActiveRelay? relay; @@ -69,19 +70,178 @@ class ServerHub { /// Replaces the relay (null: the new media is not relayed). The old /// buffer is dropped with it. void setRelay(ActiveRelay? next) { - relay?.dispose(); + if (!identical(relay, next)) relay?.dispose(); relay = next; } + /// The warm relay: Up next's first item relayed before it goes up, so + /// every viewer preloads it (see ARCHITECTURE "Getting ready"); null + /// when that item is not relayed (or nobody would read it). + ActiveRelay? warm; + + void _setWarm(ActiveRelay? next) { + if (!identical(warm, next)) warm?.dispose(); + warm = next; + } + + /// Drops both relays (server stop). + void dropRelays() { + setRelay(null); + _setWarm(null); + } + + /// `queue[0].media.relay` while the warm relay exists. + ({String id, MediaRelay relay})? _warmView() { + final w = warm, id = w?.queueId; + if (w == null || id == null) return null; + return ( + id: id, + relay: MediaRelay( + source: w.fetch ? '' : w.sourceId, fetch: w.fetch, live: w.live, tracks: w.live ? w.tracks : const []), + ); + } + + /// Keeps [warm] on Up next's first item when it is relayed: a link the + /// server fetches (its fetch starts now), a link passed on or a file + /// (its adder's app feeds it, while someone else is here to read it; + /// not the host's own file), a + /// Jellyfin item that someone here cannot play and the server's login + /// does not cover (the first member with an account feeds it). Called + /// on every change; anything else drops it. + void syncWarm() { + final first = engine.queue.items.firstOrNull; + final w = warm; + if (w != null && (first == null || w.queueId != first.id || !_keepWarm(w, first))) _setWarm(null); + if (warm == null && first != null) _setWarm(_newWarm(first)); + } + + bool _othersHere(String? memberId) => engine.members.values.any((m) => m.id != memberId); + + /// A Jellyfin [ref] someone here cannot play while someone can, outside + /// the server's login. + bool _jellyfinNeedsWarm(JellyfinRef ref) => + !_onLogin(ref) && + engine.members.values.any((m) => !m.out && m.jf[ref.serverId] != true) && + engine.members.values.any((m) => m.jf[ref.serverId] == true); + + bool _keepWarm(ActiveRelay w, QueueItem first) { + switch (first.how) { + case QueueHow.server: + return w.fetch; + case QueueHow.through: + case QueueHow.file: + final adder = engine.byId(first.by); + return adder != null && adder.token == w.sourceToken && _othersHere(adder.id); + case QueueHow.jellyfin: + final ref = first.media.jellyfin; + final src = w.sourceToken; + return ref != null && _jellyfinNeedsWarm(ref) && (src == null || engine.members.containsKey(src)); + case QueueHow.direct: + return false; + } + } + + ActiveRelay? _newWarm(QueueItem first) { + final id = first.id; + final name = sanitizeMediaName(first.media.name.isEmpty ? first.title : first.media.name); + late final ActiveRelay r; + switch (first.how) { + case QueueHow.server: + final url = Uri.tryParse(engine.queue.privateUrlOf(id) ?? ''); + if (url == null || session.secret.isEmpty) return null; + r = ActiveRelay.fetchLink(-1, name, url, ytDlp: ytDlp, queueId: id, onLiveChanged: () => relayLiveChanged(r)); + case QueueHow.through: + case QueueHow.file: + final adder = engine.byId(first.by); + // The host's own file plays from its `/m` URL, never relayed. + if (adder == null || !_othersHere(adder.id) || (adder.host && first.how == QueueHow.file)) return null; + r = ActiveRelay.warmFromMember(id, name, adder.token, + sourceId: adder.id, onLiveChanged: () => relayLiveChanged(r)); + case QueueHow.jellyfin: + final ref = first.media.jellyfin; + if (ref == null || !_jellyfinNeedsWarm(ref)) return null; + r = ActiveRelay.warmFromMember(id, name, null, onLiveChanged: () => relayLiveChanged(r)); + case QueueHow.direct: + return null; + } + return r; + } + + /// The relay `/r/...//...` reads: the current one by its media id, + /// or by the queue id it was started for, or the warm one. + ActiveRelay? relayAt(String key) { + final r = relay, w = warm; + final mediaId = int.tryParse(key); + if (mediaId != null) return r != null && r.mediaId == mediaId ? r : null; + if (key.isEmpty) return null; + if (w != null && w.queueId == key) return w; + if (r != null && r.queueId == key) return r; + return null; + } + + /// The relay member [token] may feed on its channel: the current one + /// ([queueId] null), or the one of Up next item [queueId] (warm or + /// already promoted). A warm Jellyfin item takes the first member with + /// an account on that server that dials in. + ActiveRelay? sourceOf(String token, String? queueId) { + if (queueId == null || queueId.isEmpty) { + final r = relay; + return r != null && r.sourceToken == token ? r : null; + } + final r = relayAt(queueId); + if (r == null || r.fetch) return null; + if (r.sourceToken == null && identical(r, warm)) { + final ref = engine.queue.item(queueId)?.media.jellyfin; + final m = engine.members[token]; + if (ref == null || m == null || m.jf[ref.serverId] != true) return null; + r + ..sourceToken = token + ..sourceId = m.id; + broadcastState(); + } + return r.sourceToken == token ? r : null; + } + + /// A swap of Up next item [queueId] by [token]: its warm relay, when it + /// can become the media's (the same kind of relay; a member's from its + /// own source with the same tracks), else null. Taken either way: the + /// item goes up now. + ActiveRelay? takeWarm(String? queueId, String token, {required bool fetch, List tracks = const []}) { + final w = warm; + if (queueId == null || w == null || w.queueId != queueId) return null; + warm = null; + final ok = fetch + ? w.fetch && !w.failed + : !w.fetch && w.sourceToken == token && (!w.known || _sameTracks(w.tracks, tracks)); + if (ok) { + w.ensureTracks(tracks); + return w; + } + w.dispose(); + return null; + } + + static bool _sameTracks(List a, List b) => + a.length == b.length && [for (var i = 0; i < a.length; i++) a[i] == b[i]].every((x) => x); + /// `media.relay` for a relayed swap by [token]: who serves the bytes /// and, from a member, its tracks (the server finds a fetched link's). MediaRelay relayFor(String token, MediaRelay requested) => requested.fetch ? const MediaRelay(fetch: true) : MediaRelay(source: engine.members[token]?.id ?? '', tracks: requested.tracks); + /// `media.relay` of a promoted warm relay [w] (Up next item [queueId]). + MediaRelay promotedRelay(ActiveRelay w, String queueId) => w.fetch + ? MediaRelay(fetch: true, live: w.live, tracks: w.live ? w.tracks : const [], queueId: queueId) + : MediaRelay(source: w.sourceId, live: w.live, tracks: w.tracks, queueId: queueId); + /// The relay's channel came up or went down, or a fetched link was read: - /// `media.relay.live` and `tracks` follow. + /// `media.relay.live` and `tracks` follow (the warm one's in `queue`). void relayLiveChanged(ActiveRelay r) { + if (identical(warm, r)) { + broadcastState(); + return; + } final m = engine.media; if (!identical(relay, r) || m?.relay == null) return; final was = m!.relay!; @@ -192,6 +352,7 @@ class ServerHub { } void changed() { + syncWarm(); broadcastState(); broadcastFeed(); } @@ -301,7 +462,8 @@ void mountSyncRoute( String Function(String sid, String token, SessionMedia? media, String host) urlBuilder, { String Function(String sid, String token, String itemId, String host)? relayUrlBuilder, - String Function(String sid, String token, int mediaId, String name, List tracks, String host)? + /// `key`: the media id, or the queue id of a relay started in Up next. + String Function(String sid, String token, String key, String name, List tracks, String host)? relayMediaUrlBuilder, }) { app.get('/sync', (final req) { @@ -374,15 +536,22 @@ void mountSyncRoute( // Relayed for viewers without access: from the swapper's // machine (its own sign-in never leaves it). final relayReq = parsed.relay == null || parsed.relay!.fetch ? null : parsed.relay; + // Up next's warm relay of this item carries on as its relay. + final promoted = relayReq == null || ref == null ? null : hub.takeWarm(queueId, token, fetch: false); final media = ref == null ? null : HelloMedia( - name: sanitizeMediaName(parsed.name.isEmpty ? ref.title : parsed.name), + name: promoted?.name ?? sanitizeMediaName(parsed.name.isEmpty ? ref.title : parsed.name), size: parsed.size, duration: parsed.duration, jellyfin: ref, - relay: relayReq == null ? null : hub.relayFor(token, const MediaRelay())); + relay: relayReq == null + ? null + : promoted != null + ? hub.promotedRelay(promoted, queueId!) + : hub.relayFor(token, const MediaRelay())); if (media == null || !hub.engine.applySwap(token, media, queueId: queueId)) { + promoted?.dispose(); wsErr(ws, 'bad swap media'); return; } @@ -392,7 +561,11 @@ void mountSyncRoute( duration: media.duration, jellyfin: ref, mediaId: hub.engine.mediaId); - hub.setRelay(relayReq == null ? null : _memberRelay(hub, token, media.name, const [])); + hub.setRelay(relayReq == null + ? null + : promoted != null + ? (promoted..promote(hub.engine.mediaId)) + : _memberRelay(hub, token, media.name, const [])); hub.changed(); return; } @@ -415,27 +588,38 @@ void mountSyncRoute( wsErr(ws, 'server fetch needs a server password'); return; } + // Up next's warm relay of this item carries on as its relay + // (its buffer, its feed, its `/r/...//` URLs). + final promoted = hub.takeWarm(queueId, token, fetch: relayReq.fetch, tracks: relayReq.tracks); + final shown = promoted?.name ?? name; final media = HelloMedia( - name: name, + name: shown, size: parsed.size, duration: parsed.duration, - relay: hub.relayFor(token, relayReq)); + relay: promoted != null ? hub.promotedRelay(promoted, queueId!) : hub.relayFor(token, relayReq)); if (!hub.engine.applySwap(token, media, queueId: queueId)) { + promoted?.dispose(); wsErr(ws, 'bad swap media'); return; } session.media = SessionMedia( - name: name, size: parsed.size, duration: parsed.duration, mediaId: hub.engine.mediaId); + name: shown, size: parsed.size, duration: parsed.duration, mediaId: hub.engine.mediaId); + void failed(String why) { + wsErr(ws, hub.fetchFailText(why)); + hub.fetchFailed(shown); + } late final ActiveRelay active; - active = relayReq.fetch - ? ActiveRelay.fetch(hub.engine.mediaId, name, link!, - ytDlp: hub.ytDlp, - onLiveChanged: () => hub.relayLiveChanged(active), - onFail: (why) { - wsErr(ws, hub.fetchFailText(why)); - hub.fetchFailed(name); - }) - : _memberRelay(hub, token, name, media.relay!.tracks); + if (promoted != null) { + active = promoted..promote(hub.engine.mediaId); + if (active.fetch) { + active.onFail = failed; + } + } else { + active = relayReq.fetch + ? ActiveRelay.fetchLink(hub.engine.mediaId, name, link!, + ytDlp: hub.ytDlp, onLiveChanged: () => hub.relayLiveChanged(active), onFail: failed) + : _memberRelay(hub, token, name, media.relay!.tracks); + } hub.setRelay(active); hub.changed(); return; @@ -626,8 +810,8 @@ void mountSyncRoute( relayed.mediaId == media.mediaId && relayMediaUrlBuilder != null ? (relayed.known - ? relayMediaUrlBuilder( - 'default', activeToken, relayed.mediaId, relayed.name, relayed.tracks, dialledHost) + ? relayMediaUrlBuilder('default', activeToken, relayed.queueId ?? '${relayed.mediaId}', media.name, + relayed.tracks, dialledHost) : null) : media?.file != null ? urlBuilder('default', activeToken, media, dialledHost) @@ -656,6 +840,6 @@ void mountSyncRoute( ActiveRelay _memberRelay(ServerHub hub, String token, String name, List tracks) { late final ActiveRelay r; r = ActiveRelay.fromMember(hub.engine.mediaId, name, token, - tracks: tracks, onLiveChanged: () => hub.relayLiveChanged(r)); + tracks: tracks, sourceId: hub.engine.members[token]?.id ?? '', onLiveChanged: () => hub.relayLiveChanged(r)); return r; } diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index 61dfa56..72be38b 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -94,7 +94,7 @@ class SyncServer { ); }); mountSessionMediaRoute(_app, sessions); - mountRelayRoutes(_app, sessions, () => hub.relay); + mountRelayRoutes(_app, sessions, relayAt: hub.relayAt, sourceOf: hub.sourceOf); mountAvatarRoute(_app, hub); mountArtRoute(_app, hub); mountJellyfinRelayRoute( @@ -108,8 +108,8 @@ class SyncServer { return 'http://${authority(dialledHost)}/m/$sid/$token/$name'; }, relayUrlBuilder: (sid, token, itemId, dialledHost) => 'http://${authority(dialledHost)}/j/$sid/$token/${Uri.encodeComponent(itemId)}', - relayMediaUrlBuilder: (sid, token, mediaId, name, tracks, dialledHost) => - relayViewerUrl('http://${authority(dialledHost)}/r/$sid/$token/$mediaId', name, tracks: tracks)); + relayMediaUrlBuilder: (sid, token, key, name, tracks, dialledHost) => + relayViewerUrl('http://${authority(dialledHost)}/r/$sid/$token/$key', name, tracks: tracks)); _app.fallback = (final req) => Response.notFound( body: Body.fromString('not found'), ); @@ -181,7 +181,7 @@ class SyncServer { _heartbeat = null; _relayClient?.close(force: true); _relayClient = null; - hub.setRelay(null); + hub.dropRelays(); await hub.closeAllSockets(); await _app.close(); _server = null; diff --git a/packages/server/test/relay_test.dart b/packages/server/test/relay_test.dart index 63a174b..5163f01 100644 --- a/packages/server/test/relay_test.dart +++ b/packages/server/test/relay_test.dart @@ -92,9 +92,10 @@ void main() { return m; } - Future startSource(_Member m, {RelayMedia? from}) async { + Future startSource(_Member m, {RelayMedia? from, String? queueId}) async { final src = RelaySource( - Uri.parse('ws://127.0.0.1:${server.boundPort}/r/default/${m.welcome.token}/source'), + Uri.parse('ws://127.0.0.1:${server.boundPort}/r/default/${m.welcome.token}/source' + '${queueId == null ? '' : '?q=$queueId'}'), from ?? RelayMedia.single(FileRelayBytes(file)), backoff: (_) => const Duration(milliseconds: 100)); await src.start(); @@ -444,6 +445,158 @@ void main() { }); }); + group('warm relay (Up next)', () { + late HttpServer origin; + var originBytes = 0; + setUp(() async { + originBytes = 0; + origin = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + origin.listen((req) async { + final r = resolveMediaRequest(rangeHeader: req.headers.value('range'), size: bytes.length, filename: 'x.mkv'); + req.response.statusCode = r.status; + r.headers.forEach(req.response.headers.set); + if (r.status == 200 || r.status == 206) { + final end = r.end ?? bytes.length - 1; + originBytes += end - r.start + 1; + req.response.add(bytes.sublist(r.start, end + 1)); + } + await req.response.close(); + }); + }); + tearDown(() => origin.close(force: true)); + + List queueOf(_Member m) => [for (final q in (m.lastState?['queue'] as List?) ?? const []) q as Map]; + Map? headRelay(_Member m) => (queueOf(m).firstOrNull?['media'] as Map?)?['relay'] as Map?; + + Future add(_Member m, Map item, {bool next = false}) async { + final had = {for (final q in queueOf(m)) q['id']}; + m.send({'t': 'queue', 'op': 'add', 'at': next ? 'next' : 'end', 'item': item}); + await _until(() => queueOf(m).any((q) => !had.contains(q['id'])), 'queued'); + return queueOf(m).firstWhere((q) => !had.contains(q['id']))['id'] as String; + } + + Map through(String key) => { + 'media': {'name': 'ep 01.mkv', 'size': bytes.length, 'duration': 0}, + 'how': 'through', + 'title': 'Ep 1', + 'key': key, + }; + + String qUrl(_Member v, String queueId) => + 'http://127.0.0.1:${server.boundPort}/r/default/${v.welcome.token}/$queueId/${Uri.encodeComponent('ep 01.mkv')}'; + + test('a link the server fetches: preloaded from its queue id, promoted at the swap with its buffer', () async { + final alice = await join('alice'); + final bob = await join('bob'); + final id = await add(alice, { + 'media': {'name': 'ep 01.mkv', 'size': 0, 'duration': 0, 'url': 'http://127.0.0.1:${origin.port}/x.mkv'}, + 'how': 'server', + 'title': 'Ep 1', + }); + await _until(() => headRelay(bob)?['live'] == true, 'warm relay live'); + expect(headRelay(bob), {'fetch': true, 'live': true}); + expect(jsonEncode(bob.lastState), isNot(contains('x.mkv'))); + expect((await get(qUrl(bob, id))).$3, bytes); + // Small: the warm budget is far over this file, so all of it is in. + final fetched = originBytes; + alice.send({ + 't': 'swap', + 'queueId': id, + 'media': {'name': 'Ep 1', 'size': 0, 'duration': 0, 'relay': {'fetch': true}}, + }); + await _until(() => (bob.media?['relay'] as Map?)?['queueId'] == id, 'promoted'); + expect(bob.media!['relay'], {'fetch': true, 'live': true, 'queueId': id}); + expect(bob.media!['name'], 'ep 01.mkv', reason: 'the URL name the preload used'); + expect(queueOf(bob), isEmpty); + expect(server.hub.warm, isNull); + // The same buffer: both URLs serve it, nothing is fetched again. + expect((await get(qUrl(bob, id))).$3, bytes); + expect((await get(viewerUrl(bob, mediaIdOf(bob)))).$3, bytes); + expect(originBytes, fetched); + final carol = await join('carol'); + expect(carol.welcome.mediaUrl, qUrl(carol, id)); + }); + + test('a link passed on: its adder feeds it before the swap, and the swap keeps that source', () async { + final alice = await join('alice'); + final bob = await join('bob'); + final id = await add(alice, through('k-abcdefgh')); + expect(headRelay(bob), {'source': alice.welcome.memberId}); + final src = await startSource(alice, queueId: id); + await _until(() => headRelay(bob)?['live'] == true, 'warm relay live'); + expect((await get(qUrl(bob, id), range: 'bytes=0-99')).$3, bytes.sublist(0, 100)); + // Only its source may feed it. + final impostor = await startSource(bob, queueId: id); + await impostor.done.timeout(const Duration(seconds: 5)); + alice.send({ + 't': 'swap', + 'queueId': id, + 'media': {'name': 'ep 01.mkv', 'size': bytes.length, 'duration': 0, 'relay': {}}, + }); + await _until(() => (bob.media?['relay'] as Map?)?['queueId'] == id, 'promoted'); + expect(bob.media!['relay'], {'source': alice.welcome.memberId, 'live': true, 'queueId': id}); + expect((await get(qUrl(bob, id))).$3, bytes); + expect(src.bytesSent, lessThan(bytes.length * 1.3)); + // Still the source of the media it now relays. + await Future.delayed(const Duration(milliseconds: 200)); + expect(server.hub.relay!.live, isTrue); + }); + + test('dropped when the item stops being first: its URLs 404, its source stops', () async { + final alice = await join('alice'); + final bob = await join('bob'); + final id = await add(alice, through('k-abcdefgh')); + final src = await startSource(alice, queueId: id); + await _until(() => headRelay(bob)?['live'] == true, 'warm relay live'); + await add(alice, through('k-ijklmnop'), next: true); + await src.done.timeout(const Duration(seconds: 5)); + expect((await get(qUrl(bob, id))).$1, 404); + expect(server.hub.warm!.queueId, isNot(id)); + }); + + test('not for an adder alone; dropped when its adder leaves', () async { + final alice = await join('alice'); + await add(alice, through('k-abcdefgh')); + expect(headRelay(alice), isNull); + final bob = await join('bob'); + await _until(() => headRelay(bob) != null, 'warm relay once someone could read it'); + await alice.ws.close(); + await _until(() => bob.lastState!['members'] is List && (bob.lastState!['members'] as List).length == 1, 'left'); + expect(queueOf(bob), hasLength(1)); + expect(headRelay(bob), isNull); + expect(server.hub.warm, isNull); + }); + + test('a Jellyfin item: fed by the first member with an account, for those without', () async { + final alice = await join('alice'); + final bob = await join('bob'); + void report(_Member m, bool access) => m.send({ + 't': 'report', 'position': 0, 'buffered': 0, 'eventId': 0, 'jf': {'srv1': access}, // + }); + report(alice, true); + report(bob, false); + final id = await add(alice, { + 'media': { + 'name': 'Pilot', + 'size': 0, + 'duration': 0, + 'jellyfin': {'serverId': 'srv1', 'itemId': '0123456789abcdef0123456789abcdef', 'title': 'Pilot'}, + }, + 'how': 'jellyfin', + 'title': 'Pilot', + }); + await _until(() => headRelay(bob) != null, 'warm relay'); + expect(headRelay(bob), isEmpty, reason: 'nobody feeds it yet'); + final refused = await startSource(bob, queueId: id); + await refused.done.timeout(const Duration(seconds: 5)); + await startSource(alice, queueId: id); + await _until(() => headRelay(bob)?['live'] == true, 'live'); + expect(headRelay(bob)!['source'], alice.welcome.memberId); + report(bob, true); + await _until(() => headRelay(alice) == null, 'dropped once everyone has an account'); + }); + }); + group('buffer', () { // A feed that serves [bytes] in-process and counts what it sends. late RelayBuffer buf; diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index 27cf67c..24e65e5 100644 --- a/packages/server/test/session_test.dart +++ b/packages/server/test/session_test.dart @@ -64,8 +64,8 @@ void main() { }); group('core welcome + proto', () { - test('proto 12: avatars by hash, profile intent', () { - expect(RelayConstants.protoVersion, 13); + test('proto 14: avatars by hash, profile intent', () { + expect(RelayConstants.protoVersion, 14); const v = MemberView(id: 'm1', name: 'a', ready: false, inSet: false, buffered: 0, stale: false); expect(v.toJson().containsKey('avatarHash'), isFalse); expect(v.toJson().containsKey('avatar'), isFalse);