fix(mobile): serialize channel sections sync (#3165)

### What changed?

Serializes channel-section relay synchronization so a late relay
`CLOSED` cannot overlap an in-flight retry and install duplicate
subscriptions. Pending subscription results are invalidated and
immediately closed when the manager is disposed or superseded.

### Why?

The startup retry added in #3004 could race with a late `CLOSED` or
manager disposal, leaking an untracked live subscription. This keeps
retry recovery single-flight and makes the lifecycle boundary explicit.

### How is it tested?

Build and run.

Added tests:

-
[`ChannelSectionsManager`](https://github.com/block/buzz/tree/main/mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart)
interleaving coverage for in-flight retry serialization and disposal
during subscription setup

*🤖 This PR was authored with a Buzz agent.*

Signed-off-by: npub15w828kxsxu2684ynste0uah2jwkgatd99flt7ds4523hzm8ju6cshdr8hh <a38ea3d8d03715a3d49382f2fe76ea93ac8eada52a7ebf3615a2a3716cf2e6b1@buzz.block.builderlab.xyz>
Co-authored-by: npub15w828kxsxu2684ynste0uah2jwkgatd99flt7ds4523hzm8ju6cshdr8hh <a38ea3d8d03715a3d49382f2fe76ea93ac8eada52a7ebf3615a2a3716cf2e6b1@buzz.block.builderlab.xyz>

mobile/lib/features/channels/channel_sections/channel_sections_manager.dart

@@ -53,6 +53,9 @@ class ChannelSectionsManager {
   Timer? _startupRetryTimer;
   int _startupRetryAttempt = 0;
   bool _startupFetchSucceeded = false;
+  Future<void>? _syncInFlight;
+  bool _syncAgain = false;
+  int _subscriptionGeneration = 0;
 
   ChannelSectionsManager({
     required this.pubkey,
@@ -98,12 +101,33 @@ class ChannelSectionsManager {
   /// sync must eventually land for groups to appear at all, and at the 30s
   /// delay ceiling a persistent retry is cheap. Do not "fix" this into a
   /// bounded loop — giving up permanently is the exact bug this replaces.
-  Future<void> _syncWithRelay() async {
+  Future<void> _syncWithRelay() {
+    if (_disposed) return Future.value();
+    final inFlight = _syncInFlight;
+    if (inFlight != null) {
+      _syncAgain = true;
+      return inFlight;
+    }
+
+    final sync = _runSyncWithRelay();
+    _syncInFlight = sync;
+    return sync.whenComplete(() {
+      _syncInFlight = null;
+      if (_disposed || !_syncAgain) return;
+      _syncAgain = false;
+      unawaited(_syncWithRelay());
+    });
+  }
+
+  Future<void> _runSyncWithRelay() async {
     if (!_startupFetchSucceeded) {
-      _startupFetchSucceeded = await _fetchAndMerge();
+      final fetched = await _fetchAndMerge();
+      if (_disposed) return;
+      _startupFetchSucceeded = fetched;
     }
 
     final subscribed = _unsubscribe != null || await _startLiveSubscription();
+    if (_disposed) return;
 
     if (!_startupFetchSucceeded || !subscribed) {
       _scheduleStartupRetry();
@@ -146,6 +170,8 @@ class ChannelSectionsManager {
   void dispose({bool flushPending = true}) {
     if (_disposed) return;
     _disposed = true;
+    _subscriptionGeneration++;
+    _syncAgain = false;
 
     _startupRetryTimer?.cancel();
     _startupRetryTimer = null;
@@ -270,7 +296,7 @@ class ChannelSectionsManager {
 
   /// Returns whether the fetch reached the relay (regardless of whether a
   /// remote blob exists).
-  Future<bool> _fetchAndMerge() async {
+  Future<bool> _fetchAndMerge({bool allowDisposed = false}) async {
     if (_relaySession == null) return false;
     try {
       final events = await _relaySession.fetchHistory(
@@ -283,6 +309,7 @@ class ChannelSectionsManager {
           limit: 1,
         ),
       );
+      if (_disposed && !allowDisposed) return false;
       _mergeEvents(events);
       _persist();
       if (!_disposed) _onChanged();
@@ -296,9 +323,10 @@ class ChannelSectionsManager {
 
   /// Returns whether the live subscription was established.
   Future<bool> _startLiveSubscription() async {
-    if (_relaySession == null) return false;
+    if (_relaySession == null || _disposed) return false;
+    final generation = ++_subscriptionGeneration;
     try {
-      _unsubscribe = await _relaySession.subscribe(
+      final unsubscribe = await _relaySession.subscribe(
         NostrFilter(
           kinds: const [EventKind.readState],
           authors: [pubkey],
@@ -308,8 +336,13 @@ class ChannelSectionsManager {
           limit: 1,
         ),
         _handleIncomingEvent,
-        onClosed: _handleSubscriptionClosed,
+        onClosed: (message) => _handleSubscriptionClosed(generation, message),
       );
+      if (_disposed || generation != _subscriptionGeneration) {
+        unsubscribe();
+        return false;
+      }
+      _unsubscribe = unsubscribe;
       return true;
     } catch (error) {
       debugPrint('[ChannelSectionsManager] live subscription failed: $error');
@@ -324,12 +357,13 @@ class ChannelSectionsManager {
   /// the rate-limit rejection lands later. Without this handler the manager
   /// would keep a dead subscription and never retry — the exact
   /// load-correlated cold-start failure this retry exists for.
-  void _handleSubscriptionClosed(String message) {
-    if (_disposed) return;
+  void _handleSubscriptionClosed(int generation, String message) {
+    if (_disposed || generation != _subscriptionGeneration) return;
     debugPrint(
       '[ChannelSectionsManager] live subscription closed by relay: $message',
     );
     _unsubscribe = null;
+    _subscriptionGeneration++;
     _scheduleStartupRetry();
   }
 
@@ -404,7 +438,7 @@ class ChannelSectionsManager {
     }
 
     // Read-before-write: merge remote state before publishing
-    await _fetchAndMerge();
+    await _fetchAndMerge(allowDisposed: allowDisposed);
 
     // No-op suppression: skip if nothing changed
     if (_isIdenticalToLastPublished()) return;

mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 
 import 'package:flutter_test/flutter_test.dart';
@@ -199,6 +200,46 @@ void main() {
     expect(relay.subscribeCalls, 1, reason: 'no re-subscribe after dispose');
   });
 
+  test(
+    'dispose while subscribe is pending closes the late subscription',
+    () async {
+      await setUpEnv();
+      final relay = _DelayedSubscribeRelaySession();
+      final manager = buildManager(relaySession: relay);
+
+      final initializing = manager.initialize();
+      await relay.subscribeStarted.future;
+      manager.dispose(flushPending: false);
+      relay.completeSubscribe();
+      await initializing;
+
+      expect(relay.activeListeners, 0);
+      expect(relay.unsubscribeCalls, 1);
+    },
+  );
+
+  test(
+    'a retry request during an in-flight sync does not overlap subscribe',
+    () async {
+      await setUpEnv();
+      final relay = _DelayedSubscribeRelaySession();
+      final manager = buildManager(relaySession: relay);
+
+      final initializing = manager.initialize();
+      await relay.subscribeStarted.future;
+      relay.closePendingSubscription('rate-limited: quota exceeded');
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+
+      expect(relay.subscribeCalls, 1);
+      relay.completeSubscribe();
+      await initializing;
+      await _waitUntil(() => relay.subscribeCalls == 2);
+      expect(relay.maxConcurrentSubscribes, 1);
+      expect(relay.activeListeners, 1);
+      manager.dispose(flushPending: false);
+    },
+  );
+
   test('backoff resets after full recovery so later failures start from the '
       'base delay', () async {
     await setUpEnv();
@@ -242,6 +283,60 @@ Future<void> _waitUntil(
   }
 }
 
+class _DelayedSubscribeRelaySession extends RelaySessionNotifier {
+  final subscribeStarted = Completer<void>();
+  Completer<void Function()>? _pendingSubscribe;
+  void Function(String)? _pendingOnClosed;
+  int subscribeCalls = 0;
+  int concurrentSubscribes = 0;
+  int maxConcurrentSubscribes = 0;
+  int activeListeners = 0;
+  int unsubscribeCalls = 0;
+
+  @override
+  Future<List<NostrEvent>> fetchHistory(
+    NostrFilter filter, {
+    Duration timeout = const Duration(seconds: 8),
+  }) async => const [];
+
+  @override
+  Future<void Function()> subscribe(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+  }) {
+    subscribeCalls++;
+    concurrentSubscribes++;
+    if (concurrentSubscribes > maxConcurrentSubscribes) {
+      maxConcurrentSubscribes = concurrentSubscribes;
+    }
+    if (!subscribeStarted.isCompleted) subscribeStarted.complete();
+    _pendingOnClosed = onClosed;
+    if (subscribeCalls > 1) {
+      concurrentSubscribes--;
+      activeListeners++;
+      return Future.value(_unsubscribe);
+    }
+    _pendingSubscribe = Completer<void Function()>();
+    return _pendingSubscribe!.future.whenComplete(() {
+      concurrentSubscribes--;
+    });
+  }
+
+  void closePendingSubscription(String message) =>
+      _pendingOnClosed?.call(message);
+
+  void completeSubscribe() {
+    activeListeners++;
+    _pendingSubscribe!.complete(_unsubscribe);
+  }
+
+  void _unsubscribe() {
+    activeListeners--;
+    unsubscribeCalls++;
+  }
+}
+
 /// Rejects the first [failuresBeforeSuccess] fetch and subscribe calls with
 /// the relay's rate-limit error, then succeeds — the exact failure mode seen
 /// on Android cold start where the channel-list REQ burst exhausts the