diff --git a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift index bc6c62e013..5ea6fcb7c9 100644 --- a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift +++ b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift @@ -574,6 +574,130 @@ struct BuzzPushPresentationCacheTests { ) } + @Test("Retrying a partial batch prefix preserves newer state and original scope") + func partialBatchPrefixReplayIsIdempotent() throws { + let directory = try temporaryDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let store = BuzzPushPresentationCacheStore( + containerURL: directory, + now: { Date(timeIntervalSince1970: 1_700_000_100) } + ) + let relayPubkey = try pubkey(for: relayKey) + let member = try pubkey(for: profileKey) + let addedMember = try pubkey(for: otherRelayKey) + let profiles = try [profileKey, otherRelayKey].map { key in + try signedEvent(privateKey: key, createdAt: 100, kind: 0, content: #"{"name":"Original"}"#) + } + let metadata = try ["prefix", "suffix"].map { channel in + try signedEvent( + privateKey: relayKey, createdAt: 100, kind: 39_000, + tags: [["d", channel], ["name", "Original"], ["t", "stream"]] + ) + } + let memberships = try ["prefix", "suffix"].map { channel in + try signedEvent( + privateKey: relayKey, createdAt: 100, kind: 39_002, + tags: [["d", channel], ["p", member]] + ) + } + func deliver( + _ index: Int, community: String = "original", origin: String = "https://relay.example" + ) throws { + try store.updateProfiles( + communityID: community, relayOrigin: origin, + updates: [BuzzPushProfileCacheUpdate(event: profiles[index])] + ) + try store.updateChannels( + communityID: community, relayOrigin: origin, relayMetadataPubkey: relayPubkey, + metadataEvents: [metadata[index]], membershipEvents: [memberships[index]] + ) + } + + // Native storage accepts the prefix, but the caller loses its acknowledgement. + try deliver(0) + try deliver(0, community: "other-community") + try deliver(0, origin: "https://other-relay.example") + let beforeRetry = try loadSnapshot(directory) + let untouchedProfiles = beforeRetry.profiles.filter { + $0.communityID != "original" || $0.relayOrigin != "https://relay.example" + } + let untouchedChannels = beforeRetry.channels.filter { + $0.communityID != "original" || $0.relayOrigin != "https://relay.example" + } + let newerProfile = try signedEvent( + privateKey: profileKey, createdAt: 101, kind: 0, content: #"{"name":"Newest"}"# + ) + let newerMetadata = try signedEvent( + privateKey: relayKey, createdAt: 101, kind: 39_000, + tags: [["d", "prefix"], ["name", "Newest"], ["t", "forum"]] + ) + let newerMembership = try signedEvent( + privateKey: relayKey, createdAt: 101, kind: 39_002, + tags: [["d", "prefix"], ["p", member], ["p", addedMember]] + ) + try store.updateProfiles( + communityID: "original", relayOrigin: "https://relay.example", + updates: [BuzzPushProfileCacheUpdate(event: newerProfile)] + ) + try store.updateChannels( + communityID: "original", relayOrigin: "https://relay.example", + relayMetadataPubkey: relayPubkey, + metadataEvents: [newerMetadata], membershipEvents: [newerMembership] + ) + + // Replay exact events from the accepted prefix, then deliver the missing suffix. + try deliver(0) + try deliver(1) + let completed = try loadSnapshot(directory) + try deliver(0) + try deliver(1) + let replayed = try loadSnapshot(directory) + #expect(replayed.profiles == completed.profiles) + #expect(replayed.channels == completed.channels) + #expect(replayed.profiles.count == 4) + #expect(replayed.channels.count == 4) + #expect( + replayed.profiles.filter { + $0.communityID != "original" || $0.relayOrigin != "https://relay.example" + } == untouchedProfiles) + #expect( + replayed.channels.filter { + $0.communityID != "original" || $0.relayOrigin != "https://relay.example" + } == untouchedChannels) + let profile = try #require( + replayed.profile( + communityID: "original", relayOrigin: "https://relay.example", pubkey: member + )) + #expect(profile.eventID == newerProfile.id) + #expect(profile.displayName == "Newest") + let channel = try #require( + replayed.channel( + communityID: "original", relayOrigin: "https://relay.example", channelID: "prefix" + )) + #expect(channel.eventID == newerMetadata.id) + #expect(channel.displayName == "Newest") + #expect(channel.channelType == "forum") + #expect(channel.membershipEventID == newerMembership.id) + #expect(channel.memberCount == 2) + #expect( + channel.memberDigests + == [member, addedMember].map { + BuzzPushPresentationIdentity.channelMember( + communityID: "original", channelID: "prefix", pubkey: $0 + ) + }.sorted()) + #expect( + replayed.profile( + communityID: "original", relayOrigin: "https://relay.example", pubkey: addedMember + )?.eventID == profiles[1].id) + let suffix = try #require( + replayed.channel( + communityID: "original", relayOrigin: "https://relay.example", channelID: "suffix" + )) + #expect(suffix.eventID == metadata[1].id) + #expect(suffix.membershipEventID == memberships[1].id) + } + @Test("Channel authority rotation clears membership signed by the old authority") func channelAuthorityRotationClearsMembership() throws { let directory = try temporaryDirectory() diff --git a/mobile/lib/features/channels/channel_directory.dart b/mobile/lib/features/channels/channel_directory.dart index 1282e5f1e1..dcff0a3569 100644 --- a/mobile/lib/features/channels/channel_directory.dart +++ b/mobile/lib/features/channels/channel_directory.dart @@ -82,22 +82,26 @@ final channelDirectoryLoadStatusProvider = Future> _fetchChannelMemberships( RelaySessionNotifier session, - String pubkey, -) => _fetchPaginatedChannelEvents( + String pubkey, { + void Function()? ensureCurrent, +}) => _fetchPaginatedChannelEvents( session, kind: 39002, tags: { '#p': [pubkey], }, operation: 'Channel memberships', + ensureCurrent: ensureCurrent, ); Future> _fetchChannelDirectoryMetas( - RelaySessionNotifier session, -) => _fetchPaginatedChannelEvents( + RelaySessionNotifier session, { + void Function()? ensureCurrent, +}) => _fetchPaginatedChannelEvents( session, kind: 39000, operation: 'Channel directory', + ensureCurrent: ensureCurrent, ); /// Thrown when a channel-list request is retired before it settles. @@ -391,12 +395,14 @@ Future> _fetchPaginatedChannelEvents( required int kind, required String operation, Map> tags = const {}, + void Function()? ensureCurrent, }) async { final events = []; final seenEventIds = {}; int? until; String? beforeId; for (var pageIndex = 0; pageIndex < _maxChannelDirectoryPages; pageIndex++) { + ensureCurrent?.call(); final page = await session.queryRelay([ NostrFilter( kinds: [kind], @@ -406,6 +412,7 @@ Future> _fetchPaginatedChannelEvents( extensions: {'before_id': ?beforeId}, ), ]); + ensureCurrent?.call(); if (page.isEmpty) break; var madeProgress = false; for (final event in page) { diff --git a/mobile/lib/features/channels/channel_push_cache.dart b/mobile/lib/features/channels/channel_push_cache.dart new file mode 100644 index 0000000000..4576f1b3d9 --- /dev/null +++ b/mobile/lib/features/channels/channel_push_cache.dart @@ -0,0 +1,106 @@ +part of 'channels_provider.dart'; + +extension _ChannelPushCache on ChannelsNotifier { + Future _exportPushCache( + String? communityID, + List metadata, + List membership, + ) async { + if (defaultTargetPlatform != TargetPlatform.iOS || communityID == null) { + return; + } + if (_pushCacheExporting) { + // A refresh can contain different channels. Do not replace its predecessor + // or retain another raw batch: refetch the complete scope after it drains. + _pushCacheDirty = true; + return; + } + _pushCacheExporting = true; + try { + await _pushExport.export( + () => cacheBuzzPushChannelEvents(communityID, metadata, membership), + ); + // A failed operation is terminal only for its own input. A refresh that + // arrived meanwhile still owns an independent dirty obligation. + while (_lifecycleRef.mounted && _pushCacheDirty) { + _pushCacheDirty = false; + final session = _lifecycleRef.read(relaySessionProvider.notifier); + final community = _lifecycleRef.read(activeCommunityProvider).value?.id; + final relay = _lifecycleRef.read(relayConfigProvider).baseUrl; + final pubkey = _lifecycleRef.read(myPubkeyProvider); + if (community == null || pubkey == null) return; + bool current() => + _lifecycleRef.mounted && + _lifecycleRef.read(activeCommunityProvider).value?.id == + community && + _lifecycleRef.read(relayConfigProvider).baseUrl == relay && + _lifecycleRef.read(myPubkeyProvider) == pubkey && + identical( + _lifecycleRef.read(relaySessionProvider.notifier), + session, + ); + + void ensureCurrent() { + if (!current()) throw const _StaleChannelRefresh(); + } + + // This fetch has no UI writes or channel-refresh generation changes. + // New ordinary refreshes can proceed and mark this snapshot dirty again. + // Start once. Admission retries await this same snapshot rather than + // repeating relay reads while the native export queue remains full. + final snapshot = () async { + try { + if (!current()) return null; + final members = await _fetchChannelMemberships( + session, + pubkey, + ensureCurrent: ensureCurrent, + ); + if (!current()) return null; + final ids = members + .map((event) => event.getTagValue('d')) + .whereType() + .toSet() + .toList(); + final memberMetadata = ids.isEmpty + ? const [] + : await session.fetchHistory(NostrFilters.channelMetadata(ids)); + if (!current()) return null; + final memberSnapshots = ids.isEmpty + ? const [] + : await session.fetchHistory( + NostrFilter( + kinds: const [39002], + tags: {'#d': ids}, + limit: ids.length, + ), + ); + if (!current()) return null; + return ( + metadata: memberMetadata, + membership: [...members, ...memberSnapshots], + ); + } catch (_) { + if (current()) rethrow; + // Retired fetches must neither write nor report an error in the + // replacement scope. Its ordinary refresh owns any new dirty work. + } + return null; + }(); + await _pushExport.export(() async { + final events = await snapshot; + if (events == null || !current()) return; + await cacheBuzzPushChannelEvents( + community, + events.metadata, + events.membership, + ); + }); + // Only new producer input can set dirty again; failure alone never + // schedules another attempt after recovery's bounded retries. + } + } finally { + _pushCacheExporting = false; + } + } +} diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 8e724cc3a8..c5a77ba930 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -3,10 +3,12 @@ import 'dart:convert'; import 'dart:math'; import 'package:flutter/widgets.dart'; +import 'package:flutter/foundation.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/community/community_provider.dart'; import '../../shared/push/push_presentation_cache.dart'; +import '../../shared/push/push_presentation_export_recovery.dart'; import '../../shared/relay/relay.dart'; import '../../shared/theme/theme_provider.dart'; import '../../shared/utils/string_utils.dart'; @@ -22,6 +24,7 @@ import 'unread_badge/observed_unread_event.dart'; import 'unread_badge/should_notify_for_event.dart'; part 'channel_directory.dart'; +part 'channel_push_cache.dart'; part 'channel_member_snapshots.dart'; part 'channels_provider_lifecycle.dart'; @@ -41,6 +44,10 @@ const _authoredRootIdsPrefix = 'buzz-thread-authored.v1'; /// for any visible channel event kind. Chunks stay within the relay's explicit /// channel cap and incoming events bump `lastMessageAt` for their channel. class ChannelsNotifier extends AsyncNotifier> { + final _pushExport = PushPresentationExportRecovery(); + bool _pushCacheExporting = false; + bool _pushCacheDirty = false; + static const _backstopInterval = Duration(seconds: 60); final Map _liveSubscriptionsByChunk = {}; @@ -300,7 +307,7 @@ class ChannelsNotifier extends AsyncNotifier> { // linkage validation and member-count hydration. if (memberEvents.isNotEmpty) _cacheMemberSnapshots(memberEvents); unawaited( - cacheBuzzPushChannelEvents(communityID, dedupedMetas, [ + _exportPushCache(communityID, dedupedMetas, [ ...memberships, ...memberEvents, ]), diff --git a/mobile/lib/shared/profile/user_cache_provider.dart b/mobile/lib/shared/profile/user_cache_provider.dart index 3240e00cab..4a034315d0 100644 --- a/mobile/lib/shared/profile/user_cache_provider.dart +++ b/mobile/lib/shared/profile/user_cache_provider.dart @@ -4,6 +4,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../community/community_provider.dart'; import '../push/push_presentation_cache.dart'; +import '../push/push_presentation_export_recovery.dart'; import '../relay/relay.dart'; import 'user_profile.dart'; import 'profile_event_parser.dart'; @@ -11,9 +12,14 @@ import 'profile_event_parser.dart'; /// In-memory cache of user profiles, fetched in batches from the relay. /// /// Lookups requested via [get] or [preload] are coalesced into a single -/// kind:0 batch query (NIP-01 `authors` filter) every 50ms. +/// kind:0 batch query (NIP-01 `authors` filter) every 50ms. Larger requests +/// drain sequentially in pages bounded by the relay response limit. class UserCacheNotifier extends Notifier> { + // The relay clamps each history response to DEFAULT_MAX_PAGE_LIMIT (1000). + static const _maximumProfilesPerQuery = 1000; + final Set _pending = {}; + final _pushExport = PushPresentationExportRecovery(); final Map _profileEventOrders = {}; int _generation = 0; bool _flushInFlight = false; @@ -81,11 +87,15 @@ class UserCacheNotifier extends Notifier> { final generation = _generation; try { final session = ref.read(relaySessionProvider.notifier); - final events = await session.fetchHistory( - NostrFilters.profilesBatch(normalized), - ); - if (!_isCurrent(generation)) return false; - return await _verifyAndMerge(events, generation); + for (final batch in _profileQueryBatches(normalized)) { + if (!_isCurrent(generation)) return false; + final events = await session.fetchHistory( + NostrFilters.profilesBatch(batch), + ); + if (!_isCurrent(generation)) return false; + if (!await _verifyAndMerge(events, generation)) return false; + } + return true; } catch (_) { return false; } @@ -123,25 +133,50 @@ class UserCacheNotifier extends Notifier> { try { final communityID = ref.read(activeCommunityProvider).value?.id; final session = ref.read(relaySessionProvider.notifier); - final events = await session.fetchHistory( - NostrFilters.profilesBatch(pubkeys), - ); - - if (!_isCurrent(generation)) return; - if (!await _verifyAndMerge(events, generation)) return; - if (communityID != null) { - unawaited(cacheBuzzPushProfileEvents(communityID, events)); + var remaining = pubkeys.length; + for (final batch in _profileQueryBatches(pubkeys)) { + if (!_isCurrent(generation)) return; + final events = await session.fetchHistory( + NostrFilters.profilesBatch(batch), + ); + if (!_isCurrent(generation)) return; + if (!await _verifyAndMerge(events, generation)) return; + remaining -= batch.length; + if (remaining == 0) { + // All requested pages are ready. Native export does not gate readers. + succeeded = true; + completer?.complete(true); + } + if (communityID != null) { + // Drain each page before fetching another, retaining at most one raw + // response while later requests coalesce as pubkeys in _pending. + await _pushExport.export( + () => cacheBuzzPushProfileEvents(communityID, events), + ); + } } - succeeded = true; } catch (_) { // Silently fail — non-gating callers will just show pubkeys. } finally { - completer?.complete(succeeded); + if (completer != null && !completer.isCompleted) { + completer.complete(succeeded); + } _flushInFlight = false; if (ref.mounted && _pending.isNotEmpty) _scheduleBatch(); } } + Iterable> _profileQueryBatches(List pubkeys) sync* { + for ( + var start = 0; + start < pubkeys.length; + start += _maximumProfilesPerQuery + ) { + final end = start + _maximumProfilesPerQuery; + yield pubkeys.sublist(start, end < pubkeys.length ? end : pubkeys.length); + } + } + void _scheduleBatch() { if (_flushInFlight) return; _batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending); diff --git a/mobile/lib/shared/push/push_presentation_cache.dart b/mobile/lib/shared/push/push_presentation_cache.dart index 82cc78a89d..99af19d864 100644 --- a/mobile/lib/shared/push/push_presentation_cache.dart +++ b/mobile/lib/shared/push/push_presentation_cache.dart @@ -11,6 +11,19 @@ const _pushPresentationChannel = MethodChannel('buzz/push'); const _maximumAvatarSourceBytes = 512 * 1024; const _maximumAvatarPNGBytes = 64 * 1024; Future _avatarEncodeTail = Future.value(); +Future _presentationExportTail = Future.value(); +const _maximumPresentationExports = 8; +// Match BuzzPushPresentationCacheStore's per-write admission limits. +const _maximumProfilesPerWrite = 256; +const _maximumChannelsPerWrite = 512; +int _outstandingPresentationExports = 0; + +/// Temporary admission failure; detached producers must retain and retry input. +class PushPresentationExportQueueFull extends StateError { + /// Creates a saturation error for the shared eight-export budget. + PushPresentationExportQueueFull() + : super('Push presentation export queue is full (8 outstanding exports)'); +} /// The latest best-effort App Group presentation-cache failure. final pushPresentationCacheError = ValueNotifier(null); @@ -34,6 +47,9 @@ bool isVerifiedPushPresentationEvent(NostrEvent event) { } /// Exports raw verified kind-0 events. Native code verifies them again before storage. +/// Fails with [StateError] when eight exports are already outstanding. +/// Native persistence failures retry with bounded backoff, then propagate and +/// stop any remaining chunks. Other native errors propagate immediately. Future cacheBuzzPushProfileEvents( String communityID, Iterable events, @@ -41,20 +57,30 @@ Future cacheBuzzPushProfileEvents( if (defaultTargetPlatform != TargetPlatform.iOS || communityID.isEmpty) { return; } - final verified = _newestVerifiedEvents( - events, - kind: 0, - scope: (event) => event.pubkey.toLowerCase(), - ).values.toList(); - if (verified.isEmpty) return; - await _invokeBestEffort({ - 'section': 'profiles', - 'communityId': communityID, - 'events': [for (final event in verified) event.toJson()], + final batch = events.toList(growable: false); + if (batch.isEmpty) return; + await _serializePresentationExport(() async { + final verified = await compute( + _selectPushProfileEvents, + batch, + debugLabel: 'buzz-push-profile-cache', + ); + if (verified.isEmpty) return; + final retryBudget = _NativeWriteRetryBudget(); + for (final chunk in _boundedChunks(verified, _maximumProfilesPerWrite)) { + await _invokeVerifiedChunk({ + 'section': 'profiles', + 'communityId': communityID, + 'events': [for (final event in chunk) event.toJson()], + }, retryBudget); + } }); } /// Exports verified channel metadata and membership for native authority checks. +/// Fails with [StateError] when eight exports are already outstanding. +/// Native persistence failures retry with bounded backoff, then propagate and +/// stop any remaining chunks. Other native errors propagate immediately. Future cacheBuzzPushChannelEvents( String? communityID, Iterable metadataEvents, @@ -65,20 +91,89 @@ Future cacheBuzzPushChannelEvents( communityID.isEmpty) { return; } - final batch = selectPushChannelEvents(metadataEvents, membershipEvents); - final verifiedMetadata = batch.metadata; - final verifiedMembership = batch.membership; - if (verifiedMetadata.isEmpty && verifiedMembership.isEmpty) return; - await _invokeBestEffort({ - 'section': 'channels', - 'communityId': communityID, - 'metadataEvents': [for (final event in verifiedMetadata) event.toJson()], - 'membershipEvents': [ - for (final event in verifiedMembership) event.toJson(), - ], + final batch = ( + metadata: metadataEvents.toList(growable: false), + membership: membershipEvents.toList(growable: false), + ); + if (batch.metadata.isEmpty && batch.membership.isEmpty) return; + await _serializePresentationExport(() async { + final verified = await compute( + _selectPushChannelBatch, + batch, + debugLabel: 'buzz-push-channel-cache', + ); + if (verified.metadata.isEmpty && verified.membership.isEmpty) return; + final metadata = { + for (final event in verified.metadata) event.getTagValue('d')!: event, + }; + final membership = { + for (final event in verified.membership) event.getTagValue('d')!: event, + }; + final channelIDs = {...metadata.keys, ...membership.keys}.toList(); + final retryBudget = _NativeWriteRetryBudget(); + // Keep each channel's metadata and membership in the same native write. + // All chunks retain this export's FIFO slot until handoff is complete. + for (final ids in _boundedChunks(channelIDs, _maximumChannelsPerWrite)) { + await _invokeVerifiedChunk({ + 'section': 'channels', + 'communityId': communityID, + 'metadataEvents': [ + for (final id in ids) + if (metadata[id] case final event?) event.toJson(), + ], + 'membershipEvents': [ + for (final id in ids) + if (membership[id] case final event?) event.toJson(), + ], + }, retryBudget); + } }); } +Iterable> _boundedChunks(List values, int maximum) sync* { + for (var start = 0; start < values.length; start += maximum) { + final end = start + maximum; + yield values.sublist(start, end < values.length ? end : values.length); + } +} + +// Share one worker slot across profile/channel exports and retain FIFO native +// handoff, even across community changes. The two producers are coalesced profile +// fetches and channel refreshes. Eight outstanding batches allow a short burst +// while bounding retained batch count; individual batch sizes remain caller-owned. +// Saturation fails explicitly rather than acknowledging an export we cannot retain. +Future _serializePresentationExport( + Future Function() export, +) async { + if (_outstandingPresentationExports >= _maximumPresentationExports) { + throw PushPresentationExportQueueFull(); + } + _outstandingPresentationExports++; + final previous = _presentationExportTail; + final release = Completer(); + _presentationExportTail = release.future; + await previous; + try { + await export(); + } finally { + // Keep the queue usable while propagating a worker failure to its caller. + _outstandingPresentationExports--; + release.complete(); + } +} + +List _selectPushProfileEvents(List events) => + _newestVerifiedEvents( + events, + kind: 0, + scope: (event) => event.pubkey.toLowerCase(), + ).values.toList(); + +({List metadata, List membership}) +_selectPushChannelBatch( + ({List metadata, List membership}) batch, +) => selectPushChannelEvents(batch.metadata, batch.membership); + /// Selects the newest paired verified channel metadata and membership events. @visibleForTesting ({List metadata, List membership}) @@ -162,18 +257,52 @@ Future cacheBuzzPushAvatarFromLoadedBytes( try { final png = await _boundedAvatarPNG(sourceBytes); if (png == null) return; - await _invokeBestEffort({ + await _invokeSnapshot({ 'section': 'avatar', 'communityId': communityID, 'sourceUrl': sourceURL, 'png': png, - }); + }, bestEffort: true); } finally { release.complete(); } } -Future _invokeBestEffort(Map arguments) async { +// One bounded backoff budget for the entire export, not one per chunk. +class _NativeWriteRetryBudget { + static const _delays = [250, 500, 1000, 2000, 4000]; + int _used = 0; + + Duration? takeDelay() => + _used == _delays.length ? null : Duration(milliseconds: _delays[_used++]); +} + +// Retry only native persistence failures, retaining the verified payload and +// the surrounding export's FIFO slot. Never repeat relay reads or verification. +Future _invokeVerifiedChunk( + Map arguments, + _NativeWriteRetryBudget budget, +) async { + final retryableCode = arguments['section'] == 'profiles' + ? 'profile_cache_failed' + : 'channel_cache_failed'; + while (true) { + try { + await _invokeSnapshot(arguments); + return; + } on PlatformException catch (error) { + if (error.code != retryableCode) rethrow; + final delay = budget.takeDelay(); + if (delay == null) rethrow; + await Future.delayed(delay); + } + } +} + +Future _invokeSnapshot( + Map arguments, { + bool bestEffort = false, +}) async { try { await _pushPresentationChannel.invokeMethod( 'syncPushSnapshot', @@ -186,6 +315,9 @@ Future _invokeBestEffort(Map arguments) async { pushPresentationCacheError.value = error.toString(); debugPrint('Push presentation cache update failed: $error'); debugPrintStack(stackTrace: stackTrace); + // Event exports must stop at the failed chunk and report failure to their + // owner. Avatar updates preserve their existing detached best-effort policy. + if (!bestEffort) rethrow; } } diff --git a/mobile/lib/shared/push/push_presentation_export_recovery.dart b/mobile/lib/shared/push/push_presentation_export_recovery.dart new file mode 100644 index 0000000000..764697c39c --- /dev/null +++ b/mobile/lib/shared/push/push_presentation_export_recovery.dart @@ -0,0 +1,49 @@ +import 'package:flutter/foundation.dart'; + +import 'push_presentation_cache.dart'; + +/// Latest terminal export failure, retained across unrelated successful writes. +/// +/// Unlike [pushPresentationCacheError], this records an export that could not +/// reach native storage. Retries are bounded and do not guarantee delivery. +final pushPresentationExportError = ValueNotifier(null); + +/// Retries one export while its producer applies backpressure to new input. +/// +/// Queue saturation retries the exact operation five times over 7.75 seconds. +/// Exhausted retries or a worker failure return false and record a terminal +/// error. Producers must serialize calls and retain dirty input upstream; this +/// helper does not queue, replace, or merge unverified event batches. +class PushPresentationExportRecovery { + /// Exports once, recovering transient saturation without detached errors. + Future export(Future Function() operation) async { + try { + await operation(); + return true; + } on PushPresentationExportQueueFull { + // The producer retains this operation until recovery completes. + } catch (error, stack) { + return _failed(error, stack); + } + + for (final milliseconds in [250, 500, 1000, 2000, 4000]) { + await Future.delayed(Duration(milliseconds: milliseconds)); + try { + await operation(); + return true; + } on PushPresentationExportQueueFull catch (error, stack) { + if (milliseconds == 4000) return _failed(error, stack); + } catch (error, stack) { + return _failed(error, stack); + } + } + return false; + } + + bool _failed(Object error, StackTrace stack) { + pushPresentationExportError.value = error.toString(); + debugPrint('Push presentation export could not be delivered: $error'); + debugPrintStack(stackTrace: stack); + return false; + } +} diff --git a/mobile/test/shared/profile/user_cache_batch_limit_test.dart b/mobile/test/shared/profile/user_cache_batch_limit_test.dart new file mode 100644 index 0000000000..452a8df656 --- /dev/null +++ b/mobile/test/shared/profile/user_cache_batch_limit_test.dart @@ -0,0 +1,183 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:buzz/shared/community/community.dart'; +import 'package:buzz/shared/community/community_provider.dart'; +import 'package:buzz/shared/profile/user_cache_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter/foundation.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + const bridge = MethodChannel('buzz/push'); + final messenger = + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + + test( + 'pending profiles drain in bounded queries after delayed export', + () async { + debugDefaultTargetPlatformOverride = TargetPlatform.iOS; + final entered = Completer(); + final release = Completer(); + messenger.setMockMethodCallHandler(bridge, (_) async { + if (!entered.isCompleted) { + entered.complete(); + await release.future; + } + return null; + }); + addTearDown(() { + messenger.setMockMethodCallHandler(bridge, null); + debugDefaultTargetPlatformOverride = null; + }); + final signed = NostrEvent.fromJson( + nostr.Event.from( + kind: 0, + createdAt: 1, + content: '{"name":"First"}', + tags: [], + secretKey: '1'.padLeft(64, '0'), + ).toMap(), + ); + final session = _ClampingSession(first: signed); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + activeCommunityProvider.overrideWith( + (ref) async => Community( + id: 'original', + name: 'Synthetic', + relayUrl: 'https://example.invalid', + addedAt: DateTime(2026), + ), + ), + ], + ); + addTearDown(container.dispose); + await container.read(activeCommunityProvider.future); + final cache = container.read(userCacheProvider.notifier); + expect(await cache.preload([signed.pubkey]), isTrue); + await entered.future.timeout(const Duration(seconds: 5)); + final keys = List.generate(1001, (i) => 'profile-$i'); + var completed = false; + final pending = cache.preload(keys).then((result) { + completed = true; + return result; + }); + final duplicate = cache.preload(keys); + try { + await Future.delayed(const Duration(milliseconds: 60)); + expect(session.filters, hasLength(1)); + expect(completed, isFalse); + } finally { + release.complete(); + } + expect(await pending, isTrue); + expect(await duplicate, isTrue); + expect(container.read(userCacheProvider).keys, containsAll(keys)); + expect(session.filters.map((f) => f.authors!.length), [1, 1000, 1]); + expect(session.filters.map((f) => f.limit), [1, 1000, 1]); + expect(session.maxRunning, 1); + }, + ); + + for (final preload in [true, false]) { + for (final outcome in ['success', 'failure', 'community change']) { + test( + '${preload ? "preload" : "refresh"} waits for final bounded query: $outcome', + () async { + final started = Completer(); + final release = Completer(); + final session = _ClampingSession( + beforeFetch: (index) async { + if (index != 1) return; + started.complete(); + await release.future; + if (outcome == 'failure') throw StateError('relay unavailable'); + }, + ); + final container = ProviderContainer( + overrides: [relaySessionProvider.overrideWith(() => session)], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + final keys = List.generate(1001, (i) => 'profile-$i'); + bool? result; + final done = (preload ? cache.preload(keys) : cache.refresh(keys)) + .then((value) => result = value); + await started.future.timeout(const Duration(seconds: 5)); + expect(result, isNull); + expect(container.read(userCacheProvider), hasLength(1000)); + if (outcome == 'community change') { + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://next.invalid'); + container.read(userCacheProvider); + } + release.complete(); + await done; + expect(result, outcome == 'success'); + expect(session.filters.map((f) => f.authors!.length), [1000, 1]); + expect(session.maxRunning, 1); + expect( + container.read(userCacheProvider), + hasLength( + outcome == 'community change' + ? 0 + : outcome == 'failure' + ? 1000 + : 1001, + ), + ); + }, + ); + } + } +} + +class _ClampingSession extends RelaySessionNotifier { + _ClampingSession({this.first, this.beforeFetch}); + final NostrEvent? first; + final Future Function(int)? beforeFetch; + final filters = []; + int running = 0; + int maxRunning = 0; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) async { + final index = filters.length; + filters.add(filter); + running++; + if (running > maxRunning) maxRunning = running; + try { + await beforeFetch?.call(index); + return [ + for (final key in filter.authors!.take(1000)) + if (key == first?.pubkey) + first! + else + NostrEvent( + id: key, + pubkey: key, + createdAt: 1, + kind: 0, + tags: [], + content: jsonEncode({'name': key}), + sig: 'synthetic', + ), + ]; + } finally { + running--; + } + } +} diff --git a/mobile/test/shared/push/push_presentation_export_test.dart b/mobile/test/shared/push/push_presentation_export_test.dart new file mode 100644 index 0000000000..3d4baf63af --- /dev/null +++ b/mobile/test/shared/push/push_presentation_export_test.dart @@ -0,0 +1,608 @@ +import 'dart:async'; +import 'dart:isolate'; + +import 'package:buzz/shared/push/push_presentation_cache.dart'; +import 'package:buzz/shared/push/push_presentation_export_recovery.dart'; +import 'package:buzz/shared/relay/nostr_models.dart'; +import 'package:flutter/foundation.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; + +const _channel = MethodChannel('buzz/push'); +const _secret = + '0000000000000000000000000000000000000000000000000000000000000001'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + final calls = []; + final messenger = + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + + setUp(() { + calls.clear(); + debugDefaultTargetPlatformOverride = TargetPlatform.iOS; + pushPresentationCacheError.value = null; + pushPresentationExportError.value = null; + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + return null; + }); + }); + tearDown(() { + messenger.setMockMethodCallHandler(_channel, null); + debugDefaultTargetPlatformOverride = null; + pushPresentationCacheError.value = null; + pushPresentationExportError.value = null; + }); + + for (final profiles in [true, false]) { + for (final failedChunk in [0, 1]) { + test( + 'native chunk transient failure retries ${profiles ? "profiles" : "channels"} chunk $failedChunk in FIFO slot', + () async { + final limit = profiles ? 256 : 512; + final observations = ReceivePort(); + addTearDown(observations.close); + var workerReads = 0; + final caller = Isolate.current.debugName; + observations.listen((name) { + if (name != caller) workerReads++; + }); + final events = [ + for (var i = 0; i <= limit; i++) + _signed( + profiles ? 0 : 39000, + 100, + channelID: 'channel-$i', + secretKey: profiles + ? (i + 1).toRadixString(16).padLeft(64, '0') + : _secret, + ), + ]; + events[0] = _ObservedVerificationEvent( + events[0], + observations.sendPort, + ); + final memberships = profiles + ? [] + : [ + for (var i = 0; i <= limit; i++) + _signed(39002, 100, channelID: 'channel-$i'), + ]; + var failed = false; + final successfulChunks = []; + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + final args = call.arguments as Map; + if (args['communityId'] == 'following-community') return null; + final chunk = + (args[profiles ? 'events' : 'metadataEvents'] as List).length == + limit + ? 0 + : 1; + if (chunk == failedChunk && !failed) { + failed = true; + throw PlatformException( + code: profiles + ? 'profile_cache_failed' + : 'channel_cache_failed', + ); + } + successfulChunks.add(args); + return null; + }); + final exported = PushPresentationExportRecovery().export( + () => profiles + ? cacheBuzzPushProfileEvents('original-community', events) + : cacheBuzzPushChannelEvents( + 'original-community', + events, + memberships, + ), + ); + final following = cacheBuzzPushProfileEvents('following-community', [ + _signed(0, 200), + ]); + expect(await exported, isTrue); + await following; + expect(calls.map((call) => call.arguments['communityId']), [ + 'original-community', + 'original-community', + 'original-community', + 'following-community', + ]); + expect( + calls[failedChunk + 1].arguments, + equals(calls[failedChunk].arguments), + reason: 'retry must reuse the exact failed verified chunk', + ); + expect(successfulChunks, hasLength(2)); + expect([ + for (final chunk in successfulChunks) + ...chunk[profiles ? 'events' : 'metadataEvents'] as List, + ], unorderedEquals(events.map((event) => event.toJson()))); + if (!profiles) { + expect([ + for (final chunk in successfulChunks) + ...chunk['membershipEvents'] as List, + ], unorderedEquals(memberships.map((event) => event.toJson()))); + } + await Future.delayed(Duration.zero); + expect( + workerReads, + 1, + reason: 'native retries must not repeat signature verification', + ); + expect(pushPresentationExportError.value, isNull); + }, + ); + } + } + + for (final profiles in [true, false]) { + test( + 'permanent first ${profiles ? "profile" : "channel"} chunk failure exhausts retries and stops export', + () async { + final limit = profiles ? 256 : 512; + final events = [ + for (var i = 0; i <= limit; i++) + _signed( + profiles ? 0 : 39000, + 100, + channelID: 'channel-$i', + secretKey: profiles + ? (i + 1).toRadixString(16).padLeft(64, '0') + : _secret, + ), + ]; + final memberships = profiles + ? [] + : [ + for (var i = 0; i <= limit; i++) + _signed(39002, 100, channelID: 'channel-$i'), + ]; + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + if (call.arguments['communityId'] == 'failed-community') { + throw PlatformException( + code: profiles ? 'profile_cache_failed' : 'channel_cache_failed', + ); + } + return null; + }); + final recovery = PushPresentationExportRecovery(); + final succeeded = await recovery.export( + () => profiles + ? cacheBuzzPushProfileEvents('failed-community', events) + : cacheBuzzPushChannelEvents( + 'failed-community', + events, + memberships, + ), + ); + expect(succeeded, isFalse); + expect( + calls, + hasLength(6), + reason: 'initial attempt plus five retries, with no later chunk', + ); + expect( + pushPresentationCacheError.value, + contains(profiles ? 'profile_cache_failed' : 'channel_cache_failed'), + ); + final terminal = pushPresentationExportError.value; + expect( + terminal, + contains(profiles ? 'profile_cache_failed' : 'channel_cache_failed'), + ); + await cacheBuzzPushProfileEvents('following-community', [ + _signed(0, 200), + ]); + expect(calls, hasLength(7)); + for (final retry in calls.take(6)) { + expect(retry.arguments, equals(calls.first.arguments)); + } + expect(calls.last.arguments['communityId'], 'following-community'); + expect(pushPresentationExportError.value, terminal); + }, + ); + } + + test('native retry budget is shared across all chunks', () async { + final events = [ + for (var i = 1; i <= 257; i++) + _signed(0, 100, secretKey: i.toRadixString(16).padLeft(64, '0')), + ]; + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + final rows = call.arguments['events'] as List; + if (calls.length == 1 || rows.length == 1) { + throw PlatformException(code: 'profile_cache_failed'); + } + return null; + }); + expect( + await PushPresentationExportRecovery().export( + () => cacheBuzzPushProfileEvents('community', events), + ), + isFalse, + ); + expect(calls, hasLength(7)); + expect(calls.map((call) => (call.arguments['events'] as List).length), [ + 256, + 256, + 1, + 1, + 1, + 1, + 1, + ]); + expect(pushPresentationExportError.value, contains('profile_cache_failed')); + }); + + test( + 'native argument errors are terminal without persistence retries', + () async { + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + throw PlatformException(code: 'invalid_arguments'); + }); + expect( + await PushPresentationExportRecovery().export( + () => cacheBuzzPushProfileEvents('community', [_signed(0, 100)]), + ), + isFalse, + ); + expect(calls, hasLength(1)); + expect(pushPresentationExportError.value, contains('invalid_arguments')); + }, + ); + + test('unavailable native bridge remains an intentional no-op', () async { + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + throw MissingPluginException('synthetic non-Runner embedding'); + }); + expect( + await PushPresentationExportRecovery().export( + () => cacheBuzzPushProfileEvents('community', [_signed(0, 100)]), + ), + isTrue, + ); + expect(calls, hasLength(1)); + expect(pushPresentationCacheError.value, isNull); + expect(pushPresentationExportError.value, isNull); + }); + + for (final profiles in [true, false]) { + test( + 'large ${profiles ? "profile" : "channel"} exports fit native bounds', + () async { + final limit = profiles ? 256 : 512; + final events = [ + for (var i = 0; i <= limit; i++) + _signed( + profiles ? 0 : 39000, + 100, + channelID: 'channel-$i', + secretKey: profiles + ? (i + 1).toRadixString(16).padLeft(64, '0') + : _secret, + ), + ]; + // Deliberately reverse roster order: each chunk must pair by channel ID. + final memberships = profiles + ? [] + : [ + for (var i = limit; i >= 0; i--) + _signed(39002, 100, channelID: 'channel-$i'), + ]; + final received = []; + final receivedMemberships = []; + messenger.setMockMethodCallHandler(_channel, (call) async { + final args = call.arguments as Map; + if (args['communityId'] == 'following-community') { + calls.add(call); + return null; + } + final metadata = args[profiles ? 'events' : 'metadataEvents'] as List; + final rosters = profiles + ? [] + : args['membershipEvents'] as List; + if (metadata.length > limit || rosters.length > limit) { + throw PlatformException(code: 'invalid_arguments'); + } + expect(args['communityId'], 'bounded-community'); + if (!profiles) { + String id(dynamic event) => + (event['tags'] as List).firstWhere((tag) => tag[0] == 'd')[1] + as String; + expect(rosters.map(id).toSet(), metadata.map(id).toSet()); + } + received.addAll(metadata); + receivedMemberships.addAll(rosters); + calls.add(call); + return null; + }); + final largeExport = profiles + ? cacheBuzzPushProfileEvents('bounded-community', events) + : cacheBuzzPushChannelEvents( + 'bounded-community', + events, + memberships, + ); + final followingExport = cacheBuzzPushProfileEvents( + 'following-community', + [_signed(0, 200)], + ); + await Future.wait([largeExport, followingExport]); + expect(pushPresentationCacheError.value, isNull); + expect(calls.map((call) => call.arguments['communityId']), [ + 'bounded-community', + 'bounded-community', + 'following-community', + ]); + expect( + received, + unorderedEquals(events.map((event) => event.toJson())), + ); + expect( + receivedMemberships, + unorderedEquals(memberships.map((event) => event.toJson())), + ); + }, + ); + } + + for (final kind in [0, 39000]) { + test( + 'kind $kind verifies off the calling isolate before native export', + () async { + final verification = Completer(); + final observations = ReceivePort(); + addTearDown(observations.close); + final callerIsolateName = Isolate.current.debugName; + observations.listen((name) { + if (name != callerIsolateName && !verification.isCompleted) { + verification.complete(); + } + }); + final event = _ObservedVerificationEvent( + _signed(kind, 100), + observations.sendPort, + ); + var timerRan = false; + Timer.run(() => timerRan = true); + if (kind == 0) { + await cacheBuzzPushProfileEvents('community', [event]); + } else { + await cacheBuzzPushChannelEvents('community', [event], []); + } + // The signal originates from reading the real signature in the worker. + // Restoring synchronous verification cannot produce it. This is a + // liveness timeout, not a device-dependent elapsed-time performance gate. + await verification.future.timeout(const Duration(seconds: 5)); + expect(timerRan, isTrue); + expect(calls.single.method, 'syncPushSnapshot'); + expect(calls.single.arguments['communityId'], 'community'); + final key = kind == 0 ? 'events' : 'metadataEvents'; + expect(calls.single.arguments[key], [event.toJson()]); + }, + ); + } + + test( + 'profile export retains older valid candidates and signed payloads', + () async { + final older = _signed(0, 100); + final newestValid = _signed(0, 200); + await cacheBuzzPushProfileEvents('profiles-community', [ + older, + _tampered(_signed(0, 300)), + newestValid, + _signed(39000, 400), + ]); + expect(calls.single.arguments, { + 'section': 'profiles', + 'communityId': 'profiles-community', + 'events': [newestValid.toJson()], + }); + }, + ); + + test( + 'channel export preserves valid fallback and membership scope pairing', + () async { + final metadata = _signed(39000, 200); + final membership = _signed(39002, 200); + await cacheBuzzPushChannelEvents( + 'channels-community', + [ + _signed(39000, 100), + _tampered(_signed(39000, 300)), + metadata, + _signed(39000, 400, channelID: 'unpaired'), + ], + [_tampered(_signed(39002, 300)), membership, _signed(39002, 100)], + ); + expect(calls.single.arguments, { + 'section': 'channels', + 'communityId': 'channels-community', + 'metadataEvents': [metadata.toJson()], + 'membershipEvents': [membership.toJson()], + }); + }, + ); + + test( + 'native handoff and completion retain FIFO across sections and communities', + () async { + final firstEntered = Completer(); + final releaseFirst = Completer(); + var firstReleased = false; + var overlappingHandoff = false; + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + if (calls.length == 1) { + firstEntered.complete(); + await releaseFirst.future; + firstReleased = true; + } else { + overlappingHandoff |= !firstReleased; + } + return null; + }); + final completed = []; + final first = cacheBuzzPushProfileEvents('first', [ + _signed(0, 100), + ]).then((_) => completed.add('first')); + await firstEntered.future; + final second = cacheBuzzPushChannelEvents('second', [ + _signed(39000, 100), + ], []).then((_) => completed.add('second')); + final third = cacheBuzzPushProfileEvents('third', [ + _signed(0, 200), + ]).then((_) => completed.add('third')); + try { + await Future.delayed(Duration.zero); + expect(completed, isEmpty); + expect(calls.length, 1); + } finally { + releaseFirst.complete(); + await Future.wait([first, second, third]); + } + expect(overlappingHandoff, isFalse); + expect(completed, ['first', 'second', 'third']); + expect(calls.map((call) => call.arguments['communityId']), [ + 'first', + 'second', + 'third', + ]); + }, + ); + + test( + 'saturation rejects explicitly and admitted exports drain before recovery', + () async { + final entered = Completer(); + final release = Completer(); + messenger.setMockMethodCallHandler(_channel, (call) async { + calls.add(call); + if (calls.length == 1) { + entered.complete(); + await release.future; + } + return null; + }); + final event = _signed(0, 100); + final admitted = [ + cacheBuzzPushProfileEvents('0', [event]), + ]; + await entered.future; + try { + for (var i = 1; i < 8; i++) { + admitted.add(cacheBuzzPushProfileEvents('$i', [event])); + } + await expectLater( + cacheBuzzPushProfileEvents('rejected', [ + event, + ]).timeout(const Duration(seconds: 5)), + throwsStateError, + ); + expect(calls.length, 1); + } finally { + release.complete(); + await Future.wait(admitted); + } + expect(calls.map((call) => call.arguments['communityId']), [ + for (var i = 0; i < 8; i++) '$i', + ]); + await cacheBuzzPushProfileEvents('recovered', [event]); + expect(calls.last.arguments['communityId'], 'recovered'); + expect(calls.length, 9); + }, + ); + + test( + 'worker submission failure propagates and releases the next export', + () async { + final unsendable = ReceivePort(); + addTearDown(unsendable.close); + final failed = cacheBuzzPushProfileEvents('failed', [ + _UnsendableEvent(_signed(0, 100), unsendable), + ]); + final failure = expectLater(failed, throwsArgumentError); + final next = cacheBuzzPushProfileEvents('next', [_signed(0, 200)]); + await failure; + await next.timeout(const Duration(seconds: 5)); + expect(calls.length, 1); + expect(calls.single.arguments['communityId'], 'next'); + }, + ); +} + +NostrEvent _signed( + int kind, + int createdAt, { + String channelID = 'channel', + String secretKey = _secret, +}) => NostrEvent.fromJson( + nostr.Event.from( + kind: kind, + createdAt: createdAt, + content: kind == 0 ? '{"name":"Synthetic profile"}' : '', + tags: kind == 0 + ? [] + : [ + ['d', channelID], + ], + secretKey: secretKey, + ).toMap(), +); + +NostrEvent _tampered(NostrEvent event) => NostrEvent( + id: event.id, + pubkey: event.pubkey, + createdAt: event.createdAt, + kind: event.kind, + tags: event.tags, + content: 'Tampered', + sig: event.sig, +); + +class _ObservedVerificationEvent extends NostrEvent { + final SendPort observations; + + _ObservedVerificationEvent(NostrEvent event, this.observations) + : super( + id: event.id, + pubkey: event.pubkey, + createdAt: event.createdAt, + kind: event.kind, + tags: event.tags, + content: event.content, + sig: event.sig, + ); + + @override + String get sig { + observations.send(Isolate.current.debugName); + return super.sig; + } +} + +class _UnsendableEvent extends NostrEvent { + final ReceivePort port; + + _UnsendableEvent(NostrEvent event, this.port) + : super( + id: event.id, + pubkey: event.pubkey, + createdAt: event.createdAt, + kind: event.kind, + tags: event.tags, + content: event.content, + sig: event.sig, + ); +} diff --git a/mobile/test/shared/push/push_presentation_recovery_test.dart b/mobile/test/shared/push/push_presentation_recovery_test.dart new file mode 100644 index 0000000000..9d58cebe36 --- /dev/null +++ b/mobile/test/shared/push/push_presentation_recovery_test.dart @@ -0,0 +1,707 @@ +import 'dart:async'; +import 'dart:isolate'; + +import 'package:buzz/features/channels/channels_provider.dart'; +import 'package:buzz/shared/community/community.dart'; +import 'package:buzz/shared/community/community_provider.dart'; +import 'package:buzz/shared/profile/user_cache_provider.dart'; +import 'package:buzz/shared/profile/profile_event_parser.dart'; +import 'package:buzz/shared/push/push_presentation_cache.dart'; +import 'package:buzz/shared/push/push_presentation_export_recovery.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter/foundation.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter/widgets.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; + +const _bridge = MethodChannel('buzz/push'); + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + final messenger = + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + setUp(() { + debugDefaultTargetPlatformOverride = TargetPlatform.iOS; + pushPresentationExportError.value = null; + }); + tearDown(() { + debugDefaultTargetPlatformOverride = null; + pushPresentationExportError.value = null; + messenger.setMockMethodCallHandler(_bridge, null); + }); + + for (final profiles in [true, false]) { + final section = profiles ? 'profiles' : 'channels'; + test( + '$section producer recovers its saturated detached snapshot', + () async { + final blocked = Completer(); + final entered = Completer(); + final delivered = Completer>(); + messenger.setMockMethodCallHandler(_bridge, (call) async { + final args = call.arguments as Map; + if (args['communityId'] == 'blocker' && !entered.isCompleted) { + entered.complete(); + await blocked.future; + } + if (args['communityId'] == 'original' && !delivered.isCompleted) { + delivered.complete(args); + } + return null; + }); + final filler = _signed(0); + final queued = [ + cacheBuzzPushProfileEvents('blocker', [filler]), + ]; + await entered.future; + for (var i = 1; i < 8; i++) { + queued.add(cacheBuzzPushProfileEvents('queued-$i', [filler])); + } + final event = _signed(profiles ? 0 : 39000); + final membership = _signed(39002); + var communityID = 'original'; + final container = _container( + _Session(event, membership), + communityID: () => communityID, + ); + addTearDown(container.dispose); + Future? preload; + try { + await container.read(activeCommunityProvider.future); + if (profiles) { + final ready = Completer(); + final subscription = container.listen(userCacheProvider, (_, next) { + if (next.containsKey(event.pubkey) && !ready.isCompleted) { + ready.complete(); + } + }); + addTearDown(subscription.close); + preload = container.read(userCacheProvider.notifier).preload([ + event.pubkey, + ]); + await ready.future.timeout(const Duration(seconds: 5)); + } else { + await container.read(channelsProvider.future); + } + expect(delivered.isCompleted, isFalse); + communityID = 'replacement'; + container.invalidate(activeCommunityProvider); + await container.read(activeCommunityProvider.future); + } finally { + blocked.complete(); + await Future.wait(queued); + if (preload != null) expect(await preload, isTrue); + } + final payload = await delivered.future.timeout( + const Duration(seconds: 5), + ); + expect(payload['section'], section); + expect(payload['communityId'], 'original'); + expect(payload[profiles ? 'events' : 'metadataEvents'], [ + event.toJson(), + ]); + if (!profiles) { + expect(payload['membershipEvents'], [membership.toJson()]); + } + expect(pushPresentationExportError.value, isNull); + }, + ); + + test( + '$section producer preserves distinct input beyond eight arrivals', + () async { + final entered = Completer(); + final release = Completer(); + final payloads = >[]; + final delivered = Completer(); + final expected = {}; + messenger.setMockMethodCallHandler(_bridge, (call) async { + final args = call.arguments as Map; + if (args['communityId'] == 'blocker' && !entered.isCompleted) { + entered.complete(); + await release.future; + } + if (args['communityId'] == 'original') { + payloads.add(args); + final ids = { + for (final payload in payloads) + for (final event + in payload[profiles ? 'events' : 'metadataEvents'] as List) + event['id'] as String, + }; + if (expected.length == 13 && + ids.containsAll(expected) && + !delivered.isCompleted) { + delivered.complete(); + } + } + return null; + }); + final queued = [ + cacheBuzzPushProfileEvents('blocker', [_signed(0)]), + ]; + await entered.future; + for (var i = 1; i < 8; i++) { + queued.add(cacheBuzzPushProfileEvents('queued-$i', [_signed(0)])); + } + final first = _signed(profiles ? 0 : 39000); + expected.add(first.id); + final session = _Session(first, _signed(39002)); + final container = _container(session); + addTearDown(container.dispose); + final loads = >[]; + List? visibleBefore; + var subscriptionsBefore = 0; + try { + await container.read(activeCommunityProvider.future); + if (profiles) { + final ready = Completer(); + final subscription = container.listen(userCacheProvider, (_, next) { + if (next.containsKey(first.pubkey) && !ready.isCompleted) { + ready.complete(); + } + }); + addTearDown(subscription.close); + loads.add( + container.read(userCacheProvider.notifier).preload([ + first.pubkey, + ]), + ); + await ready.future.timeout(const Duration(seconds: 5)); + expect(await loads.single, isTrue); + } else { + await container.read(channelsProvider.future); + } + for (var i = 2; i <= 13; i++) { + final event = _signed( + profiles ? 0 : 39000, + key: i, + channel: 'channel-$i', + ); + session.extra.add(event); + expected.add(event.id); + if (profiles) { + loads.add( + container.read(userCacheProvider.notifier).preload([ + event.pubkey, + ]), + ); + } else { + session.extra.add(_signed(39002, channel: 'channel-$i')); + await container.read(channelsProvider.notifier).refresh(); + } + } + // Include a newer valid event and an invalid newest candidate for an + // entity in a pending partial batch. Selection must keep the valid one. + final newer = _signed( + profiles ? 0 : 39000, + key: 2, + channel: 'channel-2', + createdAt: 200, + ); + expected.remove( + _signed(profiles ? 0 : 39000, key: 2, channel: 'channel-2').id, + ); + expected.add(newer.id); + session.extra.addAll([ + newer, + NostrEvent( + id: 'invalid-newest', + pubkey: newer.pubkey, + createdAt: 300, + kind: newer.kind, + tags: newer.tags, + content: 'tampered', + sig: newer.sig, + ), + ]); + if (profiles) { + // Let the 50 ms fetch-coalescing timer fire if backpressure is + // removed. This is scheduling coverage, not a latency threshold. + await Future.delayed(const Duration(milliseconds: 60)); + expect( + session.profileFetches, + 1, + reason: 'pending keys must remain upstream while export waits', + ); + } else { + for (var turn = 0; turn < 20; turn++) { + await Future.delayed(Duration.zero); + } + visibleBefore = container + .read(channelsProvider) + .requireValue + .map((channel) => channel.id) + .toList(); + subscriptionsBefore = session.subscriptions; + } + } finally { + release.complete(); + await Future.wait(queued); + } + await Future.wait(loads); + await delivered.future.timeout(const Duration(seconds: 5)); + expect(pushPresentationExportError.value, isNull); + if (!profiles) { + expect( + container + .read(channelsProvider) + .requireValue + .map((channel) => channel.id), + visibleBefore, + ); + expect(session.subscriptions, subscriptionsBefore); + expect( + session.directoryLoads, + 0, + reason: 'notification recovery must not fetch the open directory', + ); + expect( + payloads.last['membershipEvents'], + unorderedEquals([ + session.membership.toJson(), + for (final event in session.extra) + if (event.kind == 39002) event.toJson(), + ]), + ); + } + }, + ); + + test( + '$section producer handles detached worker submission failure', + () async { + final port = ReceivePort(); + addTearDown(port.close); + final event = _Unsendable(_signed(profiles ? 0 : 39000), port); + final failure = Completer(); + void onError() { + if (pushPresentationExportError.value != null && + !failure.isCompleted) { + failure.complete(); + } + } + + pushPresentationExportError.addListener(onError); + addTearDown(() => pushPresentationExportError.removeListener(onError)); + var nativeCalls = 0; + messenger.setMockMethodCallHandler(_bridge, (_) async { + nativeCalls++; + return null; + }); + final container = _container( + _Session(event, _signed(39002)), + bypassProfileWorker: true, + ); + addTearDown(container.dispose); + await container.read(activeCommunityProvider.future); + if (profiles) { + await container.read(userCacheProvider.notifier).preload([ + event.pubkey, + ]); + } else { + await container.read(channelsProvider.future); + } + await failure.future.timeout(const Duration(seconds: 5)); + expect(nativeCalls, 0); + final terminalError = pushPresentationExportError.value; + await cacheBuzzPushProfileEvents('unrelated-success', [_signed(0)]); + expect(nativeCalls, 1); + expect(pushPresentationExportError.value, terminalError); + }, + ); + } + + for (final mode in ['reentrant', 'retired', 'failure']) { + test('channel dirty refetch handles $mode work', () async { + final nativeEntered = Completer(); + final releaseNative = Completer(); + final refetchEntered = Completer(); + final releaseRefetch = Completer(); + final complete = Completer(); + final payloads = >[]; + messenger.setMockMethodCallHandler(_bridge, (call) async { + final args = call.arguments as Map; + payloads.add(args); + if (payloads.length == 1) { + nativeEntered.complete(); + await releaseNative.future; + } else if ((mode == 'reentrant' && payloads.length == 3) || + (mode == 'retired' && args['communityId'] == 'replacement')) { + if (!complete.isCompleted) complete.complete(); + } + return null; + }); + void errorChanged() { + if (mode == 'failure' && + pushPresentationExportError.value != null && + !complete.isCompleted) { + complete.complete(); + } + } + + pushPresentationExportError.addListener(errorChanged); + addTearDown( + () => pushPresentationExportError.removeListener(errorChanged), + ); + final session = _Session(_signed(39000), _signed(39002)); + var community = 'original'; + final container = _container(session, communityID: () => community); + addTearDown(container.dispose); + await container.read(activeCommunityProvider.future); + await container.read(channelsProvider.future); + await nativeEntered.future; + await container.read(channelsProvider.notifier).refresh(); + session.beforeMetadata = () async { + refetchEntered.complete(); + await releaseRefetch.future; + if (mode == 'failure') throw StateError('synthetic refetch failure'); + }; + releaseNative.complete(); + await refetchEntered.future.timeout(const Duration(seconds: 5)); + try { + if (mode == 'reentrant') { + session.extra.addAll([ + _signed(39000, channel: 'later'), + _signed(39002, channel: 'later'), + ]); + await container.read(channelsProvider.notifier).refresh(); + } else if (mode == 'retired') { + community = 'replacement'; + session.event = _signed(39000, channel: 'replacement-channel'); + session.membership = _signed(39002, channel: 'replacement-channel'); + container.invalidate(activeCommunityProvider); + await container.read(activeCommunityProvider.future); + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://replacement.invalid'); + await container.read(channelsProvider.future); + await container.read(channelsProvider.notifier).refresh(); + } + } finally { + releaseRefetch.complete(); + } + await complete.future.timeout(const Duration(seconds: 5)); + for (var turn = 0; turn < 20; turn++) { + await Future.delayed(Duration.zero); + } + if (mode == 'failure') { + expect(session.metadataLoads, 3); + expect(session.directoryLoads, 0); + expect(payloads, hasLength(1)); + expect( + pushPresentationExportError.value, + contains('synthetic refetch failure'), + ); + } else if (mode == 'retired') { + expect( + payloads.where((payload) => payload['communityId'] == 'original'), + hasLength(1), + ); + expect(payloads.last['metadataEvents'], [session.event.toJson()]); + expect(payloads.last['membershipEvents'], [ + session.membership.toJson(), + ]); + expect(pushPresentationExportError.value, isNull); + } else { + expect(session.metadataLoads, 5); + expect(session.directoryLoads, 0); + expect(payloads.last['metadataEvents'], hasLength(2)); + expect(payloads.last['membershipEvents'], hasLength(2)); + expect(pushPresentationExportError.value, isNull); + } + }); + } + + for (final failedOperation in ['initial export', 'dirty refetch']) { + test( + 'pending channel input drains after exhausted $failedOperation', + () async { + final blockerEntered = Completer(); + final releaseBlocker = Completer(); + final initialEntered = Completer(); + final releaseInitial = Completer(); + final refetchEntered = Completer(); + final releaseRefetch = Completer(); + final terminal = Completer(); + late _Session session; + List? readsBeforeAdmissionRetries; + List? readsAtTerminal; + final latestDelivered = Completer>(); + final latest = _signed(39000, channel: 'latest'); + final latestMembership = _signed(39002, channel: 'latest'); + messenger.setMockMethodCallHandler(_bridge, (call) async { + final args = call.arguments as Map; + if (args['communityId'] == 'blocker' && !blockerEntered.isCompleted) { + blockerEntered.complete(); + await releaseBlocker.future; + } + if (args['communityId'] == 'original') { + if (failedOperation == 'dirty refetch' && + !initialEntered.isCompleted) { + initialEntered.complete(); + await releaseInitial.future; + } + if ((args['metadataEvents'] as List).any( + (event) => event['id'] == latest.id, + ) && + !latestDelivered.isCompleted) { + latestDelivered.complete(args); + } + } + return null; + }); + void onFailure() { + if (pushPresentationExportError.value != null && + !terminal.isCompleted) { + readsAtTerminal = session.snapshotReadCounts; + terminal.complete(); + } + } + + pushPresentationExportError.addListener(onFailure); + addTearDown( + () => pushPresentationExportError.removeListener(onFailure), + ); + final queued = >[]; + Future saturate() async { + queued.add(cacheBuzzPushProfileEvents('blocker', [_signed(0)])); + await blockerEntered.future; + for (var i = 1; i < 8; i++) { + queued.add(cacheBuzzPushProfileEvents('queued-$i', [_signed(0)])); + } + } + + session = _Session(_signed(39000), _signed(39002)); + final container = _container(session); + addTearDown(container.dispose); + try { + if (failedOperation == 'initial export') await saturate(); + await container.read(activeCommunityProvider.future); + await container.read(channelsProvider.future); + if (failedOperation == 'dirty refetch') { + await initialEntered.future; + await container.read(channelsProvider.notifier).refresh(); + session.beforeMetadata = () async { + refetchEntered.complete(); + await releaseRefetch.future; + }; + releaseInitial.complete(); + await refetchEntered.future.timeout(const Duration(seconds: 5)); + await saturate(); + } + session.extra.addAll([latest, latestMembership]); + await container.read(channelsProvider.notifier).refresh(); + readsBeforeAdmissionRetries = session.snapshotReadCounts; + if (!releaseRefetch.isCompleted) releaseRefetch.complete(); + await terminal.future.timeout(const Duration(seconds: 12)); + expect(pushPresentationExportError.value, contains('queue is full')); + if (failedOperation == 'dirty refetch') { + // Membership pages and metadata were fetched before the paused + // metadata response. Only the remaining member snapshot is read; + // all five admission retries must reuse those same raw events. + expect( + [ + for (var i = 0; i < 3; i++) + readsAtTerminal![i] - readsBeforeAdmissionRetries[i], + ], + [0, 0, 1], + ); + } + // Capacity returns only after the predecessor has exhausted every retry. + releaseBlocker.complete(); + await Future.wait(queued); + final payload = await latestDelivered.future.timeout( + const Duration(seconds: 5), + ); + expect(payload['communityId'], 'original'); + expect(payload['metadataEvents'], contains(equals(latest.toJson()))); + expect( + payload['membershipEvents'], + contains(equals(latestMembership.toJson())), + ); + expect(session.directoryLoads, 0); + expect(pushPresentationExportError.value, contains('queue is full')); + } finally { + if (!releaseBlocker.isCompleted) releaseBlocker.complete(); + if (!releaseInitial.isCompleted) releaseInitial.complete(); + if (!releaseRefetch.isCompleted) releaseRefetch.complete(); + await Future.wait(queued); + } + }, + ); + } + + test('bounded retries terminate and a later operation can succeed', () { + fakeAsync((clock) { + final recovery = PushPresentationExportRecovery(); + var attempts = 0; + bool? result; + recovery + .export(() async { + attempts++; + throw PushPresentationExportQueueFull(); + }) + .then((value) => result = value); + clock.flushMicrotasks(); + clock.elapse(const Duration(seconds: 8)); + expect(result, isFalse); + expect(attempts, 6); + final terminal = pushPresentationExportError.value; + expect(terminal, isNotNull); + clock.elapse(const Duration(days: 1)); + expect(attempts, 6); + bool? recovered; + recovery.export(() async {}).then((value) => recovered = value); + clock.flushMicrotasks(); + expect(recovered, isTrue); + expect(pushPresentationExportError.value, terminal); + }); + }); +} + +ProviderContainer _container( + _Session session, { + String Function()? communityID, + bool bypassProfileWorker = false, +}) => ProviderContainer( + retry: (_, _) => null, + overrides: [ + if (bypassProfileWorker) + profileEventBatchParserProvider.overrideWithValue( + (events) async => events.map(parseProfileEvent).toList(), + ), + relaySessionProvider.overrideWith(() => session), + appLifecycleProvider.overrideWith(_Lifecycle.new), + myPubkeyProvider.overrideWith((ref) => 'me'), + activeCommunityProvider.overrideWith( + (ref) async => Community( + id: communityID?.call() ?? 'original', + name: 'Synthetic', + relayUrl: 'https://example.invalid', + addedAt: DateTime(2026), + ), + ), + ], +); + +class _Session extends RelaySessionNotifier { + _Session(this.event, this.membership); + NostrEvent event; + NostrEvent membership; + final extra = []; + int profileFetches = 0; + int subscriptions = 0; + int directoryLoads = 0; + int metadataLoads = 0; + int membershipPages = 0; + int memberSnapshotLoads = 0; + List get snapshotReadCounts => [ + membershipPages, + metadataLoads, + memberSnapshotLoads, + ]; + Future Function()? beforeMetadata; + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) async { + if (filter.kinds.contains(0)) profileFetches++; + if (filter.kinds.contains(39002) && filter.tags['#d'] != null) { + memberSnapshotLoads++; + } + if (filter.kinds.contains(39000)) { + metadataLoads++; + final hook = beforeMetadata; + beforeMetadata = null; + if (hook != null) await hook(); + } + return [ + for (final candidate in [event, membership, ...extra]) + if (filter.kinds.contains(candidate.kind) && + (filter.authors == null || + filter.authors!.contains(candidate.pubkey))) + candidate, + ]; + } + + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async { + if (filters.any( + (filter) => filter.kinds.contains(39002) && filter.tags['#p'] != null, + )) { + membershipPages++; + } + if (filters.any( + (filter) => + filter.kinds.contains(39000) && + filter.tags['#d'] == null && + filter.extensions['before_id'] == null, + )) { + directoryLoads++; + } + return [ + for (final filter in filters) + ...await fetchHistory(filter, timeout: timeout), + ]; + } + + @override + Future subscribe( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String)? onClosed, + }) async { + subscriptions++; + return () {}; + } +} + +class _Lifecycle extends AppLifecycleNotifier { + @override + AppLifecycleState build() => AppLifecycleState.resumed; +} + +NostrEvent _signed( + int kind, { + int key = 1, + String channel = 'channel', + int createdAt = 100, +}) => NostrEvent.fromJson( + nostr.Event.from( + kind: kind, + createdAt: createdAt, + content: kind == 0 ? '{"name":"Synthetic"}' : '', + tags: kind == 0 + ? [] + : [ + ['d', channel], + ['name', 'Synthetic'], + if (kind == 39002) ['p', 'me'], + ], + secretKey: key.toRadixString(16).padLeft(64, '0'), + ).toMap(), +); + +class _Unsendable extends NostrEvent { + final ReceivePort port; + _Unsendable(NostrEvent event, this.port) + : super( + id: event.id, + pubkey: event.pubkey, + createdAt: event.createdAt, + kind: event.kind, + tags: event.tags, + content: event.content, + sig: event.sig, + ); +}