From 81cac1214d4ef3198ad36adb81ea44ecddf8b55f Mon Sep 17 00:00:00 2001 From: foxyz Date: Wed, 7 Oct 2026 00:55:14 +0300 Subject: [PATCH] Server on an isolate, dedicated server binary, room -> server (proto 5) - serve(Map config, events) is the one server entrypoint: the app hosts it via Isolate.spawn (crash surfaced on the Host card, logs forwarded as [server]); bin/server.dart is the dedicated server, compiled into the bundle as watch-together-server. --headless gone. - Host identity via a per-hosting hostKey in hello, replacing the loopback heuristic. - Rename room -> server across code, files and wire (t:"state", /health name); protocol bumped to 5. - CI compiles the server binary; isolate smoke tests in server and desktop suites. Docs updated. Co-Authored-By: Claude Opus 5.5 --- .gitea/workflows/ci.yml | 5 + AGENTS.md | 4 +- README.md | 28 ++- apps/desktop/README.md | 3 +- apps/desktop/lib/main.dart | 11 -- .../lib/src/playback/playback_controller.dart | 2 +- .../lib/src/screens/home/connect_flow.dart | 6 + .../lib/src/screens/home/home_page.dart | 11 ++ .../lib/src/screens/session_screen.dart | 12 +- apps/desktop/lib/src/session/idle.dart | 2 +- .../lib/src/session/session_coordinator.dart | 69 ++++---- .../desktop/lib/src/state/server_isolate.dart | 148 ++++++++++++++++ apps/desktop/lib/src/state/session_state.dart | 84 ++++++--- .../sync/{room_diff.dart => state_diff.dart} | 64 +++---- apps/desktop/lib/src/sync/sync_client.dart | 24 +-- apps/desktop/test/coord_presence_test.dart | 29 +-- apps/desktop/test/server_isolate_test.dart | 165 ++++++++++++++++++ apps/desktop/test/session_logic_test.dart | 2 +- ...om_diff_test.dart => state_diff_test.dart} | 76 ++++---- apps/desktop/test/sync_playback_e2e_test.dart | 10 +- apps/desktop/test/video_reload_e2e_test.dart | 12 +- deploy/watch-together.service | 18 +- docs/ARCHITECTURE.md | 93 ++++++---- docs/DEVELOPMENT.md | 37 +++- docs/HOW-IT-WORKS.md | 4 +- docs/PROTOCOL.md | 33 ++-- docs/ROADMAP.md | 29 ++- packages/core/lib/src/config.dart | 10 +- packages/core/lib/src/constants.dart | 4 +- packages/core/lib/src/protocol.dart | 30 ++-- packages/core/lib/src/sync_engine.dart | 14 +- packages/core/test/core_test.dart | 26 ++- packages/core/test/sync_engine_test.dart | 38 ++-- packages/server/bin/server.dart | 98 ++++++++--- packages/server/lib/server.dart | 5 +- packages/server/lib/src/serve.dart | 47 +++++ .../src/{room_hub.dart => server_hub.dart} | 31 ++-- packages/server/lib/src/session.dart | 2 +- .../{room_server.dart => sync_server.dart} | 25 +-- packages/server/pubspec.yaml | 2 +- packages/server/test/hardening_test.dart | 48 +++-- packages/server/test/lifecycle_test.dart | 4 +- packages/server/test/serve_isolate_test.dart | 70 ++++++++ ...m_auth_test.dart => server_auth_test.dart} | 29 +-- packages/server/test/server_test.dart | 4 +- packages/server/test/session_test.dart | 4 +- tool/e2e_check.dart | 12 +- tool/package_linux.sh | 12 +- 48 files changed, 1073 insertions(+), 423 deletions(-) create mode 100644 apps/desktop/lib/src/state/server_isolate.dart rename apps/desktop/lib/src/sync/{room_diff.dart => state_diff.dart} (62%) create mode 100644 apps/desktop/test/server_isolate_test.dart rename apps/desktop/test/{room_diff_test.dart => state_diff_test.dart} (66%) create mode 100644 packages/server/lib/src/serve.dart rename packages/server/lib/src/{room_hub.dart => server_hub.dart} (95%) rename packages/server/lib/src/{room_server.dart => sync_server.dart} (83%) create mode 100644 packages/server/test/serve_isolate_test.dart rename packages/server/test/{room_auth_test.dart => server_auth_test.dart} (94%) diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 7f60dac..b079af6 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -31,3 +31,8 @@ jobs: working-directory: packages/server - run: dart test working-directory: packages/server + # Compiling the dedicated server proves it stays Flutter-free. + - run: dart compile exe bin/server.dart -o build/watch-together-server + working-directory: packages/server + - run: build/watch-together-server --help + working-directory: packages/server diff --git a/AGENTS.md b/AGENTS.md index 321bc85..e41337a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,9 +15,7 @@ AGENTS.md. Source of truth: the code, then the docs below. - Each fact lives in exactly one file; others link to it. - Code and its docs change in the same commit. A wire change also bumps `RelayConstants.protoVersion` and edits PROTOCOL.md. -- User-facing wording says "server" (there is only one), never "room". - Code identifiers and wire names keep their real names until the rename - in ROADMAP item 1. +- Say "server" (there is only one), never "room", in UI, code and docs. ## Wishlist drafts diff --git a/README.md b/README.md index 9daa16d..5c0c4ac 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,8 @@ telling you to update. Home, Host card. Pick a port, an optional password (blank = anyone can join), and optionally a video: a link or a file path. Your name and avatar come from your profile. The app starts a -server inside itself and joins it. Friends need to reach your machine: open +server inside itself and joins it; you are the host and nobody can kick you. +Leaving or closing the app stops the server. Friends need to reach your machine: open the TCP port on your router (or use a LAN or VPN address). Friends open nothing. If you pick a local file, the server streams it to everyone from your disk. @@ -44,9 +45,28 @@ Nothing is uploaded or stored anywhere. ## Dedicated server -Being reworked: today the server is started by the app (host) and the -standalone server is a dev tool. The planned dedicated server is a separate -binary built from `packages/server`. See [docs/ROADMAP.md](docs/ROADMAP.md). +For an always-on box (no display needed): the release tarball also contains +`watch-together-server`, a small standalone server without the app. Run it, +then everyone (you included) uses Join. + +``` +./watch-together-server --port 8080 --secret +``` + +| Flag | Default | Meaning | +|---|---|---| +| `--port`, `-p` | 8080 | TCP port, all IPv4 interfaces | +| `--secret` | empty | password; empty = open server (a warning is printed) | +| `--name` | `server` | name reported by `/health` | +| `--media` | none | local file to serve as the starting video | +| `--advertise-host` | none | host put in file URLs instead of the one each viewer dialled | +| `--config` | none | YAML or JSON file with the same settings (`port`, `secret`, `name`, `mediaPath`, `advertiseHost`); replaces the flags | + +It prints `READY port=... name=...` once listening and stops cleanly on +Ctrl+C or SIGTERM. A dedicated server has no host: anyone can kick anyone. +To run it as a service, unpack the tarball to `/opt/watch-together` and use +[deploy/watch-together.service](deploy/watch-together.service) (copy it to +`/etc/systemd/system/`, edit the flags, `systemctl enable --now watch-together`). ## Join diff --git a/apps/desktop/README.md b/apps/desktop/README.md index 2181574..38ef689 100644 --- a/apps/desktop/README.md +++ b/apps/desktop/README.md @@ -1,5 +1,6 @@ # apps/desktop -The Flutter desktop app (the only binary users run). Layout and rules: +The Flutter desktop app (what users run; hosting runs the server from +`packages/server` on an isolate). Layout and rules: [../../docs/ARCHITECTURE.md](../../docs/ARCHITECTURE.md). Build and test: [../../docs/DEVELOPMENT.md](../../docs/DEVELOPMENT.md). diff --git a/apps/desktop/lib/main.dart b/apps/desktop/lib/main.dart index b9e4d3a..e48c3aa 100644 --- a/apps/desktop/lib/main.dart +++ b/apps/desktop/lib/main.dart @@ -19,10 +19,6 @@ Future main(List args) async { exit(exitCode); } initLogging(); - if (args.contains('--headless')) { - await runHeadless(args); - return; - } FlutterError.onError = (details) { debugPrint('FLUTTER ERROR: ${details.exceptionAsString()}\n${details.stack}'); FlutterError.presentError(details); @@ -67,13 +63,6 @@ Future main(List args) async { runApp(const WatchTogetherApp()); } -Future runHeadless(List args) async { - final session = SessionState.instance; - await session.startHeadless(args); - await ProcessSignal.sigint.watch().first; - await session.stopServer(); -} - /// Close button / Alt+F4: leave gracefully first (others see "left", a /// hosted server shuts down), bounded so a hung socket can't keep the /// window open. diff --git a/apps/desktop/lib/src/playback/playback_controller.dart b/apps/desktop/lib/src/playback/playback_controller.dart index 6e52077..38e3855 100644 --- a/apps/desktop/lib/src/playback/playback_controller.dart +++ b/apps/desktop/lib/src/playback/playback_controller.dart @@ -7,7 +7,7 @@ import 'package:media_kit_video/media_kit_video.dart'; import 'package:signals/signals_flutter.dart'; import '../state/settings.dart'; -import '../sync/room_diff.dart'; +import '../sync/state_diff.dart'; import 'frame_watchdog.dart'; /// Owns one mpv [Player] (and its [VideoController]) per session: created diff --git a/apps/desktop/lib/src/screens/home/connect_flow.dart b/apps/desktop/lib/src/screens/home/connect_flow.dart index 1c3c33e..7a9de55 100644 --- a/apps/desktop/lib/src/screens/home/connect_flow.dart +++ b/apps/desktop/lib/src/screens/home/connect_flow.dart @@ -39,6 +39,12 @@ class ConnectFlow extends ChangeNotifier { if (!_disposed) notifyListeners(); } + void showError(ConnectMode mode, CardError error) { + errors[mode] = error; + notices[mode] = null; + _notify(); + } + void clearError(ConnectMode mode) { if (errors[mode] == null && notices[mode] == null) return; errors[mode] = null; diff --git a/apps/desktop/lib/src/screens/home/home_page.dart b/apps/desktop/lib/src/screens/home/home_page.dart index 7550c2d..43bbc9a 100644 --- a/apps/desktop/lib/src/screens/home/home_page.dart +++ b/apps/desktop/lib/src/screens/home/home_page.dart @@ -33,6 +33,7 @@ class _HomePageState extends State { late final _media = TextEditingController(text: _f.linkUrl.isNotEmpty ? _f.linkUrl : _f.mediaPath); late final void Function() _stopKickWatch; + late final void Function() _stopCrashWatch; bool _dialogOpen = false; String? _lastKickShown; @@ -49,11 +50,21 @@ class _HomePageState extends State { } WidgetsBinding.instance.addPostFrameCallback((_) => _showKickedDialog(kick)); }); + // Back from a session whose hosted server crashed: say so on the Host card. + _stopCrashWatch = effect(() { + final crash = SessionState.instance.serverCrash.value; + if (crash == null) return; + WidgetsBinding.instance.addPostFrameCallback((_) { + SessionState.instance.serverCrash.value = null; + if (mounted) _flow.showError(ConnectMode.host, CardError('Server crashed', crash)); + }); + }); } @override void dispose() { _stopKickWatch(); + _stopCrashWatch(); _flow.dispose(); for (final c in [_joinHost, _joinPort, _joinSecret, _hostPort, _hostSecret, _media]) { c.dispose(); diff --git a/apps/desktop/lib/src/screens/session_screen.dart b/apps/desktop/lib/src/screens/session_screen.dart index 27f594e..501ee21 100644 --- a/apps/desktop/lib/src/screens/session_screen.dart +++ b/apps/desktop/lib/src/screens/session_screen.dart @@ -354,7 +354,7 @@ class _SessionScreenState extends State { final coord = SessionState.instance.coordinator; return SignalBuilder( builder: (context) { - final r = coord.room.value; + final r = coord.serverState.value; final waiting = [ for (final m in r?.members ?? []) if (!m.ready) m, @@ -648,7 +648,7 @@ class _SheetPanel extends StatelessWidget { padding: const EdgeInsets.fromLTRB(16, 12, 16, 16), child: SignalBuilder( builder: (context) { - final r = coord.room.value; + final r = coord.serverState.value; final log = coord.feedLog.value; final theme = Theme.of(context); final sectionStyle = theme.textTheme.titleSmall?.copyWith(fontWeight: FontWeight.bold); @@ -669,7 +669,7 @@ class _SheetPanel extends StatelessWidget { children: [ Text('Server', style: theme.textTheme.titleMedium?.copyWith(fontWeight: FontWeight.bold)), const SizedBox(width: 8), - _StatusChip(playing: playing, hasRoom: r != null), + _StatusChip(playing: playing, hasState: r != null), ], ), const SizedBox(height: 2), @@ -780,14 +780,14 @@ class _SheetPanel extends StatelessWidget { class _StatusChip extends StatelessWidget { final bool playing; - final bool hasRoom; - const _StatusChip({required this.playing, required this.hasRoom}); + final bool hasState; + const _StatusChip({required this.playing, required this.hasState}); @override Widget build(BuildContext context) { final Color dot; final String label; - if (!hasRoom) { + if (!hasState) { dot = Colors.grey; label = '…'; } else if (playing) { diff --git a/apps/desktop/lib/src/session/idle.dart b/apps/desktop/lib/src/session/idle.dart index 37f5d18..2fd0a6c 100644 --- a/apps/desktop/lib/src/session/idle.dart +++ b/apps/desktop/lib/src/session/idle.dart @@ -19,7 +19,7 @@ class IdleTracker { } /// Called once per tick. Warn and unready only matter while self is - /// ready; while the room plays the ready flag gates nothing. + /// ready; while playing the ready flag gates nothing. IdleEvent tick(DateTime now, {required bool playing, required bool selfReady}) { if (playing) { activity(now); diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index c745592..82c4b2f 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -6,7 +6,7 @@ import 'package:sync_relay_core/core.dart'; import '../playback/playback_controller.dart'; import '../sfx/sfx.dart'; -import '../sync/room_diff.dart'; +import '../sync/state_diff.dart'; import '../sync/sync_client.dart'; import 'health.dart'; import 'idle.dart'; @@ -42,6 +42,8 @@ class ConnectParams { final String avatar; final String mediaPath; final String linkUrl; + /// Sent in hello by the hosting app's own client (see AppConfig.hostKey). + final String hostKey; const ConnectParams({ required this.host, required this.port, @@ -50,6 +52,7 @@ class ConnectParams { this.avatar = '', this.mediaPath = '', this.linkUrl = '', + this.hostKey = '', }); } @@ -62,8 +65,8 @@ class ToastItem { const _toastActions = {'join', 'leave', 'kick', 'swap', 'chat'}; -String? roomMediaUrl(RoomBroadcast room) { - final media = room.media; +String? stateMediaUrl(StateBroadcast state) { + final media = state.media; if (media == null) return null; final url = media['url']; if (url is String && url.isNotEmpty) return url; @@ -74,11 +77,11 @@ class SessionCoordinator { final SfxService sfx; final SyncClient sync = SyncClient(); final PlaybackController playback = PlaybackController(); - final RoomDiff diff = RoomDiff(); + final StateDiff diff = StateDiff(); final health = signal(const HealthStatus(HealthSeverity.info, 'idle')); - final room = signal(null); + final serverState = signal(null); final feedLog = signal>([]); final toasts = signal>([]); final reconnecting = signal(false); @@ -86,7 +89,7 @@ class SessionCoordinator { final starving = signal(null); late final Computed selfMember = computed(() { - final r = room.value; + final r = serverState.value; if (r == null || sync.memberId.isEmpty) return null; for (final m in r.members) { if (m.id == sync.memberId) return m; @@ -95,7 +98,7 @@ class SessionCoordinator { }); Timer? _ticker; - StreamSubscription? _roomSub; + StreamSubscription? _stateSub; StreamSubscription? _feedSub; StreamSubscription? _kickSub; StreamSubscription? _errSub; @@ -122,7 +125,7 @@ class SessionCoordinator { double? _pendingSeekTarget; DateTime? _pendingSeekAt; int _seekRetries = 0; - DateTime? _lastRoomAt; + DateTime? _lastStateAt; final IdleTracker _idle = IdleTracker(DateTime.now().toUtc()); int _toastId = 0; ConnectParams? _params; @@ -224,13 +227,14 @@ class SessionCoordinator { secret: params.secret, name: params.name, avatar: params.avatar, + hostKey: params.hostKey, ); // Whoever bumped the epoch (cancel/leave) also closed the socket. if (epoch != _epoch) throw const ConnectCancelled(); reconnecting.value = false; _everConnected = true; health.value = const HealthStatus(HealthSeverity.ok, 'connected'); - _lastRoomAt = DateTime.now().toUtc(); + _lastStateAt = DateTime.now().toUtc(); final url = welcome.mediaUrl; if (url != null && url.isNotEmpty) { await _openMedia(url); @@ -257,7 +261,7 @@ class SessionCoordinator { } void _subscribe() { - _roomSub ??= sync.rooms.listen(_onRoom); + _stateSub ??= sync.states.listen(_onState); _feedSub ??= sync.feeds.listen(_onFeed); _kickSub ??= sync.kicked.listen(_onKicked); _errSub ??= sync.errors.listen((msg) { @@ -291,14 +295,14 @@ class SessionCoordinator { return 'http://$_wsHost:$_wsPort/m/default/${sync.token}/${Uri.encodeComponent(name)}'; } - Future _onRoom(RoomBroadcast r) async { + Future _onState(StateBroadcast r) async { final now = DateTime.now().toUtc(); - _lastRoomAt = now; - room.value = r; + _lastStateAt = now; + serverState.value = r; _trackSelfReady(r, now); if (r.mediaId != _mediaId) { _mediaId = r.mediaId; - var url = roomMediaUrl(r); + var url = stateMediaUrl(r); if ((url == null || url.isEmpty) && sync.token.isNotEmpty) { final name = (r.media?['name'] as String?) ?? ''; if (name.isNotEmpty) url = _fileUrlFor(name); @@ -325,7 +329,7 @@ class SessionCoordinator { ]; } if (allowed && !_wasPlayAllowed && r.transport == 'playing') { - final edge = roomPositionAt(r, now); + final edge = serverPositionAt(r, now); actions = [ for (final a in actions) if (a is! TransportAction) a, SeekAction(edge), @@ -347,7 +351,7 @@ class SessionCoordinator { /// 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(RoomBroadcast r, DateTime now) { + void _trackSelfReady(StateBroadcast r, DateTime now) { final self = selfMember.peek(); if (self == null) return; if (r.mediaId != _mediaId) _restoringReady = false; @@ -364,7 +368,7 @@ class SessionCoordinator { } } - void _updateStarving(RoomBroadcast r, double bufferedAhead) { + void _updateStarving(StateBroadcast r, double bufferedAhead) { if (r.transport == 'playing') { starving.value = null; return; @@ -418,7 +422,7 @@ class SessionCoordinator { if (kind != null) { var readyIndex = 0; if (f.action == 'ready') { - readyIndex = room.peek()?.members.where((m) => m.ready).length ?? 0; + readyIndex = serverState.peek()?.members.where((m) => m.ready).length ?? 0; } unawaited(sfx.play(SfxCue(kind, readyIndex: readyIndex))); } @@ -469,7 +473,7 @@ class SessionCoordinator { final now = DateTime.now().toUtc(); toasts.value = toasts.peek().where((t) => now.difference(t.at).inSeconds < 4).toList(); - final r = room.peek(); + final r = serverState.peek(); if (sync.token.isNotEmpty && r != null && _playerReady) { sync.sendReport( position: playback.positionSecs, @@ -481,10 +485,10 @@ class SessionCoordinator { } if (!_manualClose && kickReason.peek() == null && - _lastRoomAt != null && - now.difference(_lastRoomAt!).inSeconds >= 5) { - if (room.peek() != null) { - room.value = null; + _lastStateAt != null && + now.difference(_lastStateAt!).inSeconds >= 5) { + if (serverState.peek() != null) { + serverState.value = null; starving.value = null; _wasPlayAllowed = false; _mediaId = -1; @@ -502,7 +506,7 @@ class SessionCoordinator { // PLAYING: the ready flag gates nothing then, and watching counts as // activity so a pause after a long watch starts a fresh idle clock. final event = _idle.tick(now, - playing: room.peek()?.transport == 'playing', + playing: serverState.peek()?.transport == 'playing', selfReady: selfMember.peek()?.ready ?? false); switch (event) { case IdleEvent.warn: @@ -523,7 +527,7 @@ class SessionCoordinator { } } - void _detectUserSeek(RoomBroadcast r) { + void _detectUserSeek(StateBroadcast r) { if (_openUrl == null) { _lastPos = playback.positionSecs; return; @@ -533,7 +537,7 @@ class SessionCoordinator { final jumped = (pos - _lastPos).abs(); _lastPos = pos; if (jumped <= 2.0) return; - if ((pos - roomPositionAt(r, now)).abs() < 2.0) return; + if ((pos - serverPositionAt(r, now)).abs() < 2.0) return; final recentSync = _pendingSeekAt != null && now.difference(_pendingSeekAt!).inMilliseconds < 1000; if (recentSync) return; @@ -541,7 +545,7 @@ class SessionCoordinator { sync.sendSeek(pos); } - Future _checkSeekLanded(RoomBroadcast r) async { + Future _checkSeekLanded(StateBroadcast r) async { final target = _pendingSeekTarget; final at = _pendingSeekAt; if (target == null || at == null) return; @@ -589,10 +593,10 @@ class SessionCoordinator { } void nudge(double deltaSecs) { - final r = room.peek(); + final r = serverState.peek(); if (r == null || _openUrl == null) return; final target = - roomPositionAt(r, DateTime.now().toUtc()) + deltaSecs; + serverPositionAt(r, DateTime.now().toUtc()) + deltaSecs; if (target.isNaN || target < 0) return; markInput(); sync.sendSeek(target); @@ -651,12 +655,13 @@ class SessionCoordinator { secret: params.secret, name: params.name, avatar: params.avatar, + hostKey: params.hostKey, ); // Whoever bumped the epoch also closed (or replaced) the socket. if (stale()) return; reconnecting.value = false; _everConnected = true; - _lastRoomAt = DateTime.now().toUtc(); + _lastStateAt = DateTime.now().toUtc(); health.value = const HealthStatus(HealthSeverity.ok, 'connected'); final url = welcome.mediaUrl; if (url != null && url.isNotEmpty && url != _openUrl) { @@ -743,7 +748,7 @@ class SessionCoordinator { await sync.disconnect(); } catch (_) {} if (!silent) { - room.value = null; + serverState.value = null; starving.value = null; health.value = const HealthStatus(HealthSeverity.info, 'idle'); } @@ -751,7 +756,7 @@ class SessionCoordinator { Future dispose() async { await leave(silent: true); - await _roomSub?.cancel(); + await _stateSub?.cancel(); await _feedSub?.cancel(); await _kickSub?.cancel(); await _errSub?.cancel(); diff --git a/apps/desktop/lib/src/state/server_isolate.dart b/apps/desktop/lib/src/state/server_isolate.dart new file mode 100644 index 0000000..7740e2b --- /dev/null +++ b/apps/desktop/lib/src/state/server_isolate.dart @@ -0,0 +1,148 @@ +import 'dart:async'; +import 'dart:isolate'; + +import 'package:flutter/foundation.dart' show visibleForTesting; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; + +Future _serverMain((Map, SendPort) args) => + serve(args.$1, events: args.$2); + +/// The hosted server, running [serve] on its own isolate. Its log lines +/// go to [onLog]; if the isolate dies without being asked to stop, +/// [onCrash] gets the cause once. +class ServerIsolate { + final Isolate _isolate; + final SendPort _stop; + final int port; + final ReceivePort _events; + final ReceivePort _errors; + final ReceivePort _exit; + final Completer _exited = Completer(); + bool _stopping = false; + String? _lastError; + void Function(String error)? _onCrash; + + ServerIsolate._(this._isolate, this._stop, this.port, this._events, + this._errors, this._exit); + + @visibleForTesting + Isolate get isolate => _isolate; + + bool get isRunning => !_exited.isCompleted; + + /// Spawns the server and waits until it listens. Throws a + /// [StateError] with the server's message if it cannot start. + static Future start( + AppConfig config, { + required void Function(String line) onLog, + required void Function(String error) onCrash, + Duration timeout = const Duration(seconds: 15), + }) async { + final events = ReceivePort('server events'); + final errors = ReceivePort('server errors'); + final exit = ReceivePort('server exit'); + final startup = Completer<(int, SendPort)>(); + String? lastError; + var exited = false; + ServerIsolate? self; + + void fail(String message) { + if (!startup.isCompleted) startup.completeError(StateError(message)); + } + + events.listen((msg) { + if (msg is! Map) return; + switch (msg['t']) { + case 'log': + final err = msg['error']; + onLog('${msg['level']} ${msg['logger']}: ${msg['message']}' + '${err == null ? '' : ' ($err)'}'); + case 'ready': + if (!startup.isCompleted) { + startup.complete((msg['port'] as int, msg['stop'] as SendPort)); + } + case 'error': + fail('${msg['message']}'); + } + }); + // errorsAreFatal is off: a stray async error is logged and the server + // keeps going, as it did in-process. Only a dead isolate is a crash. + errors.listen((msg) { + final text = msg is List && msg.isNotEmpty ? '${msg.first}' : '$msg'; + lastError = text; + self?._lastError = text; + onLog('SEVERE uncaught: $text'); + }); + exit.listen((_) { + exited = true; + fail(lastError ?? 'server exited before it was ready'); + self?._onExit(); + }); + + final Isolate isolate; + try { + isolate = await Isolate.spawn( + _serverMain, (config.toMap(), events.sendPort), + errorsAreFatal: false, + onError: errors.sendPort, + onExit: exit.sendPort, + debugName: 'server'); + } catch (_) { + events.close(); + errors.close(); + exit.close(); + rethrow; + } + final (int, SendPort) ready; + try { + ready = await startup.future.timeout(timeout); + } catch (e) { + if (!exited) isolate.kill(priority: Isolate.immediate); + events.close(); + errors.close(); + exit.close(); + if (e is TimeoutException) { + throw StateError('server did not start within ${timeout.inSeconds} s'); + } + rethrow; + } + final s = ServerIsolate._(isolate, ready.$2, ready.$1, events, errors, exit) + .._onCrash = onCrash + .._lastError = lastError; + self = s; + if (exited) { + // Died between ready and here: a failed start, not a crash. + s._stopping = true; + s._onExit(); + throw StateError(lastError ?? 'server exited right after starting'); + } + return s; + } + + void _onExit() { + if (_exited.isCompleted) return; + _exited.complete(); + _events.close(); + _errors.close(); + _exit.close(); + final crash = _onCrash; + _onCrash = null; + if (!_stopping) crash?.call(_lastError ?? 'the server stopped unexpectedly'); + } + + /// Asks the server to close its sockets and exit; kills the isolate if it + /// has not exited within [timeout]. + Future stop({Duration timeout = const Duration(seconds: 3)}) async { + if (_exited.isCompleted) return; + _stopping = true; + _stop.send({'t': 'stop'}); + try { + await _exited.future.timeout(timeout); + } on TimeoutException { + _isolate.kill(priority: Isolate.immediate); + await _exited.future.timeout(const Duration(seconds: 2), + onTimeout: _onExit); + } + } +} diff --git a/apps/desktop/lib/src/state/session_state.dart b/apps/desktop/lib/src/state/session_state.dart index ad4b310..c15319d 100644 --- a/apps/desktop/lib/src/state/session_state.dart +++ b/apps/desktop/lib/src/state/session_state.dart @@ -1,16 +1,17 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:math'; import 'package:flutter/widgets.dart'; import 'package:signals/signals_flutter.dart'; import 'package:sync_relay_core/core.dart'; -import 'package:sync_relay_server/server.dart'; import '../sfx/sfx.dart'; import '../session/health.dart'; import '../session/session_coordinator.dart'; import '../widgets/avatar_pipeline.dart'; import 'connect_prefs.dart'; +import 'server_isolate.dart'; enum AppRoute { home, @@ -128,6 +129,21 @@ List migrateRecents( return [RecentServer(hosted: false, host: host.trim(), port: port, secret: secret)]; } +/// The hosted server died during setup; shown on the Host card. +class ServerCrashed implements Exception { + final String cause; + const ServerCrashed(this.cause); + @override + String toString() => 'server crashed: $cause'; +} + +/// A fresh per-hosting key: the app's own client sends it in hello to be +/// recognised as the (unkickable) host. +String newHostKey() { + final r = Random.secure(); + return base64Url.encode([for (var i = 0; i < 24; i++) r.nextInt(256)]); +} + class SessionState { static final SessionState instance = SessionState._(); SessionState._(); @@ -135,11 +151,17 @@ class SessionState { final route = signal(AppRoute.home); final recents = signal>(const []); final status = signal(const HealthStatus(HealthSeverity.info, 'idle')); - final room = signal(null); final SfxService sfx = const NoopSfxService(); late final SessionCoordinator coordinator = SessionCoordinator(sfx: sfx); final lastForm = ConnectForm(); - RoomServer? _server; + /// Set when the hosted server died mid-session; Home shows it on the + /// Host card and clears it. + final serverCrash = signal(null); + ServerIsolate? _server; + String? _setupCrash; + + @visibleForTesting + ServerIsolate? get hostedServer => _server; void Function()? _stopKickWatch; // Bumped by host/join/cancelPending: a cancelled setup notices at its // next step and unwinds instead of connecting. @@ -208,18 +230,25 @@ class SessionState { const HealthStatus(HealthSeverity.info, 'starting server…'); await stopServer(); if (op != _op) throw const ConnectCancelled(); - final server = RoomServer(AppConfig( - port: params.port, - secret: params.secret, - name: 'room', - mediaPath: params.mediaPath, - )); + final hostKey = newHostKey(); + _setupCrash = null; + late final ServerIsolate server; try { - await server.start(); + server = await ServerIsolate.start( + AppConfig( + port: params.port, + secret: params.secret, + name: 'server', + mediaPath: params.mediaPath, + hostKey: hostKey, + ), + onLog: (line) => debugPrint('[server] $line'), + onCrash: (cause) => _onServerCrash(server, cause), + ); } catch (e, st) { debugPrint('server start failed: $e\n$st'); status.value = HealthStatus( - HealthSeverity.error, 'cannot bind port ${params.port}: $e'); + HealthSeverity.error, 'cannot start server on port ${params.port}: $e'); rethrow; } if (op != _op) { @@ -228,20 +257,23 @@ class SessionState { } _server = server; status.value = - HealthStatus(HealthSeverity.ok, 'hosting on port ${server.boundPort}'); + HealthStatus(HealthSeverity.ok, 'hosting on port ${server.port}'); try { await coordinator.connect(ConnectParams( host: '127.0.0.1', - port: server.boundPort, + port: server.port, secret: params.secret, name: params.name, avatar: params.avatar, linkUrl: params.linkUrl, + hostKey: hostKey, )); if (op != _op) throw const ConnectCancelled(); - } catch (_) { + } catch (e) { // Own join failed or was cancelled: don't leave the port open. if (identical(_server, server)) await stopServer(); + final crash = _setupCrash; + if (crash != null && e is! ConnectCancelled) throw ServerCrashed(crash); rethrow; } _watchKick(); @@ -251,6 +283,21 @@ class SessionState { } } + /// The server isolate died without being stopped. In a session: show it + /// and leave. During setup the own join fails and [host] reports it. + void _onServerCrash(ServerIsolate server, String cause) { + if (!identical(_server, server)) return; + final message = 'server crashed: $cause'; + debugPrint('[server] $message'); + if (route.peek() != AppRoute.session) { + _setupCrash = cause; + return; + } + coordinator.health.value = HealthStatus(HealthSeverity.error, message); + serverCrash.value = message; + unawaited(leave().catchError((Object e) => debugPrint('leave after crash failed: $e'))); + } + Future join(ConnectParams params, ConnectForm form) async { final op = ++_op; _rememberForm(form); @@ -305,15 +352,6 @@ class SessionState { ..linkUrl = form.linkUrl; } - Future startHeadless(List args) async { - int port = 8080; - for (var i = 0; i < args.length - 1; i++) { - if (args[i] == '--port') port = int.tryParse(args[i + 1]) ?? port; - } - _server = RoomServer(AppConfig(port: port, secret: '', name: 'room')); - await _server!.start(); - } - Future stopServer() async { _stopKickWatch?.call(); _stopKickWatch = null; diff --git a/apps/desktop/lib/src/sync/room_diff.dart b/apps/desktop/lib/src/sync/state_diff.dart similarity index 62% rename from apps/desktop/lib/src/sync/room_diff.dart rename to apps/desktop/lib/src/sync/state_diff.dart index 2f5105f..9cc5c21 100644 --- a/apps/desktop/lib/src/sync/room_diff.dart +++ b/apps/desktop/lib/src/sync/state_diff.dart @@ -19,9 +19,9 @@ class RateAction extends PlayerAction { const RateAction(this.rate); } -double roomPositionAt(RoomBroadcast room, DateTime now) { - if (room.transport != 'playing') return room.position; - return room.position + now.difference(room.serverTime).inMilliseconds / 1000.0; +double serverPositionAt(StateBroadcast state, DateTime now) { + if (state.transport != 'playing') return state.position; + return state.position + now.difference(state.serverTime).inMilliseconds / 1000.0; } bool seekLanded({required double ownPosition, required double target}) { @@ -31,9 +31,9 @@ bool seekLanded({required double ownPosition, required double target}) { return true; } -bool localPlayAllowed(RoomBroadcast room, String memberId) { +bool localPlayAllowed(StateBroadcast state, String memberId) { if (memberId.isEmpty) return false; - for (final m in room.members) { + for (final m in state.members) { if (m.id == memberId) return m.inSet && m.ready; } return false; @@ -45,7 +45,7 @@ bool localPlayAllowed(RoomBroadcast room, String memberId) { bool shouldReassertReady(MemberView? self) => self != null && self.ready && !self.inSet && !self.stale; -class RoomDiff { +class StateDiff { static const debounceWindow = Duration(seconds: 2); int _lastEventId = -1; @@ -57,14 +57,14 @@ class RoomDiff { bool get slowing => _slowing; - List diff(RoomBroadcast room, double ownPosition, DateTime now) { + List diff(StateBroadcast state, double ownPosition, DateTime now) { final out = []; - final roomPos = roomPositionAt(room, now); + final serverPos = serverPositionAt(state, now); - if (room.mediaId != _lastMediaId && _lastMediaId != -1) { - _lastEventId = room.eventId; - _lastTransport = room.transport; - _lastMediaId = room.mediaId; + if (state.mediaId != _lastMediaId && _lastMediaId != -1) { + _lastEventId = state.eventId; + _lastTransport = state.transport; + _lastMediaId = state.mediaId; _slowing = false; _deviatingSince = null; out.add(const SeekAction(0)); @@ -74,39 +74,39 @@ class RoomDiff { } if (_lastEventId == -1) { - _lastEventId = room.eventId; - _lastTransport = room.transport; - _lastMediaId = room.mediaId; + _lastEventId = state.eventId; + _lastTransport = state.transport; + _lastMediaId = state.mediaId; _deviatingSince = null; - out.add(SeekAction(roomPos)); - out.add(TransportAction(room.transport == 'playing')); + out.add(SeekAction(serverPos)); + out.add(TransportAction(state.transport == 'playing')); return out; } - if (room.eventId != _lastEventId) { - _lastEventId = room.eventId; + if (state.eventId != _lastEventId) { + _lastEventId = state.eventId; _slowing = false; _deviatingSince = null; out.add(const RateAction(1.0)); - out.add(SeekAction(roomPos)); - if (room.transport != _lastTransport) { - _lastTransport = room.transport; - out.add(TransportAction(room.transport == 'playing')); + out.add(SeekAction(serverPos)); + if (state.transport != _lastTransport) { + _lastTransport = state.transport; + out.add(TransportAction(state.transport == 'playing')); } return out; } - if (room.transport != _lastTransport) { - _lastTransport = room.transport; + if (state.transport != _lastTransport) { + _lastTransport = state.transport; _slowing = false; _deviatingSince = null; out.add(const RateAction(1.0)); - out.add(TransportAction(room.transport == 'playing')); - out.add(SeekAction(roomPos)); + out.add(TransportAction(state.transport == 'playing')); + out.add(SeekAction(serverPos)); return out; } - final drift = ownPosition - roomPos; + final drift = ownPosition - serverPos; if (_slowing) { if (drift.abs() < 0.1) { _slowing = false; @@ -114,10 +114,10 @@ class RoomDiff { } return out; } - switch (correctionFor(ownPosition: ownPosition, roomPosition: roomPos)) { + switch (correctionFor(ownPosition: ownPosition, serverPosition: serverPos)) { case Correction.seek: - if (_deviatingEvent != room.eventId || _deviatingSince == null) { - _deviatingEvent = room.eventId; + if (_deviatingEvent != state.eventId || _deviatingSince == null) { + _deviatingEvent = state.eventId; _deviatingSince = now; break; } @@ -125,7 +125,7 @@ class RoomDiff { break; } _deviatingSince = null; - out.add(SeekAction(roomPos)); + out.add(SeekAction(serverPos)); case Correction.slowDown: _deviatingSince = null; _slowing = true; diff --git a/apps/desktop/lib/src/sync/sync_client.dart b/apps/desktop/lib/src/sync/sync_client.dart index d065605..e958fd2 100644 --- a/apps/desktop/lib/src/sync/sync_client.dart +++ b/apps/desktop/lib/src/sync/sync_client.dart @@ -9,13 +9,13 @@ enum SyncStatus { idle, connecting, open, kicked, closed, error } class SyncClient { WebSocketChannel? _channel; StreamSubscription? _sub; - final StreamController _rooms = StreamController.broadcast(); + final StreamController _states = StreamController.broadcast(); final StreamController _feeds = StreamController.broadcast(); final StreamController _kicked = StreamController.broadcast(); final StreamController _errors = StreamController.broadcast(); final StreamController _status = StreamController.broadcast(); - Stream get rooms => _rooms.stream; + Stream get states => _states.stream; Stream get feeds => _feeds.stream; Stream get kicked => _kicked.stream; Stream get errors => _errors.stream; @@ -24,7 +24,7 @@ class SyncClient { String token = ''; String memberId = ''; String? mediaUrl; - RoomBroadcast? lastRoom; + StateBroadcast? lastState; Completer? _pendingWelcome; Future connect({ @@ -33,6 +33,7 @@ class SyncClient { required String name, String avatar = '', HelloMedia? media, + String hostKey = '', }) async { await disconnect(); _status.add(SyncStatus.connecting); @@ -61,6 +62,7 @@ class SyncClient { secret: secret, user: HelloUser(name: name, avatar: avatar), media: media ?? const HelloMedia(name: '', size: 0, duration: 0), + hostKey: hostKey, ); ch.sink.add(jsonEncode(hello.toJson())); _pendingWelcome = welcomeCompleter; @@ -74,8 +76,8 @@ class SyncClient { mediaUrl = welcome.mediaUrl; if (welcome.state != null) { try { - lastRoom = RoomBroadcast.fromJson(welcome.state!); - _rooms.add(lastRoom!); + lastState = StateBroadcast.fromJson(welcome.state!); + _states.add(lastState!); } catch (_) {} } _status.add(SyncStatus.open); @@ -113,11 +115,11 @@ class SyncClient { return; } switch (t) { - case 'room': + case 'state': try { - final room = RoomBroadcast.fromJson(decoded); - lastRoom = room; - _rooms.add(room); + final state = StateBroadcast.fromJson(decoded); + lastState = state; + _states.add(state); } catch (_) {} case 'feed': try { @@ -219,12 +221,12 @@ class SyncClient { token = ''; memberId = ''; mediaUrl = null; - lastRoom = null; + lastState = null; } Future dispose() async { await disconnect(); - await _rooms.close(); + await _states.close(); await _feeds.close(); await _kicked.close(); await _errors.close(); diff --git a/apps/desktop/test/coord_presence_test.dart b/apps/desktop/test/coord_presence_test.dart index 7732a3c..eaa1982 100644 --- a/apps/desktop/test/coord_presence_test.dart +++ b/apps/desktop/test/coord_presence_test.dart @@ -25,11 +25,11 @@ Future _poll( void main() { test( - 'dead server clears room state and reconnects instead of freezing', + 'dead server clears server state and reconnects instead of freezing', timeout: const Timeout(Duration(minutes: 2)), () async { MediaKit.ensureInitialized(); - final server = RoomServer( + final server = SyncServer( const AppConfig(port: 0, secret: 'presence', name: 'presence')); await server.start(); final coord = SessionCoordinator(); @@ -41,7 +41,7 @@ void main() { name: 'watcher', )); await coord.playback.player.setVolume(0); - await _poll(() => coord.room.peek() != null, 'room state to arrive'); + await _poll(() => coord.serverState.peek() != null, 'server state to arrive'); expect(coord.reconnecting.peek(), isFalse); await server.stop(); @@ -50,8 +50,8 @@ void main() { 'reconnect loop to engage', ); await _poll( - () => coord.room.peek() == null, - 'stale room (ghost members) to clear', + () => coord.serverState.peek() == null, + 'stale state (ghost members) to clear', ); coord.cancelReconnect(); expect(coord.reconnecting.peek(), isFalse); @@ -66,12 +66,12 @@ void main() { timeout: const Timeout(Duration(minutes: 2)), () async { MediaKit.ensureInitialized(); - final first = RoomServer( + final first = SyncServer( const AppConfig(port: 0, secret: 'race', name: 'race')); await first.start(); final port = first.boundPort; final coord = SessionCoordinator(); - RoomServer? second; + SyncServer? second; final observer = SyncClient(); try { await coord.connect(ConnectParams( @@ -79,7 +79,7 @@ void main() { await coord.playback.player.setVolume(0); await first.stop(); await _poll(() => coord.reconnecting.peek(), 'reconnect to engage'); - second = RoomServer( + second = SyncServer( AppConfig(port: port, secret: 'race', name: 'race')); await second.start(); var joins = 0; @@ -93,7 +93,7 @@ void main() { // Within the 1.5 s delay: the first attempt must die with Cancel. coord.cancelReconnect(); coord.retry(); - await _poll(() => !coord.reconnecting.peek() && coord.room.peek() != null, + await _poll(() => !coord.reconnecting.peek() && coord.serverState.peek() != null, 'reconnect to land'); await Future.delayed(const Duration(seconds: 4)); await sub.cancel(); @@ -116,13 +116,13 @@ void main() { timeout: const Timeout(Duration(minutes: 2)), () async { MediaKit.ensureInitialized(); - final first = RoomServer( + final first = SyncServer( const AppConfig(port: 0, secret: 'force', name: 'force')); await first.start(); final port = first.boundPort; final coord = SessionCoordinator(); final friend = SyncClient(); - RoomServer? second; + SyncServer? second; try { await coord.connect(ConnectParams( host: '127.0.0.1', port: port, secret: 'force', name: 'watcher')); @@ -138,7 +138,7 @@ void main() { await friend.disconnect(); await first.stop(); await _poll(() => coord.reconnecting.peek(), 'reconnect to engage'); - second = RoomServer( + second = SyncServer( AppConfig(port: port, secret: 'force', name: 'force')); await second.start(); await _poll( @@ -158,7 +158,7 @@ void main() { timeout: const Timeout(Duration(minutes: 1)), () async { MediaKit.ensureInitialized(); - final server = RoomServer( + final server = SyncServer( const AppConfig(port: 0, secret: '', name: 'cancel')); await server.start(); final coord = SessionCoordinator(); @@ -185,7 +185,8 @@ void main() { timeout: const Timeout(Duration(minutes: 1)), () async { MediaKit.ensureInitialized(); - Hive.init('/home/foxyz/.claude/jobs/4898998e/tmp/hive_coord_test'); + Hive.init( + '${Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path}/hive_coord_test'); final probe = await ServerSocket.bind(InternetAddress.anyIPv4, 0); final port = probe.port; await probe.close(); diff --git a/apps/desktop/test/server_isolate_test.dart b/apps/desktop/test/server_isolate_test.dart new file mode 100644 index 0000000..be1fd36 --- /dev/null +++ b/apps/desktop/test/server_isolate_test.dart @@ -0,0 +1,165 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:hive_ce/hive.dart'; +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:watch_together/src/session/session_coordinator.dart'; +import 'package:watch_together/src/state/server_isolate.dart'; +import 'package:watch_together/src/state/session_state.dart'; + +Future _poll( + FutureOr Function() cond, + String what, { + Duration timeout = const Duration(seconds: 15), +}) 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'); +} + +/// A bare WebSocket viewer: no player, just the wire. +class _Raw { + final WebSocket ws; + final msgs = >[]; + _Raw(this.ws) { + ws.listen((m) => msgs.add(Map.from(jsonDecode('$m') as Map))); + } + + static Future<_Raw> join(int port, String name, {String secret = ''}) async { + final r = _Raw(await WebSocket.connect('ws://127.0.0.1:$port/sync')); + r.ws.add(jsonEncode(HelloMessage( + proto: RelayConstants.protoVersion, + secret: secret, + user: HelloUser(name: name), + media: const HelloMedia(name: '', size: 0, duration: 0)) + .toJson())); + await r.next('welcome'); + return r; + } + + Future> next(String type, {bool Function(Map)? where}) async { + late Map found; + await _poll(() { + for (final m in msgs) { + if (m['t'] == type && (where == null || where(m))) { + found = m; + msgs.remove(m); + return true; + } + } + return false; + }, '$type from server'); + return found; + } +} + +Future _freePort() async { + final s = await ServerSocket.bind(InternetAddress.anyIPv4, 0); + final port = s.port; + await s.close(); + return port; +} + +void main() { + test('server isolate: logs, raw join, crash reported, stop frees port', + timeout: const Timeout(Duration(minutes: 1)), () async { + final logs = []; + final crashes = []; + final server = await ServerIsolate.start( + const AppConfig(port: 0, secret: '', name: 'iso'), + onLog: logs.add, + onCrash: crashes.add); + final viewer = await _Raw.join(server.port, 'viewer'); + await viewer.next('state'); + await _poll(() => logs.any((l) => l.contains('join viewer')), 'join log'); + await viewer.ws.close(); + server.isolate.kill(priority: Isolate.immediate); + await _poll(() => crashes.isNotEmpty, 'crash callback'); + expect(server.isRunning, isFalse); + await server.stop(); // no-op on a dead isolate + + final clean = await ServerIsolate.start( + const AppConfig(port: 0, secret: '', name: 'iso'), + onLog: (_) {}, + onCrash: crashes.add); + final port = clean.port; + await clean.stop(); + expect(crashes, hasLength(1), reason: 'a requested stop is not a crash'); + final rebind = await ServerSocket.bind(InternetAddress.anyIPv4, port); + await rebind.close(); + }); + + test('bind failure surfaces as a start error', + timeout: const Timeout(Duration(seconds: 30)), () async { + final blocker = await ServerSocket.bind(InternetAddress.anyIPv4, 0); + try { + await expectLater( + ServerIsolate.start(AppConfig(port: blocker.port, secret: '', name: 'x'), + onLog: (_) {}, onCrash: (_) {}), + throwsA(isA())); + } finally { + await blocker.close(); + } + }); + + test('host path: isolate server, host unkickable, crash leaves the session', + timeout: const Timeout(Duration(minutes: 2)), () async { + MediaKit.ensureInitialized(); + Hive.init( + '${Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path}/hive_isolate_test'); + final session = SessionState.instance; + final port = await _freePort(); + _Raw? friend; + try { + await session.host( + ConnectParams(host: '', port: port, secret: 'pw', name: 'host'), + ConnectForm()); + expect(session.route.peek(), AppRoute.session); + await session.coordinator.playback.player.setVolume(0); + + friend = await _Raw.join(port, 'friend', secret: 'pw'); + final state = await friend.next('state', + where: (m) => (m['members'] as List).any((x) => (x as Map)['name'] == 'host')); + final hostId = ((state['members'] as List) + .cast() + .firstWhere((m) => m['name'] == 'host'))['id']; + friend.ws.add(jsonEncode(KickRequest(target: '$hostId').toJson())); + final err = await friend.next('err'); + expect(err['message'], "can't kick the host"); + + // The host can kick the friend (plain loopback peers are not hosts). + final friendId = ((state['members'] as List) + .cast() + .firstWhere((m) => m['name'] == 'friend'))['id']; + session.coordinator.kick('$friendId'); + await friend.next('kicked'); + + session.hostedServer!.isolate.kill(priority: Isolate.immediate); + await _poll(() => session.route.peek() == AppRoute.home, 'leave after crash'); + expect(session.serverCrash.peek(), startsWith('server crashed')); + expect(session.hostedServer, isNull); + // The killed isolate's socket stays bound in-process; hosting again on + // the same port must still work. + session.serverCrash.value = null; + await session.host( + ConnectParams(host: '', port: port, secret: 'pw', name: 'host'), + ConnectForm()); + expect(session.route.peek(), AppRoute.session); + final again = await _Raw.join(port, 'again', secret: 'pw'); + await again.ws.close(); + await session.leave(); + expect(session.hostedServer, isNull); + } finally { + await friend?.ws.close(); + await session.coordinator.leave(silent: true); + await session.stopServer(); + } + }); +} diff --git a/apps/desktop/test/session_logic_test.dart b/apps/desktop/test/session_logic_test.dart index 8319d04..4a7a11a 100644 --- a/apps/desktop/test/session_logic_test.dart +++ b/apps/desktop/test/session_logic_test.dart @@ -2,7 +2,7 @@ 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/room_diff.dart'; +import 'package:watch_together/src/sync/state_diff.dart'; MemberView _self({bool ready = true, bool inSet = true, bool stale = false}) => MemberView( diff --git a/apps/desktop/test/room_diff_test.dart b/apps/desktop/test/state_diff_test.dart similarity index 66% rename from apps/desktop/test/room_diff_test.dart rename to apps/desktop/test/state_diff_test.dart index 767e540..c81b08b 100644 --- a/apps/desktop/test/room_diff_test.dart +++ b/apps/desktop/test/state_diff_test.dart @@ -1,15 +1,15 @@ import 'package:flutter_test/flutter_test.dart'; import 'package:sync_relay_core/core.dart'; -import 'package:watch_together/src/sync/room_diff.dart'; +import 'package:watch_together/src/sync/state_diff.dart'; -RoomBroadcast _room({ +StateBroadcast _state({ int eventId = 0, String transport = 'held', double position = 10.0, int mediaId = 1, }) { final now = DateTime.now().toUtc(); - return RoomBroadcast( + return StateBroadcast( eventId: eventId, transport: transport, position: position, @@ -19,10 +19,10 @@ RoomBroadcast _room({ ); } -RoomBroadcast _roomWithMembers(List members, +StateBroadcast _stateWithMembers(List members, {String transport = 'playing'}) { final now = DateTime.now().toUtc(); - return RoomBroadcast( + return StateBroadcast( eventId: 3, transport: transport, position: 10.0, @@ -33,50 +33,50 @@ RoomBroadcast _roomWithMembers(List members, } void main() { - test('initial diff seeks to room edge and sets transport', () { - final diff = RoomDiff(); - final actions = diff.diff(_room(), 0, DateTime.now().toUtc()); + test('initial diff seeks to server edge and sets transport', () { + final diff = StateDiff(); + final actions = diff.diff(_state(), 0, DateTime.now().toUtc()); expect(actions.whereType().single.position, closeTo(10.0, 1.0)); expect(actions.whereType().single.playing, isFalse); }); test('same event small drift does nothing', () { - final diff = RoomDiff(); + final diff = StateDiff(); final now = DateTime.now().toUtc(); - diff.diff(_room(), 0, now); - final actions = diff.diff(_room(), 10.5, now); + diff.diff(_state(), 0, now); + final actions = diff.diff(_state(), 10.5, now); expect(actions, isEmpty); }); test('ahead drift latches slowdown, resets inside 0.1', () { - final diff = RoomDiff(); + final diff = StateDiff(); final now = DateTime.now().toUtc(); - diff.diff(_room(), 0, now); - var actions = diff.diff(_room(), 13.0, now); + diff.diff(_state(), 0, now); + var actions = diff.diff(_state(), 13.0, now); expect(actions.whereType().single.rate, 0.95); - actions = diff.diff(_room(), 12.0, now); + actions = diff.diff(_state(), 12.0, now); expect(actions, isEmpty); - actions = diff.diff(_room(), 10.05, now); + actions = diff.diff(_state(), 10.05, now); expect(actions.whereType().single.rate, 1.0); }); test('eventId jump seeks and resets rate', () { - final diff = RoomDiff(); + final diff = StateDiff(); final now = DateTime.now().toUtc(); - diff.diff(_room(), 0, now); - diff.diff(_room(), 13.0, now); + diff.diff(_state(), 0, now); + diff.diff(_state(), 13.0, now); final actions = - diff.diff(_room(eventId: 1, position: 90.0), 13.0, now); + diff.diff(_state(eventId: 1, position: 90.0), 13.0, now); expect(actions.whereType().single.position, closeTo(90.0, 1.0)); expect(actions.whereType().single.rate, 1.0); }); test('media swap seeks zero and pauses', () { - final diff = RoomDiff(); + final diff = StateDiff(); final now = DateTime.now().toUtc(); - diff.diff(_room(transport: 'playing'), 10.0, now); + diff.diff(_state(transport: 'playing'), 10.0, now); final actions = diff.diff( - _room(eventId: 2, transport: 'held', position: 0, mediaId: 2), + _state(eventId: 2, transport: 'held', position: 0, mediaId: 2), 50.0, now); expect(actions.whereType().single.position, 0); @@ -84,25 +84,25 @@ void main() { }); group('seek debounce while playing', () { - RoomBroadcast playing({int eventId = 5, double position = 100.0}) => _room( + StateBroadcast playing({int eventId = 5, double position = 100.0}) => _state( eventId: eventId, transport: 'playing', position: position); test('fresh big drift is swallowed first, seeks after window', () { - final diff = RoomDiff(); + final diff = StateDiff(); final t0 = DateTime.now().toUtc(); diff.diff(playing(), 100.0, t0); var actions = diff.diff(playing(), 120.0, t0); expect(actions.whereType(), isEmpty, reason: 'user scrub must not be yanked back before detector runs'); actions = diff.diff( - playing(), 120.0, t0.add(RoomDiff.debounceWindow + const Duration(milliseconds: 100))); + playing(), 120.0, t0.add(StateDiff.debounceWindow + const Duration(milliseconds: 100))); expect(actions.whereType().single.position, closeTo(102.0, 3.0), reason: 'genuine drift still converges after the window'); }); test('new eventId seeks immediately (echo of promoted seek)', () { - final diff = RoomDiff(); + final diff = StateDiff(); final t0 = DateTime.now().toUtc(); diff.diff(playing(), 100.0, t0); diff.diff(playing(), 120.0, t0); @@ -112,13 +112,13 @@ void main() { }); test('held transport debounces the same way (detector promotes)', () { - final diff = RoomDiff(); + final diff = StateDiff(); final t0 = DateTime.now().toUtc(); - diff.diff(_room(), 10.0, t0); - var actions = diff.diff(_room(), 60.0, t0); + diff.diff(_state(), 10.0, t0); + var actions = diff.diff(_state(), 60.0, t0); expect(actions.whereType(), isEmpty); - actions = diff.diff(_room(), 60.0, - t0.add(RoomDiff.debounceWindow + const Duration(milliseconds: 100))); + actions = diff.diff(_state(), 60.0, + t0.add(StateDiff.debounceWindow + const Duration(milliseconds: 100))); expect(actions.whereType().single.position, closeTo(10.0, 3.0)); }); @@ -148,12 +148,12 @@ void main() { stale: false); test('only in-set ready members may follow transport to playing', () { - final room = _roomWithMembers([ready, catchingUp, unreadyInSet]); - expect(localPlayAllowed(room, 'm1'), isTrue); - expect(localPlayAllowed(room, 'm2'), isFalse); - expect(localPlayAllowed(room, 'm3'), isFalse); - expect(localPlayAllowed(room, 'nope'), isFalse); - expect(localPlayAllowed(room, ''), isFalse); + final state = _stateWithMembers([ready, catchingUp, unreadyInSet]); + expect(localPlayAllowed(state, 'm1'), isTrue); + expect(localPlayAllowed(state, 'm2'), isFalse); + expect(localPlayAllowed(state, 'm3'), isFalse); + expect(localPlayAllowed(state, 'nope'), isFalse); + expect(localPlayAllowed(state, ''), isFalse); }); }); } diff --git a/apps/desktop/test/sync_playback_e2e_test.dart b/apps/desktop/test/sync_playback_e2e_test.dart index 4593b25..ee04371 100644 --- a/apps/desktop/test/sync_playback_e2e_test.dart +++ b/apps/desktop/test/sync_playback_e2e_test.dart @@ -37,10 +37,10 @@ void main() { timeout: const Timeout(Duration(minutes: 4)), () async { MediaKit.ensureInitialized(); - final server = RoomServer(const AppConfig( + final server = SyncServer(const AppConfig( port: _port, secret: _secret, - name: 'room', + name: 'server', mediaPath: _mediaPath, )); await server.start(); @@ -67,10 +67,10 @@ void main() { reason: 'media must open paused, not autoplay'); expect(viewer.playback.player.state.playing, isFalse); - await _poll(() => host.room.peek() != null, 'host room state'); + await _poll(() => host.serverState.peek() != null, 'host server state'); host.toggleReady(); await Future.delayed(const Duration(seconds: 5)); - expect(host.room.peek()?.transport, 'held', + expect(host.serverState.peek()?.transport, 'held', reason: 'one Space with company present must not solo-play'); expect(host.playback.player.state.playing, isFalse); expect(viewer.playback.player.state.playing, isFalse); @@ -78,7 +78,7 @@ void main() { viewer.toggleReady(); await _poll( () => - host.room.peek()?.transport == 'playing' && + host.serverState.peek()?.transport == 'playing' && viewer.playback.player.state.playing && host.playback.player.state.playing, 'both players playing once quorum readies', diff --git a/apps/desktop/test/video_reload_e2e_test.dart b/apps/desktop/test/video_reload_e2e_test.dart index 978184c..3b2bd30 100644 --- a/apps/desktop/test/video_reload_e2e_test.dart +++ b/apps/desktop/test/video_reload_e2e_test.dart @@ -5,7 +5,7 @@ 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/room_diff.dart'; +import 'package:watch_together/src/sync/state_diff.dart'; const _mediaPath = '/home/foxyz/Videos/chiyo.mp4'; @@ -29,7 +29,7 @@ void main() { timeout: const Timeout(Duration(minutes: 2)), () async { MediaKit.ensureInitialized(); - final server = RoomServer(const AppConfig( + final server = SyncServer(const AppConfig( port: 0, secret: 'reload', name: 'reload', mediaPath: _mediaPath)); await server.start(); final c = SessionCoordinator(); @@ -40,7 +40,7 @@ void main() { secret: 'reload', name: 'solo')); await c.playback.player.setVolume(0); - await _poll(() => c.room.peek() != null, 'room state'); + await _poll(() => c.serverState.peek() != null, 'server state'); c.toggleReady(); await _poll( () => c.playback.player.state.playing && c.playback.positionSecs > 2, @@ -54,14 +54,14 @@ void main() { expect(c.selfMember.peek()?.ready, isTrue); await _poll(() { - final r = c.room.peek(); + final r = c.serverState.peek(); if (r == null || !c.playback.player.state.playing) return false; - final want = roomPositionAt(r, DateTime.now().toUtc()); + final want = serverPositionAt(r, DateTime.now().toUtc()); return (c.playback.positionSecs - want).abs() < 1.5 && c.playback.positionSecs > 2; }, 'reloaded player playing at the synced position'); expect(c.selfMember.peek()?.ready, isTrue); - expect(c.room.peek()?.transport, 'playing'); + expect(c.serverState.peek()?.transport, 'playing'); await c.leave(); expect(c.playback.alive.peek(), isFalse); diff --git a/deploy/watch-together.service b/deploy/watch-together.service index 7d3f740..00be23a 100644 --- a/deploy/watch-together.service +++ b/deploy/watch-together.service @@ -1,10 +1,22 @@ +# Dedicated watch-together server (no display needed). +# Install: unpack the release tarball to /opt/watch-together (the unit only +# uses the watch-together-server binary in it), copy this file to +# /etc/systemd/system/, then `systemctl enable --now watch-together`. +# Flags: --port (default 8080), --secret (empty = open +# server, anyone who can reach the port can join), --name, --media , +# --advertise-host , --config . See README. +# Prefer --config with a root-only file over --secret here: unit files and +# process lists are readable by other users. + [Unit] -Description=watch-together headless server -After=network.target +Description=watch-together dedicated server +After=network-online.target +Wants=network-online.target [Service] -ExecStart=/opt/watch-together/server --port 8080 +ExecStart=/opt/watch-together/watch-together-server --port 8080 --secret change-me Restart=on-failure +DynamicUser=true NoNewPrivileges=true [Install] diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 702399a..474f4a0 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -7,40 +7,56 @@ Plain overview first: [HOW-IT-WORKS.md](HOW-IT-WORKS.md). Wire format: | Path | What | Rule | |---|---|---| -| `packages/core` | protocol messages, sync engine (`RoomEngine`), constants, config, Range parsing, logging | pure Dart, no Flutter | -| `packages/server` | Relic HTTP + WebSocket server (`RoomServer`, `RoomHub`, `Session`, media endpoint), `bin/server.dart` | pure Dart, no Flutter | -| `apps/desktop` | Flutter app: UI, `SyncClient`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), updater | the only shipped binary | +| `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging | pure Dart, no Flutter | +| `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | +| `apps/desktop` | Flutter app: UI, `SyncClient`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, updater | ships with `watch-together-server` | | `tool` | harnesses: `range_logger`, `shaper`, `fake_viewer`, `e2e_check` | dev only | -| `deploy` | systemd unit for the standalone server (stale until the dedicated server exists) | | +| `deploy` | systemd unit for the dedicated server | | Guard: `packages/core/test/flutter_free_test.dart` fails if anything under `core/lib` imports Flutter, `dart:ui` or `signals_flutter`; CI runs analyze + tests for core and server on every push. Server has no such test but -the same rule; it is checked by CI building without Flutter. The client +the same rule; CI checks it with `dart compile exe` of `bin/server.dart`. The client WebSocket (`web_socket_channel`) lives in `apps/desktop`; Relic has no client. ## Server lifecycle -Today (in-process): `SessionState.host()` creates a `RoomServer`, starts it -on the chosen port (bound to all IPv4 interfaces), then the app joins it over -`127.0.0.1` like any other client. Leaving or closing the window stops it. -`RoomServer` runs a 1 s heartbeat: stale sweep, closed-socket sweep, silent -socket drop, room + feed broadcast. There is one session, id `default`. +One entrypoint: `serve(config, events:)` in `packages/server`. `config` is +an `AppConfig.toMap()` (primitives only, so it crosses isolates). It starts a +`SyncServer` (bound to all IPv4 interfaces) and reports on the `events` port: +`ready` (with the port and a `stop` SendPort), `error` (start failed), and +every log record as `log`. Sending `{'t':'stop'}` to the `stop` port closes +all sockets and returns, and the isolate exits. -`--headless` does not work: the Flutter binary needs a display and exits with -"cannot open display" on a headless box. `packages/server/bin/server.dart` is -the only working standalone server (flags `--port`, `--secret`, `--name`, -`--media`, `--advertise-host`, `--config`; default port 8080). +- **Hosting from the app**: `SessionState.host()` spawns an isolate running + `serve` (`ServerIsolate`), waits for `ready`, then joins it over + `127.0.0.1` like any other client. Server logs go to the app's log with a + `[server]` prefix. The isolate runs with `errorsAreFatal: false`: an + uncaught async error is logged and the server keeps going. If the isolate + dies anyway, the session shows "server crashed: ..." and leaves; Home + shows it on the Host card (during setup the host attempt fails with it). + Stop (leave, close, cancel during setup, a failed own join, being kicked) + sends `stop`, waits up to 3 s for exit, then kills the isolate. +- **Dedicated server**: `bin/server.dart` compiled with `dart compile exe` + and shipped next to the app as `watch-together-server`. It parses flags + (`--port`, `--secret`, `--name`, `--media`, `--advertise-host`, + `--config`; default port 8080) into the same map and calls `serve`; + SIGINT and SIGTERM stop it. A Flutter binary cannot run without a + display, which is why the server is its own build. -Planned: hosting from the app runs the server on a Dart isolate (ports and -sockets stay in the same process, UI stays responsive); the dedicated server -is a separate `dart compile exe` of `packages/server`. See [ROADMAP.md](ROADMAP.md). +`SyncServer` runs a 1 s heartbeat: stale sweep, closed-socket sweep, silent +socket drop, state + feed broadcast. There is one session, id `default`. Passwords: the default secret is empty, which means an open server (a warning is logged). Otherwise one shared secret, compared at hello. -The host is the member that connected from a loopback address; it cannot be -kicked. Kick = revoke token, send `kicked`, close; the -person may rejoin at once with a new token. + +Host: when hosting, the app generates a random `hostKey` per hosting, puts +it in the server config and sends it in its own `hello`. The member whose +key matches is the host and cannot be kicked. The address a connection +comes from plays no part (so a reverse proxy or tunnel on the same machine +does not make everyone the host). The dedicated server has no key, so it +has no host. Kick = revoke token, send `kicked`, close; the person may +rejoin at once with a new token. ## Sync model @@ -63,20 +79,20 @@ Loosely modeled on Syncplay, with a different readiness design. 4 s buffered or stale flips the server to HELD (feed action `pause` by `server`). Recovery resumes by itself when the gate passes. - **Seek**: allowed from any member in any transport; the server writes the - position and bumps `eventId`; flags are untouched. Clients diff each `room` - broadcast by `eventId` (`RoomDiff`): new event means seek to the computed - room position, transport change means play/pause. Between events: drift + position and bumps `eventId`; 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 room) and sent as a `seek` intent; a retry - (`_checkSeekLanded`) re-applies a room seek that has not landed after 1.5 s. + 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. - **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. 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. -- **Client policies**: no room broadcast for 5 s means connection lost, clear +- **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 restored from the server's state. Cancel during setup aborts the join. @@ -96,14 +112,14 @@ Additive wire changes only; bump `proto` and PROTOCOL.md in the same commit. Two sources, same sync. - **Link**: the media field holds an http(s) URL. It travels as `media.url` - in `swap` and `room`, and as `mediaUrl` in `welcome`. Each app plays it + in `swap` and `state`, and as `mediaUrl` in `welcome`. Each app plays it directly; the server is only the clock. -- **Host local file**: `RoomServer.setLocalFile` / `mediaPath` makes the file +- **Host local file**: `SyncServer.setLocalFile` / `mediaPath` makes the file the session media. `GET /m///` serves it with a Range table, gated by the viewer's token. The server builds the URL per viewer (`http://:/m/default//`), so tokens never reach other viewers. `` is the authority the client dialled; `advertiseHost` - (default empty) only overrides it when set explicitly. When a `room` + (default empty) only overrides it when set explicitly. When a `state` broadcast has no `media.url`, the client builds the URL itself from the host it connected to. Bytes stream file to socket and are counted as streamed (`bytesServed`); nothing is stored. The host plays through loopback like @@ -146,10 +162,10 @@ 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 -`RoomDiff` actions or the play gate (`localPlayAllowed`: self in-set and -ready). A room seek within 0.5 s of the current position is skipped. `leave()` +`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()` stops the player (otherwise audio keeps playing). On media change -(`mediaId` bump) the coordinator opens the new URL and `RoomDiff` seeks to 0, +(`mediaId` bump) the coordinator opens the new URL and `StateDiff` seeks to 0, holds and resets rate. Frozen picture: `FrameWatchdog` samples about once a second. It hashes a @@ -186,7 +202,7 @@ stalls silently. The app has a `HealthStatus` signal (severity + message) 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`, `room`, +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. @@ -217,9 +233,12 @@ this repo, written by `tool/release.sh`. - No roles, friends trust model: simplest thing that fits the use. - Blank password = open server: the owner chooses between open and private. - The host cannot be kicked: the server dies with the host anyway. -- Server ships inside the app (and, planned, as a separate compiled server): - a Flutter binary cannot run headless, so a dedicated server needs its own - build of `packages/server`. +- Server runs inside the app on an isolate, and ships as a separate compiled + server: a Flutter binary cannot run headless, so a dedicated server needs + its own build of `packages/server`. An isolate keeps the UI responsive and + shares the process (sockets, file paths) without a second executable. +- Host by key, not by address: loopback detection broke behind a + same-machine reverse proxy or tunnel. - Server owns the clock; reports never move it: avoids ping-pong between clients. - Quorum gate: a lone viewer must be able to watch; two or more must start together. - Relic for the server, `web_socket_channel` for the client; packages are @@ -228,5 +247,5 @@ this repo, written by `tool/release.sh`. (e.g. a reverse proxy) is a documentation item. - One session per server, id `default`: the map is keyed for later but unused. - No server storage: config and logs only; media never touches the server disk. -- Avatars ride in the 1 Hz `room` broadcast (cap 256 KiB base64 per avatar): +- Avatars ride in the 1 Hz `state` broadcast (cap 256 KiB base64 per avatar): simple, costs host uplink; hash-based serving is on the roadmap. diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index c209091..f9e86fe 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -12,11 +12,14 @@ mpv libraries come from `media_kit_libs_video`. cd apps/desktop && flutter pub get && flutter run -d linux cd apps/desktop && flutter build linux --release --dart-define=APP_VERSION= tool/package_linux.sh [--skip-build] [--noupload] # dist/*.tar.gz +cd packages/server && dart compile exe bin/server.dart -o watch-together-server ``` `APP_VERSION` defaults to `0.0.0-dev`; `package_linux.sh` stamps it from the desktop pubspec. Setting `DISCORD_WEBHOOK_URL` makes the packager post the -tarball to Discord (never commit it). +tarball to Discord (never commit it). The packager always compiles the +dedicated server (with the Flutter SDK's `dart`) into the tarball as +`watch-together-server`, also with `--skip-build`. ## Dev loop @@ -26,8 +29,10 @@ Run a server without the app, then point apps at it: dart run packages/server/bin/server.dart --port 8080 --secret x [--media FILE] ``` -Prints `READY port=... room=...` and stops on SIGINT. Then Join from one or -more apps at `127.0.0.1:8080`. Or just Host from the app. mpv env override: +This is the same `serve()` the app runs when hosting and the same code as +the dedicated binary. Prints `READY port=... name=...` and stops on SIGINT +or SIGTERM. Then Join from one or more apps at `127.0.0.1:8080`. Or just +Host from the app. mpv env override: `SYNC_RELAY_HWDEC` (wins over the Settings choice). `SYNC_RELAY_VO` has no effect with embedded video: media_kit forces `vo=libmpv`. @@ -43,13 +48,18 @@ and `e2e_check` (simulated viewers against a live server). - `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 what you changed. -- CI (`.gitea/workflows/ci.yml`): analyze + test for core and server only. - Desktop tests are not in CI. +- CI (`.gitea/workflows/ci.yml`): analyze + test for core and server, plus + `dart compile exe` of the dedicated server (fails if Flutter creeps into + `packages/server`). Desktop tests are not in CI. +- `packages/server/test/serve_isolate_test.dart` runs `serve` in a real + isolate; `apps/desktop/test/server_isolate_test.dart` covers the app's + host path (isolate server, raw WebSocket join, crash, rehost). +- Desktop tests that write files use `WT_TEST_TMP` (default: system temp). Traps: - Never pump the session screen or media_kit `Video` in `flutter test`: the runner hangs (kill the `flutter_tester` by PID). Test pure logic - (`RoomDiff`, coordinator without widgets) instead. + (`StateDiff`, coordinator without widgets) instead. - Never depend on a server-side WebSocket `onDone`; assert presence with deadline polling. - `sub.asFuture()` can resolve before `onDone`; poll with a deadline. @@ -73,9 +83,24 @@ Tokens come from the environment (`RELEASE_TOKEN`, or Actions-provided updater skips the check when absent. Workflows live in `.gitea/workflows` (the forge reads `.gitea` before `.github`). +## Debugging the hosted server + +When hosting, the server runs on its own isolate (`ServerIsolate`, +`apps/desktop/lib/src/state/server_isolate.dart`), in the same process. +Breakpoints in `packages/server` work under `flutter run`: the debugger +attaches to spawned isolates too (it shows as `server`). Its log lines are +printed with a `[server]` prefix. Hot reload reloads the server isolate's +code too, but its state (members, clock, sockets) survives until hosting +restarts; changes to startup code (`SyncServer.start`, routes) need a new +Host. Without the app: `dart run packages/server/bin/server.dart`. + ## Gotchas - Relic gives Range as a typed `RangeHeader`; convert from unit + first range. +- A killed isolate (`Isolate.kill`) does not close its listening socket: + the port stays bound in the process until exit. `SyncServer` binds with + `shared: true` so a new hosted server in the same process can take it + over; a normal stop closes it. - Relic server-side `onDone` does not fire for dead peers. Presence is the `isClosed` sweep + `onError` forget + `lastSeen` timeouts. - `web_socket_channel` 3.x: `channel.ready` errors are unhandled on refused diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index c4a1397..f43f8e8 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -33,8 +33,8 @@ A five-minute tour. Details: [ARCHITECTURE.md](ARCHITECTURE.md), wire format: - **Host from the app**: the app you click Host in also runs the server. Simplest. The server lives as long as that app. -- **Dedicated server**: a separate server program on an always-on box. - Planned, see [ROADMAP.md](ROADMAP.md). +- **Dedicated server**: `watch-together-server`, a separate program in the + release tarball, for an always-on box. See the README. Friends join either way with the Join card on Home. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 7e5a1f4..054d0fe 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,31 +1,32 @@ -# Protocol (proto = 4) +# Protocol (proto = 5) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: -`packages/core/lib/src/protocol.dart` and `packages/server/lib/src/room_hub.dart`. +`packages/core/lib/src/protocol.dart` and `packages/server/lib/src/server_hub.dart`. Any wire change bumps `RelayConstants.protoVersion` and edits this file in -the same commit. Changes are additive. +the same commit. Changes are additive where possible; a client and server +with different `proto` refuse each other at `hello` (`proto mismatch`). ## Transport - WebSocket at `/sync`, JSON text frames, each an object with a `t` field. - The first client frame must be `hello` within 10 s. Server pings every 5 s. -- Match incoming frames by `t`, never by order: `welcome` can race `room`. +- Match incoming frames by `t`, never by order: `welcome` can race `state`. - Failures: server sends `{"t":"err","message":...}` and closes. Messages: `bad hello: ...`, `proto mismatch: got X, want Y`, `wrong secret`, `rejected: ` (e.g. avatar too large), `bad message: ...` (a frame with a non-finite number, e.g. NaN position), `bad swap media`, `swap needs a url or the hosted file` (empty swap). -- `GET /health` returns JSON (`ok`, `room`, `proto`, `viewers`, `bytesServed`, +- `GET /health` returns JSON (`ok`, `name`, `proto`, `viewers`, `bytesServed`, `uptimeSecs`; unauthenticated). `GET /m///` serves the host's file (Range table in ARCHITECTURE.md). ## Client to server ``` -{"t":"hello","proto":4,"secret":"","user":{"name":"","avatar":""}, - "media":{"name":"","size":0,"duration":0,"url":null}} +{"t":"hello","proto":5,"secret":"","user":{"name":"","avatar":""}, + "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"..."} {"t":"ready","ready":true,"source":"manual"} source: manual | auto {"t":"force","target":"all"|"","ready":true} {"t":"seek","position":123.4} @@ -37,8 +38,12 @@ the same commit. Changes are additive. Field notes: - `hello.secret`: the server password; a server with an empty secret is open. - The member connecting from a loopback address is the host. `hello.media` is carried but not used by the server today. +- `hello.hostKey` (optional, omitted when empty): the hosting app + generates a random key per hosting, gives it to its server and sends it + 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. - `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. @@ -49,8 +54,8 @@ Field notes: ## Server to client ``` -{"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","state":{}} -{"t":"room","eventId":7,"transport":"playing"|"held","position":12.3, +{"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","state":{}} +{"t":"state","eventId":7,"transport":"playing"|"held","position":12.3, "serverTime":"","mediaId":1, "media":{name,size,duration,url?}|absent, "members":[{"id","name","avatar","ready","inSet","buffered","stale"}]} @@ -64,8 +69,8 @@ Field notes: media URL. `mediaUrl` is omitted when there is no media; it is a built `/m/...` URL for a host file (host part = what the client dialled), otherwise the shared link. - `welcome.state` has the same shape as `room`. -- `room` is sent every second and on every change. `position` is the value at + `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`. `eventId` increases on seek, swap and transport changes. `mediaId` increases on swap. @@ -80,7 +85,7 @@ Field notes: ## Join and reconnect -`hello` then `welcome`, then the app opens `mediaUrl`, seeks to the room +`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. @@ -90,4 +95,4 @@ the first `ready`. A socket that dies loses its token; reconnect is a fresh Server-held queue (`queue-add`, `queue-skip`) would auto-advance like a swap. `RelayConstants.chunkBytes`, `creditWindowChunks`, `maxConcurrentOpen` and the `open`, `ack`, `cancel`, `eof`, `buffered`, `state`, `event` kinds in -`MessageKind` are unused leftovers. +`MessageKind` are unused leftovers (the `state` frame does not use the enum). diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 25af291..2f18d6c 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -6,17 +6,7 @@ off by deleting it in the same commit as the code. Wire changes also edit ## Backlog -1. **Isolate server + dedicated server.** Replace host detection by loopback - peer (breaks behind a same-machine reverse proxy or tunnel) with an explicit - host token. Run the server on a Dart isolate - when hosting from the app; build the dedicated server as a separate - `dart compile exe` of `packages/server` (a Flutter binary cannot be - headless); drop `--headless`; rename "room" to "server" in code - (`RoomServer`, `RoomHub`, `RoomBroadcast`, wire `t:'room'`, `room` in - `/health`, config defaults) with the proto bump that implies; refresh - `deploy/` and the packaging. - -2. **Sync fixes.** +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 @@ -34,22 +24,23 @@ off by deleting it in the same commit as the code. Wire changes also edit a workaround (remove it then). - Ghost member after an unclean drop until `dropAfter`. -3. **Rendering, then shaders.** media_kit falls back to software rendering +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). -4. **Windows build + CI.** There is no Windows runner. Try +3. **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 - Windows apply path exists but is untested. Desktop tests are not in CI - either. + Windows apply path exists but is untested. The tarball's + `watch-together-server` needs a Windows `dart compile exe` too. Desktop + tests are not in CI either. -5. **Jellyfin** (epic, below). +4. **Jellyfin** (epic, below). -6. **Smaller items** +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. @@ -63,8 +54,8 @@ off by deleting it in the same commit as the code. Wire changes also edit OOM, no stuck session, drift recovers. - Chat history pane (today: activity log only). - Avatar caching: serve avatar bytes once by hash instead of base64 in - every `room` broadcast. - - The `room` broadcast carries the full media URL, which can include an + every `state` broadcast. + - The `state` broadcast carries the full media URL, which can include an `api_key`. - TLS: document a reverse proxy in front of the server. - Playlist: `queue-add`, `queue-skip`, auto-advance like a swap. diff --git a/packages/core/lib/src/config.dart b/packages/core/lib/src/config.dart index e721961..4fa4c4c 100644 --- a/packages/core/lib/src/config.dart +++ b/packages/core/lib/src/config.dart @@ -15,6 +15,9 @@ class AppConfig { /// Optional explicit host for file-media URLs; empty means use the host the /// client dialled. final String advertiseHost; + /// Secret the hosting app's own client sends in hello to be marked host + /// (unkickable). Empty means no member is the host (dedicated server). + final String hostKey; const AppConfig({ required this.port, @@ -23,20 +26,22 @@ class AppConfig { this.mediaPath = '', this.logLevel = 'INFO', this.advertiseHost = '', + this.hostKey = '', }); factory AppConfig.defaults() => - const AppConfig(port: 8080, secret: '', name: 'room'); + const AppConfig(port: 8080, secret: '', name: 'server'); factory AppConfig.fromMap(Map map) { try { return AppConfig( port: (map['port'] as num?)?.toInt() ?? 8080, secret: (map['secret'] as String?) ?? '', - name: (map['name'] as String?) ?? 'room', + name: (map['name'] as String?) ?? 'server', mediaPath: (map['mediaPath'] as String?) ?? '', logLevel: (map['logLevel'] as String?) ?? 'INFO', advertiseHost: (map['advertiseHost'] as String?) ?? '', + hostKey: (map['hostKey'] as String?) ?? '', ); } catch (e) { throw ConfigException('invalid config map: $e'); @@ -50,6 +55,7 @@ class AppConfig { 'mediaPath': mediaPath, 'logLevel': logLevel, 'advertiseHost': advertiseHost, + 'hostKey': hostKey, }; static Future loadFile(String path) async { diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 104e970..8ad9db8 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 = 4; + static const int protoVersion = 5; static const int chunkBytes = 256 * 1024; static const int creditWindowChunks = 8; @@ -19,7 +19,7 @@ abstract class RelayConstants { // leaves wide headroom over any 128x128 pipeline PNG (raw bitmap // <= 64 KiB, worst-case base64 ~95 KiB chars) — deliberate over-size // for private-server use. PNG keeps transparency; JPEG did not. - // Trade: avatars ride in the 1 Hz room broadcast, so worst-case host + // Trade: avatars ride in the 1 Hz state broadcast, so worst-case host // uplink scales with this cap. Follow-up (not now): send avatar bytes // once, reference by hash afterwards. static const int maxAvatarBytes = 256 * 1024; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index b7d733a..b38cc49 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -67,7 +67,15 @@ class HelloMessage { final String secret; final HelloUser user; final HelloMedia media; - const HelloMessage({required this.proto, required this.secret, required this.user, required this.media}); + /// The hosting app's own client proves it is the host with this; empty + /// for everyone else. + final String hostKey; + const HelloMessage( + {required this.proto, + required this.secret, + required this.user, + required this.media, + this.hostKey = ''}); Map toJson() => { 't': 'hello', @@ -75,6 +83,7 @@ class HelloMessage { 'secret': secret, 'user': user.toJson(), 'media': media.toJson(), + if (hostKey.isNotEmpty) 'hostKey': hostKey, }; factory HelloMessage.fromJson(Map json) { @@ -87,16 +96,17 @@ class HelloMessage { secret: (json['secret'] as String?) ?? '', user: HelloUser.fromJson(Map.from(user)), media: HelloMedia.fromJson(Map.from(media)), + hostKey: (json['hostKey'] as String?) ?? '', ); } } -class RoomState { +class SyncState { final double position; final bool paused; final int mediaId; final DateTime serverTime; - const RoomState({required this.position, required this.paused, required this.mediaId, required this.serverTime}); + const SyncState({required this.position, required this.paused, required this.mediaId, required this.serverTime}); Map toJson() => { 'position': position, @@ -105,7 +115,7 @@ class RoomState { 'serverTime': serverTime.toIso8601String(), }; - factory RoomState.fromJson(Map json) => RoomState( + factory SyncState.fromJson(Map json) => SyncState( position: finiteDouble(json['position'], 'position'), paused: (json['paused'] as bool?) ?? true, mediaId: finiteInt(json['mediaId'], 'mediaId', fallback: 0), @@ -296,7 +306,7 @@ class MemberView { ); } -class RoomBroadcast { +class StateBroadcast { final int eventId; final String transport; final double position; @@ -304,7 +314,7 @@ class RoomBroadcast { final int mediaId; final Map? media; final List members; - const RoomBroadcast({ + const StateBroadcast({ required this.eventId, required this.transport, required this.position, @@ -315,7 +325,7 @@ class RoomBroadcast { }); Map toJson() => { - 't': 'room', + 't': 'state', 'eventId': eventId, 'transport': transport, 'position': position, @@ -325,11 +335,11 @@ class RoomBroadcast { 'members': members.map((m) => m.toJson()).toList(), }; - factory RoomBroadcast.fromJson(Map json) { - if (json['t'] != 'room') throw const ProtocolException('not a room broadcast'); + factory StateBroadcast.fromJson(Map json) { + if (json['t'] != 'state') throw const ProtocolException('not a state broadcast'); final members = json['members']; final media = json['media']; - return RoomBroadcast( + return StateBroadcast( eventId: finiteInt(json['eventId'], 'eventId', fallback: 0), transport: (json['transport'] as String?) ?? 'held', position: finiteDouble(json['position'], 'position'), diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index 657b8f5..76d1795 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -1,14 +1,14 @@ import 'constants.dart'; import 'protocol.dart'; -typedef RoomClock = DateTime Function(); +typedef SyncClock = DateTime Function(); enum Transport { playing, held } enum Correction { none, slowDown, seek } -Correction correctionFor({required double ownPosition, required double roomPosition}) { - final drift = ownPosition - roomPosition; +Correction correctionFor({required double ownPosition, required double serverPosition}) { + final drift = ownPosition - serverPosition; if (drift.abs() > RelayConstants.driftSeekThresholdSecs) return Correction.seek; if (drift > RelayConstants.slowDownThresholdSecs) return Correction.slowDown; return Correction.none; @@ -52,8 +52,8 @@ class EngineFeed { } } -class RoomEngine { - final RoomClock clock; +class SyncEngine { + final SyncClock clock; final Duration staleAfter; final Duration dropAfter; final EngineFeed feed = EngineFeed(); @@ -67,7 +67,7 @@ class RoomEngine { HelloMedia? media; int _nextId = 1; - RoomEngine({RoomClock? clock, Duration? staleAfter, Duration? dropAfter}) + SyncEngine({SyncClock? clock, Duration? staleAfter, Duration? dropAfter}) : clock = clock ?? (() => DateTime.now().toUtc()), staleAfter = staleAfter ?? const Duration(seconds: 15), dropAfter = dropAfter ?? const Duration(seconds: 30), @@ -249,7 +249,7 @@ class RoomEngine { } } - RoomBroadcast broadcast() => RoomBroadcast( + StateBroadcast broadcast() => StateBroadcast( eventId: eventId, transport: transport == Transport.playing ? 'playing' : 'held', position: position, diff --git a/packages/core/test/core_test.dart b/packages/core/test/core_test.dart index ceb204a..1006aa8 100644 --- a/packages/core/test/core_test.dart +++ b/packages/core/test/core_test.dart @@ -15,7 +15,7 @@ void main() { test('fromMap tolerates missing keys', () { final c = AppConfig.fromMap({}); expect(c.port, 8080); - expect(c.name, 'room'); + expect(c.name, 'server'); }); test('round-trips through map', () { @@ -24,8 +24,10 @@ void main() { secret: 's', name: 'n', mediaPath: 'm', - advertiseHost: '1.2.3.4'); + advertiseHost: '1.2.3.4', + hostKey: 'hk'); final back = AppConfig.fromMap(c.toMap()); + expect(back.hostKey, 'hk'); expect(back.port, 1234); expect(back.secret, 's'); expect(back.mediaPath, 'm'); @@ -64,15 +66,29 @@ void main() { final back = HelloMessage.fromJson(Map.from(msg.toJson())); expect(back.user.name, 'alice'); expect(back.media.size, 1500000000); + expect(back.hostKey, isEmpty); + expect(msg.toJson().containsKey('hostKey'), isFalse); + }); + + test('hello carries hostKey when set', () { + const msg = HelloMessage( + proto: 1, + secret: '', + user: HelloUser(name: 'host'), + media: HelloMedia(name: '', size: 0, duration: 0), + hostKey: 'hk', + ); + final back = HelloMessage.fromJson(Map.from(msg.toJson())); + expect(back.hostKey, 'hk'); }); test('hello rejects wrong type', () { expect(() => HelloMessage.fromJson({'t': 'bye'}), throwsA(isA())); }); - test('room state round-trips', () { - final s = RoomState(position: 12.5, paused: false, mediaId: 3, serverTime: DateTime.utc(2026, 1, 1)); - final back = RoomState.fromJson(Map.from(s.toJson())); + test('sync state round-trips', () { + final s = SyncState(position: 12.5, paused: false, mediaId: 3, serverTime: DateTime.utc(2026, 1, 1)); + final back = SyncState.fromJson(Map.from(s.toJson())); expect(back.position, 12.5); expect(back.paused, isFalse); expect(back.mediaId, 3); diff --git a/packages/core/test/sync_engine_test.dart b/packages/core/test/sync_engine_test.dart index 35f70db..a6cc3b1 100644 --- a/packages/core/test/sync_engine_test.dart +++ b/packages/core/test/sync_engine_test.dart @@ -5,12 +5,12 @@ import 'package:test/test.dart'; void main() { late DateTime t; - late RoomEngine engine; + late SyncEngine engine; DateTime clock() => t; setUp(() { t = DateTime.utc(2026, 1, 1); - engine = RoomEngine(clock: clock); + engine = SyncEngine(clock: clock); }); String join(String token, String name) => engine.applyJoin(token, name, ''); @@ -37,12 +37,12 @@ void main() { }); group('membership + gate', () { - test('join lands out-of-set, room stays held', () { + test('join lands out-of-set, playback stays held', () { join('a', 'alice'); expect(engine.transport, Transport.held); expect(engine.members['a']!.inSet, isFalse); - final room = engine.broadcast(); - expect(room.members.single.inSet, isFalse); + final state = engine.broadcast(); + expect(state.members.single.inSet, isFalse); }); test('single ready + buffered auto-plays', () { @@ -54,7 +54,7 @@ void main() { expect(engine.transport, Transport.playing); }); - test('late joiner does not halt a playing room', () { + test('late joiner does not halt playback', () { join('a', 'alice'); engine.applyReady('a', true); buffer('a', 4.0); @@ -77,7 +77,7 @@ void main() { expect(engine.members['b']!.inSet, isTrue); }); - test('disconnect removes from quorum, room resumes', () { + test('disconnect removes from quorum, playback resumes', () { join('a', 'alice'); join('b', 'bob'); engine.applyReady('a', true); @@ -218,7 +218,7 @@ void main() { expect(engine.members['a']!.ready, isTrue); }); - test('unpause uses room clock, laggards seek forward', () { + test('unpause uses server clock, laggards seek forward', () { join('a', 'alice'); join('b', 'bob'); engine.applyReady('a', true); @@ -355,29 +355,29 @@ void main() { group('tiered correction', () { test('big drift seeks either direction', () { - expect(correctionFor(ownPosition: 100, roomPosition: 0), Correction.seek); - expect(correctionFor(ownPosition: 0, roomPosition: 100), Correction.seek); + expect(correctionFor(ownPosition: 100, serverPosition: 0), Correction.seek); + expect(correctionFor(ownPosition: 0, serverPosition: 100), Correction.seek); }); test('ahead beyond 1.5s slows to 0.95x', () { - expect(correctionFor(ownPosition: 3, roomPosition: 0), Correction.slowDown); - expect(correctionFor(ownPosition: 1.6, roomPosition: 0), Correction.slowDown); + expect(correctionFor(ownPosition: 3, serverPosition: 0), Correction.slowDown); + expect(correctionFor(ownPosition: 1.6, serverPosition: 0), Correction.slowDown); }); test('deadzone does nothing', () { - expect(correctionFor(ownPosition: 1.0, roomPosition: 0), Correction.none); - expect(correctionFor(ownPosition: 0, roomPosition: 3.0), Correction.none); - expect(correctionFor(ownPosition: 5.1, roomPosition: 0), Correction.seek); - expect(correctionFor(ownPosition: 5.0, roomPosition: 0), Correction.slowDown); + expect(correctionFor(ownPosition: 1.0, serverPosition: 0), Correction.none); + expect(correctionFor(ownPosition: 0, serverPosition: 3.0), Correction.none); + expect(correctionFor(ownPosition: 5.1, serverPosition: 0), Correction.seek); + expect(correctionFor(ownPosition: 5.0, serverPosition: 0), Correction.slowDown); }); }); group('broadcast shape', () { - test('room JSON round-trips with member views', () { + test('state JSON round-trips with member views', () { join('a', 'alice'); - final room = engine.broadcast(); + final state = engine.broadcast(); final back = - RoomBroadcast.fromJson(Map.from(room.toJson())); + StateBroadcast.fromJson(Map.from(state.toJson())); expect(back.transport, 'held'); expect(back.members.single.name, 'alice'); expect(back.members.single.id, 'm1'); diff --git a/packages/server/bin/server.dart b/packages/server/bin/server.dart index b8b0350..6fd8b86 100644 --- a/packages/server/bin/server.dart +++ b/packages/server/bin/server.dart @@ -1,47 +1,91 @@ +import 'dart:async'; import 'dart:io'; +import 'dart:isolate'; import 'package:args/args.dart'; import 'package:logging/logging.dart' as pkg_logging; import 'package:sync_relay_core/core.dart'; import 'package:sync_relay_server/server.dart'; +/// Dedicated server: `dart compile exe` of this file is shipped as +/// `watch-together-server`. Runs [serve] until SIGINT or SIGTERM. Future main(List args) async { final parser = ArgParser() - ..addOption('port', abbr: 'p', defaultsTo: '8080') - ..addOption('secret', defaultsTo: '') - ..addOption('name', defaultsTo: 'room') - ..addOption('media', defaultsTo: '') - ..addOption('advertise-host', defaultsTo: '') - ..addOption('config') - ..addFlag('help', abbr: 'h', negatable: false); - final parsed = parser.parse(args); + ..addOption('port', abbr: 'p', defaultsTo: '8080', help: 'TCP port to listen on (all IPv4 interfaces).') + ..addOption('secret', defaultsTo: '', help: 'Shared password. Empty means an open server: anyone who can reach it can join.') + ..addOption('name', defaultsTo: 'server', help: 'Server name, reported by /health.') + ..addOption('media', defaultsTo: '', help: 'Local file to serve as the starting media.') + ..addOption('advertise-host', defaultsTo: '', help: 'Host put in file-media URLs instead of the one each client dialled.') + ..addOption('config', help: 'YAML or JSON config file (keys as in AppConfig); overrides the other flags.') + ..addFlag('help', abbr: 'h', negatable: false, help: 'Show this help.'); + final ArgResults parsed; + try { + parsed = parser.parse(args); + } on FormatException catch (e) { + stderr.writeln('${e.message}\n\n${parser.usage}'); + exitCode = 64; + return; + } if (parsed['help'] as bool) { // ignore: avoid_print - print(parser.usage); + print('watch-together-server (proto ${RelayConstants.protoVersion})\n\n${parser.usage}'); return; } initLogging(level: pkg_logging.Level.INFO); AppConfig config; final configPath = parsed['config'] as String?; - if (configPath != null) { - config = await AppConfig.loadFile(configPath); - } else { - config = AppConfig( - port: int.tryParse('${parsed['port']}') ?? 8080, - secret: '${parsed['secret']}', - name: '${parsed['name']}', - mediaPath: '${parsed['media']}', - advertiseHost: '${parsed['advertise-host']}', - ); + try { + if (configPath != null) { + config = await AppConfig.loadFile(configPath); + } else { + final port = int.tryParse('${parsed['port']}'); + if (port == null || port < 0 || port > 65535) { + throw ConfigException('bad --port: ${parsed['port']}'); + } + config = AppConfig( + port: port, + secret: '${parsed['secret']}', + name: '${parsed['name']}', + mediaPath: '${parsed['media']}', + advertiseHost: '${parsed['advertise-host']}', + ); + } + } on ConfigException catch (e) { + stderr.writeln(e.message); + exitCode = 64; + return; + } + if (config.secret.isEmpty) { + stderr.writeln( + 'WARNING: no --secret set: this is an open server, anyone who can reach port ${config.port} can join.'); } - final server = RoomServer(config); - await server.start(); - - // ignore: avoid_print - print('READY port=${server.boundPort} room=${config.name}'); - - await ProcessSignal.sigint.watch().first; - await server.stop(); + final events = ReceivePort('server events'); + final done = serve(config.toMap(), events: events.sendPort); + final signals = >[]; + await for (final msg in events) { + if (msg is! Map) continue; + if (msg['t'] == 'error') { + stderr.writeln('server failed to start: ${msg['message']}'); + exitCode = 1; + break; + } + if (msg['t'] == 'ready') { + final stop = msg['stop'] as SendPort; + // ignore: avoid_print + print('READY port=${msg['port']} name=${config.name}'); + void onSignal(ProcessSignal _) => stop.send({'t': 'stop'}); + signals.add(ProcessSignal.sigint.watch().listen(onSignal)); + if (!Platform.isWindows) { + signals.add(ProcessSignal.sigterm.watch().listen(onSignal)); + } + break; + } + } + await done; + events.close(); + for (final s in signals) { + await s.cancel(); + } } diff --git a/packages/server/lib/server.dart b/packages/server/lib/server.dart index 8b479f5..92640cf 100644 --- a/packages/server/lib/server.dart +++ b/packages/server/lib/server.dart @@ -1,4 +1,5 @@ export 'src/media_endpoint.dart'; -export 'src/room_hub.dart'; -export 'src/room_server.dart'; +export 'src/serve.dart'; +export 'src/server_hub.dart'; export 'src/session.dart'; +export 'src/sync_server.dart'; diff --git a/packages/server/lib/src/serve.dart b/packages/server/lib/src/serve.dart new file mode 100644 index 0000000..39b00b7 --- /dev/null +++ b/packages/server/lib/src/serve.dart @@ -0,0 +1,47 @@ +import 'dart:async'; +import 'dart:isolate'; + +import 'package:logging/logging.dart' as pkg_logging; +import 'package:sync_relay_core/core.dart'; + +import 'sync_server.dart'; + +/// The one server entrypoint: the dedicated binary and the app's hosting +/// isolate both run this. [config] holds primitives only (an +/// [AppConfig.toMap]) so it can cross an isolate boundary. +/// +/// On [events] it reports `{'t':'ready','port':p,'stop':SendPort}` or +/// `{'t':'error','message':m}`, and forwards log records as +/// `{'t':'log','level','logger','message'[,'error']}`. Sending +/// `{'t':'stop'}` to the `stop` port stops the server; the returned future +/// completes once it is down. Without [events] a start failure throws. +Future serve(Map config, {SendPort? events}) async { + StreamSubscription? logs; + if (events != null) { + logs = pkg_logging.Logger.root.onRecord.listen((r) => events.send({ + 't': 'log', + 'level': r.level.name, + 'logger': r.loggerName, + 'message': r.message, + if (r.error != null) 'error': '${r.error}', + })); + } + final SyncServer server; + try { + server = SyncServer(AppConfig.fromMap(config)); + await server.start(); + } catch (e) { + await logs?.cancel(); + if (events == null) rethrow; + events.send({'t': 'error', 'message': '$e'}); + return; + } + final control = ReceivePort('serve control'); + events?.send({'t': 'ready', 'port': server.boundPort, 'stop': control.sendPort}); + await for (final msg in control) { + if (msg is Map && msg['t'] == 'stop') break; + } + control.close(); + await server.stop(); + await logs?.cancel(); +} diff --git a/packages/server/lib/src/room_hub.dart b/packages/server/lib/src/server_hub.dart similarity index 95% rename from packages/server/lib/src/room_hub.dart rename to packages/server/lib/src/server_hub.dart index cccfd98..38df5c3 100644 --- a/packages/server/lib/src/room_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -7,20 +7,25 @@ import 'package:web_socket/web_socket.dart'; import 'session.dart'; -final _hubLog = createLogger('room_hub'); +final _hubLog = createLogger('server_hub'); -class RoomHub { +class ServerHub { final SessionMap sessions; final String expectedSecret; - final RoomEngine engine; + /// See [AppConfig.hostKey]; empty means nobody is the host. + final String hostKey; + final SyncEngine engine; final Map sockets = {}; - RoomHub(this.sessions, this.expectedSecret, {RoomEngine? engine}) - : engine = engine ?? RoomEngine(); + ServerHub(this.sessions, this.expectedSecret, + {SyncEngine? engine, this.hostKey = ''}) + : engine = engine ?? SyncEngine(); + + bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey; Session get session => sessions.getOrCreate('default', expectedSecret); - void broadcastRoom() { + void broadcastState() { final raw = jsonEncode(engine.broadcast().toJson()); for (final ws in sockets.values.toList()) { ws.trySendText(raw); @@ -81,7 +86,7 @@ class RoomHub { } void changed() { - broadcastRoom(); + broadcastState(); broadcastFeed(); } @@ -114,18 +119,13 @@ String sanitizeMediaName(String name) { return 'media'; } -bool _isLoopback(String address) => - address == '::1' || - address.startsWith('127.') || - address.startsWith('::ffff:127.'); - void wsErr(RelicWebSocket ws, String message) { ws.trySendText(jsonEncode({'t': 'err', 'message': message})); } void mountSyncRoute( RelicApp app, - RoomHub hub, + ServerHub hub, String Function(String sid, String token, SessionMedia? media, String host) urlBuilder, ) { @@ -134,7 +134,6 @@ void mountSyncRoute( final dialledHost = req.url.hasPort ? '${req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host}:${req.url.port}' : (req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host); - final fromLoopback = _isLoopback(req.connectionInfo.remote.address.toString()); return WebSocketUpgrade((final ws) async { ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs); final firstText = Completer(); @@ -221,7 +220,7 @@ void mountSyncRoute( final report = ReportMessage.fromJson(decoded); hub.engine.applyReport( token, report.position, report.buffered, report.eventId); - hub.broadcastRoom(); + hub.broadcastState(); case 'chat': final chat = ChatMessage.fromJson(decoded); hub.broadcastChat(viewer?.name ?? '?', chat.text); @@ -309,7 +308,7 @@ void mountSyncRoute( await close(); return; } - session.tokens[token]!.isHost = fromLoopback; + 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; diff --git a/packages/server/lib/src/session.dart b/packages/server/lib/src/session.dart index 4526b60..92cc668 100644 --- a/packages/server/lib/src/session.dart +++ b/packages/server/lib/src/session.dart @@ -10,7 +10,7 @@ class ViewerToken { final String name; final String avatar; bool kicked = false; - /// Connected from a loopback address: the host's own in-process client. + /// Sent the server's host key in hello: the hosting app's own client. bool isHost = false; int bytesServed = 0; DateTime lastSeen; diff --git a/packages/server/lib/src/room_server.dart b/packages/server/lib/src/sync_server.dart similarity index 83% rename from packages/server/lib/src/room_server.dart rename to packages/server/lib/src/sync_server.dart index ec8f23f..785407d 100644 --- a/packages/server/lib/src/room_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -7,22 +7,22 @@ import 'package:relic/relic.dart'; import 'package:sync_relay_core/core.dart'; import 'media_endpoint.dart'; -import 'room_hub.dart'; +import 'server_hub.dart'; import 'session.dart'; -class RoomServer { +class SyncServer { final AppConfig config; - final pkg_logging.Logger _log = createLogger('room_server'); + final pkg_logging.Logger _log = createLogger('sync_server'); final RelicApp _app = RelicApp(); final SessionMap sessions = SessionMap(); - late final RoomHub hub; + late final ServerHub hub; RelicServer? _server; DateTime? _startedAt; Timer? _heartbeat; bool _routesMounted = false; - RoomServer(this.config) { - hub = RoomHub(sessions, config.secret); + SyncServer(this.config) { + hub = ServerHub(sessions, config.secret, hostKey: config.hostKey); } bool get isRunning => _server != null; @@ -50,7 +50,7 @@ class RoomServer { return Response.ok( body: Body.fromString(jsonEncode({ 'ok': true, - 'room': config.name, + 'name': config.name, 'proto': RelayConstants.protoVersion, 'viewers': viewers, 'bytesServed': bytes, @@ -93,16 +93,21 @@ class RoomServer { _server = await _app.serve( address: InternetAddress.anyIPv4, port: config.port, + // A killed isolate never closes its listening socket, so the port stays + // bound in this process. Shared lets the next server in the same + // process bind it again (it then gets all connections); another + // process is still refused. + shared: true, ); _startedAt = DateTime.now(); _heartbeat = Timer.periodic(const Duration(seconds: 1), (_) { hub.engine.sweepStale(); hub.sweepClosedSockets(); hub.dropSilentSockets(); - hub.broadcastRoom(); + hub.broadcastState(); hub.broadcastFeed(); }); - _log.info('room "${config.name}" listening on $boundPort'); + _log.info('server "${config.name}" listening on $boundPort'); } Future stop() async { @@ -112,6 +117,6 @@ class RoomServer { await hub.closeAllSockets(); await _app.close(); _server = null; - _log.info('room "${config.name}" stopped'); + _log.info('server "${config.name}" stopped'); } } diff --git a/packages/server/pubspec.yaml b/packages/server/pubspec.yaml index 1dc1045..4d2fd78 100644 --- a/packages/server/pubspec.yaml +++ b/packages/server/pubspec.yaml @@ -1,5 +1,5 @@ name: sync_relay_server -description: Pure-Dart room + media server. Embeddable, no Flutter. +description: Pure-Dart sync + media server. Embeddable, no Flutter. publish_to: none version: 0.1.0 environment: diff --git a/packages/server/test/hardening_test.dart b/packages/server/test/hardening_test.dart index abd7e28..b97a88f 100644 --- a/packages/server/test/hardening_test.dart +++ b/packages/server/test/hardening_test.dart @@ -16,14 +16,15 @@ class _Peer { } static Future<_Peer> join(String host, int port, String name, - {String secret = ''}) async { + {String secret = '', String hostKey = ''}) async { final ws = await WebSocket.connect('ws://$host:$port/sync'); final p = _Peer(ws); ws.add(jsonEncode(HelloMessage( proto: RelayConstants.protoVersion, secret: secret, user: HelloUser(name: name), - media: const HelloMedia(name: '', size: 0, duration: 0)) + media: const HelloMedia(name: '', size: 0, duration: 0), + hostKey: hostKey) .toJson())); await p.next('welcome'); return p; @@ -57,13 +58,13 @@ Future _lan() async { } void main() { - late RoomServer server; + late SyncServer server; late File file; setUp(() async { file = File('${Directory.systemTemp.path}/hard_${DateTime.now().microsecondsSinceEpoch}.bin') ..writeAsBytesSync(List.filled(64, 1)); - server = RoomServer(AppConfig(port: 0, secret: '', name: 't', mediaPath: file.path)); + server = SyncServer(AppConfig(port: 0, secret: '', name: 't', mediaPath: file.path)); await server.start(); }); @@ -122,22 +123,37 @@ void main() { await a.ws.close(); }); - test('host (loopback) cannot be kicked; others can', () async { - final lan = await _lan(); - if (lan == null) { - markTestSkipped('no LAN address'); - return; - } - final host = await _Peer.join('127.0.0.1', server.boundPort, 'host'); - final friend = await _Peer.join(lan, server.boundPort, 'friend'); - final hostId = server.hub.engine.members.values.firstWhere((m) => m.name == 'host').id; - friend.send(jsonEncode(KickRequest(target: hostId).toJson())); + test('host (hostKey) cannot be kicked; loopback without it can', () async { + await server.stop(); + server = SyncServer(AppConfig(port: 0, secret: '', name: 't', hostKey: 'hk')); + await server.start(); + final host = + await _Peer.join('127.0.0.1', server.boundPort, 'host', hostKey: 'hk'); + final friend = await _Peer.join('127.0.0.1', server.boundPort, 'friend'); + final guesser = + await _Peer.join('127.0.0.1', server.boundPort, 'guesser', hostKey: 'nope'); + String idOf(String name) => + server.hub.engine.members.values.firstWhere((m) => m.name == name).id; + friend.send(jsonEncode(KickRequest(target: idOf('host')).toJson())); final err = await friend.next('err'); expect(err['message'], "can't kick the host"); expect(host.closed, isFalse); - await expectLater(server.hub.kickById(hostId, 'x'), throwsA(isA())); + await expectLater( + server.hub.kickById(idOf('host'), 'x'), throwsA(isA())); + // Same machine, no key (or a wrong one): an ordinary member. + host.send(jsonEncode(KickRequest(target: idOf('guesser')).toJson())); + await guesser.next('kicked'); + await server.hub.kickById(idOf('friend'), 'x'); + await friend.next('kicked'); await host.ws.close(); - await friend.ws.close(); + }); + + test('no hostKey configured: nobody is the host', () async { + final a = await _Peer.join('127.0.0.1', server.boundPort, 'a', hostKey: ''); + final b = await _Peer.join('127.0.0.1', server.boundPort, 'b', hostKey: 'x'); + expect(server.session.tokens.values.where((t) => t.isHost), isEmpty); + await a.ws.close(); + await b.ws.close(); }); test('empty secret accepts any hello secret', () async { diff --git a/packages/server/test/lifecycle_test.dart b/packages/server/test/lifecycle_test.dart index 9d85f8d..93a325b 100644 --- a/packages/server/test/lifecycle_test.dart +++ b/packages/server/test/lifecycle_test.dart @@ -60,10 +60,10 @@ Future _pollBool( void main() { group('presence lifecycle live', () { - late RoomServer server; + late SyncServer server; setUp(() async { - server = RoomServer( + server = SyncServer( const AppConfig(port: 0, secret: 'cycle', name: 'cycle')); await server.start(); }); diff --git a/packages/server/test/serve_isolate_test.dart b/packages/server/test/serve_isolate_test.dart new file mode 100644 index 0000000..35c4025 --- /dev/null +++ b/packages/server/test/serve_isolate_test.dart @@ -0,0 +1,70 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; + +Future _run((Map, SendPort) args) => + serve(args.$1, events: args.$2); + +void main() { + test('serve in a real isolate: ready, /health, stop, clean exit', () async { + final events = ReceivePort(); + final exit = ReceivePort(); + final errors = ReceivePort(); + final config = const AppConfig(port: 0, secret: '', name: 'iso', hostKey: 'hk').toMap(); + await Isolate.spawn(_run, (config, events.sendPort), + onExit: exit.sendPort, onError: errors.sendPort); + final crashes = []; + errors.listen(crashes.add); + final logs = []; + final ready = Completer(); + events.listen((m) { + if (m is! Map) return; + if (m['t'] == 'log') logs.add(m); + if (m['t'] == 'ready' && !ready.isCompleted) ready.complete(m); + if (m['t'] == 'error' && !ready.isCompleted) ready.completeError('${m['message']}'); + }); + final r = await ready.future.timeout(const Duration(seconds: 10)); + final port = r['port'] as int; + expect(port, greaterThan(0)); + + final http = HttpClient(); + final resp = await (await http.get('127.0.0.1', port, '/health')).close(); + final body = jsonDecode(await resp.transform(utf8.decoder).join()) as Map; + http.close(); + expect(body['ok'], isTrue); + expect(body['name'], 'iso'); + expect(body['proto'], RelayConstants.protoVersion); + expect(logs.any((l) => '${l['message']}'.contains('listening on $port')), isTrue); + + (r['stop'] as SendPort).send({'t': 'stop'}); + await exit.first.timeout(const Duration(seconds: 10)); + expect(crashes, isEmpty); + // The port is free again. + final again = await ServerSocket.bind(InternetAddress.anyIPv4, port); + await again.close(); + events.close(); + errors.close(); + }); + + test('serve reports a bind failure as an error event', () async { + final blocker = await ServerSocket.bind(InternetAddress.anyIPv4, 0); + final events = ReceivePort(); + final exit = ReceivePort(); + await Isolate.spawn( + _run, + (AppConfig(port: blocker.port, secret: '', name: 'x').toMap(), + events.sendPort), + onExit: exit.sendPort); + final first = await events + .firstWhere((m) => m is Map && m['t'] != 'log') + .timeout(const Duration(seconds: 10)) as Map; + expect(first['t'], 'error'); + await exit.first.timeout(const Duration(seconds: 10)); + await blocker.close(); + }); +} diff --git a/packages/server/test/room_auth_test.dart b/packages/server/test/server_auth_test.dart similarity index 94% rename from packages/server/test/room_auth_test.dart rename to packages/server/test/server_auth_test.dart index 07888bd..1456143 100644 --- a/packages/server/test/room_auth_test.dart +++ b/packages/server/test/server_auth_test.dart @@ -8,28 +8,19 @@ import 'package:sync_relay_server/server.dart'; import 'package:test/test.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; -Future _lanAddress() async { - for (final i in await NetworkInterface.list(type: InternetAddressType.IPv4)) { - for (final a in i.addresses) { - if (!a.isLoopback) return a.address; - } - } - return null; -} - void main() { - group('room auth live', () { + group('server auth live', () { late File file; late Uint8List bytes; - late RoomServer server; + late SyncServer server; late HttpClient http; setUp(() async { bytes = Uint8List.fromList(List.generate(8192, (i) => i % 251)); file = File( - '${Directory.systemTemp.path}/room_test_${DateTime.now().microsecondsSinceEpoch}.bin'); + '${Directory.systemTemp.path}/server_test_${DateTime.now().microsecondsSinceEpoch}.bin'); await file.writeAsBytes(bytes); - server = RoomServer(AppConfig( + server = SyncServer(AppConfig( port: 0, secret: 'shh', name: 'test', mediaPath: file.path)); await server.start(); http = HttpClient(); @@ -71,14 +62,14 @@ void main() { } } - test('health reports room, proto, viewers, bytes', () async { + test('health reports name, proto, viewers, bytes', () async { final req = await http.get('127.0.0.1', server.boundPort, '/health'); final resp = await req.close(); expect(resp.statusCode, 200); final body = await resp.transform(utf8.decoder).join(); final decoded = jsonDecode(body) as Map; expect(decoded['ok'], isTrue); - expect(decoded['room'], 'test'); + expect(decoded['name'], 'test'); expect(decoded['proto'], RelayConstants.protoVersion); expect(decoded['viewers'], 0); expect(decoded['bytesServed'], 0); @@ -247,12 +238,6 @@ void main() { }); test('kick via socket: peer kicked, unknown target errors', () async { - final lan = await _lanAddress(); - if (lan == null) { - markTestSkipped('no non-loopback IPv4 to act as a non-host peer'); - return; - } - // Loopback peers are the host and unkickable, so bob dials the LAN address. Future<(WebSocketChannel, String, String, List)> join(String who, [String host = '127.0.0.1']) async { final ch = WebSocketChannel.connect( @@ -304,7 +289,7 @@ void main() { } final (a, _, _, messagesA) = await join('alice'); - final (b, tokenB, idB, messagesB) = await join('bob', lan); + final (b, tokenB, idB, messagesB) = await join('bob'); try { a.sink.add(jsonEncode( KickRequest(target: idB, reason: 'afk?').toJson())); diff --git a/packages/server/test/server_test.dart b/packages/server/test/server_test.dart index e70fdf7..0076dcd 100644 --- a/packages/server/test/server_test.dart +++ b/packages/server/test/server_test.dart @@ -23,7 +23,7 @@ void main() { }); test('server starts, answers /health, stops cleanly', () async { - final server = RoomServer(const AppConfig(port: 0, secret: 't', name: 'test')); + final server = SyncServer(const AppConfig(port: 0, secret: 't', name: 'test')); await server.start(); expect(server.isRunning, isTrue); final port = server.boundPort; @@ -45,7 +45,7 @@ void main() { }); test('unknown path returns 404', () async { - final server = RoomServer(const AppConfig(port: 0, secret: 't', name: 'test')); + final server = SyncServer(const AppConfig(port: 0, secret: 't', name: 'test')); await server.start(); final client = HttpClient(); try { diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index 9c746b3..261e05c 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 bumped to 4 for readiness + room fan-out', () { - expect(RelayConstants.protoVersion, 4); + test('proto 5: state broadcast + hello hostKey', () { + expect(RelayConstants.protoVersion, 5); }); test('welcome round-trips with token + member id', () { diff --git a/tool/e2e_check.dart b/tool/e2e_check.dart index 449f7a1..434206d 100644 --- a/tool/e2e_check.dart +++ b/tool/e2e_check.dart @@ -11,7 +11,7 @@ void check(String name, bool ok, [String detail = '']) { } class FakeViewer { - final RoomEngine engine; + final SyncEngine engine; final String token; double ownPosition; double buffered; @@ -37,7 +37,7 @@ class FakeViewer { Future main(List args) async { var simTime = DateTime.utc(2026, 1, 1); - final engine = RoomEngine(clock: () => simTime); + final engine = SyncEngine(clock: () => simTime); engine.media = const HelloMedia(name: 'Movie.mkv', size: 4000000000, duration: 7200.0); void step([int seconds = 1]) => simTime = simTime.add(Duration(seconds: seconds)); @@ -49,7 +49,7 @@ Future main(List args) async { engine.applyReady(t, true); engine.applyReport(t, 0, 6.0, engine.eventId); } - check('room playing after all ready+buffered', engine.transport == Transport.playing); + check('playing after all ready+buffered', engine.transport == Transport.playing); final alice = FakeViewer(engine, 'a'); final bob = FakeViewer(engine, 'b'); @@ -83,7 +83,7 @@ Future main(List args) async { } engine.applyJoin('d', 'dan', ''); - check('mid-play join does not halt room', engine.transport == Transport.playing); + check('mid-play join does not halt playback', engine.transport == Transport.playing); final dan = FakeViewer(engine, 'd'); dan.ownPosition = engine.positionAt(); dan.tick(); @@ -101,7 +101,7 @@ Future main(List args) async { } engine.applyReady('b', false); final frozen = engine.positionAt(); - check('unready pauses room', engine.transport == Transport.held); + check('unready pauses playback', engine.transport == Transport.held); for (var s = 0; s < 30; s++) { step(); engine.applyReport('a', frozen, 6.0, engine.eventId); @@ -122,7 +122,7 @@ Future main(List args) async { engine.applyReady('c', true); engine.applyReport('c', engine.positionAt(), 6.0, engine.eventId); } - check('post-flap room converges to playing', engine.transport == Transport.playing); + check('post-flap converges to playing', engine.transport == Transport.playing); engine.applySwap('a', const HelloMedia(name: 'Ep02.mkv', size: 100, duration: 1200.0)); check('swap holds + resets edge', engine.transport == Transport.held && engine.positionAt() == 0); diff --git a/tool/package_linux.sh b/tool/package_linux.sh index a3af49d..6418246 100755 --- a/tool/package_linux.sh +++ b/tool/package_linux.sh @@ -4,7 +4,9 @@ # Usage: tool/package_linux.sh [--skip-build] [--noupload|--no-upload] # Output: dist/watch-together--linux-x64.tar.gz containing one # top-level folder with the executable, lib/ and data/ as built, plus -# update.sh (runs the binary with --upgrade). The in-app updater strips +# update.sh (runs the binary with --upgrade) and watch-together-server (the +# dedicated server, `dart compile exe` of packages/server/bin/server.dart, +# built every run, also with --skip-build). The in-app updater strips # that top-level folder — keep the layout in sync with updater_test.dart. # Upload: set DISCORD_WEBHOOK_URL in the environment (never commit it). # The tarball must stay under Discord's upload limit (25 MB free tier). @@ -46,6 +48,14 @@ rm -rf "$STAGE" mkdir -p "$STAGE/$FOLDER" "$DIST" cp -r "$BUNDLE/." "$STAGE/$FOLDER/" install -m 755 "$APP/lib/src/update/bundle/update.sh" "$STAGE/$FOLDER/update.sh" +# Prefer the dart bundled with the Flutter SDK that builds the app. +DART="dart" +if FLUTTER="$(command -v flutter)"; then + CANDIDATE="$(dirname "$(readlink -f "$FLUTTER")")/dart" + [[ -x "$CANDIDATE" ]] && DART="$CANDIDATE" +fi +(cd "$ROOT/packages/server" && "$DART" pub get >/dev/null && + "$DART" compile exe bin/server.dart -o "$STAGE/$FOLDER/watch-together-server") tar -czf "$TARBALL" -C "$STAGE" "$FOLDER" rm -rf "$STAGE" ls -la "$TARBALL"