jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(mobile): verify notification cache events in background (#7650)

Browse Source 
On iOS, Buzz stores verified profile and channel information for the
notification extension. The extension uses this cache to show sender and
channel names. Loading profiles or channels also updates the cache.

Cache updates currently check event signatures on the UI thread, even
when the caller does not wait for the update. This can delay taps and
screen updates. This change moves those checks to a background worker.
Signature checks, event selection, community boundaries, and the order
of native writes are preserved.

The worker accepts up to eight batches. When it is busy, later profile
requests stay coalesced before fetching; profiles already loaded remain
available immediately. Overlapping channel updates trigger a fresh
joined-channel cache snapshot after the current export finishes, without
delaying or restarting the channel screen. Queue saturation gets up to
five retries per operation. Pending updates get their own recovery
attempt even if the preceding export fails. Recovery fetches
joined-channel data once and retries cache admission without repeating
relay reads. It does not fetch the open-channel directory. Verified
payloads are split into native-sized writes (256 profiles or 512 paired
channels), keeping the same export ordering. Native profile/channel
write failures retry the same verified chunk while retaining the
export's queue position. All chunks share five retries with 7.75 seconds
total backoff; other native errors fail immediately. Terminal failures
are recorded, and later successful writes do not hide them. Captured
batches keep their original community. Pending channel refetches are
retired on a community switch and refreshed when the user returns. The
cache remains best effort; prolonged failures can still prevent
delivery.

Tests through the production APIs reduced UI delay from 27–31 ms to
below 0.3 ms for four profiles, and from 59–60 ms to below 0.3 ms for
four channel and membership pairs. The native payloads were identical.
This does not measure overall screen speed.

Validation: 2,225 mobile tests passed (one skipped), along with 18
focused export tests, 12 recovery tests, 39 native Swift tests, 35
focused profile tests, and static analysis. Production regressions cover
13-entity bursts, invalid newer events, community rebuilds, reentrant
refreshes, and terminal failures. Regression tests fail when producer
recovery controls, terminal-failure draining, or member-only recovery
scope are removed.

---------

Signed-off-by: Tom Brow <tomb@block.xyz>
This commit is contained in:
Tom Brow
2026-09-15 16:01:46 -07:00
committed by GitHub
parent 76f7e5ef0c
commit b8aa0233f9
10 changed files with 2003 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift
View File

@@ -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()

15
mobile/lib/features/channels/channel_directory.dart
View File

@@ -82,22 +82,26 @@ final channelDirectoryLoadStatusProvider =
Future<List<NostrEvent>> _fetchChannelMemberships(
RelaySessionNotifier session,
String pubkey,
) => _fetchPaginatedChannelEvents(
String pubkey, {
void Function()? ensureCurrent,
}) => _fetchPaginatedChannelEvents(
session,
kind: 39002,
tags: {
'#p': [pubkey],
},
operation: 'Channel memberships',
ensureCurrent: ensureCurrent,
);
Future<List<NostrEvent>> _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<List<NostrEvent>> _fetchPaginatedChannelEvents(
required int kind,
required String operation,
Map<String, List<String>> tags = const {},
void Function()? ensureCurrent,
}) async {
final events = <NostrEvent>[];
final seenEventIds = <String>{};
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<List<NostrEvent>> _fetchPaginatedChannelEvents(
extensions: {'before_id': ?beforeId},
),
]);
ensureCurrent?.call();
if (page.isEmpty) break;
var madeProgress = false;
for (final event in page) {

106
mobile/lib/features/channels/channel_push_cache.dart Normal file
View File

@@ -0,0 +1,106 @@
part of 'channels_provider.dart';
extension _ChannelPushCache on ChannelsNotifier {
Future<void> _exportPushCache(
String? communityID,
List<NostrEvent> metadata,
List<NostrEvent> 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<String>()
.toSet()
.toList();
final memberMetadata = ids.isEmpty
? const <NostrEvent>[]
: await session.fetchHistory(NostrFilters.channelMetadata(ids));
if (!current()) return null;
final memberSnapshots = ids.isEmpty
? const <NostrEvent>[]
: 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;
}
}
}

