Sessions leave redacted logs; viewers wait for a member relay's source before opening it

Logs: one redaction pass (query strings to names, member tokens,
passwords, cookies, auth headers) for every line; daily files with size
and age caps in the app (hosted server included) and the dedicated
server (--log-dir); swaps, relay channels, per-member /r /m /j outcomes,
server fetches, mpv load errors, yt-dlp runs. Settings > About: Copy
diagnostics, Open logs folder.

Bilibili through someone else: the viewer opened /r/ before the source
dialled in; the server holds that request and mpv gives up after 5 s
("could not be opened"). Viewers now open a member relay once it is
live, the putter starts its source before its own player, and track
sizes are read in parallel.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
foxyzandClaude Opus 5.5 committed 2026-10-12 19:05:32 +03:00
1 parent 572104f49e
commit 4ab37c55cf
25 files changed
+1122 -106

No files matched your search

+5
View File
@@ -236,6 +236,7 @@ then everyone (you included) uses Join.
| `--[no-]yt-dlp` | on | read web pages (YouTube, X, …) with yt-dlp when the server fetches a link | | `--[no-]yt-dlp` | on | read web pages (YouTube, X, …) with yt-dlp when the server fetches a link |
| `--yt-dlp-dir` | next to the binary | writable folder for the server's yt-dlp (systemd: the unit's `StateDirectory`) | | `--yt-dlp-dir` | next to the binary | writable folder for the server's yt-dlp (systemd: the unit's `StateDirectory`) |
| `--yt-dlp-cookies` | none | cookies.txt or cookie-extension JSON export for that yt-dlp (sites that refuse guests); copied into the yt-dlp folder at start, removed from it when not given. **An account login**: keep it private, use a throwaway account for YouTube | | `--yt-dlp-cookies` | none | cookies.txt or cookie-extension JSON export for that yt-dlp (sites that refuse guests); copied into the yt-dlp folder at start, removed from it when not given. **An account login**: keep it private, use a throwaway account for YouTube |
| `--log-dir` | `logs` next to `--config`, else next to the binary (systemd: the unit's `StateDirectory`) | folder for the log files (`server-YYYY-MM-DD.log`, kept 7 days; passwords, tokens and cookies left out) |
| `--config` | none | YAML or JSON file with the same settings (`port`, `secret`, `name`, `mediaPath`, `advertiseHost`, `ytDlp`, `ytDlpDir`, `ytDlpCookies`); replaces the flags | | `--config` | none | YAML or JSON file with the same settings (`port`, `secret`, `name`, `mediaPath`, `advertiseHost`, `ytDlp`, `ytDlpDir`, `ytDlpCookies`); replaces the flags |
| `--upgrade` | | update the install folder to the latest release ([Install and update](#install-and-update)), then exit | | `--upgrade` | | update the install folder to the latest release ([Install and update](#install-and-update)), then exit |
| `--check-update` | | print whether a newer release exists, then exit | | `--check-update` | | print whether a newer release exists, then exit |
@@ -246,6 +247,10 @@ 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 [deploy/watch-together.service](deploy/watch-together.service) (copy it to
`/etc/systemd/system/`, edit the flags, `systemctl enable --now watch-together`). `/etc/systemd/system/`, edit the flags, `systemctl enable --now watch-together`).
Something not playing? Settings → About → **Copy diagnostics** puts the
app's recent log (passwords, tokens and cookies left out) on the
clipboard: paste it to whoever helps you, from both sides if you can.
## Join ## Join
Join in the strip. Servers you got into are listed first (up to 6), with Join in the strip. Servers you got into are listed first (up to 6), with
+2
View File
@@ -9,6 +9,7 @@ import 'package:window_manager/window_manager.dart';
import 'app.dart'; import 'app.dart';
import 'src/playback/yt_dlp_app.dart'; import 'src/playback/yt_dlp_app.dart';
import 'src/platform/app_log.dart';
import 'src/platform/desktop_entry.dart'; import 'src/platform/desktop_entry.dart';
import 'src/platform/window_memory.dart'; import 'src/platform/window_memory.dart';
import 'src/screens/home/ui.dart'; import 'src/screens/home/ui.dart';
@@ -41,6 +42,7 @@ Future<void> main(List<String> args) async {
), ),
); );
WidgetsFlutterBinding.ensureInitialized(); WidgetsFlutterBinding.ensureInitialized();
await startAppLogging();
await Hive.initFlutter(); await Hive.initFlutter();
await AppSettings.instance.ensureLoaded(); await AppSettings.instance.ensureLoaded();
await windowManager.ensureInitialized(); await windowManager.ensureInitialized();
@@ -0,0 +1,98 @@
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 '../playback/yt_dlp_app.dart';
import '../screens/home/ui.dart' show appName, displayVersion;
import '../state/session_state.dart';
// The app's log on disk (<app support>/logs/app-YYYY-MM-DD.log; core's
// logging.dart writes, rotates and redacts it), debugPrint routed into it,
// and the "Copy diagnostics" text of Settings → About.
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<void> startAppLogging() async {
try {
final base = await getApplicationSupportDirectory();
final dir = '${base.path}${Platform.pathSeparator}logs';
if (startLogFiles(dir) != null) appLogDir = dir;
} catch (e) {
_appLog.warning('no log files: $e');
}
// Everything the app prints for diagnosis lands in the file too (the
// log's console output replaces the print).
debugPrint = (String? message, {int? wrapWidth}) {
if (message == null || message.isEmpty) return;
final bad = RegExp(r'fail|error|crash|exception', caseSensitive: false).hasMatch(message);
bad ? _appLog.warning(message) : _appLog.info(message);
};
PlatformDispatcher.instance.onError = (error, stack) {
_appLog.severe('uncaught: $error', error, stack);
return false;
};
_appLog.info('$appName ${displayVersion()} on ${Platform.operatingSystem} '
'${Platform.operatingSystemVersion}, pid $pid, Dart ${Platform.version.split(' ').first}');
}
/// The text "Copy diagnostics" gives: versions, then the last [since] of
/// the app log (its hosted server's records included), already redacted.
Future<String> diagnosticsText({Duration since = const Duration(hours: 6)}) async {
final b = StringBuffer()
..writeln('$appName ${displayVersion()}')
..writeln('OS: ${Platform.operatingSystem} ${Platform.operatingSystemVersion}')
..writeln('Dart: ${Platform.version.split(' ').first}');
try {
final mpv = await SessionState.instance.coordinator.playback.mpvVersion().timeout(const Duration(seconds: 2));
b.writeln('mpv: ${mpv ?? 'no player yet'}');
} catch (_) {
b.writeln('mpv: unknown');
}
try {
final y = await AppYtDlp.instance.ytDlp.timeout(const Duration(seconds: 2));
final cookies = y.cookies.read(count: false);
b.writeln('yt-dlp: ${y.version ?? 'not ready'}${y.lastError == null ? '' : ' (${y.lastError})'}, '
'cookies: ${cookies == null ? 'none' : 'set'}');
} catch (_) {
b.writeln('yt-dlp: unknown');
}
final files = logFiles;
b.writeln('Log: ${files?.currentPath ?? 'none (console only)'}');
b.writeln('--- last ${since.inHours} h of the log ---');
b.write(files?.recent(since) ?? '');
return redactLog(b.toString());
}
/// Saves [text] next to the logs as `diagnostics-<time>.txt`; its path,
/// or null when it could not be written.
Future<String?> saveDiagnostics(String text) async {
final dir = appLogDir;
if (dir == null) return null;
final t = DateTime.now();
String two(int n) => n.toString().padLeft(2, '0');
final path = '$dir${Platform.pathSeparator}diagnostics-${t.year}${two(t.month)}${two(t.day)}-'
'${two(t.hour)}${two(t.minute)}${two(t.second)}.txt';
try {
// Older ones go after the log files' week.
for (final e in Directory(dir).listSync()) {
if (e is File &&
e.uri.pathSegments.last.startsWith('diagnostics-') &&
t.difference(e.statSync().modified) > const Duration(days: 7)) {
e.deleteSync();
}
}
await File(path).writeAsString(text);
return path;
} catch (e) {
_appLog.warning('cannot save diagnostics: $e');
return null;
}
}
@@ -8,7 +8,7 @@ import 'package:flutter/widgets.dart';
import 'package:media_kit/media_kit.dart'; import 'package:media_kit/media_kit.dart';
import 'package:media_kit_video/media_kit_video.dart'; import 'package:media_kit_video/media_kit_video.dart';
import 'package:signals/signals_flutter.dart'; import 'package:signals/signals_flutter.dart';
import 'package:sync_relay_core/core.dart' show JellyfinStreamType; import 'package:sync_relay_core/core.dart' show JellyfinStreamType, LogThrottle, createLogger;
import '../jellyfin/track_prefs.dart' show MpvTrack; import '../jellyfin/track_prefs.dart' show MpvTrack;
import '../shaders/shader_assets.dart'; import '../shaders/shader_assets.dart';
@@ -361,6 +361,7 @@ class PlaybackController {
// an error there while shaders are on is a shader that failed to load // an error there while shaders are on is a shader that failed to load
// or compile (mpv then renders without it). // or compile (mpv then renders without it).
void _onLog(PlayerLog log) { void _onLog(PlayerLog log) {
_toAppLog(log);
if (log.prefix == 'ytdl_hook' && (log.level == 'error' || log.level == 'fatal')) { if (log.prefix == 'ytdl_hook' && (log.level == 'error' || log.level == 'fatal')) {
final text = log.text.trim().replaceFirst(RegExp(r'^ERROR:\s*'), ''); final text = log.text.trim().replaceFirst(RegExp(r'^ERROR:\s*'), '');
if (text.isNotEmpty) { if (text.isNotEmpty) {
@@ -380,6 +381,8 @@ class PlaybackController {
if (!_loadErrorSent && _loadFailed(log, duration.peek())) { if (!_loadErrorSent && _loadFailed(log, duration.peek())) {
_loadErrorSent = true; _loadErrorSent = true;
final http = _httpError; final http = _httpError;
_mpvLog.warning('cannot open the media: [${log.prefix}] ${log.text.trim()}'
'${http == null ? ' (no HTTP error logged; a request that timed out after 5 s looks like this)' : ' ($http)'}');
onLoadError?.call(http == null ? 'the video stream could not be opened' : 'the video server answered $http'); onLoadError?.call(http == null ? 'the video stream could not be opened' : 'the video server answered $http');
return; return;
} }
@@ -402,6 +405,23 @@ class PlaybackController {
} }
} }
// mpv's warnings and errors (the session player's; ffmpeg's through
// whichever player logs them) go to the app log, a few per kind at a
// time: a refused request makes ffmpeg retry and warn on every try.
static final _mpvLog = createLogger('mpv');
static final _mpvThrottle = LogThrottle(burst: 5);
static void _toAppLog(PlayerLog log, {String player = ''}) {
final level = log.level;
if (level != 'warn' && level != 'error' && level != 'fatal') return;
final text = log.text.trim();
if (text.isEmpty) return;
final pass = _mpvThrottle.allow('$player${log.prefix} $level ${text.length > 40 ? text.substring(0, 40) : text}');
if (pass == null) return;
final line = '$pass$player[${log.prefix}] $text';
level == 'warn' ? _mpvLog.info(line) : _mpvLog.warning(line);
}
/// [log] says the media could not be opened: before a [duration] is /// [log] says the media could not be opened: before a [duration] is
/// known, mpv's fatal "nothing to play" or a demuxer error, or a yt-dlp /// known, mpv's fatal "nothing to play" or a demuxer error, or a yt-dlp
/// hook error. /// hook error.
@@ -438,6 +458,19 @@ class PlaybackController {
} }
} }
/// `mpv-version` and `ffmpeg-version` of the session player, for
/// diagnostics; null when there is no player.
Future<String?> mpvVersion() async {
final p = _player;
if (p == null) return null;
try {
final dynamic platform = p.platform;
return '${await platform?.getProperty('mpv-version')}, ffmpeg ${await platform?.getProperty('ffmpeg-version')}';
} catch (_) {
return null;
}
}
/// Opens [url] paused. [headers] go with every request for it (mpv /// Opens [url] paused. [headers] go with every request for it (mpv
/// `http-header-fields`): a Jellyfin token travels there, not in the URL. /// `http-header-fields`): a Jellyfin token travels there, not in the URL.
/// Each media gets its own player: when [preload] holds the same [url] /// Each media gets its own player: when [preload] holds the same [url]
@@ -708,6 +741,7 @@ class PlaybackController {
// that is gone (a promotion disposes it) the next one created, which // that is gone (a promotion disposes it) the next one created, which
// can be this preload. Its HTTP errors still name the session's. // can be this preload. Its HTTP errors still name the session's.
if (log.prefix == 'ffmpeg') return _onLog(log); if (log.prefix == 'ffmpeg') return _onLog(log);
_toAppLog(log, player: 'preload ');
if (_loadFailed(log, p.state.duration)) pre.fail(); if (_loadFailed(log, p.state.duration)) pre.fail();
}); });
try { try {
@@ -1,12 +1,14 @@
import 'package:flutter/material.dart'; import 'package:flutter/material.dart';
import '../../platform/app_log.dart';
import '../../update/update_controller.dart'; import '../../update/update_controller.dart';
import '../../widgets/copy_open.dart'; import '../../widgets/copy_open.dart';
import 'settings_page.dart'; import 'settings_page.dart';
import 'ui.dart'; import 'ui.dart';
import 'update_indicator.dart'; import 'update_indicator.dart';
/// Name, version, update state and the source link (Settings → About). /// Name, version, update state, the source link and the logs (Settings →
/// About).
class AboutPanel extends StatelessWidget { class AboutPanel extends StatelessWidget {
final UpdateController? updates; final UpdateController? updates;
const AboutPanel({super.key, this.updates}); const AboutPanel({super.key, this.updates});
@@ -41,8 +43,45 @@ class AboutPanel extends StatelessWidget {
), ),
], ],
), ),
const SizedBox(height: kGap),
Text('Something went wrong? Copy diagnostics and send them to whoever helps you: '
'versions and the last hours of the log, passwords, tokens and cookies left out.',
style: t.bodySmall?.copyWith(color: Colors.grey)),
const SizedBox(height: 8),
Wrap(
spacing: 8,
runSpacing: 8,
children: [
FilledButton.tonalIcon(
onPressed: _copyDiagnostics,
icon: const Icon(Icons.bug_report_outlined, size: 18),
label: const Text('Copy diagnostics'),
),
OutlinedButton.icon(
onPressed: appLogDir == null ? null : () => _openLogs(),
icon: const Icon(Icons.folder_open, size: 18),
label: const Text('Open logs folder'),
),
],
),
], ],
), ),
); );
} }
static Future<void> _copyDiagnostics() async {
final text = await diagnosticsText();
final saved = await saveDiagnostics(text);
await copyToClipboard(text,
done: saved == null ? 'Diagnostics copied' : 'Diagnostics copied (also saved in the logs folder)');
}
static Future<void> _openLogs() async {
final dir = appLogDir;
if (dir == null) return;
// The file manager opens a folder path the way a browser opens a link.
if (!await openInBrowser(Uri.directory(dir).toString())) {
await copyToClipboard(dir, done: "Couldn't open it — folder path copied");
}
}
} }
@@ -33,6 +33,8 @@ import 'now_playing.dart';
import 'now_playing_info.dart'; import 'now_playing_info.dart';
import 'switch_trace.dart'; import 'switch_trace.dart';
final _log = createLogger('session');
/// Thrown when a connect is superseded by cancel/leave before it finished. /// Thrown when a connect is superseded by cancel/leave before it finished.
class ConnectCancelled implements Exception { class ConnectCancelled implements Exception {
const ConnectCancelled(); const ConnectCancelled();
@@ -387,7 +389,7 @@ class SessionCoordinator {
} }
void _setPlayerError(String raw) { void _setPlayerError(String raw) {
debugPrint('player cannot open media $_mediaId: $raw'); _log.warning('player cannot open media $_mediaId: $raw');
_playerErrorFor = _mediaId; _playerErrorFor = _mediaId;
_playerErrorDuration = playback.duration.peek(); _playerErrorDuration = playback.duration.peek();
playerError.value = playErrorText(raw); playerError.value = playErrorText(raw);
@@ -563,6 +565,8 @@ class SessionCoordinator {
/// the player held before (a reconnect) keeps it. /// the player held before (a reconnect) keeps it.
Future<bool> _openMedia(String url, {Map<String, String>? headers}) async { Future<bool> _openMedia(String url, {Map<String, String>? headers}) async {
if (url == _openUrl) return true; if (url == _openUrl) return true;
_log.info('media $_mediaId: open ${_urlKind(url, headers)}: ${logUrl(url)}'
'${headers == null || headers.isEmpty ? '' : ' (headers: ${headers.keys.join(', ')})'}');
final keepPlayer = _playerMediaId == _openedFor; final keepPlayer = _playerMediaId == _openedFor;
_openedFor = _playerMediaId; _openedFor = _playerMediaId;
_openUrl = url; _openUrl = url;
@@ -610,6 +614,23 @@ class SessionCoordinator {
_sendReport(r); _sendReport(r);
} }
/// What [url] is, for the log.
String _urlKind(String url, Map<String, String>? headers) {
if (url.startsWith('edl://')) return 'relay (tracks)';
final u = Uri.tryParse(url);
if (u == null || !(u.isScheme('http') || u.isScheme('https'))) return 'local file';
if (u.host == _wsHost && u.port == _wsPort && u.pathSegments.isNotEmpty) {
return switch (u.pathSegments.first) {
'r' => 'relay',
'm' => "the host's file",
'j' => "Jellyfin through the server's login",
_ => 'server URL',
};
}
if (headers != null && headers.isNotEmpty) return 'Jellyfin, own sign-in';
return 'link, fetched here';
}
/// A web link (not this server's): mpv's yt-dlp hook gets the app's /// A web link (not this server's): mpv's yt-dlp hook gets the app's
/// managed yt-dlp, downloaded first if needed; a media file name does /// managed yt-dlp, downloaded first if needed; a media file name does
/// not wait for that. /// not wait for that.
@@ -656,6 +677,12 @@ class SessionCoordinator {
/// The server-fetched media this viewer opens once it is live. /// The server-fetched media this viewer opens once it is live.
int _awaitLiveFor = -1; int _awaitLiveFor = -1;
/// A relay nobody feeds yet: the server's fetch before it is read, or a
/// member's before its channel is up while that member is still here
/// (once it left, the server's buffer is all there is: open that).
bool _waitForSource(StateBroadcast r, MediaRelay relay) =>
!relay.live && (relay.fetch || r.members.any((m) => m.id == relay.source));
/// Relayed through a member whose channel is down: say who everyone is /// Relayed through a member whose channel is down: say who everyone is
/// waiting for (the buffer still plays what it has). /// waiting for (the buffer still plays what it has).
void _relayStatus(StateBroadcast r) { void _relayStatus(StateBroadcast r) {
@@ -881,21 +908,27 @@ class SessionCoordinator {
memberRelayUrl: memberRelayUrl:
relay != null && !mine ? _relayMediaUrlFor(name, r.mediaId, const [], relay.queueId) : null); relay != null && !mine ? _relayMediaUrlFor(name, r.mediaId, const [], relay.queueId) : null);
} else if (origin is _FileOrigin) { } else if (origin is _FileOrigin) {
await _openMedia(origin.path); // The source first: viewers wait on it, not on this player.
_startRelaySource(r.mediaId, RelayMedia.single(FileRelayBytes(File(origin.path))), _startRelaySource(r.mediaId, RelayMedia.single(FileRelayBytes(File(origin.path))),
announce: true, queueId: relay?.queueId ?? ''); announce: true, queueId: relay?.queueId ?? '');
await _openMedia(origin.path);
} else if (origin is _LinkOrigin) { } else if (origin is _LinkOrigin) {
await _openMedia(origin.url);
_startRelaySource( _startRelaySource(
r.mediaId, LinkRelayMedia(Uri.parse(origin.url), ytDlp: origin.ytDlp, probe: origin.probe), r.mediaId, LinkRelayMedia(Uri.parse(origin.url), ytDlp: origin.ytDlp, probe: origin.probe),
announce: true, queueId: relay?.queueId ?? ''); announce: true, queueId: relay?.queueId ?? '');
await _openMedia(origin.url);
} else if (relay != null && !mine) { } else if (relay != null && !mine) {
if (relay.fetch && !relay.live) { if (_waitForSource(r, relay)) {
// The server is still getting it (yt-dlp takes seconds; mpv // The server is still getting it (yt-dlp takes seconds), or the
// gives up on a silent request after 5): open once it is live. // member passing it on has not dialled in yet (its app opens the
// link first): mpv gives up on a silent request after 5 s, so
// open once it is live.
_openUrl = null; _openUrl = null;
_awaitLiveFor = r.mediaId; _awaitLiveFor = r.mediaId;
health.value = const HealthStatus(HealthSeverity.info, 'the server is getting the video…'); final who = r.members.where((m) => m.id == relay.source).firstOrNull?.name;
_log.info('media ${r.mediaId}: waiting for ${relay.fetch ? 'the server' : who} before opening the relay');
health.value = HealthStatus(HealthSeverity.info,
relay.fetch ? 'the server is getting the video…' : 'waiting for $who to pass the video on…');
} else { } else {
await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks, relay.queueId)); await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks, relay.queueId));
} }
@@ -915,7 +948,7 @@ class SessionCoordinator {
_seekGate.reset(); _seekGate.reset();
_clearPendingSeek(); _clearPendingSeek();
if (_mediaId == r.mediaId) settledMediaId.value = r.mediaId; if (_mediaId == r.mediaId) settledMediaId.value = r.mediaId;
} else if (_awaitLiveFor == r.mediaId && (stateMediaRelay(r)?.live ?? false)) { } else if (_awaitLiveFor == r.mediaId && stateMediaRelay(r) != null && !_waitForSource(r, stateMediaRelay(r)!)) {
_awaitLiveFor = -1; _awaitLiveFor = -1;
final relay = stateMediaRelay(r)!; final relay = stateMediaRelay(r)!;
await _openMedia( await _openMedia(
@@ -57,9 +57,9 @@ class ServerIsolate {
if (msg is! Map) return; if (msg is! Map) return;
switch (msg['t']) { switch (msg['t']) {
case 'log': case 'log':
final err = msg['error']; final err = msg['error'], stack = msg['stack'];
onLog('${msg['level']} ${msg['logger']}: ${msg['message']}' onLog('${msg['level']} ${msg['logger']}: ${msg['message']}'
'${err == null ? '' : ' ($err)'}'); '${err == null ? '' : ' ($err)'}${stack == null ? '' : '\n$stack'}');
case 'ready': case 'ready':
if (!startup.isCompleted) { if (!startup.isCompleted) {
startup.complete((msg['port'] as int, msg['stop'] as SendPort)); startup.complete((msg['port'] as int, msg['stop'] as SendPort));
@@ -379,7 +379,8 @@ class SessionState {
ytDlpDir: await _sharedYtDlpDir(), ytDlpDir: await _sharedYtDlpDir(),
), ),
loopbackOnly: soloSession, loopbackOnly: soloSession,
onLog: (line) => debugPrint('[server] $line'), // Its records join the app's log as `server.<logger>`.
onLog: (line) => relogLine('server', line),
onCrash: (cause) => _onServerCrash(server, cause), onCrash: (cause) => _onServerCrash(server, cause),
); );
} catch (e, st) { } catch (e, st) {
+41
View File
@@ -1,4 +1,5 @@
import 'dart:async'; import 'dart:async';
import 'dart:convert';
import 'dart:io'; import 'dart:io';
import 'package:flutter_test/flutter_test.dart'; import 'package:flutter_test/flutter_test.dart';
@@ -169,6 +170,46 @@ void main() {
} }
}); });
test('a source that dials in after mpv would have given up (5 s): viewers wait for it, then play',
timeout: const Timeout(Duration(minutes: 2)), () async {
final file = await _media();
final alice = await member('alice');
// Bob is a bare client: he puts a file up and starts passing it on
// only 7 s later (a slow first upstream read, a slow own player).
final ws = await WebSocket.connect('ws://127.0.0.1:${server.boundPort}/sync');
final welcome = Completer<Map>();
ws.listen((f) {
final j = jsonDecode('$f') as Map;
if (j['t'] == 'welcome' && !welcome.isCompleted) welcome.complete(j);
}, onError: (_) {});
ws.add(jsonEncode(const HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'pw',
user: HelloUser(name: 'bob'),
media: HelloMedia(name: '', size: 0, duration: 0))
.toJson()));
final token = (await welcome.future.timeout(const Duration(seconds: 5)))['token'] as String;
final size = await File(file).length();
ws.add(jsonEncode({
't': 'swap',
'media': {'name': 'late.mkv', 'size': size, 'duration': 0, 'relay': <String, Object?>{}},
}));
await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source.isNotEmpty ?? false, 'relay state');
await Future.delayed(const Duration(seconds: 7));
expect(alice.playerError.peek(), isEmpty, reason: 'no "could not be opened" while the source is on its way');
expect(alice.health.peek().message, contains('waiting for bob'));
final src = RelaySource(Uri.parse('ws://127.0.0.1:${server.boundPort}/r/default/$token/source'),
RelayMedia.single(FileRelayBytes(File(file))));
await src.start();
try {
await _poll(() => alice.playback.durationSecs > 19, 'alice opened it once the source was up');
expect(alice.playerError.peek(), isEmpty);
} finally {
await src.close();
await ws.close();
}
});
test('the source drops uncleanly and resumes: it keeps its copy and keeps serving', test('the source drops uncleanly and resumes: it keeps its copy and keeps serving',
timeout: const Timeout(Duration(minutes: 2)), () async { timeout: const Timeout(Duration(minutes: 2)), () async {
final file = await _media(); final file = await _media();
+73 -4
View File
@@ -33,8 +33,8 @@ size (`ServerIsolate.setFile`): the host's in-session File swap then sends a
- **Hosting from the app**: `SessionState.host()` spawns an isolate running - **Hosting from the app**: `SessionState.host()` spawns an isolate running
`serve` (`ServerIsolate`), waits for `ready`, then joins it over `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 `127.0.0.1` like any other client. Its log records go to the app's log
`[server]` prefix. The isolate runs with `errorsAreFatal: false`: an as `server.<logger>` (see "Logging"). The isolate runs with `errorsAreFatal: false`: an
uncaught async error is logged and the server keeps going. If the isolate uncaught async error is logged and the server keeps going. If the isolate
dies anyway, the session shows "server crashed: ..." and leaves; the dies anyway, the session shows "server crashed: ..." and leaves; the
shell says so in a dialog (during setup the host attempt fails with it). shell says so in a dialog (during setup the host attempt fails with it).
@@ -61,6 +61,7 @@ size (`ServerIsolate.setFile`): the host's in-session File swap then sends a
and shipped next to the app as `watch-together-server`. It parses flags and shipped next to the app as `watch-together-server`. It parses flags
(`--port`, `--secret`, `--name`, `--media`, `--advertise-host`, (`--port`, `--secret`, `--name`, `--media`, `--advertise-host`,
`--jellyfin <file>`, `--[no-]yt-dlp`, `--yt-dlp-dir`, `--yt-dlp-cookies`, `--config`; default port 8080) into the same map and calls `serve` `--jellyfin <file>`, `--[no-]yt-dlp`, `--yt-dlp-dir`, `--yt-dlp-cookies`, `--config`; default port 8080) into the same map and calls `serve`
(`--log-dir` only picks where its log files go, see "Logging")
(`--upgrade` / `--check-update` run the [updater](#updater) instead); (`--upgrade` / `--check-update` run the [updater](#updater) instead);
SIGINT and SIGTERM stop it. A Flutter binary cannot run without a SIGINT and SIGTERM stop it. A Flutter binary cannot run without a
display, which is why the server is its own build. display, which is why the server is its own build.
@@ -621,6 +622,16 @@ How bytes move:
with none. A manual swap to something else keeps it (the item is with none. A manual swap to something else keeps it (the item is
still first). A failed warm fetch is not promoted: the swap fetches still first). A failed warm fetch is not promoted: the swap fetches
again and reports the reason. again and reports the reason.
- Before the source's channel is first up, a viewer request gets no
answer (the server cannot send headers before it knows the size), and
media_kit's mpv gives up on a silent request after 5 s
(`network-timeout`) with "could not be opened". So viewers open a
member relay's `/r/` URL only once `relay.live` is set, as for a
server fetch (`_waitForSource`; "waiting for <name> to pass the video
on…" meanwhile), unless the source already left (then they open what
the server has); Up next preloads wait the same way. The putter starts
its source before opening its own player, and the source reads its
tracks' sizes at once.
- Source away: viewers keep playing from the buffer; a request waits up - Source away: viewers keep playing from the buffer; a request waits up
to 60 s for missing bytes. The source reconnects with backoff (same to 60 s for missing bytes. The source reconnects with backoff (same
token after a resume); a different size (or piece count and roles) token after a resume); a different size (or piece count and roles)
@@ -784,8 +795,7 @@ all three link modes. Nothing is bundled or installed by hand.
cannot); when the server cannot read it, the putter gets an `err` with cannot); when the server cannot read it, the putter gets an `err` with
the reason. Viewers open a fetched the reason. Viewers open a fetched
link's `/r/` URL only once `relay.live` is set ("the server is getting link's `/r/` URL only once `relay.live` is set ("the server is getting
the video…" until then): resolving takes seconds and media_kit's mpv the video…" until then; see "Relay" for why).
gives up on a silent request after 5 s (`network-timeout`).
## Jellyfin in a session ## Jellyfin in a session
@@ -1492,6 +1502,65 @@ Window chrome is frameless (`window_manager`): the party strip is the drag
area and holds the min/max/close buttons (`widgets/window_buttons.dart`); area and holds the min/max/close buttons (`widgets/window_buttons.dart`);
fullscreen has no drag area and shows the buttons in the player's top bar. fullscreen has no drag area and shows the buttons in the player's top bar.
## Logging
`packages/core/lib/src/logging.dart`, one setup for the app and the
dedicated server (`package:logging` loggers named by area: `relay`,
`yt-dlp`, `link`, `server_hub`, `media`, `queue`, `session`, `mpv`,
`app`).
- **Where**: the app writes `<app support>/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.<logger>` (`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
prints the folder at start. Both also print every line to stdout.
- **Files** (`LogFiles`): one per day, the next `.1.log`, `.2.log` once a
file passes 10 MiB; files older than 7 days, then the oldest beyond
100 MiB in all, are deleted whenever one is opened. A line is
`2026-10-12 18:44:27.202 LEVEL logger: message`, the error and stack
trace on the lines below. Writes are synchronous appends (nothing lost
in a crash). Lines from before the folder is known (the app finds it
once Flutter is up) are kept and written first.
- **Redaction** (`redactLog`): every line, in one place
(`formatLogLine`), before console or file: URL query strings cut to
their parameter names (signed CDN links, Jellyfin `api_key`),
`user:pass@` in URLs, member tokens in `/r /m /j /a /i` paths, and the
values of passwords, secrets, tokens, API keys, cookies,
`Authorization` / `X-Emby-*` headers (`name=value`, `"name":"value"`,
`Name="value"`, `Name: value`) and `Bearer` credentials. Code logs
members as `name (id)`, never their token, headers by name only, and
yt-dlp's cookie source as `--cookies <file>`.
- **Floods** (`LogThrottle`): a few lines per key per 30 s, the next one
says how many were dropped: mpv's warnings and errors, upstream
refusals per host and status, each member's media requests per route
and outcome.
- **What is logged**: the server: joins and leaves, every swap (who, how
it plays, a link with its query cut), relay channels up and down (who,
tracks and sizes, how long, the close frame), relay live changes, warm
relays created, promoted and dropped, server fetches (start, tracks,
time, failure reason), each member's `/r /m /j` requests (status,
range, bytes, time to headers: a 503 after 60 s is a source that never
came up), broken viewer reads, a member's `report.error` as it changes,
queue ops, and relay stats every 30 s (source rate, need, buffered,
each reader's rate). The app: `debugPrint` (routed into the log at
start), uncaught errors, each media opened (relay, the host's file,
Jellyfin, a link fetched here), mpv's warnings and errors (`mpv`; why a
media could not be opened, and whether ffmpeg logged an HTTP status),
probes, yt-dlp runs (arguments, time, exit code, the tail of stderr,
the picked formats with protocol, host and header names), the relay
source (media opened and how long it took, channel up and down,
reconnects, upstream refusals with the header names sent, re-resolves
of expired format URLs).
- **Getting it**: Settings → About → "Copy diagnostics" puts the app and
OS versions, mpv/ffmpeg and yt-dlp versions, whether yt-dlp has
cookies, and the last 6 hours of the log (up to 4 MiB) on the
clipboard, already redacted, and saves the same text as
`diagnostics-<time>.txt` in the logs folder (kept 7 days); "Open logs
folder" opens it.
## Health reporting ## Health reporting
Every failure must show in the UI with a cause and a next step; nothing Every failure must show in the UI with a cause and a next step; nothing
+19 -2
View File
@@ -127,8 +127,8 @@ updater skips the check when absent. Workflows live in `.gitea/workflows`
When hosting, the server runs on its own isolate (`ServerIsolate`, When hosting, the server runs on its own isolate (`ServerIsolate`,
`apps/desktop/lib/src/state/server_isolate.dart`), in the same process. `apps/desktop/lib/src/state/server_isolate.dart`), in the same process.
Breakpoints in `packages/server` work under `flutter run`: the debugger Breakpoints in `packages/server` work under `flutter run`: the debugger
attaches to spawned isolates too (it shows as `server`). Its log lines are attaches to spawned isolates too (it shows as `server`). Its log lines
printed with a `[server]` prefix. Hot reload reloads the server isolate's carry `server.<logger>` (e.g. `server.relay`). Hot reload reloads the server isolate's
code too, but its state (members, clock, sockets) survives until hosting code too, but its state (members, clock, sockets) survives until hosting
restarts; changes to startup code (`SyncServer.start`, routes) need a new restarts; changes to startup code (`SyncServer.start`, routes) need a new
Host. Hot restart kills the server isolate (flutter_tools kills every Host. Hot restart kills the server isolate (flutter_tools kills every
@@ -136,6 +136,23 @@ non-UI isolate first), so a `[server] server crashed: the server stopped
unexpectedly` line just before it is expected. Without the app: unexpectedly` line just before it is expected. Without the app:
`dart run packages/server/bin/server.dart`. `dart run packages/server/bin/server.dart`.
## Logs
After a session, read the log files rather than reproducing
("Logging" in [ARCHITECTURE.md](ARCHITECTURE.md) has what goes in them):
the app's are `~/.local/share/moe.senkochan.watch_together/logs/app-*.log`
on Linux (on Windows under `%APPDATA%`; Settings → About → "Open logs
folder" finds it anywhere), with the hosted server's lines
in the same file; a dedicated server prints its folder at start
(`--log-dir`). Users send theirs with Settings → About → "Copy
diagnostics". For a "can't play" report, grep both sides around the
time: `swap by` (what went up and how), `relay channel` (when the
source dialled in), `media:` (each viewer's requests: status and time to
headers), `mpv:` (`cannot open the media` and ffmpeg's lines), `yt-dlp:`
(the resolve and its picked formats) and `upstream` (CDN refusals with
the headers sent). A viewer that got no bytes while the source was
slow to come up shows mpv's error with "no HTTP error logged".
## Gotchas ## Gotchas
- Relic gives Range as a typed `RangeHeader`; convert from unit + first range. - Relic gives Range as a typed `RangeHeader`; convert from unit + first range.
+16
View File
@@ -4,6 +4,7 @@ import 'dart:io';
import 'dart:typed_data'; import 'dart:typed_data';
import 'constants.dart'; import 'constants.dart';
import 'logging.dart';
import 'relay.dart'; import 'relay.dart';
import 'relay_pieces.dart'; import 'relay_pieces.dart';
import 'yt_dlp.dart'; import 'yt_dlp.dart';
@@ -170,6 +171,8 @@ class LinkProbe {
: null; : null;
} }
final _log = createLogger('link');
const _probeLimit = 512 * 1024; const _probeLimit = 512 * 1024;
/// Finds out what [url] is: a `Range: bytes=0-0` GET; a media file is /// Finds out what [url] is: a `Range: bytes=0-0` GET; a media file is
@@ -177,6 +180,17 @@ const _probeLimit = 512 * 1024;
/// yt-dlp cannot read it, the page's own title. Throws /// yt-dlp cannot read it, the page's own title. Throws
/// [LinkProbeException] when the link cannot be reached at all. /// [LinkProbeException] when the link cannot be reached at all.
Future<LinkProbe> probeLink(Uri url, {YtDlp? ytDlp, HttpClient? client}) async { Future<LinkProbe> probeLink(Uri url, {YtDlp? ytDlp, HttpClient? client}) async {
final p = await _probeLink(url, ytDlp: ytDlp, client: client);
_log.info('probe ${logUrl(url)}: ${p.isPage ? 'page' : 'file of ${p.size} bytes'}'
'${p.status == 0 ? '' : ', HTTP ${p.status}'}'
'${p.info == null ? '' : ', yt-dlp read it'}'
'${p.ytDlpError == null ? '' : ', yt-dlp: ${p.ytDlpError}'}'
'${p.problem == null ? '' : '; problem: ${p.problem}'}'
'${p.relayProblem == null ? '' : '; relay: ${p.relayProblem}'}');
return p;
}
Future<LinkProbe> _probeLink(Uri url, {YtDlp? ytDlp, HttpClient? client}) async {
final c = client ?? (HttpClient()..connectionTimeout = const Duration(seconds: 10)); final c = client ?? (HttpClient()..connectionTimeout = const Duration(seconds: 10));
try { try {
String? body; String? body;
@@ -210,6 +224,7 @@ Future<LinkProbe> probeLink(Uri url, {YtDlp? ytDlp, HttpClient? client}) async {
} }
} }
if (getError != null) { if (getError != null) {
_log.warning('probe ${logUrl(url)}: cannot open it: $getError${ytErr == null ? '' : ' (yt-dlp: $ytErr)'}');
throw LinkProbeException('cannot open the link: $getError${ytErr == null ? '' : ' (yt-dlp: $ytErr)'}'); throw LinkProbeException('cannot open the link: $getError${ytErr == null ? '' : ' (yt-dlp: $ytErr)'}');
} }
// A server honouring the Range sent one byte of the page: get it all. // A server honouring the Range sent one byte of the page: get it all.
@@ -300,6 +315,7 @@ class PageRelayMedia implements RelayMedia {
Future<void> _refresh(int gen) { Future<void> _refresh(int gen) {
if (gen != _gen) return Future.value(); if (gen != _gen) return Future.value();
return _refreshing ??= () async { return _refreshing ??= () async {
_log.info('relay of ${logUrl(page)}: a format URL expired (403/410), resolving the page again');
try { try {
const changed = LinkProbeException('the video changed on the page'); const changed = LinkProbeException('the video changed on the page');
final info = await _info(); final info = await _info();
+361 -13
View File
@@ -1,19 +1,367 @@
import 'dart:convert';
import 'dart:io';
import 'package:logging/logging.dart'; import 'package:logging/logging.dart';
// One log for the app (its hosted server's records included) and one for
// the dedicated server: every record goes through [redactLog] once, then
// to the console and, once [startLogFiles] ran, to a rotating file
// (ARCHITECTURE "Logging").
Logger createLogger(String name) => Logger(name); Logger createLogger(String name) => Logger(name);
void initLogging({Level level = Level.INFO}) { var _listening = false;
var _console = true;
LogFiles? _files;
// Lines logged before the files are open (the app finds its folder only
// after Flutter is up): written first once they are.
final _early = <String>[];
const _earlyMax = 2000;
/// The open log files, once [startLogFiles] ran.
LogFiles? get logFiles => _files;
/// Routes every log record to the console ([console]) and, after
/// [startLogFiles], to the files. Safe to call again (only [level] and
/// [console] change).
void initLogging({Level level = Level.INFO, bool console = true}) {
Logger.root.level = level; Logger.root.level = level;
Logger.root.onRecord.listen((r) { _console = console;
// ignore: avoid_print if (_listening) return;
print('${r.time.toIso8601String()} ${r.level.name} ${r.loggerName}: ${r.message}'); _listening = true;
if (r.error != null) { Logger.root.onRecord.listen((r) =>
// ignore: avoid_print writeLogLine(formatLogLine(r.time, r.level.name, r.loggerName, r.message, error: r.error, stack: r.stackTrace)));
print(' error: ${r.error}'); }
}
if (r.stackTrace != null) { /// Writes the files in [dir] from now on: `<prefix>-YYYY-MM-DD.log`,
// ignore: avoid_print /// rotated by day and [LogFiles.maxFileBytes], pruned by age and total
print('${r.stackTrace}'); /// size. Lines logged before go in first. A folder that cannot be written
} /// leaves logging on the console only (and says so there).
}); LogFiles? startLogFiles(String dir, {String prefix = 'app'}) {
if (_files != null) return _files;
try {
final f = LogFiles(dir, prefix: prefix)..open();
_files = f;
for (final l in _early) {
f.write(l);
}
_early.clear();
return f;
} catch (e) {
// ignore: avoid_print
print('log files off: cannot write $dir ($e)');
_early.clear();
return null;
}
}
/// One finished, redacted line to the console and the file.
void writeLogLine(String line) {
if (_console) {
// ignore: avoid_print
print(line);
}
final f = _files;
if (f != null) {
f.write(line);
} else if (_early.length < _earlyMax) {
_early.add(line);
}
}
/// `2026-10-12 18:44:27.202 WARNING relay: message`, then the error and
/// stack trace on their own lines; all of it through [redactLog].
String formatLogLine(DateTime time, String level, String logger, String message, {Object? error, Object? stack}) {
final b = StringBuffer('${logTimestamp(time)} $level $logger: $message');
if (error != null) b.write('\n error: $error');
final s = '${stack ?? ''}'.trimRight();
if (s.isNotEmpty) b.write('\n$s');
return redactLog(b.toString());
}
String _two(int n) => n.toString().padLeft(2, '0');
/// Local time as the log writes it (`2026-10-12 18:44:27.202`).
String logTimestamp(DateTime t) => '${t.year}-${_two(t.month)}-${_two(t.day)} '
'${_two(t.hour)}:${_two(t.minute)}:${_two(t.second)}.${t.millisecond.toString().padLeft(3, '0')}';
/// The time a log line starts with, or null for a continuation line.
DateTime? logLineTime(String line) {
final m = _lineTime.firstMatch(line);
if (m == null) return null;
return DateTime.tryParse('${m[1]}T${m[2]}');
}
final _lineTime = RegExp(r'^(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}\.\d{3}) ');
/// A record from another isolate (the hosted server's, `LEVEL logger:
/// message`) logged here under [tag]: `server.relay` at its own level.
void relogLine(String tag, String line) {
final m = RegExp(r'^([A-Z]+) ([^:\s]+): ').firstMatch(line);
final level = Level.LEVELS.where((l) => l.name == m?[1]).firstOrNull;
if (m == null || level == null) {
Logger(tag).info(line);
return;
}
Logger('$tag.${m[2]}').log(level, line.substring(m.end));
}
// ------------------------------------------------------------- redaction
// Names whose value is a secret wherever they appear as `name=value`,
// `"name":"value"` or `Name="value"` (Jellyfin's MediaBrowser header).
const _secretNames = r'(?:x-emby-token|x-mediabrowser-token|x-emby-authorization|authorization|set-cookie|cookie|'
r'api_?key|access_?token|refresh_?token|id_?token|auth_?token|token|password|passwd|secret|host_?key|'
r'sessdata|bili_jct|client_?secret)';
final _jsonSecret = RegExp('"($_secretNames)"\\s*:\\s*"(?:[^"\\\\]|\\\\.)*"', caseSensitive: false);
final _quotedSecret = RegExp('\\b($_secretNames)\\s*=\\s*"[^"]*"', caseSensitive: false);
// Header-like `Name: value` runs to the end of the line (a cookie
// header's `;`-separated pairs, MediaBrowser's comma list).
final _headerSecret = RegExp(
'\\b(authorization|cookie|set-cookie|x-emby-authorization|x-emby-token|x-mediabrowser-token)\\s*:\\s*[^\\n]*',
caseSensitive: false);
final _pairSecret = RegExp('\\b($_secretNames)\\s*=\\s*[^&\\s"\',;)}\\]]+', caseSensitive: false);
final _bearer = RegExp(r'\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]{8,}', caseSensitive: false);
final _url = RegExp(r'''\b(?:https?|wss?)://[^\s"'<>]+''', caseSensitive: false);
// A member token in a server path: /r/<sid>/<token>/..., /m, /j, /a, /i.
final _pathToken = RegExp(r'(/[rmjai]/[^/\s?#]+/)[0-9a-fA-F]{32}\b');
final _userInfo = RegExp(r'^([a-z]+://)[^/@\s]+@', caseSensitive: false);
/// [s] with nothing secret left: every URL's query string down to its
/// parameter names (signed CDN links, `api_key`), user:password@ in URLs,
/// member tokens in `/r/ /m/ /j/ /a/ /i/` paths, and the values of
/// passwords, tokens, API keys, cookies and auth headers. Applied to
/// every log line ([formatLogLine]).
String redactLog(String s) {
if (s.isEmpty) return s;
var out = s.replaceAllMapped(_url, (m) => _redactUrl(m[0]!));
out = out.replaceAllMapped(_pathToken, (m) => '${m[1]}***');
out = out.replaceAllMapped(_jsonSecret, (m) => '"${m[1]}":"***"');
out = out.replaceAllMapped(_headerSecret, (m) => '${m[1]}: ***');
out = out.replaceAllMapped(_quotedSecret, (m) => '${m[1]}="***"');
out = out.replaceAllMapped(_pairSecret, (m) => '${m[1]}=***');
out = out.replaceAllMapped(_bearer, (m) => '${m[1]} ***');
return out;
}
String _redactUrl(String url) {
var u = url.replaceFirstMapped(_userInfo, (m) => '${m[1]}***@');
final hash = u.indexOf('#');
if (hash >= 0) u = u.substring(0, hash);
final q = u.indexOf('?');
if (q < 0) return u;
final names = u
.substring(q + 1)
.split('&')
.map((p) => p.split('=').first)
.where((n) => n.isNotEmpty)
.map((n) => n.length > 40 ? '${n.substring(0, 40)}…' : n)
.toList();
return '${u.substring(0, q)}?${names.isEmpty ? '' : '${names.join('&')}=…'}';
}
/// [url] for a log line: scheme, host and path, the query cut to its
/// parameter names (see [redactLog]).
String logUrl(Object? url) => redactLog('$url');
// --------------------------------------------------------------- flood control
/// At most [burst] lines per key in each [window]; the next line let
/// through says how many were dropped. For floods (ffmpeg retrying a
/// refused request, a viewer's range reads).
class LogThrottle {
final int burst;
final Duration window;
final DateTime Function() _now;
final _keys = <String, ({DateTime start, int count, int dropped})>{};
LogThrottle({this.burst = 5, this.window = const Duration(seconds: 30), DateTime Function()? now})
: _now = now ?? DateTime.now;
/// Null to drop the line; else what to put before it: '' or
/// `(12 more like this dropped) `.
String? allow(String key) {
final now = _now();
final k = _keys[key];
if (k == null || now.difference(k.start) >= window) {
final dropped = k?.dropped ?? 0;
_keys.remove(key);
_keys[key] = (start: now, count: 1, dropped: 0);
if (_keys.length > 500) _keys.remove(_keys.keys.first);
return dropped > 0 ? '($dropped more like this dropped) ' : '';
}
if (k.count < burst) {
_keys[key] = (start: k.start, count: k.count + 1, dropped: k.dropped);
return '';
}
_keys[key] = (start: k.start, count: k.count, dropped: k.dropped + 1);
return null;
}
}
// ------------------------------------------------------------------ files
/// Rotating log files in [dir]: `<prefix>-YYYY-MM-DD.log` (then `.1.log`,
/// `.2.log` ... once one passes [maxFileBytes]); files older than
/// [keepDays] and the oldest beyond [maxTotalBytes] are deleted whenever a
/// file is opened.
class LogFiles {
final String dir;
final String prefix;
final int maxFileBytes;
final int maxTotalBytes;
final int keepDays;
final DateTime Function() _now;
LogFiles(this.dir,
{this.prefix = 'app',
this.maxFileBytes = 10 * 1024 * 1024,
this.maxTotalBytes = 100 * 1024 * 1024,
this.keepDays = 7,
DateTime Function()? now})
: _now = now ?? DateTime.now;
RandomAccessFile? _out;
String? _day;
var _size = 0;
String? _path;
/// The file being written.
String? get currentPath => _path;
String _dayOf(DateTime t) => '${t.year}-${_two(t.month)}-${_two(t.day)}';
String _dayOfFile(File f) => f.uri.pathSegments.last.substring(prefix.length + 1, prefix.length + 11);
/// Opens today's file (creating [dir]); throws when it cannot.
void open() {
Directory(dir).createSync(recursive: true);
_roll(_dayOf(_now()));
}
void _roll(String day) {
_out?.closeSync();
_out = null;
_day = day;
for (var i = 0;; i++) {
final path = '$dir${Platform.pathSeparator}$prefix-$day${i == 0 ? '' : '.$i'}.log';
final f = File(path);
final size = f.existsSync() ? f.lengthSync() : 0;
if (size < maxFileBytes) {
_out = f.openSync(mode: FileMode.append);
_size = size;
_path = path;
break;
}
}
prune();
}
/// Appends [line]; a new day or a full file moves on to the next file.
void write(String line) {
try {
final day = _dayOf(_now());
if (_out == null || day != _day || _size >= maxFileBytes) _roll(day);
final bytes = utf8.encode('$line\n');
_out!.writeFromSync(bytes);
_size += bytes.length;
} catch (_) {
// A full disk or a removed folder must not take the app down.
}
}
/// This prefix's files, oldest first.
List<File> files() {
final d = Directory(dir);
if (!d.existsSync()) return const [];
final name = RegExp('^${RegExp.escape(prefix)}-(\\d{4}-\\d{2}-\\d{2})(?:\\.(\\d+))?\\.log\$');
final found = <(String, int, File)>[];
for (final e in d.listSync()) {
if (e is! File) continue;
final m = name.firstMatch(e.uri.pathSegments.last);
if (m != null) found.add((m[1]!, int.tryParse(m[2] ?? '') ?? 0, e));
}
found.sort((a, b) => a.$1 != b.$1 ? a.$1.compareTo(b.$1) : a.$2.compareTo(b.$2));
return [for (final f in found) f.$3];
}
/// Deletes files older than [keepDays], then the oldest while all of
/// them pass [maxTotalBytes]; never the one being written.
void prune() {
final cutoff = _dayOf(_now().subtract(Duration(days: keepDays)));
final kept = <File>[];
for (final f in files()) {
if (_dayOfFile(f).compareTo(cutoff) < 0 && f.path != _path) {
_delete(f);
} else {
kept.add(f);
}
}
var total = kept.fold<int>(0, (a, f) => a + _len(f));
for (final f in kept) {
if (total <= maxTotalBytes) break;
if (f.path == _path) continue;
total -= _len(f);
_delete(f);
}
}
static int _len(File f) {
try {
return f.lengthSync();
} catch (_) {
return 0;
}
}
static void _delete(File f) {
try {
f.deleteSync();
} catch (_) {}
}
/// The lines of the last [since] (a record's continuation lines with
/// it), oldest first; at most about [maxBytes], the newest kept.
String recent(Duration since, {int maxBytes = 4 * 1024 * 1024}) {
final from = _now().subtract(since);
final fromDay = _dayOf(from);
final out = <String>[];
var size = 0;
var done = false;
for (final f in files().reversed) {
if (done || _dayOfFile(f).compareTo(fromDay) < 0) break;
List<String> lines;
try {
lines = const LineSplitter().convert(utf8.decode(f.readAsBytesSync(), allowMalformed: true));
} catch (_) {
continue;
}
// Newest first; a record's continuation lines come before its head.
final block = <String>[];
for (final l in lines.reversed) {
block.add(l);
final t = logLineTime(l);
if (t == null) continue;
if (t.isBefore(from)) {
done = true;
break;
}
for (final b in block) {
size += b.length + 1;
out.add(b);
}
block.clear();
if (size >= maxBytes) {
done = true;
break;
}
}
}
return out.reversed.join('\n');
}
void close() {
try {
_out?.closeSync();
} catch (_) {}
_out = null;
}
} }
+35 -5
View File
@@ -35,6 +35,8 @@ import 'relay_pieces.dart';
// the tracks; a cancelled pull's unacked bytes are written off. // the tracks; a cancelled pull's unacked bytes are written off.
final _log = createLogger('relay'); final _log = createLogger('relay');
final _upstreamThrottle = LogThrottle(burst: 3);
final _readThrottle = LogThrottle(burst: 3);
/// The `<name>` segment of a `/r/` URL: the media name URL-encoded, its /// The `<name>` segment of a `/r/` URL: the media name URL-encoded, its
/// slashes made `_` first (a page title can hold a URL, and an encoded /// slashes made `_` first (a page title can hold a URL, and an encoded
@@ -104,6 +106,15 @@ class UpstreamHttpException extends HttpException {
UpstreamHttpException(this.status, Uri url) : super('upstream HTTP $status', uri: url); UpstreamHttpException(this.status, Uri url) : super('upstream HTTP $status', uri: url);
} }
/// Logs an upstream answer that is not the media (status, what was
/// asked, the names of the headers sent), a few per host and status at a
/// time.
void logUpstreamRefusal(int status, Uri url, String what, Iterable<String> headerNames) {
final pass = _upstreamThrottle.allow('${url.host} $status');
if (pass == null) return;
_log.warning('${pass}upstream $status for $what of ${logUrl(url)} (sent headers: ${headerNames.join(', ')})');
}
/// An HTTP(S) resource with Range support, e.g. a link or a Jellyfin /// An HTTP(S) resource with Range support, e.g. a link or a Jellyfin
/// stream; [headers] (auth) never leave this machine. A read is one GET /// stream; [headers] (auth) never leave this machine. A read is one GET
/// per [chunkSize] bytes, back to back (see /// per [chunkSize] bytes, back to back (see
@@ -127,6 +138,7 @@ class HttpRelayBytes implements RelayBytes {
final res = await req.close().timeout(const Duration(seconds: 20)); final res = await req.close().timeout(const Duration(seconds: 20));
if (res.statusCode != 200 && res.statusCode != 206) { if (res.statusCode != 200 && res.statusCode != 206) {
await res.drain<void>().catchError((_) {}); await res.drain<void>().catchError((_) {});
logUpstreamRefusal(res.statusCode, url, range, headers.keys);
throw UpstreamHttpException(res.statusCode, url); throw UpstreamHttpException(res.statusCode, url);
} }
return res; return res;
@@ -144,6 +156,7 @@ class HttpRelayBytes implements RelayBytes {
final total = res.headers.value(HttpHeaders.contentRangeHeader)?.split('/').last; final total = res.headers.value(HttpHeaders.contentRangeHeader)?.split('/').last;
final n = int.tryParse(total ?? '') ?? (res.statusCode == 200 ? res.contentLength : -1); final n = int.tryParse(total ?? '') ?? (res.statusCode == 200 ? res.contentLength : -1);
if (n <= 0) throw HttpException('upstream size unknown', uri: url); if (n <= 0) throw HttpException('upstream size unknown', uri: url);
_log.info('upstream ${logUrl(url)}: ${res.statusCode} $type, $n bytes (headers: ${headers.keys.join(', ')})');
return _size = n; return _size = n;
} }
@@ -354,8 +367,19 @@ class RelaySource {
/// Opens the media (a page is resolved) and starts serving; throws /// Opens the media (a page is resolved) and starts serving; throws
/// when it cannot be read. /// when it cannot be read.
Future<void> start() async { Future<void> start() async {
final tracks = await media.open(); final sw = Stopwatch()..start();
final infos = [for (final t in tracks) await relayTrackInfo(t)]; final List<RelayTrack> tracks;
final List<Map<String, Object?>> infos;
try {
tracks = await media.open();
// The tracks' first upstream reads at once: viewers wait on them.
infos = await Future.wait([for (final t in tracks) relayTrackInfo(t)]);
} catch (e) {
_log.warning('relay source ${logUrl(url)}: cannot open the media after ${sw.elapsedMilliseconds} ms: $e');
rethrow;
}
_log.info('relay source ${logUrl(url)}: media opened in ${sw.elapsedMilliseconds} ms '
'(${infos.map((i) => i['size'] != null ? '${i['kind'] == '' ? 'file' : i['kind']} ${i['size']} bytes' : '${i['kind']} ${(i['pieces'] as List?)?.length} pieces').join(' + ')})');
_tracks = tracks; _tracks = tracks;
unawaited(_loop({ unawaited(_loop({
't': 'info', 't': 'info',
@@ -365,6 +389,7 @@ class RelaySource {
Future<void> _loop(Map<String, Object?> info) async { Future<void> _loop(Map<String, Object?> info) async {
var attempt = 0; var attempt = 0;
final where = logUrl(url);
while (!_closed) { while (!_closed) {
try { try {
final ws = await WebSocket.connect(url.toString()).timeout(const Duration(seconds: 15)); final ws = await WebSocket.connect(url.toString()).timeout(const Duration(seconds: 15));
@@ -373,17 +398,21 @@ class RelaySource {
break; break;
} }
_ws = ws; _ws = ws;
_log.info('relay channel up: $where${attempt > 0 ? ' (after $attempt retries)' : ''}');
attempt = 0; attempt = 0;
ws.add(jsonEncode(info)); ws.add(jsonEncode(info));
await for (final m in ws) { await for (final m in ws) {
if (m is String) _onText(ws, m); if (m is String) _onText(ws, m);
} }
if (ws.closeCode == relayCloseNotSource) { if (ws.closeCode == relayCloseNotSource) {
_log.info('relay channel: not the source any more, stopping'); _log.info('relay channel: not the source any more, stopping ($where, sent $bytesSent bytes)');
_closed = true; _closed = true;
} else if (!_closed) {
_log.warning('relay channel down: $where, code ${ws.closeCode} ${ws.closeReason ?? ''}; reconnecting');
} }
} catch (e) { } catch (e) {
_log.fine('relay channel: $e'); final pass = attempt < 3 || attempt % 10 == 0 ? '' : null;
if (pass != null) _log.warning('relay channel: cannot reach $where (attempt ${attempt + 1}): $e');
} }
_ws = null; _ws = null;
_abandonAll(); _abandonAll();
@@ -477,7 +506,8 @@ class RelaySource {
if (!p.cancelled) ws.add(jsonEncode({'t': 'end', 'id': p.id})); if (!p.cancelled) ws.add(jsonEncode({'t': 'end', 'id': p.id}));
} catch (e) { } catch (e) {
if (!p.cancelled) { if (!p.cancelled) {
_log.warning('relay read ${p.id} failed: $e'); final pass = _readThrottle.allow('$e');
if (pass != null) _log.warning('${pass}relay read ${p.id} failed: $e');
try { try {
ws.add(jsonEncode({'t': 'fail', 'id': p.id, 'reason': '$e'})); ws.add(jsonEncode({'t': 'fail', 'id': p.id, 'reason': '$e'}));
} catch (_) {} } catch (_) {}
+2
View File
@@ -343,6 +343,7 @@ Future<UpstreamPieces> fetchHlsPlaylist(HttpClient client, Uri url, Map<String,
final res = await req.close().timeout(const Duration(seconds: 20)); final res = await req.close().timeout(const Duration(seconds: 20));
if (res.statusCode != 200) { if (res.statusCode != 200) {
await res.drain<void>().catchError((_) {}); await res.drain<void>().catchError((_) {});
logUpstreamRefusal(res.statusCode, url, 'the playlist', headers.keys);
throw UpstreamHttpException(res.statusCode, url); throw UpstreamHttpException(res.statusCode, url);
} }
final text = await res.transform(utf8.decoder).join().timeout(const Duration(seconds: 30)); final text = await res.transform(utf8.decoder).join().timeout(const Duration(seconds: 30));
@@ -361,6 +362,7 @@ Stream<List<int>> readPiece(HttpClient client, PieceSource s, Map<String, String
final res = await req.close().timeout(const Duration(seconds: 20)); final res = await req.close().timeout(const Duration(seconds: 20));
if (res.statusCode != 200 && res.statusCode != 206) { if (res.statusCode != 200 && res.statusCode != 206) {
await res.drain<void>().catchError((_) {}); await res.drain<void>().catchError((_) {});
logUpstreamRefusal(res.statusCode, s.url, 'a piece', headers.keys);
throw UpstreamHttpException(res.statusCode, s.url); throw UpstreamHttpException(res.statusCode, s.url);
} }
if (len == null) { if (len == null) {
+54 -3
View File
@@ -51,6 +51,11 @@ class YtDlpFormat {
this.protocol = 'https', this.protocol = 'https',
this.fragments}); this.fragments});
/// For the log: no URL query, header names only.
String describe() => '$id $protocol $ext${height == null ? '' : ' ${height}p'} @${url.host}'
' headers=[${headers.keys.join(',')}]${chunkSize == null ? '' : ' chunk=$chunkSize'}'
'${fragments == null ? '' : ' fragments=${fragments!.length}'}';
/// Relayed as pieces (see relay_pieces.dart), not as a byte stream. /// Relayed as pieces (see relay_pieces.dart), not as a byte stream.
bool get isPieces => _piecesProtocols.contains(protocol); bool get isPieces => _piecesProtocols.contains(protocol);
@@ -128,6 +133,17 @@ class YtDlpInfo {
this.formats = const {}, this.formats = const {},
}); });
/// One line for the log: extractor, duration, live, and each picked
/// track's format (id, protocol, container, height, CDN host, the names
/// of the headers it is fetched with).
String describe() {
final t = tracks.isEmpty
? 'nothing a relay can read'
: tracks.map((t) => '${t.kind.isEmpty ? 'file' : t.kind} ${t.format.describe()}').join(' + ');
return '$extractor${isLive ? ' live' : ''}, ${duration?.toStringAsFixed(0) ?? '?'} s, '
'${formats.length} relayable formats; tracks: $t';
}
/// Parses `yt-dlp -J` output; a playlist answer uses its first entry. /// Parses `yt-dlp -J` output; a playlist answer uses its first entry.
static YtDlpInfo parse(Map<String, Object?> j) { static YtDlpInfo parse(Map<String, Object?> j) {
final entries = j['entries']; final entries = j['entries'];
@@ -568,7 +584,10 @@ class YtDlp {
Future<YtDlpInfo> resolve(String url, {Duration timeout = const Duration(seconds: 25)}) async { Future<YtDlpInfo> resolve(String url, {Duration timeout = const Duration(seconds: 25)}) async {
final exe = await ensure(); final exe = await ensure();
if (exe == null) throw YtDlpException(lastError ?? 'yt-dlp is not available'); if (exe == null) throw YtDlpException(lastError ?? 'yt-dlp is not available');
final proc = await Process.start(exe, resolveArgs(url)); final args = resolveArgs(url);
final sw = Stopwatch()..start();
_log.info('resolve ${logUrl(url)} (yt-dlp $version, ${_argsForLog(args)})');
final proc = await Process.start(exe, args);
final out = proc.stdout.transform(utf8.decoder).join(); final out = proc.stdout.transform(utf8.decoder).join();
final err = proc.stderr.transform(utf8.decoder).join(); final err = proc.stderr.transform(utf8.decoder).join();
final int code; final int code;
@@ -576,18 +595,50 @@ class YtDlp {
code = await proc.exitCode.timeout(timeout); code = await proc.exitCode.timeout(timeout);
} on TimeoutException { } on TimeoutException {
proc.kill(); proc.kill();
_log.warning('resolve ${logUrl(url)}: no answer in ${timeout.inSeconds} s, killed');
throw YtDlpException('yt-dlp took longer than ${timeout.inSeconds} s'); throw YtDlpException('yt-dlp took longer than ${timeout.inSeconds} s');
} }
if (code != 0) throw YtDlpException(ytDlpErrorText(await err)); if (code != 0) {
final e = await err;
_log.warning('resolve ${logUrl(url)}: exit $code after ${sw.elapsedMilliseconds} ms; stderr tail:\n'
'${_tail(e)}');
throw YtDlpException(ytDlpErrorText(e));
}
try { try {
final j = jsonDecode(await out); final j = jsonDecode(await out);
if (j is! Map) throw const FormatException('not an object'); if (j is! Map) throw const FormatException('not an object');
return YtDlpInfo.parse(Map<String, Object?>.from(j)); final info = YtDlpInfo.parse(Map<String, Object?>.from(j));
_log.info('resolved ${logUrl(url)} in ${sw.elapsedMilliseconds} ms: ${info.describe()}');
return info;
} on FormatException catch (e) { } on FormatException catch (e) {
_log.warning('resolve ${logUrl(url)}: unreadable output (${e.message})');
throw YtDlpException('unreadable yt-dlp output: ${e.message}'); throw YtDlpException('unreadable yt-dlp output: ${e.message}');
} }
} }
/// [args] for the log: the cookie source named, not its path or spec.
static String _argsForLog(List<String> args) {
final out = <String>[];
for (var i = 0; i < args.length; i++) {
final a = args[i];
if ((a == '--cookies' || a == '--cookies-from-browser') && i + 1 < args.length) {
out.add(a == '--cookies' ? '--cookies <file>' : '--cookies-from-browser <browser>');
i++;
} else if (a == '--') {
break;
} else {
out.add(a);
}
}
return out.join(' ');
}
/// The last lines of yt-dlp's stderr, indented.
static String _tail(String stderr, {int lines = 8}) {
final all = const LineSplitter().convert(stderr).where((l) => l.trim().isNotEmpty).toList();
return all.skip(all.length > lines ? all.length - lines : 0).map((l) => ' $l').join('\n');
}
/// The command line [resolve] runs. /// The command line [resolve] runs.
List<String> resolveArgs(String url) => List<String> resolveArgs(String url) =>
['-J', '--no-playlist', '--no-warnings', '--ignore-config', ...cookies.args(), '--', url]; ['-J', '--no-playlist', '--no-warnings', '--ignore-config', ...cookies.args(), '--', url];
+89
View File
@@ -0,0 +1,89 @@
import 'dart:io';
import 'package:sync_relay_core/core.dart';
import 'package:test/test.dart';
void main() {
group('redactLog', () {
const tok = '50989dda73c45a4710751f8aff4a1982';
test('signed URLs keep scheme, host, path and parameter names only', () {
final out = redactLog('GET https://upos-hz-mirrorakam.akamaized.net/a/b-30080.m4s?deadline=1791826934&upsig=0f61b79b&hdnts=exp~acl -> 403');
expect(out, 'GET https://upos-hz-mirrorakam.akamaized.net/a/b-30080.m4s?deadline&upsig&hdnts=… -> 403');
expect(redactLog('https://jf.example/Videos/1/stream?static=true&api_key=abc123'),
'https://jf.example/Videos/1/stream?static&api_key=***');
expect(redactLog('http://user:pw@host/x#frag'), 'http://***@host/x');
});
test('member tokens in server paths', () {
expect(redactLog('open http://1.2.3.4:8080/r/default/$tok/2/0/name'), 'open http://1.2.3.4:8080/r/default/***/2/0/name');
expect(redactLog('ws://h:1/r/default/$tok/source?q=7'), 'ws://h:1/r/default/***/source?q=…');
expect(redactLog('/m/default/$tok/file.mkv range=bytes=0-'), '/m/default/***/file.mkv range=bytes=0-');
expect(redactLog('edl://!no_clip;%160%http://h/r/default/$tok/2/0/n;!new_stream;!no_clip;%160%http://h/r/default/$tok/2/1/n'),
isNot(contains(tok)));
});
test('secret values by name: pairs, JSON, quoted, headers, bearer', () {
expect(redactLog('api_key=abc123&x=1'), 'api_key=***&x=1');
expect(redactLog('{"token":"$tok","name":"bob","secret":"hunter2"}'), '{"token":"***","name":"bob","secret":"***"}');
expect(redactLog('MediaBrowser Client="WT", Device="pc", Token="deadbeef"'), contains('Token="***"'));
expect(redactLog('headers: {Cookie: SESSDATA=abc; bili_jct=x}'), 'headers: {Cookie: ***');
expect(redactLog('Authorization: Bearer abcdefgh12345678'), 'Authorization: ***');
expect(redactLog('sent Bearer abcdefgh12345678 to it'), 'sent Bearer *** to it');
expect(redactLog('password=pw1 hostKey=k1 X-Emby-Token=t1'), 'password=*** hostKey=*** X-Emby-Token=***');
});
test('leaves ordinary text alone', () {
const plain = 'yt-dlp cookies: 5 for youtube.com; relay pull 3 failed: upstream HTTP 403; tokens: 2 members';
expect(redactLog(plain), plain);
});
test('formatLogLine redacts the error and stack too', () {
final line = formatLogLine(DateTime(2026, 1, 2, 3, 4, 5, 6), 'WARNING', 'relay', 'read failed',
error: 'HttpException: upstream HTTP 403, uri = https://cdn/x.m4s?sig=abc');
expect(line, '2026-01-02 03:04:05.006 WARNING relay: read failed\n error: HttpException: upstream HTTP 403, uri = https://cdn/x.m4s?sig=…');
expect(logLineTime(line), DateTime(2026, 1, 2, 3, 4, 5, 6));
});
});
test('LogThrottle lets a burst through, then counts what it dropped', () {
var now = DateTime(2026);
final t = LogThrottle(burst: 2, window: const Duration(seconds: 10), now: () => now);
expect([t.allow('a'), t.allow('a'), t.allow('a'), t.allow('b')], ['', '', null, '']);
now = now.add(const Duration(seconds: 11));
expect(t.allow('a'), '(1 more like this dropped) ');
});
group('LogFiles', () {
late Directory dir;
setUp(() async => dir = await Directory.systemTemp.createTemp('logfiles'));
tearDown(() async => dir.delete(recursive: true));
test('rolls by size and day, prunes by age and total, reads recent lines', () {
var now = DateTime(2026, 10, 12, 10);
File('${dir.path}/app-2026-10-01.log').writeAsStringSync('old\n');
File('${dir.path}/server-2026-10-01.log').writeAsStringSync('not ours\n');
final f = LogFiles(dir.path, maxFileBytes: 100, maxTotalBytes: 1000, now: () => now)..open();
expect(File('${dir.path}/app-2026-10-01.log').existsSync(), isFalse);
expect(File('${dir.path}/server-2026-10-01.log').existsSync(), isTrue);
String line(String m) => formatLogLine(now, 'INFO', 't', m);
f.write(line('a' * 60));
f.write(line('b'));
expect(f.currentPath, endsWith('app-2026-10-12.log'));
f.write(line('b2'));
expect(f.currentPath, endsWith('app-2026-10-12.1.log'));
now = DateTime(2026, 10, 13, 9);
f.write(line('c'));
f.write(' continuation of c');
expect(f.currentPath, endsWith('app-2026-10-13.log'));
final recent = f.recent(const Duration(hours: 2));
expect(recent, '${line('c')}\n continuation of c');
expect(f.recent(const Duration(days: 2)).split('\n').first, contains('a' * 60));
f.close();
// Total cap: the oldest go first.
final small = LogFiles(dir.path, maxFileBytes: 100, maxTotalBytes: 150, now: () => now)..open();
expect(small.files().map((x) => x.uri.pathSegments.last), ['app-2026-10-12.1.log', 'app-2026-10-13.log']);
small.close();
});
});
}
+11 -1
View File
@@ -30,6 +30,9 @@ Future<void> main(List<String> args) async {
help: 'cookies.txt or browser-extension JSON export to log that yt-dlp in (sites that refuse guests). ' help: 'cookies.txt or browser-extension JSON export to log that yt-dlp in (sites that refuse guests). '
'An account login: keep it private. Copied into --yt-dlp-dir at start.') 'An account login: keep it private. Copied into --yt-dlp-dir at start.')
..addOption('config', help: 'YAML or JSON config file (keys as in AppConfig); overrides the other flags.') ..addOption('config', help: 'YAML or JSON config file (keys as in AppConfig); overrides the other flags.')
..addOption('log-dir',
help: 'Folder for the rotating log files (server-YYYY-MM-DD.log, kept 7 days). '
'Default: logs next to --config, else logs next to this binary (systemd: \$STATE_DIRECTORY/logs).')
..addFlag('upgrade', ..addFlag('upgrade',
negatable: false, negatable: false,
help: 'Update this install (the whole folder, app included) to the latest release, then exit. ' help: 'Update this install (the whole folder, app included) to the latest release, then exit. '
@@ -58,8 +61,14 @@ Future<void> main(List<String> args) async {
} }
initLogging(level: pkg_logging.Level.INFO); initLogging(level: pkg_logging.Level.INFO);
AppConfig config;
final configPath = parsed['config'] as String?; final configPath = parsed['config'] as String?;
final logDir = (parsed['log-dir'] as String?) ??
(configPath != null ? '${File(configPath).absolute.parent.path}${Platform.pathSeparator}logs' : serverDataDir('logs'));
final files = startLogFiles(logDir, prefix: 'server');
if (files != null) stdout.writeln('logs: ${files.dir}');
pkg_logging.Logger('server').info('watch-together-server $appVersion (proto ${RelayConstants.protoVersion}) '
'on ${Platform.operatingSystem} ${Platform.operatingSystemVersion}, pid $pid');
AppConfig config;
try { try {
if (configPath != null) { if (configPath != null) {
config = await AppConfig.loadFile(configPath); config = await AppConfig.loadFile(configPath);
@@ -114,6 +123,7 @@ Future<void> main(List<String> args) async {
if (msg is! Map) continue; if (msg is! Map) continue;
if (msg['t'] == 'error') { if (msg['t'] == 'error') {
stderr.writeln('server failed to start: ${msg['message']}'); stderr.writeln('server failed to start: ${msg['message']}');
pkg_logging.Logger('server').severe('failed to start: ${msg['message']}');
exitCode = 1; exitCode = 1;
break; break;
} }
+3 -2
View File
@@ -80,15 +80,16 @@ void mountJellyfinRelayRoute(
upRes = await upReq.close().timeout(timeout); upRes = await upReq.close().timeout(timeout);
} catch (e) { } catch (e) {
_relayLog.warning('upstream ${ref.itemId}: $e'); _relayLog.warning('upstream ${ref.itemId}: $e');
onLog?.call(token, MediaLog(method: req.method.name, path: '/j/$sid/$itemId', rangeHeader: range,
status: 502, bytesWritten: 0, durationMs: stopwatch.elapsedMilliseconds));
return _plain(502); return _plain(502);
} }
final shortToken = token.length > 4 ? '…${token.substring(token.length - 4)}' : token;
void log(int status, int bytes) => onLog?.call( void log(int status, int bytes) => onLog?.call(
token, token,
MediaLog( MediaLog(
method: req.method.name, method: req.method.name,
path: '/j/$sid/$shortToken/$itemId', path: '/j/$sid/$itemId',
rangeHeader: range, rangeHeader: range,
status: status, status: status,
bytesWritten: bytes, bytesWritten: bytes,
+26 -2
View File
@@ -27,6 +27,24 @@ class MediaLog {
'$method $path range=${rangeHeader ?? '-'} -> $status bytes=$bytesWritten ${durationMs}ms'; '$method $path range=${rangeHeader ?? '-'} -> $status bytes=$bytesWritten ${durationMs}ms';
} }
final _requestLog = createLogger('media');
/// Each member's media requests (`/m`, `/r`, `/j`) for the log: a few per
/// member, route and outcome every 30 s (mpv reads in many ranges), every
/// failure kind at least once per window.
class MediaRequestLog {
final _throttle = LogThrottle(burst: 4, window: const Duration(seconds: 30));
void call(String who, MediaLog log) {
final bad = log.status >= 400;
final route = log.path.split('/').where((s) => s.isNotEmpty).firstOrNull ?? '';
final pass = _throttle.allow('$who $route ${bad ? log.status : 'ok'}');
if (pass == null) return;
final line = '$pass$who: $log';
bad ? _requestLog.warning(line) : _requestLog.info(line);
}
}
class MediaResolution { class MediaResolution {
final int status; final int status;
final Map<String, String> headers; final Map<String, String> headers;
@@ -187,14 +205,20 @@ void mountSessionMediaRoute(
} }
final res = resolveMediaRequest(rangeHeader: rangeHeader, size: size, filename: media.name); final res = resolveMediaRequest(rangeHeader: rangeHeader, size: size, filename: media.name);
if (res.status != 200 && res.status != 206) { if (res.status != 200 && res.status != 206) {
onLog?.call(token, MediaLog(
method: req.method.name,
path: '/m/$sid/${media.name}',
rangeHeader: rangeHeader,
status: res.status,
bytesWritten: 0,
durationMs: stopwatch.elapsedMilliseconds));
return _gatedResponse(res); return _gatedResponse(res);
} }
final end = res.end ?? size - 1; final end = res.end ?? size - 1;
final byteCount = end - res.start + 1; final byteCount = end - res.start + 1;
final shortToken = token.length > 4 ? '…${token.substring(token.length - 4)}' : token;
onLog?.call(token, MediaLog( onLog?.call(token, MediaLog(
method: req.method.name, method: req.method.name,
path: '/m/$sid/$shortToken/${media.name}', path: '/m/$sid/${media.name}',
rangeHeader: rangeHeader, rangeHeader: rangeHeader,
status: res.status, status: res.status,
bytesWritten: byteCount, bytesWritten: byteCount,
+8
View File
@@ -6,6 +6,8 @@ import 'server_hub.dart';
// `queue` messages (Up next, see PROTOCOL.md "Queue"): what an added item // `queue` messages (Up next, see PROTOCOL.md "Queue"): what an added item
// may carry, and the ops applied to the engine's queue. // may carry, and the ops applied to the engine's queue.
final _log = createLogger('queue');
final _key = RegExp(r'^[A-Za-z0-9_-]{8,64}$'); final _key = RegExp(r'^[A-Za-z0-9_-]{8,64}$');
bool _httpUrl(String? url) { bool _httpUrl(String? url) {
@@ -76,9 +78,11 @@ void handleQueueIntent(ServerHub hub, String token, RelicWebSocket ws, Map<Strin
final error = engine.applyQueueAdd(token, cleaned.item, final error = engine.applyQueueAdd(token, cleaned.item,
next: intent.next, privateUrl: cleaned.privateUrl, thumb: thumb); next: intent.next, privateUrl: cleaned.privateUrl, thumb: thumb);
if (error != null) { if (error != null) {
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}) refused: $error');
wsErr(ws, error); wsErr(ws, error);
return; return;
} }
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}${intent.next ? ', next' : ''})');
changed = true; changed = true;
case 'move': case 'move':
changed = engine.applyQueueMove(token, intent.id, intent.to); changed = engine.applyQueueMove(token, intent.id, intent.to);
@@ -89,5 +93,9 @@ void handleQueueIntent(ServerHub hub, String token, RelicWebSocket ws, Map<Strin
case 'skip': case 'skip':
changed = engine.applyQueuePlay(token, ''); changed = engine.applyQueuePlay(token, '');
} }
if (intent.op != 'add') {
_log.info('${hub.who(token)}: ${intent.op}${intent.id.isEmpty ? '' : ' ${intent.id}'}'
'${intent.op == 'move' ? ' to ${intent.to}' : ''}${changed ? '' : ' (no change)'}');
}
if (changed) hub.changed(); if (changed) hub.changed();
} }
+80 -47
View File
@@ -203,7 +203,9 @@ class ActiveRelay {
final r = ActiveRelay._(mediaId, name, null, onLiveChanged, fetch: true, queueId: queueId, warm: queueId != null) final r = ActiveRelay._(mediaId, name, null, onLiveChanged, fetch: true, queueId: queueId, warm: queueId != null)
..onFail = onFail; ..onFail = onFail;
final media = r._media = LinkRelayMedia(url, ytDlp: ytDlp); final media = r._media = LinkRelayMedia(url, ytDlp: ytDlp);
_log.info('relay fetch of "$name"${queueId == null ? '' : ' (warm, Up next $queueId)'}: ${logUrl(url)}');
unawaited(() async { unawaited(() async {
final sw = Stopwatch()..start();
try { try {
final tracks = await media.open(); final tracks = await media.open();
final infos = <_TrackInfo>[ final infos = <_TrackInfo>[
@@ -213,12 +215,14 @@ class ActiveRelay {
: (kind: t.kind, size: await t.bytes!.size(), pieces: null), : (kind: t.kind, size: await t.bytes!.size(), pieces: null),
]; ];
if (r._disposed) return; if (r._disposed) return;
_log.info('relay fetch of "$name" ready in ${sw.elapsedMilliseconds} ms: '
'${infos.map((t) => t.size != null ? '${t.kind.isEmpty ? 'file' : t.kind} ${t.size} bytes' : '${t.kind} ${t.pieces!.length} pieces').join(' + ')}');
r._setTracks(relayTrackSpecs(tracks)); r._setTracks(relayTrackSpecs(tracks));
r._local = [for (var i = 0; i < tracks.length; i++) LocalRelayFeed(tracks[i], r.buffers[i])]; r._local = [for (var i = 0; i < tracks.length; i++) LocalRelayFeed(tracks[i], r.buffers[i])];
r._attachAll(r._local, infos); r._attachAll(r._local, infos);
onLiveChanged(); onLiveChanged();
} catch (e) { } catch (e) {
_log.warning('relay fetch: cannot read the link: $e'); _log.warning('relay fetch of "$name": cannot read the link after ${sw.elapsedMilliseconds} ms: $e');
if (!r._disposed) { if (!r._disposed) {
r.failed = true; r.failed = true;
r.onFail?.call('$e'); r.onFail?.call('$e');
@@ -370,9 +374,12 @@ void mountRelayRoutes(
return WebSocketUpgrade((final ws) async { return WebSocketUpgrade((final ws) async {
final r = found; final r = found;
if (r == null) { if (r == null) {
_log.warning('relay channel refused: not the source of ${queueId == null ? 'the current media' : 'Up next item $queueId'}');
await _closeNotSource(ws); await _closeNotSource(ws);
return; return;
} }
final opened = DateTime.now();
String? closed;
ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs); ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs);
var attached = false; var attached = false;
try { try {
@@ -397,7 +404,7 @@ void mountRelayRoutes(
return; return;
} }
attached = true; attached = true;
_log.info('relay channel up (${r.name}, ' _log.info('relay channel up (${r.name}${r.warm ? ', warm' : ''}, from ${r.sourceId}, '
'${infos.map((t) => t.size != null ? '${t.size} bytes' : '${t.pieces!.length} pieces').join(' + ')})'); '${infos.map((t) => t.size != null ? '${t.size} bytes' : '${t.pieces!.length} pieces').join(' + ')})');
r._attachChannel(ws, infos); r._attachChannel(ws, infos);
case 'end': case 'end':
@@ -413,6 +420,8 @@ void mountRelayRoutes(
} }
} }
} }
} else if (e is CloseReceived) {
closed = '${e.code ?? '-'} ${e.reason}'.trim();
} else if (e is BinaryDataReceived && attached) { } else if (e is BinaryDataReceived && attached) {
try { try {
// Pull ids are unique across tracks: only the owner takes it. // Pull ids are unique across tracks: only the owner takes it.
@@ -423,11 +432,13 @@ void mountRelayRoutes(
} catch (_) {} } catch (_) {}
} }
} }
} catch (_) { } catch (e) {
// channel broke _log.warning('relay channel broke (${r.name}): $e');
} finally { } finally {
r._detachChannel(ws); r._detachChannel(ws);
_log.info('relay channel down (${r.name})'); _log.info('relay channel down (${r.name}, from ${r.sourceId}) after '
'${DateTime.now().difference(opened).inSeconds} s, ${closed == null ? 'no close frame' : 'closed $closed'}'
'${r._disposed ? ', relay dropped' : ''}');
} }
}); });
}); });
@@ -438,83 +449,97 @@ void mountRelayRoutes(
final token = req.pathParameters.raw[#token] ?? ''; final token = req.pathParameters.raw[#token] ?? '';
final key = req.pathParameters.raw[#mediaId] ?? ''; final key = req.pathParameters.raw[#mediaId] ?? '';
final track = tracked ? int.tryParse(req.pathParameters.raw[#seg] ?? '') ?? -1 : null; final track = tracked ? int.tryParse(req.pathParameters.raw[#seg] ?? '') ?? -1 : null;
final session = sessions.get(sid); final pieceIndex = piece ? req.pathParameters.raw[#piece] ?? '' : null;
if (session == null) return _plain(410);
switch (session.gate(token)) {
case MediaGate.ok:
break;
default:
return _plain(401);
}
// A player still on an older media must not get the new one's bytes.
final r = relayAt(key);
if (r == null) return _plain(404);
final notConnected = Response(503,
body: Body.fromString('relay source not connected'),
headers: Headers.build((h) => h['retry-after'] = ['2']));
if (!await r.whenKnown()) return notConnected;
final buffer = r.bufferAt(track);
if (buffer == null || (piece && buffer is! PieceBuffer)) return _plain(404);
if (!await buffer.ready()) return notConnected;
final typed = req.headers.range; final typed = req.headers.range;
String? rangeHeader; String? rangeHeader;
if (typed != null && typed.ranges.isNotEmpty) { if (typed != null && typed.ranges.isNotEmpty) {
final x = typed.ranges.first; final x = typed.ranges.first;
rangeHeader = '${typed.unit}=${x.start?.toString() ?? ''}-${x.end?.toString() ?? ''}'; rangeHeader = '${typed.unit}=${x.start?.toString() ?? ''}-${x.end?.toString() ?? ''}';
} }
// Every outcome goes to the log (throttled per member there); a 503
// after a long wait is a source that never came up.
Response logged(Response res, {int bytes = 0}) {
onLog?.call(
token,
MediaLog(
method: req.method.name,
path: '/r/$sid/$key/${track == null ? '' : '$track/'}${pieceIndex == null ? '' : 'p/$pieceIndex'}',
rangeHeader: rangeHeader,
status: res.statusCode,
bytesWritten: bytes,
durationMs: stopwatch.elapsedMilliseconds,
));
return res;
}
final session = sessions.get(sid);
if (session == null) return logged(_plain(410));
switch (session.gate(token)) {
case MediaGate.ok:
break;
default:
return logged(_plain(401));
}
// A player still on an older media must not get the new one's bytes.
final r = relayAt(key);
if (r == null) return logged(_plain(404));
Response notConnected() => logged(Response(503,
body: Body.fromString('relay source not connected'),
headers: Headers.build((h) => h['retry-after'] = ['2'])));
if (!await r.whenKnown()) return notConnected();
final buffer = r.bufferAt(track);
if (buffer == null || (piece && buffer is! PieceBuffer)) return logged(_plain(404));
if (!await buffer.ready()) return notConnected();
if (buffer is PieceBuffer) { if (buffer is PieceBuffer) {
if (!piece) { if (!piece) {
buffer.opened(token); buffer.opened(token);
return Response(200, return logged(Response(200,
body: Body.fromString(buffer.playlist ?? '', mimeType: MimeType('application', 'vnd.apple.mpegurl')), body: Body.fromString(buffer.playlist ?? '', mimeType: MimeType('application', 'vnd.apple.mpegurl')),
headers: Headers.build((h) => h['cache-control'] = ['no-store'])); headers: Headers.build((h) => h['cache-control'] = ['no-store'])));
} }
final index = int.tryParse(req.pathParameters.raw[#piece] ?? ''); final index = int.tryParse(pieceIndex ?? '');
if (index == null || index < 0 || index >= buffer.list!.length) return _plain(404); if (index == null || index < 0 || index >= buffer.list!.length) return logged(_plain(404));
final data = await buffer.piece(token, index); final data = await buffer.piece(token, index);
if (data == null) return notConnected; if (data == null) return notConnected();
final res = resolveMediaRequest(rangeHeader: rangeHeader, size: data.length, filename: 'piece'); final res = resolveMediaRequest(rangeHeader: rangeHeader, size: data.length, filename: 'piece');
final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v])); final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v]));
if (res.status != 200 && res.status != 206) { if (res.status != 200 && res.status != 206) {
return Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers); return logged(Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers));
} }
final end = res.end ?? data.length - 1; final end = res.end ?? data.length - 1;
final n = end - res.start + 1; final n = end - res.start + 1;
buffer.bytesToViewers += n; buffer.bytesToViewers += n;
session.addBytes(token, n); session.addBytes(token, n);
return Response(res.status, return logged(
Response(res.status,
body: Body.fromData(Uint8List.sublistView(data, res.start, end + 1), mimeType: MimeType.octetStream), body: Body.fromData(Uint8List.sublistView(data, res.start, end + 1), mimeType: MimeType.octetStream),
headers: headers); headers: headers),
bytes: n);
} }
final size = (buffer as RelayBuffer).size; final size = (buffer as RelayBuffer).size;
final res = resolveMediaRequest(rangeHeader: rangeHeader, size: size, filename: r.name); final res = resolveMediaRequest(rangeHeader: rangeHeader, size: size, filename: r.name);
final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v])); final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v]));
if (res.status != 200 && res.status != 206) { if (res.status != 200 && res.status != 206) {
return Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers); return logged(Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers));
} }
final end = res.end ?? size - 1; final end = res.end ?? size - 1;
final shortToken = token.length > 4 ? '…${token.substring(token.length - 4)}' : token; final Stream<Uint8List> body = buffer
onLog?.call( .read(token, res.start, end, onSent: (n) => session.addBytes(token, n))
token, .transform(StreamTransformer.fromHandlers(handleError: (e, st, sink) {
MediaLog( _readFails(token, '${r.name} track ${track ?? 0} from ${res.start}: $e');
method: req.method.name, sink.addError(e, st);
path: '/r/$sid/$shortToken/$key/${track == null ? '' : '$track/'}${r.name}', }));
rangeHeader: rangeHeader,
status: res.status,
bytesWritten: end - res.start + 1,
durationMs: stopwatch.elapsedMilliseconds,
));
final Stream<Uint8List> body =
buffer.read(token, res.start, end, onSent: (n) => session.addBytes(token, n));
MimeType mime; MimeType mime;
try { try {
mime = MimeType.parse(res.headers['content-type'] ?? ''); mime = MimeType.parse(res.headers['content-type'] ?? '');
} catch (_) { } catch (_) {
mime = MimeType.octetStream; mime = MimeType.octetStream;
} }
return Response(res.status, return logged(
Response(res.status,
body: Body.fromDataStream(body, mimeType: mime, contentLength: end - res.start + 1), body: Body.fromDataStream(body, mimeType: mime, contentLength: end - res.start + 1),
headers: headers); headers: headers),
bytes: end - res.start + 1);
} }
// `:seg` is the name of a single file, the track of several (the router // `:seg` is the name of a single file, the track of several (the router
@@ -525,3 +550,11 @@ void mountRelayRoutes(
} }
Response _plain(int status) => Response(status, body: Body.fromString('status $status')); Response _plain(int status) => Response(status, body: Body.fromString('status $status'));
final _readFailThrottle = LogThrottle(burst: 3);
/// A viewer's read of a relay broke (the source never sent the bytes).
void _readFails(String token, String what) {
final pass = _readFailThrottle.allow(token);
if (pass != null) _log.warning('${pass}relay read for a viewer broke: $what');
}
+2 -1
View File
@@ -12,7 +12,7 @@ import 'sync_server.dart';
/// ///
/// On [events] it reports `{'t':'ready','port':p,'stop':SendPort}` or /// On [events] it reports `{'t':'ready','port':p,'stop':SendPort}` or
/// `{'t':'error','message':m}`, and forwards log records as /// `{'t':'error','message':m}`, and forwards log records as
/// `{'t':'log','level','logger','message'[,'error']}`. Sending /// `{'t':'log','level','logger','message'[,'error'][,'stack']}`. Sending
/// `{'t':'stop'}` to the `stop` port stops the server; the returned future /// `{'t':'stop'}` to the `stop` port stops the server; the returned future
/// completes once it is down. `{'t':'file','path':p,'reply':SendPort}` makes /// completes once it is down. `{'t':'file','path':p,'reply':SendPort}` makes
/// a file on this machine the hosted file (the hosting app's in-session file /// a file on this machine the hosted file (the hosting app's in-session file
@@ -29,6 +29,7 @@ Future<void> serve(Map<String, Object?> config, {SendPort? events}) async {
'logger': r.loggerName, 'logger': r.loggerName,
'message': r.message, 'message': r.message,
if (r.error != null) 'error': '${r.error}', if (r.error != null) 'error': '${r.error}',
if (r.stackTrace != null) 'stack': '${r.stackTrace}',
})); }));
} }
final SyncServer server; final SyncServer server;
+60 -3
View File
@@ -70,17 +70,32 @@ class ServerHub {
/// Replaces the relay (null: the new media is not relayed). The old /// Replaces the relay (null: the new media is not relayed). The old
/// buffer is dropped with it. /// buffer is dropped with it.
void setRelay(ActiveRelay? next) { void setRelay(ActiveRelay? next) {
if (!identical(relay, next)) relay?.dispose(); if (!identical(relay, next)) {
final old = relay;
if (old != null) _hubLog.info('relay of "${old.name}" dropped (sent ${_sent(old)})');
old?.dispose();
}
relay = next; relay = next;
} }
static String _sent(ActiveRelay r) =>
'${(r.buffers.fold<int>(0, (a, b) => a + b.bytesFromFeed) / 1048576).toStringAsFixed(1)} MiB in';
/// The warm relay: Up next's first item relayed before it goes up, so /// The warm relay: Up next's first item relayed before it goes up, so
/// every viewer preloads it (see ARCHITECTURE "Getting ready"); null /// every viewer preloads it (see ARCHITECTURE "Getting ready"); null
/// when that item is not relayed (or nobody would read it). /// when that item is not relayed (or nobody would read it).
ActiveRelay? warm; ActiveRelay? warm;
void _setWarm(ActiveRelay? next) { void _setWarm(ActiveRelay? next) {
if (!identical(warm, next)) warm?.dispose(); if (!identical(warm, next)) {
final old = warm;
if (old != null) _hubLog.info('warm relay of Up next ${old.queueId} ("${old.name}") dropped (${_sent(old)})');
old?.dispose();
if (next != null) {
_hubLog.info('warm relay of Up next ${next.queueId} ("${next.name}"): '
'${next.fetch ? 'the server fetches it' : next.sourceId.isEmpty ? 'from the first member with an account' : 'from ${next.sourceId}'}');
}
}
warm = next; warm = next;
} }
@@ -214,9 +229,11 @@ class ServerHub {
? w.fetch && !w.failed ? w.fetch && !w.failed
: !w.fetch && w.sourceToken == token && (!w.known || _sameTracks(w.tracks, tracks)); : !w.fetch && w.sourceToken == token && (!w.known || _sameTracks(w.tracks, tracks));
if (ok) { if (ok) {
_hubLog.info('warm relay of Up next $queueId promoted (${_sent(w)}, live ${w.live})');
w.ensureTracks(tracks); w.ensureTracks(tracks);
return w; return w;
} }
_hubLog.info('warm relay of Up next $queueId not promoted (${w.failed ? 'its fetch failed' : 'another kind or source'})');
w.dispose(); w.dispose();
return null; return null;
} }
@@ -246,11 +263,15 @@ class ServerHub {
if (!identical(relay, r) || m?.relay == null) return; if (!identical(relay, r) || m?.relay == null) return;
final was = m!.relay!; final was = m!.relay!;
if (was.live == r.live && was.tracks.join(',') == r.tracks.join(',')) return; if (was.live == r.live && was.tracks.join(',') == r.tracks.join(',')) return;
if (was.live != r.live) _hubLog.info('relay of "${r.name}" ${r.live ? 'live' : 'not live (source away)'}');
engine.media = m.copyWith(relay: was.copyWith(live: r.live, tracks: r.tracks)); engine.media = m.copyWith(relay: was.copyWith(live: r.live, tracks: r.tracks));
changed(); changed();
} }
final _readRates = ReadRates(); final _readRates = ReadRates();
var _statTicks = 0;
static String _kbs(int bytesPerSec) => '${(bytesPerSec / 1024).round()} KiB/s';
/// Once a second: each member's read rate (`members[].rate`) and the /// Once a second: each member's read rate (`members[].rate`) and the
/// relay's source rate and bitrate (`media.relay.rate` / `need`). Goes /// relay's source rate and bitrate (`media.relay.rate` / `need`). Goes
@@ -268,6 +289,12 @@ class ServerHub {
if (r == null || m == null || was == null || r.mediaId != engine.mediaId) return; if (r == null || m == null || was == null || r.mediaId != engine.mediaId) return;
r.sampleRate(); r.sampleRate();
final rate = r.rate, need = r.need(m.duration); final rate = r.rate, need = r.need(m.duration);
if (++_statTicks % 30 == 0 && r.live) {
final readers = engine.members.values.where((x) => x.rate > 0).map((x) => '${x.name} ${_kbs(x.rate)}');
_hubLog.info('relay stats "${r.name}": source ${_kbs(rate)}, needs ${_kbs(need)}, '
'buffered ${(r.cachedBytes / 1048576).toStringAsFixed(1)} MiB, ${_sent(r)}'
'${readers.isEmpty ? '' : '; reading: ${readers.join(', ')}'}');
}
if (rate != was.rate || need != was.need) { if (rate != was.rate || need != was.need) {
engine.media = m.copyWith(relay: was.copyWith(rate: rate, need: need)); engine.media = m.copyWith(relay: was.copyWith(rate: rate, need: need));
} }
@@ -275,6 +302,14 @@ class ServerHub {
bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey; bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey;
/// A member for the log: `name (id)`, never its token.
String who(String token) {
final m = engine.members[token];
if (m != null) return '${m.name} (${m.id})';
final v = session.tokens[token];
return v != null ? '${v.name} (no member)' : 'unknown member';
}
/// The member a reconnecting client may take over with its old token: /// The member a reconnecting client may take over with its old token:
/// still known. /// still known.
EngineMember? resumable(String token) { EngineMember? resumable(String token) {
@@ -365,6 +400,7 @@ class ServerHub {
} }
void forgetSocket(String token) { void forgetSocket(String token) {
if (engine.members.containsKey(token)) _hubLog.info('leave ${who(token)}');
engine.applyLeave(token); engine.applyLeave(token);
session.tokens.remove(token); session.tokens.remove(token);
sockets.remove(token); sockets.remove(token);
@@ -383,6 +419,17 @@ class ServerHub {
} }
} }
/// How a swap's media plays, for the log (a link's query cut to names).
String _swapKind(HelloMedia m) {
final relay = m.relay;
final url = (m.url ?? '').isEmpty ? '' : ' ${logUrl(m.url)}';
if (m.jellyfin != null) return 'Jellyfin item ${m.jellyfin!.itemId}${relay == null ? '' : ', relayed from the swapper'}';
if (relay == null) return url.isEmpty ? 'hosted file' : 'link, everyone fetches it:$url';
if (relay.fetch) return 'link, the server fetches it:$url';
final tracks = relay.tracks.isEmpty ? 'one file' : relay.tracks.map((t) => t.toJson().values.join('/')).join(' + ');
return 'through the swapper ($tracks, ${m.size} bytes)';
}
/// `GET /a/<sid>/<token>/<hash>`: a current member's picture by its /// `GET /a/<sid>/<token>/<hash>`: a current member's picture by its
/// content hash, for members (valid token) only. The bytes for a /// content hash, for members (valid token) only. The bytes for a
/// hash never change, so clients may cache them for good. /// hash never change, so clients may cache them for good.
@@ -519,6 +566,8 @@ void mountSyncRoute(
final swap = SwapIntent.fromJson(decoded); final swap = SwapIntent.fromJson(decoded);
var parsed = swap.media; var parsed = swap.media;
final queueId = swap.queueId; final queueId = swap.queueId;
_hubLog.info('swap by ${hub.who(token)}: "${parsed.name}" ${_swapKind(parsed)}'
'${queueId == null ? '' : ' (Up next $queueId)'}');
if (queueId != null) { if (queueId != null) {
// Already put up by someone else: nothing to do. // Already put up by someone else: nothing to do.
if (!hub.engine.queue.has(queueId)) return; if (!hub.engine.queue.has(queueId)) return;
@@ -682,6 +731,12 @@ void mountSyncRoute(
case 'report': case 'report':
if (token == null) return; if (token == null) return;
final report = ReportMessage.fromJson(decoded); final report = ReportMessage.fromJson(decoded);
final before = hub.engine.members[token]?.error ?? '';
if (report.error != before && report.mediaId == hub.engine.mediaId) {
_hubLog.warning(report.error.isEmpty
? '${hub.who(token)}: player error cleared'
: '${hub.who(token)}: cannot play media ${report.mediaId}: ${report.error}');
}
hub.engine.applyReport( hub.engine.applyReport(
token, report.position, report.buffered, report.eventId, token, report.position, report.buffered, report.eventId,
duration: report.duration, mediaId: report.mediaId, jf: report.jf, error: report.error); duration: report.duration, mediaId: report.mediaId, jf: report.jf, error: report.error);
@@ -708,8 +763,10 @@ void mountSyncRoute(
handleQueueIntent(hub, token, ws, decoded); handleQueueIntent(hub, token, ws, decoded);
} }
} on ProtocolException catch (e) { } on ProtocolException catch (e) {
_hubLog.info('${token == null ? 'a socket' : hub.who(token)}: bad $type message: ${e.message}');
wsErr(ws, 'bad message: ${e.message}'); wsErr(ws, 'bad message: ${e.message}');
} catch (_) { } catch (e, st) {
_hubLog.warning('${token == null ? 'a socket' : hub.who(token)}: $type message failed: $e', e, st);
return; return;
} }
} }
+17 -10
View File
@@ -15,15 +15,19 @@ import 'session.dart';
/// [configured], else systemd's `$STATE_DIRECTORY/yt-dlp`, else `yt-dlp` /// [configured], else systemd's `$STATE_DIRECTORY/yt-dlp`, else `yt-dlp`
/// next to the server binary when that is writable (not under `dart run`), /// next to the server binary when that is writable (not under `dart run`),
/// else `~/.cache/watch-together/yt-dlp`. /// else `~/.cache/watch-together/yt-dlp`.
String serverYtDlpDir(String configured) { String serverYtDlpDir(String configured) => configured.isNotEmpty ? configured : serverDataDir('yt-dlp');
if (configured.isNotEmpty) return configured;
/// A writable folder [name] for the dedicated server's own files:
/// `$STATE_DIRECTORY/<name>` (systemd), else next to the binary if
/// writable, else `~/.cache/watch-together/<name>`.
String serverDataDir(String name) {
final sep = Platform.pathSeparator; final sep = Platform.pathSeparator;
final state = (Platform.environment['STATE_DIRECTORY'] ?? '').split(':').first; final state = (Platform.environment['STATE_DIRECTORY'] ?? '').split(':').first;
if (state.isNotEmpty) return '$state${sep}yt-dlp'; if (state.isNotEmpty) return '$state$sep$name';
final exe = File(Platform.resolvedExecutable); final exe = File(Platform.resolvedExecutable);
final name = exe.uri.pathSegments.last.toLowerCase(); final exeName = exe.uri.pathSegments.last.toLowerCase();
if (name != 'dart' && name != 'dart.exe' && name != 'dartaotruntime') { if (exeName != 'dart' && exeName != 'dart.exe' && exeName != 'dartaotruntime') {
final dir = Directory('${exe.parent.path}${sep}yt-dlp'); final dir = Directory('${exe.parent.path}$sep$name');
try { try {
dir.createSync(recursive: true); dir.createSync(recursive: true);
final probe = File('${dir.path}$sep.writable')..writeAsStringSync(''); final probe = File('${dir.path}$sep.writable')..writeAsStringSync('');
@@ -32,7 +36,7 @@ String serverYtDlpDir(String configured) {
} catch (_) {} } catch (_) {}
} }
final home = Platform.environment['HOME'] ?? Platform.environment['USERPROFILE'] ?? '.'; final home = Platform.environment['HOME'] ?? Platform.environment['USERPROFILE'] ?? '.';
return '$home$sep.cache${sep}watch-together${sep}yt-dlp'; return '$home$sep.cache${sep}watch-together$sep$name';
} }
class SyncServer { class SyncServer {
@@ -93,12 +97,15 @@ class SyncServer {
})), })),
); );
}); });
mountSessionMediaRoute(_app, sessions); final requests = MediaRequestLog();
mountRelayRoutes(_app, sessions, relayAt: hub.relayAt, sourceOf: hub.sourceOf); void logRequest(String token, MediaLog log) => requests(hub.who(token), log);
mountSessionMediaRoute(_app, sessions, onLog: logRequest);
mountRelayRoutes(_app, sessions, relayAt: hub.relayAt, sourceOf: hub.sourceOf, onLog: logRequest);
mountAvatarRoute(_app, hub); mountAvatarRoute(_app, hub);
mountArtRoute(_app, hub); mountArtRoute(_app, hub);
mountJellyfinRelayRoute( mountJellyfinRelayRoute(
_app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient(), next: () => hub.nextJellyfinRef); _app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient(),
next: () => hub.nextJellyfinRef, onLog: logRequest);
// advertiseHost is only an explicit override; otherwise echo what the client dialled. // advertiseHost is only an explicit override; otherwise echo what the client dialled.
String authority(String dialledHost) => config.advertiseHost.isNotEmpty String authority(String dialledHost) => config.advertiseHost.isNotEmpty
? '${config.advertiseHost}:$boundPort' ? '${config.advertiseHost}:$boundPort'