From 36fa0fe318e34eeb916a77e5833a7fbd736ad24c Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 13 Nov 2026 13:32:53 -0800 Subject: [PATCH] perf(mobile): subscribe to presence only for people on screen (#9444) The mobile app subscribed to every presence event in the community, though it only shows presence for people on screen. It now subscribes only to the people it tracks. ### Validation **Impact:** on an iPhone 12 mini signed in to Block's Builderlab, 5 minutes idle on Messages: | | Presence downloaded | All WebSocket traffic | | --- | --- | --- | | Before | 2.9 MB (8153 events) | 3.1 MB | | After | 3.6 KB (10 events) | 66 KB | That's about 35 MB per hour saved at Builderlab's presence rate (about 27 events per second). Measured on an instrumented profile build of the first revision, [3f789d1](https://github.com/block/buzz/commit/3f789d16b836dc66dd18c0f2b90044cf3937aea2). The later swap fix doesn't change what is subscribed. **Device runs (automated, profile builds on the 12 mini):** - First revision: cold start, opening and leaving the home list, a Control Center pull, and a resume after 60 s in the background. Presence dots kept updating for people on screen. - Final head [9844c43](https://github.com/block/buzz/commit/9844c438dd): cold start, opening two channels, a 60 s background resume, and scrolling the home list through the DM section. Every DM row showed its presence dot (online and away), including rows newly scrolled into view. Opening a DM and its details sheet worked as before. **Added tests:** - Swapping subscriptions: a late replacement is promoted without clearing anyone's dot and re-reads the newly covered people; a retryable close or an outright failure keeps the current stream; a second track while one is opening doesn't open a third subscription. Each guard was removed in turn and the test failed each time. - A widening that keeps failing is retried each poll, and expired leases are still corrected. - Replacements the relay never answers are dropped and retried, and after the current stream also closes, the next poll covers and reads everyone. - A first subscription still retrying when it opens waits for its ready, then reads everyone and widens. **Updated test:** the filter test now expects no subscription while nobody is tracked, then `authors: [alice]`, then `[alice, bob]`. The full mobile test suite passes. ### How it works - The live subscription filters by `authors`: everyone shown with presence this session. The cache never untracks; it resets on community or account change. Nothing is subscribed while nobody is tracked. - When someone new is tracked, a replacement subscription opens for the wider set. The current one keeps delivering until the replacement's first EOSE, then it's swapped in and the old one closed. Newly covered people get a fresh snapshot at that point. - If the replacement fails or closes first, it's dropped and nothing on screen changes. The 60-second poll retries the widening and keeps reading the people already covered. - One replacement opens at a time; people tracked meanwhile join the next round. ### Regressions - A person tracked mid-session gets live presence after one extra subscription round trip (up to about a second on a slow link). People already on screen keep live presence throughout the swap. - Each widening resends the full tracked list (about 67 bytes per person), so a few hundred people cost tens of KB per widening, against 35 MB per hour saved. The relay indexes `authors` filters up to 4096 authors, so this is cheap for it. --------- Signed-off-by: Tom Brow Co-authored-by: Tom Brow Co-authored-by: studio --- .../profile/presence_cache_provider.dart | 168 ++++++++-- .../profile/presence_cache_provider_test.dart | 313 +++++++++++++++++- 2 files changed, 448 insertions(+), 33 deletions(-) diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index 8a56dc1226..8feac12831 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -27,9 +27,18 @@ class PresenceCacheNotifier extends Notifier> { final Map _timestamps = {}; final Map _snapshotAt = {}; final List _seenIds = []; + // Keys the live subscription asks for. Only tracked keys are subscribed: + // a community's presence heartbeats arrive many times a second. + final Set _subscribedKeys = {}; void Function()? _presenceUnsub; + // Drops the replacement being opened, until it is promoted or abandoned. + void Function()? _abandonPending; Timer? _poll; Timer? _queued; + Timer? _resubscribe; + int _nextSubscription = 0; + int _activeSubscription = 0; + bool _subscribeAttempted = false; int _generation = 0; int _epoch = 0; bool _querying = false; @@ -53,6 +62,9 @@ class PresenceCacheNotifier extends Notifier> { _timestamps.clear(); _snapshotAt.clear(); _seenIds.clear(); + _subscribedKeys.clear(); + _activeSubscription = 0; + _subscribeAttempted = false; _pending ..clear() ..addAll(_tracked); @@ -60,16 +72,26 @@ class PresenceCacheNotifier extends Notifier> { _generation++; _presenceUnsub?.call(); _presenceUnsub = null; + _abandonPending?.call(); + _abandonPending = null; _poll?.cancel(); _queued?.cancel(); _queued = null; + _resubscribe?.cancel(); + _resubscribe = null; }); if (sessionState.status == SessionStatus.connected && lifecycle == AppLifecycleState.resumed) { _poll = Timer.periodic(const Duration(seconds: 60), (_) { - if (_presenceUnsub == null) { + // 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)); - } 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); } @@ -82,10 +104,32 @@ class PresenceCacheNotifier extends Notifier> { /// Request initial presence for exact keys. Repeated render calls coalesce. void track(List pubkeys) { + var added = false; for (final key in pubkeys.map((key) => key.trim().toLowerCase())) { - if (key.isNotEmpty && _tracked.add(key)) _pending.add(key); + if (key.isNotEmpty && _tracked.add(key)) { + _pending.add(key); + added = true; + } } - _schedule(_generation); + // While live (the poll runs only then), a new key needs a subscription + // that covers it. A snapshot read before that one opens can miss a change, + // so the key is read again once it is covered. + if (added && (_poll?.isActive ?? false)) { + _queueResubscribe(_generation); + } else { + _schedule(_generation); + } + } + + void _queueResubscribe(int generation) { + _resubscribe ??= Timer(Duration.zero, () { + _resubscribe = null; + // Without a subscription, the next poll opens one with every key. + if (generation == _generation && + (_presenceUnsub != null || !_subscribeAttempted)) { + unawaited(_subscribe(generation)); + } + }); } void _schedule(int generation) { @@ -105,31 +149,115 @@ class PresenceCacheNotifier extends Notifier> { } Future _subscribe(int generation) async { - if (_opening || generation != _generation) return; + if (_opening || + generation != _generation || + _tracked.isEmpty || + (_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; - _retrying = false; + _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 the first subscription reports before it becomes active. + var retrying = false; + 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; + _abandonPending = 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; + _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) _abandonPending = abandon; try { - final unsub = await ref + unsub = await ref .read(relaySessionProvider.notifier) .subscribeWithStatus( - const NostrFilter(kinds: [EventKind.presenceUpdate], limit: 0), + NostrFilter( + kinds: const [EventKind.presenceUpdate], + authors: keys, + limit: 0, + ), (event) { - if (generation == _generation && !closed && !_retrying) { - _handlePresenceEvent(event); - } + if (current() && !_retrying) _handlePresenceEvent(event); }, onStatusChanged: (status) { if (generation != _generation || closed) return; + if (pending) { + status == RelaySubscriptionStatus.ready ? promote() : abandon(); + return; + } + if (subscription != _activeSubscription) { + retrying = status == RelaySubscriptionStatus.retrying; + return; + } _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; + if (pending) { + abandon(); + return; + } closed = true; - _presenceUnsub?.call(); + if (subscription != _activeSubscription) return; _presenceUnsub = null; + _subscribedKeys.clear(); _invalidateSnapshot(); }, ); @@ -137,15 +265,17 @@ class PresenceCacheNotifier extends Notifier> { unsub(); return; } - _presenceUnsub = unsub; - _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 219afd8f3c..8eb4f3497b 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'; @@ -17,6 +20,7 @@ void main() { // Track alice, then emit her initial 'online' status. container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); expect(container.read(presenceCacheProvider)['alice'], 'online'); @@ -35,6 +39,7 @@ void main() { // Track only alice. container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); // Emit event for bob (untracked). relaySession.emit(_presence('bob', 'online')); @@ -52,6 +57,7 @@ void main() { await _pumpEventQueue(); container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); expect(container.read(presenceCacheProvider)['alice'], 'online'); @@ -71,6 +77,7 @@ void main() { await _pumpEventQueue(); container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); // Listen for state changes after initial setup. @@ -84,20 +91,228 @@ void main() { expect(stateChangeCount, 0); }); - test('subscribes to kind:20001 with limit 0', () async { + test('subscribes to kind:20001 for tracked pubkeys only', () async { final relaySession = _RecordingRelaySessionNotifier(); final container = _buildContainer(relaySession: relaySession); addTearDown(container.dispose); container.read(presenceCacheProvider); await _pumpEventQueue(); + // Nothing is tracked, so nothing is worth a community-wide stream. + expect(relaySession.filters, isEmpty); - // Should have subscribed with the correct filter. + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); expect(relaySession.filters, hasLength(1)); expect(relaySession.filters.single.kinds, [EventKind.presenceUpdate]); + expect(relaySession.filters.single.authors, ['alice']); expect(relaySession.filters.single.limit, 0); + + // A newly tracked key replaces the subscription with a wider one. + container.read(presenceCacheProvider.notifier).track(['bob', 'alice']); + await _pumpEventQueue(); + expect(relaySession.filters, hasLength(1)); + expect(relaySession.filters.single.authors, ['alice', 'bob']); + relaySession.emit(_presence('bob', 'online')); + 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; + // 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. + relaySession.snapshot['alice'] = 'online'; + subscription.retrying(); + await _pumpEventQueue(); + expect(relaySession.queried, isEmpty); + + // Its recovery reads both and widens to cover bob live. + subscription.ready(); + await _pumpEventQueue(); + 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(); + 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. @@ -112,6 +327,7 @@ void main() { 'deadbeef', 'cafebabe', ]); + await _pumpEventQueue(); // Seed cafebabe -> offline, then set deadbeef online. relaySession.emit(_presence('cafebabe', 'offline')); @@ -127,8 +343,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, @@ -155,7 +371,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); @@ -166,29 +387,93 @@ 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 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(); + onClosed?.call('blocked: test'); + opened.completeError(Exception('blocked: test')); + } +} + class _FakeAppLifecycleNotifier extends AppLifecycleNotifier { @override AppLifecycleState build() => AppLifecycleState.resumed;