9
mobile/lib/features/channels/channels_provider.dart
View File

@@ -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<List<Channel>> {
final _pushExport = PushPresentationExportRecovery();
bool _pushCacheExporting = false;
bool _pushCacheDirty = false;
static const _backstopInterval = Duration(seconds: 60);
final Map<String, _LiveChunkSubscription> _liveSubscriptionsByChunk = {};
@@ -300,7 +307,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
// linkage validation and member-count hydration.
if (memberEvents.isNotEmpty) _cacheMemberSnapshots(memberEvents);
unawaited(
cacheBuzzPushChannelEvents(communityID, dedupedMetas, [
_exportPushCache(communityID, dedupedMetas, [
...memberships,
...memberEvents,
]),

67
mobile/lib/shared/profile/user_cache_provider.dart
View File

@@ -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<Map<String, UserProfile>> {
// The relay clamps each history response to DEFAULT_MAX_PAGE_LIMIT (1000).
static const _maximumProfilesPerQuery = 1000;
final Set<String> _pending = {};
final _pushExport = PushPresentationExportRecovery();
final Map<String, ({int createdAt, String eventId})> _profileEventOrders = {};
int _generation = 0;
bool _flushInFlight = false;
@@ -81,11 +87,15 @@ class UserCacheNotifier extends Notifier<Map<String, UserProfile>> {
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<Map<String, UserProfile>> {
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<List<String>> _profileQueryBatches(List<String> 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);

180
mobile/lib/shared/push/push_presentation_cache.dart
View File

@@ -11,6 +11,19 @@ const _pushPresentationChannel = MethodChannel('buzz/push');
const _maximumAvatarSourceBytes = 512 * 1024;
const _maximumAvatarPNGBytes = 64 * 1024;
Future<void> _avatarEncodeTail = Future.value();
Future<void> _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<String?>(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<void> cacheBuzzPushProfileEvents(
String communityID,
Iterable<NostrEvent> events,
@@ -41,20 +57,30 @@ Future<void> 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<void> cacheBuzzPushChannelEvents(
String? communityID,
Iterable<NostrEvent> metadataEvents,
@@ -65,20 +91,89 @@ Future<void> 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<List<T>> _boundedChunks<T>(List<T> 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<void> _serializePresentationExport(
Future<void> Function() export,
) async {
if (_outstandingPresentationExports >= _maximumPresentationExports) {
throw PushPresentationExportQueueFull();
}
_outstandingPresentationExports++;
final previous = _presentationExportTail;
final release = Completer<void>();
_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<NostrEvent> _selectPushProfileEvents(List<NostrEvent> events) =>
_newestVerifiedEvents(
events,
kind: 0,
scope: (event) => event.pubkey.toLowerCase(),
).values.toList();
({List<NostrEvent> metadata, List<NostrEvent> membership})
_selectPushChannelBatch(
({List<NostrEvent> metadata, List<NostrEvent> membership}) batch,
) => selectPushChannelEvents(batch.metadata, batch.membership);
/// Selects the newest paired verified channel metadata and membership events.
@visibleForTesting
({List<NostrEvent> metadata, List<NostrEvent> membership})
@@ -162,18 +257,52 @@ Future<void> 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<void> _invokeBestEffort(Map<String, Object> 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<void> _invokeVerifiedChunk(
Map<String, Object> 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<void>.delayed(delay);
}
}
}
Future<void> _invokeSnapshot(
Map<String, Object> arguments, {
bool bestEffort = false,
}) async {
try {
await _pushPresentationChannel.invokeMethod<void>(
'syncPushSnapshot',
@@ -186,6 +315,9 @@ Future<void> _invokeBestEffort(Map<String, Object> 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;
}
}

49
mobile/lib/shared/push/push_presentation_export_recovery.dart Normal file
View File

@@ -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<String?>(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<bool> export(Future<void> 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<void>.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;
}
}

183
mobile/test/shared/profile/user_cache_batch_limit_test.dart Normal file
View File

@@ -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<void>();
final release = Completer<void>();
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<void>.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<void>();
final release = Completer<void>();
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<void> Function(int)? beforeFetch;
final filters = <NostrFilter>[];
int running = 0;
int maxRunning = 0;
@override
SessionState build() => const SessionState(status: SessionStatus.connected);
@override
Future<List<NostrEvent>> 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--;
}
}
}

608
mobile/test/shared/push/push_presentation_export_test.dart Normal file
View File

@@ -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 = <MethodCall>[];
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
? <NostrEvent>[]
: [
for (var i = 0; i <= limit; i++)
_signed(39002, 100, channelID: 'channel-$i'),
];
var failed = false;
final successfulChunks = <Map>[];
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<void>.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
? <NostrEvent>[]
: [
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
? <NostrEvent>[]
: [
for (var i = limit; i >= 0; i--)
_signed(39002, 100, channelID: 'channel-$i'),
];
final received = <dynamic>[];
final receivedMemberships = <dynamic>[];
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
? <dynamic>[]
: 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<void>();
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<void>();
final releaseFirst = Completer<void>();
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 = <String>[];
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<void>.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<void>();
final release = Completer<void>();
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,
);
}

707
mobile/test/shared/push/push_presentation_recovery_test.dart Normal file
View File

@@ -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<void>();
final entered = Completer<void>();
final delivered = Completer<Map<dynamic, dynamic>>();
messenger.setMockMethodCallHandler(_bridge, (call) async {
final args = call.arguments as Map<dynamic, dynamic>;
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<bool>? preload;
try {
await container.read(activeCommunityProvider.future);
if (profiles) {
final ready = Completer<void>();
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<void>();
final release = Completer<void>();
final payloads = <Map<dynamic, dynamic>>[];
final delivered = Completer<void>();
final expected = <String>{};
messenger.setMockMethodCallHandler(_bridge, (call) async {
final args = call.arguments as Map<dynamic, dynamic>;
if (args['communityId'] == 'blocker' && !entered.isCompleted) {
entered.complete();
await release.future;
}
if (args['communityId'] == 'original') {
payloads.add(args);
final ids = <String>{
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 = <Future<bool>>[];
List<String>? visibleBefore;
var subscriptionsBefore = 0;
try {
await container.read(activeCommunityProvider.future);
if (profiles) {
final ready = Completer<void>();
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<void>.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<void>.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>();
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<void>();
final releaseNative = Completer<void>();
final refetchEntered = Completer<void>();
final releaseRefetch = Completer<void>();
final complete = Completer<void>();
final payloads = <Map<dynamic, dynamic>>[];
messenger.setMockMethodCallHandler(_bridge, (call) async {
final args = call.arguments as Map<dynamic, dynamic>;
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<void>.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<void>();
final releaseBlocker = Completer<void>();
final initialEntered = Completer<void>();
final releaseInitial = Completer<void>();
final refetchEntered = Completer<void>();
final releaseRefetch = Completer<void>();
final terminal = Completer<void>();
late _Session session;
List<int>? readsBeforeAdmissionRetries;
List<int>? readsAtTerminal;
final latestDelivered = Completer<Map<dynamic, dynamic>>();
final latest = _signed(39000, channel: 'latest');
final latestMembership = _signed(39002, channel: 'latest');
messenger.setMockMethodCallHandler(_bridge, (call) async {
final args = call.arguments as Map<dynamic, dynamic>;
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<void>>[];
Future<void> 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 = <NostrEvent>[];
int profileFetches = 0;
int subscriptions = 0;
int directoryLoads = 0;
int metadataLoads = 0;
int membershipPages = 0;
int memberSnapshotLoads = 0;
List<int> get snapshotReadCounts => [
membershipPages,
metadataLoads,
memberSnapshotLoads,
];
Future<void> Function()? beforeMetadata;
@override
SessionState build() => const SessionState(status: SessionStatus.connected);
@override
Future<List<NostrEvent>> 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<List<NostrEvent>> queryRelay(
List<NostrFilter> 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<void Function()> 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,
);
}