From d8d487c189036528507fc079a79f6c461b2db706 Mon Sep 17 00:00:00 2001 From: foxyz Date: Wed, 7 Oct 2026 01:20:18 +0300 Subject: [PATCH] Sync fixes: clean seeks, clock offset, end of file, 2.5 min read-ahead (proto 6) - User seeks are intercepted at the player (SyncPlayer), previewed locally, sent once 300 ms after the drag ends; own echoes are matched by seekSeq and never re-applied; the landed check can't reapply a stale target. 1 s position poll removed. - ping/pong clock offset (min-RTT ClockSync); positions use server time. - Reports carry duration; buffer gate is min(4 s, remaining - 2 s); new finished transport instead of a HELD stall at the end. - cache/readahead 150 s, 256 MiB demuxer cache, pins read back. - Resumed reports put a stale ready member back in the set; hello.resume takes over the old member after an unclean drop (no ghost). - Docs: roadmap rendering findings (HW path works; shaders verified). Co-Authored-By: Claude Opus 5.5 --- .../lib/src/playback/playback_controller.dart | 51 ++++- .../lib/src/session/session_coordinator.dart | 184 +++++++++++------- apps/desktop/lib/src/sync/state_diff.dart | 34 +++- apps/desktop/lib/src/sync/sync_client.dart | 37 +++- apps/desktop/lib/src/sync/user_seek.dart | 109 +++++++++++ apps/desktop/test/clock_offset_test.dart | 52 +++++ apps/desktop/test/end_of_file_e2e_test.dart | 76 ++++++++ apps/desktop/test/readahead_e2e_test.dart | 70 +++++++ apps/desktop/test/seek_drag_e2e_test.dart | 164 ++++++++++++++++ apps/desktop/test/session_logic_test.dart | 26 --- apps/desktop/test/state_diff_test.dart | 35 ++++ apps/desktop/test/user_seek_test.dart | 68 +++++++ docs/ARCHITECTURE.md | 131 +++++++++---- docs/DEVELOPMENT.md | 9 +- docs/HOW-IT-WORKS.md | 6 +- docs/PROTOCOL.md | 50 +++-- docs/ROADMAP.md | 47 ++--- packages/core/lib/core.dart | 1 + packages/core/lib/src/clock_sync.dart | 45 +++++ packages/core/lib/src/constants.dart | 6 +- packages/core/lib/src/protocol.dart | 82 +++++++- packages/core/lib/src/sync_engine.dart | 107 ++++++++-- packages/core/test/core_test.dart | 63 ++++++ packages/core/test/sync_engine_test.dart | 135 +++++++++++++ packages/server/lib/src/server_hub.dart | 64 ++++-- packages/server/lib/src/sync_server.dart | 2 +- packages/server/test/session_test.dart | 4 +- packages/server/test/sync_wave_test.dart | 171 ++++++++++++++++ 28 files changed, 1592 insertions(+), 237 deletions(-) create mode 100644 apps/desktop/lib/src/sync/user_seek.dart create mode 100644 apps/desktop/test/clock_offset_test.dart create mode 100644 apps/desktop/test/end_of_file_e2e_test.dart create mode 100644 apps/desktop/test/readahead_e2e_test.dart create mode 100644 apps/desktop/test/seek_drag_e2e_test.dart create mode 100644 apps/desktop/test/user_seek_test.dart create mode 100644 packages/core/lib/src/clock_sync.dart create mode 100644 packages/server/test/sync_wave_test.dart diff --git a/apps/desktop/lib/src/playback/playback_controller.dart b/apps/desktop/lib/src/playback/playback_controller.dart index 38e3855..f493041 100644 --- a/apps/desktop/lib/src/playback/playback_controller.dart +++ b/apps/desktop/lib/src/playback/playback_controller.dart @@ -8,6 +8,7 @@ import 'package:signals/signals_flutter.dart'; import '../state/settings.dart'; import '../sync/state_diff.dart'; +import '../sync/user_seek.dart'; import 'frame_watchdog.dart'; /// Owns one mpv [Player] (and its [VideoController]) per session: created @@ -15,7 +16,18 @@ import 'frame_watchdog.dart'; /// [recreate] for "Reload video". Never reuse a player across sessions — /// a GPU-side freeze sticks to the player, not to the widget. class PlaybackController { - Player? _player; + /// Read-ahead target: about two minutes rides out short outages and + /// makes skipping an intro instant. + static const readAheadSecs = 150; + /// Forward demuxer cache. Holds [readAheadSecs] up to ~14 Mbit/s (a + /// 1.4 GB / 24 min file is ~8); higher bitrates get less time. In RAM: + /// media_kit sets cache-on-disk but no cache-dir, so mpv falls back. + static const maxBytes = 256 * 1024 * 1024; + /// Back buffer (media_kit sets it to bufferSize too): enough for a short + /// seek back without holding another 256 MiB. + static const maxBackBytes = 64 * 1024 * 1024; + + SyncPlayer? _player; VideoController? _video; Timer? _watchTimer; bool _probing = false; @@ -37,6 +49,13 @@ class PlaybackController { Future Function()? shownFrame; Player get player => _player!; + + /// Seeks from the video controls (seek bar) go here instead of to mpv. + void Function(Duration position)? onUserSeek; + + /// Pinned mpv properties whose read-back differs from what was set + /// (mpv ignores invalid values silently), as `name: wanted -> got`. + final List pinMismatches = []; bool get _ready => _player != null; /// The [VideoController] of the current player. Created lazily by the @@ -64,12 +83,13 @@ class PlaybackController { Future init() async { if (_player != null) return; - _player = Player( + _player = SyncPlayer( configuration: PlayerConfiguration( - bufferSize: 64 * 1024 * 1024, + bufferSize: maxBytes, libass: true, vo: Platform.environment['SYNC_RELAY_VO'], - )); + )) + ..onUserSeek = (d) => onUserSeek?.call(d); await _pinBufferProps(); _watchdog.reset(); _watchTimer = Timer.periodic(const Duration(seconds: 1), (_) => _watch()); @@ -94,14 +114,23 @@ class PlaybackController { try { final dynamic platform = player.platform; await platform?.setProperty(property, value); + final got = '${await platform?.getProperty(property)}'; + if (!pinMatches(value, got)) { + pinMismatches.add('$property: $value -> $got'); + debugPrint('pin $property=$value not applied, mpv has "$got"'); + } } catch (e, st) { + pinMismatches.add('$property: $value -> error $e'); debugPrint('pin $property=$value failed: $e\n$st'); } } Future _pinBufferProps() async { - await _pin('demuxer-readahead-secs', '10'); - await _pin('cache-secs', '10'); + pinMismatches.clear(); + await _pin('demuxer-readahead-secs', '$readAheadSecs'); + await _pin('cache-secs', '$readAheadSecs'); + await _pin('demuxer-max-bytes', '$maxBytes'); + await _pin('demuxer-max-back-bytes', '$maxBackBytes'); final hwdec = _hwdecOverride; if (hwdec != null) { await _pin('hwdec', hwdec); @@ -214,7 +243,8 @@ class PlaybackController { switch (action) { case SeekAction(:final position): if ((positionSecs - position).abs() < 0.5) break; - await player.seek(Duration(milliseconds: (position * 1000).round())); + await _player! + .syncSeek(Duration(milliseconds: (position * 1000).round())); case TransportAction(:final playing): if (playing) { await player.play(); @@ -269,3 +299,10 @@ class PlaybackController { } } } + +/// mpv echoes numbers in its own format (`150` reads back `150.000000`). +bool pinMatches(String wanted, String got) { + if (wanted == got) return true; + final a = double.tryParse(wanted), b = double.tryParse(got); + return a != null && b != null && a == b; +} diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index 82c4b2f..4161fcd 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -8,6 +8,7 @@ import '../playback/playback_controller.dart'; import '../sfx/sfx.dart'; import '../sync/state_diff.dart'; import '../sync/sync_client.dart'; +import '../sync/user_seek.dart'; import 'health.dart'; import 'idle.dart'; @@ -113,7 +114,6 @@ class SessionCoordinator { // Set after a reconnect until the server echoes the restored ready, so // the fresh member's ready=false doesn't overwrite [_lastReady]. bool _restoringReady = false; - DateTime? _reassertAt; // Bumped by every connect/leave/cancel/retry: stale attempts no-op. int _epoch = 0; Timer? _reconnectTimer; @@ -121,10 +121,19 @@ class SessionCoordinator { String? _openUrl; String _wsHost = ''; int _wsPort = 0; - double _lastPos = 0; - double? _pendingSeekTarget; + // Server seek applied to the player (event id and when), checked once it + // should have landed. + int? _pendingSeekEvent; DateTime? _pendingSeekAt; int _seekRetries = 0; + int _lastEventId = -1; + final UserSeekGate _seekGate = UserSeekGate(); + Timer? _seekTimer; + // True while a media open is in flight: the player's duration may still + // belong to the previous media then. + bool _opening = false; + // Token of the current connection, offered as `resume` on reconnect. + String _resumeToken = ''; DateTime? _lastStateAt; final IdleTracker _idle = IdleTracker(DateTime.now().toUtc()); int _toastId = 0; @@ -140,6 +149,7 @@ class SessionCoordinator { if (_playerReady) return; try { playback.onFrozen = () => unawaited(reloadVideo(auto: true)); + playback.onUserSeek = (d) => _userSeek(d.inMicroseconds / 1e6); await playback.init(); _playerReady = true; } catch (e, st) { @@ -166,12 +176,7 @@ class SessionCoordinator { await playback.recreate(); final url = _openUrl; _openUrl = null; - _pendingSeekTarget = null; - _pendingSeekAt = null; - _seekRetries = 0; - // New player starts at 0: without this the first tick would read - // the drop to 0 as a user seek and send everyone there. - _lastPos = 0; + _clearPendingSeek(); if (url != null) await _openMedia(url); } catch (e, st) { debugPrint('video reload failed: $e\n$st'); @@ -212,6 +217,10 @@ class SessionCoordinator { _restoringReady = false; _wasPlayAllowed = false; _everConnected = false; + _resumeToken = ''; + _seekGate.reset(); + _clearPendingSeek(); + _lastEventId = -1; markInput(); _subscribe(); feedLog.value = []; @@ -233,6 +242,7 @@ class SessionCoordinator { if (epoch != _epoch) throw const ConnectCancelled(); reconnecting.value = false; _everConnected = true; + _resumeToken = sync.token; health.value = const HealthStatus(HealthSeverity.ok, 'connected'); _lastStateAt = DateTime.now().toUtc(); final url = welcome.mediaUrl; @@ -280,6 +290,7 @@ class SessionCoordinator { if (url == _openUrl) return; _openUrl = url; diff.reset(); + _opening = true; try { await playback.openUrl(url); health.value = @@ -288,6 +299,8 @@ class SessionCoordinator { debugPrint('open media failed: $e'); health.value = HealthStatus(HealthSeverity.error, 'cannot open media: $e'); + } finally { + _opening = false; } } @@ -296,10 +309,12 @@ class SessionCoordinator { } Future _onState(StateBroadcast r) async { - final now = DateTime.now().toUtc(); - _lastStateAt = now; + _lastStateAt = DateTime.now().toUtc(); serverState.value = r; - _trackSelfReady(r, now); + _trackSelfReady(r); + _seekGate.observe(r, sync.memberId); + final newEvent = r.eventId != _lastEventId; + _lastEventId = r.eventId; if (r.mediaId != _mediaId) { _mediaId = r.mediaId; var url = stateMediaUrl(r); @@ -312,14 +327,16 @@ class SessionCoordinator { } else { _openUrl = null; } - _pendingSeekTarget = null; - _seekRetries = 0; + _seekGate.reset(); + _clearPendingSeek(); } if (!_playerReady || _openUrl == null) { _wasPlayAllowed = false; _updateStarving(r, 0); return; } + // Server clock: serverTime in the state is the server's. + final now = sync.serverNow(); final allowed = localPlayAllowed(r, sync.memberId); var actions = diff.diff(r, playback.positionSecs, now); if (r.transport == 'playing' && !allowed) { @@ -335,23 +352,28 @@ class SessionCoordinator { SeekAction(edge), const TransportAction(true), ]; - _pendingSeekTarget = edge; - _pendingSeekAt = now; } _wasPlayAllowed = allowed; - for (final a in actions) { - if (a is SeekAction) { - _pendingSeekTarget = a.position; - _pendingSeekAt = now; - } + // The user is seeking (or our seek has not echoed yet), or this is the + // echo of our own seek arriving: the player is where the user put it. + // Later drift corrections within that event still apply. + if (_seekGate.holdsPlayer(DateTime.now().toUtc()) || + (newEvent && _seekGate.isOwnEcho(r, sync.memberId))) { + actions = [for (final a in actions) if (a is! SeekAction) a]; + } + if (actions.any((a) => a is SeekAction)) { + _pendingSeekEvent = r.eventId; + _pendingSeekAt = DateTime.now().toUtc(); + _seekRetries = 0; } await playback.apply(actions); _updateStarving(r, playback.bufferedSecsAhead); } /// Mirrors the server's ready flag for self into [_lastReady] (force - /// by others included) and re-enters the set after a stale spell. - void _trackSelfReady(StateBroadcast r, DateTime now) { + /// by others included). After a stale spell the server puts a ready + /// member back in the set by itself. + void _trackSelfReady(StateBroadcast r) { final self = selfMember.peek(); if (self == null) return; if (r.mediaId != _mediaId) _restoringReady = false; @@ -360,12 +382,6 @@ class SessionCoordinator { } else { _lastReady = self.ready; } - if (shouldReassertReady(self) && - (_reassertAt == null || - now.difference(_reassertAt!) >= const Duration(seconds: 3))) { - _reassertAt = now; - sync.sendReady(true, source: 'auto'); - } } void _updateStarving(StateBroadcast r, double bufferedAhead) { @@ -373,6 +389,10 @@ class SessionCoordinator { starving.value = null; return; } + if (r.transport == 'finished') { + starving.value = 'finished — seek back or change media'; + return; + } for (final m in r.members) { if (m.inSet && m.buffered < RelayConstants.minBufferSecs) { starving.value = @@ -474,13 +494,17 @@ class SessionCoordinator { toasts.value = toasts.peek().where((t) => now.difference(t.at).inSeconds < 4).toList(); final r = serverState.peek(); + if (sync.token.isNotEmpty) sync.sendPing(); if (sync.token.isNotEmpty && r != null && _playerReady) { + // Duration only once the open media is the server's current one. + final known = !_opening && _openUrl != null && _mediaId == r.mediaId; sync.sendReport( position: playback.positionSecs, buffered: playback.bufferedSecsAhead, eventId: r.eventId, + duration: known ? playback.durationSecs : 0, + mediaId: known ? r.mediaId : 0, ); - _detectUserSeek(r); _checkSeekLanded(r); } if (!_manualClose && @@ -492,8 +516,7 @@ class SessionCoordinator { starving.value = null; _wasPlayAllowed = false; _mediaId = -1; - _pendingSeekTarget = null; - _seekRetries = 0; + _clearPendingSeek(); diff.reset(); } if (!reconnecting.peek()) { @@ -527,46 +550,70 @@ class SessionCoordinator { } } - void _detectUserSeek(StateBroadcast r) { - if (_openUrl == null) { - _lastPos = playback.positionSecs; - return; - } - final now = DateTime.now().toUtc(); - final pos = playback.positionSecs; - final jumped = (pos - _lastPos).abs(); - _lastPos = pos; - if (jumped <= 2.0) return; - if ((pos - serverPositionAt(r, now)).abs() < 2.0) return; - final recentSync = _pendingSeekAt != null && - now.difference(_pendingSeekAt!).inMilliseconds < 1000; - if (recentSync) return; - markInput(); - sync.sendSeek(pos); + void _clearPendingSeek() { + _pendingSeekEvent = null; + _pendingSeekAt = null; + _seekRetries = 0; } + /// A server seek that has not landed after 1.5 s is applied once more, + /// to where the server is now (never the old target: it went stale + /// while playing). Dropped when a newer event or a user seek took over. Future _checkSeekLanded(StateBroadcast r) async { - final target = _pendingSeekTarget; + final event = _pendingSeekEvent; final at = _pendingSeekAt; - if (target == null || at == null) return; - if (DateTime.now().toUtc().difference(at).inMilliseconds < 1500) return; + if (event == null || at == null) return; + final now = DateTime.now().toUtc(); + if (event != r.eventId || _seekGate.holdsPlayer(now)) { + _clearPendingSeek(); + return; + } + if (now.difference(at).inMilliseconds < 1500) return; + final target = serverPositionAt(r, sync.serverNow()); if (seekLanded(ownPosition: playback.positionSecs, target: target)) { - _pendingSeekTarget = null; - _pendingSeekAt = null; - _seekRetries = 0; + _clearPendingSeek(); return; } if (_seekRetries < 1) { _seekRetries++; - _pendingSeekAt = DateTime.now().toUtc(); + _pendingSeekAt = now; await playback.apply([SeekAction(target)]); } else { - _pendingSeekTarget = null; - _pendingSeekAt = null; - _seekRetries = 0; + _clearPendingSeek(); } } + /// Every user seek (seek bar, arrow keys, [seekTo]) lands here: the + /// player moves at once (scrub preview), the server hears one `seek` + /// once the user stops for [UserSeekGate.debounce]. + void _userSeek(double position) { + if (!position.isFinite || position < 0) return; + if (serverState.peek() == null || _openUrl == null) return; + markInput(); + final now = DateTime.now().toUtc(); + _seekGate.request(position, now); + _clearPendingSeek(); + unawaited(playback.apply([SeekAction(position)])); + _seekTimer?.cancel(); + _seekTimer = Timer(_seekGate.debounce, _flushUserSeek); + } + + void _flushUserSeek() { + _seekTimer = null; + final settled = _seekGate.takeSettled(DateTime.now().toUtc()); + if (settled == null) return; + var position = settled.position; + // Playing on since the request: send where the player is now, so the + // server's clock and ours agree when the echo comes back. + final own = playback.positionSecs; + if (_playerReady && + playback.player.state.playing && + (own - position).abs() < 2.0) { + position = own; + } + sync.sendSeek(position, seq: settled.seq); + } + void toggleReady() { final self = selfMember.peek(); // Ready but out of the set (dropped out while stale) gates like @@ -587,19 +634,17 @@ class SessionCoordinator { sync.sendForce('all', true); } - void seekTo(double position) { - markInput(); - sync.sendSeek(position); - } + void seekTo(double position) => _userSeek(position); + /// Arrow keys: relative to a seek still in flight, so repeated presses + /// add up. void nudge(double deltaSecs) { final r = serverState.peek(); if (r == null || _openUrl == null) return; - final target = - serverPositionAt(r, DateTime.now().toUtc()) + deltaSecs; - if (target.isNaN || target < 0) return; - markInput(); - sync.sendSeek(target); + final base = _seekGate.target ?? serverPositionAt(r, sync.serverNow()); + final target = base + deltaSecs; + if (target.isNaN) return; + _userSeek(target < 0 ? 0 : target); } void sendChat(String text) { @@ -656,11 +701,13 @@ class SessionCoordinator { name: params.name, avatar: params.avatar, hostKey: params.hostKey, + resume: _resumeToken, ); // Whoever bumped the epoch also closed (or replaced) the socket. if (stale()) return; reconnecting.value = false; _everConnected = true; + _resumeToken = sync.token; _lastStateAt = DateTime.now().toUtc(); health.value = const HealthStatus(HealthSeverity.ok, 'connected'); final url = welcome.mediaUrl; @@ -739,6 +786,9 @@ class SessionCoordinator { _reconnectTimer = null; _ticker?.cancel(); _ticker = null; + _seekTimer?.cancel(); + _seekTimer = null; + _seekGate.reset(); reconnecting.value = false; _playerReady = false; try { diff --git a/apps/desktop/lib/src/sync/state_diff.dart b/apps/desktop/lib/src/sync/state_diff.dart index 9cc5c21..1193cb7 100644 --- a/apps/desktop/lib/src/sync/state_diff.dart +++ b/apps/desktop/lib/src/sync/state_diff.dart @@ -19,9 +19,13 @@ class RateAction extends PlayerAction { const RateAction(this.rate); } -double serverPositionAt(StateBroadcast state, DateTime now) { +/// Server position at [serverNow], which must be on the server's clock +/// (`SyncClient.serverNow()`, local time corrected by the clock offset): +/// `state.serverTime` is stamped by the server. +double serverPositionAt(StateBroadcast state, DateTime serverNow) { if (state.transport != 'playing') return state.position; - return state.position + now.difference(state.serverTime).inMilliseconds / 1000.0; + return state.position + + serverNow.difference(state.serverTime).inMicroseconds / 1e6; } bool seekLanded({required double ownPosition, required double target}) { @@ -39,12 +43,6 @@ bool localPlayAllowed(StateBroadcast state, String memberId) { return false; } -/// Self is ready but out of the set and fresh again: the server's stale -/// sweep dropped it from the set during a signal gap and nothing puts it -/// back. Re-sending ready restores it without the user pressing Space. -bool shouldReassertReady(MemberView? self) => - self != null && self.ready && !self.inSet && !self.stale; - class StateDiff { static const debounceWindow = Duration(seconds: 2); @@ -57,6 +55,7 @@ class StateDiff { bool get slowing => _slowing; + /// [now] is server time (see [serverPositionAt]). List diff(StateBroadcast state, double ownPosition, DateTime now) { final out = []; final serverPos = serverPositionAt(state, now); @@ -73,6 +72,25 @@ class StateDiff { return out; } + // End of media: everyone stops where they are (at or just before the + // end). No seek to the very end and no drift correction. + if (state.transport == 'finished') { + final first = _lastEventId == -1; + final changed = first || + state.eventId != _lastEventId || + state.transport != _lastTransport; + _lastEventId = state.eventId; + _lastTransport = state.transport; + _lastMediaId = state.mediaId; + _slowing = false; + _deviatingSince = null; + if (!changed) return out; + out.add(const RateAction(1.0)); + if (first) out.add(SeekAction(serverPos > 1 ? serverPos - 1 : 0)); + out.add(const TransportAction(false)); + return out; + } + if (_lastEventId == -1) { _lastEventId = state.eventId; _lastTransport = state.transport; diff --git a/apps/desktop/lib/src/sync/sync_client.dart b/apps/desktop/lib/src/sync/sync_client.dart index e958fd2..dfcceaa 100644 --- a/apps/desktop/lib/src/sync/sync_client.dart +++ b/apps/desktop/lib/src/sync/sync_client.dart @@ -7,6 +7,18 @@ import 'package:web_socket_channel/web_socket_channel.dart'; enum SyncStatus { idle, connecting, open, kicked, closed, error } class SyncClient { + /// Local clock; injectable so tests can run a client whose clock is off. + final DateTime Function() _clock; + SyncClient({DateTime Function()? clock}) + : _clock = clock ?? (() => DateTime.now().toUtc()); + + /// Offset to the server's clock, from ping/pong round trips. + final ClockSync clockSync = ClockSync(); + + /// Now on the server's clock: use this, not the local clock, against + /// `serverTime` in a state. + DateTime serverNow() => clockSync.serverNow(_clock()); + WebSocketChannel? _channel; StreamSubscription? _sub; final StreamController _states = StreamController.broadcast(); @@ -34,8 +46,10 @@ class SyncClient { String avatar = '', HelloMedia? media, String hostKey = '', + String resume = '', }) async { await disconnect(); + clockSync.reset(); _status.add(SyncStatus.connecting); final ch = WebSocketChannel.connect(wsUri); unawaited(ch.ready.then((_) {}, onError: (_) {})); @@ -63,6 +77,7 @@ class SyncClient { user: HelloUser(name: name, avatar: avatar), media: media ?? const HelloMedia(name: '', size: 0, duration: 0), hostKey: hostKey, + resume: resume, ); ch.sink.add(jsonEncode(hello.toJson())); _pendingWelcome = welcomeCompleter; @@ -81,6 +96,7 @@ class SyncClient { } catch (_) {} } _status.add(SyncStatus.open); + sendPing(); return welcome; } catch (e) { _pendingWelcome = null; @@ -138,6 +154,14 @@ class SyncClient { } catch (_) { _kicked.add(''); } + case 'pong': + try { + final pong = PongMessage.fromJson(decoded); + clockSync.add( + sentMs: pong.clientMs, + serverMs: pong.serverMs, + receivedMs: _clock().millisecondsSinceEpoch); + } catch (_) {} case 'err': _errors.add('${decoded['message'] ?? 'error'}'); case 'chat': @@ -170,9 +194,14 @@ class SyncClient { _send(ForceIntent(target: target, ready: ready).toJson()); } - void sendSeek(double position) { + void sendSeek(double position, {int seq = 0}) { if (position.isNaN || position < 0) return; - _send(SeekIntent(position: position).toJson()); + _send(SeekIntent(position: position, seq: seq).toJson()); + } + + /// One clock sample; the coordinator sends one per second. + void sendPing() { + _send(PingMessage(clientMs: _clock().millisecondsSinceEpoch).toJson()); } void sendSwap(HelloMedia media) { @@ -183,6 +212,8 @@ class SyncClient { required double position, required double buffered, required int eventId, + double duration = 0, + int mediaId = 0, }) { final safeBuffered = buffered.isNaN || buffered < 0 ? 0.0 : buffered; _send( @@ -190,6 +221,8 @@ class SyncClient { position: position, buffered: safeBuffered, eventId: eventId, + duration: duration.isFinite && duration > 0 ? duration : 0, + mediaId: mediaId, ).toJson(), ); } diff --git a/apps/desktop/lib/src/sync/user_seek.dart b/apps/desktop/lib/src/sync/user_seek.dart new file mode 100644 index 0000000..db099fc --- /dev/null +++ b/apps/desktop/lib/src/sync/user_seek.dart @@ -0,0 +1,109 @@ +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; + +/// A [Player] whose public [seek] belongs to the user. media_kit's seek bar +/// (and anything else in the video controls) calls `player.seek`; with +/// [onUserSeek] set those calls become explicit events for the sync layer +/// instead of moving mpv behind its back. Sync-driven seeks use [syncSeek]. +class SyncPlayer extends Player { + SyncPlayer({super.configuration}); + + void Function(Duration position)? onUserSeek; + + @override + Future seek(Duration duration) async { + final hook = onUserSeek; + if (hook == null) return super.seek(duration); + hook(duration); + } + + /// Seek mpv directly (preview while scrubbing, server corrections). + Future syncSeek(Duration duration) => super.seek(duration); +} + +/// Turns a burst of user seeks (a seek bar drag fires one per pointer move, +/// arrow keys one per press) into one `seek` intent, and tells the +/// coordinator when server seeks must not touch the player: +/// +/// - while the user is still seeking ([debounce] since the last request), +/// - while our sent seek has no echo yet ([echoTimeout] at most), +/// - for the echo of our own seek (`state.seekBy` is us): the player is +/// already there, re-seeking would only hitch or yank. +class UserSeekGate { + final Duration debounce; + final Duration echoTimeout; + + UserSeekGate({ + this.debounce = const Duration(milliseconds: RelayConstants.seekDebounceMs), + this.echoTimeout = const Duration(seconds: 3), + }); + + double? _target; + DateTime? _requestedAt; + int _seq = 0; + int _sentSeq = 0; + DateTime? _sentAt; + + /// Latest requested position, while a request or its echo is pending. + double? get target => _target; + + /// The user asked for [position] at [now]. + void request(double position, DateTime now) { + _target = position; + _requestedAt = now; + } + + bool scrubbing(DateTime now) => + _requestedAt != null && now.difference(_requestedAt!) < debounce; + + /// The settled request to send, once [debounce] has passed since the last + /// one; null otherwise. Returns its seq, which the server echoes. + ({double position, int seq})? takeSettled(DateTime now) { + final t = _target; + if (t == null || _requestedAt == null || scrubbing(now)) return null; + _requestedAt = null; + _sentSeq = ++_seq; + _sentAt = now; + return (position: t, seq: _sentSeq); + } + + bool get _awaitingEcho => _sentAt != null; + + /// Server seeks are ignored while this holds. + bool holdsPlayer(DateTime now) { + if (_requestedAt != null) return true; // not sent yet + final sent = _sentAt; + if (sent == null) return false; + if (now.difference(sent) >= echoTimeout) { + _sentAt = null; + _target = null; + return false; + } + return true; + } + + /// The state's current event is a seek this client sent. + bool isOwnEcho(StateBroadcast state, String memberId) { + final seq = state.seekSeq; + return memberId.isNotEmpty && + state.seekBy == memberId && + seq != null && + seq > 0 && + seq <= _seq; + } + + /// Feed every state: the echo of the last sent seek ends the hold. + void observe(StateBroadcast state, String memberId) { + if (!_awaitingEcho || _requestedAt != null) return; + if (isOwnEcho(state, memberId) && state.seekSeq! >= _sentSeq) { + _sentAt = null; + _target = null; + } + } + + void reset() { + _target = null; + _requestedAt = null; + _sentAt = null; + } +} diff --git a/apps/desktop/test/clock_offset_test.dart b/apps/desktop/test/clock_offset_test.dart new file mode 100644 index 0000000..b01819e --- /dev/null +++ b/apps/desktop/test/clock_offset_test.dart @@ -0,0 +1,52 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:watch_together/src/sync/state_diff.dart'; +import 'package:watch_together/src/sync/sync_client.dart'; + +Future _poll(FutureOr Function() cond, String what) async { + final end = DateTime.now().add(const Duration(seconds: 10)); + while (DateTime.now().isBefore(end)) { + if (await cond()) return; + await Future.delayed(const Duration(milliseconds: 50)); + } + fail('timed out waiting for $what'); +} + +void main() { + test('a client whose clock is 3 s off still extrapolates the server position', + () async { + final server = SyncServer(const AppConfig(port: 0, secret: '', name: 'clk')); + await server.start(); + const skew = Duration(seconds: 3); + final client = SyncClient(clock: () => DateTime.now().toUtc().add(skew)); + try { + await client.connect( + wsUri: Uri.parse('ws://127.0.0.1:${server.boundPort}/sync'), + secret: '', + name: 'skewed'); + client.sendReady(true); + client.sendReport(position: 0, buffered: 10, eventId: 0); + await _poll(() => client.lastState?.transport == 'playing', 'playing'); + for (var i = 0; i < 5; i++) { + client.sendPing(); + await Future.delayed(const Duration(milliseconds: 50)); + } + await _poll(() => client.clockSync.hasSample, 'a pong'); + expect(client.clockSync.offsetMs, closeTo(-skew.inMilliseconds, 50)); + final st = client.lastState!; + final truth = server.hub.engine.positionAt(); + // The bug: extrapolating with the local clock is off by the skew. + final naive = serverPositionAt(st, DateTime.now().toUtc().add(skew)); + expect(naive - truth, closeTo(3, 0.1)); + // The fix: extrapolate on the server's clock. + final corrected = serverPositionAt(st, client.serverNow()); + expect(corrected, closeTo(truth, 0.1)); + } finally { + await client.dispose(); + await server.stop(); + } + }); +} diff --git a/apps/desktop/test/end_of_file_e2e_test.dart b/apps/desktop/test/end_of_file_e2e_test.dart new file mode 100644 index 0000000..34e3413 --- /dev/null +++ b/apps/desktop/test/end_of_file_e2e_test.dart @@ -0,0 +1,76 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:watch_together/src/session/session_coordinator.dart'; + +const _mediaPath = '/home/foxyz/Videos/chiyo.mp4'; // 13.8 s + +Future _poll( + FutureOr Function() cond, + String what, { + Duration timeout = const Duration(seconds: 30), +}) async { + final deadline = DateTime.now().add(timeout); + while (DateTime.now().isBefore(deadline)) { + if (await cond()) return; + await Future.delayed(const Duration(milliseconds: 100)); + } + fail('timed out waiting for $what'); +} + +void main() { + test( + 'plays through the last 4 s to finished; seek back plays again', + timeout: const Timeout(Duration(minutes: 2)), + () async { + MediaKit.ensureInitialized(); + final server = SyncServer(const AppConfig( + port: 0, secret: 'eof', name: 'eof', mediaPath: _mediaPath)); + await server.start(); + final c = SessionCoordinator(); + try { + await c.connect(ConnectParams( + host: '127.0.0.1', + port: server.boundPort, + secret: 'eof', + name: 'solo')); + await c.playback.player.setVolume(0); + await _poll(() => c.serverState.peek() != null, 'server state'); + await _poll( + () => (c.serverState.peek()?.media?['duration'] as num? ?? 0) > 13, + 'server learns the duration from reports'); + c.toggleReady(); + await _poll(() => c.playback.player.state.playing, 'playing'); + c.seekTo(7); + final transports = []; + await _poll(() { + transports.add(c.serverState.peek()?.transport ?? ''); + return c.serverState.peek()?.transport == 'finished'; + }, 'finished at the end'); + // ignore: avoid_print + print('transports: ${transports.toSet()} ' + 'end at ${c.playback.positionSecs.toStringAsFixed(2)}'); + expect(transports, isNot(contains('held')), + reason: 'never held in the last seconds'); + expect(c.serverState.peek()!.position, closeTo(13.77, 0.1)); + await _poll(() => !c.playback.player.state.playing, 'player stopped'); + expect(c.playback.positionSecs, greaterThan(12)); + expect(c.starving.peek(), contains('finished')); + + c.seekTo(2); + await _poll( + () => + c.serverState.peek()?.transport == 'playing' && + c.playback.player.state.playing && + c.playback.positionSecs < 6, + 'seek back from finished plays again'); + } finally { + await c.dispose(); + await server.stop(); + } + }, + ); +} diff --git a/apps/desktop/test/readahead_e2e_test.dart b/apps/desktop/test/readahead_e2e_test.dart new file mode 100644 index 0000000..f95d0f2 --- /dev/null +++ b/apps/desktop/test/readahead_e2e_test.dart @@ -0,0 +1,70 @@ +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:watch_together/src/playback/playback_controller.dart'; + +final _tmp = Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path; + +/// 240 s at a constant 8 Mbit/s: the bitrate of a 1.4 GB / 24 min episode. +Future _media() async { + final path = '$_tmp/readahead_240s.mp4'; + if (await File(path).exists()) return path; + final r = await Process.run('ffmpeg', [ + '-y', '-v', 'error', + '-f', 'lavfi', '-i', 'testsrc2=size=640x360:rate=24', + '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', + '-t', '240', '-c:v', 'libx264', '-preset', 'ultrafast', '-g', '48', + '-b:v', '8M', '-minrate', '8M', '-maxrate', '8M', '-bufsize', '4M', + '-x264-params', 'nal-hrd=cbr', '-pix_fmt', 'yuv420p', '-c:a', 'aac', + '-movflags', '+faststart', path, + ]); + if (r.exitCode != 0) fail('ffmpeg failed: ${r.stderr}'); + return path; +} + +void main() { + test('pins read back as set', () { + expect(pinMatches('150', '150.000000'), isTrue); + expect(pinMatches('268435456', '268435456'), isTrue); + expect(pinMatches('150', '10.000000'), isFalse); + expect(pinMatches('a=1,b=2', 'a=1,b=2'), isTrue); + }); + + test( + 'read-ahead grows far past 10 s from the server media endpoint', + timeout: const Timeout(Duration(minutes: 2)), + () async { + MediaKit.ensureInitialized(); + final media = await _media(); + final server = SyncServer(AppConfig( + port: 0, secret: '', name: 'readahead', mediaPath: media)); + await server.start(); + final p = PlaybackController(); + try { + await p.init(); + expect(p.pinMismatches, isEmpty, + reason: 'every pinned mpv property took effect'); + final token = server.session.issueToken('probe', ''); + await p.openUrl( + 'http://127.0.0.1:${server.boundPort}/m/default/$token/readahead.mp4'); + var best = 0.0; + final end = DateTime.now().add(const Duration(seconds: 30)); + while (DateTime.now().isBefore(end) && best < 120) { + await Future.delayed(const Duration(milliseconds: 500)); + best = p.bufferedSecsAhead; + } + // ignore: avoid_print + print('buffered ahead: ${best.toStringAsFixed(1)} s, ' + 'served ${server.session.totalBytesServed >> 20} MiB'); + expect(best, greaterThan(120), + reason: 'paused player reads ahead about two minutes'); + } finally { + await p.dispose(); + await server.stop(); + } + }, + ); +} diff --git a/apps/desktop/test/seek_drag_e2e_test.dart b/apps/desktop/test/seek_drag_e2e_test.dart new file mode 100644 index 0000000..20b7136 --- /dev/null +++ b/apps/desktop/test/seek_drag_e2e_test.dart @@ -0,0 +1,164 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:watch_together/src/session/session_coordinator.dart'; +import 'package:watch_together/src/sync/state_diff.dart'; + +final _tmp = Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path; + +/// 120 s, keyframe every second (seeks land fast). Needs an audio track: +/// without a video output attached (headless) mpv refuses a video-only file. +Future _media() async { + final path = '$_tmp/seek_drag_120s.mp4'; + if (await File(path).exists()) return path; + final r = await Process.run('ffmpeg', [ + '-y', '-v', 'error', '-f', 'lavfi', '-i', 'testsrc2=size=320x240:rate=24', + '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', + '-t', '120', '-c:v', 'libx264', '-preset', 'ultrafast', '-g', '24', + '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-movflags', '+faststart', path, + ]); + if (r.exitCode != 0) fail('ffmpeg failed: ${r.stderr}'); + return path; +} + +Future _poll( + FutureOr Function() cond, + String what, { + Duration timeout = const Duration(seconds: 30), +}) async { + final deadline = DateTime.now().add(timeout); + while (DateTime.now().isBefore(deadline)) { + if (await cond()) return; + await Future.delayed(const Duration(milliseconds: 100)); + } + fail('timed out waiting for $what'); +} + +void main() { + test( + 'seek bar drag: one seek to the release point, no yank back', + timeout: const Timeout(Duration(minutes: 3)), + () async { + MediaKit.ensureInitialized(); + final media = await _media(); + final server = SyncServer(AppConfig( + port: 0, secret: 'drag', name: 'drag', mediaPath: media)); + await server.start(); + final c = SessionCoordinator(); + try { + await c.connect(ConnectParams( + host: '127.0.0.1', + port: server.boundPort, + secret: 'drag', + name: 'solo')); + await c.playback.player.setVolume(0); + await _poll(() => c.serverState.peek() != null, 'server state'); + c.toggleReady(); + await _poll( + () => c.playback.player.state.playing && c.playback.positionSecs > 1, + 'solo playback'); + + // What media_kit's MaterialDesktopSeekBar does: player.seek on every + // pointer move (here 2.5 s of dragging from 20 s to 80 s), then once + // more on release. + final positions = []; + final sampler = Timer.periodic(const Duration(milliseconds: 100), + (_) => positions.add(c.playback.positionSecs)); + const moves = 130; // ~4 s of dragging + for (var i = 0; i <= moves; i++) { + final at = 20 + 60 * i / moves; + unawaited(c.playback.player.seek(Duration(milliseconds: (at * 1000).round()))); + await Future.delayed(const Duration(milliseconds: 30)); + } + await c.playback.player.seek(const Duration(seconds: 80)); + final releasedAt = positions.length; + await Future.delayed(const Duration(seconds: 5)); + sampler.cancel(); + + final seeks = c.feedLog.peek().where((f) => f.action == 'seek').length; + final r = c.serverState.peek()!; + final serverPos = serverPositionAt(r, DateTime.now().toUtc()); + final during = positions.sublist(0, releasedAt); + final after = positions.sublist(releasedAt); + // ignore: avoid_print + print('seek feeds=$seeks server=${serverPos.toStringAsFixed(2)} ' + 'local=${c.playback.positionSecs.toStringAsFixed(2)} ' + 'during=${during.map((p) => p.toStringAsFixed(1)).join(' ')} ' + 'after release=${after.map((p) => p.toStringAsFixed(1)).join(' ')}'); + for (var i = 1; i < during.length; i++) { + expect(during[i], greaterThan(during[i - 1] - 2), + reason: 'drag never yanked backwards (sample $i)'); + } + expect(seeks, 1, reason: 'one seek per drag, sent on release'); + expect(serverPos, inInclusiveRange(80, 87), + reason: 'server follows the release point'); + expect((c.playback.positionSecs - serverPos).abs(), lessThan(1.0)); + for (final p in after) { + expect(p, greaterThanOrEqualTo(79), reason: 'no yank back after release'); + } + } finally { + await c.dispose(); + await server.stop(); + } + }, + ); + + test( + 'drag on one client: the other follows with exactly one seek', + timeout: const Timeout(Duration(minutes: 3)), + () async { + MediaKit.ensureInitialized(); + final media = await _media(); + final server = SyncServer(AppConfig( + port: 0, secret: 'drag2', name: 'drag2', mediaPath: media)); + await server.start(); + final a = SessionCoordinator(); + final b = SessionCoordinator(); + try { + for (final (c, name) in [(a, 'a'), (b, 'b')]) { + await c.connect(ConnectParams( + host: '127.0.0.1', + port: server.boundPort, + secret: 'drag2', + name: name)); + await c.playback.player.setVolume(0); + } + await _poll(() => a.serverState.peek()?.members.length == 2, 'both in'); + a.toggleReady(); + b.toggleReady(); + await _poll( + () => a.playback.player.state.playing && b.playback.player.state.playing, + 'both playing'); + final bSamples = []; + final sampler = Timer.periodic(const Duration(milliseconds: 100), + (_) => bSamples.add(b.playback.positionSecs)); + for (var i = 0; i <= 100; i++) { + unawaited(a.playback.player + .seek(Duration(milliseconds: (30000 + 400 * i)))); + await Future.delayed(const Duration(milliseconds: 30)); + } + await a.playback.player.seek(const Duration(seconds: 70)); + await Future.delayed(const Duration(seconds: 4)); + sampler.cancel(); + final seeks = b.feedLog.peek().where((f) => f.action == 'seek').length; + // ignore: avoid_print + print('b saw $seeks seek(s); a=${a.playback.positionSecs.toStringAsFixed(2)} ' + 'b=${b.playback.positionSecs.toStringAsFixed(2)}'); + expect(seeks, 1); + expect(b.playback.positionSecs, greaterThan(70)); + expect((a.playback.positionSecs - b.playback.positionSecs).abs(), + lessThan(1.0)); + expect(bSamples.where((p) => p > 31 && p < 69), isEmpty, + reason: 'b never jumps to mid-drag positions'); + } finally { + await a.dispose(); + await b.dispose(); + await server.stop(); + } + }, + ); +} diff --git a/apps/desktop/test/session_logic_test.dart b/apps/desktop/test/session_logic_test.dart index 4a7a11a..debe1df 100644 --- a/apps/desktop/test/session_logic_test.dart +++ b/apps/desktop/test/session_logic_test.dart @@ -1,19 +1,6 @@ import 'package:flutter_test/flutter_test.dart'; -import 'package:sync_relay_core/core.dart'; import 'package:watch_together/src/session/idle.dart'; import 'package:watch_together/src/session/session_coordinator.dart'; -import 'package:watch_together/src/sync/state_diff.dart'; - -MemberView _self({bool ready = true, bool inSet = true, bool stale = false}) => - MemberView( - id: 'm1', - name: 'me', - avatar: '', - ready: ready, - inSet: inSet, - buffered: 5, - stale: stale, - ); void main() { group('IdleTracker', () { @@ -104,17 +91,4 @@ void main() { expect(n.contains('SECRET'), isFalse); }); }); - - group('shouldReassertReady', () { - test('ready, fresh, out of set (after stale sweep) -> reassert', () { - expect(shouldReassertReady(_self(inSet: false)), isTrue); - }); - test('normal states -> no', () { - expect(shouldReassertReady(null), isFalse); - expect(shouldReassertReady(_self()), isFalse); - expect(shouldReassertReady(_self(ready: false, inSet: false)), isFalse); - expect(shouldReassertReady(_self(ready: false)), isFalse); - expect(shouldReassertReady(_self(inSet: false, stale: true)), isFalse); - }); - }); } diff --git a/apps/desktop/test/state_diff_test.dart b/apps/desktop/test/state_diff_test.dart index c81b08b..dd4061b 100644 --- a/apps/desktop/test/state_diff_test.dart +++ b/apps/desktop/test/state_diff_test.dart @@ -156,4 +156,39 @@ void main() { expect(localPlayAllowed(state, ''), isFalse); }); }); + + group('finished', () { + test('stops in place: pause, no seek to the end, no drift seeks', () { + final diff = StateDiff(); + final t = DateTime.now().toUtc(); + diff.diff(_state(eventId: 1, transport: 'playing', position: 100), 100, t); + final end = _state(eventId: 2, transport: 'finished', position: 120); + final actions = diff.diff(end, 119.4, t); + expect(actions.whereType(), isEmpty); + expect(actions.whereType().single.playing, isFalse); + // Far off and later: still nothing. + expect(diff.diff(end, 50, t.add(const Duration(seconds: 10))), isEmpty); + }); + + test('late joiner lands just before the end, paused', () { + final diff = StateDiff(); + final actions = diff.diff( + _state(eventId: 5, transport: 'finished', position: 120), 0, + DateTime.now().toUtc()); + expect(actions.whereType().single.position, 119); + expect(actions.whereType().single.playing, isFalse); + }); + }); + + test('serverPositionAt extrapolates on the server clock', () { + final st = StateBroadcast( + eventId: 1, + transport: 'playing', + position: 10, + serverTime: DateTime.utc(2026, 1, 1, 12), + mediaId: 1, + members: const [], + ); + expect(serverPositionAt(st, DateTime.utc(2026, 1, 1, 12, 0, 2, 500)), 12.5); + }); } diff --git a/apps/desktop/test/user_seek_test.dart b/apps/desktop/test/user_seek_test.dart new file mode 100644 index 0000000..8c8130c --- /dev/null +++ b/apps/desktop/test/user_seek_test.dart @@ -0,0 +1,68 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:watch_together/src/sync/user_seek.dart'; + +StateBroadcast _state({int eventId = 1, String? seekBy, int? seekSeq}) => + StateBroadcast( + eventId: eventId, + transport: 'playing', + position: 0, + serverTime: DateTime.utc(2026), + mediaId: 1, + members: const [], + seekBy: seekBy, + seekSeq: seekSeq, + ); + +void main() { + final t0 = DateTime.utc(2026, 1, 1); + DateTime at(int ms) => t0.add(Duration(milliseconds: ms)); + + test('a drag is one seek, sent after the user stops', () { + final g = UserSeekGate(debounce: const Duration(milliseconds: 300)); + for (var i = 0; i < 100; i++) { + g.request(20.0 + i, at(i * 30)); + expect(g.takeSettled(at(i * 30 + 10)), isNull, reason: 'still dragging'); + expect(g.holdsPlayer(at(i * 30 + 10)), isTrue); + } + final sent = g.takeSettled(at(99 * 30 + 300)); + expect(sent!.position, 119.0); + expect(sent.seq, 1); + expect(g.takeSettled(at(99 * 30 + 600)), isNull, reason: 'sent once'); + }); + + test('holds until the echo of the last seek, then releases', () { + final g = UserSeekGate(debounce: const Duration(milliseconds: 300)); + g.request(50, at(0)); + final first = g.takeSettled(at(300))!; + g.request(60, at(400)); + final second = g.takeSettled(at(700))!; + expect(second.seq, first.seq + 1); + // Echo of the older seek: ours, but the hold stays. + final old = _state(eventId: 2, seekBy: 'm1', seekSeq: first.seq); + expect(g.isOwnEcho(old, 'm1'), isTrue); + g.observe(old, 'm1'); + expect(g.holdsPlayer(at(800)), isTrue); + g.observe(_state(eventId: 3, seekBy: 'm1', seekSeq: second.seq), 'm1'); + expect(g.holdsPlayer(at(900)), isFalse); + expect(g.target, isNull); + }); + + test('someone else\'s seek is not ours', () { + final g = UserSeekGate(); + g.request(5, at(0)); + g.takeSettled(at(1000)); + expect(g.isOwnEcho(_state(seekBy: 'm2', seekSeq: 1), 'm1'), isFalse); + expect(g.isOwnEcho(_state(), 'm1'), isFalse); + }); + + test('a lost echo stops holding the player after the timeout', () { + final g = UserSeekGate( + debounce: const Duration(milliseconds: 300), + echoTimeout: const Duration(seconds: 3)); + g.request(5, at(0)); + g.takeSettled(at(300)); + expect(g.holdsPlayer(at(3000)), isTrue); + expect(g.holdsPlayer(at(3400)), isFalse); + }); +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 474f4a0..311b7d9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -62,36 +62,65 @@ rejoin at once with a new token. Loosely modeled on Syncplay, with a different readiness design. -- **Transport** is `playing` or `held`. The server owns the clock: +- **Transport** is `playing`, `held` or `finished`. The server owns the clock: `positionAt(now) = position + (playing ? now - serverTime : 0)`. - Reports from clients never move it. Only seek, swap and gate transitions - write `position`/`serverTime` and bump `eventId`. -- **Reports**: clients send `position`, `buffered`, `eventId` every 1 s. The - server stores `buffered` and liveness; a report with `eventId` older than - current has its position ignored. + Reports from clients never move it. Only seek, swap, gate transitions and + the end of media write `position`/`serverTime` and bump `eventId`. +- **Clock offset**: `serverTime` is on the server's clock. Each client sends + a `ping` with its clock every second and estimates the offset NTP style + (`ClockSync`: `offset = s - (t0 + t3) / 2` of the sample with the smallest + round trip among the last 16). All extrapolation uses + `SyncClient.serverNow()` (local clock plus offset), so a client whose clock + is off still lands on the server position. +- **Reports**: clients send `position`, `buffered`, `eventId` every 1 s, plus + the media `duration` and its `mediaId` once the player knows it. The + server stores `buffered` and liveness, and takes the first reported + duration for the current media (swap resets it); a report with `eventId` + older than current has its position ignored. - **Ready gate** (HELD to PLAYING): the in-set is non-empty, every in-set - member is ready, fresh and has `buffered >= 4 s` (`minBufferSecs`), and - quorum holds: in-set size >= 2, unless at most one member is fresh. So a - lone viewer plays alone; with a second viewer present one Space waits for the - other. Joining members start with `inSet=false, ready=false` and never block. - Disconnect removes from the set; swap clears everyone. + member is ready, fresh and has `buffered >= need`, and quorum holds: in-set + size >= 2, unless at most one member is fresh. `need` is 4 s + (`minBufferSecs`), or near the end what is left minus 2 s (`endSlackSecs`, + floor 0), since mpv cannot buffer past the end. So a lone viewer plays + alone; with a second viewer present one Space waits for the other. Joining + members start with `inSet=false, ready=false` and never block. Disconnect + removes from the set; swap clears everyone. - **Pause**: while PLAYING, any in-set member who becomes not ready, under - 4 s buffered or stale flips the server to HELD (feed action `pause` by `server`). - Recovery resumes by itself when the gate passes. + `need` buffered or stale flips the server to HELD (feed action `pause` by + `server`). Recovery resumes by itself when the gate passes. +- **End**: while PLAYING with a known duration, reaching it sets FINISHED + (feed `finished` by `server`, position = duration). Clients pause where they + are; FINISHED never auto-resumes. A seek turns it into HELD at the new + position (resuming at once if the gate passes); a swap resets to HELD. - **Seek**: allowed from any member in any transport; the server writes the - position and bumps `eventId`; flags are untouched. Clients diff each `state` + position, bumps `eventId` and records who sent it (`seekBy`, `seekSeq` in + `state` for that event); flags are untouched. Clients diff each `state` broadcast by `eventId` (`StateDiff`): new event means seek to the computed server position, transport change means play/pause. Between events: drift over 5 s for more than 2 s seeks, ahead by more than 1.5 s sets rate 0.95 - until level, otherwise nothing. Local seeks are detected by the coordinator - polling mpv position once per second (`_detectUserSeek`, jump over 2 s and - more than 2 s from the server position) and sent as a `seek` intent; a retry - (`_checkSeekLanded`) re-applies a server seek that has not landed after 1.5 s. - Known roughness: see ROADMAP. + until level, otherwise nothing. +- **User seeks**: the player is a `SyncPlayer` whose public `seek` is the + user's: media_kit's seek bar (one `seek` per pointer move while dragging), + the arrow keys and `seekTo` all become explicit events. The player moves at + once (scrub preview); `UserSeekGate` sends one `seek` intent with a fresh + `seq` after 300 ms without another request (`seekDebounceMs`). While the + user is seeking, and until the echo of the last sent seq arrives (3 s at + most), server seeks never touch the player; the echo of our own seek is not + applied either (the player is already there). A server seek that has not + landed after 1.5 s is re-applied once (`_checkSeekLanded`), to the server + position recomputed now, and only while its event is still current and the + user is not seeking. - **Liveness**: WS ping every 5 s; silent over 15 s means stale (out of the gate, `stale` feed); over 30 s the socket is closed and the member removed. + A stale member that was ready goes back into the set with its next frame. Closed sockets are polled with `isClosed` every second: Relic's server-side `onDone` does not fire for dead peers, so presence never relies on it. +- **Resume**: a reconnecting client sends its previous token as + `hello.resume`. If the server still has that member (an unclean drop it has + not noticed), the new socket takes it over: same token, member id, ready + flag and media URL (the player keeps streaming), no join/leave feed, and the + dead socket is closed without removing the member. An unknown or kicked + token joins as a new member. - **Client policies**: no state broadcast for 5 s means connection lost, clear the server state and reconnect every 1.5 s until cancelled (an epoch counter makes a cancelled attempt stale) or kicked. After a reconnect the ready flag is @@ -102,8 +131,7 @@ Loosely modeled on Syncplay, with a different readiness design. auto-unready at 10 min, and never less than 2 min after the warning. - **Deviations from Syncplay**: clock never follows the slowest watcher; no pause-on-leave; anyone can force ready/unready (all or one); quorum gate; - the clock is server time, and remote clock offset is not corrected (clients - assume their clock matches the server's). + the clock is server time (clients correct for their offset to it). Additive wire changes only; bump `proto` and PROTOCOL.md in the same commit. @@ -149,12 +177,21 @@ open-ended), then every seek is a fresh `bytes=N-` connection. A server outage of a few seconds is ridden out from mpv's cache with no error. Longer outages exhaust the cache and rely on reconnect (not chaos-tested yet). -mpv settings (`PlaybackController`): `Player` with `bufferSize` 64 MiB, -`libass` on; pinned via runtime property: `demuxer-readahead-secs=10`, -`cache-secs=10`, `stream-lavf-o=reconnect_streamed=1,reconnect_on_http_error=1,reconnect_delay_max=4` -(comma form; this mpv build rejects colon-joined). media_kit has no public -`setProperty`; the code calls it through a guarded dynamic call on -`player.platform`. Hardware decoding comes from Settings (auto, off, custom) +mpv settings (`PlaybackController`): `Player` with `bufferSize` 256 MiB +(media_kit's `bufferSize` sets both `demuxer-max-bytes` and +`demuxer-max-back-bytes`), `libass` on; pinned via runtime property: +`demuxer-readahead-secs=150`, `cache-secs=150` (about two minutes of +read-ahead: rides out short drops, intro skips are instant), +`demuxer-max-bytes=268435456` (150 s up to ~14 Mbit/s; a 1.4 GB / 24 min +file is ~8), `demuxer-max-back-bytes=67108864`, +`stream-lavf-o=reconnect_streamed=1,reconnect_on_http_error=1,reconnect_delay_max=4` +(comma form; this mpv build rejects colon-joined). The cache is in RAM: +media_kit sets `cache-on-disk` without a `cache-dir`, and mpv falls back. mpv +ignores invalid values silently, so every pin is read back and a mismatch is +logged and kept in `pinMismatches`. Measured against the `/m` endpoint +(`readahead_e2e_test`): a paused player buffers ~147 s ahead (10 s before). +media_kit has no public `setProperty`/`getProperty`; the code calls them +through a guarded dynamic call on `player.platform`. Hardware decoding comes from Settings (auto, off, custom) and is passed to the `VideoController`; env `SYNC_RELAY_HWDEC` (e.g. `no`) overrides it. It is read per player, so a change applies on the next join or Reload video. @@ -162,8 +199,8 @@ Reload video. Player lifecycle: the `Player` and its `VideoController` live for one session: created on join, disposed on leave, replaced by `recreate()` for "Reload video" (button or Ctrl+R; volume kept). `openUrl` opens with `play: false`; play/pause/seek/rate come only from -`StateDiff` actions or the play gate (`localPlayAllowed`: self in-set and -ready). A server seek within 0.5 s of the current position is skipped. `leave()` +`StateDiff` actions, the play gate (`localPlayAllowed`: self in-set and +ready) or the user's own seeks (see Sync model). A server seek within 0.5 s of the current position is skipped. `leave()` stops the player (otherwise audio keeps playing). On media change (`mediaId` bump) the coordinator opens the new URL and `StateDiff` seeks to 0, holds and resets rate. @@ -183,17 +220,30 @@ mounted and skip painting (`Visibility` with `maintainState`). ## Desktop UI -One `HomeScreen` (`screens/home/`) with a nav rail: Home (profile chip, Join -card, Host card, up to 6 Recent server chips for one-click rejoin), Settings -(profile, hardware decoding, updates, shortcuts, shaders "coming soon"), -Integrations (Jellyfin "coming soon") and About. The update indicator sits on -Home. Settings persist in two Hive boxes: `connect_form` (last form and -recents; saved even when a join fails) and `settings`. Link media names are -derived from the URL path and never contain a query string. +One `HomeScreen` (`screens/home/`) with a single top bar (`top_bar.dart`): +profile chip, Join and Host tabs, the update indicator, Settings / +Integrations / About buttons, and the window drag area and buttons. Only the +selected view is mounted; form state lives in `HomeShell`. Join +(`join_tab.dart`): one compact address/port/password row, then up to 6 +Recent server tiles in a grid that adds columns as the window widens. While +the Join tab is visible each tile polls the server's `GET /health` every +10 s (`health_probe.dart`) for online, name, viewer count and a "needs +update" badge when `proto` differs from ours. Host (`host_tab.dart`): the +media picker (`media_picker.dart`: Jellyfin placeholder, Link, File; a +reusable widget driven by a `MediaPickerController`) fills the main area; +port, password and Host sit in a 340 px column on the right, below the +picker under 860 px. Text fields have fixed or capped widths (720 px max) so +they stay compact on ultrawide screens. Submit state (busy, cancel, +per-tab errors, server crash) is `ConnectFlow`. Settings (profile, hardware +decoding, updates, shortcuts, shaders "coming soon"), Integrations +(Jellyfin "coming soon") and About open in the same body. Settings persist +in two Hive boxes: `connect_form` (last form and recents; saved even when a +join fails) and `settings`. Link media names are derived from the URL path +and never contain a query string. Window chrome is frameless (`window_manager`): every screen brings its own -drag area and min/max/close buttons (`widgets/window_chrome_bar.dart`, -`window_buttons.dart`); the session drag layer is `window_drag_layer.dart`. +drag area and min/max/close buttons (Home's top bar; the session uses +`widgets/window_chrome_bar.dart`; both use `window_buttons.dart`); the session drag layer is `window_drag_layer.dart`. ## Health reporting @@ -203,8 +253,9 @@ for connecting, reconnecting, wrong password, version mismatch, kicked, host offline, and starvation ("waiting for X - N.N s buffered"). Errors are logged with a stack trace at the boundary, then turned into a displayable state. The server's `GET /health` returns JSON (`ok`, `name`, -`proto`, `viewers`, `bytesServed`, `uptimeSecs`); it is unauthenticated and -the app does not read it. +`proto`, `viewers`, `bytesServed`, `uptimeSecs`); it is unauthenticated, so +it carries nothing sensitive. Home's recent-server tiles poll it for live +status. ## Updater diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index f9e86fe..17acd96 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -45,8 +45,13 @@ and `e2e_check` (simulated viewers against a live server). - `cd packages/core && dart test`, `cd packages/server && dart test`. - `cd apps/desktop && flutter test test/`. -- `sync_playback_e2e_test.dart` and `coord_presence_test.dart` use real mpv; - the e2e one also expects `~/Videos/chiyo.mp4`. Run only tests related to +- Real-mpv tests: `sync_playback_e2e`, `coord_presence`, `video_reload_e2e`, + `seek_drag_e2e`, `end_of_file_e2e`, `readahead_e2e`. 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 + open video-only files, so test media needs an audio track. Tests always log + media_kit "S/W rendering" because they have no display; that says nothing + about the app. Run only tests related to what you changed. - CI (`.gitea/workflows/ci.yml`): analyze + test for core and server, plus `dart compile exe` of the dedicated server (fails if Flutter creeps into diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index f43f8e8..b99a64b 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -47,7 +47,9 @@ Friends join either way with the Join card on Home. says we are, and buffers. A new arrival is not ready yet and never holds anyone back. 3. **Ready gate.** Press Space to say you are ready. The server starts - playback only when everyone who is ready has at least 4 seconds buffered. + playback only when everyone who is ready has at least 4 seconds buffered + (or what is left, near the end). At the end everyone stops and the server + stays finished until someone seeks back or changes the video. If two or more people are ready, one person alone cannot start it for the others. A lone viewer can play alone. 4. **Play.** The server owns the clock. Each app compares its mpv position @@ -61,7 +63,7 @@ Friends join either way with the Join card on Home. 7. **Leave.** Closing or leaving drops you from the server; the others keep going. If you vanish without saying goodbye, you are marked stale after 15 s (stops blocking) and removed after 30 s. The app reconnects by itself and - restores your ready flag. + takes over your old place (same member, ready flag and video). There are no roles: anyone can seek, swap, force-ready others and kick anyone except the host. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 054d0fe..bfc350c 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,4 +1,4 @@ -# Protocol (proto = 5) +# Protocol (proto = 6) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: @@ -25,13 +25,16 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). ## Client to server ``` -{"t":"hello","proto":5,"secret":"","user":{"name":"","avatar":""}, - "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"..."} +{"t":"hello","proto":6,"secret":"","user":{"name":"","avatar":""}, + "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"...", + "resume":""} {"t":"ready","ready":true,"source":"manual"} source: manual | auto {"t":"force","target":"all"|"","ready":true} -{"t":"seek","position":123.4} +{"t":"seek","position":123.4,"seq":3} {"t":"swap","media":{"name":"","size":0,"duration":0,"url":"..."}} -{"t":"report","position":123.1,"buffered":6.2,"eventId":41} +{"t":"report","position":123.1,"buffered":6.2,"eventId":41, + "duration":1440.5,"mediaId":2} +{"t":"ping","c":1767225600000} {"t":"chat","text":"..."} {"t":"kick","target":"","reason":""} ``` @@ -44,6 +47,18 @@ Field notes: from its own client. The member whose key equals the server's is the host and cannot be kicked. A server without a key (the dedicated server) has no host. Where the connection comes from does not matter. +- `hello.resume` (optional, omitted when empty): the token from this + client's previous `welcome`, sent on reconnect. If the server still has + that member and it is not kicked, the connection takes it over: `welcome` + returns the same `token` and `memberId`, the member keeps its flags, and + there is no `join`/`leave` feed. Otherwise it is ignored and a new member + joins. +- `seek.seq` (optional, default 0): increasing per client; echoed in + `state.seekSeq` so the sender recognises its own seek. +- `report.duration` / `report.mediaId` (optional, omitted when 0): the open + media's length in seconds and the `mediaId` it belongs to. The server takes + the first one whose `mediaId` is current while its media duration is 0. +- `ping.c`: the client's clock in epoch ms; answered with `pong`. - `user.avatar`: base64 image, at most 256 KiB of characters. `name` is trimmed, empty becomes `?`, cut to 32 characters. - `chat.text` is cut to 500 characters. @@ -55,10 +70,12 @@ Field notes: ``` {"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","state":{}} -{"t":"state","eventId":7,"transport":"playing"|"held","position":12.3, +{"t":"state","eventId":7,"transport":"playing"|"held"|"finished","position":12.3, "serverTime":"","mediaId":1, "media":{name,size,duration,url?}|absent, - "members":[{"id","name","avatar","ready","inSet","buffered","stale"}]} + "members":[{"id","name","avatar","ready","inSet","buffered","stale"}], + "seekBy":"m1","seekSeq":3} +{"t":"pong","c":1767225600000,"s":1767225600012} {"t":"feed","actor":"","action":"","target":"","reason":"","at":""} {"t":"chat","text":"...","from":""} {"t":"kicked","reason":""} @@ -71,12 +88,19 @@ Field notes: the shared link. `welcome.state` has the same shape as `state`. - `state` is sent every second and on every change. `position` is the value at - `serverTime`; clients extrapolate while `transport` is `playing`. + `serverTime` (server clock); clients extrapolate while `transport` is + `playing`, on the server's clock (see `pong`). `finished`: the media + reached its end (`position` = duration); it stays until a seek or swap. `eventId` increases on seek, swap and transport changes. - `mediaId` increases on swap. + `mediaId` increases on swap. `media.duration` is 0 until a client reports + it. `seekBy` (member id) and `seekSeq` are present only while the current + `eventId` came from a seek. +- `pong.c` echoes the ping's `c`; `pong.s` is the server's clock in epoch ms + (the clock `serverTime` uses). With `t3` the local receive time, the offset + is `s - (c + t3) / 2`, accurate to half the round trip. - `feed.action` values: `ready`, `unready`, `force-ready`, `force-unready`, `force-play`, `force-hold`, `kick`, `join`, `leave`, `stale`, `swap`, - `seek`, `pause`, `resume`, `chat`. `target` and `reason` are optional; + `seek`, `pause`, `resume`, `finished`, `chat`. `target` and `reason` are optional; `reason` carries `auto`/`manual` for ready and `silent too long` or `no signal` for leave/stale. - Chat is delivered twice: a `chat` frame with the text and a textless @@ -87,8 +111,10 @@ Field notes: `hello` then `welcome`, then the app opens `mediaUrl`, seeks to the server position and buffers. A joining member is `ready=false, inSet=false` until -the first `ready`. A socket that dies loses its token; reconnect is a fresh -`hello`, after which the client re-sends its prior ready flag. +the first `ready`. A socket the server sees close loses its token; reconnect +is a fresh `hello` (with `resume` = the old token), after which the client +re-sends its prior ready flag. If the old member was still there (unclean +drop), `resume` takes it over and the token and media URL stay the same. ## Reserved, not implemented diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 2f18d6c..1bdea2d 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -6,31 +6,20 @@ off by deleting it in the same commit as the code. Wire changes also edit ## Backlog -1. **Sync fixes.** - - Jumpy seek. Root cause: the media_kit seek bar calls `player.seek` - continuously while dragging; the coordinator only notices through its - 1 s position poll (`_detectUserSeek`), so it fights the server echo and - the seek-landed retry (`_checkSeekLanded`). - - End-of-file stall. The 4 s buffered gate cannot be met in the last 4 s of - a file, the server never learns the duration, and there is no finished - state. Add duration reporting and a finished state. - - Tiny read-ahead. Our own pins (`cache-secs=10`, - `demuxer-readahead-secs=10` in `playback_controller.dart`) cap it; want - about two minutes for large files. - - Remote clock offset is not corrected: clients assume their clock equals - the server's. - - Stale-ready. A ready member who went stale (no reports for 15 s) should - return to the ready set when reports resume; the client re-sends ready as - a workaround (remove it then). - - Ghost member after an unclean drop until `dropAfter`. +1. **Shaders.** Per-viewer local setting using mpv `glsl-shaders` (set after + the player is created), shipping Anime4K presets (bloc97, MIT). Shaders only + apply on media_kit's hardware path (verified with a test shader); warn + when the software fallback is active. The hardware path is fragile: + media_kit borrows whatever EGL context GDK left current on the main + thread (upstream media-kit#1404), so a `VideoController` created before + the first frame, an X11 session, or a Flutter/GDK change can silently + fall back to software. Log which path media_kit picked; a vendored + media_kit_video fix (build the EGL display from GDK's native display) is + the durable answer if it regresses. The video freeze after GPU idle is not + explained by this; next suspect is media_kit's context switching on the + raster thread with NVIDIA. -2. **Rendering, then shaders.** media_kit falls back to software rendering - on the dev machine. Suspect Impeller on Linux against media_kit's GL - texture path; investigate first, since shaders need hardware rendering and - it may also cause the video freeze. Then shaders: per-viewer local setting - using mpv `glsl-shaders`, shipping Anime4K presets (bloc97, MIT). - -3. **Windows build + CI.** There is no Windows runner. Try +2. **Windows build + CI.** There is no Windows runner. Try `ubuntu2610/flutter_build` (LLVM-MinGW + Wine `gen_snapshot`) on the Linux runner, fallback a GitHub Actions mirror. The manifest needs per-platform URLs and `release_publish.sh` must handle a second platform. The updater's @@ -38,12 +27,12 @@ off by deleting it in the same commit as the code. Wire changes also edit `watch-together-server` needs a Windows `dart compile exe` too. Desktop tests are not in CI either. -4. **Jellyfin** (epic, below). +3. **Jellyfin** (epic, below). -5. **Smaller items** - - Link conveniences: Ctrl+V pastes into the media field; a link swap in - session updates the remembered media; show the open link with an - open-in-browser button. +4. **Smaller items** + - Link conveniences: a link swap in session updates the remembered media; + show the open link with an open-in-browser button; the session's swap + panel reuses Home's `MediaPicker`. - Time UI: remaining time, scrubber preview. - mpv / mpv.net shortcut parity. - Sounds: replace `NoopSfxService` (cues already fire). Candidates: diff --git a/packages/core/lib/core.dart b/packages/core/lib/core.dart index 96b151d..66cffa9 100644 --- a/packages/core/lib/core.dart +++ b/packages/core/lib/core.dart @@ -1,6 +1,7 @@ export 'src/constants.dart'; export 'src/errors.dart'; export 'src/logging.dart'; +export 'src/clock_sync.dart'; export 'src/config.dart'; export 'src/protocol.dart'; export 'src/media_source.dart'; diff --git a/packages/core/lib/src/clock_sync.dart b/packages/core/lib/src/clock_sync.dart new file mode 100644 index 0000000..a1b16f7 --- /dev/null +++ b/packages/core/lib/src/clock_sync.dart @@ -0,0 +1,45 @@ +/// Estimates how far the server's clock is from ours (NTP style). +/// +/// Each ping/pong round trip gives one sample: the client sent at `t0`, the +/// server stamped `s`, the reply arrived at `t3` (client clock). Assuming the +/// two legs take equally long, `offset = s - (t0 + t3) / 2` with an error of +/// at most `rtt / 2`. The sample with the smallest round trip among the last +/// [window] is the most trustworthy, so that one is used. +class ClockSync { + final int window; + final List<({int rttMs, double offsetMs})> _samples = []; + + ClockSync({this.window = 16}); + + /// Server clock minus local clock, in ms; 0 until the first sample. + double get offsetMs { + if (_samples.isEmpty) return 0; + var best = _samples.first; + for (final s in _samples) { + if (s.rttMs < best.rttMs) best = s; + } + return best.offsetMs; + } + + /// Round trip of the sample in use, null without samples. + int? get rttMs { + if (_samples.isEmpty) return null; + return _samples.map((s) => s.rttMs).reduce((a, b) => a < b ? a : b); + } + + bool get hasSample => _samples.isNotEmpty; + + /// [sentMs] and [receivedMs] are local epoch ms, [serverMs] the server's. + void add({required int sentMs, required int serverMs, required int receivedMs}) { + final rtt = receivedMs - sentMs; + if (rtt < 0) return; + _samples.add((rttMs: rtt, offsetMs: serverMs - (sentMs + receivedMs) / 2)); + if (_samples.length > window) _samples.removeAt(0); + } + + void reset() => _samples.clear(); + + /// The server's clock now, given the local clock [local]. + DateTime serverNow(DateTime local) => + local.add(Duration(microseconds: (offsetMs * 1000).round())); +} diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 8ad9db8..71e6102 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 = 5; + static const int protoVersion = 6; static const int chunkBytes = 256 * 1024; static const int creditWindowChunks = 8; @@ -10,6 +10,10 @@ abstract class RelayConstants { static const double slowDownThresholdSecs = 1.5; static const double slowDownRate = 0.95; static const double minBufferSecs = 4.0; + /// Near the end the buffer requirement is what is left minus this, so a + /// client slightly ahead of the clock (or whose last packet ends a bit + /// early) still passes. + static const double endSlackSecs = 2.0; static const double resumeHysteresisSecs = 4.0; static const int stateHeartbeatSecs = 1; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index b38cc49..08b06a3 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -70,12 +70,17 @@ class HelloMessage { /// The hosting app's own client proves it is the host with this; empty /// for everyone else. final String hostKey; + /// Token from this client's previous `welcome`, sent on reconnect: if the + /// server still has that member (an unclean drop it has not noticed), the + /// connection takes it over instead of leaving a ghost. + final String resume; const HelloMessage( {required this.proto, required this.secret, required this.user, required this.media, - this.hostKey = ''}); + this.hostKey = '', + this.resume = ''}); Map toJson() => { 't': 'hello', @@ -84,6 +89,7 @@ class HelloMessage { 'user': user.toJson(), 'media': media.toJson(), if (hostKey.isNotEmpty) 'hostKey': hostKey, + if (resume.isNotEmpty) 'resume': resume, }; factory HelloMessage.fromJson(Map json) { @@ -97,6 +103,7 @@ class HelloMessage { user: HelloUser.fromJson(Map.from(user)), media: HelloMedia.fromJson(Map.from(media)), hostKey: (json['hostKey'] as String?) ?? '', + resume: (json['resume'] as String?) ?? '', ); } } @@ -212,13 +219,19 @@ class ForceIntent { class SeekIntent { final double position; - const SeekIntent({required this.position}); + /// Client-chosen, increasing per connection; echoed in `state.seekSeq` so + /// the sender recognises its own seek. + final int seq; + const SeekIntent({required this.position, this.seq = 0}); - Map toJson() => {'t': 'seek', 'position': position}; + Map toJson() => + {'t': 'seek', 'position': position, if (seq != 0) 'seq': seq}; factory SeekIntent.fromJson(Map json) { if (json['t'] != 'seek') throw const ProtocolException('not a seek intent'); - return SeekIntent(position: finiteDouble(json['position'], 'position')); + return SeekIntent( + position: finiteDouble(json['position'], 'position'), + seq: finiteInt(json['seq'], 'seq')); } } @@ -240,10 +253,25 @@ class ReportMessage { final double position; final double buffered; final int eventId; - const ReportMessage({required this.position, required this.buffered, required this.eventId}); + /// Length of the open media in seconds, 0 while unknown. + final double duration; + /// The `mediaId` the client has open; [duration] belongs to it. + final int mediaId; + const ReportMessage( + {required this.position, + required this.buffered, + required this.eventId, + this.duration = 0, + this.mediaId = 0}); - Map toJson() => - {'t': 'report', 'position': position, 'buffered': buffered, 'eventId': eventId}; + Map toJson() => { + 't': 'report', + 'position': position, + 'buffered': buffered, + 'eventId': eventId, + if (duration > 0) 'duration': duration, + if (mediaId != 0) 'mediaId': mediaId, + }; factory ReportMessage.fromJson(Map json) { if (json['t'] != 'report') throw const ProtocolException('not a report'); @@ -251,10 +279,40 @@ class ReportMessage { position: finiteDouble(json['position'], 'position'), buffered: finiteDouble(json['buffered'], 'buffered'), eventId: finiteInt(json['eventId'], 'eventId', fallback: 0), + duration: finiteDouble(json['duration'], 'duration'), + mediaId: finiteInt(json['mediaId'], 'mediaId'), ); } } +/// Clock probe: the client sends its own clock in [clientMs]; the server +/// answers with a [PongMessage] carrying it back plus its clock. +class PingMessage { + final int clientMs; + const PingMessage({required this.clientMs}); + + Map toJson() => {'t': 'ping', 'c': clientMs}; + + factory PingMessage.fromJson(Map json) { + if (json['t'] != 'ping') throw const ProtocolException('not a ping'); + return PingMessage(clientMs: finiteInt(json['c'], 'c')); + } +} + +class PongMessage { + final int clientMs; + final int serverMs; + const PongMessage({required this.clientMs, required this.serverMs}); + + Map toJson() => {'t': 'pong', 'c': clientMs, 's': serverMs}; + + factory PongMessage.fromJson(Map json) { + if (json['t'] != 'pong') throw const ProtocolException('not a pong'); + return PongMessage( + clientMs: finiteInt(json['c'], 'c'), serverMs: finiteInt(json['s'], 's')); + } +} + class ChatMessage { final String text; const ChatMessage({required this.text}); @@ -314,6 +372,10 @@ class StateBroadcast { final int mediaId; final Map? media; final List members; + /// Member id and seq of the seek that made [eventId]; null when the + /// current event is not a seek. + final String? seekBy; + final int? seekSeq; const StateBroadcast({ required this.eventId, required this.transport, @@ -322,6 +384,8 @@ class StateBroadcast { required this.mediaId, this.media, required this.members, + this.seekBy, + this.seekSeq, }); Map toJson() => { @@ -333,6 +397,8 @@ class StateBroadcast { 'mediaId': mediaId, if (media != null) 'media': media, 'members': members.map((m) => m.toJson()).toList(), + if (seekBy != null) 'seekBy': seekBy, + if (seekSeq != null) 'seekSeq': seekSeq, }; factory StateBroadcast.fromJson(Map json) { @@ -352,6 +418,8 @@ class StateBroadcast { .map((m) => MemberView.fromJson(Map.from(m))) .toList() : const [], + seekBy: json['seekBy'] as String?, + seekSeq: json['seekSeq'] == null ? null : finiteInt(json['seekSeq'], 'seekSeq'), ); } } diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index 76d1795..1bda9de 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -3,7 +3,7 @@ import 'protocol.dart'; typedef SyncClock = DateTime Function(); -enum Transport { playing, held } +enum Transport { playing, held, finished } enum Correction { none, slowDown, seek } @@ -66,6 +66,10 @@ class SyncEngine { int mediaId = 1; HelloMedia? media; int _nextId = 1; + // Who sent the seek that produced [_seekEventId], for the echo check. + String? _seekBy; + int _seekSeq = 0; + int _seekEventId = -1; SyncEngine({SyncClock? clock, Duration? staleAfter, Duration? dropAfter}) : clock = clock ?? (() => DateTime.now().toUtc()), @@ -84,6 +88,36 @@ class SyncEngine { bool isStale(EngineMember member, [DateTime? at]) => (at ?? now).difference(member.lastSeen) > staleAfter; + /// Media length in seconds, null while nobody has reported it. + double? get duration { + final d = media?.duration ?? 0; + return d > 0 ? d : null; + } + + /// Buffer every in-set member needs: [RelayConstants.minBufferSecs], or + /// less near the end where that much is not left to buffer. + double requiredBufferSecs([DateTime? at]) { + final d = duration; + const full = RelayConstants.minBufferSecs; + if (d == null) return full; + final left = d - positionAt(at) - RelayConstants.endSlackSecs; + return left.clamp(0.0, full).toDouble(); + } + + /// Any frame from a member: fresh again, and a ready member the stale + /// sweep took out of the set goes back in (ready && !inSet only arises + /// from that sweep). + void _seen(EngineMember member) { + member.lastSeen = now; + if (member.ready && !member.inSet) member.inSet = true; + } + + /// Heartbeat: stale sweep, end of media. + void tick() { + sweepStale(); + evaluate(); + } + String applyJoin(String token, String name, String avatar) { final id = 'm${_nextId++}'; members[token] = @@ -92,6 +126,14 @@ class SyncEngine { return id; } + /// A reconnect took over [token]'s member (see `hello.resume`). + void applyResume(String token) { + final member = members[token]; + if (member == null) return; + _seen(member); + evaluate(); + } + String? applyLeave(String token) { final removed = members.remove(token); if (removed == null) return null; @@ -110,7 +152,7 @@ class SyncEngine { void applyReady(String token, bool ready, {String source = 'manual'}) { final member = members[token]; if (member == null) return; - member.lastSeen = now; + _seen(member); if (member.ready == ready && member.inSet) { evaluate(); return; @@ -153,15 +195,23 @@ class SyncEngine { evaluate(); } - void applySeek(String token, double seekPosition) { + void applySeek(String token, double seekPosition, {int seq = 0}) { final member = members[token]; if (member == null) return; if (!seekPosition.isFinite || seekPosition < 0) return; - member.lastSeen = now; + _seen(member); position = seekPosition; serverTime = now; eventId++; + _seekBy = member.id; + _seekSeq = seq; + _seekEventId = eventId; feed.add(FeedMessage(actor: member.name, action: 'seek', at: now)); + // Seeking back from the end: held, and resumes once the gate passes. + if (transport == Transport.finished) { + transport = Transport.held; + evaluate(); + } } /// Returns false (no state change) for media with non-finite numbers. @@ -183,12 +233,19 @@ class SyncEngine { return true; } - void applyReport(String token, double reportPosition, double buffered, int reportEventId) { + void applyReport(String token, double reportPosition, double buffered, int reportEventId, + {double duration = 0, int mediaId = 0}) { final member = members[token]; if (member == null) return; if (!reportPosition.isFinite) return; - member.lastSeen = now; + _seen(member); member.buffered = !buffered.isFinite || buffered < 0 ? 0 : buffered; + final m = media; + // First report of the length wins; only for the media the client has open. + if (m != null && m.duration <= 0 && mediaId == this.mediaId && + duration.isFinite && duration > 0) { + media = HelloMedia(name: m.name, size: m.size, duration: duration, url: m.url); + } if (reportEventId < eventId) return; evaluate(); } @@ -221,10 +278,21 @@ class SyncEngine { void evaluate() { final t = now; + if (transport == Transport.finished) return; + final need = requiredBufferSecs(t); if (transport == Transport.playing) { + final d = duration; + if (d != null && positionAt(t) >= d) { + position = d; + serverTime = t; + eventId++; + transport = Transport.finished; + feed.add(FeedMessage(actor: 'server', action: 'finished', at: t)); + return; + } final blocked = members.values .where((m) => m.inSet) - .any((m) => !m.ready || m.buffered < RelayConstants.minBufferSecs || isStale(m, t)); + .any((m) => !m.ready || m.buffered < need || isStale(m, t)); if (blocked) { position = positionAt(t); serverTime = t; @@ -240,7 +308,7 @@ class SyncEngine { members.values.where((m) => !isStale(m, t)).length; final quorum = inSet.length >= 2 || freshCount <= 1; final allGood = inSet.every( - (m) => m.ready && m.buffered >= RelayConstants.minBufferSecs && !isStale(m, t)); + (m) => m.ready && m.buffered >= need && !isStale(m, t)); if (allGood && quorum) { serverTime = t; eventId++; @@ -249,13 +317,18 @@ class SyncEngine { } } - StateBroadcast broadcast() => StateBroadcast( - eventId: eventId, - transport: transport == Transport.playing ? 'playing' : 'held', - position: position, - serverTime: serverTime, - mediaId: mediaId, - media: media?.toJson(), - members: members.values.map((m) => m.view(stale: isStale(m))).toList(), - ); + StateBroadcast broadcast() { + final bySeek = _seekEventId == eventId; + return StateBroadcast( + eventId: eventId, + transport: transport.name, + position: position, + serverTime: serverTime, + mediaId: mediaId, + media: media?.toJson(), + members: members.values.map((m) => m.view(stale: isStale(m))).toList(), + seekBy: bySeek ? _seekBy : null, + seekSeq: bySeek ? _seekSeq : null, + ); + } } diff --git a/packages/core/test/core_test.dart b/packages/core/test/core_test.dart index 1006aa8..02f2ec2 100644 --- a/packages/core/test/core_test.dart +++ b/packages/core/test/core_test.dart @@ -130,4 +130,67 @@ void main() { expect(RangeRequest.parse('bytes=5000-6000', 1000), isNull); }); }); + + group('ClockSync', () { + test('uses the min-RTT sample, not the average', () { + final c = ClockSync(); + expect(c.offsetMs, 0); + // Server is 5000 ms ahead. Symmetric 10 ms trip: exact. + c.add(sentMs: 1000, serverMs: 6010, receivedMs: 1020); + // Asymmetric slow trips (300 ms out, 10 back) would skew an average. + c.add(sentMs: 2000, serverMs: 7300, receivedMs: 2310); + c.add(sentMs: 3000, serverMs: 8400, receivedMs: 3410); + expect(c.rttMs, 20); + expect(c.offsetMs, 5000); + final local = DateTime.utc(2026, 1, 1); + expect(c.serverNow(local), local.add(const Duration(seconds: 5))); + }); + + test('old samples age out of the window', () { + final c = ClockSync(window: 2); + c.add(sentMs: 0, serverMs: 105, receivedMs: 10); // offset 100 + c.add(sentMs: 0, serverMs: 250, receivedMs: 100); + c.add(sentMs: 0, serverMs: 250, receivedMs: 100); + expect(c.offsetMs, 200); + }); + + test('negative round trip ignored', () { + final c = ClockSync()..add(sentMs: 10, serverMs: 0, receivedMs: 5); + expect(c.hasSample, isFalse); + }); + }); + + group('proto 6 fields', () { + Map rt(Map j) => + Map.from(jsonDecode(jsonEncode(j)) as Map); + + test('round-trip', () { + final hello = HelloMessage.fromJson(rt(const HelloMessage( + proto: 6, + secret: '', + user: HelloUser(name: 'a'), + media: HelloMedia(name: '', size: 0, duration: 0), + resume: 'tok') + .toJson())); + expect(hello.resume, 'tok'); + final seek = SeekIntent.fromJson(rt(const SeekIntent(position: 3, seq: 9).toJson())); + expect(seek.seq, 9); + final report = ReportMessage.fromJson(rt(const ReportMessage( + position: 1, buffered: 2, eventId: 3, duration: 1440.5, mediaId: 4) + .toJson())); + expect(report.duration, 1440.5); + expect(report.mediaId, 4); + final pong = PongMessage.fromJson( + rt(const PongMessage(clientMs: 5, serverMs: 7).toJson())); + expect((pong.clientMs, pong.serverMs), (5, 7)); + expect(PingMessage.fromJson(rt(const PingMessage(clientMs: 5).toJson())).clientMs, 5); + }); + + test('older frames without the new fields still parse', () { + expect(SeekIntent.fromJson({'t': 'seek', 'position': 1}).seq, 0); + final r = ReportMessage.fromJson( + {'t': 'report', 'position': 1, 'buffered': 2, 'eventId': 3}); + expect((r.duration, r.mediaId), (0.0, 0)); + }); + }); } diff --git a/packages/core/test/sync_engine_test.dart b/packages/core/test/sync_engine_test.dart index a6cc3b1..81a64ad 100644 --- a/packages/core/test/sync_engine_test.dart +++ b/packages/core/test/sync_engine_test.dart @@ -402,4 +402,139 @@ void main() { expect(rpBack.eventId, 3); }); }); + + group('stale ready rejoins', () { + test('ready member swept while stale is back in the set when reports resume', () { + join('a', 'alice'); + join('b', 'bob'); + engine.applyReady('a', true); + engine.applyReady('b', true); + buffer('a', 6.0); + buffer('b', 6.0); + expect(engine.transport, Transport.playing); + // bob goes silent past staleAfter; alice keeps reporting. + for (var i = 0; i < 16; i++) { + t = t.add(const Duration(seconds: 1)); + buffer('a', 6.0); + engine.sweepStale(); + } + expect(engine.members['b']!.inSet, isFalse); + expect(engine.members['b']!.ready, isTrue); + // bob's reports resume: he must be gated again without pressing Space. + buffer('b', 6.0); + expect(engine.members['b']!.inSet, isTrue); + expect(engine.broadcast().members.firstWhere((m) => m.id == 'm2').inSet, + isTrue); + }); + + test('unready member does not rejoin on reports', () { + join('a', 'alice'); + buffer('a', 6.0); + expect(engine.members['a']!.inSet, isFalse); + }); + }); + + group('end of file', () { + setUp(() { + engine.media = const HelloMedia(name: 'f', size: 1, duration: 20); + }); + + void playTo(double secs, {required double Function(double pos) buffered}) { + while (engine.positionAt() < secs) { + t = t.add(const Duration(milliseconds: 500)); + final pos = engine.positionAt(); + engine.applyReport('a', pos, buffered(pos), engine.eventId); + engine.tick(); + } + } + + test('last seconds play out with less than 4 s left, then finished', () { + join('a', 'alice'); + engine.applyReady('a', true); + buffer('a', 6.0); + expect(engine.transport, Transport.playing); + // mpv's read-ahead cannot exceed what is left of the file. + double left(double pos) => (20 - pos).clamp(0, 6).toDouble(); + playTo(19.5, buffered: left); + expect(engine.transport, Transport.playing); + t = t.add(const Duration(seconds: 1)); + engine.tick(); + expect(engine.transport, Transport.finished); + expect(engine.positionAt(), 20); + expect(engine.broadcast().transport, 'finished'); + // Finished never auto-resumes, however ready everyone is. + t = t.add(const Duration(seconds: 5)); + buffer('a', 0); + engine.tick(); + expect(engine.transport, Transport.finished); + }); + + test('seek back from finished resumes for ready members', () { + join('a', 'alice'); + engine.applyReady('a', true); + buffer('a', 6.0); + t = t.add(const Duration(seconds: 21)); + engine.tick(); + expect(engine.transport, Transport.finished); + engine.applySeek('a', 5); + buffer('a', 6.0); + expect(engine.transport, Transport.playing); + expect(engine.positionAt(), 5); + }); + + test('swap after finished resets to held', () { + join('a', 'alice'); + engine.applyReady('a', true); + buffer('a', 6.0); + t = t.add(const Duration(seconds: 21)); + engine.tick(); + expect(engine.transport, Transport.finished); + engine.applySwap( + 'a', const HelloMedia(name: 'g', size: 1, duration: 0, url: 'http://x/g')); + expect(engine.transport, Transport.held); + expect(engine.media!.duration, 0); + }); + + test('duration learned from reports for the current media only', () { + engine.media = const HelloMedia(name: 'f', size: 1, duration: 0); + join('a', 'alice'); + engine.applyReport('a', 0, 1, engine.eventId, + duration: 99, mediaId: engine.mediaId - 1); + expect(engine.media!.duration, 0); + engine.applyReport('a', 0, 1, engine.eventId, + duration: 42.5, mediaId: engine.mediaId); + expect(engine.media!.duration, 42.5); + expect(engine.broadcast().media!['duration'], 42.5); + }); + + test('requirement is min(4 s, remaining) with slack', () { + join('a', 'alice'); + engine.applySeek('a', 18); + expect(engine.requiredBufferSecs(), lessThan(RelayConstants.minBufferSecs)); + engine.applySeek('a', 5); + expect(engine.requiredBufferSecs(), RelayConstants.minBufferSecs); + engine.media = const HelloMedia(name: 'f', size: 1, duration: 0); + expect(engine.requiredBufferSecs(), RelayConstants.minBufferSecs); + }); + }); + + group('seek attribution', () { + test('state carries who sent the seek and its seq, only for that event', () { + join('a', 'alice'); + engine.applySeek('a', 10, seq: 7); + var s = engine.broadcast(); + expect(s.seekBy, 'm1'); + expect(s.seekSeq, 7); + final back = StateBroadcast.fromJson(Map.from( + jsonDecode(jsonEncode(s.toJson())) as Map)); + expect(back.seekBy, 'm1'); + expect(back.seekSeq, 7); + // A later event (gate) is not attributed to the seek. + engine.applyReady('a', true); + buffer('a', 6.0); + s = engine.broadcast(); + expect(s.transport, 'playing'); + expect(s.seekBy, isNull); + }); + }); } diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 38df5c3..34233e7 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -23,6 +23,15 @@ class ServerHub { bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey; + /// The member a reconnecting client may take over with its old token: + /// still known and not kicked. + EngineMember? resumable(String token) { + if (token.isEmpty) return null; + final viewer = session.tokens[token]; + if (viewer == null || viewer.kicked) return null; + return engine.members[token]; + } + Session get session => sessions.getOrCreate('default', expectedSecret); void broadcastState() { @@ -90,6 +99,13 @@ class ServerHub { broadcastFeed(); } + /// Like [forgetSocket], but only while [ws] is still the member's socket: + /// after a resume the old socket's close must not remove the member. + void forgetSocketOf(String token, RelicWebSocket ws) { + if (!identical(sockets[token], ws)) return; + forgetSocket(token); + } + void forgetSocket(String token) { engine.applyLeave(token); final viewer = session.tokens[token]; @@ -175,7 +191,7 @@ void mountSyncRoute( case 'seek': if (token == null) return; final intent = SeekIntent.fromJson(decoded); - hub.engine.applySeek(token, intent.position); + hub.engine.applySeek(token, intent.position, seq: intent.seq); hub.changed(); case 'swap': if (token == null) return; @@ -219,8 +235,15 @@ void mountSyncRoute( if (token == null) return; final report = ReportMessage.fromJson(decoded); hub.engine.applyReport( - token, report.position, report.buffered, report.eventId); - hub.broadcastState(); + token, report.position, report.buffered, report.eventId, + duration: report.duration, mediaId: report.mediaId); + hub.changed(); + case 'ping': + final ping = PingMessage.fromJson(decoded); + ws.trySendText(jsonEncode(PongMessage( + clientMs: ping.clientMs, + serverMs: hub.engine.now.millisecondsSinceEpoch) + .toJson())); case 'chat': final chat = ChatMessage.fromJson(decoded); hub.broadcastChat(viewer?.name ?? '?', chat.text); @@ -258,13 +281,13 @@ void mountSyncRoute( } }, onError: (_) { - if (token != null) hub.forgetSocket(token); + if (token != null) hub.forgetSocketOf(token, ws); }, onDone: () { if (!firstText.isCompleted) { firstText.completeError(StateError('closed before hello')); } else if (token != null) { - hub.forgetSocket(token); + hub.forgetSocketOf(token, ws); } }, ); @@ -300,23 +323,36 @@ void mountSyncRoute( await close(); return; } - try { - token = session.issueToken(hello.user.name, hello.user.avatar); - } catch (e) { - _hubLog.warning('reject ${hello.user.name}: $e'); - wsErr(ws, 'rejected: $e'); - await close(); - return; + final resumed = hub.resumable(hello.resume); + String memberId; + if (resumed != null) { + // Unclean drop the server has not noticed yet: take the member over + // (same token, id, ready flag) and drop the dead socket quietly. + token = hello.resume; + memberId = resumed.id; + final old = hub.sockets.remove(token); + if (old != null) unawaited(old.tryClose()); + hub.engine.applyResume(token); + _hubLog.info('resume ${hello.user.name} ($memberId)'); + } else { + try { + token = session.issueToken(hello.user.name, hello.user.avatar); + } catch (e) { + _hubLog.warning('reject ${hello.user.name}: $e'); + wsErr(ws, 'rejected: $e'); + await close(); + return; + } + memberId = hub.engine.applyJoin(token, hello.user.name, hello.user.avatar); } session.tokens[token]!.isHost = hub.isHostKey(hello.hostKey); - final memberId = hub.engine.applyJoin(token, hello.user.name, hello.user.avatar); final activeToken = token; hub.sockets[activeToken] = ws; session.tokens[activeToken]!.onKick = (reason) async { ws.trySendText(jsonEncode(KickedMessage(reason: reason).toJson())); await Future.delayed(const Duration(milliseconds: 200)); await ws.tryClose(); - hub.forgetSocket(activeToken); + hub.forgetSocketOf(activeToken, ws); }; final media = session.media; final welcome = WelcomeMessage( diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index 785407d..833a2c1 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -101,7 +101,7 @@ class SyncServer { ); _startedAt = DateTime.now(); _heartbeat = Timer.periodic(const Duration(seconds: 1), (_) { - hub.engine.sweepStale(); + hub.engine.tick(); hub.sweepClosedSockets(); hub.dropSilentSockets(); hub.broadcastState(); diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index 261e05c..0c3c03b 100644 --- a/packages/server/test/session_test.dart +++ b/packages/server/test/session_test.dart @@ -80,8 +80,8 @@ void main() { }); group('core welcome + proto', () { - test('proto 5: state broadcast + hello hostKey', () { - expect(RelayConstants.protoVersion, 5); + test('proto 6: state broadcast + hello hostKey', () { + expect(RelayConstants.protoVersion, 6); }); test('welcome round-trips with token + member id', () { diff --git a/packages/server/test/sync_wave_test.dart b/packages/server/test/sync_wave_test.dart new file mode 100644 index 0000000..5bc765d --- /dev/null +++ b/packages/server/test/sync_wave_test.dart @@ -0,0 +1,171 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; + +/// TCP relay that can go silent without closing: the server sees a socket +/// that is open but never talks again (an unclean drop it cannot detect). +class _FreezableProxy { + final ServerSocket _listener; + final List _open = []; + bool frozen = false; + _FreezableProxy._(this._listener); + + int get port => _listener.port; + + static Future<_FreezableProxy> start(int target) async { + final l = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final p = _FreezableProxy._(l); + l.listen((client) async { + final upstream = await Socket.connect(InternetAddress.loopbackIPv4, target); + p._open..add(client)..add(upstream); + client.listen((d) { + if (!p.frozen) upstream.add(d); + }, onError: (_) {}, onDone: () {}); + upstream.listen((d) { + if (!p.frozen) client.add(d); + }, onError: (_) {}, onDone: () {}); + }); + return p; + } + + Future close() async { + for (final s in _open) { + s.destroy(); + } + await _listener.close(); + } +} + +class _Client { + final WebSocket ws; + final frames = >[]; + final _welcome = Completer>(); + _Client(this.ws) { + ws.listen((m) { + final d = Map.from(jsonDecode('$m') as Map); + frames.add(d); + if (d['t'] == 'welcome' && !_welcome.isCompleted) _welcome.complete(d); + }, onError: (_) {}, onDone: () {}); + } + + StateBroadcast? get lastState { + for (final f in frames.reversed) { + if (f['t'] == 'state') return StateBroadcast.fromJson(f); + } + return null; + } + + void send(Map j) => ws.add(jsonEncode(j)); + + static Future<(_Client, WelcomeMessage)> join(int port, String name, + {String resume = ''}) async { + final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync'); + final c = _Client(ws); + c.send(HelloMessage( + proto: RelayConstants.protoVersion, + secret: 'w', + user: HelloUser(name: name), + media: const HelloMedia(name: '', size: 0, duration: 0), + resume: resume, + ).toJson()); + final w = await c._welcome.future.timeout(const Duration(seconds: 5)); + return (c, WelcomeMessage.fromJson(w)); + } +} + +Future _until(FutureOr Function() cond, String what, + {Duration timeout = const Duration(seconds: 10)}) async { + final end = DateTime.now().add(timeout); + while (DateTime.now().isBefore(end)) { + if (await cond()) return; + await Future.delayed(const Duration(milliseconds: 50)); + } + fail('timed out waiting for $what'); +} + +void main() { + late SyncServer server; + setUp(() async { + server = SyncServer(const AppConfig(port: 0, secret: 'w', name: 'wave')); + await server.start(); + }); + tearDown(() => server.stop()); + + test('reconnect after an unclean drop takes over the old member, no ghost', + () async { + final proxy = await _FreezableProxy.start(server.boundPort); + final other = (await _Client.join(server.boundPort, 'other')).$1; + final (first, welcome) = await _Client.join(proxy.port, 'alice'); + first.send(const ReadyIntent(ready: true).toJson()); + await _until(() => server.hub.engine.members[welcome.token]?.ready == true, + 'ready'); + proxy.frozen = true; // unclean drop: the server's socket stays open + + final (second, again) = + await _Client.join(server.boundPort, 'alice', resume: welcome.token); + expect(again.memberId, welcome.memberId); + expect(again.token, welcome.token, + reason: 'same token keeps the media URL valid'); + expect(server.hub.engine.members.length, 2, + reason: 'alice once (taken over) plus other, no ghost'); + expect(server.hub.engine.members[welcome.token]!.ready, isTrue, + reason: 'ready survives the takeover'); + // The frozen old socket closing later must not remove the new one. + await proxy.close(); + await Future.delayed(const Duration(seconds: 2)); + expect(server.hub.engine.members.containsKey(welcome.token), isTrue); + expect(server.hub.sockets.containsKey(welcome.token), isTrue); + second.send(const ChatMessage(text: 'still here').toJson()); + await _until( + () => other.frames.any((f) => f['t'] == 'chat' && f['text'] == 'still here'), + 'chat from the resumed member'); + await other.ws.close(); + await second.ws.close(); + }); + + test('unknown or kicked resume token joins as a new member', () async { + final (a, wa) = await _Client.join(server.boundPort, 'a', resume: 'nope'); + expect(wa.token, isNot('nope')); + final (b, wb) = await _Client.join(server.boundPort, 'b'); + await server.hub.kickById(wb.memberId, 'bye'); + final (c, wc) = await _Client.join(server.boundPort, 'b', resume: wb.token); + expect(wc.token, isNot(wb.token)); + expect(wc.memberId, isNot(wb.memberId)); + await a.ws.close(); + await b.ws.close(); + await c.ws.close(); + }); + + test('ping gets a pong with the client stamp and the server clock', () async { + final (c, _) = await _Client.join(server.boundPort, 'p'); + final before = DateTime.now().millisecondsSinceEpoch; + c.send(const PingMessage(clientMs: 1234).toJson()); + await _until(() => c.frames.any((f) => f['t'] == 'pong'), 'pong'); + final pong = PongMessage.fromJson(c.frames.firstWhere((f) => f['t'] == 'pong')); + expect(pong.clientMs, 1234); + expect(pong.serverMs, greaterThanOrEqualTo(before)); + await c.ws.close(); + }); + + test('seek echo carries seekBy/seekSeq; report teaches the duration', + () async { + final (c, w) = await _Client.join(server.boundPort, 's'); + c.send(const SeekIntent(position: 12, seq: 3).toJson()); + await _until(() => c.lastState?.seekSeq == 3, 'seek echo'); + expect(c.lastState!.seekBy, w.memberId); + server.hub.engine.media = const HelloMedia(name: 'f', size: 1, duration: 0); + c.send(ReportMessage( + position: 12, + buffered: 5, + eventId: c.lastState!.eventId, + duration: 1440, + mediaId: c.lastState!.mediaId) + .toJson()); + await _until(() => c.lastState?.media?['duration'] == 1440, 'duration'); + await c.ws.close(); + }); +}