diff --git a/mobile/lib/features/channels/channel_directory.dart b/mobile/lib/features/channels/channel_directory.dart index c8c5b945a2..d45a8db731 100644 --- a/mobile/lib/features/channels/channel_directory.dart +++ b/mobile/lib/features/channels/channel_directory.dart @@ -136,8 +136,40 @@ class _StaleChannelRefresh implements Exception { /// This token is captured once at the start of every ordinary, directory, and /// reconnect refresh. It is re-checked after each relay await, so an older /// request cannot regain ownership by reaching subscription setup last. -class _ChannelRefreshFence { +/// What a directory load checks before each write: the scope it serves. +abstract interface class _DirectoryLoadFence { + /// Relay-and-identity scope that started the load. + String get scope; + + /// Throws [_StaleChannelRefresh] once the load must write nothing. + void ensureCurrent(); +} + +/// Fences a directory-only load to its scope alone. +/// +/// Unlike [_ChannelRefreshFence] it holds no refresh generation: a directory +/// read does not depend on memberships, so it neither retires an in-flight +/// list refresh nor is retired by one. A community or identity switch, or +/// disposal, still retires it, so its response never reaches another scope. +class _DirectoryOnlyFence implements _DirectoryLoadFence { + @override + final String scope; + + final _ChannelRefreshCoordinator _coordinator; + + _DirectoryOnlyFence(this._coordinator, this.scope); + + @override + void ensureCurrent() { + if (!_coordinator.isMounted() || scope != _coordinator.currentScope()) { + throw const _StaleChannelRefresh(); + } + } +} + +class _ChannelRefreshFence implements _DirectoryLoadFence { /// Relay-and-identity scope that started the refresh. + @override final String scope; final _ChannelRefreshCoordinator _coordinator; @@ -155,6 +187,7 @@ class _ChannelRefreshFence { /// /// Call after every await and immediately before every write to metadata, /// cache, load status, subscriptions or provider state. + @override void ensureCurrent() { if (!isCurrent) throw const _StaleChannelRefresh(); } @@ -390,6 +423,16 @@ class _ChannelRefreshCoordinator { /// Generation of the most recently issued or retired request. int get generation => _generation; + // Directory loads are ordered among themselves: only the newest one writes + // the directory or its status. A directory-only load is not retired by a + // list refresh, so [beginRefresh] must leave its loading status alone. + int _directoryGeneration = 0; + int? _directoryOnlyGeneration; + + /// Starts a directory-only load for the active scope. + _DirectoryOnlyFence beginDirectoryOnly() => + _DirectoryOnlyFence(this, currentScope()); + _ChannelRefreshCoordinator({ required this.isMounted, required this.currentScope, @@ -416,6 +459,16 @@ class _ChannelRefreshCoordinator { /// wire cannot be written into the new scope. void retireInFlight() => _generation++; + /// Retires every in-flight request, directory-only loads included, for a + /// relay or identity change. A directory-only load survives ordinary list + /// refreshes, but not a scope change: its loading status must not look + /// owned when the user returns to the old scope. + void retireScope() { + _generation++; + _directoryGeneration++; + _directoryOnlyGeneration = null; + } + /// Starts a fenced refresh without issuing the directory query. /// /// Used by callers that must carry the scope across later awaits even when @@ -433,7 +486,8 @@ class _ChannelRefreshCoordinator { fence.ensureCurrent(); if (!fetchesDirectory) { final status = loadStatus(); - if (status.isLoading(scope)) { + if (status.isLoading(scope) && + _directoryOnlyGeneration != _directoryGeneration) { // This refresh takes ownership of the shared generation, so the // in-flight directory response will be discarded. Mark it idle, not // error: nothing failed, so a caller may load it again, and search @@ -456,18 +510,25 @@ class _ChannelRefreshCoordinator { /// Fetches the directory under [fence], or throws if the fence is retired. /// /// Returns null when the request failed inside the current scope, which means - /// "retain the cached discovery". The fence is re-checked after the await and - /// before every write, on both the success and the failure path. + /// "retain the cached discovery", or when a newer directory load started, + /// which then owns the directory and its status. The fence is re-checked + /// after the await and before every write, on both the success and the + /// failure path. Future?> loadDirectory( RelaySessionNotifier session, - _ChannelRefreshFence fence, + _DirectoryLoadFence fence, ) async { + final directoryGeneration = ++_directoryGeneration; + _directoryOnlyGeneration = fence is _DirectoryOnlyFence + ? directoryGeneration + : null; loadStatus().markLoading(fence.scope); final List metas; try { metas = await _fetchChannelDirectoryMetas(session); } catch (error, stackTrace) { fence.ensureCurrent(); + if (directoryGeneration != _directoryGeneration) return null; loadStatus().markError(fence.scope); debugPrint( '[ChannelsNotifier] channel directory refresh failed; retaining ' @@ -476,6 +537,7 @@ class _ChannelRefreshCoordinator { return null; } fence.ensureCurrent(); + if (directoryGeneration != _directoryGeneration) return null; loadStatus().markLoaded(fence.scope); return metas; } @@ -621,3 +683,100 @@ extension ObservedThreadActivity on ChannelsNotifier { } } } + +/// Directory-only loads and the list edits they share with full refreshes. +/// +/// Lives in this part file because `channels_provider.dart` sits against the +/// repository-wide 1200-line file ceiling enforced by `just file-size-check`. +extension _ChannelsNotifierDirectory on ChannelsNotifier { + /// Downloads only the open-channel directory and merges it into the list + /// [current] returns once it lands, for a list that already has its + /// memberships. Joined channels keep their entries; live updates keep them + /// current. Returns null when nothing should be written: the load failed + /// (its status says so), a newer directory load replaced it, the scope + /// changed, or there is no list to merge into. + Future?> _loadDirectoryOnly( + List? Function() current, + ) async { + final fence = _refreshCoordinator.beginDirectoryOnly(); + try { + final metas = await _refreshCoordinator.loadDirectory( + _lifecycleRef.read(relaySessionProvider.notifier), + fence, + ); + if (metas == null) return null; + _directoryMetas = metas; + _directoryMetasScope = fence.scope; + } on _StaleChannelRefresh { + return null; + } + final previous = current(); + if (previous == null) return null; + final channels = [...previous]; + _replaceDirectoryChannels(channels); + _sortChannels(channels); + _invalidateChangedDetails(channels, previous); + return channels; + } + + /// Rebuilds every channel in [channels] the user is not a member of from + /// [_directoryMetas], with a full refresh's visibility rules: the latest + /// revision of each open, non-DM channel. Joined channels are untouched. + void _replaceDirectoryChannels(List channels) { + final previous = { + for (final channel in channels) + if (!channel.isMember) channel.id: channel, + }; + channels.removeWhere((channel) => !channel.isMember); + final memberIds = {for (final channel in channels) channel.id}; + for (final event in _latestChannelEvents(_directoryMetas, kind: 39000)) { + final data = ChannelData.fromEvent(event); + if (memberIds.contains(data.id) || + data.visibility == 'private' || + data.channelType == 'dm') { + continue; + } + final channel = _channelFromMeta(event, data, isMember: false); + final lastMessageAt = previous[data.id]?.lastMessageAt; + channels.add( + lastMessageAt == null + ? channel + : channel.copyWith(lastMessageAt: lastMessageAt), + ); + } + } + + /// Invalidates the details of channels whose identity or archive state + /// changed from [previous], so opening one does not show stale Solo or + /// archive state. Returns [previous] by id. + Map _invalidateChangedDetails( + List channels, + List? previous, + ) { + final prevById = { + for (final c in previous ?? const []) c.id: c, + }; + for (final channel in channels) { + final prev = prevById[channel.id]; + if (prev == null || + prev.isArchived != channel.isArchived || + prev.channelType != channel.channelType || + prev.visibility != channel.visibility || + prev.createdBy != channel.createdBy || + !listEquals(prev.labels, channel.labels)) { + _lifecycleRef.invalidate(channelDetailsProvider(channel.id)); + } + } + return prevById; + } +} + +/// Orders the list by channel type, then case-insensitively by name to match +/// desktop's `localeCompare` ordering. +void _sortChannels(List channels) => channels.sort((left, right) { + final typeOrder = + (_channelTypeOrder[left.channelType] ?? 99) - + (_channelTypeOrder[right.channelType] ?? 99); + if (typeOrder != 0) return typeOrder; + return left.name.toLowerCase().compareTo(right.name.toLowerCase()); +}); diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 0c9e9a9c6b..089e6e5c0a 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -227,7 +227,7 @@ class ChannelsNotifier extends AsyncNotifier> { _confirmedJoins.clear(); // Retire any in-flight directory request: its response describes the // previous relay or identity and must not reach this scope's state. - _refreshCoordinator.retireInFlight(); + _refreshCoordinator.retireScope(); } final snapshot = (_hasLoaded ? state.value : null) ?? @@ -429,6 +429,7 @@ class ChannelsNotifier extends AsyncNotifier> { _directoryMetasScope = fence.scope; } } + final directoryMetas = _directoryMetas; // Merge and dedupe by `d` tag. Kind:39000 is parameterized-replaceable, // but stale revisions from before the relay's d_tag backfill can linger. @@ -579,14 +580,11 @@ class ChannelsNotifier extends AsyncNotifier> { } } - channels.sort((left, right) { - final typeOrder = - (_channelTypeOrder[left.channelType] ?? 99) - - (_channelTypeOrder[right.channelType] ?? 99); - if (typeOrder != 0) return typeOrder; - // Case-insensitive to match desktop's `localeCompare` ordering. - return left.name.toLowerCase().compareTo(right.name.toLowerCase()); - }); + // A directory-only load that landed during the awaits above is newer. + if (!identical(directoryMetas, _directoryMetas)) { + _replaceDirectoryChannels(channels); + } + _sortChannels(channels); // Details are a separate cache. Invalidate changes that would otherwise // overwrite fresh Solo identity or archive state when opening a channel. @@ -596,20 +594,7 @@ class ChannelsNotifier extends AsyncNotifier> { // after the await. fence.ensureCurrent(); - final prevById = { - for (final c in state.value ?? const []) c.id: c, - }; - for (final channel in channels) { - final prev = prevById[channel.id]; - if (prev == null || - prev.isArchived != channel.isArchived || - prev.channelType != channel.channelType || - prev.visibility != channel.visibility || - prev.createdBy != channel.createdBy || - !listEquals(prev.labels, channel.labels)) { - ref.invalidate(channelDetailsProvider(channel.id)); - } - } + final prevById = _invalidateChangedDetails(channels, state.value); // Layer live delivery onto the already-built snapshot without making EOSE // part of the provider's readiness path. Every refresh queues a generation; @@ -1080,11 +1065,14 @@ class ChannelsNotifier extends AsyncNotifier> { fresh)) { return; } - await retryDirectory(); + await retryDirectory(directoryOnly: true); } /// Retries channel discovery while retaining the current channel list. - Future retryDirectory() async { + /// + /// With [directoryOnly], a list that has loaded its memberships reads only + /// the directory again; joined channels keep their entries. + Future retryDirectory({bool directoryOnly = false}) async { final directoryStatus = ref.read( channelDirectoryLoadStatusProvider.notifier, ); @@ -1115,6 +1103,15 @@ class ChannelsNotifier extends AsyncNotifier> { } if (!ref.mounted || generation != _buildGeneration) return; final previousChannels = state.value; + if (directoryOnly && _hasLoaded && previousChannels != null) { + // The list has its memberships, so only the directory is read again. + final saveSnapshot = _saveSnapshot; + final channels = await _loadDirectoryOnly(() => state.value); + if (channels == null) return; + state = AsyncData(channels); + saveSnapshot?.call(channels, _hiddenDmIds); + return; + } try { state = AsyncData( await _fetch( diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 2dd0595f04..fef1108e77 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -1908,6 +1908,282 @@ void main() { expect(session.metadataPageRequestCount, initialDirectoryRequests); }); + group('search directory load on a loaded list', () { + const channelC = '33333333-3333-4333-8333-333333333333'; + + Future loaded(_FakeRelaySession session) async { + final container = _buildContainer(session: session); + addTearDown(container.dispose); + await container.read(channelsProvider.future); + // Let the first load's detached unread catch-up finish. + await _settle(); + return container; + } + + List listedIds(ProviderContainer container) => container + .read(channelsProvider) + .requireValue + .map((channel) => channel.id) + .toList(); + + test('reads only the directory and keeps joined channels', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'discoverable'), + ], + ); + final container = await loaded(session); + final joined = container.read(channelsProvider).requireValue.single; + final memberships = session.membershipRequestCount; + final histories = session.historyFilters.length; + final batches = session.queryBatches.length; + + session.metadata = [ + _meta(id: _channelA, name: 'general'), + _meta( + id: _channelB, + name: 'discoverable', + visibility: 'private', + createdAt: 2, + ), + _meta(id: channelC, name: 'new-open'), + ]; + await container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + + // No roster, member metadata, hidden-DM, Huddle or message reads. + expect(session.membershipRequestCount, memberships); + expect(session.historyFilters, hasLength(histories)); + expect(session.queryBatches, hasLength(batches)); + expect(session.directoryQueryFilters, isNotEmpty); + expect(listedIds(container), unorderedEquals([_channelA, channelC])); + final listed = container.read(channelsProvider).requireValue; + expect(listed.firstWhere((c) => c.id == _channelA), same(joined)); + expect(listed.firstWhere((c) => c.id == channelC).isMember, isFalse); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loaded, + ); + }); + + test('a membership refresh during the load is kept', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'discoverable'), + _meta(id: _channelD, name: 'joined-elsewhere'), + ], + ); + final container = await loaded(session); + + session.pauseNextDirectoryQuery(); + final directory = container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + await session.nextDirectoryQueryStarted; + + // Joining on another device lands while the directory is in flight. + session.memberships = [ + _membership(_channelA, myPk), + _membership(_channelD, myPk), + ]; + await container.read(channelsProvider.notifier).refresh(); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loading, + ); + + session.resumePausedDirectoryQuery(); + await directory; + await _settle(); + + final listed = container.read(channelsProvider).requireValue; + expect( + listed.map((c) => c.id), + unorderedEquals([_channelA, _channelB, _channelD]), + ); + expect(listed.firstWhere((c) => c.id == _channelD).isMember, isTrue); + expect(listed.firstWhere((c) => c.id == _channelB).isMember, isFalse); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loaded, + ); + }); + + test( + 'a refresh that merged the directory before it landed keeps it', + () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'discoverable'), + ], + ); + final container = await loaded(session); + + // The refresh merges the directory it has (none) and then waits for + // its latest messages while the search load completes. + final gate = Completer(); + session.latestBatchGate = gate; + final refresh = container.read(channelsProvider.notifier).refresh(); + await _waitUntil(() => session.latestBatchGate == null); + await container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + expect(listedIds(container), unorderedEquals([_channelA, _channelB])); + + gate.complete(); + await refresh; + + expect(listedIds(container), unorderedEquals([_channelA, _channelB])); + }, + ); + + test('a community switch discards the response', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'community-a-open'), + ], + ); + final container = await loaded(session); + final communityB = channelDirectoryScope( + 'https://community-b.example', + container.read(myPubkeyProvider), + ); + + session.pauseNextDirectoryQuery(); + final directory = container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + await session.nextDirectoryQueryStarted; + + session.memberships = [_membership(_channelD, myPk)]; + session.metadata = [_meta(id: _channelD, name: 'community-b-general')]; + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://community-b.example'); + await container.read(channelsProvider.future); + + session.resumePausedDirectoryQuery(); + await directory; + await _settle(); + + expect(listedIds(container), [_channelD]); + expect( + container.read(channelDirectoryLoadStatusProvider).scope, + isNot(communityB), + ); + }); + + test( + 'returning to a community during a discarded load loads again', + () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'community-a-open'), + ], + ); + final container = await loaded(session); + final communityA = container.read(relayConfigProvider).baseUrl; + + session.pauseNextDirectoryQuery(); + final directory = container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + await session.nextDirectoryQueryStarted; + + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://community-b.example'); + await container.read(channelsProvider.future); + container + .read(relayConfigProvider.notifier) + .update(baseUrl: communityA); + await container.read(channelsProvider.future); + await _settle(); + final directoryQueries = session.directoryQueryFilters.length; + await container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + + expect( + session.directoryQueryFilters.length, + greaterThan(directoryQueries), + ); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loaded, + ); + + // The old request ending later leaves the newer load's status alone. + session.resumePausedDirectoryQuery(); + await directory; + await _settle(); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loaded, + ); + }, + ); + + test( + 'returning to a community after a discarded load loads again', + () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'community-a-open'), + ], + ); + final container = await loaded(session); + final communityA = container.read(relayConfigProvider).baseUrl; + + session.pauseNextDirectoryQuery(); + final directory = container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + await session.nextDirectoryQueryStarted; + + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://community-b.example'); + await container.read(channelsProvider.future); + session.resumePausedDirectoryQuery(); + await directory; + await _settle(); + + container + .read(relayConfigProvider.notifier) + .update(baseUrl: communityA); + await container.read(channelsProvider.future); + await _settle(); + final directoryQueries = session.directoryQueryFilters.length; + await container + .read(channelsProvider.notifier) + .ensureDirectoryLoaded(maxAge: Duration.zero); + + expect( + session.directoryQueryFilters.length, + greaterThan(directoryQueries), + ); + expect( + container.read(channelDirectoryLoadStatusProvider).status, + ChannelDirectoryLoadStatus.loaded, + ); + }, + ); + }); + test('membership refresh does not refetch a loaded directory', () async { final session = _FakeRelaySession( memberships: [_membership(_channelA, myPk)], @@ -3408,6 +3684,9 @@ class _FakeRelaySession extends RelaySessionNotifier { /// While set, member metadata reads wait for it to complete. Completer? metadataGate; + // Holds the next latest-message batch, which a list refresh awaits after it + // has merged the directory. + Completer? latestBatchGate; final List> queryBatches = []; Object? messageBatchError; Object? latestBatchError; @@ -3430,6 +3709,7 @@ class _FakeRelaySession extends RelaySessionNotifier { Completer? _pausedSubscribe; Completer? _subscribeStarted; Completer? _pausedDirectory; + bool _pauseClaimed = false; Completer? _directoryStarted; Completer? _pausedHiddenDm; Completer? _hiddenDmStarted; @@ -3752,11 +4032,15 @@ class _FakeRelaySession extends RelaySessionNotifier { // active when the response is released. final directorySnapshot = List.of(metadata); final paused = _pausedDirectory; - if (paused != null) { + // Only the first query after the pause waits; a later one meanwhile + // (a new scope's load) answers at once. + if (paused != null && !_pauseClaimed) { + _pauseClaimed = true; _directoryStarted!.complete(); await paused.future; _pausedDirectory = null; _directoryStarted = null; + _pauseClaimed = false; } if (directoryFailures > 0) { directoryFailures--; @@ -3790,6 +4074,11 @@ class _FakeRelaySession extends RelaySessionNotifier { final isUnreadCatchUp = filters.isNotEmpty && filters.every((filter) => filter.since != null); if (!isUnreadCatchUp && latestBatchError != null) throw latestBatchError!; + if (!isUnreadCatchUp) { + final gate = latestBatchGate; + latestBatchGate = null; + await gate?.future; + } final messageSnapshot = List.of(recentMessages); if (isUnreadCatchUp) { // Claim the parked slot so the refresh that follows the switch can run