External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): recover stale relay sessions (#4372)

Browse Source 
### Summary

Fixes [this
issue](buzz://message?channel=e62570dd-33ad-42c5-b92b-75f2689f9694&id=b726c366abfe62429ee3cdcd34d0c0fb98c33c3ea053480585bed71745412b56):
> I often don’t see my bot responses until after I post. they’re usually
time stamped correctly so I think it’s just a refresh issue?

### What changed?

Buzz Mobile now reconnects relay sessions after the app has remained
backgrounded beyond the existing 5-second grace period, even when the
session still reports a stale `connected` state. This makes resume
recovery independent of whether iOS runs the grace timer before or after
delivering `resumed`.

Reconnection is now based on elapsed background time rather than a
direct socket-health probe.
- If the app was backgrounded for at least the 5-second grace period,
the socket is presumed dead and the session reconnects regardless of
reported status.
- If it was backgrounded for less than that, a reported `connected`
status is still trusted.

In the sub-5-second window the socket is either genuinely alive, which
is the common case for a momentary background, or it is dead and the
client ping detects it within the two-interval worst case described
below. That is now a degraded-latency path, not a silent-forever path.

The mobile relay socket now uses `IOWebSocketChannel.connect` with a
30-second `pingInterval`. An unanswered ping closes the Dart socket
through the existing disconnect and reconnect path.

Detection takes up to two ping intervals, so about 60 seconds worst
case, not 30. One interval of idleness elapses and a ping is sent, then
a second interval elapses with no pong and the socket closes. Any
inbound pong restarts the first stage, so the clock measures idleness
rather than running on a fixed cadence.

### Why?

Buzz iOS can sometimes stop showing new bot or agent responses after a
phone has been locked for 5 to 10 minutes. When the user later posts a
message, the missing responses can appear all at once. iOS may suspend
Buzz before the short delayed cleanup that would normally close its
connection has a chance to run. Before this change, Buzz trusted the
resulting stale healthy status on resume and skipped reconnecting, so
the missing responses stayed hidden until a later post exposed the dead
connection.

A state-machine test with a stubbed connection reproduced this reported
pattern and showed that it matches this failure mode: the failed post
triggered a reconnect that fetched the missing messages. The same test
also checked the other candidate explanation, the bug tracked in
[#3053](https://github.com/block/buzz/pull/3053), where the relay has
closed the app's subscription. That state does not produce the pattern.
Posting succeeds and the user's own message appears, but nothing looks
for the missed messages, so they stay hidden. The test confirmed that
the missed messages were still available to fetch in that state, so the
missing step was a trigger to fetch them. This was not an end-to-end
reproduction on an iOS device or a live relay.

The new resume check covers the normal lock and unlock path. If the app
was backgrounded for less than the 5-second grace period, it still
trusts a connection marked as healthy. A dead connection in that window
is instead detected by the ping check, which can take up to about 60
seconds but prevents the app from remaining silently stuck. The ping
only runs while iOS is running the app, so it does not detect a
connection that died during suspension; the resume check owns the lock
and unlock path.

A pre-existing path also runs the same resume handling when network
connectivity returns while the app is already in the foreground. Because
the app was not backgrounded, this change does not alter that path,
which still trusts a connection marked as healthy and relies on the
slower ping check.

Recovery from a subscription that the relay explicitly closes remains in
[#3053](https://github.com/block/buzz/pull/3053), and the two changes
overlap in one file. Changes to how missed messages are backfilled or
replayed are out of scope.

### How is it tested?

Full mobile suite at base and head. Both runs have the same known
macOS-host-only failure in `ChannelDetailPage keeps follow mode off
while a tall newest message stays visible` at line 1053:

- Base: 1,021 passed, 1 skipped, 1 failed
- Head: 1,025 passed, 1 skipped, 1 failed

Added tests:

-
[`relay_session_test.dart`](https://github.com/block/buzz/tree/main/mobile/test/shared/relay/relay_session_test.dart):
long-background resume reconnect and within-grace control
-
[`relay_socket_liveness_test.dart`](https://github.com/block/buzz/tree/main/mobile/test/shared/relay/relay_socket_liveness_test.dart):
silent-peer disconnect and idle-but-healthy control

Mutation checks confirm that removing elapsed-background resume recovery
fails with one socket instead of two, and removing `pingInterval` leaves
the silent peer connected. Restored production code passes both
mutations' regression tests and the healthy idle control.

Signed-off-by: Tom Brow <tomb@block.xyz>
Co-authored-by: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz>
This commit is contained in:
Tom Brow
2026-08-03 12:28:50 -07:00
committed by GitHub
co-authored by npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp
parent 651f637275
commit ce56e34411
6 changed files with 259 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/lib/shared/relay/relay_session.dart
+17 -4
View File
@@ -89,17 +89,20 @@ class RelaySessionNotifier extends Notifier<SessionState> {
RelaySessionNotifier({
http.Client? httpClient,
RelaySocketFactory socketFactory = RelaySocket.new,
DateTime Function()? now,
RelayRateLimitGate? rateLimitGate,
RelayTimerFactory retryTimerFactory = Timer.new,
Future<void> Function(Duration) replayDelay = Future.delayed,
}) : _httpClient = httpClient,
_socketFactory = socketFactory,
_now = now ?? DateTime.now,
_rateLimitGate = rateLimitGate ?? RelayRateLimitGate(),
_retryTimerFactory = retryTimerFactory,
_replayDelay = replayDelay;
final http.Client? _httpClient;
final RelaySocketFactory _socketFactory;
final DateTime Function() _now;
final RelayRateLimitGate _rateLimitGate;
final RelayTimerFactory _retryTimerFactory;
final Future<void> Function(Duration) _replayDelay;
@@ -111,6 +114,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
static const _replayBatchSize = 8;
static const _replayInterBatchDelay = Duration(milliseconds: 50);
static const _maxRecentDeliveryKeys = 5000;
static const _backgroundGraceDuration = Duration(seconds: 5);
RelaySocket? _socket;
final Map<String, _HistorySubscription> _historySubscriptions = {};
@@ -122,6 +126,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
Timer? _reconnectTimer;
Timer? _flushTimer;
Timer? _backgroundGraceTimer;
DateTime? _backgroundedAt;
int _reconnectDelayMs = _baseReconnectDelayMs;
int _subIdCounter = 0;
bool _disposed = false;
@@ -394,8 +399,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
/// Called by the app lifecycle provider when the app goes to background.
void onAppPaused() {
_backgroundedAt = _now();
_backgroundGraceTimer?.cancel();
_backgroundGraceTimer = Timer(const Duration(seconds: 5), _pauseNow);
_backgroundGraceTimer = Timer(_backgroundGraceDuration, _pauseNow);
}
void _pauseNow() {
@@ -411,12 +417,18 @@ class RelaySessionNotifier extends Notifier<SessionState> {
/// Called by the app lifecycle provider when the app returns to foreground.
void onAppResumed() {
_paused = false;
final backgroundedAt = _backgroundedAt;
_backgroundedAt = null;
_backgroundGraceTimer?.cancel();
_backgroundGraceTimer = null;
// If still connected, nothing to do — the socket survived the background
// grace window.
if (state.status == SessionStatus.connected) return;
final backgroundedLongEnoughToRequireReconnect =
backgroundedAt != null &&
_now().difference(backgroundedAt) >= _backgroundGraceDuration;
if (!backgroundedLongEnoughToRequireReconnect &&
state.status == SessionStatus.connected) {
return;
}
// Cancel any in-flight reconnect backoff timer so we reconnect immediately
// instead of waiting for the (possibly large) exponential delay.
@@ -908,6 +920,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
_reconnectTimer?.cancel();
_flushTimer?.cancel();
_backgroundGraceTimer?.cancel();
_backgroundedAt = null;
_cancelAllClosedRetries();
_rateLimitGate.reset();
_visibleChannelsByOwner.clear();
mobile/lib/shared/relay/relay_socket.dart
+11 -1
View File
@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:flutter/foundation.dart';
import 'package:nostr/nostr.dart' as nostr;
import 'package:web_socket_channel/io.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'nostr_models.dart';
@@ -30,6 +31,12 @@ Exception classifyRelayAuthFailure(String message) {
}
class RelaySocket {
/// Interval for sending a ping and awaiting its pong before disconnecting.
static const pingInterval = Duration(seconds: 30);
@visibleForTesting
static Duration debugPingInterval = pingInterval;
final String _wsUrl;
final String? _nsec;
final void Function(List<dynamic> message) _onMessage;
@@ -63,7 +70,10 @@ class RelaySocket {
_state = SocketState.connecting;
try {
_channel = WebSocketChannel.connect(Uri.parse(_wsUrl));
_channel = IOWebSocketChannel.connect(
Uri.parse(_wsUrl),
pingInterval: debugPingInterval,
);
await _channel!.ready;
} catch (e) {
_state = SocketState.disconnected;
mobile/pubspec.lock
+1 -1
View File
@@ -274,7 +274,7 @@ packages:
source: hosted
version: "0.3.5+2"
crypto:
dependency: transitive
dependency: "direct dev"
description:
name: crypto
sha256: c8ea0233063ba03258fbcf2ca4d6dadfefe14f02fab57702265467a19f27fadf
mobile/pubspec.yaml
+1
View File
@@ -47,6 +47,7 @@ dev_dependencies:
flutter_test:
sdk: flutter
flutter_lints: ^6.0.0
crypto: ^3.0.7
custom_lint: ^0.8.0
riverpod_lint: ^3.1.0
mocktail: ^1.0.4
mobile/test/shared/relay/relay_session_test.dart
+118 -1
View File
@@ -432,6 +432,120 @@ void main() {
expect(session.state.status, SessionStatus.disconnected);
});
test(
'resume reconnects a stale connected session after a long pause',
() async {
final sockets = <_ControlledRelaySocket>[];
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 = _ControlledRelaySocket(
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(minutes: 5));
session.onAppResumed();
await Future<void>.delayed(Duration.zero);
expect(sockets, hasLength(2));
expect(sockets.first.disposeCalls, 1);
expect(session.state.status, SessionStatus.reconnecting);
},
);
test(
'resume keeps a connected session within the background grace period',
() async {
final sockets = <_ControlledRelaySocket>[];
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 = _ControlledRelaySocket(
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);
expect(sockets, hasLength(1));
expect(sockets.single.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>[];
@@ -1143,6 +1257,7 @@ class _AuthenticatedAuthNotifier extends AuthNotifier {
class _ControlledRelaySocket extends RelaySocket {
final void Function() _connected;
final void Function(Object? error) _disconnected;
int disposeCalls = 0;
_ControlledRelaySocket({
required super.wsUrl,
@@ -1157,7 +1272,9 @@ class _ControlledRelaySocket extends RelaySocket {
Future<void> connect() async {}
@override
void dispose() {}
void dispose() {
disposeCalls++;
}
void connectSuccessfully() => _connected();
mobile/test/shared/relay/relay_socket_liveness_test.dart
+111
View File
@@ -0,0 +1,111 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:buzz/shared/relay/relay_socket.dart';
import 'package:crypto/crypto.dart';
import 'package:flutter_test/flutter_test.dart';
/// A server that completes the WS handshake then never speaks again: no pongs,
/// no close frame. Only a client-side ping timeout can notice.
Future<ServerSocket> _silentAfterHandshakeServer() async {
final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
server.listen((client) {
client.listen(
(data) {
final match = RegExp(
r'Sec-WebSocket-Key: (.*)\r\n',
caseSensitive: false,
).firstMatch(String.fromCharCodes(data));
if (match == null) return;
final accept = base64.encode(
sha1
.convert(
utf8.encode(
'${match.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',
);
},
onError: (_) {},
onDone: () {},
);
});
return server;
}
void main() {
const testPingInterval = Duration(milliseconds: 150);
setUp(() {
RelaySocket.debugPingInterval = testPingInterval;
});
tearDown(() {
RelaySocket.debugPingInterval = RelaySocket.pingInterval;
});
test('detects a peer that stops answering pings', () async {
final server = await _silentAfterHandshakeServer();
final disconnected = Completer<Object?>();
final socket = RelaySocket(
wsUrl: 'ws://127.0.0.1:${server.port}',
nsec: null,
onMessage: (_) {},
onConnected: () {},
onDisconnected: (error) {
if (!disconnected.isCompleted) disconnected.complete(error);
},
);
unawaited(socket.connect());
var detected = true;
try {
await disconnected.future.timeout(testPingInterval * 4);
} on TimeoutException {
detected = false;
}
expect(
detected,
isTrue,
reason:
'RelaySocket must surface an unanswered ping through onDisconnected',
);
socket.dispose();
await server.close();
});
test('keeps an idle but healthy peer connected', () async {
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
server.transform(WebSocketTransformer()).listen((ws) {
// A healthy relay answers pings without sending application data.
ws.listen((_) {}, onError: (_) {}, onDone: () {});
});
var tornDown = false;
final socket = RelaySocket(
wsUrl: 'ws://127.0.0.1:${server.port}',
nsec: null,
onMessage: (_) {},
onConnected: () {},
onDisconnected: (_) => tornDown = true,
);
unawaited(socket.connect());
await Future<void>.delayed(testPingInterval * 4);
expect(tornDown, isFalse);
await socket.disconnect();
await server.close(force: true);
});
}