Protocol 12: Up next (a shared queue) and where the video comes from

Up next: a server-side queue anyone edits by item id (add at end or
next, move, remove, play now, skip); each item says who added it and
how it plays. At the real end the first item swaps in after the end
card's countdown, put up by whoever can (its adder for through-me links
and files, someone with access for Jellyfin, else anyone); items that
can't play are skipped with the reason. The next item's link is
resolved ahead. Greyed "Continues with" episodes below. A chip with the
controls opens it as a popover (leaf_widgets, local path dependency for
now); "Add to queue" next to Put it up everywhere.

Delivery: the server measures the source's rate, the media's bitrate
and each member's read rate; the members card shows a source line
(who passes it on, through what, keeping up or not, source left) and
each member's route and buffer. A slow source is told once per media.
A failed server fetch is announced to everyone.

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

packages/server/lib/server.dart

@@ -6,3 +6,4 @@ export 'src/sync_server.dart';
 export 'src/jellyfin_relay.dart';
 export 'src/relay_buffer.dart';
 export 'src/relay_routes.dart';
+export 'src/relay_stats.dart';

packages/server/lib/src/piece_buffer.dart

@@ -5,6 +5,7 @@ import 'dart:typed_data';
 import 'package:sync_relay_core/core.dart';
 
 import 'relay_buffer.dart';
+import 'relay_stats.dart';
 
 final _log = createLogger('relay');
 
