Protocol 14: a warm relay preloads relayed Up next items for everyone

The server keeps one small relay (32 MiB ahead, 8 behind) for Up next's
first relayed item, keyed by its queue id at a URL that stays the same
after the swap: fetched by the server for server-fetch links, fed by
the adder's app for through-me links and files, by a member with an
account for member-passed Jellyfin items. Every app preloads from it;
the swap promotes it with its buffer. Dropped when the item stops
being first or its adder leaves. Measured switch: ~1.6 s cold, 20-40 ms
preloaded. ROADMAP gains the research item on server-side fetching of
member-resolved URLs.

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

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 (`<queue id>|<token>`), 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);

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<RelayTrackSpec> 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<RelayTrackSpec> 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<RelayTrackSpec> 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;

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 = <Object>[];
+      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) {