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 <tomb@squareup.com>
Co-authored-by: studio <dc2fc481343672df31a689840c09299972e704dac592e0da6f18486646a7ee3e@buzz.block.builderlab.xyz>

mobile/lib/features/profile/presence_cache_provider.dart

@@ -31,8 +31,8 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
   // a community's presence heartbeats arrive many times a second.
   final Set<String> _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<Map<String, String>> {
       _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<Map<String, String>> {
     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<Map<String, String>> {
         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<Map<String, String>> {
 
     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<Map<String, String>> {
               }
               _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;

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();