@@ -67,7 +68,18 @@ class PieceBuffer implements TrackBuffer {
   final _waiters = <int, List<Completer<void>>>{};
   final _aux = <int, int>{}; // init/key piece a request waits on -> since (us)
   final _viewers = <String, _Viewer>{};
-  _PiecePull? _pull;
+  @override
+  FeedClock clock = FeedClock();
+  _PiecePull? _pullNow;
+  _PiecePull? get _pull => _pullNow;
+  set _pull(_PiecePull? p) {
+    if ((_pullNow == null) != (p == null)) p == null ? clock.stop() : clock.start();
+    _pullNow = p;
+  }
+
+  /// The track's average bytes per second from the segments in so far;
+  /// null before a minute of it has arrived.
+  double? get bytesPerSec => _knownSecs >= 60 ? _knownBytes / _knownSecs : null;
   DateTime? _retryAt;
   var _disposed = false;
   Completer<void> _kick = Completer();

packages/server/lib/src/queue_ops.dart

@@ -0,0 +1,93 @@
+import 'package:relic/relic.dart';
+import 'package:sync_relay_core/core.dart';
+
+import 'server_hub.dart';
+
+// `queue` messages (Up next, see PROTOCOL.md "Queue"): what an added item
+// may carry, and the ops applied to the engine's queue.
+
+final _key = RegExp(r'^[A-Za-z0-9_-]{8,64}$');
+
+bool _httpUrl(String? url) {
+  final u = Uri.tryParse(url ?? '');
+  return u != null && (u.isScheme('http') || u.isScheme('https')) && u.host.isNotEmpty;
+}
+
+/// An added item as the server keeps it: names cleaned like a swap's,
+/// how it plays checked against what it carries, a server-fetched link's
+/// URL taken out ([privateUrl]: never broadcast). Throws
+/// [ProtocolException] with the reason for the sender.
+({QueueItem item, String? privateUrl}) cleanQueueAdd(QueueItem raw, {required bool serverHasPassword}) {
+  final media = raw.media;
+  var name = sanitizeMediaName(media.name);
+  final ref = media.jellyfin;
+  final QueueItem item;
+  String? privateUrl;
+  if (ref != null || raw.how == QueueHow.jellyfin) {
+    final clean = ref == null ? null : sanitizeJellyfinRef(ref);
+    if (clean == null) throw const ProtocolException('bad queue item');
+    if (name.isEmpty) name = clean.title;
+    item = QueueItem(
+        media: HelloMedia(name: name, size: 0, duration: 0, jellyfin: clean), how: QueueHow.jellyfin, title: '');
+  } else {
+    switch (raw.how) {
+      case QueueHow.direct:
+        if (!_httpUrl(media.url)) throw const ProtocolException('a queued link needs an http(s) url');
+        item = QueueItem(media: HelloMedia(name: name, size: 0, duration: 0, url: media.url), how: raw.how, title: '');
+      case QueueHow.server:
+        if (!_httpUrl(media.url)) throw const ProtocolException('relay fetch needs an http(s) url');
+        if (!serverHasPassword) throw const ProtocolException('server fetch needs a server password');
+        privateUrl = media.url;
+        item = QueueItem(
+            media: HelloMedia(name: name, size: 0, duration: 0, relay: const MediaRelay(fetch: true)),
+            how: raw.how,
+            title: '');
+      case QueueHow.through:
+      case QueueHow.file:
+        if (!_key.hasMatch(raw.key)) throw const ProtocolException('a queued item of yours needs a key');
+        item = QueueItem(
+            media: HelloMedia(name: name, size: media.size < 0 ? 0 : media.size, duration: 0),
+            how: raw.how,
+            title: '',
+            key: raw.key);
+      case QueueHow.jellyfin:
+        throw const ProtocolException('bad queue item');
+    }
+  }
+  var title = cleanQueueText(sanitizeMediaName(raw.title));
+  if (title.isEmpty) title = cleanQueueText(item.media.name.isEmpty ? 'media' : item.media.name);
+  return (
+    item: item.copyWith(
+        title: title, sub: cleanQueueText(sanitizeMediaName(raw.sub)), duration: raw.duration),
+    privateUrl: privateUrl,
+  );
+}
+
+/// Applies one `queue` message from [token]; a refused add answers the
+/// sender with `err` (the socket stays open).
+void handleQueueIntent(ServerHub hub, String token, RelicWebSocket ws, Map<String, Object?> decoded) {
+  final intent = QueueIntent.fromJson(decoded);
+  final engine = hub.engine;
+  var changed = false;
+  switch (intent.op) {
+    case 'add':
+      final cleaned = cleanQueueAdd(intent.item!, serverHasPassword: hub.session.secret.isNotEmpty);
+      final thumb = intent.thumb.isEmpty ? null : decodePicture(intent.thumb, RelayConstants.maxPosterBytes);
+      final error = engine.applyQueueAdd(token, cleaned.item,
+          next: intent.next, privateUrl: cleaned.privateUrl, thumb: thumb);
+      if (error != null) {
+        wsErr(ws, error);
+        return;
+      }
+      changed = true;
+    case 'move':
+      changed = engine.applyQueueMove(token, intent.id, intent.to);
+    case 'remove':
+      changed = engine.applyQueueRemove(token, intent.id, why: intent.why);
+    case 'play':
+      changed = engine.applyQueuePlay(token, intent.id);
+    case 'skip':
+      changed = engine.applyQueuePlay(token, '');
+  }
+  if (changed) hub.changed();
+}

packages/server/lib/src/relay_buffer.dart

@@ -5,6 +5,8 @@ import 'dart:typed_data';
 
 import 'package:sync_relay_core/core.dart';
 
+import 'relay_stats.dart';
+
 final _log = createLogger('relay');
 
 /// One track's buffer on the server: a byte stream ([RelayBuffer]) or
@@ -16,6 +18,9 @@ abstract interface class TrackBuffer {
   int get cachedBytes;
   int get bytesFromFeed;
   int get bytesToViewers;
+  /// Runs while this buffer has a pull in flight; the relay gives its
+  /// tracks one, for the source's rate.
+  abstract FeedClock clock;
   /// Waits for the feed's first connect, up to the wait limit.
   Future<bool> ready();
   void detach(RelayFeed feed);
@@ -162,7 +167,15 @@ class RelayBuffer implements TrackBuffer {
   final _waiters = <int, List<Completer<void>>>{};
   final _blocked = <int, int>{}; // slice -> wait start (us)
   final _viewers = <String, _Viewer>{};
-  _Pull? _pull;
+  @override
+  FeedClock clock = FeedClock();
+  _Pull? _pullNow;
+  _Pull? get _pull => _pullNow;
+  set _pull(_Pull? p) {
+    if ((_pullNow == null) != (p == null)) p == null ? clock.stop() : clock.start();
+    _pullNow = p;
+  }
+
   var _reqSeq = 0;
   var _disposed = false;
   Completer<void> _kick = Completer();

packages/server/lib/src/relay_routes.dart

@@ -10,6 +10,7 @@ import 'package:web_socket/web_socket.dart';
 import 'media_endpoint.dart';
 import 'piece_buffer.dart';
 import 'relay_buffer.dart';
+import 'relay_stats.dart';
 import 'session.dart';
 
 final _log = createLogger('relay');
@@ -56,9 +57,39 @@ class ActiveRelay {
   var _local = const <LocalRelayFeed>[];
   RelicWebSocket? _channel;
   var _channelFeeds = const <ChannelRelayFeed>[];
+  /// 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);
 
+  /// Takes the bytes the source sent so far, for [rate]; the server calls
+  /// it once a second.
+  void sampleRate() => _rate.sample(buffers.fold(0, (a, b) => a + b.bytesFromFeed), clock.busy);
+
+  /// Bytes per second the source delivers while the buffer asks for bytes
+  /// (`media.relay.rate`); 0 until measured.
+  int get rate => _rate.rate;
+
+  /// 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.
+  int need(double duration) {
+    if (buffers.isEmpty) return 0;
+    var total = 0.0;
+    for (final b in buffers) {
+      if (b is RelayBuffer) {
+        if (b.size <= 0 || duration <= 0) return 0;
+        total += b.size / duration;
+      } else if (b is PieceBuffer) {
+        final r = b.bytesPerSec;
+        if (r == null) return 0;
+        total += r;
+      }
+    }
+    return roundRate(total);
+  }
+
   int _nextId() => ++_ids;
 
   void _setTracks(List<RelayTrackSpec> tracks) {
@@ -69,6 +100,9 @@ class ActiveRelay {
             for (final t in this.tracks)
               t.pieces ? PieceBuffer(nextId: _nextId, rate: _piecesRate(t.kind)) : RelayBuffer(nextId: _nextId),
           ];
+    for (final b in buffers) {
+      b.clock = clock;
+    }
     if (!_known.isCompleted) _known.complete();
   }
 

packages/server/lib/src/relay_stats.dart

@@ -0,0 +1,103 @@
+/// How fast bytes move through the server, coarse enough for `state`
+/// (sent every second): the relay source's delivery rate and each
+/// member's read rate.
+library;
+
+/// [bytesPerSec] kept to two significant digits (under 10 kB/s: 0), so
+/// the state does not change with every byte.
+int roundRate(double bytesPerSec) {
+  if (!bytesPerSec.isFinite || bytesPerSec < 10000) return 0;
+  var unit = 1;
+  while (bytesPerSec / unit >= 100) {
+    unit *= 10;
+  }
+  return (bytesPerSec / unit).round() * unit;
+}
+
+/// Time a relay's feed had at least one pull in flight. The tracks of a
+/// media share one (they share the source's uplink), so its time counts
+/// once however many tracks pull at the same moment.
+class FeedClock {
+  final Stopwatch _watch;
+  var _open = 0;
+  FeedClock([Stopwatch? watch]) : _watch = watch ?? Stopwatch();
+
+  void start() {
+    if (_open++ == 0) _watch.start();
+  }
+
+  void stop() {
+    if (_open == 0) return;
+    if (--_open == 0) _watch.stop();
+  }
+
+  Duration get busy => _watch.elapsed;
+}
+
+/// The source's delivery rate: bytes over the time a pull was in flight,
+/// so a buffer that is full (the source idles) does not read as a slow
+/// source. Smoothed over about [window] of busy time; null until
+/// [settle] of busy time has been measured.
+class SourceRate {
+  final Duration window;
+  final Duration settle;
+  SourceRate({this.window = const Duration(seconds: 4), this.settle = const Duration(seconds: 2)});
+
+  int _bytes = 0;
+  Duration _busy = Duration.zero;
+  Duration _measured = Duration.zero;
+  double? _rate;
+
+  /// Takes the feed's totals now: all bytes so far, all busy time so far.
+  /// Gaps under a quarter second of busy time wait for the next sample
+  /// (a pull's first-byte wait would dominate them).
+  void sample(int bytes, Duration busy) {
+    final db = bytes - _bytes;
+    final dt = busy - _busy;
+    if (db < 0 || dt < Duration.zero) {
+      // Totals went back (a new feed): start over.
+      _bytes = bytes;
+      _busy = busy;
+      return;
+    }
+    if (dt < const Duration(milliseconds: 250)) return;
+    _bytes = bytes;
+    _busy = busy;
+    final secs = dt.inMicroseconds / 1e6;
+    final now = db / secs;
+    final r = _rate;
+    final w = (secs / (window.inMicroseconds / 1e6)).clamp(0.0, 1.0);
+    _rate = r == null ? now : r + (now - r) * w;
+    _measured += dt;
+  }
+
+  /// Bytes per second, rounded ([roundRate]); 0 while not settled.
+  int get rate => _measured < settle || _rate == null ? 0 : roundRate(_rate!);
+}
+
+/// Each member's read rate from the server: the change of its
+/// `bytesServed` over the last [window].
+class ReadRates {
+  final Duration window;
+  ReadRates({this.window = const Duration(seconds: 5)});
+
+  final _seen = <String, List<(DateTime, int)>>{};
+
+  /// Records [served] totals by token at [now]; returns each token's rate
+  /// (rounded). Tokens not in [served] are forgotten.
+  Map<String, int> sample(DateTime now, Map<String, int> served) {
+    _seen.removeWhere((t, _) => !served.containsKey(t));
+    final out = <String, int>{};
+    for (final e in served.entries) {
+      final list = _seen.putIfAbsent(e.key, () => []);
+      list.add((now, e.value));
+      while (list.length > 1 && now.difference(list[1].$1) >= window) {
+        list.removeAt(0);
+      }
+      final first = list.first;
+      final secs = now.difference(first.$1).inMicroseconds / 1e6;
+      out[e.key] = secs <= 0 ? 0 : roundRate((e.value - first.$2) / secs);
+    }
+    return out;
+  }
+}

packages/server/lib/src/server_hub.dart

@@ -6,7 +6,9 @@ import 'package:relic/relic.dart';
 import 'package:sync_relay_core/core.dart';
 import 'package:web_socket/web_socket.dart';
 
+import 'queue_ops.dart';
 import 'relay_routes.dart';
+import 'relay_stats.dart';
 import 'session.dart';
 
 final _hubLog = createLogger('server_hub');
@@ -80,6 +82,29 @@ class ServerHub {
     changed();
   }
 
+  final _readRates = ReadRates();
+
+  /// Once a second: each member's read rate (`members[].rate`) and the
+  /// relay's source rate and bitrate (`media.relay.rate` / `need`). Goes
+  /// out with the next state.
+  void sampleRates() {
+    final tokens = session.tokens;
+    final rates = _readRates.sample(engine.now, {
+      for (final t in engine.members.keys) t: tokens[t]?.bytesServed ?? 0,
+    });
+    for (final m in engine.members.values) {
+      m.rate = rates[m.token] ?? 0;
+    }
+    final r = relay, m = engine.media;
+    final was = m?.relay;
+    if (r == null || m == null || was == null || r.mediaId != engine.mediaId) return;
+    r.sampleRate();
+    final rate = r.rate, need = r.need(m.duration);
+    if (rate != was.rate || need != was.need) {
+      engine.media = m.copyWith(relay: was.copyWith(rate: rate, need: need));
+    }
+  }
+
   bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey;
 
   /// The member a reconnecting client may take over with its old token:
@@ -151,6 +176,13 @@ class ServerHub {
     broadcastFeed();
   }
 
+  /// The server's own fetch of [name] failed: everyone hears it (the
+  /// swapper also gets the reason as an error), not just "getting it…".
+  void fetchFailed(String name) {
+    engine.feed.add(FeedMessage(actor: 'server', action: 'fetch-fail', target: name, at: engine.now));
+    broadcastFeed();
+  }
+
   void changed() {
     broadcastState();
     broadcastFeed();
@@ -314,7 +346,19 @@ void mountSyncRoute(
               hub.changed();
             case 'swap':
               if (token == null) return;
-              final parsed = SwapIntent.fromJson(decoded).media;
+              final swap = SwapIntent.fromJson(decoded);
+              var parsed = swap.media;
+              final queueId = swap.queueId;
+              if (queueId != null) {
+                // Already put up by someone else: nothing to do.
+                if (!hub.engine.queue.has(queueId)) return;
+                // A server-fetched item's link never left the server.
+                final hidden = hub.engine.queue.privateUrlOf(queueId);
+                if (hidden != null && (parsed.url ?? '').isEmpty) {
+                  parsed = HelloMedia(
+                      name: parsed.name, size: parsed.size, duration: parsed.duration, url: hidden, relay: parsed.relay);
+                }
+              }
               if (parsed.jellyfin != null) {
                 // By reference only: no URL (it could carry a token) is
                 // stored or broadcast for Jellyfin media.
@@ -330,7 +374,7 @@ void mountSyncRoute(
                         duration: parsed.duration,
                         jellyfin: ref,
                         relay: relayReq == null ? null : hub.relayFor(token, const MediaRelay()));
-                if (media == null || !hub.engine.applySwap(token, media)) {
+                if (media == null || !hub.engine.applySwap(token, media, queueId: queueId)) {
                   wsErr(ws, 'bad swap media');
                   return;
                 }
@@ -368,7 +412,7 @@ void mountSyncRoute(
                     size: parsed.size,
                     duration: parsed.duration,
                     relay: hub.relayFor(token, relayReq));
-                if (!hub.engine.applySwap(token, media)) {
+                if (!hub.engine.applySwap(token, media, queueId: queueId)) {
                   wsErr(ws, 'bad swap media');
                   return;
                 }
@@ -379,7 +423,10 @@ void mountSyncRoute(
                     ? ActiveRelay.fetch(hub.engine.mediaId, name, link!,
                         ytDlp: hub.ytDlp,
                         onLiveChanged: () => hub.relayLiveChanged(active),
-                        onFail: (why) => wsErr(ws, hub.fetchFailText(why)))
+                        onFail: (why) {
+                          wsErr(ws, hub.fetchFailText(why));
+                          hub.fetchFailed(name);
+                        })
                     : _memberRelay(hub, token, name, media.relay!.tracks);
                 hub.setRelay(active);
                 hub.changed();
@@ -399,7 +446,7 @@ void mountSyncRoute(
                 wsErr(ws, 'swap needs a url or the hosted file');
                 return;
               }
-              if (!hub.engine.applySwap(token, intent.media)) {
+              if (!hub.engine.applySwap(token, intent.media, queueId: queueId)) {
                 wsErr(ws, 'bad swap media');
                 return;
               }
@@ -464,6 +511,9 @@ void mountSyncRoute(
             case 'nudge':
               if (token == null) return;
               hub.nudge(token, NudgeIntent.fromJson(decoded).target);
+            case 'queue':
+              if (token == null) return;
+              handleQueueIntent(hub, token, ws, decoded);
           }
         } on ProtocolException catch (e) {
           wsErr(ws, 'bad message: ${e.message}');

packages/server/lib/src/sync_server.dart

@@ -148,6 +148,7 @@ class SyncServer {
       hub.engine.tick();
       hub.sweepClosedSockets();
       hub.dropSilentSockets();
+      hub.sampleRates();
       hub.broadcastState();
       hub.broadcastFeed();
     });