diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index 68cd67ca2a..8feac12831 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -31,8 +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; + // Drops the replacement being opened, until it is promoted or abandoned. + void Function()? _abandonPending; Timer? _poll; Timer? _queued; Timer? _resubscribe; @@ -72,8 +72,8 @@ class PresenceCacheNotifier extends Notifier> { _generation++; _presenceUnsub?.call(); _presenceUnsub = null; - _pendingClose?.call(); - _pendingClose = null; + _abandonPending?.call(); + _abandonPending = null; _poll?.cancel(); _queued?.cancel(); _queued = null; @@ -83,6 +83,9 @@ class PresenceCacheNotifier extends Notifier> { if (sessionState.status == SessionStatus.connected && lifecycle == AppLifecycleState.resumed) { _poll = Timer.periodic(const Duration(seconds: 60), (_) { + // A replacement the relay never answered must not hold the slot, so a + // lost REQ costs at most one poll interval. + _abandonPending?.call(); if (_presenceUnsub == null || !_subscribedKeys.containsAll(_tracked)) { unawaited(_subscribe(generation)); } @@ -179,7 +182,7 @@ class PresenceCacheNotifier extends Notifier> { subscription == _activeSubscription; void promote() { pending = false; - _pendingClose = null; + _abandonPending = null; _opening = false; final previous = _presenceUnsub; final added = keys.where((key) => !_subscribedKeys.contains(key)); @@ -205,14 +208,15 @@ class PresenceCacheNotifier extends Notifier> { void abandon() { pending = false; - _pendingClose = null; - _opening = false; + _abandonPending = null; close(); + if (generation != _generation) return; + _opening = false; // Keys only it would cover get a snapshot; the poll retries the widening. _schedule(generation); } - if (pending) _pendingClose = close; + if (pending) _abandonPending = abandon; try { unsub = await ref .read(relaySessionProvider.notifier) @@ -237,7 +241,12 @@ class PresenceCacheNotifier extends Notifier> { } _retrying = status == RelaySubscriptionStatus.retrying; _invalidateSnapshot(); - if (!_retrying) _schedule(generation); + if (_retrying) return; + // Keys tracked during the backoff need a wider subscription. + if (!_subscribedKeys.containsAll(_tracked)) { + _queueResubscribe(generation); + } + _schedule(generation); }, onClosed: (_) { if (generation != _generation || closed) return; diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 3c82334e28..8eb4f3497b 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -128,6 +128,10 @@ void main() { container.read(presenceCacheProvider.notifier).track(['alice']); await _pumpEventQueue(); final subscription = relaySession.subscriptions.single; + // Bob is tracked while alice's first subscription is still opening. + container.read(presenceCacheProvider.notifier).track(['bob']); + await _pumpEventQueue(); + expect(relaySession.subscriptions, [subscription]); // A retryable CLOSED releases the open before the stream is live, so a // snapshot now could miss a change before the retry's EOSE. @@ -136,15 +140,72 @@ void main() { await _pumpEventQueue(); expect(relaySession.queried, isEmpty); + // Its recovery reads both and widens to cover bob live. subscription.ready(); await _pumpEventQueue(); - expect(relaySession.queried, [ - ['alice'], - ]); + expect( + relaySession.queried.expand((keys) => keys), + containsAll(['alice', 'bob']), + ); + expect(relaySession.subscriptions.last.filter.authors, ['alice', 'bob']); expect(container.read(presenceCacheProvider)['alice'], 'online'); }, ); + test('the poll recovers from a replacement the relay never answers', () { + 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); + final first = relaySession.subscriptions.single; + + // Bob's replacement opens on the timeout and never gets EOSE or CLOSED. + relaySession.autoReady = false; + notifier.track(['bob']); + time.elapse(Duration.zero); + final lost = relaySession.subscriptions.last; + lost.open(); + time.elapse(Duration.zero); + + // The next poll drops it and widens again; alice's stream stays. + time.elapse(const Duration(seconds: 60)); + expect(relaySession.subscriptions, hasLength(2)); + expect(relaySession.subscriptions, isNot(contains(lost))); + final retry = relaySession.subscriptions.last; + expect(retry.filter.authors, ['alice', 'bob']); + retry.open(); + time.elapse(Duration.zero); + + // That one is lost too, and then alice's stream closes for good. + first.closeTerminal(); + time.elapse(Duration.zero); + expect(relaySession.subscriptions, [retry]); + + // One poll later a fresh subscription covers everyone and reads them. + relaySession.autoReady = true; + relaySession.queried.clear(); + time.elapse(const Duration(seconds: 60)); + expect(relaySession.subscriptions, hasLength(1)); + expect(relaySession.subscriptions.single.filter.authors, [ + 'alice', + 'bob', + ]); + expect( + relaySession.queried.expand((keys) => keys), + containsAll(['alice', 'bob']), + ); + expect(container.read(presenceCacheProvider), { + 'alice': 'offline', + 'bob': 'offline', + }); + }); + }); + test('the poll still reads a live stream while its widening fails', () { fakeAsync((time) { final relaySession = _RecordingRelaySessionNotifier(); @@ -399,6 +460,12 @@ class _FakeSubscription { onStatusChanged(RelaySubscriptionStatus.retrying); } + /// A terminal CLOSED on a subscription that is already open. + void closeTerminal() { + _remove(); + onClosed?.call('blocked: test'); + } + /// A terminal CLOSED before the open future resolves. void fail() { _remove();