jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-05 00:41:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): resume reconnect replay from the last live point and spread reconnects (#8659)

Browse Source 
## Summary

A relay rolling deploy closes every socket with 1012. On reconnect,
mobile replayed each live subscription's **original filter** (startup
`since`, full `limit`) unless that subscription had received an event,
so quiet subscriptions re-read history back to app start every time.
Reconnects used a fixed 1s delay with no jitter. The delay also reset to
1s on every successful connect, so a client bounced from pod to pod
during a deploy kept coming back at the same moment as every other
client.

**This PR makes two changes:**

1. **Resume cursor.** If a subscription reached EOSE on a connection, it
records `liveUntil`: the time of the last frame received from the relay
on that connection, capped at the clock at close. The replay then uses
`since = max(original since, lastSeen − 5s, liveUntil − 960s)`, and
never goes past the filter's `until`.
- **Why 960s:** the relay accepts `created_at` up to ±900s from its
clock (`MAX_TIMESTAMP_DRIFT_SECS` in
`crates/buzz-relay/src/handlers/ingest.rs`). An event stored after the
close can therefore be backdated by that much, and the extra 60s covers
client clock skew.
- **Only relay frames count as liveness.** Close and error paths never
extend `liveUntil`, so a close noticed after the device sleeps doesn't
overstate how long the subscription was live.
- **Ambiguous EOSE is ignored.** If a subscription has a second REQ on
the same socket, that subscription doesn't advance `liveUntil` for the
rest of the socket. A CLOSED ends liveness for the rest of that socket,
because the retry is a second REQ. Subscription IDs are never reused, so
an old EOSE can't mark a re-subscribe as live.
- **No cursor:** a subscription that never reached EOSE keeps its
previous cursor, or replays as before.
2. **Reconnect spreading.**
- After a 1012 or 1001 close, the first attempt waits a uniform random
delay in [0, 10s]. A client already backing off waits the longer of that
delay and its backoff.
   - Ordinary backoff gets symmetric ±25% jitter, capped at 30s.
- The backoff resets only after a connection stays up for 60s, not on
connect. This matches desktop.
- Foreground resume and a manual `reconnect()` still reset to the base
delay, as before.

**Simpler?** The pure reconnect math (cursor, jitter, spread, stability)
lives in a new `relay_reconnect_policy.dart` and is table-tested on its
own. `relay_session.dart` was near its 1200-line cap, so its five small
private bookkeeping classes moved verbatim to
`relay_session_subscriptions.dart`. Net concept count is unchanged.

**Identified debt (not changed here):**
- Dart's WebSocket ping/pong isn't visible to the session, so a
long-idle but healthy connection that drops replays back to its last
application frame − 960s. This is still bounded, and much cheaper than
replaying from app start.
- The unread catch-up in `channels_provider.dart` runs one REQ per
active channel (`since = readAt + 1`, `limit: 1000`) on reconnect and
every 60s.
- A retryable CLOSED is retried with a 30s cap, no jitter and no attempt
limit.
- `lastSeenCreatedAt` advances on stored events received before EOSE. If
a relay returns stored events newest-first and the socket dies before
EOSE, the next replay can skip older stored events.
- The `lastSeen − 5s` term keeps its existing gap for events backdated
after a close.

### Related issue
N/A. This is part of a set of reconnect-load reductions. Companion PRs:
relay read-pool warm (#8651), desktop (#8652), agent harness (#8653).

### Testing
- Mobile gate on a Linux workstation, at the head SHA:
  - `dart format --set-exit-if-changed`: pass.
  - `flutter analyze`: no issues.
  - Gateway recipe tests: pass.
- `flutter test`: 3881 pass, 1 fail.
`push_gateway_origin_validator_test` fails the same way on `main` on
that host, due to a host git-config error.
- New `relay_reconnect_resume_test.dart`:
  - pure table tests for the cursor and the delay
  - a real-socket test that reads the 1012 close code
- journeys through `RelaySessionNotifier`. They check that a close or
error after an hour asleep doesn't extend liveness, that a second REQ or
a delayed EOSE for an unsubscribed ID can't advance the cursor, and that
a backdated event inside the 960s window is still replayed.
- 22 production mutations, each failing the new tests. They cover every
cursor term, the 960s window, the `until` clamp, the liveness cap, the
ambiguous-EOSE guard, CLOSED handling, jitter, the 1012 spread and
floor, the backoff reset, and close-code capture.
- `fake_relay.dart` now draws a fixed jitter midpoint, so existing
journey tests stay deterministic.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Luke Tornquist
2026-10-28 20:35:12 -04:00
committed by GitHub
parent bad9d26142
commit 974495fd4a
7 changed files with 818 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
mobile/lib/shared/relay/relay.dart
View File

@@ -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';

94
mobile/lib/shared/relay/relay_reconnect_policy.dart Normal file
View File

@@ -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);
}

