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

perf(mobile): subscribe to presence only for people on screen (#9444)

Browse Source 
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](3f789d16b8).
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 <tomb@squareup.com>
Co-authored-by: Tom Brow <tomb@squareup.com>
Co-authored-by: studio <dc2fc481343672df31a689840c09299972e704dac592e0da6f18486646a7ee3e@buzz.block.builderlab.xyz>
This commit is contained in:
Tom Brow
2026-11-13 13:32:53 -08:00
committed by GitHub
parent 359839f70b
commit 36fa0fe318
2 changed files with 448 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
mobile/lib/features/profile/presence_cache_provider.dart
View File

@@ -27,9 +27,18 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
final Map<String, int> _timestamps = {};
final Map<String, int> _snapshotAt = {};
final List<String> _seenIds = [];
// Keys the live subscription asks for. Only tracked keys are subscribed:
// a community's presence heartbeats arrive many times a second.
final Set<String> _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<Map<String, String>> {
_timestamps.clear();
_snapshotAt.clear();
_seenIds.clear();
_subscribedKeys.clear();
_activeSubscription = 0;
_subscribeAttempted = false;
_pending
..clear()
..addAll(_tracked);
@@ -60,16 +72,26 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
_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<Map<String, String>> {
/// Request initial presence for exact keys. Repeated render calls coalesce.
void track(List<String> 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<Map<String, String>> {
}
Future<void> _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<Map<String, String>> {
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;
}
}

313
mobile/test/features/profile/presence_cache_provider_test.dart
View File

@@ -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<String, String> 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<NostrFilter> filters = [];
final List<void Function(NostrEvent)> _listeners = [];
final List<_FakeSubscription> subscriptions = [];
final List<List<String>> queried = [];
final Map<String, String> 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<List<NostrEvent>> queryRelay(
List<NostrFilter> 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<void Function()>();
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;