Stillkill/sync_replay_project
1
0 Fork 0
forked from Foxyz/sync_replay_project
Code Pull Requests Activity

Restructure step 1: lasting identity, presence and protocol 16

Browse Source 
Server and core: version ranges on both sides with a coded refusal; a
client id maps to the same member and token for the server's life; a
drop is `gone` for a 10-minute grace keeping ready, presence, queue
items, relay and avatar; `bye` leaves at once; ordered takeover (seq,
replaced/stale); presence joining/in/away/gone with one PresencePolicy;
joiners never pause playback and hold a held gate at most ~15 s;
state.hold; feed ids, seq and replay; the relay lives through its
source's grace and continues on re-dial; queue `by` lasting, auto
items and cancel; MediaRef carried; client ids never logged.

App: persisted client id and seq; Connection with unstable/lost/
replaced states, backoff with jitter, no lost/back pair per flap; a drop
keeps the session (state, player, media, relay source); presence in the
strip and members card; sit out = away intent; feed matched by id;
auto sit-out (idle when not playing, minimised for a while; never when
alone) replaces auto-unready; bye on close and Leave, not on update
restart; relay source re-dials through its grace.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
foxyzandClaude Opus 5.5 committed 2026-10-15 21:39:47 +03:00
1 parent 51c1c5ab78
commit 9f2586b7ff
85 files changed
+5198 -1902
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/server/lib/src/queue_ops.dart
+20 -10
View File
@@ -30,25 +30,28 @@ bool _httpUrl(String? url) {
if (clean == null) throw const ProtocolException('bad queue item');
if (name.isEmpty) name = clean.title;
item = QueueItem(
media: HelloMedia(name: name, size: 0, duration: 0, jellyfin: clean), how: QueueHow.jellyfin, title: '');
media: HelloMedia(name: name, size: 0, duration: 0, jellyfin: clean, ref: media.ref),
how: QueueHow.jellyfin,
title: '');
} else {
switch (raw.how) {
case QueueHow.direct:
if (!_httpUrl(media.url)) throw const ProtocolException('a queued link needs an http(s) url');
item = QueueItem(media: HelloMedia(name: name, size: 0, duration: 0, url: media.url), how: raw.how, title: '');
item = QueueItem(
media: HelloMedia(name: name, size: 0, duration: 0, url: media.url, ref: media.ref), how: raw.how, title: '');
case QueueHow.server:
if (!_httpUrl(media.url)) throw const ProtocolException('relay fetch needs an http(s) url');
if (!serverHasPassword) throw const ProtocolException('server fetch needs a server password');
privateUrl = media.url;
item = QueueItem(
media: HelloMedia(name: name, size: 0, duration: 0, relay: const MediaRelay(fetch: true)),
media: HelloMedia(name: name, size: 0, duration: 0, relay: const MediaRelay(fetch: true), ref: media.ref),
how: raw.how,
title: '');
case QueueHow.through:
case QueueHow.file:
if (!_key.hasMatch(raw.key)) throw const ProtocolException('a queued item of yours needs a key');
item = QueueItem(
media: HelloMedia(name: name, size: media.size < 0 ? 0 : media.size, duration: 0),
media: HelloMedia(name: name, size: media.size < 0 ? 0 : media.size, duration: 0, ref: media.ref),
how: raw.how,
title: '',
key: raw.key);
@@ -60,7 +63,11 @@ bool _httpUrl(String? url) {
if (title.isEmpty) title = cleanQueueText(item.media.name.isEmpty ? 'media' : item.media.name);
return (
item: item.copyWith(
title: title, sub: cleanQueueText(sanitizeMediaName(raw.sub)), duration: raw.duration),
title: title,
sub: cleanQueueText(sanitizeMediaName(raw.sub)),
duration: raw.duration,
auto: raw.auto,
forMedia: raw.auto ? raw.forMedia : 0),
privateUrl: privateUrl,
);
}
@@ -75,14 +82,15 @@ void handleQueueIntent(ServerHub hub, String token, RelicWebSocket ws, Map<Strin
case 'add':
final cleaned = cleanQueueAdd(intent.item!, serverHasPassword: hub.session.secret.isNotEmpty);
final thumb = intent.thumb.isEmpty ? null : decodePicture(intent.thumb, RelayConstants.maxPosterBytes);
final error = engine.applyQueueAdd(token, cleaned.item,
final refused = engine.applyQueueAdd(token, cleaned.item,
next: intent.next, privateUrl: cleaned.privateUrl, thumb: thumb);
if (error != null) {
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}) refused: $error');
wsErr(ws, error);
if (refused != null) {
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}) refused: ${refused.message}');
wsErr(ws, refused.code, refused.message, forType: 'queue');
return;
}
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}${intent.next ? ', next' : ''})');
_log.info('${hub.who(token)}: add "${cleaned.item.title}" (${cleaned.item.how.name}'
'${cleaned.item.auto ? ', continues media ${cleaned.item.forMedia}' : intent.next ? ', next' : ''})');
changed = true;
case 'move':
changed = engine.applyQueueMove(token, intent.id, intent.to);
@@ -92,6 +100,8 @@ void handleQueueIntent(ServerHub hub, String token, RelicWebSocket ws, Map<Strin
changed = engine.applyQueuePlay(token, intent.id);
case 'skip':
changed = engine.applyQueuePlay(token, '');
case 'cancel':
changed = engine.applyQueueCancel(token);
}
if (intent.op != 'add') {
_log.info('${hub.who(token)}: ${intent.op}${intent.id.isEmpty ? '' : ' ${intent.id}'}'
packages/server/lib/src/relay_routes.dart
+52 -4
View File
@@ -75,6 +75,10 @@ class ActiveRelay {
RelicWebSocket? _channel;
var _channelFeeds = const <ChannelRelayFeed>[];
List<_TrackInfo>? _infos;
/// The source's first `info` on this relay: a later channel (the source
/// back within its grace) must send the same one, else the relay ends
/// (4002).
_SourceInfo? _firstInfo;
/// Shared by the tracks: time any of them waits on the source.
final clock = FeedClock();
final _rate = SourceRate();
@@ -295,7 +299,9 @@ class ActiveRelay {
void _attachChannel(RelicWebSocket ws, List<_TrackInfo> infos) {
final old = _channel;
if (old != null && !identical(old, ws)) unawaited(old.tryClose());
// A second channel for the same relay replaces the first (the
// source's app reconnected before the old link died).
if (old != null && !identical(old, ws)) unawaited(_closeWith(old, CloseCodes.relayReplaced, 'replaced'));
_channel = ws;
final named = tracks.isNotEmpty;
final feeds = _channelFeeds = [
@@ -317,6 +323,33 @@ class ActiveRelay {
}
}
/// A source's `info` as a whole: its tracks, and the optional `name` and
/// `duration` it gives. Two are the same media when the tracks agree
/// (kinds, byte sizes, piece shapes) and the name and duration agree
/// where both give one.
typedef _SourceInfo = ({List<_TrackInfo> tracks, String name, double duration});
bool _sameInfo(_SourceInfo a, _SourceInfo b) {
if (a.tracks.length != b.tracks.length) return false;
for (var i = 0; i < a.tracks.length; i++) {
final x = a.tracks[i], y = b.tracks[i];
if (x.kind != y.kind || x.size != y.size || (x.pieces == null) != (y.pieces == null)) return false;
if (x.pieces != null && !x.pieces!.sameShape(y.pieces!)) return false;
}
if (a.name.isNotEmpty && b.name.isNotEmpty && a.name != b.name) return false;
if (a.duration > 0 && b.duration > 0 && (a.duration - b.duration).abs() > 0.5) return false;
return true;
}
_SourceInfo _sourceInfo(Map j, List<_TrackInfo> tracks) {
final d = j['duration'];
return (
tracks: tracks,
name: j['name'] is String ? j['name'] as String : '',
duration: d is num && d.isFinite && d > 0 ? d.toDouble() : 0,
);
}
/// The tracks of a channel's `info` frame (`size`, or `tracks[]` each
/// with a `size` or, `"as":"pieces"`, a piece list), or null when
/// malformed.
@@ -342,9 +375,11 @@ List<_TrackInfo>? _infoTracks(Map j) {
return out;
}
Future<void> _closeNotSource(RelicWebSocket ws) async {
Future<void> _closeNotSource(RelicWebSocket ws) => _closeWith(ws, CloseCodes.relayNotSource, 'not the source');
Future<void> _closeWith(RelicWebSocket ws, int code, String reason) async {
try {
await ws.close(relayCloseNotSource, 'not the source');
await ws.close(code, reason);
} catch (_) {
await ws.tryClose();
}
@@ -357,12 +392,15 @@ Future<void> _closeNotSource(RelicWebSocket ws) async {
/// with tracks (a pieces track's media playlist), and
/// `.../<key>/<track>/p/<i>` piece `i` of a pieces track. `<key>` is the
/// media id, or the queue id of a warm (or promoted) relay. [relayAt]
/// finds the relay of a key, [sourceOf] the relay a token may feed.
/// finds the relay of a key, [sourceOf] the relay a token may feed;
/// [onOtherMedia] hears of a source back with another media than its
/// first `info` (the channel is closed with 4002; the relay should end).
void mountRelayRoutes(
RelicApp app,
SessionMap sessions, {
required ActiveRelay? Function(String key) relayAt,
required ActiveRelay? Function(String token, String? queueId) sourceOf,
void Function(ActiveRelay relay)? onOtherMedia,
void Function(String token, MediaLog log)? onLog,
}) {
app.get('/r/:sid/:token/source', (final req) {
@@ -397,12 +435,22 @@ void mountRelayRoutes(
case 'info':
if (attached) continue;
final infos = _infoTracks(j);
final first = r._firstInfo;
if (infos != null && first != null && !_sameInfo(first, _sourceInfo(j, infos))) {
// The source came back with another media: the relay
// ends (its buffer is of the first one).
_log.warning('relay channel: the source is back with another media (${r.name}); the relay ends');
await _closeWith(ws, CloseCodes.relayOtherMedia, 'other media');
onOtherMedia?.call(r);
return;
}
if (infos == null || !r._channelInfo(infos)) {
// Not the media this swap named: the source stops.
_log.warning('relay channel: info does not match the media (${r.name})');
await _closeNotSource(ws);
return;
}
r._firstInfo ??= _sourceInfo(j, infos);
attached = true;
_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(' + ')})');
packages/server/lib/src/server_hub.dart
+261 -89
View File
@@ -1,5 +1,6 @@
import 'dart:async';
import 'dart:convert';
import 'dart:math';
import 'dart:typed_data';
import 'package:relic/relic.dart';
@@ -19,15 +20,91 @@ class ServerHub {
/// See [AppConfig.hostKey]; empty means nobody is the host.
final String hostKey;
final SyncEngine engine;
/// The `/sync` socket of each member that has one, by token; a `gone`
/// member has none.
final Map<String, RelicWebSocket> sockets = {};
/// The server's Jellyfin login for the `/j/...` relay; null = no relay.
JellyfinCredential? jellyfin;
/// Reads web pages for "the server fetches it"; null = files only.
final YtDlp? ytDlp;
/// Random per server boot (`welcome.instance`).
final String instance = _randomId();
ServerHub(this.sessions, this.expectedSecret,
{SyncEngine? engine, this.hostKey = '', this.jellyfin, this.ytDlp})
: engine = engine ?? SyncEngine();
: engine = engine ?? SyncEngine() {
this.engine
..onGone = _memberGone
..onLeft = _memberLeft;
}
static String _randomId() {
final r = Random.secure();
return base64Url.encode(List<int>.generate(12, (_) => r.nextInt(256))).replaceAll('=', '');
}
/// A member went `gone`: a socket still open (silent) is closed; a
/// relay it serves goes on from its buffer (`media.relay.sourceGone`).
void _memberGone(EngineMember m) {
_hubLog.info('gone ${m.name} (${m.id})');
final ws = sockets.remove(m.token);
if (ws != null) unawaited(closeWith(ws, CloseCodes.silent, 'silent'));
refreshRelaySource();
}
/// A member left for good (`bye`, grace over): its token stops working,
/// a relay it serves ends.
void _memberLeft(EngineMember m) {
_hubLog.info('leave ${m.name} (${m.id})');
final ws = sockets.remove(m.token);
if (ws != null) unawaited(closeWith(ws, CloseCodes.left, 'left'));
session.tokens.remove(m.token);
final r = relay;
if (r != null && !r.fetch && r.sourceToken == m.token) {
_endRelay(r, 'its source left', sourceGone: true, since: m.goneSince ?? engine.now);
}
}
/// The current relay ends (its source left, or sent another media):
/// the buffer goes, `/r` answers 404, `media.relay` stays with `live`
/// false.
void _endRelay(ActiveRelay r, String why, {bool sourceGone = false, DateTime? since}) {
if (!identical(relay, r)) return;
_hubLog.info('relay of "${r.name}" ended: $why');
setRelay(null);
final m = engine.media;
final was = m?.relay;
if (m != null && was != null) {
engine.media = m.copyWith(
relay: was.copyWith(live: false, rate: 0, sourceGone: sourceGone, sourceGoneSince: since));
}
}
/// The source of [r] sent an `info` unlike its first (another media):
/// the relay ends (warm: dropped, made again from the next `info`).
void relayOtherMedia(ActiveRelay r) {
if (identical(warm, r)) {
_setWarm(null);
} else {
_endRelay(r, 'its source sent another media');
}
changed();
}
/// `media.relay.sourceGone` / `sourceGoneSince` follow the current
/// relay's source member (`gone` or back).
void refreshRelaySource() {
final r = relay, m = engine.media;
final was = m?.relay;
if (r == null || m == null || was == null || r.fetch || r.sourceToken == null) return;
final src = engine.members[r.sourceToken];
final gone = src == null || src.presence == Presence.gone;
if (gone == was.sourceGone) return;
_hubLog.info('relay of "${r.name}": source ${gone ? 'gone, viewers play what is buffered' : 'back'}');
engine.media = m.copyWith(relay: was.copyWith(sourceGone: gone, sourceGoneSince: src?.goneSince ?? engine.now));
}
bool get _sourceGone => engine.media?.relay?.sourceGone ?? false;
/// The current media is a Jellyfin ref on the server [jellyfin] logs in to.
bool get relayAvailable => _onLogin(session.media?.jellyfin);
@@ -61,8 +138,8 @@ class ServerHub {
: '$head: $why';
}
StateBroadcast stateBroadcast() =>
engine.broadcast(relay: relayAvailable, relayNext: relayNextAvailable, warm: _warmView());
StateBroadcast stateBroadcast() => engine.broadcast(
relay: relayAvailable, relayNext: relayNextAvailable, warm: _warmView(), sourceGone: _sourceGone);
/// The relay of the current media (`/r/...`), if it is relayed.
ActiveRelay? relay;
@@ -109,10 +186,17 @@ class ServerHub {
({String id, MediaRelay relay})? _warmView() {
final w = warm, id = w?.queueId;
if (w == null || id == null) return null;
final src = w.sourceToken == null ? null : engine.members[w.sourceToken];
final gone = src?.presence == Presence.gone;
return (
id: id,
relay: MediaRelay(
source: w.fetch ? '' : w.sourceId, fetch: w.fetch, live: w.live, tracks: w.live ? w.tracks : const []),
source: w.fetch ? '' : w.sourceId,
fetch: w.fetch,
live: w.live,
tracks: w.live ? w.tracks : const [],
sourceGone: gone,
sourceGoneSince: gone ? src?.goneSince : null),
);
}
@@ -130,13 +214,15 @@ class ServerHub {
if (warm == null && first != null) _setWarm(_newWarm(first));
}
bool _othersHere(String? memberId) => engine.members.values.any((m) => m.id != memberId);
bool _othersHere(String? memberId) =>
engine.members.values.any((m) => m.id != memberId && m.presence != Presence.gone);
/// A Jellyfin [ref] someone here cannot play while someone can, outside
/// the server's login.
bool _jellyfinNeedsWarm(JellyfinRef ref) =>
!_onLogin(ref) &&
engine.members.values.any((m) => !m.out && m.jf[ref.serverId] != true) &&
engine.members.values
.any((m) => engine.policy.countsFor(m.presence, Purpose.preload) && m.jf[ref.serverId] != true) &&
engine.members.values.any((m) => m.jf[ref.serverId] == true);
bool _keepWarm(ActiveRelay w, QueueItem first) {
@@ -302,7 +388,7 @@ class ServerHub {
bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey;
/// A member for the log: `name (id)`, never its token.
/// A member for the log: `name (id)`, never its token or client id.
String who(String token) {
final m = engine.members[token];
if (m != null) return '${m.name} (${m.id})';
@@ -310,13 +396,6 @@ class ServerHub {
return v != null ? '${v.name} (no member)' : 'unknown member';
}
/// The member a reconnecting client may take over with its old token:
/// still known.
EngineMember? resumable(String token) {
if (token.isEmpty || !session.tokens.containsKey(token)) return null;
return engine.members[token];
}
Session get session => sessions.getOrCreate('default', expectedSecret);
void broadcastState() {
@@ -326,6 +405,19 @@ class ServerHub {
}
}
/// Once a second: closed sockets, then the engine's sweep (silence,
/// joining, gone notices, grace), relay source, warm relay, rates, and
/// a state for everyone.
void heartbeat() {
sweepClosedSockets();
engine.tick();
refreshRelaySource();
syncWarm();
sampleRates();
broadcastState();
broadcastFeed();
}
void broadcastFeed() {
for (final feed in engine.feed.drain()) {
final raw = jsonEncode(feed.toJson());
@@ -335,14 +427,8 @@ class ServerHub {
}
}
void dropSilentSockets() {
for (final token in engine.sweepDropped()) {
final ws = sockets[token];
if (ws != null) unawaited(ws.tryClose());
forgetSocket(token);
}
}
/// Relic's server-side `onDone` does not fire for every dead peer:
/// sockets are polled with `isClosed` too.
void sweepClosedSockets() {
for (final entry in sockets.entries.toList()) {
var closed = false;
@@ -351,7 +437,7 @@ class ServerHub {
} catch (_) {
closed = true;
}
if (closed) forgetSocket(entry.key);
if (closed) socketClosed(entry.key, entry.value);
}
}
@@ -365,24 +451,23 @@ class ServerHub {
}
}
void broadcastChat(String from, String text) {
void broadcastChat(String from, String text, {String fromId = ''}) {
final clean = text.length > RelayConstants.maxChatLength
? text.substring(0, RelayConstants.maxChatLength)
: text;
final raw = jsonEncode(
ChatMessage(text: clean).toJson()..addAll({'from': from}));
ChatMessage(text: clean).toJson()..addAll({'from': from, if (fromId.isNotEmpty) 'fromId': fromId}));
for (final ws in sockets.values.toList()) {
ws.trySendText(raw);
}
engine.feed.add(
FeedMessage(actor: from, action: 'chat', at: engine.now));
engine.feed.add(FeedMessage(actor: from, actorId: fromId, action: 'chat', at: engine.now));
broadcastFeed();
}
/// The server's own fetch of [name] failed: everyone hears it (the
/// swapper also gets the reason as an error), not just "getting it…".
void fetchFailed(String name) {
engine.feed.add(FeedMessage(actor: 'server', action: 'fetch-fail', target: name, at: engine.now));
engine.feed.add(FeedMessage(actor: 'server', actorId: 'server', action: 'fetch-fail', target: name, at: engine.now));
broadcastFeed();
}
@@ -392,18 +477,19 @@ class ServerHub {
broadcastFeed();
}
/// Like [forgetSocket], but only while [ws] is still the member's socket:
/// after a resume the old socket's close must not remove the member.
void forgetSocketOf(String token, RelicWebSocket ws) {
/// [ws], [token]'s socket, closed: the member is `gone` (kept for the
/// grace). A socket that is no longer the member's (taken over) changes
/// nothing.
void socketClosed(String token, RelicWebSocket ws) {
if (!identical(sockets[token], ws)) return;
forgetSocket(token);
sockets.remove(token);
engine.applyGone(token);
changed();
}
void forgetSocket(String token) {
if (engine.members.containsKey(token)) _hubLog.info('leave ${who(token)}');
engine.applyLeave(token);
session.tokens.remove(token);
sockets.remove(token);
/// `bye`: [token]'s member leaves now.
void bye(String token) {
engine.applyLeave(token, reason: 'bye');
changed();
}
@@ -478,8 +564,20 @@ String sanitizeMediaName(String name) {
return 'media';
}
void wsErr(RelicWebSocket ws, String message) {
ws.trySendText(jsonEncode({'t': 'err', 'message': message}));
/// Sends `err` [code] with [message] (plain text for people) and, when a
/// frame failed, its type as `for`. The socket stays open.
void wsErr(RelicWebSocket ws, ErrCode code, String message, {String? forType}) {
ws.trySendText(jsonEncode(ErrorFrame(code: code, message: message, forType: forType).toJson()));
}
/// Closes [ws] with [code] (see [CloseCodes]); a socket that is already
/// gone is fine.
Future<void> closeWith(RelicWebSocket ws, int code, String reason) async {
try {
await ws.close(code, reason);
} catch (_) {
await ws.tryClose();
}
}
/// Jellyfin ids are path segments of upstream requests: keep them plain.
@@ -532,9 +630,12 @@ void mountSyncRoute(
}
void handleIntent(String text) {
// A socket another hello took over says nothing for the member.
if (token != null && !identical(hub.sockets[token], ws)) return;
final session = hub.session;
final viewer = token == null ? null : session.tokens[token];
viewer?.touch(hub.engine.now);
if (token != null) hub.engine.touch(token);
Map<String, Object?> decoded;
try {
final parsed = jsonDecode(text);
@@ -575,7 +676,12 @@ void mountSyncRoute(
final hidden = hub.engine.queue.privateUrlOf(queueId);
if (hidden != null && (parsed.url ?? '').isEmpty) {
parsed = HelloMedia(
name: parsed.name, size: parsed.size, duration: parsed.duration, url: hidden, relay: parsed.relay);
name: parsed.name,
size: parsed.size,
duration: parsed.duration,
url: hidden,
relay: parsed.relay,
ref: parsed.ref);
}
}
if (parsed.jellyfin != null) {
@@ -594,6 +700,7 @@ void mountSyncRoute(
size: parsed.size,
duration: parsed.duration,
jellyfin: ref,
ref: parsed.ref,
relay: relayReq == null
? null
: promoted != null
@@ -601,7 +708,7 @@ void mountSyncRoute(
: hub.relayFor(token, const MediaRelay()));
if (media == null || !hub.engine.applySwap(token, media, queueId: queueId)) {
promoted?.dispose();
wsErr(ws, 'bad swap media');
wsErr(ws, ErrCode.bad, 'bad swap media', forType: 'swap');
return;
}
session.media = SessionMedia(
@@ -627,14 +734,14 @@ void mountSyncRoute(
final link = Uri.tryParse(parsed.url ?? '');
if (relayReq.fetch &&
(link == null || !(link.isScheme('http') || link.isScheme('https')) || link.host.isEmpty)) {
wsErr(ws, 'relay fetch needs an http(s) url');
wsErr(ws, ErrCode.bad, 'relay fetch needs an http(s) url', forType: 'swap');
return;
}
// The server would fetch any address it can reach (its own
// network, a VPN) for whoever asks: only for members who
// know the password.
if (relayReq.fetch && session.secret.isEmpty) {
wsErr(ws, 'server fetch needs a server password');
wsErr(ws, ErrCode.rejected, 'server fetch needs a server password', forType: 'swap');
return;
}
// Up next's warm relay of this item carries on as its relay
@@ -645,16 +752,17 @@ void mountSyncRoute(
name: shown,
size: parsed.size,
duration: parsed.duration,
ref: parsed.ref,
relay: promoted != null ? hub.promotedRelay(promoted, queueId!) : hub.relayFor(token, relayReq));
if (!hub.engine.applySwap(token, media, queueId: queueId)) {
promoted?.dispose();
wsErr(ws, 'bad swap media');
wsErr(ws, ErrCode.bad, 'bad swap media', forType: 'swap');
return;
}
session.media = SessionMedia(
name: shown, size: parsed.size, duration: parsed.duration, mediaId: hub.engine.mediaId);
void failed(String why) {
wsErr(ws, hub.fetchFailText(why));
wsErr(ws, ErrCode.rejected, hub.fetchFailText(why), forType: 'swap');
hub.fetchFailed(shown);
}
late final ActiveRelay active;
@@ -678,17 +786,18 @@ void mountSyncRoute(
name: sanitizeMediaName(parsed.name),
size: parsed.size,
duration: parsed.duration,
url: parsed.url));
url: parsed.url,
ref: parsed.ref));
final current = session.media;
final matchesFile = current?.file != null &&
current!.name == intent.media.name &&
current.size == intent.media.size;
if (!matchesFile && (intent.media.url ?? '').isEmpty) {
wsErr(ws, 'swap needs a url or the hosted file');
wsErr(ws, ErrCode.bad, 'swap needs a url or the hosted file', forType: 'swap');
return;
}
if (!hub.engine.applySwap(token, intent.media, queueId: queueId)) {
wsErr(ws, 'bad swap media');
wsErr(ws, ErrCode.bad, 'bad swap media', forType: 'swap');
return;
}
hub.setRelay(null);
@@ -714,7 +823,7 @@ void mountSyncRoute(
final intent = ProfileIntent.fromJson(decoded);
final avatar = intent.avatar;
if (avatar != null && avatar.length > RelayConstants.maxAvatarBytes) {
wsErr(ws, 'rejected: avatar too large');
wsErr(ws, ErrCode.rejected, 'rejected: avatar too large', forType: 'profile');
return;
}
final name = intent.name == null ? null : cleanMemberName(intent.name!);
@@ -749,12 +858,19 @@ void mountSyncRoute(
.toJson()));
case 'chat':
final chat = ChatMessage.fromJson(decoded);
hub.broadcastChat(viewer?.name ?? '?', chat.text);
case 'out':
hub.broadcastChat(viewer?.name ?? '?', chat.text, fromId: hub.engine.members[token]?.id ?? '');
case 'presence':
if (token == null) return;
final intent = OutIntent.fromJson(decoded);
hub.engine.applyOut(token, intent.target, intent.out);
final intent = PresenceIntent.fromJson(decoded);
final refused = hub.engine.applyPresence(token, intent.target, intent.state, reason: intent.reason);
if (refused != null) {
wsErr(ws, refused, 'only a member itself comes back in', forType: 'presence');
}
hub.changed();
case 'bye':
if (token == null) return;
_hubLog.info('bye from ${hub.who(token)}');
hub.bye(token);
case 'nudge':
if (token == null) return;
hub.nudge(token, NudgeIntent.fromJson(decoded).target);
@@ -764,7 +880,7 @@ void mountSyncRoute(
}
} 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, ErrCode.bad, 'bad message: ${e.message}', forType: type is String ? type : null);
} catch (e, st) {
_hubLog.warning('${token == null ? 'a socket' : hub.who(token)}: $type message failed: $e', e, st);
return;
@@ -781,13 +897,13 @@ void mountSyncRoute(
}
},
onError: (_) {
if (token != null) hub.forgetSocketOf(token, ws);
if (token != null) hub.socketClosed(token, ws);
},
onDone: () {
if (!firstText.isCompleted) {
firstText.completeError(StateError('closed before hello'));
} else if (token != null) {
hub.forgetSocketOf(token, ws);
hub.socketClosed(token, ws);
}
},
);
@@ -798,64 +914,121 @@ void mountSyncRoute(
await close();
return;
}
HelloMessage hello;
Future<void> refuse(ErrorFrame err, int code, String log) async {
_hubLog.warning('reject: $log');
ws.trySendText(jsonEncode(err.toJson()));
try {
await sub.cancel();
} catch (_) {}
await closeWith(ws, code, err.code?.name ?? 'refused');
}
// The version first, read leniently: another version's hello may
// not parse as ours.
final Map<String, Object?> raw0;
try {
final decoded = jsonDecode(raw);
if (decoded is! Map) throw const ProtocolException('hello must be an object');
hello = HelloMessage.fromJson(Map<String, Object?>.from(decoded));
raw0 = Map<String, Object?>.from(decoded);
} catch (e) {
_hubLog.warning('reject: bad hello: $e');
wsErr(ws, 'bad hello: $e');
await close();
await refuse(ErrorFrame(code: ErrCode.bad, message: 'bad hello: $e', forType: 'hello'), CloseCodes.refused,
'bad hello: $e');
return;
}
if (hello.proto != RelayConstants.protoVersion) {
_hubLog.warning(
'reject ${hello.user.name}: proto mismatch: got ${hello.proto}, want ${RelayConstants.protoVersion}');
wsErr(ws, 'proto mismatch: got ${hello.proto}, want ${RelayConstants.protoVersion}');
await close();
final range = HelloMessage.protoRangeOf(raw0);
final proto = range == null ? null : pickProto(range.min, range.max);
if (proto == null) {
final got = range?.max ?? 0;
await refuse(ErrorFrame.version(got: got, server: appVersion), CloseCodes.version,
'proto mismatch: got ${range == null ? 'none' : '${range.min}..${range.max}'}, '
'want ${RelayConstants.protoMin}..${RelayConstants.protoMax}');
return;
}
HelloMessage hello;
try {
hello = HelloMessage.fromJson(raw0);
} catch (e) {
await refuse(ErrorFrame(code: ErrCode.bad, message: 'bad hello: $e', forType: 'hello'), CloseCodes.refused,
'bad hello: $e');
return;
}
final session = hub.session;
if (session.secret.isNotEmpty && hello.secret != session.secret) {
_hubLog.warning('reject ${hello.user.name}: wrong secret');
wsErr(ws, 'wrong secret');
await close();
await refuse(const ErrorFrame(code: ErrCode.secret, message: 'wrong secret', forType: 'hello'),
CloseCodes.secret, '${cleanMemberName(hello.user.name)}: wrong secret');
return;
}
final resumed = hub.resumable(hello.resume);
String memberId;
if (resumed != null) {
// Unclean drop the server has not noticed yet: take the member over
// (same token, id, ready flag) and drop the dead socket quietly.
token = hello.resume;
memberId = resumed.id;
final old = hub.sockets.remove(token);
if (old != null) unawaited(old.tryClose());
hub.engine.applyResume(token);
_hubLog.info('resume ${hello.user.name} ($memberId)');
if (hello.user.avatar.length > RelayConstants.maxAvatarBytes) {
await refuse(
const ErrorFrame(code: ErrCode.rejected, message: 'rejected: avatar too large', forType: 'hello'),
CloseCodes.refused,
'${cleanMemberName(hello.user.name)}: avatar too large');
return;
}
final known = hub.engine.byClient(hello.client);
if (known != null && hello.seq < known.seq) {
await refuse(
const ErrorFrame(
code: ErrCode.stale, message: 'an older connection of this app (a newer one holds it)', forType: 'hello'),
CloseCodes.stale,
'${known.name} (${known.id}): an older hello than the member\'s');
return;
}
final String memberId;
final bool resumed;
final about = '${hello.app.isEmpty ? 'no version' : hello.app}${hello.platform.isEmpty ? '' : ' on ${hello.platform}'}';
if (known != null) {
// This app again (a reconnect, or opened anew): it takes its
// member over, same token and id; an older socket still open is
// told and closed.
token = known.token;
memberId = known.id;
resumed = true;
final old = hub.sockets[known.token];
hub.sockets[known.token] = ws;
if (old != null && !identical(old, ws)) {
old.trySendText(jsonEncode(const ErrorFrame(
code: ErrCode.replaced, message: 'this member was opened from another connection of the same app')
.toJson()));
unawaited(closeWith(old, CloseCodes.replaced, 'replaced'));
}
final back = hub.engine.applyTakeover(known.token,
seq: hello.seq, app: hello.app, platform: hello.platform, caps: hello.caps);
final name = cleanMemberName(hello.user.name);
session.tokens[known.token]?.name = name;
hub.engine.applyProfile(known.token, name: name, avatar: hello.user.avatar);
_hubLog.info('${back ? 'return' : 'takeover'} ${known.name} ($memberId), $about');
} else {
try {
token = session.issueToken(hello.user.name, hello.user.avatar);
} catch (e) {
_hubLog.warning('reject ${hello.user.name}: $e');
wsErr(ws, 'rejected: $e');
await close();
await refuse(ErrorFrame(code: ErrCode.rejected, message: 'rejected: $e', forType: 'hello'),
CloseCodes.refused, '${cleanMemberName(hello.user.name)}: $e');
return;
}
memberId = hub.engine.applyJoin(token, session.tokens[token]!.name, hello.user.avatar);
memberId = hub.engine.applyJoin(token, session.tokens[token]!.name, hello.user.avatar,
client: hello.client, seq: hello.seq, app: hello.app, platform: hello.platform, caps: hello.caps);
resumed = false;
hub.sockets[token] = ws;
_hubLog.info('join ${session.tokens[token]!.name} ($memberId), $about');
}
final isHost = hub.isHostKey(hello.hostKey);
session.tokens[token]!.isHost = isHost;
hub.engine.members[token]?.host = isHost;
final activeToken = token;
hub.sockets[activeToken] = ws;
final isHost = hub.isHostKey(hello.hostKey);
session.tokens[activeToken]!.isHost = isHost;
hub.engine.members[activeToken]?.host = isHost;
hub.refreshRelaySource();
final media = session.media;
final relay = hub.relayAvailable;
final relayed = hub.relay;
final feedAfter = hello.feedAfter;
final welcome = WelcomeMessage(
token: activeToken,
memberId: memberId,
proto: proto,
resumed: resumed,
instance: hub.instance,
grace: hub.engine.liveness.grace.inSeconds,
feed: feedAfter == null ? const [] : hub.engine.feed.after(feedAfter),
// Jellyfin media resolves per member (own account first); only
// relayed files and links start from the /r/ URL, and never for
// the source itself (it plays its own copy).
@@ -885,7 +1058,6 @@ void mountSyncRoute(
return;
}
welcomed = true;
_hubLog.info('join ${hello.user.name} ($memberId)');
hub.changed();
try {
await sub.asFuture();
packages/server/lib/src/sync_server.dart
+9 -9
View File
@@ -67,7 +67,7 @@ class SyncServer {
/// [ytDlp] replaces the server's own (tests). [clock] and [liveness]
/// go to the engine: a test passes a fake clock to step through the
/// silence limits ([Liveness]) without waiting.
/// silence, grace and joiner limits ([Liveness]) without waiting.
SyncServer(this.config,
{YtDlp? ytDlp, this.loopbackOnly = false, this.logDir, SyncClock? clock, Liveness liveness = const Liveness()}) {
hub = ServerHub(sessions, config.secret,
@@ -103,7 +103,11 @@ class SyncServer {
body: Body.fromString(jsonEncode({
'ok': true,
'name': config.name,
'proto': RelayConstants.protoVersion,
'proto': RelayConstants.protoMax,
'protoMin': RelayConstants.protoMin,
'protoMax': RelayConstants.protoMax,
'version': appVersion,
'tls': false,
'viewers': viewers,
'bytesServed': bytes,
'uptimeSecs': uptime,
@@ -113,7 +117,8 @@ class SyncServer {
final requests = MediaRequestLog();
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);
mountRelayRoutes(_app, sessions,
relayAt: hub.relayAt, sourceOf: hub.sourceOf, onOtherMedia: hub.relayOtherMedia, onLog: logRequest);
mountAvatarRoute(_app, hub);
mountArtRoute(_app, hub);
mountSessionLogRoutes(_app, hub, () => _sessionLog);
@@ -174,12 +179,7 @@ class SyncServer {
);
_startedAt = DateTime.now();
_heartbeat = Timer.periodic(const Duration(seconds: 1), (_) {
hub.engine.tick();
hub.sweepClosedSockets();
hub.dropSilentSockets();
hub.sampleRates();
hub.broadcastState();
hub.broadcastFeed();
hub.heartbeat();
_sessionLog?.flush();
});
_log.info('server "${config.name}" listening on $boundPort');
packages/server/test/flaky_link_test.dart
+41 -21
View File
@@ -1,9 +1,6 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence): both
// tests pin today's rules (a silent member is stale, then removed; a reset
// socket loses its member, so the redial is a new one, P1); with grace they
// expect `gone` and the same member back.
//
// A client reaching the server through FlakyProxy (DEVELOPMENT.md "Tests").
// A client reaching the server through FlakyProxy (DEVELOPMENT.md "Tests"):
// a dead link makes the member `gone` (kept with everything it had), and
// the same app dialling again gets the same member back (P1).
import 'dart:async';
import 'dart:convert';
@@ -22,7 +19,7 @@ class _Client {
Timer? _reports;
late final StreamSubscription<dynamic> _sub;
_Client(int port, {String resume = ''})
_Client(int port, {int seq = 1})
: ch = WebSocketChannel.connect(Uri.parse('ws://127.0.0.1:$port/sync')) {
ch.ready.then((_) {}, onError: (_) {});
_sub = ch.stream.listen((m) => frames.add(Map<String, Object?>.from(jsonDecode('$m') as Map)),
@@ -31,11 +28,15 @@ class _Client {
proto: RelayConstants.protoVersion,
secret: 'shh',
user: const HelloUser(name: 'alice'),
media: const HelloMedia(name: 'f.bin', size: 1, duration: 10.0),
resume: resume)
client: clientId,
seq: seq)
.toJson()));
}
static const clientId = 'flakyLinkTestClient01';
void send(Map<String, Object?> j) => ch.sink.add(jsonEncode(j));
Map<String, Object?>? get welcome => frames.where((f) => f['t'] == 'welcome').firstOrNull;
String get token => welcome!['token'] as String;
String get memberId => welcome!['memberId'] as String;
@@ -73,7 +74,7 @@ void main() {
setUp(() async {
// Short limits so the demo runs in a couple of seconds.
server = SyncServer(AppConfig(port: 0, secret: 'shh', name: 'test'),
liveness: const Liveness(staleAfter: Duration(milliseconds: 600), dropAfter: Duration(milliseconds: 1500)));
liveness: const Liveness(silence: Duration(milliseconds: 600)));
await server.start();
proxy = await FlakyProxy.start(server.boundPort);
});
@@ -87,25 +88,34 @@ void main() {
await server.stop();
});
_Client dial({String resume = ''}) {
final c = _Client(proxy.port, resume: resume);
_Client dial({int seq = 1}) {
final c = _Client(proxy.port, seq: seq);
clients.add(c);
return c;
}
test('a blackholed client goes stale, then is removed; it never hears of it', () async {
test('a blackholed client goes gone with its ready flag, queue item and relay; it comes back as itself', () async {
final c = dial();
await until(() => c.welcome != null, 'welcome');
c.startReports();
final engine = server.hub.engine;
EngineMember? member() => engine.members[c.token];
c.send(const ReadyIntent(ready: true).toJson());
c.send(const QueueIntent.add(QueueItem(
media: HelloMedia(name: 'f.mkv', size: 10, duration: 0), how: QueueHow.file, title: 'f', key: 'key-of-f-1'))
.toJson());
c.send(const SwapIntent(media: HelloMedia(name: 'mine.mkv', size: 100, duration: 0, relay: MediaRelay())).toJson());
await until(() => engine.queue.length == 1 && server.hub.relay != null && member()!.ready, 'ready, queued, relayed');
await Future<void>.delayed(const Duration(milliseconds: 1200));
expect(engine.isStale(member()!), isFalse, reason: 'reports through the proxy keep it fresh');
expect(member()!.presence, isNot(Presence.gone), reason: 'reports through the proxy keep it present');
proxy.last!.blackhole();
final seen = c.frames.length;
await until(() => member() != null && engine.isStale(member()!), 'stale');
await until(() => member() == null, 'removed');
await until(() => member()?.presence == Presence.gone, 'gone');
expect(member()!.ready, isTrue);
expect(engine.queue.items.single.by, c.memberId);
expect(server.hub.relay, isNotNull, reason: 'the relay lives through the grace (D6)');
expect(engine.media!.relay!.sourceGone, isTrue);
expect(server.hub.sockets.containsKey(c.token), isFalse);
// Half-open: the server's close never reached the client.
await Future<void>.delayed(const Duration(milliseconds: 300));
@@ -113,20 +123,30 @@ void main() {
expect(c.frames.length, seen);
expect(proxy.last!.isClosed, isFalse);
expect(proxy.last!.dropped, greaterThan(0));
final again = dial(seq: 2);
await until(() => again.welcome != null, 'welcome after redial');
expect(again.welcome!['resumed'], isTrue);
expect([again.memberId, again.token], [c.memberId, c.token]);
expect(engine.members, hasLength(1));
expect(member()!.presence, Presence.joining, reason: 'media is up: it loads, then comes in (D2)');
expect(member()!.ready, isTrue);
expect(engine.media!.relay!.sourceGone, isFalse);
final actions = [for (final f in again.frames) if (f['t'] == 'feed') f['action']];
expect(actions, isNot(contains('join')), reason: 'a return, not a join');
});
test('a reset client sees the drop at once and redials; today it comes back as a new member', () async {
test('a reset client sees the drop at once and redials into the same member', () async {
final c = dial();
await until(() => c.welcome != null, 'welcome');
final first = c.memberId;
proxy.last!.reset();
await until(() => c.ended, 'client sees the reset');
await until(() => !server.hub.engine.members.containsKey(c.token), 'server forgets the socket');
await until(() => server.hub.engine.members[c.token]?.presence == Presence.gone, 'server sees it gone');
final again = dial(resume: c.token);
final again = dial(seq: 2);
await until(() => again.welcome != null, 'welcome after redial');
expect(proxy.links, hasLength(2));
expect(again.memberId, isNot(first), reason: 'P1: no grace yet');
expect([again.memberId, again.token], [c.memberId, c.token]);
expect(server.hub.engine.members, hasLength(1));
});
}
packages/server/test/hardening_test.dart
+16 -16
View File
@@ -1,5 +1,3 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence): "anyone
// sits anyone out" (Away); the rest stays.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -25,7 +23,6 @@ class _Peer {
proto: RelayConstants.protoVersion,
secret: secret,
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0),
hostKey: hostKey)
.toJson()));
await p.next('welcome');
@@ -81,6 +78,7 @@ void main() {
a.send('{"t":"report","position":1e400,"buffered":1e400,"eventId":0}');
final err = await a.next('err');
expect('${err['message']}', contains('finite'));
expect([err['code'], err['for']], ['bad', 'seek']);
expect(server.hub.engine.position, 0);
// still alive: a valid seek round-trips through a broadcast
a.send('{"t":"seek","position":5}');
@@ -125,7 +123,7 @@ void main() {
await a.ws.close();
});
test('anyone sits anyone out (host too); nudges reach only the target, capped', () async {
test('anyone sets anyone away (host too), only oneself comes back in; nudges reach only the target, capped', () async {
await server.stop();
server = SyncServer(AppConfig(port: 0, secret: '', name: 't', hostKey: 'hk'));
await server.start();
@@ -137,19 +135,23 @@ void main() {
server.hub.engine.members.values.firstWhere((m) => m.name == name).id;
EngineMember of(String name) =>
server.hub.engine.members.values.firstWhere((m) => m.name == name);
friend.send(jsonEncode(OutIntent(target: idOf('host'), out: true).toJson()));
friend.send(jsonEncode(PresenceIntent(target: idOf('host'), state: Presence.away).toJson()));
var feed = await friend.next('feed');
while (feed['action'] != 'out') {
while (feed['action'] != 'away') {
feed = await friend.next('feed');
}
expect([feed['actor'], feed['target']], ['friend', 'host']);
expect(of('host').out, isTrue);
expect(of('host').inSet, isFalse);
expect([feed['actor'], feed['target'], feed['actorId'], feed['targetId']],
['friend', 'host', idOf('friend'), idOf('host')]);
expect(of('host').presence, Presence.away);
expect(server.hub.engine.countsForGate(of('host')), isFalse);
// Only oneself comes back.
friend.send(jsonEncode(OutIntent(target: idOf('host'), out: false).toJson()));
host.send(jsonEncode(OutIntent(target: idOf('host'), out: false).toJson()));
friend.send(jsonEncode(PresenceIntent(target: idOf('host'), state: Presence.watching).toJson()));
final err = await friend.next('err');
expect([err['code'], err['for']], ['forbidden', 'presence']);
expect(of('host').presence, Presence.away);
host.send(jsonEncode(const PresenceIntent(state: Presence.watching).toJson()));
await Future.delayed(const Duration(milliseconds: 200));
expect(of('host').out, isFalse);
expect(of('host').presence, Presence.watching, reason: 'no media up: nothing to load');
// 5 at once: 3 pass (per second), only to the target.
for (var i = 0; i < 5; i++) {
@@ -224,8 +226,7 @@ void main() {
ws.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: '',
user: const HelloUser(name: 'a'),
media: const HelloMedia(name: '', size: 0, duration: 0))
user: const HelloUser(name: 'a'))
.toJson()));
final w = await p.next('welcome');
expect('${w['mediaUrl']}', startsWith('http://127.0.0.1:${server.boundPort}/m/'));
@@ -239,8 +240,7 @@ void main() {
ws2.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: '',
user: const HelloUser(name: 'c'),
media: const HelloMedia(name: '', size: 0, duration: 0))
user: const HelloUser(name: 'c'))
.toJson()));
final w2 = await p2.next('welcome');
expect('${w2['mediaUrl']}', startsWith('http://$lan:${server.boundPort}/m/'));
packages/server/test/jellyfin_relay_live_test.dart
+1 -2
View File
@@ -54,8 +54,7 @@ void main() {
ws.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: '',
user: const HelloUser(name: 'live'),
media: const HelloMedia(name: '', size: 0, duration: 0))
user: const HelloUser(name: 'live'))
.toJson()));
final welcome = await frames.firstWhere((f) => f['t'] == 'welcome').timeout(const Duration(seconds: 5));
ws.add(jsonEncode(SwapIntent(
packages/server/test/jellyfin_relay_test.dart
+1 -3
View File
@@ -133,9 +133,7 @@ class _Client {
c.send(HelloMessage(
proto: RelayConstants.protoVersion,
secret: '',
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0),
).toJson());
user: HelloUser(name: name)).toJson());
return (c, await c._welcome.future.timeout(const Duration(seconds: 5)));
}
}
packages/server/test/lifecycle_test.dart
+14 -8
View File
@@ -1,5 +1,3 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence): a
// socket close no longer removes the member (explicit `bye`, grace).
import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -74,14 +72,22 @@ void main() {
await server.stop();
});
test('clean client close removes member within seconds, not timeout',
() async {
test('a clean close is not a leave: the member is gone at once, kept', () async {
final client = await _ProbeClient.hello(server.boundPort, 'alice');
await _pollBool(
() => server.hub.engine.members.length == 1, 'join visible');
await _pollBool(() => server.hub.engine.members.length == 1, 'join visible');
await client.ws.close();
await _pollBool(
() => server.hub.engine.members.isEmpty, 'member removed');
await _pollBool(() => server.hub.engine.members.values.single.presence == Presence.gone, 'gone');
expect(server.session.tokens, hasLength(1), reason: 'its token works through the grace');
expect(server.hub.sockets, isEmpty);
});
test('bye leaves at once: member, token and socket go', () async {
final client = await _ProbeClient.hello(server.boundPort, 'alice');
await _pollBool(() => server.hub.engine.members.length == 1, 'join visible');
client.ws.add(jsonEncode(const ByeMessage().toJson()));
final outcome = await client.closed.future.timeout(const Duration(seconds: 5));
expect(outcome, contains('code=${CloseCodes.left}'));
expect(server.hub.engine.members, isEmpty);
expect(server.session.tokens, isEmpty);
expect(server.hub.sockets, isEmpty);
});
packages/server/test/presence_test.dart
+192
View File
@@ -0,0 +1,192 @@
// Identity, grace and the feed across reconnects, on a fake clock: the
// server's limits ([Liveness]) are stepped through, not waited for.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:logging/logging.dart';
import 'package:sync_relay_core/core.dart';
import 'package:sync_relay_server/server.dart';
import 'package:test/test.dart';
class _Peer {
final WebSocket ws;
final frames = <Map<String, Object?>>[];
late final WelcomeMessage welcome;
_Peer(this.ws) {
ws.listen((m) => frames.add(Map<String, Object?>.from(jsonDecode('$m') as Map)), onError: (_) {});
}
static Future<_Peer> join(int port, String name, {String client = '', int seq = 0, int? feedAfter}) async {
final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync');
final p = _Peer(ws);
ws.add(jsonEncode(HelloMessage(
secret: 's', user: HelloUser(name: name), client: client, seq: seq, app: '1.3.0', feedAfter: feedAfter)
.toJson()));
final end = DateTime.now().add(const Duration(seconds: 5));
while (!p.frames.any((f) => f['t'] == 'welcome')) {
if (DateTime.now().isAfter(end)) fail('no welcome: ${p.frames}');
await Future<void>.delayed(const Duration(milliseconds: 20));
}
p.welcome = WelcomeMessage.fromJson(p.frames.firstWhere((f) => f['t'] == 'welcome'));
return p;
}
List<Map<String, Object?>> get feed => [for (final f in frames) if (f['t'] == 'feed') f];
}
Future<void> _until(bool Function() ok, String what) async {
final end = DateTime.now().add(const Duration(seconds: 5));
while (!ok()) {
if (DateTime.now().isAfter(end)) fail('timed out: $what');
await Future<void>.delayed(const Duration(milliseconds: 20));
}
}
void main() {
const install = 'presenceTestInstall001';
late DateTime t;
late SyncServer server;
final peers = <_Peer>[];
Future<_Peer> join(String name, {String client = '', int seq = 0, int? feedAfter}) async {
final p = await _Peer.join(server.boundPort, name, client: client, seq: seq, feedAfter: feedAfter);
peers.add(p);
return p;
}
/// Moves the server's clock on and runs one heartbeat.
void step(Duration d) {
t = t.add(d);
server.hub.heartbeat();
}
setUp(() async {
t = DateTime.utc(2026, 1, 1);
server = SyncServer(const AppConfig(port: 0, secret: 's', name: 't'), clock: () => t);
await server.start();
});
tearDown(() async {
for (final p in peers) {
await p.ws.close();
}
peers.clear();
await server.stop();
});
test('a gone member leaves when its grace runs out; it comes back as a new member', () async {
final alice = await join('alice', client: install, seq: 1);
final bob = await join('bob');
await alice.ws.close();
await _until(() => server.hub.engine.byClient(install)?.presence == Presence.gone, 'gone');
expect(alice.welcome.grace, RelayConstants.graceSecs);
// Bob keeps talking; alice says nothing.
for (var i = 0; i < 59; i++) {
server.hub.engine.touch(bob.welcome.token);
step(const Duration(seconds: 10));
}
expect(server.hub.engine.byClient(install), isNotNull, reason: '9 min 50 s: still kept');
expect(server.session.tokens.containsKey(alice.welcome.token), isTrue);
server.hub.engine.touch(bob.welcome.token);
step(const Duration(seconds: 10));
expect(server.hub.engine.byClient(install), isNull);
expect(server.session.tokens.containsKey(alice.welcome.token), isFalse, reason: 'its URLs stop working');
await _until(() => bob.feed.any((f) => f['action'] == 'leave'), 'leave line');
final leave = bob.feed.firstWhere((f) => f['action'] == 'leave');
expect([leave['actorId'], leave['reason']], [alice.welcome.memberId, 'grace']);
final again = await join('alice', client: install, seq: 2);
expect(again.welcome.resumed, isFalse);
expect(again.welcome.memberId, isNot(alice.welcome.memberId));
});
test('a quick return says nothing; a slow one says gone, then return; never join', () async {
final alice = await join('alice', client: install, seq: 1);
final bob = await join('bob');
await alice.ws.close();
await _until(() => server.hub.engine.byClient(install)?.presence == Presence.gone, 'gone');
await _until(
() => bob.frames.any((f) =>
f['t'] == 'state' && (f['members'] as List).any((m) => (m as Map)['presence'] == 'gone')),
'presence in state at once');
step(const Duration(seconds: 5));
final quick = await join('alice', client: install, seq: 2);
await Future<void>.delayed(const Duration(milliseconds: 200));
expect(quick.welcome.resumed, isTrue);
expect(bob.feed.map((f) => f['action']), isNot(anyOf(contains('gone'), contains('return'), contains('leave'))));
expect(bob.feed.where((f) => f['action'] == 'join'), hasLength(1), reason: "bob's own join only");
await quick.ws.close();
await _until(() => server.hub.engine.byClient(install)?.presence == Presence.gone, 'gone again');
step(const Duration(seconds: 10));
await _until(() => bob.feed.any((f) => f['action'] == 'gone'), 'gone line after 10 s');
await join('alice', client: install, seq: 3);
await _until(() => bob.feed.any((f) => f['action'] == 'return'), 'return line');
final ret = bob.feed.firstWhere((f) => f['action'] == 'return');
expect([ret['actor'], ret['actorId']], ['alice', alice.welcome.memberId]);
});
test('feed lines carry ids and seq; a reconnect replays what came after feedAfter', () async {
final alice = await join('alice', client: install, seq: 1);
final bob = await join('bob');
bob.ws.add(jsonEncode(const ReadyIntent(ready: true).toJson()));
await _until(() => alice.feed.any((f) => f['action'] == 'ready'), 'ready line');
final seqs = [for (final f in alice.feed) f['seq'] as int];
expect(seqs, orderedEquals([...seqs]..sort()));
final ready = alice.feed.firstWhere((f) => f['action'] == 'ready');
expect(ready['actorId'], bob.welcome.memberId);
final last = seqs.last;
await alice.ws.close();
bob.ws.add(jsonEncode(const ChatMessage(text: 'while you were out').toJson()));
bob.ws.add(jsonEncode(const ReadyIntent(ready: false).toJson()));
await _until(() => server.hub.engine.feed.lastSeq >= last + 2, 'lines while away');
final again = await join('alice', client: install, seq: 2, feedAfter: last);
final replayed = again.welcome.feed;
expect(replayed.map((f) => f.action), containsAllInOrder(['chat', 'unready']));
expect(replayed.every((f) => f.replayed && f.seq > last), isTrue);
expect(again.feed, isEmpty, reason: 'replayed only inside welcome');
});
test('a server restart: resumed false, another instance', () async {
final alice = await join('alice', client: install, seq: 1);
final before = alice.welcome.instance;
expect(before, isNotEmpty);
await server.stop();
server = SyncServer(const AppConfig(port: 0, secret: 's', name: 't'), clock: () => t);
await server.start();
final again = await join('alice', client: install, seq: 2);
expect(again.welcome.resumed, isFalse);
expect(again.welcome.instance, isNot(before));
});
test('members carry presence and app; the client id is never broadcast nor logged', () async {
final records = <String>[];
final sub = Logger.root.onRecord.listen((r) => records.add('${r.message} ${r.error ?? ''}'));
addTearDown(sub.cancel);
final alice = await join('alice', client: install, seq: 3);
final old = await WebSocket.connect('ws://127.0.0.1:${server.boundPort}/sync');
old.add(jsonEncode(const HelloMessage(secret: 's', user: HelloUser(name: 'alice'), client: install, seq: 1).toJson()));
await old.drain<void>().timeout(const Duration(seconds: 5));
expect(old.closeCode, CloseCodes.stale);
await alice.ws.close();
await _until(() => server.hub.engine.byClient(install)?.presence == Presence.gone, 'gone');
final again = await join('alice', client: install, seq: 4);
await _until(() => again.frames.any((f) => f['t'] == 'state'), 'state');
final state = again.frames.lastWhere((f) => f['t'] == 'state');
final me = (state['members'] as List).cast<Map>().single;
expect([me['presence'], me['app']], ['in', '1.3.0']);
final everything = [...records, ...again.frames.map(jsonEncode), ...alice.frames.map(jsonEncode)].join('\n');
expect(records, isNotEmpty);
expect(everything, isNot(contains(install)));
expect(redactLog('bad hello: {"client":"$install"}'), isNot(contains(install)));
});
test('server close code for a refused secret; the err names its code', () async {
final ws = await WebSocket.connect('ws://127.0.0.1:${server.boundPort}/sync');
final frames = <String>[];
ws.listen((m) => frames.add('$m'), onError: (_) {});
ws.add(jsonEncode(const HelloMessage(secret: 'nope', user: HelloUser(name: 'x')).toJson()));
await _until(() => ws.closeCode != null, 'closed');
expect(ws.closeCode, CloseCodes.secret);
expect((jsonDecode(frames.single) as Map)['code'], 'secret');
});
}
packages/server/test/queue_test.dart
+23 -4
View File
@@ -1,5 +1,3 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence): queue
// `by` and adder-only keys follow the lasting client id.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -15,14 +13,15 @@ class _Peer {
ws.listen((m) => msgs.add(Map<String, Object?>.from(jsonDecode('$m') as Map)));
}
static Future<_Peer> join(int port, String name) async {
static Future<_Peer> join(int port, String name, {String client = '', int seq = 0}) async {
final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync');
final p = _Peer(ws);
p.send(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 's',
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0))
client: client,
seq: seq)
.toJson());
await p.until((m) => m['t'] == 'welcome');
return p;
@@ -122,4 +121,24 @@ void main() {
expect(s.queue.single.media.url, isNull, reason: 'an adder-only item carries no path or link');
expect(s.queue.single.key, 'k1234567890');
});
test("an adder's items keep its lasting id through a drop and wait while it is gone", () async {
const install = 'queueTestInstallId0001';
final a = await _Peer.join(server.boundPort, 'alice', client: install, seq: 1);
final b = await _Peer.join(server.boundPort, 'bob');
a.send(const QueueIntent.add(QueueItem(
media: HelloMedia(name: 'f.mkv', size: 10, duration: 0), how: QueueHow.file, title: 'f', key: 'k1234567890'))
.toJson());
final added = (await b.state((s) => s.queue.isNotEmpty)).queue.single;
final aliceId = added.by;
await a.ws.close();
// Gone: still a member, so whoever would put the item up waits for it
// instead of skipping it.
final whileGone = await b.state((s) => s.members.any((m) => m.id == aliceId && m.presence == Presence.gone));
expect(whileGone.queue.single.by, aliceId);
final again = await _Peer.join(server.boundPort, 'alice', client: install, seq: 2);
final back = await again.state((s) => s.members.any((m) => m.id == aliceId && m.presence != Presence.gone));
expect(back.queue.single.by, aliceId, reason: 'the same member: its items are its own again');
expect(back.members.where((m) => m.name == 'alice'), hasLength(1));
});
}
packages/server/test/relay_test.dart
+79 -6
View File
@@ -17,7 +17,7 @@ class _Member {
late WelcomeMessage welcome;
_Member(this.ws);
static Future<_Member> join(int port, String name, {String secret = 'pw'}) async {
static Future<_Member> join(int port, String name, {String secret = 'pw', String client = '', int seq = 0}) async {
final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync');
final m = _Member(ws);
final w = Completer<String>();
@@ -29,8 +29,8 @@ class _Member {
proto: RelayConstants.protoVersion,
secret: secret,
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0),
).toJson()));
client: client,
seq: seq).toJson()));
m.welcome = WelcomeMessage.fromJson(
Map<String, Object?>.from(jsonDecode(await w.future.timeout(const Duration(seconds: 5))) as Map));
return m;
@@ -86,8 +86,8 @@ void main() {
await tmp.delete(recursive: true);
});
Future<_Member> join(String name) async {
final m = await _Member.join(server.boundPort, name);
Future<_Member> join(String name, {String client = '', int seq = 0}) async {
final m = await _Member.join(server.boundPort, name, client: client, seq: seq);
cleanup.add(() => m.ws.close());
return m;
}
@@ -251,6 +251,75 @@ void main() {
expect(body, bytes.sublist(RelayBuffer.slice * 4));
});
group('the source drops (D6)', () {
const install = 'relaySourceInstall0001';
Map? relayOf(_Member m) => m.media?['relay'] as Map?;
test('gone: the relay stays, sourceGone, the buffer serves; back with the same info it continues', () async {
final alice = await join('alice', client: install, seq: 1);
final bob = await join('bob');
await swapRelay(alice);
final src = await startSource(alice);
final url = viewerUrl(bob, mediaIdOf(alice));
expect((await get(url, range: 'bytes=0-${RelayBuffer.slice * 2 - 1}')).$3, bytes.sublist(0, RelayBuffer.slice * 2));
await alice.ws.close();
await src.close();
await src.done;
await _until(() => relayOf(bob)?['sourceGone'] == true, 'sourceGone');
expect(relayOf(bob)!['sourceGoneSince'], isA<String>());
expect(server.hub.relay, isNotNull, reason: 'the relay lives through the grace');
expect((await get(url, range: 'bytes=0-99')).$3, bytes.sublist(0, 100), reason: 'what is buffered serves');
final pending = get(url, range: 'bytes=${RelayBuffer.slice * 4}-');
await Future<void>.delayed(const Duration(milliseconds: 300));
final again = await join('alice', client: install, seq: 2);
expect([again.welcome.resumed, again.welcome.token], [true, alice.welcome.token]);
await _until(() => relayOf(bob)?['sourceGone'] == null, 'source back');
await startSource(again);
final (status, _, body) = await pending.timeout(const Duration(seconds: 10));
expect(status, 206);
expect(body, bytes.sublist(RelayBuffer.slice * 4), reason: 'the rest is pulled from the source again');
});
test('back with another media: closed with 4002, the relay ends', () async {
final alice = await join('alice');
final bob = await join('bob');
await swapRelay(alice);
final src = await startSource(alice);
await _until(() => server.hub.relay?.live == true, 'live');
await src.close();
await src.done;
final other = File('${tmp.path}/other.mkv')..writeAsBytesSync(List.filled(1000, 3));
final wrong = await startSource(alice, from: RelayMedia.single(FileRelayBytes(other)));
await wrong.done.timeout(const Duration(seconds: 5));
await _until(() => server.hub.relay == null, 'relay ended');
await _until(() => relayOf(bob) != null && relayOf(bob)!['live'] == null, 'not live');
expect((await get(viewerUrl(bob, mediaIdOf(alice)), range: 'bytes=0-9')).$1, 404);
});
test('a second channel for the same relay replaces the first (4003)', () async {
final alice = await join('alice');
await swapRelay(alice);
final first = await startSource(alice);
await _until(() => server.hub.relay?.live == true, 'live');
await startSource(alice);
await first.done.timeout(const Duration(seconds: 5));
await Future<void>.delayed(const Duration(milliseconds: 200));
expect(server.hub.relay?.live, isTrue, reason: 'the second one feeds it');
});
test('its source leaving for good ends the relay', () async {
final alice = await join('alice');
final bob = await join('bob');
await swapRelay(alice);
final src = await startSource(alice);
await _until(() => server.hub.relay?.live == true, 'live');
alice.send(const ByeMessage().toJson());
await src.done.timeout(const Duration(seconds: 5));
await _until(() => relayOf(bob)?['sourceGone'] == true && relayOf(bob)?['live'] == null, 'ended');
expect(server.hub.relay, isNull);
});
});
test('two tracks: each byte-exact at its own URL, welcome is an EDL of both', () async {
final alice = await join('alice');
final bob = await join('bob');
@@ -554,13 +623,17 @@ void main() {
expect(server.hub.warm!.queueId, isNot(id));
});
test('not for an adder alone; dropped when its adder leaves', () async {
test('not for an adder alone; kept while its adder is gone, dropped when it leaves', () async {
final alice = await join('alice');
await add(alice, through('k-abcdefgh'));
expect(headRelay(alice), isNull);
final bob = await join('bob');
await _until(() => headRelay(bob) != null, 'warm relay once someone could read it');
await alice.ws.close();
await _until(() => headRelay(bob)?['sourceGone'] == true, 'adder gone');
expect(server.hub.warm, isNotNull, reason: 'its adder may come back within the grace');
server.hub.engine.applyLeave(alice.welcome.token, reason: 'bye');
server.hub.changed();
await _until(() => bob.lastState!['members'] is List && (bob.lastState!['members'] as List).length == 1, 'left');
expect(queueOf(bob), hasLength(1));
expect(headRelay(bob), isNull);
packages/server/test/server_auth_test.dart
+45 -32
View File
@@ -1,6 +1,3 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence):
// "proto mismatch rejected" (min/max range, refusal frame) and "silent
// socket dropped and member removed" (Gone with grace); the rest stays.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -43,15 +40,15 @@ void main() {
Future<String> hello(String secret,
{int proto = RelayConstants.protoVersion,
int? protoMin,
Duration timeout = const Duration(seconds: 5)}) async {
final ch = WebSocketChannel.connect(
Uri.parse('ws://127.0.0.1:${server.boundPort}/sync'));
final msg = HelloMessage(
proto: proto,
protoMin: protoMin,
secret: secret,
user: const HelloUser(name: 'alice'),
media:
const HelloMedia(name: 'f.bin', size: 8192, duration: 10.0)).toJson();
user: const HelloUser(name: 'alice')).toJson();
ch.sink.add(jsonEncode(msg));
try {
final first = await ch.stream.first.timeout(timeout);
@@ -73,7 +70,9 @@ void main() {
final decoded = jsonDecode(body) as Map;
expect(decoded['ok'], isTrue);
expect(decoded['name'], 'test');
expect(decoded['proto'], RelayConstants.protoVersion);
expect(decoded['proto'], RelayConstants.protoMax);
expect([decoded['protoMin'], decoded['protoMax'], decoded['version'], decoded['tls']],
[RelayConstants.protoMin, RelayConstants.protoMax, appVersion, false]);
expect(decoded['viewers'], 0);
expect(decoded['bytesServed'], 0);
expect(decoded.containsKey('media'), isFalse, reason: 'no media name leak');
@@ -81,7 +80,7 @@ void main() {
test('wrong secret gets err, right secret gets welcome+token', () async {
final bad = await hello('wrong');
expect(jsonDecode(bad)['t'], 'err');
expect([jsonDecode(bad)['t'], jsonDecode(bad)['code'], jsonDecode(bad)['message']], ['err', 'secret', 'wrong secret']);
final good = await hello('shh');
final decoded = jsonDecode(good) as Map;
@@ -96,10 +95,32 @@ void main() {
expect(server.session.tokens.length, 1);
});
test('proto mismatch rejected', () async {
final reply =
await hello('shh', proto: RelayConstants.protoVersion + 99);
expect(jsonDecode(reply)['t'], 'err');
test('version ranges: a common version is taken, none is refused with code version', () async {
final newer = jsonDecode(await hello('shh', protoMin: RelayConstants.protoMin, proto: RelayConstants.protoMax + 3));
expect([newer['t'], newer['proto']], ['welcome', RelayConstants.protoMax], reason: 'the highest common one');
final ahead = jsonDecode(await hello('shh', protoMin: RelayConstants.protoMax + 1, proto: RelayConstants.protoMax + 2));
expect([ahead['t'], ahead['code'], ahead['min'], ahead['max'], ahead['server']],
['err', 'version', RelayConstants.protoMin, RelayConstants.protoMax, appVersion]);
expect(server.session.tokens, hasLength(1));
});
test('a proto 15 hello gets the stable refusal it can show, then the version close code', () async {
final ch = WebSocketChannel.connect(Uri.parse('ws://127.0.0.1:${server.boundPort}/sync'));
// What a 1.2.x app sends: no protoMin, a media object, a resume token.
ch.sink.add(jsonEncode({
't': 'hello',
'proto': 15,
'secret': 'shh',
'user': {'name': 'old', 'avatar': ''},
'media': {'name': '', 'size': 0, 'duration': 0},
'resume': 'deadbeef',
}));
final frames = await ch.stream.toList().timeout(const Duration(seconds: 5));
final err = jsonDecode('${frames.single}') as Map;
expect(err['t'], 'err');
expect(err['code'], 'version');
expect(err['message'], startsWith('proto mismatch: got 15'));
expect(ch.closeCode, CloseCodes.version);
expect(server.session.tokens, isEmpty);
});
@@ -136,7 +157,7 @@ void main() {
await resp.drain();
});
test('silent socket dropped and member removed', () async {
test('a silent socket is closed and its member is gone, not removed', () async {
final ch = WebSocketChannel.connect(
Uri.parse('ws://127.0.0.1:${server.boundPort}/sync'));
final messages = <String>[];
@@ -148,9 +169,7 @@ void main() {
final msg = HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'shh',
user: const HelloUser(name: 'ghost'),
media: const HelloMedia(
name: 'f.bin', size: 8192, duration: 10.0)).toJson();
user: const HelloUser(name: 'ghost')).toJson();
ch.sink.add(jsonEncode(msg));
final deadline = DateTime.now().add(const Duration(seconds: 5));
while (messages.isEmpty && DateTime.now().isBefore(deadline)) {
@@ -160,18 +179,16 @@ void main() {
(jsonDecode(messages.firstWhere((m) => (jsonDecode(m) as Map)['t'] == 'welcome')) as Map)['token']
as String;
server.hub.engine.members[token]!.lastSeen =
DateTime.now().toUtc().subtract(const Duration(seconds: 61));
server.hub.dropSilentSockets();
DateTime.now().toUtc().subtract(const Duration(seconds: 16));
server.hub.engine.tick();
final closeDeadline = DateTime.now().add(const Duration(seconds: 5));
while ((!wsClosed ||
server.hub.engine.members.containsKey(token) ||
server.session.tokens.containsKey(token)) &&
DateTime.now().isBefore(closeDeadline)) {
while (!wsClosed && DateTime.now().isBefore(closeDeadline)) {
await Future.delayed(const Duration(milliseconds: 50));
}
expect(wsClosed, isTrue);
expect(server.hub.engine.members.containsKey(token), isFalse);
expect(server.session.tokens.containsKey(token), isFalse);
expect(ch.closeCode, CloseCodes.silent);
expect(server.hub.engine.members[token]!.presence, Presence.gone);
expect(server.session.tokens.containsKey(token), isTrue);
try {
await sub.cancel();
} catch (_) {}
@@ -193,9 +210,7 @@ void main() {
ch.sink.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'shh',
user: const HelloUser(name: 'hb'),
media: const HelloMedia(
name: 'f.bin', size: 8192, duration: 10.0))
user: const HelloUser(name: 'hb'))
.toJson()));
await Future.delayed(const Duration(seconds: 1));
final hbToken = server.session.tokens.keys
@@ -223,8 +238,7 @@ void main() {
ch.sink.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'shh',
user: HelloUser(name: name),
media: const HelloMedia(name: 'f.bin', size: 8192, duration: 10.0)).toJson()));
user: HelloUser(name: name)).toJson()));
final deadline = DateTime.now().add(const Duration(seconds: 5));
while (!messages.any((m) => m['t'] == 'welcome')) {
if (DateTime.now().isAfter(deadline)) fail('no welcome');
@@ -329,8 +343,7 @@ void main() {
ch.sink.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'shh',
user: HelloUser(name: 'alice', avatar: base64Encode(png)),
media: const HelloMedia(name: 'f.bin', size: 8192, duration: 10.0)).toJson()));
user: HelloUser(name: 'alice', avatar: base64Encode(png))).toJson()));
final welcome = await waitFor((m) => m['t'] == 'welcome');
final token = welcome['token'] as String;
final first = member(Map<String, Object?>.from(welcome['state'] as Map));
@@ -376,7 +389,7 @@ void main() {
ch.sink.add(jsonEncode(
ProfileIntent(avatar: 'x' * (RelayConstants.maxAvatarBytes + 1)).toJson()));
final err = await waitFor((m) => m['t'] == 'err');
expect(err['message'], 'rejected: avatar too large');
expect([err['code'], err['for'], err['message']], ['rejected', 'profile', 'rejected: avatar too large']);
} finally {
await sub.cancel();
try {
packages/server/test/session_log_test.dart
+1 -2
View File
@@ -71,8 +71,7 @@ void main() {
ch.sink.add(jsonEncode(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 's',
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0))
user: HelloUser(name: name))
.toJson()));
while (await msgs.moveNext()) {
final m = jsonDecode('${msgs.current}') as Map;
packages/server/test/session_test.dart
+5 -5
View File
@@ -64,17 +64,17 @@ void main() {
});
group('core welcome + proto', () {
test('proto 15: avatars by hash, profile intent', () {
expect(RelayConstants.protoVersion, 15);
const v = MemberView(id: 'm1', name: 'a', ready: false, inSet: false, buffered: 0, stale: false);
test('avatars by hash, profile intent', () {
expect(RelayConstants.protoVersion, RelayConstants.protoMax);
const v = MemberView(id: 'm1', name: 'a', ready: false, inSet: false, buffered: 0);
expect(v.toJson().containsKey('avatarHash'), isFalse);
expect(v.toJson().containsKey('avatar'), isFalse);
const h = MemberView(
id: 'm1', name: 'a', avatarHash: 'abc', ready: false, inSet: false, buffered: 0, stale: false);
id: 'm1', name: 'a', avatarHash: 'abc', ready: false, inSet: false, buffered: 0);
expect(MemberView.fromJson(Map<String, Object?>.from(h.toJson())).avatarHash, 'abc');
// host: only on the hosting app's member, absent otherwise.
expect(v.toJson().containsKey('host'), isFalse);
const host = MemberView(id: 'm2', name: 'h', ready: false, inSet: false, buffered: 0, stale: false, host: true);
const host = MemberView(id: 'm2', name: 'h', ready: false, inSet: false, buffered: 0, host: true);
expect(host.toJson()['host'], isTrue);
expect(MemberView.fromJson(Map<String, Object?>.from(host.toJson())).host, isTrue);
expect(MemberView.fromJson(Map<String, Object?>.from(v.toJson())).host, isFalse);
packages/server/test/sync_wave_test.dart
+67 -33
View File
@@ -1,6 +1,3 @@
// restructure: rewrite in step 1 (Connection + Identity + Presence): the
// resume-token takeover becomes a lasting client id; ping/pong and the
// seek echo stay.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -47,12 +44,15 @@ class _Client {
final WebSocket ws;
final frames = <Map<String, Object?>>[];
final _welcome = Completer<Map<String, Object?>>();
final closed = Completer<int?>();
_Client(this.ws) {
ws.listen((m) {
final d = Map<String, Object?>.from(jsonDecode('$m') as Map);
frames.add(d);
if (d['t'] == 'welcome' && !_welcome.isCompleted) _welcome.complete(d);
}, onError: (_) {}, onDone: () {});
}, onError: (_) {}, onDone: () {
if (!closed.isCompleted) closed.complete(ws.closeCode);
});
}
StateBroadcast? get lastState {
@@ -65,18 +65,24 @@ class _Client {
void send(Map<String, Object?> j) => ws.add(jsonEncode(j));
static Future<(_Client, WelcomeMessage)> join(int port, String name,
{String resume = ''}) async {
{String client = '', int seq = 0}) async {
final c = await dial(port, name, client: client, seq: seq);
final w = await c._welcome.future.timeout(const Duration(seconds: 5));
return (c, WelcomeMessage.fromJson(w));
}
/// Sends the hello and returns at once (a refused one gets no welcome).
static Future<_Client> dial(int port, String name, {String client = '', int seq = 0}) async {
final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync');
final c = _Client(ws);
c.send(HelloMessage(
proto: RelayConstants.protoVersion,
secret: 'w',
user: HelloUser(name: name),
media: const HelloMedia(name: '', size: 0, duration: 0),
resume: resume,
).toJson());
final w = await c._welcome.future.timeout(const Duration(seconds: 5));
return (c, WelcomeMessage.fromJson(w));
proto: RelayConstants.protoVersion,
secret: 'w',
user: HelloUser(name: name),
client: client,
seq: seq)
.toJson());
return c;
}
}
@@ -98,42 +104,70 @@ void main() {
});
tearDown(() => server.stop());
test('reconnect after an unclean drop takes over the old member, no ghost',
() async {
const alice = 'aliceInstallIdAAAAAAAA';
test('an unclean drop: the same app takes its member over (id, token, ready), no ghost', () async {
final proxy = await _FreezableProxy.start(server.boundPort);
final other = (await _Client.join(server.boundPort, 'other')).$1;
final (first, welcome) = await _Client.join(proxy.port, 'alice');
final (first, welcome) = await _Client.join(proxy.port, 'alice', client: alice, seq: 1);
expect(welcome.resumed, isFalse);
expect(welcome.instance, server.hub.instance);
expect(welcome.proto, RelayConstants.protoMax);
expect(welcome.grace, RelayConstants.graceSecs);
first.send(const ReadyIntent(ready: true).toJson());
await _until(() => server.hub.engine.members[welcome.token]?.ready == true,
'ready');
await _until(() => server.hub.engine.members[welcome.token]?.ready == true, 'ready');
proxy.frozen = true; // unclean drop: the server's socket stays open
final (second, again) =
await _Client.join(server.boundPort, 'alice', resume: welcome.token);
final (second, again) = await _Client.join(server.boundPort, 'alice', client: alice, seq: 2);
expect(again.resumed, isTrue);
expect(again.memberId, welcome.memberId);
expect(again.token, welcome.token,
reason: 'same token keeps the media URL valid');
expect(server.hub.engine.members.length, 2,
reason: 'alice once (taken over) plus other, no ghost');
expect(server.hub.engine.members[welcome.token]!.ready, isTrue,
reason: 'ready survives the takeover');
// The frozen old socket closing later must not remove the new one.
expect(again.token, welcome.token, reason: 'same token keeps the media URL valid');
expect(server.hub.engine.members.length, 2, reason: 'alice once (taken over) plus other, no ghost');
expect(server.hub.engine.members[welcome.token]!.ready, isTrue, reason: 'ready survives the takeover');
// The frozen old socket closing later must not touch the member.
await proxy.close();
await Future.delayed(const Duration(seconds: 2));
expect(server.hub.engine.members.containsKey(welcome.token), isTrue);
expect(server.hub.engine.members[welcome.token]!.presence, isNot(Presence.gone));
expect(server.hub.sockets.containsKey(welcome.token), isTrue);
second.send(const ChatMessage(text: 'still here').toJson());
await _until(
() => other.frames.any((f) => f['t'] == 'chat' && f['text'] == 'still here'),
await _until(() => other.frames.any((f) => f['t'] == 'chat' && f['text'] == 'still here'),
'chat from the resumed member');
await other.ws.close();
await second.ws.close();
});
test('unknown resume token joins as a new member', () async {
final (a, wa) = await _Client.join(server.boundPort, 'a', resume: 'nope');
expect(wa.token, isNot('nope'));
test('a takeover tells the older socket: err replaced, then its close code', () async {
final (first, welcome) = await _Client.join(server.boundPort, 'alice', client: alice, seq: 1);
final (second, again) = await _Client.join(server.boundPort, 'alice', client: alice, seq: 1);
expect([again.resumed, again.memberId, again.token], [true, welcome.memberId, welcome.token],
reason: 'the same seq takes over too');
expect(await first.closed.future.timeout(const Duration(seconds: 5)), CloseCodes.replaced);
final err = first.frames.lastWhere((f) => f['t'] == 'err');
expect(err['code'], 'replaced');
// Frames from the old socket (had it stayed open) no longer count.
expect(server.hub.sockets[welcome.token], isNotNull);
await second.ws.close();
});
test("an older hello than the member's gets err stale and is closed", () async {
final (current, welcome) = await _Client.join(server.boundPort, 'alice', client: alice, seq: 5);
final older = await _Client.dial(server.boundPort, 'alice', client: alice, seq: 4);
expect(await older.closed.future.timeout(const Duration(seconds: 5)), CloseCodes.stale);
expect(older.frames.single['code'], 'stale');
expect(server.hub.sockets.containsKey(welcome.token), isTrue, reason: 'the current one stays');
expect(server.hub.engine.members, hasLength(1));
await current.ws.close();
});
test('without a client id, or with another, it is a new member', () async {
final (a, wa) = await _Client.join(server.boundPort, 'a');
final (b, wb) = await _Client.join(server.boundPort, 'a');
final (c, wc) = await _Client.join(server.boundPort, 'a', client: 'someOtherInstallId0001');
expect({wa.memberId, wb.memberId, wc.memberId}, hasLength(3));
expect([wa.resumed, wb.resumed, wc.resumed], [false, false, false]);
await a.ws.close();
await b.ws.close();
await c.ws.close();
});
test('ping gets a pong with the client stamp and the server clock', () async {