fix(mobile): bound startup channel history load

Aggregate initial last-message discovery through bounded HTTP query chunks,
and pace unread catch-up with cancellation while preserving per-channel live
subscriptions. Add regression coverage for relay routing scope.

Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>

crates/buzz-relay/src/handlers/req.rs

@@ -1544,6 +1544,60 @@ mod tests {
         ]));
     }
 
+    #[test]
+    fn extracted_channel_scope_drives_live_fan_out() {
+        use crate::subscription::SubscriptionRegistry;
+        use buzz_core::StoredEvent;
+        use chrono::Utc;
+        use nostr::{EventBuilder, Keys, Kind, Tag};
+
+        fn stored_channel_event(channel_id: uuid::Uuid) -> StoredEvent {
+            let event = EventBuilder::new(Kind::TextNote, "test")
+                .tags([Tag::parse(["h", &channel_id.to_string()]).expect("valid h tag")])
+                .sign_with_keys(&Keys::generate())
+                .expect("sign");
+            StoredEvent::with_received_at(event, Utc::now(), Some(channel_id), true)
+        }
+
+        let channel_a = uuid::Uuid::new_v4();
+        let channel_b = uuid::Uuid::new_v4();
+        let channel_a_filter = filter_with_channel(channel_a).kind(Kind::TextNote);
+        let registry = SubscriptionRegistry::new();
+        let conn_id = uuid::Uuid::new_v4();
+        registry.register(
+            conn_id,
+            "channel-a".into(),
+            vec![channel_a_filter.clone()],
+            extract_channel_id_from_filters(std::slice::from_ref(&channel_a_filter)),
+        );
+
+        assert_eq!(
+            registry.fan_out(&stored_channel_event(channel_a)),
+            vec![(conn_id, "channel-a".into())],
+        );
+        assert!(registry
+            .fan_out(&stored_channel_event(channel_b))
+            .is_empty());
+
+        let aggregate_filter = Filter::new().kind(Kind::TextNote).custom_tags(
+            SingleLetterTag::lowercase(Alphabet::H),
+            [channel_a.to_string(), channel_b.to_string()],
+        );
+        let aggregate_registry = SubscriptionRegistry::new();
+        aggregate_registry.register(
+            uuid::Uuid::new_v4(),
+            "aggregate".into(),
+            vec![aggregate_filter.clone()],
+            extract_channel_id_from_filters(&[aggregate_filter]),
+        );
+        assert!(aggregate_registry
+            .fan_out(&stored_channel_event(channel_a))
+            .is_empty(),);
+        assert!(aggregate_registry
+            .fan_out(&stored_channel_event(channel_b))
+            .is_empty(),);
+    }
+
     #[test]
     fn test_extract_channel_id_single_channel() {
         let channel_id = uuid::Uuid::new_v4();

mobile/lib/features/channels/channels_provider.dart

@@ -20,6 +20,10 @@ import 'unread_badge/should_notify_for_event.dart';
 
 const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2};
 const _unreadCatchUpLimit = 1000;
+const _lastMessageQueryBatchSize = 100;
+const _lastMessageQueryTotalTimeout = Duration(seconds: 8);
+const _unreadCatchUpConcurrency = 4;
+const _unreadCatchUpStartInterval = Duration(milliseconds: 125);
 const _participatedRootIdsPrefix = 'buzz-thread-participation.v1';
 const _authoredRootIdsPrefix = 'buzz-thread-authored.v1';
 
@@ -36,13 +40,19 @@ const _authoredRootIdsPrefix = 'buzz-thread-authored.v1';
 class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
   static const _backstopInterval = Duration(seconds: 60);
 
-  final Map<String, void Function()> _unsubscribersByChannel = {};
+  final Map<String, ({Object attempt, void Function() unsubscribe})>
+  _liveSubscriptionsByChannel = {};
+  final Map<String, Completer<void>> _pendingSubscriptionCancellations = {};
   Future<void> _liveSubscriptionQueue = Future.value();
   List<Channel> _desiredLiveChannels = const [];
   Set<String> _desiredLiveChannelIds = const {};
   int _subscriptionVersion = 0;
   String? _subscriptionRelayBaseUrl;
   Timer? _backstopTimer;
+  int _unreadCatchUpGeneration = 0;
+  Set<String> _unreadCatchUpChannelIds = const {};
+  bool _bufferingBootstrapLiveEvents = false;
+  final List<NostrEvent> _bootstrapLiveEvents = [];
   final Map<String, int> _latestObservedByChannel = {};
   final Map<String, Map<String, ObservedUnreadEvent>>
   _observedUnreadEventsByChannel = {};
@@ -86,7 +96,17 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     final waitingForInitialConnection =
         sessionState.status != SessionStatus.connected;
     ref.listen(relaySessionProvider, (previous, next) {
-      if (next.status != SessionStatus.connected) return;
+      if (next.status != SessionStatus.connected) {
+        _cancelPendingLiveSubscriptions();
+        if (previous?.status == SessionStatus.connected) {
+          // In-flight history requests cannot always be aborted at the transport
+          // boundary. Invalidate their generation so late responses are ignored,
+          // and clear the completed-set marker so reconnect catch-up restarts.
+          _unreadCatchUpGeneration++;
+          _unreadCatchUpChannelIds = const {};
+        }
+        return;
+      }
       if (waitingForInitialConnection &&
           !_hasLoaded &&
           !connected.isCompleted) {
@@ -110,6 +130,10 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       _observedUnreadEventsByChannel.clear();
       _backstopTimer?.cancel();
       _backstopTimer = null;
+      _unreadCatchUpGeneration++;
+      _unreadCatchUpChannelIds = const {};
+      _bufferingBootstrapLiveEvents = false;
+      _bootstrapLiveEvents.clear();
     });
 
     if (sessionState.status != SessionStatus.connected) {
@@ -273,6 +297,26 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       }
     }
 
+    // Establish live callbacks before the bounded HTTP snapshot. During initial
+    // build there is no published channel list to mutate, so buffer callbacks
+    // and replay them onto the completed snapshot before returning it.
+    final bufferBootstrapLiveEvents = subscribeLive && !_hasLoaded;
+    if (bufferBootstrapLiveEvents) {
+      _bootstrapLiveEvents.clear();
+      _bufferingBootstrapLiveEvents = true;
+    }
+    Future<void>? bootstrapLiveSync;
+    if (subscribeLive) {
+      if (bufferBootstrapLiveEvents) {
+        // Subscription admission is deliberately paced and may take longer than
+        // the bounded snapshot. Run both concurrently; any not-yet-ready channel
+        // is covered by unread catch-up after the live pass completes.
+        bootstrapLiveSync = _subscribeLive(channels);
+      } else {
+        await _subscribeLive(channels);
+      }
+    }
+
     // Step 3: fetch the most recent message per channel to populate lastMessageAt.
     // kind:39000 metadata doesn't carry message timestamps, so channels load with
     // lastMessageAt: null. Without this, unread detection and badge computation
@@ -283,29 +327,38 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
         for (final channel in channels)
           if (channel.isMember && !channel.isArchived) channel,
       ];
-      final channelById = {
-        for (final channel in activeChannels) channel.id: channel,
-      };
-      final events = await _fetchLastMessageEvents(session, activeChannels);
       final lastMessageMap = <String, int>{};
       final mutedChannelIds = _mutedChannelIds();
