Skip to content

Commit c193cd4

Browse files
npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmpbrow
npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp
and
brow
committed
fix(mobile): recover stale relay sessions
Co-authored-by: Tom Brow <[email protected]> Signed-off-by: Tom Brow <[email protected]>
1 parent fc598f5 commit c193cd4

6 files changed

Lines changed: 260 additions & 8 deletions

File tree

‎mobile/lib/shared/relay/relay_session.dart‎

Lines changed: 18 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -78,17 +78,21 @@ class RelaySessionNotifier extends Notifier<SessionState> {
7878
RelaySessionNotifier({
7979
http.Client? httpClient,
8080
RelaySocketFactory socketFactory = RelaySocket.new,
81+
DateTime Function()? now,
8182
}) : _httpClient = httpClient,
82-
_socketFactory = socketFactory;
83+
_socketFactory = socketFactory,
84+
_now = now ?? DateTime.now;
8385

8486
final http.Client? _httpClient;
8587
final RelaySocketFactory _socketFactory;
88+
final DateTime Function() _now;
8689

8790
static const _baseReconnectDelayMs = 1000;
8891
static const _maxReconnectDelayMs = 30000;
8992
static const _eventBatchMs = 16;
9093
static const _reconnectReplaySkewSeconds = 5;
9194
static const _maxRecentDeliveryKeys = 5000;
95+
static const _backgroundGraceDuration = Duration(seconds: 5);
9296

9397
RelaySocket? _socket;
9498
final Map<String, _HistorySubscription> _historySubscriptions = {};
@@ -99,6 +103,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
99103
Timer? _reconnectTimer;
100104
Timer? _flushTimer;
101105
Timer? _backgroundGraceTimer;
106+
DateTime? _backgroundedAt;
102107
int _reconnectDelayMs = _baseReconnectDelayMs;
103108
int _subIdCounter = 0;
104109
bool _disposed = false;
@@ -315,8 +320,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
315320

316321
/// Called by the app lifecycle provider when the app goes to background.
317322
void onAppPaused() {
323+
_backgroundedAt = _now();
318324
_backgroundGraceTimer?.cancel();
319-
_backgroundGraceTimer = Timer(const Duration(seconds: 5), _pauseNow);
325+
_backgroundGraceTimer = Timer(_backgroundGraceDuration, _pauseNow);
320326
}
321327

322328
void _pauseNow() {
@@ -331,12 +337,18 @@ class RelaySessionNotifier extends Notifier<SessionState> {
331337
/// Called by the app lifecycle provider when the app returns to foreground.
332338
void onAppResumed() {
333339
_paused = false;
340+
final backgroundedAt = _backgroundedAt;
341+
_backgroundedAt = null;
334342
_backgroundGraceTimer?.cancel();
335343
_backgroundGraceTimer = null;
336344

337-
// If still connected, nothing to do — the socket survived the background
338-
// grace window.
339-
if (state.status == SessionStatus.connected) return;
345+
final backgroundedLongEnoughToRequireReconnect =
346+
backgroundedAt != null &&
347+
_now().difference(backgroundedAt) >= _backgroundGraceDuration;
348+
if (!backgroundedLongEnoughToRequireReconnect &&
349+
state.status == SessionStatus.connected) {
350+
return;
351+
}
340352

341353
// Cancel any in-flight reconnect backoff timer so we reconnect immediately
342354
// instead of waiting for the (possibly large) exponential delay.
@@ -650,6 +662,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
650662
_reconnectTimer?.cancel();
651663
_flushTimer?.cancel();
652664
_backgroundGraceTimer?.cancel();
665+
_backgroundedAt = null;
653666
_cancelAllHistory(null);
654667
_rejectAllPending(null);
655668
_recentDeliveryKeys.clear();

‎mobile/lib/shared/relay/relay_socket.dart‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import 'dart:convert';
33

44
import 'package:flutter/foundation.dart';
55
import 'package:nostr/nostr.dart' as nostr;
6+
import 'package:web_socket_channel/io.dart';
67
import 'package:web_socket_channel/web_socket_channel.dart';
78

89
import 'nostr_models.dart';
@@ -30,6 +31,12 @@ Exception classifyRelayAuthFailure(String message) {
3031
}
3132

3233
class RelaySocket {
34+
/// Interval for sending a ping and awaiting its pong before disconnecting.
35+
static const pingInterval = Duration(seconds: 30);
36+
37+
@visibleForTesting
38+
static Duration debugPingInterval = pingInterval;
39+
3340
final String _wsUrl;
3441
final String? _nsec;
3542
final void Function(List<dynamic> message) _onMessage;
@@ -63,7 +70,10 @@ class RelaySocket {
6370
_state = SocketState.connecting;
6471

6572
try {
66-
_channel = WebSocketChannel.connect(Uri.parse(_wsUrl));
73+
_channel = IOWebSocketChannel.connect(
74+
Uri.parse(_wsUrl),
75+
pingInterval: debugPingInterval,
76+
);
6777
await _channel!.ready;
6878
} catch (e) {
6979
_state = SocketState.disconnected;

‎mobile/pubspec.lock‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ packages:
274274
source: hosted
275275
version: "0.3.5+2"
276276
crypto:
277-
dependency: transitive
277+
dependency: "direct dev"
278278
description:
279279
name: crypto
280280
sha256: c8ea0233063ba03258fbcf2ca4d6dadfefe14f02fab57702265467a19f27fadf

‎mobile/pubspec.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ dev_dependencies:
4747
flutter_test:
4848
sdk: flutter
4949
flutter_lints: ^6.0.0
50+
crypto: ^3.0.7
5051
custom_lint: ^0.8.0
5152
riverpod_lint: ^3.1.0
5253
mocktail: ^1.0.4

‎mobile/test/shared/relay/relay_session_test.dart‎

Lines changed: 118 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,120 @@ void main() {
260260
expect(session.state.status, SessionStatus.disconnected);
261261
});
262262

263+
test(
264+
'resume reconnects a stale connected session after a long pause',
265+
() async {
266+
final sockets = <_ControlledRelaySocket>[];
267+
final keychain = nostr.Keys.generate();
268+
var now = DateTime(2026, 8, 2, 12);
269+
final session = RelaySessionNotifier(
270+
now: () => now,
271+
socketFactory:
272+
({
273+
required wsUrl,
274+
required nsec,
275+
required onMessage,
276+
required onConnected,
277+
required onDisconnected,
278+
}) {
279+
final socket = _ControlledRelaySocket(
280+
wsUrl: wsUrl,
281+
nsec: nsec,
282+
onMessage: onMessage,
283+
onConnected: onConnected,
284+
onDisconnected: onDisconnected,
285+
);
286+
sockets.add(socket);
287+
return socket;
288+
},
289+
);
290+
final container = ProviderContainer(
291+
overrides: [
292+
relaySessionProvider.overrideWith(() => session),
293+
relayConfigProvider.overrideWith(
294+
() => _FakeRelayConfigNotifier(
295+
baseUrl: 'https://relay.example',
296+
nsec: keychain.nsec,
297+
),
298+
),
299+
authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
300+
],
301+
);
302+
addTearDown(container.dispose);
303+
await container.read(authProvider.future);
304+
final subscription = container.listen(relaySessionProvider, (_, _) {});
305+
addTearDown(subscription.close);
306+
await Future<void>.delayed(Duration.zero);
307+
sockets.single.connectSuccessfully();
308+
309+
session.onAppPaused();
310+
now = now.add(const Duration(minutes: 5));
311+
session.onAppResumed();
312+
await Future<void>.delayed(Duration.zero);
313+
314+
expect(sockets, hasLength(2));
315+
expect(sockets.first.disposeCalls, 1);
316+
expect(session.state.status, SessionStatus.reconnecting);
317+
},
318+
);
319+
320+
test(
321+
'resume keeps a connected session within the background grace period',
322+
() async {
323+
final sockets = <_ControlledRelaySocket>[];
324+
final keychain = nostr.Keys.generate();
325+
var now = DateTime(2026, 8, 2, 12);
326+
final session = RelaySessionNotifier(
327+
now: () => now,
328+
socketFactory:
329+
({
330+
required wsUrl,
331+
required nsec,
332+
required onMessage,
333+
required onConnected,
334+
required onDisconnected,
335+
}) {
336+
final socket = _ControlledRelaySocket(
337+
wsUrl: wsUrl,
338+
nsec: nsec,
339+
onMessage: onMessage,
340+
onConnected: onConnected,
341+
onDisconnected: onDisconnected,
342+
);
343+
sockets.add(socket);
344+
return socket;
345+
},
346+
);
347+
final container = ProviderContainer(
348+
overrides: [
349+
relaySessionProvider.overrideWith(() => session),
350+
relayConfigProvider.overrideWith(
351+
() => _FakeRelayConfigNotifier(
352+
baseUrl: 'https://relay.example',
353+
nsec: keychain.nsec,
354+
),
355+
),
356+
authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
357+
],
358+
);
359+
addTearDown(container.dispose);
360+
await container.read(authProvider.future);
361+
final subscription = container.listen(relaySessionProvider, (_, _) {});
362+
addTearDown(subscription.close);
363+
await Future<void>.delayed(Duration.zero);
364+
sockets.single.connectSuccessfully();
365+
366+
session.onAppPaused();
367+
now = now.add(const Duration(seconds: 4));
368+
session.onAppResumed();
369+
await Future<void>.delayed(Duration.zero);
370+
371+
expect(sockets, hasLength(1));
372+
expect(sockets.single.disposeCalls, 0);
373+
expect(session.state.status, SessionStatus.connected);
374+
},
375+
);
376+
263377
test('delivers the same live event to each matching subscription', () async {
264378
final session = RelaySessionNotifier();
265379
final firstEvents = <NostrEvent>[];
@@ -402,6 +516,7 @@ class _AuthenticatedAuthNotifier extends AuthNotifier {
402516
class _ControlledRelaySocket extends RelaySocket {
403517
final void Function() _connected;
404518
final void Function(Object? error) _disconnected;
519+
int disposeCalls = 0;
405520

406521
_ControlledRelaySocket({
407522
required super.wsUrl,
@@ -416,7 +531,9 @@ class _ControlledRelaySocket extends RelaySocket {
416531
Future<void> connect() async {}
417532

418533
@override
419-
void dispose() {}
534+
void dispose() {
535+
disposeCalls++;
536+
}
420537

421538
void connectSuccessfully() => _connected();
422539

‎mobile/test/shared/relay/relay_socket_liveness_test.dart‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
import 'dart:async';
2+
import 'dart:convert';
3+
import 'dart:io';
4+
5+
import 'package:buzz/shared/relay/relay_socket.dart';
6+
import 'package:crypto/crypto.dart';
7+
import 'package:flutter_test/flutter_test.dart';
8+
9+
/// A server that completes the WS handshake then never speaks again: no pongs,
10+
/// no close frame. Only a client-side ping timeout can notice.
11+
Future<ServerSocket> _silentAfterHandshakeServer() async {
12+
final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
13+
server.listen((client) {
14+
client.listen(
15+
(data) {
16+
final match = RegExp(
17+
r'Sec-WebSocket-Key: (.*)\r\n',
18+
caseSensitive: false,
19+
).firstMatch(String.fromCharCodes(data));
20+
if (match == null) return;
21+
final accept = base64.encode(
22+
sha1
23+
.convert(
24+
utf8.encode(
25+
'${match.group(1)!.trim()}258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
26+
),
27+
)
28+
.bytes,
29+
);
30+
client.write(
31+
'HTTP/1.1 101 Switching Protocols\r\n'
32+
'Upgrade: websocket\r\nConnection: Upgrade\r\n'
33+
'Sec-WebSocket-Accept: $accept\r\n\r\n',
34+
);
35+
},
36+
onError: (_) {},
37+
onDone: () {},
38+
);
39+
});
40+
return server;
41+
}
42+
43+
void main() {
44+
const testPingInterval = Duration(milliseconds: 150);
45+
46+
setUp(() {
47+
RelaySocket.debugPingInterval = testPingInterval;
48+
});
49+
50+
tearDown(() {
51+
RelaySocket.debugPingInterval = RelaySocket.pingInterval;
52+
});
53+
54+
test('detects a peer that stops answering pings', () async {
55+
final server = await _silentAfterHandshakeServer();
56+
57+
final disconnected = Completer<Object?>();
58+
final socket = RelaySocket(
59+
wsUrl: 'ws://127.0.0.1:${server.port}',
60+
nsec: null,
61+
onMessage: (_) {},
62+
onConnected: () {},
63+
onDisconnected: (error) {
64+
if (!disconnected.isCompleted) disconnected.complete(error);
65+
},
66+
);
67+
unawaited(socket.connect());
68+
69+
var detected = true;
70+
try {
71+
await disconnected.future.timeout(testPingInterval * 4);
72+
} on TimeoutException {
73+
detected = false;
74+
}
75+
76+
expect(
77+
detected,
78+
isTrue,
79+
reason:
80+
'RelaySocket must surface an unanswered ping through onDisconnected',
81+
);
82+
83+
socket.dispose();
84+
await server.close();
85+
});
86+
87+
test('keeps an idle but healthy peer connected', () async {
88+
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
89+
server.transform(WebSocketTransformer()).listen((ws) {
90+
// A healthy relay answers pings without sending application data.
91+
ws.listen((_) {}, onError: (_) {}, onDone: () {});
92+
});
93+
94+
var tornDown = false;
95+
final socket = RelaySocket(
96+
wsUrl: 'ws://127.0.0.1:${server.port}',
97+
nsec: null,
98+
onMessage: (_) {},
99+
onConnected: () {},
100+
onDisconnected: (_) => tornDown = true,
101+
);
102+
unawaited(socket.connect());
103+
104+
await Future<void>.delayed(testPingInterval * 4);
105+
106+
expect(tornDown, isFalse);
107+
108+
await socket.disconnect();
109+
await server.close(force: true);
110+
});
111+
}

0 commit comments

Comments
 (0)