Skip to content

Commit 9844c43

Browse files
square-tombstudio
andcommitted
fix(mobile): recover presence coverage after a retry or a lost replacement
Keys tracked while the first subscription was backing off got live coverage only at the next poll; its recovery now widens the subscription when tracked keys are uncovered. A replacement the relay never answers no longer holds the opening slot: each poll drops it before widening again, so a later close of the active stream is recovered at the next poll. Signed-off-by: Tom Brow <[email protected]> Co-authored-by: studio <dc2fc481343672df31a689840c09299972e704dac592e0da6f18486646a7ee3e@buzz.block.builderlab.xyz>
1 parent ba4d645 commit 9844c43

2 files changed

Lines changed: 88 additions & 12 deletions

File tree

‎mobile/lib/features/profile/presence_cache_provider.dart‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,8 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
3131
// a community's presence heartbeats arrive many times a second.
3232
final Set<String> _subscribedKeys = {};
3333
void Function()? _presenceUnsub;
34-
// Closes the replacement being opened, until it is promoted or abandoned.
35-
void Function()? _pendingClose;
34+
// Drops the replacement being opened, until it is promoted or abandoned.
35+
void Function()? _abandonPending;
3636
Timer? _poll;
3737
Timer? _queued;
3838
Timer? _resubscribe;
@@ -72,8 +72,8 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
7272
_generation++;
7373
_presenceUnsub?.call();
7474
_presenceUnsub = null;
75-
_pendingClose?.call();
76-
_pendingClose = null;
75+
_abandonPending?.call();
76+
_abandonPending = null;
7777
_poll?.cancel();
7878
_queued?.cancel();
7979
_queued = null;
@@ -83,6 +83,9 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
8383
if (sessionState.status == SessionStatus.connected &&
8484
lifecycle == AppLifecycleState.resumed) {
8585
_poll = Timer.periodic(const Duration(seconds: 60), (_) {
86+
// A replacement the relay never answered must not hold the slot, so a
87+
// lost REQ costs at most one poll interval.
88+
_abandonPending?.call();
8689
if (_presenceUnsub == null || !_subscribedKeys.containsAll(_tracked)) {
8790
unawaited(_subscribe(generation));
8891
}
@@ -179,7 +182,7 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
179182
subscription == _activeSubscription;
180183
void promote() {
181184
pending = false;
182-
_pendingClose = null;
185+
_abandonPending = null;
183186
_opening = false;
184187
final previous = _presenceUnsub;
185188
final added = keys.where((key) => !_subscribedKeys.contains(key));
@@ -205,14 +208,15 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
205208

206209
void abandon() {
207210
pending = false;
208-
_pendingClose = null;
209-
_opening = false;
211+
_abandonPending = null;
210212
close();
213+
if (generation != _generation) return;
214+
_opening = false;
211215
// Keys only it would cover get a snapshot; the poll retries the widening.
212216
_schedule(generation);
213217
}
214218

215-
if (pending) _pendingClose = close;
219+
if (pending) _abandonPending = abandon;
216220
try {
217221
unsub = await ref
218222
.read(relaySessionProvider.notifier)
@@ -237,7 +241,12 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
237241
}
238242
_retrying = status == RelaySubscriptionStatus.retrying;
239243
_invalidateSnapshot();
240-
if (!_retrying) _schedule(generation);
244+
if (_retrying) return;
245+
// Keys tracked during the backoff need a wider subscription.
246+
if (!_subscribedKeys.containsAll(_tracked)) {
247+
_queueResubscribe(generation);
248+
}
249+
_schedule(generation);
241250
},
242251
onClosed: (_) {
243252
if (generation != _generation || closed) return;

‎mobile/test/features/profile/presence_cache_provider_test.dart‎

Lines changed: 70 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,10 @@ void main() {
128128
container.read(presenceCacheProvider.notifier).track(['alice']);
129129
await _pumpEventQueue();
130130
final subscription = relaySession.subscriptions.single;
131+
// Bob is tracked while alice's first subscription is still opening.
132+
container.read(presenceCacheProvider.notifier).track(['bob']);
133+
await _pumpEventQueue();
134+
expect(relaySession.subscriptions, [subscription]);
131135

132136
// A retryable CLOSED releases the open before the stream is live, so a
133137
// snapshot now could miss a change before the retry's EOSE.
@@ -136,15 +140,72 @@ void main() {
136140
await _pumpEventQueue();
137141
expect(relaySession.queried, isEmpty);
138142

143+
// Its recovery reads both and widens to cover bob live.
139144
subscription.ready();
140145
await _pumpEventQueue();
141-
expect(relaySession.queried, [
142-
['alice'],
143-
]);
146+
expect(
147+
relaySession.queried.expand((keys) => keys),
148+
containsAll(['alice', 'bob']),
149+
);
150+
expect(relaySession.subscriptions.last.filter.authors, ['alice', 'bob']);
144151
expect(container.read(presenceCacheProvider)['alice'], 'online');
145152
},
146153
);
147154

155+
test('the poll recovers from a replacement the relay never answers', () {
156+
fakeAsync((time) {
157+
final relaySession = _RecordingRelaySessionNotifier();
158+
final container = _buildContainer(relaySession: relaySession);
159+
addTearDown(container.dispose);
160+
final notifier = container.read(presenceCacheProvider.notifier);
161+
162+
container.read(presenceCacheProvider);
163+
notifier.track(['alice']);
164+
time.elapse(Duration.zero);
165+
final first = relaySession.subscriptions.single;
166+
167+
// Bob's replacement opens on the timeout and never gets EOSE or CLOSED.
168+
relaySession.autoReady = false;
169+
notifier.track(['bob']);
170+
time.elapse(Duration.zero);
171+
final lost = relaySession.subscriptions.last;
172+
lost.open();
173+
time.elapse(Duration.zero);
174+
175+
// The next poll drops it and widens again; alice's stream stays.
176+
time.elapse(const Duration(seconds: 60));
177+
expect(relaySession.subscriptions, hasLength(2));
178+
expect(relaySession.subscriptions, isNot(contains(lost)));
179+
final retry = relaySession.subscriptions.last;
180+
expect(retry.filter.authors, ['alice', 'bob']);
181+
retry.open();
182+
time.elapse(Duration.zero);
183+
184+
// That one is lost too, and then alice's stream closes for good.
185+
first.closeTerminal();
186+
time.elapse(Duration.zero);
187+
expect(relaySession.subscriptions, [retry]);
188+
189+
// One poll later a fresh subscription covers everyone and reads them.
190+
relaySession.autoReady = true;
191+
relaySession.queried.clear();
192+
time.elapse(const Duration(seconds: 60));
193+
expect(relaySession.subscriptions, hasLength(1));
194+
expect(relaySession.subscriptions.single.filter.authors, [
195+
'alice',
196+
'bob',
197+
]);
198+
expect(
199+
relaySession.queried.expand((keys) => keys),
200+
containsAll(['alice', 'bob']),
201+
);
202+
expect(container.read(presenceCacheProvider), {
203+
'alice': 'offline',
204+
'bob': 'offline',
205+
});
206+
});
207+
});
208+
148209
test('the poll still reads a live stream while its widening fails', () {
149210
fakeAsync((time) {
150211
final relaySession = _RecordingRelaySessionNotifier();
@@ -399,6 +460,12 @@ class _FakeSubscription {
399460
onStatusChanged(RelaySubscriptionStatus.retrying);
400461
}
401462

463+
/// A terminal CLOSED on a subscription that is already open.
464+
void closeTerminal() {
465+
_remove();
466+
onClosed?.call('blocked: test');
467+
}
468+
402469
/// A terminal CLOSED before the open future resolves.
403470
void fail() {
404471
_remove();

0 commit comments

Comments
 (0)