-      for (final event in events) {
-        final channelId = event.channelId;
-        if (channelId == null) continue;
-        final channel = channelById[channelId];
-        if (channel == null) continue;
-        if (!channel.isDm &&
-            !shouldNotifyForEvent(
-              event,
-              myPk,
-              mutedChannelIds: mutedChannelIds,
-              channelId: channelId,
-            )) {
-          continue;
+
+      // Keep the HTTP request bounded for very large accounts. Each batch still
+      // replaces hundreds of simultaneous WebSocket history subscriptions with
+      // one authenticated /query request, while a failed batch does not erase
+      // successful results from the others. The whole sequence shares the same
+      // eight-second startup budget as the former individual history requests,
+      // so a slow relay cannot delay live subscription setup once per batch.
+      final queryStopwatch = Stopwatch()..start();
+      for (final batch in chunkChannelsForLastMessageQuery(activeChannels)) {
+        final remaining =
+            _lastMessageQueryTotalTimeout - queryStopwatch.elapsed;
+        if (remaining <= Duration.zero) {
+          debugPrint('[ChannelsNotifier] last-message query budget exhausted');
+          break;
         }
-        final current = lastMessageMap[channelId];
-        if (current == null || event.createdAt > current) {
-          lastMessageMap[channelId] = event.createdAt;
+        try {
+          final events = await session.queryRelay(
+            buildChannelLastMessageFilters(batch),
+            timeout: remaining,
+          );
+          lastMessageMap.addAll(
+            resolveChannelLastMessages(
+              batch,
+              events,
+              myPk: myPk,
+              mutedChannelIds: mutedChannelIds,
+            ),
+          );
+        } catch (error) {
+          debugPrint('[ChannelsNotifier] last-message batch failed: $error');
         }
       }
 
@@ -322,6 +375,21 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       }
     }
 
+    // Never let a refresh snapshot roll back a newer callback already applied
+    // to the published list while the bounded HTTP batches were in flight.
+    final publishedLastMessage = {
+      for (final channel in state.value ?? const <Channel>[])
+        if (channel.lastMessageAt != null) channel.id: channel.lastMessageAt!,
+    };
+    for (var i = 0; i < channels.length; i++) {
+      final published = publishedLastMessage[channels[i].id];
+      if (published != null &&
+          (channels[i].lastMessageAt == null ||
+              published.isAfter(channels[i].lastMessageAt!))) {
+        channels[i] = channels[i].copyWith(lastMessageAt: published);
+      }
+    }
+
     channels.sort((left, right) {
       final typeOrder =
           (_channelTypeOrder[left.channelType] ?? 99) -
@@ -351,8 +419,22 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       }
     }
 
-    if (subscribeLive) {
-      await _subscribeLive(channels);
+    if (bufferBootstrapLiveEvents) {
+      // Publish immediately after the bounded snapshot. Events from live REQs
+      // already accepted are replayed now; later callbacks update published
+      // state normally. The paced live pass continues in the background and
+      // starts unread catch-up when complete, recovering the gap for channels
+      // whose REQ was not accepted before their snapshot batch completed.
+      _bufferingBootstrapLiveEvents = false;
+      for (final event in _bootstrapLiveEvents) {
+        _applyLiveEventToChannels(channels, event, myPk);
+      }
+      _bootstrapLiveEvents.clear();
+      if (channels.length <= _lastMessageQueryBatchSize) {
+        await bootstrapLiveSync;
+      } else {
+        unawaited(bootstrapLiveSync!);
+      }
     }
     return channels;
   }
@@ -391,40 +473,13 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     _memberSnapshotsByChannelId = Map.unmodifiable(snapshots);
   }
 
