From f4ca02c3cfe60dd9b29528cb9f22fae3d5bc8fdb Mon Sep 17 00:00:00 2001 From: foxyz Date: Thu, 15 Oct 2026 20:05:45 +0300 Subject: [PATCH] Restructure step 0 done: notice table test, flaky-link proxy, liveness on the engine clock, STARSYNC_PROFILE - notice_table_test: one row per AppEvent variant through noticeFor. - FlakyProxy (core/testing.dart): pass, blackhole (half-open), delay, hold, cut after N bytes, reset; demo tests against today's server. - Liveness (stale/drop/ping) on the engine clock, so grace can be tested with a fake clock; Connection takes an injectable clock. - AppDirs with STARSYNC_PROFILE=: separate data and logs, so two instances run on one machine. - DEVELOPMENT: scenario list per restructure step; test tags follow the new step order. Co-Authored-By: Claude Opus 5.5 --- apps/desktop/lib/main.dart | 10 +- apps/desktop/lib/src/platform/app_dirs.dart | 64 ++++ apps/desktop/lib/src/platform/app_log.dart | 17 +- apps/desktop/lib/src/session/connection.dart | 17 +- apps/desktop/test/app_dirs_test.dart | 36 +++ apps/desktop/test/jellyfin_e2e_test.dart | 2 +- apps/desktop/test/jellyfin_session_test.dart | 2 +- apps/desktop/test/notice_table_test.dart | 141 +++++++++ apps/desktop/test/queue_logic_test.dart | 2 +- apps/desktop/test/queue_switch_e2e_test.dart | 2 +- docs/ARCHITECTURE.md | 26 +- docs/DEVELOPMENT.md | 48 +++ docs/ROADMAP.md | 17 +- packages/core/lib/src/constants.dart | 4 + packages/core/lib/src/sync_engine.dart | 28 +- .../core/lib/src/testing/flaky_proxy.dart | 275 ++++++++++++++++++ packages/core/lib/testing.dart | 5 + packages/core/test/flaky_proxy_test.dart | 134 +++++++++ packages/core/test/sync_engine_test.dart | 31 +- packages/core/test/watch_session_test.dart | 2 +- packages/server/lib/src/server_hub.dart | 4 +- packages/server/lib/src/sync_server.dart | 8 +- packages/server/test/flaky_link_test.dart | 132 +++++++++ 23 files changed, 959 insertions(+), 48 deletions(-) create mode 100644 apps/desktop/lib/src/platform/app_dirs.dart create mode 100644 apps/desktop/test/app_dirs_test.dart create mode 100644 apps/desktop/test/notice_table_test.dart create mode 100644 packages/core/lib/src/testing/flaky_proxy.dart create mode 100644 packages/core/lib/testing.dart create mode 100644 packages/core/test/flaky_proxy_test.dart create mode 100644 packages/server/test/flaky_link_test.dart diff --git a/apps/desktop/lib/main.dart b/apps/desktop/lib/main.dart index 1b8db0b..d86990b 100644 --- a/apps/desktop/lib/main.dart +++ b/apps/desktop/lib/main.dart @@ -7,6 +7,7 @@ import 'package:media_kit/media_kit.dart'; import 'package:sync_relay_core/core.dart'; import 'app.dart'; +import 'src/platform/app_dirs.dart'; import 'src/platform/app_log.dart'; import 'src/platform/app_shutdown.dart'; import 'src/platform/desktop_entry.dart'; @@ -41,14 +42,17 @@ Future main(List args) async { ), ); WidgetsFlutterBinding.ensureInitialized(); - await startAppLogging(); - await Hive.initFlutter(); + final dirs = await AppDirs.resolve(); + await startAppLogging(dirs); + // An absolute subDir replaces the documents folder (package:path join). + await Hive.initFlutter(dirs.hive); await AppSettings.instance.ensureLoaded(); // Close button / Alt+F4: the orderly close ([shutDownApp]), so others // see "left", a hosted server shuts down and no player outlives the // engine. final ports = SessionState.instance.ports; - await ports.shell.start(title: appName, onCloseRequest: shutDownApp); + await ports.shell.start( + title: dirs.profile == null ? appName : '$appName [${dirs.profile}]', onCloseRequest: shutDownApp); try { await SessionState.instance.loadForm(); } catch (e, st) { diff --git a/apps/desktop/lib/src/platform/app_dirs.dart b/apps/desktop/lib/src/platform/app_dirs.dart new file mode 100644 index 0000000..be027c3 --- /dev/null +++ b/apps/desktop/lib/src/platform/app_dirs.dart @@ -0,0 +1,64 @@ +import 'dart:io'; + +import 'package:path_provider/path_provider.dart'; + +/// The environment variable that picks a separate profile (DEVELOPMENT.md +/// "Dev loop": two instances on one machine). +const profileEnv = 'STARSYNC_PROFILE'; + +/// Where the app keeps its own files, resolved once in `main` ([resolve]) +/// and handed on from there. +/// +/// Without a profile it is exactly today's layout: Hive's boxes in Hive's +/// default folder (the documents folder) and the logs in +/// `/logs`. With `STARSYNC_PROFILE=` (letters, digits, +/// `-`, `_`; at most 32) the per-install data moves to +/// `/profiles//`: the Hive boxes (settings and so the +/// profile, connect form, window bounds, folders, Jellyfin sign-ins; the +/// client id from restructure step 1 on) under `hive/`, the logs under +/// `logs/`. Downloads and caches stay shared: yt-dlp and its cookies, the +/// extracted shaders, thumbnails, the AniList cache. +class AppDirs { + /// The profile name; null for the normal app. + final String? profile; + /// The app support folder (yt-dlp, shaders; logs without a profile). + final String support; + /// Hive's folder; null: Hive's default (`Hive.initFlutter()`). + final String? hive; + /// The app's log folder. + final String logs; + + const AppDirs({required this.profile, required this.support, required this.hive, required this.logs}); + + /// The layout for [support] (the app support folder) and [env]; an + /// invalid profile name is ignored ([invalidProfile] says which). + static AppDirs fromEnv({required String support, required Map env}) { + final sep = Platform.pathSeparator; + final name = validProfile(env[profileEnv]); + if (name == null) return AppDirs(profile: null, support: support, hive: null, logs: '$support${sep}logs'); + final root = '$support${sep}profiles$sep$name'; + return AppDirs(profile: name, support: support, hive: '$root${sep}hive', logs: '$root${sep}logs'); + } + + /// [raw] when it is a usable profile name, else null. + static String? validProfile(String? raw) { + final v = raw?.trim() ?? ''; + return RegExp(r'^[A-Za-z0-9_-]{1,32}$').hasMatch(v) ? v : null; + } + + /// The env value when it is set but not a usable name (for a warning). + static String? invalidProfile(Map env) { + final raw = env[profileEnv]; + return raw != null && raw.trim().isNotEmpty && validProfile(raw) == null ? raw : null; + } + + /// Resolves the layout from the platform and the process environment. + /// Needs Flutter's binding (path_provider). + static Future resolve() async { + final support = (await getApplicationSupportDirectory()).path; + final dirs = fromEnv(support: support, env: Platform.environment); + final hive = dirs.hive; + if (hive != null) await Directory(hive).create(recursive: true); + return dirs; + } +} diff --git a/apps/desktop/lib/src/platform/app_log.dart b/apps/desktop/lib/src/platform/app_log.dart index 7657dad..43c4ebd 100644 --- a/apps/desktop/lib/src/platform/app_log.dart +++ b/apps/desktop/lib/src/platform/app_log.dart @@ -2,10 +2,10 @@ import 'dart:async'; import 'dart:io'; import 'package:flutter/foundation.dart'; -import 'package:path_provider/path_provider.dart'; import 'package:sync_relay_core/core.dart'; import '../screens/home/ui.dart' show appName, displayVersion; +import 'app_dirs.dart'; import '../state/session_state.dart'; // The app's log on disk (/logs/app-YYYY-MM-DD.log; core's @@ -17,13 +17,11 @@ final _appLog = createLogger('app'); /// The logs folder, once [startAppLogging] ran. String? appLogDir; -/// Opens the log files and routes `debugPrint` and uncaught errors into -/// the log. Needs Flutter's binding (path_provider). -Future startAppLogging() async { +/// Opens the log files in [dirs]' log folder and routes `debugPrint` and +/// uncaught errors into the log. +Future startAppLogging(AppDirs dirs) async { try { - final base = await getApplicationSupportDirectory(); - final dir = '${base.path}${Platform.pathSeparator}logs'; - if (startLogFiles(dir) != null) appLogDir = dir; + if (startLogFiles(dirs.logs) != null) appLogDir = dirs.logs; } catch (e) { _appLog.warning('no log files: $e'); } @@ -39,7 +37,10 @@ Future startAppLogging() async { return false; }; _appLog.info('$appName ${displayVersion()} on ${Platform.operatingSystem} ' - '${Platform.operatingSystemVersion}, pid $pid, Dart ${Platform.version.split(' ').first}'); + '${Platform.operatingSystemVersion}, pid $pid, Dart ${Platform.version.split(' ').first}' + '${dirs.profile == null ? '' : ', profile ${dirs.profile}'}'); + final bad = AppDirs.invalidProfile(Platform.environment); + if (bad != null) _appLog.warning('$profileEnv="$bad" ignored: use letters, digits, - and _ (at most 32)'); } /// The text "Copy diagnostics" gives: versions, then the last [since] of diff --git a/apps/desktop/lib/src/session/connection.dart b/apps/desktop/lib/src/session/connection.dart index b4cbeb0..06fa879 100644 --- a/apps/desktop/lib/src/session/connection.dart +++ b/apps/desktop/lib/src/session/connection.dart @@ -196,23 +196,30 @@ class Connection { final Duration reconnectDelay; /// No state broadcast for this long means the connection is lost. final Duration silenceLimit; + /// The clock [silenceLimit] is measured on (UTC; a test passes a fake + /// one). The heartbeat that checks it stays a real 1 s timer. + final DateTime Function() now; Connection( {void Function(AppEvent event)? notify, SyncClient? client, this.hello = defaultHello, this.reconnectDelay = const Duration(milliseconds: 1500), - this.silenceLimit = const Duration(seconds: 5)}) + this.silenceLimit = const Duration(seconds: 5), + DateTime Function()? now}) : _notify = notify ?? ((_) {}), + now = now ?? _utcNow, client = client ?? SyncClient() { _subs = [ this.client.status.listen(_onStatus), - this.client.states.listen((_) => _lastStateAt = DateTime.now().toUtc()), + this.client.states.listen((_) => _lastStateAt = this.now()), ]; } late final List _subs; + static DateTime _utcNow() => DateTime.now().toUtc(); + final _state = signal(const ConnClosed()); /// Every change is notified, a repeat too. ReadonlySignal get state => _state; @@ -368,7 +375,7 @@ class Connection { _notify(const ConnectionBack()); } _resumeToken = client.token; - _lastStateAt = DateTime.now().toUtc(); + _lastStateAt = now(); _set(const ConnOnline()); _startHeartbeat(); } @@ -400,13 +407,13 @@ class Connection { /// Every second while connected: a clock sample, the diagnostics, and /// the silence check. void _beat() { - final now = DateTime.now().toUtc(); + final t = now(); if (client.token.isNotEmpty) client.sendPing(); diag ..share = AppSettings.instance.shareDiagnostics.peek() ..tick(); final last = _lastStateAt; - if (!_closed && last != null && now.difference(last) >= silenceLimit) { + if (!_closed && last != null && t.difference(last) >= silenceLimit) { _silence.add(null); if (!_state.peek().retrying) _dropped(); } diff --git a/apps/desktop/test/app_dirs_test.dart b/apps/desktop/test/app_dirs_test.dart new file mode 100644 index 0000000..e965846 --- /dev/null +++ b/apps/desktop/test/app_dirs_test.dart @@ -0,0 +1,36 @@ +// restructure: rewrite in step R (Rename): Hive moves into the support +// folder for the normal app too. +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:watch_together/src/platform/app_dirs.dart'; + +void main() { + final sep = Platform.pathSeparator; + const support = '/s'; + + test('no profile: today\'s layout', () { + final d = AppDirs.fromEnv(support: support, env: const {}); + expect(d.profile, isNull); + expect(d.hive, isNull, reason: "Hive's default folder"); + expect(d.logs, '$support${sep}logs'); + }); + + test('a profile gets its own Hive and logs folders', () { + final d = AppDirs.fromEnv(support: support, env: const {profileEnv: 'b'}); + expect(d.profile, 'b'); + expect(d.hive, '$support${sep}profiles${sep}b${sep}hive'); + expect(d.logs, '$support${sep}profiles${sep}b${sep}logs'); + expect(d.support, support, reason: 'yt-dlp and shaders stay shared'); + }); + + test('an unusable name is ignored and reported', () { + for (final bad in ['../x', 'a b', 'x' * 33, 'é']) { + final env = {profileEnv: bad}; + expect(AppDirs.fromEnv(support: support, env: env).profile, isNull, reason: bad); + expect(AppDirs.invalidProfile(env), bad); + } + expect(AppDirs.invalidProfile(const {profileEnv: ''}), isNull, reason: 'empty is just unset'); + expect(AppDirs.invalidProfile(const {profileEnv: 'ok_1-2'}), isNull); + }); +} diff --git a/apps/desktop/test/jellyfin_e2e_test.dart b/apps/desktop/test/jellyfin_e2e_test.dart index 1bbd81d..0748bb8 100644 --- a/apps/desktop/test/jellyfin_e2e_test.dart +++ b/apps/desktop/test/jellyfin_e2e_test.dart @@ -1,4 +1,4 @@ -// restructure: rewrite in step 5 (Watch tracking): the progress-reporting +// restructure: rewrite in step 6 (Watch tracking): the progress-reporting // checks; the guest-through-the-relay flow stays. import 'dart:async'; import 'dart:convert'; diff --git a/apps/desktop/test/jellyfin_session_test.dart b/apps/desktop/test/jellyfin_session_test.dart index 1afc0a4..106996d 100644 --- a/apps/desktop/test/jellyfin_session_test.dart +++ b/apps/desktop/test/jellyfin_session_test.dart @@ -1,4 +1,4 @@ -// restructure: rewrite in step 5 (What's next + Watch tracking): the +// restructure: rewrite in step 6 (What's next + Watch tracking): the // "progress" and "episodes" groups (reporting, auto-advance swapper). import 'dart:convert'; diff --git a/apps/desktop/test/notice_table_test.dart b/apps/desktop/test/notice_table_test.dart new file mode 100644 index 0000000..77e34e2 --- /dev/null +++ b/apps/desktop/test/notice_table_test.dart @@ -0,0 +1,141 @@ +// restructure: rewrite in step 1 (Connection + Identity + Presence): the +// feed rows follow the new feed actions one-to-one (away/in/gone/return); +// step 2 (Status + Notifier): every row, when noticeFor takes a context. +// +// The Notifier policy as a table: one row per AppEvent subtype, what +// noticeFor makes of it (toast text and actor, sound, attention, activity-log +// line). A policy change edits a row here on purpose. +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:sync_relay_core/core.dart' show FeedMessage; +import 'package:watch_together/src/session/notifier.dart'; +import 'package:watch_together/src/sfx/sfx.dart'; + +final _at = DateTime.utc(2026, 1, 1); + +FeedMessage _feed(String action, {String actor = 'Ann', String? target, String reason = ''}) => + FeedMessage(actor: actor, action: action, target: target, reason: reason, at: _at); + +/// One row: the event, then what it must do. +class _Row { + final String name; + final AppEvent event; + final Type type; + final String? toast; + final String? actor; + final SfxKind? sound; + final int readyIndex; + final bool attention; + /// Whether the event's own feed line goes into the activity log. + final bool logged; + const _Row(this.name, this.event, this.type, + {this.toast, this.actor, this.sound, this.readyIndex = 0, this.attention = false, this.logged = false}); +} + +/// The feed as the coordinator turns it into events: this member is 'Me', +/// two members ready in the latest state. +FeedEvent _of(FeedMessage f) => FeedEvent.of(f, selfName: 'Me', readyCount: 2); + +final _rows = <_Row>[ + // ---- the feed (FeedEvent.of, then noticeFor); every line is logged + _Row('join', _of(_feed('join')), MemberJoined, + toast: 'Ann joined', actor: 'Ann', sound: SfxKind.join, logged: true), + _Row('leave', _of(_feed('leave')), MemberLeft, + toast: 'Ann left', actor: 'Ann', sound: SfxKind.leave, logged: true), + _Row('stale', _of(_feed('stale')), MemberStale, logged: true), + _Row('out, someone else sat out', _of(_feed('out', target: 'Bob')), MemberSatOut, logged: true), + _Row('out, sat yourself out', _of(_feed('out', actor: 'Me', target: 'Me')), MemberSatOut, logged: true), + _Row('out, someone sat you out', _of(_feed('out', target: 'Me')), MemberSatOut, + toast: 'Ann sat you out', actor: 'Ann', sound: SfxKind.kick, logged: true), + _Row('back', _of(_feed('back')), MemberBack, logged: true), + _Row('ready', _of(_feed('ready')), ReadyChanged, sound: SfxKind.ready, readyIndex: 2, logged: true), + _Row('unready', _of(_feed('unready')), ReadyChanged, sound: SfxKind.unready, logged: true), + _Row('force-ready', _of(_feed('force-ready')), ForcedReady, sound: SfxKind.forceReady, logged: true), + _Row('force-unready', _of(_feed('force-unready')), ForcedReady, logged: true), + _Row('force-play', _of(_feed('force-play')), ForcedTransport, logged: true), + _Row('force-hold', _of(_feed('force-hold')), ForcedTransport, logged: true), + _Row('swap', _of(_feed('swap')), MediaSwapped, + toast: 'Ann changed media', actor: 'Ann', sound: SfxKind.swap, logged: true), + _Row('seek', _of(_feed('seek')), Seeked, logged: true), + _Row('pause', _of(_feed('pause', actor: 'server')), Paused, logged: true), + _Row('resume', _of(_feed('resume', actor: 'server')), Resumed, logged: true), + _Row('finished', _of(_feed('finished', actor: 'server')), Finished, logged: true), + _Row('chat from someone', _of(_feed('chat', reason: 'hi')), ChatPosted, + toast: 'Ann: hi', actor: 'Ann', sound: SfxKind.chat, logged: true), + _Row('own chat', _of(_feed('chat', actor: 'Me', reason: 'hi')), ChatPosted, + toast: 'Me: hi', actor: 'Me', logged: true), + _Row('rename', _of(_feed('rename', target: 'Annie')), MemberRenamed, logged: true), + _Row('avatar', _of(_feed('avatar')), AvatarChanged, logged: true), + _Row('queue-add', _of(_feed('queue-add', target: 'Ep 6')), QueueAdded, logged: true), + _Row('queue-remove', _of(_feed('queue-remove', target: 'Ep 6')), QueueRemoved, logged: true), + _Row('queue-skip with a reason', _of(_feed('queue-skip', actor: 'server', target: 'Ep 6', reason: 'gone')), + QueueSkipped, toast: 'Skipped Ep 6: gone', logged: true), + _Row('queue-skip, no title or reason', _of(_feed('queue-skip', actor: 'server')), QueueSkipped, + toast: 'Skipped an item', logged: true), + _Row('fetch-fail', _of(_feed('fetch-fail', actor: 'server', reason: '404')), FetchFailed, logged: true), + _Row('an unknown action', _of(_feed('from-the-future')), OtherFeed, logged: true), + // ---- the session + _Row('nudge', const Nudged('Ann'), Nudged, + toast: 'Ann nudged you', actor: 'Ann', sound: SfxKind.nudge, attention: true), + _Row('nudge from someone unknown', const Nudged(''), Nudged, + toast: 'Someone nudged you', sound: SfxKind.nudge, attention: true), + _Row('member cannot play', const MemberCantPlay('Ann', 'no codec'), MemberCantPlay, + toast: "Ann can't play this: no codec", actor: 'Ann'), + _Row('slow source', const SlowSource('Passing on slower than it plays'), SlowSource, + toast: 'Passing on slower than it plays'), + _Row('relay started', const RelayStarted(), RelayStarted, + toast: 'Passing the video on through the server. Keep the app open while it plays.'), + _Row('relay failed', const RelayFailed('closed'), RelayFailed, toast: 'Cannot pass the video on: closed'), + _Row('idle, unready', const IdleWarning(sitsOut: false), IdleWarning, + toast: 'idle — auto-unready in 2 min unless you press Space', sound: SfxKind.warn), + _Row('idle, sits out', const IdleWarning(sitsOut: true), IdleWarning, + toast: 'idle — sitting out in 2 min unless you press Space', sound: SfxKind.warn), + _Row('connection lost', const ConnectionLost(), ConnectionLost, sound: SfxKind.lost), + _Row('connection back', const ConnectionBack(), ConnectionBack, sound: SfxKind.back), + _Row('jellyfin, sign-in expired', const JellyfinProgressFailed(signInExpired: true), JellyfinProgressFailed, + toast: 'Jellyfin progress not saved: sign-in expired'), + _Row('jellyfin, unreachable', const JellyfinProgressFailed(signInExpired: false), JellyfinProgressFailed, + toast: 'Jellyfin progress not saved: server unreachable'), + // ---- the app + _Row('shader message', const ShaderMessage('Shaders off'), ShaderMessage, toast: 'Shaders off'), + _Row('added to up next', const AddedToUpNext('Ep 6'), AddedToUpNext, toast: 'Added Ep 6 to Up next'), + _Row('action failed', const ActionFailed('No such file'), ActionFailed, toast: 'No such file'), + _Row('schedule result', const ScheduleResult('Loaded'), ScheduleResult, toast: 'Loaded'), + _Row('problem marked in session', const ProblemMarked(inSession: true), ProblemMarked, + toast: 'Marked in the session log'), + _Row('problem marked locally', const ProblemMarked(inSession: false), ProblemMarked, + toast: "Marked in this app's log"), + _Row('copied', const Copied('Link copied'), Copied, toast: 'Link copied'), + _Row('copy failed', const CopyFailed(), CopyFailed, toast: "Couldn't copy"), + _Row('update available', const UpdateAvailable('1.3.0'), UpdateAvailable), +]; + +void main() { + group('noticeFor', () { + for (final r in _rows) { + test(r.name, () { + expect(r.event.runtimeType, r.type, reason: 'event type'); + final n = noticeFor(r.event); + expect(n.toast, r.toast, reason: 'toast'); + expect(n.actor, r.actor, reason: 'actor'); + expect(n.sound?.kind, r.sound, reason: 'sound'); + if (r.sound != null) expect(n.sound!.readyIndex, r.readyIndex, reason: 'ready pitch'); + expect(n.attention, r.attention, reason: 'attention'); + final e = r.event; + expect(n.log, r.logged ? same((e as FeedEvent).feed) : isNull, reason: 'activity log'); + }); + } + }); + + test('every concrete AppEvent subtype has a row', () { + // Read from the source, so a new event fails here until it gets a row. + final src = File('lib/src/session/app_events.dart').readAsStringSync(); + final declared = RegExp(r'^class (\w+) extends (?:FeedEvent|AppEvent)', multiLine: true) + .allMatches(src) + .map((m) => m.group(1)!) + .toSet(); + expect(declared, isNotEmpty); + expect(_rows.map((r) => r.type.toString()).toSet(), declared); + }); +} diff --git a/apps/desktop/test/queue_logic_test.dart b/apps/desktop/test/queue_logic_test.dart index 2312645..531bb16 100644 --- a/apps/desktop/test/queue_logic_test.dart +++ b/apps/desktop/test/queue_logic_test.dart @@ -1,4 +1,4 @@ -// restructure: rewrite in step 5 (What's next): who performs an item and +// restructure: rewrite in step 6 (What's next): who performs an item and // the janitor become server state. import 'package:flutter_test/flutter_test.dart'; import 'package:sync_relay_core/core.dart'; diff --git a/apps/desktop/test/queue_switch_e2e_test.dart b/apps/desktop/test/queue_switch_e2e_test.dart index 488cde6..ea64c99 100644 --- a/apps/desktop/test/queue_switch_e2e_test.dart +++ b/apps/desktop/test/queue_switch_e2e_test.dart @@ -1,4 +1,4 @@ -// restructure: rewrite in step 6 (Media pipeline): preload and the warm +// restructure: rewrite in step 5 (Media pipeline): preload and the warm // relay become one pipeline; keep the timing per source. import 'dart:async'; import 'dart:convert'; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 114b90c..eed563e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -7,7 +7,7 @@ Plain overview first: [HOW-IT-WORKS.md](HOW-IT-WORKS.md). Wire format: | Path | What | Rule | |---|---|---| -| `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging, Jellyfin client, updater, watched rule (`WatchSession`) | pure Dart, no Flutter | +| `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging, Jellyfin client, updater, watched rule (`WatchSession`); `testing.dart` (test support for all three packages, not in `core.dart`: `FlakyProxy`) | pure Dart, no Flutter | | `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint, relay buffer, Jellyfin relay), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | | `apps/desktop` | Flutter app: UI, `SyncClient`, `Connection`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, update UI (`UpdateController`) | ships with `watch-together-server` | | `tool` | harnesses: `range_logger`, `shaper`, `fake_viewer`, `e2e_check` | dev only | @@ -180,6 +180,12 @@ Loosely modeled on Syncplay, with a different readiness design. user is not seeking. - **Liveness**: WS ping every 5 s; silent over 15 s means stale (out of the gate, `stale` feed); over 30 s the socket is closed and the member removed. + The durations are a `Liveness` (core, `sync_engine.dart`; defaults + `RelayConstants.staleAfterSecs` / `dropAfterSecs` / `pingIntervalSecs`) + given to `SyncEngine(liveness:)` and `SyncServer(liveness:)`; stale and + drop are measured on the engine's clock (`SyncEngine.clock`, + `SyncServer(clock:)`) by the 1 s heartbeat sweep, so a test with a fake + clock steps through them; the ping is the socket's own real timer. A stale member goes back into the set with its next frame if it is ready (also one stale at a swap) or the server is not PLAYING. Closed sockets are polled with `isClosed` every second: Relic's server-side @@ -228,7 +234,9 @@ the diagnostics tick, the silence check: `silenceLimit`, 5 s), the diagnostics uploader (`diag`, started with the session, stopped by `close` before the socket), "Stop reconnecting" (`cancelReconnect`), "Retry" (`retry`) and setup cancel (`cancelConnect`). The `hello` is -built by an injectable `HelloBuilder` (`defaultHello`). The incoming +built by an injectable `HelloBuilder` (`defaultHello`). The silence is +measured on `Connection.now` (injectable, UTC wall clock by default); the +heartbeat and the reconnect delay are real timers. The incoming streams (`states`, `feeds`, `nudges`, `errors`), the send methods, the `token`, `memberId` and `serverNow()` are read through it. @@ -1494,7 +1502,12 @@ panel) and chat messages (Copy message, and a link in it). Settings persist in two Hive boxes: `connect_form` (last form, recent servers and links, the profile picture and colour; saved even when a join fails) and -`settings`. Link media names are a web page's title (see "yt-dlp"), else +`settings`. Where Hive and the logs live is `AppDirs` +(`platform/app_dirs.dart`, resolved once in `main`): Hive's default folder +(the documents folder) and `/logs`; with `STARSYNC_PROFILE` +(DEVELOPMENT "Dev loop") `/profiles//hive` and `logs`, +and the window title gets ` []`. yt-dlp, shaders and caches stay +shared. Link media names are a web page's title (see "yt-dlp"), else derived from the URL path, never containing a query string. Jellyfin on the desktop (`lib/src/jellyfin/`, over the core client): @@ -1684,9 +1697,10 @@ dedicated server (`package:logging` loggers named by area: `relay`, `app`). - **Where**: the app writes `/logs/app-YYYY-MM-DD.log` - (Linux `~/.local/share/moe.senkochan.watch_together/logs`); its hosted - server's records arrive over the isolate's `events` port and go in the - same file as `server.` (`relogLine`). The dedicated server + (Linux `~/.local/share/moe.senkochan.watch_together/logs`; with a + profile `/profiles//logs`, see `AppDirs` under + "Desktop UI"); its hosted server's records arrive over the isolate's + `events` port and go in the same file as `server.` (`relogLine`). The dedicated server writes `server-YYYY-MM-DD.log` in `--log-dir`, else `logs` next to `--config`, else `logs` next to the binary (systemd: `$STATE_DIRECTORY/logs`), else `~/.cache/watch-together/logs`; it diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 96a0b7a..b2dd574 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -49,6 +49,18 @@ effect with embedded video: media_kit forces `vo=libmpv`. `SYNC_RELAY_PRELOAD=0` turns Up next's preload player off; `SYNC_RELAY_TRACE_SWITCH=1` prints the timings of each media change (`SwitchTrace`). +Two instances on one machine: `STARSYNC_PROFILE=` (letters, digits, +`-`, `_`; at most 32) gives an instance its own Hive boxes (settings, +profile, connect form, window bounds, folders, Jellyfin sign-ins; the +client id from restructure step 1 on) and logs under +`/profiles//`, and ` []` in the window title; +yt-dlp, shaders and caches stay shared (`AppDirs`, ARCHITECTURE "Desktop +UI"). Without it nothing changes. Run one plain instance and one with +`STARSYNC_PROFILE=b` (`flutter run -d linux` passes the environment on, or +start the built binary twice); one hosts, the other joins +`127.0.0.1:`, or both join a `dart run` server. An unusable name is +ignored with a warning in the log. + Harnesses in `tool/` (own `pubspec.yaml`, `dart run tool/.dart`): `range_logger` (serve a file and log every Range request), `shaper` (bandwidth cap, latency, `--kill-after` for outage tests), `fake_viewer` @@ -95,6 +107,42 @@ display; that says nothing about the app. `sound_device_e2e` opens the real audio device (flutter_soloud) and plays every sound of every preset quietly (`WT_SFX_VOLUME`, default 0.1). +Bad links: `FlakyProxy` (`package:sync_relay_core/testing.dart`, so +core, server and desktop tests share it; not part of `core.dart`) sits +between a client and a server on loopback: `FlakyProxy.start(serverPort)`, +the client dials `proxy.port`, and each accepted connection is a +`FlakyLink` the test steers at any time: `pass()`, `blackhole()` (both +sockets stay open, every byte and a closing end are dropped: a half-open +link), `delay` (added latency each way, order kept), `holdNext(d, flow:)` +(the next chunk, and what queues behind it, arrives `d` late), +`cutAfter(n, flow:)` (n more bytes, then a reset) and `reset()` (RST via +SO_LINGER 0). `onConnect` steers a link before its first byte (blackhole +or refuse the next dial). It counts bytes each way and the dropped ones. +It is TCP only: a "frame" is one chunk as read (on loopback usually one +small WebSocket frame). Its own tests: core `flaky_proxy_test`; against +the server: server `flaky_link_test`. For manual runs with real apps use +`tool/shaper` instead. + +Long durations on a fake clock: `SyncServer(clock:, liveness:)` and +`SyncEngine(clock:, liveness:)` take the engine clock and the server's +silence limits (`Liveness`, ARCHITECTURE "Sync model"), so a test can +pass short limits or step a fake clock through minutes (the 1 s heartbeat +still ticks in real time and reads the clock); `Connection(now:, +silenceLimit:, reconnectDelay:)` on the app side. + +Safety-net scenarios: the core flows every restructure step must keep +working. What covers them today and which step adds the rest (steps as +in [ROADMAP.md](ROADMAP.md)): + +| Scenario | Covered today | Missing, added in | +|---|---|---| +| Drop and reconnect while sharing a relay | server `relay_test` "source drops and comes back", `sync_wave_test` (resume after an unclean drop), `flaky_link_test` (blackhole, reset); desktop `relay_e2e` "the source drops uncleanly and resumes", `coord_presence` (dead server, reconnect restores ready) | step 1: the same member back after a closed socket (grace, P1), the source re-dialling its relay (D6), the player and state kept through the drop, through `FlakyProxy` and a fake clock | +| Join mid-playback | core `sync_engine_test` "late joiner does not halt playback", "a joiner while playing never holds anyone back"; server `relay_test` "a late joiner gets its own /r/ URL" | step 1: `joining` until in sync, loading in the background while playing (D2, P3), the ~15 s hold while not playing | +| Sit out / Away and back | core `sync_engine_test` "sitting out"; server `hardening_test` "anyone sits anyone out"; desktop `server_isolate_test` (host sat out), `notice_table_test` (sat you out) | step 1: presence `away`/`in`, the ready counter (P6), auto sit-out; step 2: notifications while away (P8); step 6: not tracked as watched (P7) | +| Swap, also while someone is Away or gone | core `sync_engine_test` "swap" (a stale member stays out), `queue_test` (a swap takes its item); server `relay_test` (a swap drops the buffer); desktop `sync_playback_e2e` | step 1: a swap with a member `away` (stays away) or `gone` (in grace, its items wait); step 5: no commands reach a player mid-switch (P15) | +| Up next advance | core `queue_test` (skip, play request timeout); server `queue_test`, `relay_test` "warm relay"; desktop `jellyfin_session_test` (auto-advance swapper, end card countdown), `queue_switch_e2e` | step 5: preparing it through the media pipeline; step 6: the continuation and the countdown as server state with a shared cancel | +| Close the app cleanly, incl. `bye` | server `lifecycle_test` (clean close removes the member, server stop closes sockets); desktop `server_isolate_test` (stop frees the port), `coord_presence` (cancelled host setup releases the port) | step 1: `bye` leaves at once, any other close keeps the member in grace, update-and-restart sends no `bye`; step 5: the Windows close smoke job (P32) | + A test file whose first line is `// restructure: rewrite in step N` pins behaviour that step of the restructure ([ROADMAP.md](ROADMAP.md)) changes; the comment names the tests. Rewrite or delete them in that step, not diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 450652b..06493cb 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -85,17 +85,12 @@ the automatic continuation and countdown, relay state), the device (client id, library and folders, match corrections, watch history, settings, notification prefs, what this app is sourcing) and the integrations (Jellyfin played/resume, AniList list). -0. Safety net and seams, no visible change. Left: a `noticeFor` table - test; a flaky-link test harness (pure-Dart TCP proxy: pass, blackhole, - delay, reset, late frame); grace and silence through - the engine clock; an env override for client id and data folder (two - instances on one machine); the scenario list in DEVELOPMENT "Tests". - (Done: tests audited and tagged, P27, AppEvent + Notifier, ports, - protocol 16 planned, Connection extracted from the coordinator.) R. Rename (before step 1, so dev builds install side by side with 1.2.x): app id `dev.foxyz.starsync`, binary, Windows company `foxyz`; Hive out of `~/Documents` into the app's support folder, copied not moved; - a settings schema version with ordered migrations; an `AppDirs` port; + a settings schema version with ordered migrations; `AppDirs` (today + Hive and logs, and `STARSYNC_PROFILE`) becomes a port and the default + moves next to the profiles; the 1.3.0 archive still contains `watch_together(.exe)` so 1.2.x's self-update restart works; the old `.desktop` entry removed; the manifest kept at the old URL; VPS `ExecStart`/`update.sh` paths. @@ -156,9 +151,9 @@ Not in the restructure: P25 (ops). **Agreed since:** the queue stays in memory (no server storage). Transport: `wss://` preferred when a server offers it, `ws://` still allowed (a cleartext exception on Android); it only carries sync. -Tests: many pin behaviour the restructure drops; step 0 audits them -(delete what no longer fits, keep a few scenario tests for the core -flows). Open questions below are decided during the restructure. +Tests: many pin behaviour the restructure drops; each is tagged with the +step that rewrites it, and the safety-net scenarios (DEVELOPMENT "Tests") +must keep passing through every step. Open questions below are decided during the restructure. How: incremental on `main` (one system extracted per step, every dev build usable), `signals` plus constructor wiring through one composition root (no new framework), dev builds `1.3.0-dev.N`, everyone moves to diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 7d272d5..69686e3 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -44,6 +44,10 @@ abstract class RelayConstants { static const int stateHeartbeatSecs = 1; static const int pingIntervalSecs = 5; + /// Server liveness defaults ([Liveness]): a member silent this long is + /// stale, then removed. + static const int staleAfterSecs = 15; + static const int dropAfterSecs = 30; // Base64-char count of the avatar string, enforced at hello and // profile. 256 KiB leaves wide headroom over any 128x128 pipeline PNG diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index f49cbab..d757a73 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -9,6 +9,25 @@ import 'queue.dart'; typedef SyncClock = DateTime Function(); +/// How long the server bears a silent member. [staleAfter] and [dropAfter] +/// are measured on the engine's clock ([SyncEngine.clock]) by its 1 s +/// sweep, so a test with a fake clock steps through them without waiting; +/// [pingInterval] is the WebSocket's own timer (real time). +class Liveness { + /// No frame for longer than this: stale (out of the set, feed `stale`). + final Duration staleAfter; + /// No frame for longer than this: the member is removed (feed `leave`) + /// and its socket closed. + final Duration dropAfter; + /// WebSocket ping on every `/sync` socket. + final Duration pingInterval; + const Liveness({ + this.staleAfter = const Duration(seconds: RelayConstants.staleAfterSecs), + this.dropAfter = const Duration(seconds: RelayConstants.dropAfterSecs), + this.pingInterval = const Duration(seconds: RelayConstants.pingIntervalSecs), + }); +} + enum Transport { playing, held, finished } enum Correction { none, slowDown, seek } @@ -141,8 +160,9 @@ class EngineFeed { class SyncEngine { final SyncClock clock; - final Duration staleAfter; - final Duration dropAfter; + final Liveness liveness; + Duration get staleAfter => liveness.staleAfter; + Duration get dropAfter => liveness.dropAfter; final EngineFeed feed = EngineFeed(); final Map members = {}; @@ -167,10 +187,8 @@ class SyncEngine { int _seekSeq = 0; int _seekEventId = -1; - SyncEngine({SyncClock? clock, Duration? staleAfter, Duration? dropAfter}) + SyncEngine({SyncClock? clock, this.liveness = const Liveness()}) : clock = clock ?? (() => DateTime.now().toUtc()), - staleAfter = staleAfter ?? const Duration(seconds: 15), - dropAfter = dropAfter ?? const Duration(seconds: 30), serverTime = (clock ?? (() => DateTime.now().toUtc()))(); DateTime get now => clock(); diff --git a/packages/core/lib/src/testing/flaky_proxy.dart b/packages/core/lib/src/testing/flaky_proxy.dart new file mode 100644 index 0000000..660bc64 --- /dev/null +++ b/packages/core/lib/src/testing/flaky_proxy.dart @@ -0,0 +1,275 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +/// Which way bytes go through a [FlakyLink]: [up] client to server, +/// [down] server to client. +enum Flow { up, down, both } + +/// A TCP proxy for tests that need a bad link between a client and a +/// server (DEVELOPMENT.md "Tests"): every accepted connection is a +/// [FlakyLink] the test steers while it runs: pass through, blackhole (a +/// half-open link), added latency, a late chunk, a cut after N bytes, a +/// reset. It knows nothing about WebSocket or HTTP; a "frame" is one chunk +/// as it was read from the socket (on loopback that is usually one write, +/// e.g. one small WebSocket frame). +class FlakyProxy { + final String upstreamHost; + final int upstreamPort; + final ServerSocket _server; + + /// Every connection so far, in accept order. + final List links = []; + + /// Called with each new link before any byte goes through it, e.g. to + /// blackhole or reset the next connection. + void Function(FlakyLink link)? onConnect; + + final _accepted = StreamController.broadcast(sync: true); + + FlakyProxy._(this._server, this.upstreamHost, this.upstreamPort, this.onConnect) { + _server.listen(_accept); + } + + /// Listens on a free loopback port and forwards to + /// [upstreamHost]:[upstreamPort]. + static Future start(int upstreamPort, + {String upstreamHost = '127.0.0.1', void Function(FlakyLink link)? onConnect}) async { + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + return FlakyProxy._(server, upstreamHost, upstreamPort, onConnect); + } + + /// The port clients dial instead of the server's. + int get port => _server.port; + + /// Each new link as it is accepted. + Stream get connections => _accepted.stream; + + /// The latest link, null before the first. + FlakyLink? get last => links.isEmpty ? null : links.last; + + Future _accept(Socket client) async { + final link = FlakyLink._(client); + links.add(link); + onConnect?.call(link); + _accepted.add(link); + if (link.isClosed) return; + try { + final upstream = await Socket.connect(upstreamHost, upstreamPort); + link._attach(upstream); + } catch (_) { + link._close(rst: true); + } + } + + /// Stops accepting and closes every link. + Future close() async { + await _server.close(); + for (final l in links) { + l._close(rst: false); + } + await _accepted.close(); + } +} + +/// One proxied connection. Modes change at once and can be combined +/// ([delay] with [holdNext], say); [pass] clears them all. +class FlakyLink { + final Socket _client; + Socket? _upstream; + late final _Pipe _up = _Pipe(this, Flow.up); + late final _Pipe _down = _Pipe(this, Flow.down); + final _ready = Completer(); + final _closed = Completer(); + bool _blackholed = false; + + /// Latency added to every chunk, each way (on top of the link's own; a + /// chunk never overtakes an earlier one). + Duration delay = Duration.zero; + + FlakyLink._(this._client) { + _quiet(_client); + _client.listen(_up._onData, onDone: _up._onDone, onError: (_) => _close(rst: false)); + } + + void _attach(Socket upstream) { + _upstream = upstream; + _quiet(upstream); + if (isClosed) { + upstream.destroy(); + return; + } + upstream.listen(_down._onData, onDone: _down._onDone, onError: (_) => _close(rst: false)); + _ready.complete(); + } + + static void _quiet(Socket s) { + try { + s.setOption(SocketOption.tcpNoDelay, true); + } catch (_) {} + s.done.then((_) {}, onError: (_) {}); + } + + /// Bytes delivered client to server / server to client. + int get bytesUp => _up.forwarded; + int get bytesDown => _down.forwarded; + + /// Bytes swallowed by [blackhole] (and by a cut), both ways. + int get dropped => _up.dropped + _down.dropped; + + bool get isClosed => _closed.isCompleted; + + /// Completes when the link is gone (both ends closed, a reset, a cut). + Future get closed => _closed.future; + + /// Plain pass-through again: no blackhole, delay, held chunk or cut. + /// Bytes swallowed meanwhile stay lost (a WebSocket on it is then + /// usually broken). + void pass() { + _blackholed = false; + delay = Duration.zero; + for (final p in [_up, _down]) { + p._holdNext = null; + p._cutLeft = null; + } + } + + /// Keeps both sockets open and drops every byte both ways, a closing + /// end included: neither side learns anything (a half-open link, like a + /// laptop lid closed or a NAT forgetting the flow). Chunks already + /// delayed are dropped too. + void blackhole() => _blackholed = true; + + bool get blackholed => _blackholed; + + /// The next chunk going [flow] (and, as TCP keeps order, everything + /// behind it) arrives [late] later than it would. + void holdNext(Duration late, {Flow flow = Flow.down}) { + for (final p in _pipes(flow)) { + p._holdNext = late; + } + } + + /// Delivers [bytes] more going [flow] (counted per direction; with + /// [Flow.both] whichever gets there first), then resets the link. + void cutAfter(int bytes, {Flow flow = Flow.both}) { + for (final p in _pipes(flow)) { + p._cutLeft = bytes; + } + if (bytes <= 0) reset(); + } + + /// Aborts both ends at once: an RST where the platform lets us ask for + /// one (SO_LINGER 0), otherwise an abrupt close. Nothing queued is sent. + void reset() => _close(rst: true); + + List<_Pipe> _pipes(Flow flow) => switch (flow) { + Flow.up => [_up], + Flow.down => [_down], + Flow.both => [_up, _down], + }; + + void _close({required bool rst}) { + if (isClosed) return; + _closed.complete(); + for (final s in [_client, _upstream]) { + if (s == null) continue; + if (rst) _lingerZero(s); + s.destroy(); + } + } + + static void _lingerZero(Socket s) { + try { + // struct linger { int l_onoff = 1; int l_linger = 0; } + final v = ByteData(8) + ..setInt32(0, 1, Endian.host) + ..setInt32(4, 0, Endian.host); + final soLinger = Platform.isLinux || Platform.isAndroid ? 13 : 0x80; + s.setRawOption(RawSocketOption(RawSocketOption.levelSocket, soLinger, v.buffer.asUint8List())); + } catch (_) {} + } + + void _bothDone() { + if (_up._srcDone && _down._srcDone) _close(rst: false); + } +} + +class _Pipe { + final FlakyLink link; + final Flow flow; + _Pipe(this.link, this.flow); + + int forwarded = 0; + int dropped = 0; + int? _cutLeft; + Duration? _holdNext; + bool _srcDone = false; + Future? _chainTail; + + Socket? get _dst => flow == Flow.up ? link._upstream : link._client; + + Future get _chain => _chainTail ??= link._ready.future; + + void _then(Future Function() step) { + _chainTail = _chain.then((_) => step()).catchError((Object _) => link._close(rst: false)); + } + + void _onData(Uint8List data) { + if (link.isClosed) return; + if (link._blackholed) { + dropped += data.length; + return; + } + var chunk = data; + var cut = false; + final left = _cutLeft; + if (left != null) { + if (chunk.length >= left) { + dropped += chunk.length - left; + chunk = Uint8List.sublistView(chunk, 0, left); + _cutLeft = 0; + cut = true; + } else { + _cutLeft = left - chunk.length; + } + } + var due = DateTime.now().add(link.delay); + final hold = _holdNext; + if (hold != null) { + _holdNext = null; + due = due.add(hold); + } + _then(() async { + final wait = due.difference(DateTime.now()); + if (wait > Duration.zero) await Future.delayed(wait); + if (link.isClosed) return; + if (link._blackholed) { + dropped += chunk.length; + return; + } + final dst = _dst!; + if (chunk.isNotEmpty) { + dst.add(chunk); + forwarded += chunk.length; + } + if (cut) { + await dst.flush(); + link.reset(); + } + }); + } + + void _onDone() { + _srcDone = true; + if (link._blackholed) { + link._bothDone(); + return; + } + _then(() async { + // Half-close: the far end reads its end of stream, may still answer. + if (!link.isClosed && !link._blackholed) unawaited(_dst?.close().then((_) {}, onError: (_) {})); + link._bothDone(); + }); + } +} diff --git a/packages/core/lib/testing.dart b/packages/core/lib/testing.dart new file mode 100644 index 0000000..d2b7785 --- /dev/null +++ b/packages/core/lib/testing.dart @@ -0,0 +1,5 @@ +/// Test support shared by core, server and desktop tests; not part of +/// `core.dart` and never imported by app or server code. +library; + +export 'src/testing/flaky_proxy.dart'; diff --git a/packages/core/test/flaky_proxy_test.dart b/packages/core/test/flaky_proxy_test.dart new file mode 100644 index 0000000..f2184f5 --- /dev/null +++ b/packages/core/test/flaky_proxy_test.dart @@ -0,0 +1,134 @@ +// The flaky-link proxy itself, against a loopback echo server. +import 'dart:async'; +import 'dart:io'; + +import 'package:sync_relay_core/testing.dart'; +import 'package:test/test.dart'; + +/// A client socket through the proxy, collecting what comes back. +class _Client { + final Socket socket; + final received = []; + final done = Completer(); + _Client(this.socket) { + socket.done.then((_) {}, onError: (_) {}); + socket.listen(received.addAll, + onDone: () => done.isCompleted ? null : done.complete(null), + onError: (Object e) => done.isCompleted ? null : done.complete(e)); + } + + static Future<_Client> dial(FlakyProxy p) async => _Client(await Socket.connect('127.0.0.1', p.port)); + + void send(String s) => socket.add(s.codeUnits); + String get text => String.fromCharCodes(received); +} + +Future until(bool Function() ok, {Duration limit = const Duration(seconds: 3)}) async { + final end = DateTime.now().add(limit); + while (!ok()) { + if (DateTime.now().isAfter(end)) fail('condition not met in $limit'); + await Future.delayed(const Duration(milliseconds: 10)); + } +} + +void main() { + late ServerSocket echo; + late FlakyProxy proxy; + + setUp(() async { + echo = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + echo.listen((s) { + s.done.then((_) {}, onError: (_) {}); + s.listen(s.add, onDone: () => s.close().then((_) {}, onError: (_) {}), onError: (_) {}); + }); + proxy = await FlakyProxy.start(echo.port); + }); + + tearDown(() async { + await proxy.close(); + await echo.close(); + }); + + test('passes bytes both ways and counts them', () async { + final c = await _Client.dial(proxy); + c.send('hello'); + await until(() => c.text == 'hello'); + expect(proxy.links, hasLength(1)); + expect(proxy.last!.bytesUp, 5); + expect(proxy.last!.bytesDown, 5); + await c.socket.close(); + await proxy.last!.closed.timeout(const Duration(seconds: 3)); + }); + + test('blackhole drops everything both ways and keeps the socket open', () async { + final c = await _Client.dial(proxy); + c.send('a'); + await until(() => c.text == 'a'); + proxy.last!.blackhole(); + c.send('bcd'); + await Future.delayed(const Duration(milliseconds: 200)); + expect(c.text, 'a'); + expect(proxy.last!.dropped, 3); + expect(c.done.isCompleted, isFalse); + expect(proxy.last!.isClosed, isFalse); + }); + + test('delay holds every chunk', () async { + final c = await _Client.dial(proxy); + await until(() => proxy.last != null); + proxy.last!.delay = const Duration(milliseconds: 150); + final sw = Stopwatch()..start(); + c.send('x'); + await until(() => c.text == 'x'); + // Up and down each wait 150 ms. + expect(sw.elapsedMilliseconds, greaterThanOrEqualTo(290)); + }); + + test('holdNext delivers one chunk late and keeps order', () async { + final c = await _Client.dial(proxy); + await until(() => proxy.last != null); + proxy.last!.holdNext(const Duration(milliseconds: 300), flow: Flow.up); + final sw = Stopwatch()..start(); + c.send('1'); + await Future.delayed(const Duration(milliseconds: 50)); + c.send('2'); + await Future.delayed(const Duration(milliseconds: 100)); + expect(c.text, '', reason: 'both still held'); + await until(() => c.text == '12'); + expect(sw.elapsedMilliseconds, greaterThanOrEqualTo(290)); + }); + + test('cutAfter delivers N bytes then resets', () async { + final c = await _Client.dial(proxy); + await until(() => proxy.last != null); + proxy.last!.cutAfter(3, flow: Flow.up); + c.send('abcdef'); + await c.done.future.timeout(const Duration(seconds: 3)); + expect(proxy.last!.bytesUp, 3); + expect(proxy.last!.isClosed, isTrue); + }); + + test('reset ends the client at once', () async { + final c = await _Client.dial(proxy); + c.send('a'); + await until(() => c.text == 'a'); + proxy.last!.reset(); + await c.done.future.timeout(const Duration(seconds: 3)); + expect(proxy.last!.isClosed, isTrue); + }); + + test('onConnect can steer a link before any byte goes through', () async { + proxy.onConnect = (l) => l.reset(); + // The RST can beat the connect's own completion. + try { + final c = await _Client.dial(proxy); + await c.done.future.timeout(const Duration(seconds: 3)); + expect(c.text, ''); + } on Object catch (e) { + // dart:io throws the bare OSError here (ECONNRESET), on some + // platforms a SocketException. + expect(e, anyOf(isA(), isA())); + } + expect(proxy.last!.isClosed, isTrue); + }); +} diff --git a/packages/core/test/sync_engine_test.dart b/packages/core/test/sync_engine_test.dart index 1e1f070..e5968fc 100644 --- a/packages/core/test/sync_engine_test.dart +++ b/packages/core/test/sync_engine_test.dart @@ -2,7 +2,7 @@ // groups "sitting out", "membership + gate", "presence gate", "stale ready // rejoins" and the member rules of "swap" pin today's presence (out flag, // stale sweep, leave on drop); presence In/Away/Gone with grace replaces -// them. The clock, correction, end-of-file and wire groups stay. +// them ("liveness on the engine clock" then steps through the grace). The clock, correction, end-of-file and wire groups stay. import 'dart:convert'; import 'package:sync_relay_core/core.dart'; @@ -618,6 +618,35 @@ void main() { }); }); + group('liveness on the engine clock', () { + test('stale and drop follow Liveness and the fake clock, not wall time', () { + engine = SyncEngine( + clock: clock, + liveness: const Liveness(staleAfter: Duration(minutes: 5), dropAfter: Duration(minutes: 10))); + join('a', 'alice'); + engine.applyReady('a', true); + engine.feed.drain(); + t = t.add(const Duration(minutes: 5)); + engine.tick(); + expect(engine.members['a']!.inSet, isTrue, reason: 'not past staleAfter yet'); + t = t.add(const Duration(seconds: 1)); + engine.tick(); + expect(engine.members['a']!.inSet, isFalse); + expect(engine.feed.drain().map((f) => f.action), ['stale']); + t = t.add(const Duration(minutes: 4, seconds: 59)); + expect(engine.sweepDropped(), isEmpty, reason: 'exactly dropAfter is not past it'); + t = t.add(const Duration(seconds: 1)); + expect(engine.sweepDropped(), ['a']); + expect(engine.feed.drain().map((f) => f.action), ['leave']); + }); + + test('defaults are today\'s 15 s / 30 s', () { + expect(engine.staleAfter, const Duration(seconds: 15)); + expect(engine.dropAfter, const Duration(seconds: 30)); + expect(engine.liveness.pingInterval, const Duration(seconds: 5)); + }); + }); + group('end of file', () { setUp(() { engine.media = const HelloMedia(name: 'f', size: 1, duration: 20); diff --git a/packages/core/test/watch_session_test.dart b/packages/core/test/watch_session_test.dart index 6093c39..bafac97 100644 --- a/packages/core/test/watch_session_test.dart +++ b/packages/core/test/watch_session_test.dart @@ -1,4 +1,4 @@ -// restructure: rewrite in step 5 (D3: the watched rule moves from position coverage to time present) +// restructure: rewrite in step 6 (D3: the watched rule moves from position coverage to time present) import 'package:sync_relay_core/core.dart'; import 'package:test/test.dart'; diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 0930f00..c747359 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -519,7 +519,7 @@ void mountSyncRoute( ? '${req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host}:${req.url.port}' : (req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host); return WebSocketUpgrade((final ws) async { - ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs); + ws.pingInterval = hub.engine.liveness.pingInterval; final firstText = Completer(); String? token; var welcomed = false; @@ -534,7 +534,7 @@ void mountSyncRoute( void handleIntent(String text) { final session = hub.session; final viewer = token == null ? null : session.tokens[token]; - viewer?.touch(); + viewer?.touch(hub.engine.now); Map decoded; try { final parsed = jsonDecode(text); diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index cfd7424..39fb5fc 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -65,9 +65,13 @@ class SyncServer { /// The session log while running with a [logDir]. SessionLog? get sessionLog => _sessionLog; - /// [ytDlp] replaces the server's own (tests). - SyncServer(this.config, {YtDlp? ytDlp, this.loopbackOnly = false, this.logDir}) { + /// [ytDlp] replaces the server's own (tests). [clock] and [liveness] + /// go to the engine: a test passes a fake clock to step through the + /// silence limits ([Liveness]) without waiting. + SyncServer(this.config, + {YtDlp? ytDlp, this.loopbackOnly = false, this.logDir, SyncClock? clock, Liveness liveness = const Liveness()}) { hub = ServerHub(sessions, config.secret, + engine: SyncEngine(clock: clock, liveness: liveness), hostKey: config.hostKey, jellyfin: config.jellyfin, ytDlp: ytDlp ?? YtDlp(serverYtDlpDir(config.ytDlpDir), enabled: config.ytDlp)); diff --git a/packages/server/test/flaky_link_test.dart b/packages/server/test/flaky_link_test.dart new file mode 100644 index 0000000..7b38bbb --- /dev/null +++ b/packages/server/test/flaky_link_test.dart @@ -0,0 +1,132 @@ +// restructure: rewrite in step 1 (Connection + Identity + Presence): both +// tests pin today's rules (a silent member is stale, then removed; a reset +// socket loses its member, so the redial is a new one, P1); with grace they +// expect `gone` and the same member back. +// +// A client reaching the server through FlakyProxy (DEVELOPMENT.md "Tests"). +import 'dart:async'; +import 'dart:convert'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_core/testing.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; + +/// A bare `/sync` client: hello, then a report every 100 ms (any frame +/// keeps a member fresh). +class _Client { + final WebSocketChannel ch; + final frames = >[]; + bool ended = false; + Timer? _reports; + late final StreamSubscription _sub; + + _Client(int port, {String resume = ''}) + : ch = WebSocketChannel.connect(Uri.parse('ws://127.0.0.1:$port/sync')) { + ch.ready.then((_) {}, onError: (_) {}); + _sub = ch.stream.listen((m) => frames.add(Map.from(jsonDecode('$m') as Map)), + onDone: () => ended = true, onError: (_) => ended = true); + ch.sink.add(jsonEncode(HelloMessage( + proto: RelayConstants.protoVersion, + secret: 'shh', + user: const HelloUser(name: 'alice'), + media: const HelloMedia(name: 'f.bin', size: 1, duration: 10.0), + resume: resume) + .toJson())); + } + + Map? get welcome => frames.where((f) => f['t'] == 'welcome').firstOrNull; + String get token => welcome!['token'] as String; + String get memberId => welcome!['memberId'] as String; + + void startReports() => _reports = Timer.periodic(const Duration(milliseconds: 100), (_) { + if (!ended) { + ch.sink.add(jsonEncode(const ReportMessage(position: 0, buffered: 0, eventId: 0).toJson())); + } + }); + + Future close() async { + _reports?.cancel(); + try { + await _sub.cancel(); + } catch (_) {} + try { + await ch.sink.close().timeout(const Duration(seconds: 1)); + } catch (_) {} + } +} + +Future until(bool Function() ok, String what, {Duration limit = const Duration(seconds: 6)}) async { + final end = DateTime.now().add(limit); + while (!ok()) { + if (DateTime.now().isAfter(end)) fail('$what: not within $limit'); + await Future.delayed(const Duration(milliseconds: 20)); + } +} + +void main() { + late SyncServer server; + late FlakyProxy proxy; + final clients = <_Client>[]; + + setUp(() async { + // Short limits so the demo runs in a couple of seconds. + server = SyncServer(AppConfig(port: 0, secret: 'shh', name: 'test'), + liveness: const Liveness(staleAfter: Duration(milliseconds: 600), dropAfter: Duration(milliseconds: 1500))); + await server.start(); + proxy = await FlakyProxy.start(server.boundPort); + }); + + tearDown(() async { + for (final c in clients) { + await c.close(); + } + clients.clear(); + await proxy.close(); + await server.stop(); + }); + + _Client dial({String resume = ''}) { + final c = _Client(proxy.port, resume: resume); + clients.add(c); + return c; + } + + test('a blackholed client goes stale, then is removed; it never hears of it', () async { + final c = dial(); + await until(() => c.welcome != null, 'welcome'); + c.startReports(); + final engine = server.hub.engine; + EngineMember? member() => engine.members[c.token]; + await Future.delayed(const Duration(milliseconds: 1200)); + expect(engine.isStale(member()!), isFalse, reason: 'reports through the proxy keep it fresh'); + + proxy.last!.blackhole(); + final seen = c.frames.length; + await until(() => member() != null && engine.isStale(member()!), 'stale'); + await until(() => member() == null, 'removed'); + expect(server.hub.sockets.containsKey(c.token), isFalse); + // Half-open: the server's close never reached the client. + await Future.delayed(const Duration(milliseconds: 300)); + expect(c.ended, isFalse); + expect(c.frames.length, seen); + expect(proxy.last!.isClosed, isFalse); + expect(proxy.last!.dropped, greaterThan(0)); + }); + + test('a reset client sees the drop at once and redials; today it comes back as a new member', () async { + final c = dial(); + await until(() => c.welcome != null, 'welcome'); + final first = c.memberId; + proxy.last!.reset(); + await until(() => c.ended, 'client sees the reset'); + await until(() => !server.hub.engine.members.containsKey(c.token), 'server forgets the socket'); + + final again = dial(resume: c.token); + await until(() => again.welcome != null, 'welcome after redial'); + expect(proxy.links, hasLength(2)); + expect(again.memberId, isNot(first), reason: 'P1: no grace yet'); + expect(server.hub.engine.members, hasLength(1)); + }); +}