fix(mobile): swap in a wider presence subscription only once it is live

A replacement opened when subscribeWithStatus returned, which also happens
on a retryable CLOSED or the 500 ms timeout, so it could retire a healthy
stream or blank every dot on a late EOSE. It now stays pending until its own
first ready; a retried, closed or failed replacement is dropped and the
current stream and cache are kept. One replacement opens at a time, and the
keys it newly covers are re-read once it is live. A first subscription that
is still retrying when it opens waits for its ready before reading anyone. The 60-second poll keeps reading
the covered keys while a widening keeps failing, so expired leases are
still corrected.

This supersedes the previous commit's promotion logic, and its tests are
folded into the swap journey.

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,6 +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;
   Timer? _poll;
   Timer? _queued;
   Timer? _resubscribe;
@@ -70,6 +72,8 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
       _generation++;
       _presenceUnsub?.call();
       _presenceUnsub = null;
+      _pendingClose?.call();
+      _pendingClose = null;
       _poll?.cancel();
       _queued?.cancel();
       _queued = null;
@@ -81,7 +85,10 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
       _poll = Timer.periodic(const Duration(seconds: 60), (_) {
         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);
         }
@@ -145,20 +152,69 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
         (_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;
     _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 reported before this subscription replaces the active one.
+    // Status the first subscription reports before it becomes active.
     var retrying = false;
-    // The subscription being replaced keeps delivering until this one opens.
+    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;
+      _pendingClose = 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;
+      _pendingClose = null;
+      _opening = false;
+      close();
+      // Keys only it would cover get a snapshot; the poll retries the widening.
+      _schedule(generation);
+    }
+
+    if (pending) _pendingClose = close;
     try {
-      final unsub = await ref
+      unsub = await ref
           .read(relaySessionProvider.notifier)
           .subscribeWithStatus(
             NostrFilter(
@@ -171,6 +227,10 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
             },
             onStatusChanged: (status) {
               if (generation != _generation || closed) return;
+              if (pending) {
+                status == RelaySubscriptionStatus.ready ? promote() : abandon();
+                return;
+              }
               if (subscription != _activeSubscription) {
                 retrying = status == RelaySubscriptionStatus.retrying;
                 return;
@@ -181,9 +241,12 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
             },
             onClosed: (_) {
               if (generation != _generation || closed) return;
+              if (pending) {
+                abandon();
+                return;
+              }
               closed = true;
               if (subscription != _activeSubscription) return;
-              _presenceUnsub?.call();
               _presenceUnsub = null;
               _subscribedKeys.clear();
               _invalidateSnapshot();
@@ -193,38 +256,17 @@ class PresenceCacheNotifier extends Notifier<Map<String, String>> {
         unsub();
         return;
       }
-      final previous = _presenceUnsub;
-      // A replacement still backing off would leave nobody live: keep the
-      // working subscription, and let the next poll try a wider one again.
-      if (retrying && previous != null) {
-        closed = true;
-        unsub();
-        return;
-      }
-      // Read newly covered keys again: an earlier snapshot of them was not
-      // backed by a live subscription.
-      _pending.addAll(keys.where((key) => !_subscribedKeys.contains(key)));
-      _activeSubscription = subscription;
-      _retrying = retrying;
-      _presenceUnsub = unsub;
-      _subscribedKeys
-        ..clear()
-        ..addAll(keys);
-      previous?.call();
-      if (_retrying) {
-        _invalidateSnapshot();
-        return;
-      }
-      // Keys tracked while this one opened need a wider subscription.
-      if (!_subscribedKeys.containsAll(_tracked)) _queueResubscribe(generation);
-      _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;
     }
   }
 

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';
@@ -114,6 +117,141 @@ void main() {
     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;
+
+      // 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);
+
+      subscription.ready();
+      await _pumpEventQueue();
+      expect(relaySession.queried, [
+        ['alice'],
+      ]);
+      expect(container.read(presenceCacheProvider)['alice'], 'online');
+    },
+  );
+
+  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.
@@ -144,8 +282,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,
@@ -172,7 +310,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);
@@ -183,29 +326,87 @@ 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 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;

mobile/test/features/profile/presence_ordering_test.dart

@@ -116,39 +116,6 @@ void main() {
     }
   }
 
-  test('a person tracked during a snapshot is read again once covered', () {
-    scenario((time, container, relay) {
-      relay.ready = Completer<void>();
-      container.read(presenceCacheProvider.notifier).track(['bob']);
-      time.elapse(Duration.zero);
-      expect(relay.subscriptions, 2);
-
-      // Alice's snapshot settles and bob's is read before his subscription
-      // opens, so a change of his in between could be missed.
-      relay.results.removeAt(0).complete([]);
-      time.flushMicrotasks();
-      expect(relay.queries.last.authors, ['bob']);
-      final beforeCovered = relay.queries.length;
-
-      relay.ready!.complete();
-      time.elapse(Duration.zero);
-      relay.results.removeAt(0).complete([]);
-      time.elapse(Duration.zero);
-      expect(relay.queries.length, beforeCovered + 1);
-      expect(relay.queries.last.authors, ['bob']);
-    });
-  });
-
-  test('a replacement still backing off keeps the working subscription', () {
-    scenario((time, container, relay) {
-      relay.openingStatus = RelaySubscriptionStatus.retrying;
-      container.read(presenceCacheProvider.notifier).track(['bob']);
-      time.elapse(Duration.zero);
-      expect(relay.subscriptions, 2);
-      expect(relay.unsubscribed, [2]);
-    });
-  });
-
   test('redelivered ambiguous event arms one confirmation', () {
     scenario((time, container, relay) {
       relay.results.removeAt(0).complete([
@@ -204,8 +171,6 @@ class _Relay extends RelaySessionNotifier {
   bool failSubscribe = false;
   int subscriptions = 0;
   int unsubscribes = 0;
-  final unsubscribed = <int>[];
-  RelaySubscriptionStatus openingStatus = RelaySubscriptionStatus.ready;
 
   @override
   SessionState build() => const SessionState(status: SessionStatus.connected);
@@ -217,15 +182,12 @@ class _Relay extends RelaySessionNotifier {
     void Function(String)? onClosed,
     required void Function(RelaySubscriptionStatus) onStatusChanged,
   }) async {
-    final subscription = ++subscriptions;
+    subscriptions++;
     if (failSubscribe) throw StateError('unavailable');
     emit = onEvent;
-    onStatusChanged(openingStatus);
+    onStatusChanged(RelaySubscriptionStatus.ready);
     if (ready != null) await ready!.future;
-    return () {
-      unsubscribes++;
-      unsubscribed.add(subscription);
-    };
+    return () => unsubscribes++;
   }
 
   @override