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 <noreply@anthropic.com>
This commit is contained in:
foxyzandClaude Opus 5.5 committed 2026-10-07 00:55:14 +03:00
1 parent d8a0729029
commit 81cac1214d
48 files changed
+1061 -411

No files matched your search

+5
View File
@@ -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
+1 -3
View File
@@ -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
+24 -4
View File
@@ -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 <password>
```
| 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
+2 -1
View File
@@ -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).
-11
View File
@@ -19,10 +19,6 @@ Future<void> main(List<String> 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<void> main(List<String> args) async {
runApp(const WatchTogetherApp());
}
Future<void> runHeadless(List<String> 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.
@@ -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
@@ -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;
@@ -33,6 +33,7 @@ class _HomePageState extends State<HomePage> {
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<HomePage> {
}
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();
@@ -354,7 +354,7 @@ class _SessionScreenState extends State<SessionScreen> {
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) {
+1 -1
View File
@@ -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);
@@ -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<RoomBroadcast?>(null);
final serverState = signal<StateBroadcast?>(null);
final feedLog = signal<List<FeedMessage>>([]);
final toasts = signal<List<ToastItem>>([]);
final reconnecting = signal(false);
@@ -86,7 +89,7 @@ class SessionCoordinator {
final starving = signal<String?>(null);
late final Computed<MemberView?> 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<void> _onRoom(RoomBroadcast r) async {
Future<void> _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<void> _checkSeekLanded(RoomBroadcast r) async {
Future<void> _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<void> dispose() async {
await leave(silent: true);
await _roomSub?.cancel();
await _stateSub?.cancel();
await _feedSub?.cancel();
await _kickSub?.cancel();
await _errSub?.cancel();
@@ -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<void> _serverMain((Map<String, Object?>, 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<void> _exited = Completer<void>();
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<ServerIsolate> 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<void> 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);
}
}
}
+59 -21
View File
@@ -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<RecentServer> 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<List<RecentServer>>(const []);
final status = signal(const HealthStatus(HealthSeverity.info, 'idle'));
final room = signal<RoomBroadcast?>(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<String?>(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(
final hostKey = newHostKey();
_setupCrash = null;
late final ServerIsolate server;
try {
server = await ServerIsolate.start(
AppConfig(
port: params.port,
secret: params.secret,
name: 'room',
name: 'server',
mediaPath: params.mediaPath,
));
try {
await server.start();
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<void> join(ConnectParams params, ConnectForm form) async {
final op = ++_op;
_rememberForm(form);
@@ -305,15 +352,6 @@ class SessionState {
..linkUrl = form.linkUrl;
}
Future<void> startHeadless(List<String> 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<void> stopServer() async {
_stopKickWatch?.call();
_stopKickWatch = null;
@@ -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<PlayerAction> diff(RoomBroadcast room, double ownPosition, DateTime now) {
List<PlayerAction> diff(StateBroadcast state, double ownPosition, DateTime now) {
final out = <PlayerAction>[];
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;
+13 -11
View File
@@ -9,13 +9,13 @@ enum SyncStatus { idle, connecting, open, kicked, closed, error }
class SyncClient {
WebSocketChannel? _channel;
StreamSubscription? _sub;
final StreamController<RoomBroadcast> _rooms = StreamController.broadcast();
final StreamController<StateBroadcast> _states = StreamController.broadcast();
final StreamController<FeedMessage> _feeds = StreamController.broadcast();
final StreamController<String> _kicked = StreamController.broadcast();
final StreamController<String> _errors = StreamController.broadcast();
final StreamController<SyncStatus> _status = StreamController.broadcast();
Stream<RoomBroadcast> get rooms => _rooms.stream;
Stream<StateBroadcast> get states => _states.stream;
Stream<FeedMessage> get feeds => _feeds.stream;
Stream<String> get kicked => _kicked.stream;
Stream<String> get errors => _errors.stream;
@@ -24,7 +24,7 @@ class SyncClient {
String token = '';
String memberId = '';
String? mediaUrl;
RoomBroadcast? lastRoom;
StateBroadcast? lastState;
Completer<WelcomeMessage>? _pendingWelcome;
Future<WelcomeMessage> 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<void> dispose() async {
await disconnect();
await _rooms.close();
await _states.close();
await _feeds.close();
await _kicked.close();
await _errors.close();
+15 -14
View File
@@ -25,11 +25,11 @@ Future<void> _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();
+165
View File
@@ -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<void> _poll(
FutureOr<bool> 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 = <Map<String, Object?>>[];
_Raw(this.ws) {
ws.listen((m) => msgs.add(Map<String, Object?>.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<Map<String, Object?>> next(String type, {bool Function(Map<String, Object?>)? where}) async {
late Map<String, Object?> 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<int> _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 = <String>[];
final crashes = <String>[];
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<StateError>()));
} 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<Map>()
.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<Map>()
.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();
}
});
}
+1 -1
View File
@@ -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(
@@ -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<MemberView> members,
StateBroadcast _stateWithMembers(List<MemberView> 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<MemberView> 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<SeekAction>().single.position, closeTo(10.0, 1.0));
expect(actions.whereType<TransportAction>().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<RateAction>().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<RateAction>().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<SeekAction>().single.position, closeTo(90.0, 1.0));
expect(actions.whereType<RateAction>().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<SeekAction>().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<SeekAction>(), 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<SeekAction>().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<SeekAction>(), 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<SeekAction>().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);
});
});
}
@@ -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',
+6 -6
View File
@@ -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);
+15 -3
View File
@@ -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 <n> (default 8080), --secret <password> (empty = open
# server, anyone who can reach the port can join), --name, --media <file>,
# --advertise-host <host>, --config <file.yaml|json>. 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]
+56 -37
View File
@@ -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/<sid>/<token>/<name>` serves it with a Range
table, gated by the viewer's token. The server builds the URL per viewer
(`http://<host>:<port>/m/default/<token>/<name>`), so tokens never reach
other viewers. `<host>` 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.
+31 -6
View File
@@ -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=<x.y.z>
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
+2 -2
View File
@@ -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.
+19 -14
View File
@@ -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: <reason>` (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/<sid>/<token>/<name>`
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"|"<memberId>","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":{<room>}}
{"t":"room","eventId":7,"transport":"playing"|"held","position":12.3,
{"t":"welcome","token":"...","memberId":"m1","mediaUrl":"...","state":{<state>}}
{"t":"state","eventId":7,"transport":"playing"|"held","position":12.3,
"serverTime":"<ISO-8601 UTC>","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).
+10 -19
View File
@@ -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.
+8 -2
View File
@@ -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<String, Object?> 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<AppConfig> loadFile(String path) async {
+2 -2
View File
@@ -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;
+20 -10
View File
@@ -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<String, Object?> 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<String, Object?> json) {
@@ -87,16 +96,17 @@ class HelloMessage {
secret: (json['secret'] as String?) ?? '',
user: HelloUser.fromJson(Map<String, Object?>.from(user)),
media: HelloMedia.fromJson(Map<String, Object?>.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<String, Object?> toJson() => {
'position': position,
@@ -105,7 +115,7 @@ class RoomState {
'serverTime': serverTime.toIso8601String(),
};
factory RoomState.fromJson(Map<String, Object?> json) => RoomState(
factory SyncState.fromJson(Map<String, Object?> 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<String, Object?>? media;
final List<MemberView> members;
const RoomBroadcast({
const StateBroadcast({
required this.eventId,
required this.transport,
required this.position,
@@ -315,7 +325,7 @@ class RoomBroadcast {
});
Map<String, Object?> 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<String, Object?> json) {
if (json['t'] != 'room') throw const ProtocolException('not a room broadcast');
factory StateBroadcast.fromJson(Map<String, Object?> 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'),
+7 -7
View File
@@ -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,
+21 -5
View File
@@ -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<String, Object?>.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<String, Object?>.from(msg.toJson()));
expect(back.hostKey, 'hk');
});
test('hello rejects wrong type', () {
expect(() => HelloMessage.fromJson({'t': 'bye'}), throwsA(isA<ProtocolException>()));
});
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<String, Object?>.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<String, Object?>.from(s.toJson()));
expect(back.position, 12.5);
expect(back.paused, isFalse);
expect(back.mediaId, 3);
+19 -19
View File
@@ -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<String, Object?>.from(room.toJson()));
StateBroadcast.fromJson(Map<String, Object?>.from(state.toJson()));
expect(back.transport, 'held');
expect(back.members.single.name, 'alice');
expect(back.members.single.id, 'm1');
+61 -17
View File
@@ -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<void> main(List<String> 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?;
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: int.tryParse('${parsed['port']}') ?? 8080,
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();
final events = ReceivePort('server events');
final done = serve(config.toMap(), events: events.sendPort);
final signals = <StreamSubscription<ProcessSignal>>[];
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=${server.boundPort} room=${config.name}');
await ProcessSignal.sigint.watch().first;
await server.stop();
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();
}
}
+3 -2
View File
@@ -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';
+47
View File
@@ -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<void> serve(Map<String, Object?> config, {SendPort? events}) async {
StreamSubscription<pkg_logging.LogRecord>? 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();
}
@@ -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<String, RelicWebSocket> 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<String>();
@@ -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;
+1 -1
View File
@@ -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;
@@ -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<void> 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');
}
}
+1 -1
View File
@@ -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:
+32 -16
View File
@@ -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<String?> _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<AuthException>()));
await expectLater(
server.hub.kickById(idOf('host'), 'x'), throwsA(isA<AuthException>()));
// 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 {
+2 -2
View File
@@ -60,10 +60,10 @@ Future<void> _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();
});
@@ -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<void> _run((Map<String, Object?>, 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 = <Object?>[];
errors.listen(crashes.add);
final logs = <Map>[];
final ready = Completer<Map>();
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();
});
}
@@ -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<String?> _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<int>.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<String>)> 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()));
+2 -2
View File
@@ -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 {
+2 -2
View File
@@ -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', () {
+6 -6
View File
@@ -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<void> main(List<String> 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<void> main(List<String> 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<void> main(List<String> 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<void> main(List<String> 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<void> main(List<String> 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);
+11 -1
View File
@@ -4,7 +4,9 @@
# Usage: tool/package_linux.sh [--skip-build] [--noupload|--no-upload]
# Output: dist/watch-together-<version>-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"