-  /// Fetches each channel's independent latest-message window in one HTTP
-  /// bridge request. The relay preserves NIP-01 per-filter limits while
-  /// executing the filters with bounded concurrency, avoiding an unbounded
-  /// burst of websocket REQs on communities with many channels.
-  Future<List<NostrEvent>> _fetchLastMessageEvents(
-    RelaySessionNotifier session,
-    List<Channel> channels,
-  ) async {
-    if (channels.isEmpty) return const [];
-
-    final filters = [
-      for (final channel in channels)
-        NostrFilter(
-          kinds: EventKind.channelMessageEventKinds,
-          tags: {
-            '#h': [channel.id],
-          },
-          limit: channel.isDm ? 1 : 20,
-        ),
-    ];
-
-    return _fetchChannelHistoryBatch(
-      session,
-      filters,
-      operation: 'latest-message query',
-    );
-  }
-
   Future<List<NostrEvent>> _fetchChannelHistoryBatch(
     RelaySessionNotifier session,
     List<NostrFilter> filters, {
     required String operation,
+    required bool Function() isCancelled,
   }) async {
-    if (filters.isEmpty) return const [];
+    if (filters.isEmpty || isCancelled()) return const [];
 
     try {
       return await session.queryRelay(filters);
@@ -438,6 +493,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     const fallbackConcurrency = 4;
     final events = <NostrEvent>[];
     for (var start = 0; start < filters.length; start += fallbackConcurrency) {
+      if (isCancelled()) break;
       final end = min(start + fallbackConcurrency, filters.length);
       final results = await Future.wait(
         filters.sublist(start, end).map((filter) async {
@@ -448,6 +504,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
           }
         }),
       );
+      if (isCancelled()) break;
       for (final result in results) {
         events.addAll(result);
       }
@@ -535,6 +592,16 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
         if (channel.isMember && !channel.isArchived) channel.id,
     };
     final relayBaseUrl = ref.read(relayConfigProvider).baseUrl;
+    if (!_sameStringSet(_unreadCatchUpChannelIds, channelIds)) {
+      // Stop admitting work from the old channel set while its replacement
+      // live subscriptions are being established.
+      _unreadCatchUpGeneration++;
+    }
+    for (final entry in _pendingSubscriptionCancellations.entries.toList()) {
+      if (channelIds.contains(entry.key)) continue;
+      _pendingSubscriptionCancellations.remove(entry.key);
+      if (!entry.value.isCompleted) entry.value.complete();
+    }
     _desiredLiveChannels = channels;
     _desiredLiveChannelIds = channelIds;
     final subscriptionVersion = ++_subscriptionVersion;
@@ -556,7 +623,8 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     int subscriptionVersion,
     List<Channel> channels,
   ) async {
-    if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
+    if (!ref.mounted ||
+        ref.read(relaySessionProvider).status != SessionStatus.connected) {
       return;
     }
 
@@ -570,10 +638,14 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     }
 
     if (_subscriptionRelayBaseUrl != relayBaseUrl) {
-      for (final unsubscribe in _unsubscribersByChannel.values) {
-        unsubscribe();
+      for (final cancellation in _pendingSubscriptionCancellations.values) {
+        if (!cancellation.isCompleted) cancellation.complete();
       }
-      _unsubscribersByChannel.clear();
+      _pendingSubscriptionCancellations.clear();
+      for (final subscription in _liveSubscriptionsByChannel.values) {
+        subscription.unsubscribe();
+      }
+      _liveSubscriptionsByChannel.clear();
       _subscriptionRelayBaseUrl = relayBaseUrl;
     }
     if (ref.read(relayConfigProvider).baseUrl != relayBaseUrl) {
@@ -582,63 +654,117 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     final session = ref.read(relaySessionProvider.notifier);
     final channelIds = _desiredLiveChannelIds;
 
-    for (final entry in _unsubscribersByChannel.entries.toList()) {
+    for (final entry in _liveSubscriptionsByChannel.entries.toList()) {
       if (channelIds.contains(entry.key)) continue;
-      _unsubscribersByChannel.remove(entry.key);
-      entry.value();
+      _liveSubscriptionsByChannel.remove(entry.key);
+      entry.value.unsubscribe();
     }
 
-    for (final channelId in channelIds) {
-      if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
-        return;
-      }
-      if (_unsubscribersByChannel.containsKey(channelId)) continue;
-      try {
-        final unsubscribe = await session.subscribe(
-          NostrFilter(
-            kinds: EventKind.channelEventKinds,
-            tags: {
-              '#h': [channelId],
-            },
-            limit: 0,
-          ),
-          _handleLiveEvent,
-        );
-        if (ref.read(relaySessionProvider).status != SessionStatus.connected ||
-            !_desiredLiveChannelIds.contains(channelId) ||
-            ref.read(relayConfigProvider).baseUrl != relayBaseUrl ||
-            _subscriptionRelayBaseUrl != relayBaseUrl) {
-          unsubscribe();
-          return;
-        }
-        final replaced = _unsubscribersByChannel[channelId];
-        if (replaced != null) {
-          unsubscribe();
-          continue;
-        }
-        _unsubscribersByChannel[channelId] = unsubscribe;
-      } catch (error) {
-        debugPrint(
-          '[ChannelsNotifier] live subscription failed for $channelId: $error',
-        );
-      }
-    }
+    final newChannelIds = [
+      for (final channelId in channelIds)
+        if (!_liveSubscriptionsByChannel.containsKey(channelId)) channelId,
+    ];
+    final subscribeTasks = <Future<void> Function()>[
+      for (final channelId in newChannelIds)
+        () async {
+          if (!ref.mounted ||
+              ref.read(relaySessionProvider).status !=
+                  SessionStatus.connected) {
+            return;
+          }
+          try {
+            final cancellation = Completer<void>();
+            final attempt = Object();
+            var terminallyClosed = false;
+            _pendingSubscriptionCancellations[channelId] = cancellation;
+            final unsubscribe = await session.subscribeWhenReady(
+              NostrFilter(
+                kinds: EventKind.channelEventKinds,
+                tags: {
+                  '#h': [channelId],
+                },
+                limit: 0,
+              ),
+              _handleLiveEvent,
+              onClosed: (_) {
+                terminallyClosed = true;
+                final current = _liveSubscriptionsByChannel[channelId];
+                if (current?.attempt != attempt) return;
+                _liveSubscriptionsByChannel.remove(channelId);
+                _unreadCatchUpChannelIds = Set.unmodifiable(
+                  _unreadCatchUpChannelIds.difference({channelId}),
+                );
+              },
+              cancelled: cancellation.future,
+            );
+            _pendingSubscriptionCancellations.remove(channelId);
+            if (terminallyClosed) {
+              unsubscribe();
+              return;
+            }
+            if (!ref.mounted) {
+              unsubscribe();
+              return;
+            }
+            if (ref.read(relaySessionProvider).status !=
+                    SessionStatus.connected ||
+                !_desiredLiveChannelIds.contains(channelId) ||
+                ref.read(relayConfigProvider).baseUrl != relayBaseUrl ||
+                _subscriptionRelayBaseUrl != relayBaseUrl) {
+              unsubscribe();
+              return;
+            }
+            final replaced = _liveSubscriptionsByChannel[channelId];
+            if (replaced != null) {
+              unsubscribe();
+              return;
+            }
+            _liveSubscriptionsByChannel[channelId] = (
+              attempt: attempt,
+              unsubscribe: unsubscribe,
+            );
+          } on RelaySubscriptionCancelledException {
+            // The desired channel set, relay, or provider lifetime changed.
+          } catch (error) {
+            debugPrint(
+              '[ChannelsNotifier] live subscription failed for $channelId: $error',
+            );
+          } finally {
+            _pendingSubscriptionCancellations.remove(channelId);
+          }
+        },
+    ];
+    await runPacedTasks(
+      subscribeTasks,
+      maxConcurrent: _unreadCatchUpConcurrency,
+      startInterval: _unreadCatchUpStartInterval,
+      isCancelled: () =>
+          !ref.mounted ||
+          subscriptionVersion != _subscriptionVersion ||
+          ref.read(relaySessionProvider).status != SessionStatus.connected ||
+          ref.read(relayConfigProvider).baseUrl != relayBaseUrl,
+      delay: ref.read(channelsLiveSubscriptionDelayProvider),
+    );
 
-    if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
+    if (!ref.mounted ||
+        ref.read(relaySessionProvider).status != SessionStatus.connected) {
       return;
     }
 
     if (subscriptionVersion != _subscriptionVersion) {
       final desiredChannelIds = _desiredLiveChannelIds;
-      for (final entry in _unsubscribersByChannel.entries.toList()) {
+      for (final entry in _liveSubscriptionsByChannel.entries.toList()) {
         if (desiredChannelIds.contains(entry.key)) continue;
-        _unsubscribersByChannel.remove(entry.key);
-        entry.value();
+        _liveSubscriptionsByChannel.remove(entry.key);
+        entry.value.unsubscribe();
       }
       return;
     }
 
-    unawaited(_catchUpUnreadEvents(channels));
+    _startUnreadCatchUpIfNeeded([
+      for (final channel in channels)
+        if (_liveSubscriptionsByChannel.containsKey(channel.id)) channel,
+    ]);
 
     _backstopTimer?.cancel();
     _backstopTimer = Timer.periodic(
@@ -647,11 +773,31 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     );
   }
 
-  Future<void> _catchUpUnreadEvents(List<Channel> channels) async {
+  void _startUnreadCatchUpIfNeeded(List<Channel> channels) {
+    final catchUpChannelIds = {
+      for (final channel in channels)
+        if (channel.isMember && !channel.isArchived) channel.id,
+    };
+    if (_sameStringSet(_unreadCatchUpChannelIds, catchUpChannelIds)) return;
+    _unreadCatchUpChannelIds = Set.unmodifiable(catchUpChannelIds);
+    final catchUpGeneration = ++_unreadCatchUpGeneration;
+    unawaited(_catchUpUnreadEvents(channels, catchUpGeneration));
+  }
+
+  Future<void> _catchUpUnreadEvents(
+    List<Channel> channels,
+    int catchUpGeneration,
+  ) async {
     final myPk = ref.read(myPubkeyProvider);
     if (myPk == null) return;
 
     final session = ref.read(relaySessionProvider.notifier);
+    final relayBaseUrl = ref.read(relayConfigProvider).baseUrl;
+    bool isCancelled() =>
+        !ref.mounted ||
+        catchUpGeneration != _unreadCatchUpGeneration ||
+        ref.read(relaySessionProvider).status != SessionStatus.connected ||
+        ref.read(relayConfigProvider).baseUrl != relayBaseUrl;
     final mutedChannelIds = _mutedChannelIds();
     final ReadStateState readState;
     try {
@@ -688,7 +834,11 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
         session,
         filters,
         operation: 'unread catch-up',
+        isCancelled: isCancelled,
       );
+      if (isCancelled()) {
+        return;
+      }
 
       for (final event in events) {
         if (event.pubkey.toLowerCase() == myPk.toLowerCase()) {
@@ -716,6 +866,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
           continue;
         }
         _recordUnreadEvent(channel, event, myPk);
+        _mergeLastMessageAt(channel.id, event.createdAt);
       }
     } catch (error) {
       debugPrint('[ChannelsNotifier] unread catch-up failed: $error');
@@ -724,51 +875,80 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     state = state.whenData((channels) => List<Channel>.of(channels));
   }
 
-  void _handleLiveEvent(NostrEvent event) {
-    final channelId = event.channelId;
-    if (channelId == null) return;
-
-    final myPk = ref.read(myPubkeyProvider);
-    final mutedChannelIds = _mutedChannelIds();
-
+  void _mergeLastMessageAt(String channelId, int createdAt) {
+    final eventTime = DateTime.fromMillisecondsSinceEpoch(
+      createdAt * 1000,
+      isUtc: true,
+    );
     state = state.whenData((channels) {
-      final idx = channels.indexWhere((c) => c.id == channelId);
-      if (idx == -1) {
-        refresh();
+      final index = channels.indexWhere((channel) => channel.id == channelId);
+      if (index == -1) return channels;
+      final channel = channels[index];
+      if (channel.lastMessageAt != null &&
+          !eventTime.isAfter(channel.lastMessageAt!)) {
         return channels;
       }
       final updated = List<Channel>.of(channels);
-      final channel = updated[idx];
-
-      if (myPk != null && event.pubkey.toLowerCase() == myPk.toLowerCase()) {
-        _recordSelfThreadInterest(event, myPk);
-      }
-
-      if (myPk != null &&
-          shouldNotifyForEvent(
-            event,
-            myPk,
-            participatedRootIds: _participatedRootIds,
-            followedRootIds: _followedRootIds(),
-            authoredRootIds: _authoredRootIds,
-            mutedChannelIds: mutedChannelIds,
-            channelId: channel.id,
-          )) {
-        _recordUnreadEvent(channel, event, myPk);
-        final eventTime = DateTime.fromMillisecondsSinceEpoch(
-          event.createdAt * 1000,
-          isUtc: true,
-        );
-        if (channel.lastMessageAt == null ||
-            eventTime.isAfter(channel.lastMessageAt!)) {
-          updated[idx] = channel.copyWith(lastMessageAt: eventTime);
-        }
-      }
-
+      updated[index] = channel.copyWith(lastMessageAt: eventTime);
       return updated;
     });
   }
 
+  void _handleLiveEvent(NostrEvent event) {
+    if (_bufferingBootstrapLiveEvents) {
+      _bootstrapLiveEvents.add(event);
+      return;
+    }
+    final myPk = ref.read(myPubkeyProvider);
+
+    state = state.whenData((channels) {
+      final updated = List<Channel>.of(channels);
+      if (!_applyLiveEventToChannels(updated, event, myPk)) {
+        refresh();
+        return channels;
+      }
+      return updated;
+    });
+  }
+
+  bool _applyLiveEventToChannels(
+    List<Channel> channels,
+    NostrEvent event,
+    String? myPk,
+  ) {
+    final channelId = event.channelId;
+    if (channelId == null) return true;
+    final idx = channels.indexWhere((channel) => channel.id == channelId);
+    if (idx == -1) return false;
+    final channel = channels[idx];
+
+    if (myPk != null && event.pubkey.toLowerCase() == myPk.toLowerCase()) {
+      _recordSelfThreadInterest(event, myPk);
+    }
+
+    if (myPk != null &&
+        shouldNotifyForEvent(
+          event,
+          myPk,
+          participatedRootIds: _participatedRootIds,
+          followedRootIds: _followedRootIds(),
+          authoredRootIds: _authoredRootIds,
+          mutedChannelIds: _mutedChannelIds(),
+          channelId: channel.id,
+        )) {
+      _recordUnreadEvent(channel, event, myPk);
+      final eventTime = DateTime.fromMillisecondsSinceEpoch(
+        event.createdAt * 1000,
+        isUtc: true,
+      );
+      if (channel.lastMessageAt == null ||
+          eventTime.isAfter(channel.lastMessageAt!)) {
+        channels[idx] = channel.copyWith(lastMessageAt: eventTime);
+      }
+    }
+    return true;
+  }
+
   Set<String> _mutedChannelIds() => {
     for (final entry in ref.read(channelMutesProvider).store.channels.entries)
       if (entry.value.muted) entry.key,
@@ -894,17 +1074,27 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
     state = await AsyncValue.guard(() => _fetch(subscribeLive: true));
   }
 
+  void _cancelPendingLiveSubscriptions() {
+    for (final cancellation in _pendingSubscriptionCancellations.values) {
+      if (!cancellation.isCompleted) cancellation.complete();
+    }
+    _pendingSubscriptionCancellations.clear();
+  }
+
   void _clearLiveSubscriptions() {
     _subscriptionVersion++;
     _desiredLiveChannels = const [];
     _desiredLiveChannelIds = const {};
-    for (final unsubscribe in _unsubscribersByChannel.values) {
-      unsubscribe();
+    _cancelPendingLiveSubscriptions();
+    for (final subscription in _liveSubscriptionsByChannel.values) {
+      subscription.unsubscribe();
     }
-    _unsubscribersByChannel.clear();
+    _liveSubscriptionsByChannel.clear();
     _subscriptionRelayBaseUrl = null;
     _backstopTimer?.cancel();
     _backstopTimer = null;
+    _unreadCatchUpGeneration++;
+    _unreadCatchUpChannelIds = const {};
   }
 }
 
@@ -912,6 +1102,117 @@ final channelsProvider = AsyncNotifierProvider<ChannelsNotifier, List<Channel>>(
   ChannelsNotifier.new,
 );
 
+final channelsLiveSubscriptionDelayProvider = Provider<TaskDelay>(
+  (_) => _defaultTaskDelay,
+);
+
+List<List<Channel>> chunkChannelsForLastMessageQuery(List<Channel> channels) =>
+    [
+      for (
+        var start = 0;
+        start < channels.length;
+        start += _lastMessageQueryBatchSize
+      )
+        channels.sublist(
+          start,
+          min(start + _lastMessageQueryBatchSize, channels.length),
+        ),
+    ];
+
+List<NostrFilter> buildChannelLastMessageFilters(List<Channel> channels) => [
+  for (final channel in channels)
+    NostrFilter(
+      kinds: EventKind.channelMessageEventKinds,
+      tags: {
+        '#h': [channel.id],
+      },
+      limit: channel.isDm ? 1 : 20,
+    ),
+];
+
+Map<String, int> resolveChannelLastMessages(
+  List<Channel> channels,
+  Iterable<NostrEvent> events, {
+  required String myPk,
+  Set<String> mutedChannelIds = const {},
+}) {
+  final channelsById = {for (final channel in channels) channel.id: channel};
+  final result = <String, int>{};
+  for (final event in events) {
+    final channelId = event.channelId;
+    final channel = channelId == null ? null : channelsById[channelId];
+    if (channel == null) continue;
+    if (!channel.isDm &&
+        !shouldNotifyForEvent(
+          event,
+          myPk,
+          mutedChannelIds: mutedChannelIds,
+          channelId: channelId,
+        )) {
+      continue;
+    }
+    final current = result[channel.id];
+    if (current == null || event.createdAt > current) {
+      result[channel.id] = event.createdAt;
+    }
+  }
+  return result;
+}
+
+typedef TaskDelay = Future<void> Function(Duration duration);
+typedef TaskErrorHandler = void Function(Object error);
+
+Future<void> runPacedTasks(
+  List<Future<void> Function()> tasks, {
+  required int maxConcurrent,
+  required Duration startInterval,
+  required bool Function() isCancelled,
+  TaskDelay delay = _defaultTaskDelay,
+  TaskErrorHandler onError = _defaultTaskErrorHandler,
+}) async {
+  if (maxConcurrent < 1) throw ArgumentError.value(maxConcurrent);
+  var nextTask = 0;
+  var firstStart = true;
+  Future<void> startPermit = Future.value();
+
+  Future<bool> acquireStartPermit() async {
+    if (firstStart) {
+      firstStart = false;
+    } else {
+      startPermit = startPermit.then((_) => delay(startInterval));
+      await startPermit;
+    }
+    return !isCancelled();
+  }
+
+  Future<void> worker() async {
+    while (true) {
+      final index = nextTask++;
+      if (index >= tasks.length) return;
+      if (!await acquireStartPermit()) return;
+      try {
+        await tasks[index]();
+      } catch (error) {
+        onError(error);
+      }
+    }
+  }
+
+  await Future.wait([
+    for (var i = 0; i < min(maxConcurrent, tasks.length); i++) worker(),
+  ]);
+}
+
+Future<void> _defaultTaskDelay(Duration duration) =>
+    Future<void>.delayed(duration);
+
+void _defaultTaskErrorHandler(Object error) {
+  debugPrint('[ChannelsNotifier] paced task failed: $error');
+}
+
+bool _sameStringSet(Set<String> left, Set<String> right) =>
+    left.length == right.length && left.containsAll(right);
+
 String? _observedUnreadRootId(NostrEvent event) =>
     _isBroadcastReply(event) ? null : event.threadReference.rootId;
 

mobile/lib/shared/relay/relay_session.dart

@@ -37,10 +37,18 @@ class _HistorySubscription {
   _HistorySubscription({required this.completer, required this.timeout});
 }
 
+class RelaySubscriptionCancelledException implements Exception {
+  const RelaySubscriptionCancelledException();
+
+  @override
+  String toString() => 'Relay subscription cancelled before ready';
+}
+
 class _LiveSubscription {
   final NostrFilter filter;
   final void Function(NostrEvent) onEvent;
   final void Function(String message)? onClosed;
+  final bool waitForRelayReady;
   Completer<void>? readyCompleter;
   int? lastSeenCreatedAt;
   int closedRetryAttempt = 0;
@@ -49,6 +57,7 @@ class _LiveSubscription {
   _LiveSubscription({
     required this.filter,
     required this.onEvent,
+    required this.waitForRelayReady,
     this.onClosed,
     this.readyCompleter,
   });
@@ -264,6 +273,30 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     NostrFilter filter,
     void Function(NostrEvent) onEvent, {
     void Function(String message)? onClosed,
+  }) =>
+      _subscribe(filter, onEvent, onClosed: onClosed, waitForRelayReady: false);
+
+  /// Subscribe and resolve only after the relay confirms the REQ with EOSE.
+  /// Retryable CLOSED responses keep this pending across the scheduled retry.
+  Future<void Function()> subscribeWhenReady(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+    Future<void>? cancelled,
+  }) => _subscribe(
+    filter,
+    onEvent,
+    onClosed: onClosed,
+    waitForRelayReady: true,
+    cancelled: cancelled,
+  );
+
+  Future<void Function()> _subscribe(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    required bool waitForRelayReady,
+    void Function(String message)? onClosed,
+    Future<void>? cancelled,
   }) async {
     if (_disposed) throw StateError('Relay session is disposed');
     final subId = _nextSubId('l');
@@ -272,26 +305,44 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     _liveSubscriptions[subId] = _LiveSubscription(
       filter: filter,
       onEvent: onEvent,
+      waitForRelayReady: waitForRelayReady,
       onClosed: onClosed,
       readyCompleter: readyCompleter,
     );
 
     _sendReq(subId, filter);
 
-    // Wait for EOSE or a short fallback timeout.
-    try {
-      await readyCompleter.future.timeout(
-        const Duration(milliseconds: 500),
-        onTimeout: () {},
-      );
-    } catch (_) {
-      _liveSubscriptions.remove(subId);
-      _recentDeliveryKeys.removeWhere((key) => key.startsWith('$subId:'));
-      rethrow;
-    }
-    final liveSub = _liveSubscriptions[subId];
-    if (liveSub != null && liveSub.readyCompleter == readyCompleter) {
-      liveSub.readyCompleter = null;
+    if (waitForRelayReady) {
+      if (cancelled == null) {
+        await readyCompleter.future;
+      } else {
+        final cancelledBeforeReady = await Future.any([
+          readyCompleter.future.then((_) => false),
+          cancelled.then((_) => true),
+        ]);
+        if (cancelledBeforeReady) {
+          final liveSub = _liveSubscriptions[subId];
+          if (liveSub != null) _removeLiveSubscription(subId, liveSub);
+          _sendClose(subId);
+          throw const RelaySubscriptionCancelledException();
+        }
+      }
+    } else {
+      // Wait for EOSE or a short fallback timeout.
+      try {
+        await readyCompleter.future.timeout(
+          const Duration(milliseconds: 500),
+          onTimeout: () {},
+        );
+      } catch (_) {
+        _liveSubscriptions.remove(subId);
+        _recentDeliveryKeys.removeWhere((key) => key.startsWith('$subId:'));
+        rethrow;
+      }
+      final liveSub = _liveSubscriptions[subId];
+      if (liveSub != null && liveSub.readyCompleter == readyCompleter) {
+        liveSub.readyCompleter = null;
+      }
     }
 
     return () => _unsubscribe(subId);
@@ -694,7 +745,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
       _removeLiveSubscription(subId, liveSub);
       return;
     }
-    if (readyCompleter != null && !readyCompleter.isCompleted) {
+    if (!liveSub.waitForRelayReady &&
+        readyCompleter != null &&
+        !readyCompleter.isCompleted) {
       readyCompleter.complete();
       liveSub.readyCompleter = null;
     }
@@ -931,6 +984,12 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     final subscriptions = _liveSubscriptions.values.toList();
     _liveSubscriptions.clear();
     for (final subscription in subscriptions) {
+      final readyCompleter = subscription.readyCompleter;
+      if (readyCompleter != null && !readyCompleter.isCompleted) {
+        readyCompleter.completeError(
+          const RelaySubscriptionCancelledException(),
+        );
+      }
       subscription.closedRetryTimer?.cancel();
       subscription.closedRetryTimer = null;
     }

mobile/test/features/channels/channels_provider_test.dart

@@ -1,8 +1,10 @@
 import 'dart:async';
+import 'dart:math';
 
 import 'package:flutter/widgets.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:buzz/features/channels/channel.dart';
 import 'package:buzz/features/channels/channel_management_provider.dart';
 import 'package:buzz/features/channels/channels_provider.dart';
 import 'package:buzz/shared/relay/relay.dart';
@@ -21,6 +23,304 @@ import 'package:buzz/shared/relay/relay.dart';
 void main() {
   const myPk = 'me';
 
+  test('chunks 1,565 channels into sixteen bounded HTTP requests', () {
+    final chunks = chunkChannelsForLastMessageQuery([
+      for (var i = 0; i < 1565; i++) _channel(_generatedChannelId(i)),
+    ]);
+
+    expect(chunks, hasLength(16));
+    expect(chunks.take(15).every((chunk) => chunk.length == 100), isTrue);
+    expect(chunks.last, hasLength(65));
+  });
+
+  test('chunks aggregate last-message queries and isolates failures', () async {
+    final channelIds = List.generate(201, _generatedChannelId);
+    final session =
+        _FakeRelaySession(
+            memberships: [for (final id in channelIds) _membership(id, myPk)],
+            metadata: [for (final id in channelIds) _meta(id: id, name: id)],
+          )
+          ..queryFailureCalls.add(2)
+          ..queryEvents = [
+            _message(channelIds.first, createdAt: 10),
+            _message(channelIds[100], createdAt: 20),
+            _message(channelIds.last, createdAt: 30),
+          ];
+    final container = _buildContainer(session: session);
+    addTearDown(container.dispose);
+
+    final channels = await container.read(channelsProvider.future);
+
+    expect(session.queryFilterBatches.map((batch) => batch.length), [
+      100,
+      100,
+      1,
+    ]);
+    expect(
+      session.queryTimeouts.every(
+        (timeout) => timeout <= const Duration(seconds: 8),
+      ),
+      isTrue,
+    );
+    expect(session.queryFilterBatches.expand((batch) => batch), hasLength(201));
+    expect(
+      channels
+          .firstWhere((channel) => channel.id == channelIds.first)
+          .lastMessageAt,
+      DateTime.fromMillisecondsSinceEpoch(10000, isUtc: true),
+    );
+    expect(
+      channels
+          .firstWhere((channel) => channel.id == channelIds[100])
+          .lastMessageAt,
+      isNull,
+    );
+    expect(
+      channels
+          .firstWhere((channel) => channel.id == channelIds.last)
+          .lastMessageAt,
+      DateTime.fromMillisecondsSinceEpoch(30000, isUtc: true),
+    );
+  });
+
+  test('catch-up recovers event missed before live REQ readiness', () async {
+    final channelIds = List.generate(101, _generatedChannelId);
+    final session = _FakeRelaySession(
+      memberships: [for (final id in channelIds) _membership(id, myPk)],
+      metadata: [for (final id in channelIds) _meta(id: id, name: id)],
+    );
+    session.pauseNextSubscribe();
+    final container = _buildContainer(session: session);
+    addTearDown(container.dispose);
+
+    final channels = await container.read(channelsProvider.future);
+    expect(
+      channels
+          .firstWhere((channel) => channel.id == channelIds.first)
+          .lastMessageAt,
+      isNull,
+    );
+    expect(session.activeSubscriptionCount, lessThan(101));
+
+    final missedEvent = _message(channelIds.first, createdAt: 50);
+    session.emit(missedEvent);
+    session.unreadEvents = [missedEvent];
+    session.resumePausedSubscribe();
+
+    await _waitUntil(() => session.unreadCatchUpQueryCount > 0);
+    await _waitUntil(
+      () =>
+          container
+              .read(channelsProvider.notifier)
+              .observedUnreadEventsByChannel[channelIds.first]
+              ?.containsKey(missedEvent.id) ==
+          true,
+      attempts: 30000,
+    );
+    expect(
+      container
+          .read(channelsProvider)
+          .value!
+          .firstWhere((channel) => channel.id == channelIds.first)
+          .lastMessageAt,
+      DateTime.fromMillisecondsSinceEpoch(50000, isUtc: true),
+    );
+  });
+
+  test('loaded refresh preserves newer live timestamp', () async {
+    final channelIds = List.generate(101, _generatedChannelId);
+    final session = _FakeRelaySession(
+      memberships: [for (final id in channelIds) _membership(id, myPk)],
+      metadata: [for (final id in channelIds) _meta(id: id, name: id)],
+    )..queryEvents = [_message(channelIds.first, createdAt: 10)];
+    final container = _buildContainer(session: session);
+    addTearDown(container.dispose);
+
+    await container.read(channelsProvider.future);
+    session.pauseQueryCall(4);
+    final refresh = container.read(channelsProvider.notifier).refresh();
+    await session.nextPausedQueryStarted;
+    session.emit(_message(channelIds.first, createdAt: 50));
+    session.resumePausedQuery();
+    await refresh;
+
+    expect(
+      container
+          .read(channelsProvider)
+          .value!
+          .firstWhere((channel) => channel.id == channelIds.first)
+          .lastMessageAt,
+      DateTime.fromMillisecondsSinceEpoch(50000, isUtc: true),
+    );
+  });
+
+  test('stale unread fallback stops before its next wave', () async {
+    final channelIds = List.generate(9, _generatedChannelId);
+    final session = _FakeRelaySession(
+      memberships: [for (final id in channelIds) _membership(id, myPk)],
+      metadata: [for (final id in channelIds) _meta(id: id, name: id)],
+    )..queryFailureCalls.add(2);
+    session.pauseNextUnreadCatchUp();
+    final container = _buildContainer(session: session);
+    addTearDown(container.dispose);
+
+    await container.read(channelsProvider.future);
+    await session.nextUnreadCatchUpStarted;
+    expect(session.unreadFallbackFetchCount, 4);
+
+    session.setStatus(SessionStatus.disconnected);
+    session.resumePausedUnreadCatchUp();
+    await Future<void>.delayed(Duration.zero);
+    await Future<void>.delayed(Duration.zero);
+
+    expect(session.unreadFallbackFetchCount, 4);
+  });
+
+  test('unchanged refresh does not restart unread catch-up', () async {
+    final session = _FakeRelaySession(
+      memberships: [_membership(_channelA, myPk)],
+      metadata: [_meta(id: _channelA, name: 'general')],
+    );
+    final container = _buildContainer(session: session);
+    addTearDown(container.dispose);
+
+    await container.read(channelsProvider.future);
+    await _waitUntil(() => session.unreadCatchUpQueryCount == 1);
+    await container.read(channelsProvider.notifier).refresh();
+    await Future<void>.delayed(Duration.zero);
+
+    expect(session.unreadCatchUpQueryCount, 1);
+  });
+
+  test(
+    'disconnect discards stale catch-up and reconnect restarts it',
+    () async {
+      final session = _FakeRelaySession(
+        memberships: [_membership(_channelA, myPk)],
+        metadata: [_meta(id: _channelA, name: 'general')],
+      )..unreadEvents = [_message(_channelA, createdAt: 20)];
+      session.pauseNextUnreadCatchUp();
+      final container = _buildContainer(session: session);
+      addTearDown(container.dispose);
+
+      await container.read(channelsProvider.future);
+      await session.nextUnreadCatchUpStarted;
+
+      session.setStatus(SessionStatus.disconnected);
+      session.resumePausedUnreadCatchUp();
+      await Future<void>.delayed(Duration.zero);
+      expect(
+        container
+            .read(channelsProvider.notifier)
+            .observedUnreadEventsByChannel[_channelA],
+        isNull,
+      );
+
+      session.setStatus(SessionStatus.connected);
+      await _waitUntil(() => session.unreadCatchUpQueryCount == 2);
+      await _waitUntil(
+        () =>
+            container
+                .read(channelsProvider.notifier)
+                .observedUnreadEventsByChannel[_channelA]
+                ?.isNotEmpty ==
+            true,
+      );
+    },
+  );
+
+  test('builds one bounded last-message filter per channel type', () {
+    final filters = buildChannelLastMessageFilters([
+      _channel(_channelA),
+      _channel(_channelB, channelType: 'dm'),
+    ]);
+
+    expect(filters, hasLength(2));
+    expect(filters[0].tags['#h'], [_channelA]);
+    expect(filters[0].limit, 20);
+    expect(filters[1].tags['#h'], [_channelB]);
+    expect(filters[1].limit, 1);
+  });
+
+  test('maps newest qualifying last messages and preserves DM semantics', () {
+    final channels = [
+      _channel(_channelA),
+      _channel(_channelB, channelType: 'dm'),
+    ];
+    final result = resolveChannelLastMessages(channels, [
+      _message(_channelA, createdAt: 10, pubkey: myPk),
+      _message(_channelA, createdAt: 20),
+      _message(_channelA, createdAt: 15),
+      _message(_channelB, createdAt: 30, pubkey: myPk),
+      _message(_channelD, createdAt: 40),
+    ], myPk: myPk);
+
+    expect(result, {_channelA: 20, _channelB: 30});
+  });
+
+  test(
+    'paced tasks are lazy, bounded, failure-isolated, and cancellable',
+    () async {
+      var active = 0;
+      var maxActive = 0;
+      var created = 0;
+      var errors = 0;
+      var cancelled = false;
+      final permits = <Completer<void>>[];
+      final completions = List.generate(6, (_) => Completer<void>());
+      final tasks = [
+        for (var i = 0; i < completions.length; i++)
+          () async {
+            created++;
+            active++;
+            maxActive = max(maxActive, active);
+            if (i == 1) {
+              active--;
+              throw Exception('isolated');
+            }
+            await completions[i].future;
+            active--;
+          },
+      ];
+
+      final run = runPacedTasks(
+        tasks,
+        maxConcurrent: 4,
+        startInterval: const Duration(milliseconds: 125),
+        isCancelled: () => cancelled,
+        delay: (_) {
+          if (cancelled) return Future.value();
+          final permit = Completer<void>();
+          permits.add(permit);
+          return permit.future;
+        },
+        onError: (_) => errors++,
+      );
+      await Future<void>.delayed(Duration.zero);
+      expect(created, 1);
+      expect(permits, hasLength(1));
+
+      for (var expectedCreated = 2; expectedCreated <= 4; expectedCreated++) {
+        permits.removeAt(0).complete();
+        await Future<void>.delayed(Duration.zero);
+        expect(created, expectedCreated);
+        if (expectedCreated < 4) expect(permits, hasLength(1));
+      }
+      expect(maxActive, 3);
+
+      cancelled = true;
+      for (final completion in completions) {
+        if (!completion.isCompleted) completion.complete();
+      }
+      for (final permit in permits) {
+        if (!permit.isCompleted) permit.complete();
+      }
+      await run;
+      expect(created, 4);
+      expect(errors, 1);
+    },
+  );
+
   test(
     'seeds members from the channel-list snapshot during reconnect',
     () async {
@@ -136,6 +436,108 @@ void main() {
     },
   );
 
+  test(
+    'terminal live failure catches up only ready channels and retries gap',
+    () async {
+      final session = _FakeRelaySession(
+        memberships: [
+          _membership(_channelA, myPk),
+          _membership(_channelB, myPk),
+        ],
+        metadata: [
+          _meta(id: _channelA, name: 'general'),
+          _meta(id: _channelB, name: 'random'),
+        ],
+      )..subscribeFailuresByChannel[_channelA] = 1;
+      final container = _buildContainer(session: session);
+      addTearDown(container.dispose);
+
+      await container.read(channelsProvider.future);
+      await _waitUntil(() => session.unreadCatchUpQueryCount == 1);
+      expect(session.activeChannels, {_channelB});
+      expect(
+        session.queryFilterBatches
+            .expand((filters) => filters)
+            .where((filter) => filter.since != null)
+            .map((filter) => filter.tags['#h']!.single),
+        [_channelB],
+      );
+
+      final missedEvent = _message(_channelA, createdAt: 50);
+      session.unreadEvents = [missedEvent];
+      await container.read(channelsProvider.notifier).refresh();
+
+      await _waitUntil(
+        () => session.queryFilterBatches
+            .expand((filters) => filters)
+            .any(
+              (filter) =>
+                  filter.since != null &&
+                  filter.tags['#h']?.single == _channelA,
+            ),
+        attempts: 30000,
+      );
+      expect(session.activeChannels, {_channelA, _channelB});
+      expect(
+        container
+            .read(channelsProvider.notifier)
+            .observedUnreadEventsByChannel[_channelA]
+            ?.containsKey(missedEvent.id),
+        isTrue,
+      );
+      expect(
+        container
+            .read(channelsProvider)
+            .value!
+            .firstWhere((channel) => channel.id == _channelA)
+            .lastMessageAt,
+        DateTime.fromMillisecondsSinceEpoch(50000, isUtc: true),
+      );
+    },
+  );
+
+  test(
+    'post-ready terminal close retries subscription and catches up gap',
+    () async {
+      final session = _FakeRelaySession(
+        memberships: [_membership(_channelA, myPk)],
+        metadata: [_meta(id: _channelA, name: 'general')],
+      );
+      final container = _buildContainer(session: session);
+      addTearDown(container.dispose);
+
+      await container.read(channelsProvider.future);
+      await _waitUntil(() => session.unreadCatchUpQueryCount == 1);
+      expect(session.activeChannels, {_channelA});
+
+      session.terminallyClose(_channelA);
+      expect(session.activeChannels, isEmpty);
+      final missedEvent = _message(_channelA, createdAt: 50);
+      session.unreadEvents = [missedEvent];
+
+      await container.read(channelsProvider.notifier).refresh();
+
+      await _waitUntil(
+        () =>
+            session.activeChannels.contains(_channelA) &&
+            container
+                    .read(channelsProvider.notifier)
+                    .observedUnreadEventsByChannel[_channelA]
+                    ?.containsKey(missedEvent.id) ==
+                true,
+        attempts: 30000,
+      );
+      expect(
+        container
+            .read(channelsProvider)
+            .value!
+            .firstWhere((channel) => channel.id == _channelA)
+            .lastMessageAt,
+        DateTime.fromMillisecondsSinceEpoch(50000, isUtc: true),
+      );
+    },
+  );
+
   test(
     'live subscription diff only removes and adds changed channels',
     () async {
@@ -660,6 +1062,39 @@ const _channelA = '11111111-1111-4111-8111-111111111111';
 const _channelB = '22222222-2222-4222-8222-222222222222';
 const _channelD = '44444444-4444-4444-8444-444444444444';
 
+String _generatedChannelId(int index) {
+  final suffix = index.toRadixString(16).padLeft(12, '0');
+  return 'aaaaaaaa-aaaa-4aaa-8aaa-$suffix';
+}
+
+NostrEvent _message(
+  String channelId, {
+  required int createdAt,
+  String pubkey = 'alice',
+}) => NostrEvent(
+  id: 'message-$channelId-$createdAt',
+  pubkey: pubkey,
+  createdAt: createdAt,
+  kind: EventKind.streamMessageV2,
+  tags: [
+    ['h', channelId],
+  ],
+  content: 'message',
+  sig: 'sig',
+);
+
+Channel _channel(String id, {String channelType = 'stream'}) => Channel(
+  id: id,
+  name: id,
+  channelType: channelType,
+  visibility: 'open',
+  description: '',
+  createdBy: 'creator',
+  createdAt: DateTime.fromMillisecondsSinceEpoch(1000, isUtc: true),
+  memberCount: 1,
+  isMember: true,
+);
+
 /// Build a kind:39002 membership event tagged with the channel id and member.
 NostrEvent _membership(
   String channelId,
@@ -725,13 +1160,16 @@ ProviderContainer _buildContainer({required _FakeRelaySession session}) {
     overrides: [
       appLifecycleProvider.overrideWith(() => _FakeAppLifecycleNotifier()),
       relaySessionProvider.overrideWith(() => session),
+      channelsLiveSubscriptionDelayProvider.overrideWithValue(
+        (_) => Future<void>.value(),
+      ),
       myPubkeyProvider.overrideWithValue('me'),
     ],
   );
 }
 
-Future<void> _waitUntil(bool Function() predicate) async {
-  for (var i = 0; i < 100; i++) {
+Future<void> _waitUntil(bool Function() predicate, {int attempts = 100}) async {
+  for (var i = 0; i < attempts; i++) {
     if (predicate()) return;
     await Future<void>.delayed(Duration.zero);
   }
@@ -757,16 +1195,53 @@ class _FakeRelaySession extends RelaySessionNotifier {
 
   final List<NostrFilter> historyFilters = [];
   final List<List<NostrFilter>> queryBatches = [];
+  final List<List<NostrFilter>> queryFilterBatches = [];
+  final List<Duration> queryTimeouts = [];
+  final Set<int> queryFailureCalls = {};
+  final Map<String, int> subscribeFailuresByChannel = {};
+  List<NostrEvent> queryEvents = [];
+  int? _pausedQueryCall;
+  Completer<void>? _pausedQuery;
+  Completer<void>? _pausedQueryStarted;
   final List<NostrFilter> subscribeFilters = [];
-  final Map<int, (NostrFilter, void Function(NostrEvent))> _subscriptions = {};
+  final Map<
+    int,
+    (NostrFilter, void Function(NostrEvent), void Function(String message)?)
+  >
+  _subscriptions = {};
   int _nextSubscriptionKey = 0;
   Completer<void>? _pausedSubscribe;
   Completer<void>? _subscribeStarted;
+  Completer<void>? _pausedUnreadCatchUp;
+  Completer<void>? _unreadCatchUpStarted;
+  List<NostrEvent> unreadEvents = [];
   int unsubscribeCount = 0;
   int totalSubscribeCount = 0;
 
+  int get unreadFallbackFetchCount => historyFilters
+      .where(
+        (filter) =>
+            filter.tags['#h']?.length == 1 &&
+            filter.since != null &&
+            filter.kinds.length == EventKind.channelMessageEventKinds.length &&
+            filter.kinds.every(EventKind.channelMessageEventKinds.contains),
+      )
+      .length;
+
+  int get unreadCatchUpQueryCount => queryFilterBatches
+      .expand((filters) => filters)
+      .where(
+        (filter) =>
+            filter.tags['#h']?.length == 1 &&
+            filter.since != null &&
+            filter.kinds.length == EventKind.channelMessageEventKinds.length &&
+            filter.kinds.every(EventKind.channelMessageEventKinds.contains),
+      )
+      .length;
+
   Set<String> get activeChannels => {
-    for (final (filter, _) in _subscriptions.values) ?filter.tags['#h']?.single,
+    for (final (filter, _, _) in _subscriptions.values)
+      ?filter.tags['#h']?.single,
   };
 
   int get activeSubscriptionCount => _subscriptions.length;
@@ -793,6 +1268,46 @@ class _FakeRelaySession extends RelaySessionNotifier {
     paused.complete();
   }
 
+  Future<void> get nextUnreadCatchUpStarted async {
+    final started = _unreadCatchUpStarted;
+    if (started == null) {
+      throw StateError('No paused unread catch-up is pending');
+    }
+    await started.future;
+  }
+
+  void pauseNextUnreadCatchUp() {
+    if (_pausedUnreadCatchUp != null) {
+      throw StateError('An unread catch-up is already paused');
+    }
+    _pausedUnreadCatchUp = Completer<void>();
+    _unreadCatchUpStarted = Completer<void>();
+  }
+
+  void resumePausedUnreadCatchUp() {
+    final paused = _pausedUnreadCatchUp;
+    if (paused == null) throw StateError('No unread catch-up is paused');
+    paused.complete();
+  }
+
+  Future<void> get nextPausedQueryStarted async {
+    final started = _pausedQueryStarted;
+    if (started == null) throw StateError('No query is paused');
+    await started.future;
+  }
+
+  void pauseQueryCall(int call) {
+    _pausedQueryCall = call;
+    _pausedQuery = Completer<void>();
+    _pausedQueryStarted = Completer<void>();
+  }
+
+  void resumePausedQuery() {
+    final paused = _pausedQuery;
+    if (paused == null) throw StateError('No query is paused');
+    paused.complete();
+  }
+
   @override
   SessionState build() => const SessionState(status: SessionStatus.connected);
 
@@ -824,6 +1339,19 @@ class _FakeRelaySession extends RelaySessionNotifier {
       final ids = (filter.tags['#d'] ?? const <String>[]).toSet();
       return metadata.where((e) => ids.contains(e.getTagValue('d'))).toList();
     }
+    if (filter.since != null && filter.tags['#h']?.length == 1) {
+      final paused = _pausedUnreadCatchUp;
+      if (paused != null) {
+        _unreadCatchUpStarted!.complete();
+        await paused.future;
+        _pausedUnreadCatchUp = null;
+        _unreadCatchUpStarted = null;
+      }
+      final channelId = filter.tags['#h']!.single;
+      return unreadEvents
+          .where((event) => event.channelId == channelId)
+          .toList();
+    }
     return const [];
   }
 
@@ -832,10 +1360,39 @@ class _FakeRelaySession extends RelaySessionNotifier {
     List<NostrFilter> filters, {
     Duration timeout = const Duration(seconds: 8),
   }) async {
-    queryBatches.add(filters);
-    return recentMessages.where((event) {
+    queryBatches.add(List.of(filters));
+    queryFilterBatches.add(List.of(filters));
+    queryTimeouts.add(timeout);
+    if (_pausedQueryCall == queryFilterBatches.length) {
+      _pausedQueryStarted!.complete();
+      await _pausedQuery!.future;
+      _pausedQueryCall = null;
+      _pausedQuery = null;
+      _pausedQueryStarted = null;
+    }
+    if (queryFailureCalls.contains(queryFilterBatches.length)) {
+      throw Exception('query failed');
+    }
+    if (filters.any((filter) => filter.since != null)) {
+      final paused = _pausedUnreadCatchUp;
+      if (paused != null) {
+        _unreadCatchUpStarted!.complete();
+        await paused.future;
+        _pausedUnreadCatchUp = null;
+        _unreadCatchUpStarted = null;
+      }
+    }
+    final candidates = <NostrEvent>[
+      ...recentMessages,
+      ...queryEvents,
+      ...unreadEvents,
+    ];
+    return candidates.where((event) {
       return filters.any((filter) {
         if (!filter.kinds.contains(event.kind)) return false;
+        if (filter.since != null && event.createdAt < filter.since!) {
+          return false;
+        }
         for (final entry in filter.tags.entries) {
           final tagName = entry.key.startsWith('#')
               ? entry.key.substring(1)
@@ -854,23 +1411,45 @@ class _FakeRelaySession extends RelaySessionNotifier {
     }).toList();
   }
 
+  @override
+  Future<void Function()> subscribeWhenReady(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+    Future<void>? cancelled,
+  }) => _subscribeFake(filter, onEvent, onClosed: onClosed);
+
   @override
   Future<void Function()> subscribe(
     NostrFilter filter,
     void Function(NostrEvent) onEvent, {
     void Function(String message)? onClosed,
+  }) => _subscribeFake(filter, onEvent, onClosed: onClosed);
+
+  Future<void Function()> _subscribeFake(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
   }) async {
     totalSubscribeCount++;
     subscribeFilters.add(filter);
+    final channelId = filter.tags['#h']?.single;
+    final remainingFailures = channelId == null
+        ? 0
+        : subscribeFailuresByChannel[channelId] ?? 0;
+    if (remainingFailures > 0) {
+      subscribeFailuresByChannel[channelId!] = remainingFailures - 1;
+      throw Exception('terminal subscription failure for $channelId');
+    }
     final paused = _pausedSubscribe;
-    if (paused != null) {
+    if (paused != null && !_subscribeStarted!.isCompleted) {
       _subscribeStarted!.complete();
       await paused.future;
       _pausedSubscribe = null;
       _subscribeStarted = null;
     }
     final subscriptionKey = ++_nextSubscriptionKey;
-    _subscriptions[subscriptionKey] = (filter, onEvent);
+    _subscriptions[subscriptionKey] = (filter, onEvent, onClosed);
     return () {
       final subscription = _subscriptions.remove(subscriptionKey);
       if (subscription == null) return;
@@ -883,9 +1462,18 @@ class _FakeRelaySession extends RelaySessionNotifier {
     state = SessionState(status: status);
   }
 
+  void terminallyClose(String channelId) {
+    final entry = _subscriptions.entries.singleWhere(
+      (entry) => entry.value.$1.tags['#h']?.single == channelId,
+    );
+    _subscriptions.remove(entry.key);
+    subscribeFilters.remove(entry.value.$1);
+    entry.value.$3?.call('restricted: access revoked');
+  }
+
   /// Emit a live event to all subscribers.
   void emit(NostrEvent event) {
-    for (final (_, listener) in List.of(_subscriptions.values)) {
+    for (final (_, listener, _) in List.of(_subscriptions.values)) {
       listener(event);
     }
   }

mobile/test/shared/relay/relay_session_test.dart

@@ -749,6 +749,145 @@ void main() {
     },
   );
 
+  test(
+    'subscribeWhenReady stays pending across retryable CLOSED until retry EOSE',
+    () async {
+      final timers = <_ManualTimer>[];
+      final socket = _RecordingRelaySocket();
+      final session = RelaySessionNotifier(
+        retryTimerFactory: (duration, callback) {
+          final timer = _ManualTimer(duration, callback);
+          timers.add(timer);
+          return timer;
+        },
+      );
+      session.debugAttachSocketForTest(socket);
+      final events = <NostrEvent>[];
+      var ready = false;
+      final subscribe = session
+          .subscribeWhenReady(_channelFilter, events.add)
+          .whenComplete(() => ready = true);
+
+      session.debugHandleMessage(['CLOSED', 'l-1', 'error: relay overloaded']);
+      await Future<void>.delayed(Duration.zero);
+      expect(ready, isFalse);
+
+      timers.single.fire();
+      await Future<void>.delayed(Duration.zero);
+      expect(_reqs(socket).where((req) => req[1] == 'l-1'), hasLength(2));
+      expect(ready, isFalse);
+
+      final replayEvent = _event(createdAt: 30);
+      session.debugHandleMessage(['EVENT', 'l-1', replayEvent.toJson()]);
+      expect(events, isEmpty);
+      session.debugHandleMessage(['EOSE', 'l-1']);
+
+      final unsubscribe = await subscribe;
+      expect(events, [replayEvent]);
+      expect(ready, isTrue);
+      unsubscribe();
+    },
+  );
+
+  test(
+    'subscribeWhenReady stays pending across rate-limited CLOSED until retry EOSE',
+    () async {
+      final now = DateTime(2026);
+      final gateTimers = <_ManualTimer>[];
+      final retryTimers = <_ManualTimer>[];
+      final gate = RelayRateLimitGate(
+        now: () => now,
+        timerFactory: (duration, callback) {
+          final timer = _ManualTimer(duration, callback);
+          gateTimers.add(timer);
+          return timer;
+        },
+      );
+      final socket = _RecordingRelaySocket();
+      final session = RelaySessionNotifier(
+        rateLimitGate: gate,
+        retryTimerFactory: (duration, callback) {
+          final timer = _ManualTimer(duration, callback);
+          retryTimers.add(timer);
+          return timer;
+        },
+      );
+      session.debugAttachSocketForTest(socket);
+      var ready = false;
+      final subscribe = session
+          .subscribeWhenReady(_channelFilter, (_) {})
+          .whenComplete(() => ready = true);
+
+      session.debugHandleMessage([
+        'CLOSED',
+        'l-1',
+        'rate-limited: quota exceeded; retry in 4s',
+      ]);
+      await Future<void>.delayed(Duration.zero);
+      expect(ready, isFalse);
+
+      retryTimers.single.fire();
+      await Future<void>.delayed(Duration.zero);
+      expect(ready, isFalse);
+      gateTimers.single.fire();
+      await Future<void>.delayed(Duration.zero);
+      expect(_reqs(socket).where((req) => req[1] == 'l-1'), hasLength(2));
+      expect(ready, isFalse);
+
+      session.debugHandleMessage(['EOSE', 'l-1']);
+      final unsubscribe = await subscribe;
+      expect(ready, isTrue);
+      unsubscribe();
+    },
+  );
+
+  test(
+    'subscribeWhenReady cancellation removes the pending subscription',
+    () async {
+      final timers = <_ManualTimer>[];
+      final socket = _RecordingRelaySocket();
+      final session = RelaySessionNotifier(
+        retryTimerFactory: (duration, callback) {
+          final timer = _ManualTimer(duration, callback);
+          timers.add(timer);
+          return timer;
+        },
+      );
+      session.debugAttachSocketForTest(socket);
+      final cancelled = Completer<void>();
+      final subscribe = session.subscribeWhenReady(
+        _channelFilter,
+        (_) {},
+        cancelled: cancelled.future,
+      );
+      session.debugHandleMessage(['CLOSED', 'l-1', 'error: relay overloaded']);
+
+      cancelled.complete();
+      await expectLater(
+        subscribe,
+        throwsA(isA<RelaySubscriptionCancelledException>()),
+      );
+      expect(timers.single.isActive, isFalse);
+      expect(socket.messages, contains(equals(<dynamic>['CLOSE', 'l-1'])));
+
+      timers.single.fire();
+      await Future<void>.delayed(Duration.zero);
+      expect(_reqs(socket).where((req) => req[1] == 'l-1'), hasLength(1));
+    },
+  );
+
+  test('dispose fails a pending subscribeWhenReady', () async {
+    final session = RelaySessionNotifier();
+    final subscribe = session.subscribeWhenReady(_channelFilter, (_) {});
+
+    session.debugDispose();
+
+    await expectLater(
+      subscribe,
+      throwsA(isA<RelaySubscriptionCancelledException>()),
+    );
+  });
+
   test('CLOSED retries back off and reset after EOSE', () async {
     final timers = <_ManualTimer>[];
     final socket = _RecordingRelaySocket();