Skip to content

fix(mobile): probe socket liveness on resume #5118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aidenalmazan1228-wq wants to merge 1 commit into
base: main
Choose a base branch
from
+239 −1
Open

fix(mobile): probe socket liveness on resume #5118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion mobile/lib/shared/relay/relay_session.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,19 +93,22 @@ class RelaySessionNotifier extends Notifier<SessionState> {
RelayRateLimitGate? rateLimitGate,
RelayTimerFactory retryTimerFactory = Timer.new,
Future<void> Function(Duration) replayDelay = Future.delayed,
Duration resumeProbeTimeout = const Duration(seconds: 3),
}) : _httpClient = httpClient,
_socketFactory = socketFactory,
_now = now ?? DateTime.now,
_rateLimitGate = rateLimitGate ?? RelayRateLimitGate(),
_retryTimerFactory = retryTimerFactory,
_replayDelay = replayDelay;
_replayDelay = replayDelay,
_resumeProbeTimeout = resumeProbeTimeout;

final http.Client? _httpClient;
final RelaySocketFactory _socketFactory;
final DateTime Function() _now;
final RelayRateLimitGate _rateLimitGate;
final RelayTimerFactory _retryTimerFactory;
final Future<void> Function(Duration) _replayDelay;
final Duration _resumeProbeTimeout;

static const _baseReconnectDelayMs = 1000;
static const _maxReconnectDelayMs = 30000;
Expand Down Expand Up @@ -397,6 +400,25 @@ class RelaySessionNotifier extends Notifier<SessionState> {
await _connect(config);
}

/// Liveness probe after resume: a minimal REQ that resolves on EOSE. On
/// timeout or error the socket is presumed dead and we reconnect (live
/// subscriptions replay with the usual since-skew).
Future<void> _verifyConnectionAfterResume(int connectionGeneration) async {
try {
await fetchHistory(
const NostrFilter(kinds: [39000], limit: 1),
timeout: _resumeProbeTimeout,
);
} catch (_) {
if (_disposed || _paused) return;
if (connectionGeneration != _connectionGeneration || !_socketConnected) {
return;
}
if (state.status != SessionStatus.connected) return;
await reconnect();
}
}