153
mobile/lib/shared/relay/relay_session.dart
View File

@@ -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<NostrEvent> events = [];
final Completer<List<NostrEvent>> 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<void>? 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<NostrEvent> 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<SessionState> {
RelaySessionNotifier({
@@ -88,6 +40,8 @@ class RelaySessionNotifier extends Notifier<SessionState> {
RelayRateLimitGate? rateLimitGate,
RelayTimerFactory retryTimerFactory = Timer.new,
Future<void> 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<SessionState> {
_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<SessionState> {
final RelayRateLimitGate _rateLimitGate;
final RelayTimerFactory _retryTimerFactory;
final Future<void> 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<SessionState> {
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<SessionState> {
final subId = _nextSubId('l');
final readyCompleter = Completer<void>();
_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<SessionState> {
Future<void> _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<SessionState> {
}
error = AccountFailure.denied;
}
_handleDisconnected(generation, error);
_handleDisconnected(generation, error, closeCode: socket.closeCode);
},
);
_socket = socket;
@@ -639,14 +602,23 @@ class RelaySessionNotifier extends Notifier<SessionState> {
_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<SessionState> {
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<SessionState> {
);
_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<void> _replayLiveSubscriptions(int generation) async {
@@ -739,7 +735,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
}
_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<SessionState> {
!_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<dynamic> data) {
_lastInboundAt = _now();
if (data.isEmpty) return;
final type = data[0] as String;
@@ -823,6 +824,10 @@ class RelaySessionNotifier extends Notifier<SessionState> {
// 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<SessionState> {
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<SessionState> {
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()]);
}

64
mobile/lib/shared/relay/relay_session_subscriptions.dart Normal file
View File

@@ -0,0 +1,64 @@
part of 'relay_session.dart';
class _HistorySubscription {
final List<NostrEvent> events = [];
final Completer<List<NostrEvent>> 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<void>? 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<NostrEvent> completer;
final Timer timeout;
_PendingEvent({required this.completer, required this.timeout});
}
class _BufferedEvent {
final String subId;
final NostrEvent event;
_BufferedEvent(this.subId, this.event);
}

6
mobile/lib/shared/relay/relay_socket.dart
View File

@@ -52,9 +52,14 @@ class RelaySocket {
Completer<void>? _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<String, String> headers = const {};
@@ -108,6 +113,7 @@ class RelaySocket {
_onDisconnected(error);
},
onDone: () {
_closeCode = _channel?.closeCode;
_failAuth(null);
_resetConnection();
_onDisconnected(null);

15
mobile/test/helpers/fake_relay.dart
View File

@@ -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<Override> 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);

554
mobile/test/shared/relay/relay_reconnect_resume_test.dart Normal file
View File

@@ -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<int?>();
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<String, dynamic> _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 = <Duration>[];
final subscriptionNames = <String, String>{};
final unsubscribers = <String, void Function()>{};
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<void> settle() => Future<void>.delayed(Duration.zero);
/// Delivers [frame] from the relay over the latest socket.
void receive(List<dynamic> 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<void> 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<String> 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<String, int?> replayedSince() => {
for (final message in sockets.last.sent.where((m) => m.first == 'REQ'))
subscriptionNames[message[1] as String]!:
(message[2] as Map<String, dynamic>)['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<dynamic> message) _message;
final sent = <List<dynamic>>[];
int? _fakeCloseCode;
@override
int? get closeCode => _fakeCloseCode;
@override
Future<void> connect() async {}
@override
void send(List<dynamic> payload) => sent.add(payload);
@override
void dispose() {}
void connectSuccessfully() => _connected();
void receive(List<dynamic> 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<double> doubles) : _doubles = List.of(doubles);
final List<double> _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<AuthState> build() async =>
const AuthState(status: AuthStatus.authenticated);
}