diff --git a/mobile/lib/shared/relay/relay.dart b/mobile/lib/shared/relay/relay.dart index dd153fc928..c2490968b8 100644 --- a/mobile/lib/shared/relay/relay.dart +++ b/mobile/lib/shared/relay/relay.dart @@ -9,6 +9,7 @@ export 'relay_closed_policy.dart'; export 'relay_client.dart'; export 'relay_provider.dart'; export 'relay_rate_limit_gate.dart'; +export 'relay_reconnect_policy.dart'; export 'relay_session.dart'; export 'relay_socket.dart'; export 'signed_event_relay.dart'; diff --git a/mobile/lib/shared/relay/relay_reconnect_policy.dart b/mobile/lib/shared/relay/relay_reconnect_policy.dart new file mode 100644 index 0000000000..3d33b3d20b --- /dev/null +++ b/mobile/lib/shared/relay/relay_reconnect_policy.dart @@ -0,0 +1,94 @@ +import 'dart:math'; + +/// Reconnect delay before the first retry, in milliseconds. +const relayReconnectBaseDelayMs = 1000; + +/// Ceiling for a reconnect delay, in milliseconds. +const relayReconnectMaxDelayMs = 30000; + +/// How long a connection must stay up before a drop stops counting as part of +/// a bounce loop and the backoff returns to [relayReconnectBaseDelayMs]. Matches +/// desktop's `BACKOFF_RESET_STABLE_MS`. +const relayStableConnectionDuration = Duration(seconds: 60); + +/// Width of the uniform window the first reconnect after a server restart +/// (close code 1012) or shutdown (1001) is spread over. The relay already +/// spreads those closes over 10s; this spreads the arrivals too. +const relayServiceRestartSpreadMs = 10000; + +/// Seconds a replay reaches behind the newest delivered event. +const relayReplaySkewSeconds = 5; + +/// Seconds a replay reaches behind the moment a subscription's connection +/// closed: 900s for the relay's `MAX_TIMESTAMP_DRIFT_SECS` +/// (`crates/buzz-relay/src/handlers/ingest.rs`), plus 60s of client clock +/// skew. The relay stores events with a `created_at` up to 900s behind its own +/// clock, so one stored just after the close can be backdated that far; a +/// narrower window would miss it for a quiet subscription. +const relayLiveWindowSkewSeconds = 960; + +/// Whether [closeCode] says the relay is restarting or going away. +bool isRelayServiceRestartClose(int? closeCode) => + closeCode == 1012 || closeCode == 1001; + +/// The delay before the next reconnect attempt. +/// +/// [backoffMs] is the current exponential backoff. The delay is jittered +/// symmetrically by +/-25% and capped at [relayReconnectMaxDelayMs]. After a +/// service-restart close the first attempt (backoff still at the base) is +/// instead uniform over [0, [relayServiceRestartSpreadMs]]; a client already +/// backing off waits the longer of the two so a bounce loop cannot spin. +/// +/// [jitterUnit] and [spreadUnit] are uniform draws in [0, 1). +Duration relayReconnectDelay({ + required int backoffMs, + required int? closeCode, + required double jitterUnit, + required double spreadUnit, +}) { + final jittered = min( + (backoffMs * (0.75 + jitterUnit * 0.5)).round(), + relayReconnectMaxDelayMs, + ); + if (!isRelayServiceRestartClose(closeCode)) { + return Duration(milliseconds: jittered); + } + final spread = (spreadUnit * relayServiceRestartSpreadMs).round(); + return Duration( + milliseconds: backoffMs > relayReconnectBaseDelayMs + ? max(jittered, spread) + : spread, + ); +} + +/// Whether a connection that stayed up for [uptime] was stable. +bool isRelayConnectionStable(Duration uptime) => + uptime >= relayStableConnectionDuration; + +/// The `since` a live subscription replays with after a reconnect, or null to +/// replay the original filter unchanged. +/// +/// [lastSeenCreatedAt] is the newest event delivered. [liveUntil] is the +/// client clock, in seconds, of the last relay frame received on the +/// subscription's last connection (capped at the close), set only if that +/// subscription reached EOSE on it: everything stored was delivered by EOSE +/// and everything after it arrived live until that frame. +/// The result is the latest of the original `since`, +/// `lastSeenCreatedAt - 5s` and `liveUntil - 960s`, never past [until]. +int? relayResumeSince({ + required int? originalSince, + required int? until, + required int? lastSeenCreatedAt, + required int? liveUntil, +}) { + if (lastSeenCreatedAt == null && liveUntil == null) return originalSince; + var since = originalSince ?? 0; + if (lastSeenCreatedAt != null) { + since = max(since, lastSeenCreatedAt - relayReplaySkewSeconds); + } + if (liveUntil != null) { + since = max(since, liveUntil - relayLiveWindowSkewSeconds); + } + if (until != null) since = min(since, until); + return max(0, since); +} diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 7390274b61..cdefed9768 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -21,6 +21,7 @@ import 'relay_closed_policy.dart'; import 'relay_http_query_client.dart'; import 'relay_provider.dart'; import 'relay_rate_limit_gate.dart'; +import 'relay_reconnect_policy.dart'; import 'relay_session_types.dart'; import 'relay_socket.dart'; import 'package:buzz/shared/crypto/event_signing.dart'; @@ -28,56 +29,7 @@ import 'package:buzz/shared/crypto/event_signing.dart'; export 'relay_session_types.dart'; part 'relay_session_auth.dart'; - -class _HistorySubscription { - final List events = []; - final Completer> completer; - final Timer timeout; - - /// The filter of a request held until the socket authenticates. - NostrFilter? heldFilter; - - _HistorySubscription({required this.completer, required this.timeout}); -} - -class _LiveSubscription { - final NostrFilter filter; - final void Function(NostrEvent) onEvent; - final void Function(String message)? onClosed; - final void Function(RelaySubscriptionStatus status)? onStatusChanged; - Completer? readyCompleter; - int? lastSeenCreatedAt; - int closedRetryAttempt = 0; - Timer? closedRetryTimer; - - _LiveSubscription({ - required this.filter, - required this.onEvent, - this.onClosed, - this.onStatusChanged, - this.readyCompleter, - }); -} - -class _ClosedRetry { - final _LiveSubscription subscription; - final int generation; - _ClosedRetry({required this.subscription, required this.generation}); -} - -class _PendingEvent { - final Completer completer; - final Timer timeout; - - _PendingEvent({required this.completer, required this.timeout}); -} - -class _BufferedEvent { - final String subId; - final NostrEvent event; - - _BufferedEvent(this.subId, this.event); -} +part 'relay_session_subscriptions.dart'; class RelaySessionNotifier extends Notifier { RelaySessionNotifier({ @@ -88,6 +40,8 @@ class RelaySessionNotifier extends Notifier { RelayRateLimitGate? rateLimitGate, RelayTimerFactory retryTimerFactory = Timer.new, Future Function(Duration) replayDelay = Future.delayed, + RelayTimerFactory reconnectTimerFactory = Timer.new, + Random? random, }) : _httpQueryClient = RelayHttpQueryClient( client: httpClient, clientFactory: httpClientFactory, @@ -96,7 +50,9 @@ class RelaySessionNotifier extends Notifier { _now = now ?? DateTime.now, _rateLimitGate = rateLimitGate ?? RelayRateLimitGate(), _retryTimerFactory = retryTimerFactory, - _replayDelay = replayDelay; + _replayDelay = replayDelay, + _reconnectTimerFactory = reconnectTimerFactory, + _random = random ?? Random(); final RelayHttpQueryClient _httpQueryClient; final RelaySocketFactory _socketFactory; @@ -104,11 +60,12 @@ class RelaySessionNotifier extends Notifier { final RelayRateLimitGate _rateLimitGate; final RelayTimerFactory _retryTimerFactory; final Future Function(Duration) _replayDelay; + final RelayTimerFactory _reconnectTimerFactory; + final Random _random; - static const _baseReconnectDelayMs = 1000; - static const _maxReconnectDelayMs = 30000; + static const _baseReconnectDelayMs = relayReconnectBaseDelayMs; + static const _maxReconnectDelayMs = relayReconnectMaxDelayMs; static const _eventBatchMs = 16; - static const _reconnectReplaySkewSeconds = 5; static const _replayBatchSize = 8; static const _replayInterBatchDelay = Duration(milliseconds: 50); static const _maxRecentDeliveryKeys = 5000; @@ -125,6 +82,8 @@ class RelaySessionNotifier extends Notifier { Timer? _flushTimer; Timer? _backgroundGraceTimer; DateTime? _backgroundedAt; + DateTime? _connectedAt; + DateTime? _lastInboundAt; int _reconnectDelayMs = _baseReconnectDelayMs; int _subIdCounter = 0; bool _disposed = false; @@ -388,17 +347,18 @@ class RelaySessionNotifier extends Notifier { final subId = _nextSubId('l'); final readyCompleter = Completer(); - _liveSubscriptions[subId] = _LiveSubscription( + final subscription = _LiveSubscription( filter: filter, onEvent: onEvent, onClosed: onClosed, onStatusChanged: onStatusChanged, readyCompleter: readyCompleter, ); + _liveSubscriptions[subId] = subscription; // Before AUTH the relay closes it as auth-required, which is terminal. // The replay on connect sends it instead. - if (_socketConnected) _sendReq(subId, filter); + if (_socketConnected) _sendLiveReq(subId, subscription); try { await readyCompleter.future.timeout( @@ -585,6 +545,9 @@ class RelaySessionNotifier extends Notifier { Future _connect(RelayConfig config, {bool renewed = false}) async { if (_disposed || _ageRestricted) return; + // Whatever connection came before ends here, capped at its last frame. + _markLiveSubscriptionsClosed(); + _connectedAt = null; final generation = ++_connectionGeneration; // A replacement socket is unauthenticated until its AUTH completes. _socketConnected = false; @@ -619,7 +582,7 @@ class RelaySessionNotifier extends Notifier { } error = AccountFailure.denied; } - _handleDisconnected(generation, error); + _handleDisconnected(generation, error, closeCode: socket.closeCode); }, ); _socket = socket; @@ -639,14 +602,23 @@ class RelaySessionNotifier extends Notifier { _socketConnected = true; ref.read(hostedRelayAuthProvider).connected(ref.read(relayConfigProvider)); _hasConnectedOnce = true; - _reconnectDelayMs = _baseReconnectDelayMs; + // The backoff is not reset here: a client bounced repeatedly must keep + // backing off. [_handleDisconnected] resets it after a stable connection. + _connectedAt = _now(); + _lastInboundAt = _connectedAt; state = const SessionState(status: SessionStatus.connected); await _replayLiveSubscriptions(generation); } - void _handleDisconnected(int generation, Object? error) { + void _handleDisconnected(int generation, Object? error, {int? closeCode}) { if (_disposed || generation != _connectionGeneration) return; _socketConnected = false; + final connectedAt = _connectedAt; + _connectedAt = null; + if (connectedAt != null && + isRelayConnectionStable(_now().difference(connectedAt))) { + _reconnectDelayMs = _baseReconnectDelayMs; + } _cancelAllHistory(error); _rejectAllPending(error); _resetAllClosedRetries(); @@ -659,10 +631,10 @@ class RelaySessionNotifier extends Notifier { state = const SessionState(status: SessionStatus.disconnected); return; } - _scheduleReconnect(); + _scheduleReconnect(closeCode); } - void _scheduleReconnect() { + void _scheduleReconnect(int? closeCode) { if (_disposed || _paused) return; final attempt = state.reconnectAttempt + 1; state = SessionState( @@ -671,13 +643,37 @@ class RelaySessionNotifier extends Notifier { ); _reconnectTimer?.cancel(); - _reconnectTimer = Timer(Duration(milliseconds: _reconnectDelayMs), () { + final delay = relayReconnectDelay( + backoffMs: _reconnectDelayMs, + closeCode: closeCode, + jitterUnit: _random.nextDouble(), + spreadUnit: _random.nextDouble(), + ); + _reconnectTimer = _reconnectTimerFactory(delay, () { _reconnectDelayMs = min(_reconnectDelayMs * 2, _maxReconnectDelayMs); final config = ref.read(relayConfigProvider); _connect(config); }); } + /// Records when each subscription that was live on the closing connection + /// stopped being live, so the next replay can resume from there. + /// + /// Liveness ends at the last frame received, not at the close: after a + /// device sleep the loss is only noticed on wake, and nothing arrived in + /// between. + void _markLiveSubscriptionsClosed() { + final now = _now(); + final inbound = _lastInboundAt; + final end = inbound == null || inbound.isAfter(now) ? now : inbound; + final closedAt = end.millisecondsSinceEpoch ~/ 1000; + for (final subscription in _liveSubscriptions.values) { + if (!subscription.liveOnConnection) continue; + subscription.liveOnConnection = false; + subscription.liveUntil = closedAt; + } + } + /// Replay all live subscriptions after a reconnect, with a time skew to /// catch events that occurred during the disconnect. Future _replayLiveSubscriptions(int generation) async { @@ -739,7 +735,7 @@ class RelaySessionNotifier extends Notifier { } _pendingClosedRetries.remove(entry.key); } - _sendReq(entry.key, _replayFilter(entry.value)); + _sendLiveReq(entry.key, entry.value); } if (i + _replayBatchSize < entries.length) { await _replayDelay(_replayInterBatchDelay); @@ -751,15 +747,20 @@ class RelaySessionNotifier extends Notifier { !_disposed && generation == _connectionGeneration; NostrFilter _replayFilter(_LiveSubscription subscription) { - final since = subscription.lastSeenCreatedAt; - return since == null - ? subscription.filter - : subscription.filter.copyWithSince( - max(0, since - _reconnectReplaySkewSeconds), - ); + final filter = subscription.filter; + final since = relayResumeSince( + originalSince: filter.since, + until: filter.until, + lastSeenCreatedAt: subscription.lastSeenCreatedAt, + liveUntil: subscription.liveUntil, + ); + return since == null || since == filter.since + ? filter + : filter.copyWithSince(since); } void _handleMessage(List data) { + _lastInboundAt = _now(); if (data.isEmpty) return; final type = data[0] as String; @@ -823,6 +824,10 @@ class RelaySessionNotifier extends Notifier { // already been released. final liveSub = _liveSubscriptions[subId]; if (liveSub != null) { + // A second REQ on this socket makes its EOSE ambiguous: it may belong + // to the REQ it replaced, so it cannot vouch for the replacement. + liveSub.liveOnConnection = + liveSub.ambiguousGeneration != _connectionGeneration; _resetClosedRetry(liveSub); _flushBufferedEventsNow(); liveSub.onStatusChanged?.call(RelaySubscriptionStatus.ready); @@ -858,6 +863,8 @@ class RelaySessionNotifier extends Notifier { final liveSub = _liveSubscriptions[subId]; if (liveSub == null) return; final readyCompleter = liveSub.readyCompleter; + // The stream is interrupted until the retry's EOSE. + liveSub.liveOnConnection = false; // A deadline is terminal for one-shot history, but a live subscription // retries it: same filter under backoff, which can keep retrying at the // cap. `since` is not advanced, so events missed while closed aren't @@ -1034,6 +1041,14 @@ class RelaySessionNotifier extends Notifier { return '$prefix-$_subIdCounter'; } + void _sendLiveReq(String subId, _LiveSubscription subscription) { + if (subscription.reqGeneration == _connectionGeneration) { + subscription.ambiguousGeneration = _connectionGeneration; + } + subscription.reqGeneration = _connectionGeneration; + _sendReq(subId, _replayFilter(subscription)); + } + void _sendReq(String subId, NostrFilter filter) { _socket?.send(['REQ', subId, filter.toJson()]); } diff --git a/mobile/lib/shared/relay/relay_session_subscriptions.dart b/mobile/lib/shared/relay/relay_session_subscriptions.dart new file mode 100644 index 0000000000..9fcd0c135f --- /dev/null +++ b/mobile/lib/shared/relay/relay_session_subscriptions.dart @@ -0,0 +1,64 @@ +part of 'relay_session.dart'; + +class _HistorySubscription { + final List events = []; + final Completer> completer; + final Timer timeout; + + /// The filter of a request held until the socket authenticates. + NostrFilter? heldFilter; + + _HistorySubscription({required this.completer, required this.timeout}); +} + +class _LiveSubscription { + final NostrFilter filter; + final void Function(NostrEvent) onEvent; + final void Function(String message)? onClosed; + final void Function(RelaySubscriptionStatus status)? onStatusChanged; + Completer? readyCompleter; + int? lastSeenCreatedAt; + + /// Whether EOSE arrived on the current connection and no CLOSED has + /// interrupted the stream since. + bool liveOnConnection = false; + + /// Client clock (seconds) when the last connection this subscription was + /// live on closed. Kept across later connections that never reach EOSE. + int? liveUntil; + + /// The connection generation of this subscription's latest REQ, and the + /// one on which it sent a second REQ, leaving its EOSE ambiguous. + int? reqGeneration; + int? ambiguousGeneration; + int closedRetryAttempt = 0; + Timer? closedRetryTimer; + + _LiveSubscription({ + required this.filter, + required this.onEvent, + this.onClosed, + this.onStatusChanged, + this.readyCompleter, + }); +} + +class _ClosedRetry { + final _LiveSubscription subscription; + final int generation; + _ClosedRetry({required this.subscription, required this.generation}); +} + +class _PendingEvent { + final Completer completer; + final Timer timeout; + + _PendingEvent({required this.completer, required this.timeout}); +} + +class _BufferedEvent { + final String subId; + final NostrEvent event; + + _BufferedEvent(this.subId, this.event); +} diff --git a/mobile/lib/shared/relay/relay_socket.dart b/mobile/lib/shared/relay/relay_socket.dart index c72c36fba0..61329e8afd 100644 --- a/mobile/lib/shared/relay/relay_socket.dart +++ b/mobile/lib/shared/relay/relay_socket.dart @@ -52,9 +52,14 @@ class RelaySocket { Completer? _authCompleter; Timer? _authTimeout; String? _pendingAuthEventId; + int? _closeCode; SocketState get state => _state; + /// The WebSocket close code the relay sent, once it has closed the + /// connection; null if the connection failed without one. + int? get closeCode => _closeCode; + /// Scoped assertion supplied by the connection owner before connecting. Map headers = const {}; @@ -108,6 +113,7 @@ class RelaySocket { _onDisconnected(error); }, onDone: () { + _closeCode = _channel?.closeCode; _failAuth(null); _resetConnection(); _onDisconnected(null); diff --git a/mobile/test/helpers/fake_relay.dart b/mobile/test/helpers/fake_relay.dart index eb44c4e954..5b67559961 100644 --- a/mobile/test/helpers/fake_relay.dart +++ b/mobile/test/helpers/fake_relay.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:math'; import 'package:buzz/shared/auth/auth_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; @@ -50,6 +51,7 @@ class FakeRelay { socketFactory: _openSocket, httpClient: MockClient(_handleHttp), now: _now, + random: _MidpointRandom(), ); List get overrides => [ @@ -434,6 +436,19 @@ int _newestFirst(NostrEvent a, NostrEvent b) => a.createdAt != b.createdAt ? b.createdAt.compareTo(a.createdAt) : a.id.compareTo(b.id); +/// Always draws the midpoint, so reconnect jitter is exactly the backoff and +/// tests can elapse a known delay. +class _MidpointRandom implements Random { + @override + double nextDouble() => 0.5; + + @override + bool nextBool() => false; + + @override + int nextInt(int max) => max ~/ 2; +} + class _FixedRelayConfig extends RelayConfigNotifier { _FixedRelayConfig(this._nsec); diff --git a/mobile/test/shared/relay/relay_reconnect_resume_test.dart b/mobile/test/shared/relay/relay_reconnect_resume_test.dart new file mode 100644 index 0000000000..62bc8f3944 --- /dev/null +++ b/mobile/test/shared/relay/relay_reconnect_resume_test.dart @@ -0,0 +1,554 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:math'; + +import 'package:buzz/shared/auth/auth_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:crypto/crypto.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; + +void main() { + group('relayResumeSince', () { + // (original since, until, last seen, live until) -> resume since. + final cases = <(int?, int?, int?, int?, int?)>[ + (null, null, null, null, null), // nothing known: original filter + (500, null, null, null, 500), // original since kept + (null, null, 1000, null, 995), // last seen - 5s + (null, null, null, 1000, 40), // live until - 960s + (null, null, 1000, 1010, 995), // last seen newer than live window + (null, null, 1000, 2100, 1140), // live window newer than last seen + (2000, null, 1000, 2100, 2000), // original since is a floor + (null, null, 3, null, 0), // clamped at zero + (null, null, null, 30, 0), + (null, 900, 1000, 2100, 900), // never past the filter's until + // An event stored right after the close can be backdated 900s (the + // relay's timestamp drift bound); the window must reach at least that + // far behind the close. + (null, null, null, 5000, 4040), + ]; + test('the live window covers the relay timestamp drift bound', () { + const driftBound = 900; + expect( + relayResumeSince( + originalSince: null, + until: null, + lastSeenCreatedAt: null, + liveUntil: 5000, + )!, + lessThanOrEqualTo(5000 - driftBound), + ); + }); + + for (final (original, until, lastSeen, liveUntil, expected) in cases) { + test('since=$original until=$until seen=$lastSeen live=$liveUntil', () { + expect( + relayResumeSince( + originalSince: original, + until: until, + lastSeenCreatedAt: lastSeen, + liveUntil: liveUntil, + ), + expected, + ); + }); + } + }); + + group('relayReconnectDelay', () { + // (backoff, close code, jitter unit, spread unit) -> delay in ms. + final cases = <(int, int?, double, double, int)>[ + (1000, null, 0.0, 0.9, 750), // symmetric jitter, low edge + (1000, null, 0.5, 0.9, 1000), + (1000, null, 0.999, 0.9, 1250), // high edge + (30000, null, 0.999, 0.9, 30000), // capped + (1000, 1006, 0.5, 0.9, 1000), // abnormal close is not a restart + (1000, 1012, 0.5, 0.0, 0), // restart: first attempt uniform [0, 10s] + (1000, 1012, 0.5, 0.35, 3500), + (1000, 1001, 0.5, 0.999, 9990), + (4000, 1012, 0.5, 0.1, 4000), // already backing off: not below backoff + (2000, 1012, 0.5, 0.9, 9000), // ... but still spread over the window + ]; + for (final (backoff, code, jitter, spread, expected) in cases) { + test('backoff=$backoff code=$code jitter=$jitter spread=$spread', () { + expect( + relayReconnectDelay( + backoffMs: backoff, + closeCode: code, + jitterUnit: jitter, + spreadUnit: spread, + ).inMilliseconds, + expected, + ); + }); + } + + test('a connection is stable from 60 seconds', () { + expect(isRelayConnectionStable(const Duration(seconds: 59)), isFalse); + expect(isRelayConnectionStable(const Duration(seconds: 60)), isTrue); + }); + }); + + test('RelaySocket surfaces the relay close code', () async { + // dart:io's server refuses to send 1012, so write the close frame by hand. + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + server.listen((client) { + client.listen((data) { + final key = RegExp( + r'Sec-WebSocket-Key: (.*)\r\n', + caseSensitive: false, + ).firstMatch(String.fromCharCodes(data)); + if (key == null) return; + final accept = base64.encode( + sha1 + .convert( + utf8.encode( + '${key.group(1)!.trim()}258EAFA5-E914-47DA-95CA-C5AB0DC85B11', + ), + ) + .bytes, + ); + client.write( + 'HTTP/1.1 101 Switching Protocols\r\n' + 'Upgrade: websocket\r\nConnection: Upgrade\r\n' + 'Sec-WebSocket-Accept: $accept\r\n\r\n', + ); + client.add([0x88, 2, 1012 >> 8, 1012 & 0xff]); + }, onError: (_) {}); + }); + final closeCode = Completer(); + late final RelaySocket socket; + socket = RelaySocket( + wsUrl: 'ws://127.0.0.1:${server.port}', + nsec: null, + onMessage: (_) {}, + onConnected: () {}, + onDisconnected: (_) { + if (!closeCode.isCompleted) closeCode.complete(socket.closeCode); + }, + ); + unawaited(socket.connect()); + expect(await closeCode.future.timeout(const Duration(seconds: 5)), 1012); + socket.dispose(); + await server.close(); + }); + + group('reconnect through RelaySessionNotifier', () { + test('a bounced client keeps backing off until a connection is stable, ' + 'and a restart close spreads the first attempt', () async { + final h = await _Harness.start( + random: _ScriptedRandom([ + 0.5, 0.3, // 1012: first attempt, uniform 3s + 0.5, 0.1, // 1012 again within 60s: backoff 2s beats spread 1s + 0.0, 0.0, // plain close after a stable connection: base - 25% + 0.5, 0.0, // plain close after a short connection: doubled backoff + ]), + ); + h.sockets.last.connectSuccessfully(); + h.now = h.now.add(const Duration(seconds: 5)); + h.sockets.last.disconnectWith(null, closeCode: 1012); + expect(h.reconnectDelays.last, const Duration(seconds: 3)); + + h.timers.last.fire(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + h.now = h.now.add(const Duration(seconds: 5)); + h.sockets.last.disconnectWith(null, closeCode: 1012); + expect(h.reconnectDelays.last, const Duration(seconds: 2)); + + h.timers.last.fire(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + h.now = h.now.add(const Duration(seconds: 60)); + h.sockets.last.disconnectWith(null); + expect(h.reconnectDelays.last, const Duration(milliseconds: 750)); + + h.timers.last.fire(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + h.now = h.now.add(const Duration(seconds: 59)); + h.sockets.last.disconnectWith(null); + // Backoff was doubled to 2s when the previous attempt fired and the + // connection did not stay up long enough to reset it. + expect(h.reconnectDelays.last, const Duration(seconds: 2)); + }); + }); + + group('resume cursor through RelaySessionNotifier', () { + test('replays from the last frame received before the loss, not the ' + 'original filter or the moment the loss is noticed, and only for ' + 'subscriptions that were live', () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + + await h.subscribeLive(_filter('quiet', since: start - 3600)); + h.subscribeWithoutEose(_filter('no-eose')); + final busy = await h.subscribeLive(_filter('busy')); + final closed = await h.subscribeLive(_filter('closed')); + h.receive(['CLOSED', closed, 'error: transient']); + h.now = h.now.add(const Duration(seconds: 120)); + h.receive(['EVENT', busy, _eventJson('e1', start + 100, 'busy')]); + + // The device sleeps for 2000s; the loss is only noticed on wake. + h.now = h.now.add(const Duration(seconds: 2000)); + h.sockets.last.disconnectWith(null); + h.timers.last.fire(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + await h.settle(); + + // A backdated event stored just after the last frame (up to 900s behind + // the relay clock) is still inside the window. + expect(h.replayedSince()['quiet']!, lessThanOrEqualTo(start + 120 - 900)); + expect(h.replayedSince(), { + 'quiet': start - 840, // last frame (start + 120) - 960s + 'no-eose': null, // never live: today's behaviour + 'busy': start + 95, // last seen - 5s beats last frame - 960s + 'closed': null, // relay closed it: not live at the disconnect + }); + }); + + test( + 'a background disconnect resumes from the last frame before it', + () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + await h.subscribeLive(_filter('quiet')); + + h.session.onAppPaused(); + h.now = h.now.add(const Duration(seconds: 20)); + h.receive(['NOTICE', 'still here']); + h.now = h.now.add(const Duration(seconds: 10)); + h.session.debugPauseNow(); + h.now = h.now.add(const Duration(minutes: 10)); + h.session.onAppResumed(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + await h.settle(); + + expect(h.replayedSince(), {'quiet': start + 20 - 960}); + }, + ); + + test('a later connection that never reaches EOSE keeps the older cursor, ' + 'then advances once live again', () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + await h.subscribeLive(_filter('quiet')); + + await h.dropAfter(const Duration(seconds: 100)); + expect(h.replayedSince(), {'quiet': start + 100 - 960}); + + // The second connection dies before EOSE: the cursor must not move. + await h.dropAfter(const Duration(seconds: 100)); + expect(h.replayedSince(), {'quiet': start + 100 - 960}); + + h.receive(['EOSE', 'l-1']); + await h.dropAfter(const Duration(seconds: 100)); + expect(h.replayedSince(), {'quiet': start + 300 - 960}); + }); + + // A close or error that was buffered while the device slept is processed + // on wake. It is the transport talking, not a frame from the relay, so it + // must not count as the last time the relay was heard from. + final losses = <(String, int?, Object?)>[ + ('close 1012', 1012, null), + ('close 1006', 1006, null), + ('socket error', null, Exception('connection reset')), + ]; + for (final (label, closeCode, error) in losses) { + test( + 'a $label noticed after an hour asleep does not extend liveness', + () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + await h.subscribeLive(_filter('quiet')); + + h.now = h.now.add(const Duration(hours: 1)); + h.sockets.last.disconnectWith(error, closeCode: closeCode); + h.timers.last.fire(); + await h.settle(); + h.sockets.last.connectSuccessfully(); + await h.settle(); + + expect(h.replayedSince(), {'quiet': start - 960}); + }, + ); + } + + test('a delayed EOSE for an unsubscribed id cannot mark its replacement ' + 'live', () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + final old = await h.subscribeLive(_filter('quiet', since: start - 3600)); + h.unsubscribers[old]!(); + + // The channel is subscribed again with an older floor. The old REQ's + // EOSE was still in flight when its CLOSE went out and now arrives. + final replacement = h.subscribeWithoutEose( + _filter('quiet', since: start - 7200), + ); + expect(replacement, isNot(old)); + h.receive(['EOSE', old]); + + // The socket drops before the replacement's history arrives. + await h.dropAfter(const Duration(seconds: 100)); + expect(h.replayedSince(), {'quiet': start - 7200}); + }); + + test('a second REQ on one socket cannot advance the cursor: its EOSE may ' + 'belong to the REQ it replaced', () async { + final h = await _Harness.start(); + final start = h.nowSeconds; + h.sockets.last.connectSuccessfully(); + await h.subscribeLive(_filter('quiet')); + await h.dropAfter(const Duration(seconds: 100)); + expect(h.replayedSince(), {'quiet': start + 100 - 960}); + + // On the new socket: EOSE, then CLOSED, then a same-socket retry REQ + // whose EOSE arrives. The retry's history has not been confirmed. + h.receive(['EOSE', 'l-1']); + h.receive(['CLOSED', 'l-1', 'error: transient']); + h.retryTimers.last.fire(); + await h.settle(); + expect(h.replayedSince(), {'quiet': start + 100 - 960}); + h.receive(['EOSE', 'l-1']); + + await h.dropAfter(const Duration(seconds: 500)); + expect(h.replayedSince(), {'quiet': start + 100 - 960}); + }); + }); +} + +NostrFilter _filter(String name, {int? since}) => NostrFilter( + kinds: const [9], + tags: { + '#h': [name], + }, + limit: 0, + since: since, +); + +Map _eventJson(String id, int createdAt, String channel) => { + 'id': id, + 'pubkey': 'alice', + 'created_at': createdAt, + 'kind': 9, + 'tags': [ + ['h', channel], + ], + 'content': 'hello', + 'sig': 'sig', +}; + +class _Harness { + _Harness._(this.session, this.now); + + final RelaySessionNotifier session; + DateTime now; + final sockets = <_FakeSocket>[]; + final timers = <_ManualTimer>[]; + final retryTimers = <_ManualTimer>[]; + final reconnectDelays = []; + final subscriptionNames = {}; + final unsubscribers = {}; + + int get nowSeconds => now.millisecondsSinceEpoch ~/ 1000; + + static Future<_Harness> start({Random? random}) async { + final keys = nostr.Keys.generate(); + late final _Harness h; + final session = RelaySessionNotifier( + now: () => h.now, + random: random, + retryTimerFactory: (duration, callback) { + final timer = _ManualTimer(callback); + h.retryTimers.add(timer); + return timer; + }, + reconnectTimerFactory: (duration, callback) { + final timer = _ManualTimer(callback); + h.timers.add(timer); + h.reconnectDelays.add(duration); + return timer; + }, + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) { + final socket = _FakeSocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + ); + h.sockets.add(socket); + return socket; + }, + ); + h = _Harness._(session, DateTime.utc(2026, 10, 28, 12)); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + relayConfigProvider.overrideWith( + () => _FakeRelayConfigNotifier(nsec: keys.nsec), + ), + authProvider.overrideWith(() => _AuthenticatedAuthNotifier()), + ], + ); + addTearDown(container.dispose); + await container.read(authProvider.future); + final listener = container.listen(relaySessionProvider, (_, _) {}); + addTearDown(listener.close); + await h.settle(); + expect(h.sockets, hasLength(1)); + return h; + } + + Future settle() => Future.delayed(Duration.zero); + + /// Delivers [frame] from the relay over the latest socket. + void receive(List frame) => sockets.last.receive(frame); + + /// Lets [elapsed] pass with a last frame at its end, then drops the socket + /// and reconnects, as a reconnect timer firing and AUTH completing would. + Future dropAfter(Duration elapsed) async { + now = now.add(elapsed); + receive(['NOTICE', 'ping']); + sockets.last.disconnectWith(null); + timers.last.fire(); + await settle(); + sockets.last.connectSuccessfully(); + await settle(); + } + + /// Subscribes and delivers EOSE, returning the subscription id. + Future subscribeLive(NostrFilter filter) async { + final subId = 'l-${subscriptionNames.length + 1}'; + subscriptionNames[subId] = filter.tags['#h']!.single; + final future = session.subscribe(filter, (_) {}); + receive(['EOSE', subId]); + unsubscribers[subId] = await future; + return subId; + } + + /// Subscribes without EOSE, returning the subscription id. + String subscribeWithoutEose(NostrFilter filter) { + final subId = 'l-${subscriptionNames.length + 1}'; + subscriptionNames[subId] = filter.tags['#h']!.single; + unawaited(session.subscribe(filter, (_) {})); + return subId; + } + + /// The `since` of each subscription's REQ on the latest socket, by channel. + Map replayedSince() => { + for (final message in sockets.last.sent.where((m) => m.first == 'REQ')) + subscriptionNames[message[1] as String]!: + (message[2] as Map)['since'] as int?, + }; +} + +class _FakeSocket extends RelaySocket { + _FakeSocket({ + required super.wsUrl, + required super.nsec, + required super.onMessage, + required super.onConnected, + required super.onDisconnected, + }) : _connected = onConnected, + _disconnected = onDisconnected, + _message = onMessage; + + final void Function() _connected; + final void Function(Object? error) _disconnected; + final void Function(List message) _message; + final sent = >[]; + int? _fakeCloseCode; + + @override + int? get closeCode => _fakeCloseCode; + + @override + Future connect() async {} + + @override + void send(List payload) => sent.add(payload); + + @override + void dispose() {} + + void connectSuccessfully() => _connected(); + + void receive(List frame) => _message(frame); + + void disconnectWith(Object? error, {int? closeCode}) { + _fakeCloseCode = closeCode; + _disconnected(error); + } +} + +class _ManualTimer implements Timer { + _ManualTimer(this._callback); + + final void Function() _callback; + bool _active = true; + + void fire() { + if (!_active) return; + _active = false; + _callback(); + } + + @override + void cancel() => _active = false; + + @override + bool get isActive => _active; + + @override + int get tick => _active ? 0 : 1; +} + +/// A [Random] whose doubles come from a script. +class _ScriptedRandom implements Random { + _ScriptedRandom(List doubles) : _doubles = List.of(doubles); + + final List _doubles; + + @override + double nextDouble() => _doubles.removeAt(0); + + @override + bool nextBool() => throw UnimplementedError(); + + @override + int nextInt(int max) => throw UnimplementedError(); +} + +class _FakeRelayConfigNotifier extends RelayConfigNotifier { + _FakeRelayConfigNotifier({required String? nsec}) : _nsec = nsec; + + final String? _nsec; + + @override + RelayConfig build() => + RelayConfig(baseUrl: 'https://relay.example', nsec: _nsec); +} + +class _AuthenticatedAuthNotifier extends AuthNotifier { + @override + Future build() async => + const AuthState(status: AuthStatus.authenticated); +}