/// Called by the app lifecycle provider when the app goes to background.
void onAppPaused() {
_backgroundedAt = _now();
Expand Down Expand Up @@ -427,6 +449,14 @@ class RelaySessionNotifier extends Notifier<SessionState> {
_now().difference(backgroundedAt) >= _backgroundGraceDuration;
if (!backgroundedLongEnoughToRequireReconnect &&
state.status == SessionStatus.connected) {
// Even after a short background stint the transport can be half-open:
// iOS may drop the network on screen lock or rebind NAT on a
// Wi-Fi/cellular switch without a close frame ever reaching us. If we
// trust the flag here, the session is a zombie — live subscriptions
// stay silent until the user force-kills the app. Verify with a cheap
// REQ/EOSE round-trip instead; any failure forces a reconnect, which
// replays live subscriptions.
unawaited(_verifyConnectionAfterResume(_connectionGeneration));
return;
}

Expand Down
208 changes: 208 additions & 0 deletions mobile/test/shared/relay/relay_session_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -546,6 +546,197 @@ void main() {
},
);

test(
'resume within the grace period probes the socket and keeps it when the '
'relay answers',
() async {
final sockets = <_ProbeRecordingRelaySocket>[];
final keychain = nostr.Keys.generate();
var now = DateTime(2026, 8, 2, 12);
final session = RelaySessionNotifier(
now: () => now,
socketFactory:
({
required wsUrl,
required nsec,
required onMessage,
required onConnected,
required onDisconnected,
}) {
final socket = _ProbeRecordingRelaySocket(
wsUrl: wsUrl,
nsec: nsec,
onMessage: onMessage,
onConnected: onConnected,
onDisconnected: onDisconnected,
);
sockets.add(socket);
return socket;
},
);
final container = ProviderContainer(
overrides: [
relaySessionProvider.overrideWith(() => session),
relayConfigProvider.overrideWith(
() => _FakeRelayConfigNotifier(
baseUrl: 'https://relay.example',
nsec: keychain.nsec,
),
),
authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
],
);
addTearDown(container.dispose);
await container.read(authProvider.future);
final subscription = container.listen(relaySessionProvider, (_, _) {});
addTearDown(subscription.close);
await Future<void>.delayed(Duration.zero);
sockets.single.connectSuccessfully();

session.onAppPaused();
now = now.add(const Duration(seconds: 4));
session.onAppResumed();
await Future<void>.delayed(Duration.zero);

// The resume path issued a liveness REQ; answer it with EOSE.
final probeReq = sockets.single.sentFrames.lastWhere(
(frame) => frame.isNotEmpty && frame.first == 'REQ',
);
session.debugHandleMessage(['EOSE', probeReq[1]]);
await Future<void>.delayed(Duration.zero);

expect(sockets, hasLength(1));
expect(sockets.single.disposeCalls, 0);
expect(session.state.status, SessionStatus.connected);
},
);

test(
'resume within the grace period reconnects when the socket is half-open '
'(probe never answered)',
() async {
final sockets = <_ProbeRecordingRelaySocket>[];
final keychain = nostr.Keys.generate();
var now = DateTime(2026, 8, 2, 12);
final session = RelaySessionNotifier(
now: () => now,
resumeProbeTimeout: const Duration(milliseconds: 20),
socketFactory:
({
required wsUrl,
required nsec,
required onMessage,
required onConnected,
required onDisconnected,
}) {
final socket = _ProbeRecordingRelaySocket(
wsUrl: wsUrl,
nsec: nsec,
onMessage: onMessage,
onConnected: onConnected,
onDisconnected: onDisconnected,
);
sockets.add(socket);
return socket;
},
);
final container = ProviderContainer(
overrides: [
relaySessionProvider.overrideWith(() => session),
relayConfigProvider.overrideWith(
() => _FakeRelayConfigNotifier(
baseUrl: 'https://relay.example',
nsec: keychain.nsec,
),
),
authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
],
);
addTearDown(container.dispose);
await container.read(authProvider.future);
final subscription = container.listen(relaySessionProvider, (_, _) {});
addTearDown(subscription.close);
await Future<void>.delayed(Duration.zero);
sockets.single.connectSuccessfully();

session.onAppPaused();
now = now.add(const Duration(seconds: 4));
session.onAppResumed();

// Probe times out against the dead transport, forcing a reconnect.
await Future<void>.delayed(const Duration(milliseconds: 60));

expect(sockets, hasLength(2));
expect(sockets.first.disposeCalls, 1);
expect(session.state.status, SessionStatus.reconnecting);
},
);

test(
'stale resume probe cannot reconnect a replacement connection',
() async {
final sockets = <_ProbeRecordingRelaySocket>[];
final keychain = nostr.Keys.generate();
var now = DateTime(2026, 8, 2, 12);
final session = RelaySessionNotifier(
now: () => now,
resumeProbeTimeout: const Duration(milliseconds: 20),
socketFactory:
({
required wsUrl,
required nsec,
required onMessage,
required onConnected,
required onDisconnected,
}) {
final socket = _ProbeRecordingRelaySocket(
wsUrl: wsUrl,
nsec: nsec,
onMessage: onMessage,
onConnected: onConnected,
onDisconnected: onDisconnected,
);
sockets.add(socket);
return socket;
},
);
final container = ProviderContainer(
overrides: [
relaySessionProvider.overrideWith(() => session),
relayConfigProvider.overrideWith(
() => _FakeRelayConfigNotifier(
baseUrl: 'https://relay.example',
nsec: keychain.nsec,
),
),
authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
],
);
addTearDown(container.dispose);
await container.read(authProvider.future);
final subscription = container.listen(relaySessionProvider, (_, _) {});
addTearDown(subscription.close);
await Future<void>.delayed(Duration.zero);
sockets.single.connectSuccessfully();

session.onAppPaused();
now = now.add(const Duration(seconds: 4));
session.onAppResumed();
await Future<void>.delayed(Duration.zero);

// Replace the connection before the old probe times out.
await session.reconnect();
expect(sockets, hasLength(2));
sockets.last.connectSuccessfully();

await Future<void>.delayed(const Duration(milliseconds: 60));

expect(sockets, hasLength(2));
expect(sockets.last.disposeCalls, 0);
expect(session.state.status, SessionStatus.connected);
},
);

test('delivers the same live event to each matching subscription', () async {
final session = RelaySessionNotifier();
final firstEvents = <NostrEvent>[];
Expand Down Expand Up @@ -1281,6 +1472,23 @@ class _ControlledRelaySocket extends RelaySocket {
void disconnectWith(Object? error) => _disconnected(error);
}

class _ProbeRecordingRelaySocket extends _ControlledRelaySocket {
final List<List<dynamic>> sentFrames = [];

_ProbeRecordingRelaySocket({
required super.wsUrl,
required super.nsec,
required super.onMessage,
required super.onConnected,
required super.onDisconnected,
});

@override
void send(List<dynamic> payload) {
sentFrames.add(payload);
}
}

const _channelId = '11111111-1111-4111-8111-111111111111';

class _FakeRelayConfigNotifier extends RelayConfigNotifier {
Expand Down