diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index d5834d3aef..68cd67ca2a 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -31,6 +31,8 @@ class PresenceCacheNotifier extends Notifier> { // a community's presence heartbeats arrive many times a second. final Set _subscribedKeys = {}; void Function()? _presenceUnsub; + // Closes the replacement being opened, until it is promoted or abandoned. + void Function()? _pendingClose; Timer? _poll; Timer? _queued; Timer? _resubscribe; @@ -70,6 +72,8 @@ class PresenceCacheNotifier extends Notifier> { _generation++; _presenceUnsub?.call(); _presenceUnsub = null; + _pendingClose?.call(); + _pendingClose = null; _poll?.cancel(); _queued?.cancel(); _queued = null; @@ -81,7 +85,10 @@ class PresenceCacheNotifier extends Notifier> { _poll = Timer.periodic(const Duration(seconds: 60), (_) { if (_presenceUnsub == null || !_subscribedKeys.containsAll(_tracked)) { unawaited(_subscribe(generation)); - } else if (!_querying) { + } + // The snapshot backstop runs for a live stream even while a widening + // keeps failing, so expired leases are still corrected. + if (_presenceUnsub != null && !_querying) { _pending.addAll(_tracked); _schedule(generation); } @@ -145,20 +152,69 @@ class PresenceCacheNotifier extends Notifier> { (_presenceUnsub != null && _subscribedKeys.containsAll(_tracked))) { return; } + // One subscription opens at a time. A replacement holds this until it is + // promoted or abandoned, so tracking bursts widen it in later rounds. _opening = true; _subscribeAttempted = true; final keys = _tracked.toList()..sort(); final subscription = ++_nextSubscription; + // A replacement becomes active only on its own first ready: returning from + // subscribeWithStatus also happens on a retryable CLOSED or a timeout. + // Until then the active subscription and the cache stay as they are. + final replacing = _presenceUnsub != null; + var pending = replacing; var closed = false; - // Status reported before this subscription replaces the active one. + // Status the first subscription reports before it becomes active. var retrying = false; - // The subscription being replaced keeps delivering until this one opens. + void Function()? unsub; + void close() { + if (closed) return; + closed = true; + unsub?.call(); // otherwise released when subscribeWithStatus returns + } + bool current() => generation == _generation && !closed && subscription == _activeSubscription; + void promote() { + pending = false; + _pendingClose = null; + _opening = false; + final previous = _presenceUnsub; + final added = keys.where((key) => !_subscribedKeys.contains(key)); + _pending.addAll(added); + _activeSubscription = subscription; + _retrying = retrying; + _presenceUnsub = close; + _subscribedKeys + ..clear() + ..addAll(keys); + previous?.call(); + // Still backing off: its ready reads everything once it is live. + if (_retrying) { + _invalidateSnapshot(); + return; + } + // Keys tracked while this one opened need a wider subscription. + if (!_subscribedKeys.containsAll(_tracked)) _queueResubscribe(generation); + // Newly covered keys are read only now, so no change between an earlier + // snapshot and their live coverage is lost. + _schedule(generation); + } + + void abandon() { + pending = false; + _pendingClose = null; + _opening = false; + close(); + // Keys only it would cover get a snapshot; the poll retries the widening. + _schedule(generation); + } + + if (pending) _pendingClose = close; try { - final unsub = await ref + unsub = await ref .read(relaySessionProvider.notifier) .subscribeWithStatus( NostrFilter( @@ -171,6 +227,10 @@ class PresenceCacheNotifier extends Notifier> { }, onStatusChanged: (status) { if (generation != _generation || closed) return; + if (pending) { + status == RelaySubscriptionStatus.ready ? promote() : abandon(); + return; + } if (subscription != _activeSubscription) { retrying = status == RelaySubscriptionStatus.retrying; return; @@ -181,9 +241,12 @@ class PresenceCacheNotifier extends Notifier> { }, onClosed: (_) { if (generation != _generation || closed) return; + if (pending) { + abandon(); + return; + } closed = true; if (subscription != _activeSubscription) return; - _presenceUnsub?.call(); _presenceUnsub = null; _subscribedKeys.clear(); _invalidateSnapshot(); @@ -193,38 +256,17 @@ class PresenceCacheNotifier extends Notifier> { unsub(); return; } - final previous = _presenceUnsub; - // A replacement still backing off would leave nobody live: keep the - // working subscription, and let the next poll try a wider one again. - if (retrying && previous != null) { - closed = true; - unsub(); - return; - } - // Read newly covered keys again: an earlier snapshot of them was not - // backed by a live subscription. - _pending.addAll(keys.where((key) => !_subscribedKeys.contains(key))); - _activeSubscription = subscription; - _retrying = retrying; - _presenceUnsub = unsub; - _subscribedKeys - ..clear() - ..addAll(keys); - previous?.call(); - if (_retrying) { - _invalidateSnapshot(); - return; - } - // Keys tracked while this one opened need a wider subscription. - if (!_subscribedKeys.containsAll(_tracked)) _queueResubscribe(generation); - _schedule(generation); + // With nothing to protect, the first subscription is active once open. + if (!replacing) promote(); } catch (error) { - if (generation == _generation) { + if (generation != _generation) return; + debugPrint('[PresenceCacheNotifier] subscription failed: $error'); + if (replacing) { + if (pending) abandon(); + } else if (subscription != _activeSubscription) { + _opening = false; _invalidateSnapshot(); - debugPrint('[PresenceCacheNotifier] subscription failed: $error'); } - } finally { - if (generation == _generation) _opening = false; } } diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 150439991d..3c82334e28 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -1,3 +1,6 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; import 'package:flutter/widgets.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -114,6 +117,141 @@ void main() { expect(container.read(presenceCacheProvider)['bob'], 'online'); }); + test( + 'a first subscription still retrying when it opens waits for ready', + () async { + final relaySession = _RecordingRelaySessionNotifier()..autoReady = false; + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + final subscription = relaySession.subscriptions.single; + + // A retryable CLOSED releases the open before the stream is live, so a + // snapshot now could miss a change before the retry's EOSE. + relaySession.snapshot['alice'] = 'online'; + subscription.retrying(); + await _pumpEventQueue(); + expect(relaySession.queried, isEmpty); + + subscription.ready(); + await _pumpEventQueue(); + expect(relaySession.queried, [ + ['alice'], + ]); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }, + ); + + test('the poll still reads a live stream while its widening fails', () { + fakeAsync((time) { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + final notifier = container.read(presenceCacheProvider.notifier); + + container.read(presenceCacheProvider); + notifier.track(['alice']); + time.elapse(Duration.zero); + relaySession.emit(_presence('alice', 'online')); + relaySession.autoReady = false; + notifier.track(['bob']); + time.elapse(Duration.zero); + relaySession.subscriptions.last.retrying(); + time.elapse(Duration.zero); + expect(relaySession.subscriptions, hasLength(1)); + + // Alice's lease expired; each poll retries the widening and still + // reads her, so she is corrected to offline. + relaySession.queried.clear(); + time.elapse(const Duration(seconds: 60)); + expect(relaySession.subscriptions, hasLength(2)); + expect(relaySession.queried.expand((keys) => keys), contains('alice')); + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + }); + + test( + 'a wider subscription replaces the live one only once it is ready', + () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + final notifier = container.read(presenceCacheProvider.notifier); + Map cache() => container.read(presenceCacheProvider); + + container.read(presenceCacheProvider); + notifier.track(['alice']); + await _pumpEventQueue(); + relaySession.emit(_presence('alice', 'online')); + expect(cache(), {'alice': 'online'}); + final first = relaySession.subscriptions.single; + + // Tracking bob opens a replacement; alice's stream keeps delivering. + relaySession.autoReady = false; + notifier.track(['bob']); + await _pumpEventQueue(); + final retried = relaySession.subscriptions.last; + expect(retried.filter.authors, ['alice', 'bob']); + relaySession.emit(_presence('alice', 'away')); + expect(cache()['alice'], 'away'); + + // A second track while one is pending does not open a third. + notifier.track(['carol']); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [first, retried]); + + // A retryable CLOSED on the replacement retires it, not alice's stream, + // and her status stays on screen. + retried.retrying(); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [first]); + expect(cache()['alice'], 'away'); + relaySession.emit(_presence('alice', 'online', 1)); + expect(cache()['alice'], 'online'); + + // The next replacement resolves on the timeout, before its EOSE. + notifier.track(['dave']); + await _pumpEventQueue(); + final slow = relaySession.subscriptions.last; + expect(slow.filter.authors, ['alice', 'bob', 'carol', 'dave']); + slow.open(); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [first, slow]); + + // Bob went offline after any earlier snapshot of him and before the new + // stream covered him. Its late EOSE promotes it without blanking alice, + // and the keys it newly covers are read again. + relaySession.snapshot['bob'] = 'offline'; + relaySession.queried.clear(); + slow.ready(); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [slow]); + expect( + relaySession.queried.expand((keys) => keys), + containsAll(['bob', 'carol', 'dave']), + ); + expect( + relaySession.queried.expand((keys) => keys), + isNot(contains('alice')), + ); + expect(cache()['alice'], 'online'); + expect(cache()['bob'], 'offline'); + + // A replacement that fails outright leaves the live stream and cache. + notifier.track(['erin']); + await _pumpEventQueue(); + relaySession.subscriptions.last.fail(); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [slow]); + expect(cache()['alice'], 'online'); + relaySession.emit(_presence('alice', 'away', 1)); + expect(cache()['alice'], 'away'); + }, + ); + test('WS event uses pubkey variable, not literal string', () async { // Regression test for the map key bug where `{...state, pubkey: status}` // used the literal string "pubkey" instead of the variable's value. @@ -144,8 +282,8 @@ void main() { }); } -NostrEvent _presence(String pubkey, String status) => NostrEvent( - id: 'evt-$pubkey-$status', +NostrEvent _presence(String pubkey, String status, [int seq = 0]) => NostrEvent( + id: 'evt-$pubkey-$status-$seq', pubkey: pubkey, createdAt: 1000, kind: EventKind.presenceUpdate, @@ -172,7 +310,12 @@ ProviderContainer _buildContainer({ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { final List filters = []; - final List _listeners = []; + final List<_FakeSubscription> subscriptions = []; + final List> queried = []; + final Map snapshot = {}; + + /// When false, each subscription waits for the test to drive it. + bool autoReady = true; @override SessionState build() => const SessionState(status: SessionStatus.connected); @@ -183,29 +326,87 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, required void Function(RelaySubscriptionStatus) onStatusChanged, - }) async { + }) { + final subscription = _FakeSubscription( + this, + filter, + onEvent, + onClosed, + onStatusChanged, + ); filters.add(filter); - _listeners.add(onEvent); - return () { - filters.remove(filter); - _listeners.remove(onEvent); - }; + subscriptions.add(subscription); + if (autoReady) { + // Like an EOSE that arrives before the open future resolves. + onStatusChanged(RelaySubscriptionStatus.ready); + subscription.open(); + } + return subscription.opened.future; } @override Future> queryRelay( List filters, { Duration timeout = const Duration(seconds: 8), - }) async => []; + }) async { + final authors = filters.single.authors!; + queried.add(authors); + return [ + for (final key in authors) + if (snapshot[key] case final status?) _presence(key, status), + ]; + } /// Emit an event synchronously to all live subscribers. void emit(NostrEvent event) { - for (final listener in List.of(_listeners)) { - listener(event); + for (final subscription in List.of(subscriptions)) { + if (subscription.live) subscription.onEvent(event); } } } +/// One subscription whose open, status and failure order the test controls. +class _FakeSubscription { + _FakeSubscription( + this.relay, + this.filter, + this.onEvent, + this.onClosed, + this.onStatusChanged, + ); + + final _RecordingRelaySessionNotifier relay; + final NostrFilter filter; + final void Function(NostrEvent) onEvent; + final void Function(String message)? onClosed; + final void Function(RelaySubscriptionStatus) onStatusChanged; + final opened = Completer(); + bool live = true; + + void _remove() { + live = false; + relay.filters.remove(filter); + relay.subscriptions.remove(this); + } + + /// The open future resolves (EOSE, retryable CLOSED or the 500 ms timeout). + void open() => opened.complete(_remove); + void ready() => onStatusChanged(RelaySubscriptionStatus.ready); + + /// A retryable CLOSED: the open future resolves, then status is retrying. + void retrying() { + if (!opened.isCompleted) open(); + onStatusChanged(RelaySubscriptionStatus.retrying); + } + + /// A terminal CLOSED before the open future resolves. + void fail() { + _remove(); + onClosed?.call('blocked: test'); + opened.completeError(Exception('blocked: test')); + } +} + class _FakeAppLifecycleNotifier extends AppLifecycleNotifier { @override AppLifecycleState build() => AppLifecycleState.resumed; diff --git a/mobile/test/features/profile/presence_ordering_test.dart b/mobile/test/features/profile/presence_ordering_test.dart index 5106d7b40a..094b7d35c7 100644 --- a/mobile/test/features/profile/presence_ordering_test.dart +++ b/mobile/test/features/profile/presence_ordering_test.dart @@ -116,39 +116,6 @@ void main() { } } - test('a person tracked during a snapshot is read again once covered', () { - scenario((time, container, relay) { - relay.ready = Completer(); - container.read(presenceCacheProvider.notifier).track(['bob']); - time.elapse(Duration.zero); - expect(relay.subscriptions, 2); - - // Alice's snapshot settles and bob's is read before his subscription - // opens, so a change of his in between could be missed. - relay.results.removeAt(0).complete([]); - time.flushMicrotasks(); - expect(relay.queries.last.authors, ['bob']); - final beforeCovered = relay.queries.length; - - relay.ready!.complete(); - time.elapse(Duration.zero); - relay.results.removeAt(0).complete([]); - time.elapse(Duration.zero); - expect(relay.queries.length, beforeCovered + 1); - expect(relay.queries.last.authors, ['bob']); - }); - }); - - test('a replacement still backing off keeps the working subscription', () { - scenario((time, container, relay) { - relay.openingStatus = RelaySubscriptionStatus.retrying; - container.read(presenceCacheProvider.notifier).track(['bob']); - time.elapse(Duration.zero); - expect(relay.subscriptions, 2); - expect(relay.unsubscribed, [2]); - }); - }); - test('redelivered ambiguous event arms one confirmation', () { scenario((time, container, relay) { relay.results.removeAt(0).complete([ @@ -204,8 +171,6 @@ class _Relay extends RelaySessionNotifier { bool failSubscribe = false; int subscriptions = 0; int unsubscribes = 0; - final unsubscribed = []; - RelaySubscriptionStatus openingStatus = RelaySubscriptionStatus.ready; @override SessionState build() => const SessionState(status: SessionStatus.connected); @@ -217,15 +182,12 @@ class _Relay extends RelaySessionNotifier { void Function(String)? onClosed, required void Function(RelaySubscriptionStatus) onStatusChanged, }) async { - final subscription = ++subscriptions; + subscriptions++; if (failSubscribe) throw StateError('unavailable'); emit = onEvent; - onStatusChanged(openingStatus); + onStatusChanged(RelaySubscriptionStatus.ready); if (ready != null) await ready!.future; - return () { - unsubscribes++; - unsubscribed.add(subscription); - }; + return () => unsubscribes++; } @override