From 2572c55ad1968376f190eeb415a41de5c5c8eb1d Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Mon, 14 Sep 2026 14:42:11 -0700 Subject: [PATCH 1/2] perf(mobile): verify profile batches off the UI isolate Signed-off-by: Tom Brow --- .../shared/profile/profile_event_parser.dart | 53 ++++ .../shared/profile/user_cache_provider.dart | 148 +++++---- .../profile/user_cache_provider_test.dart | 294 ++++++++++++++++++ 3 files changed, 440 insertions(+), 55 deletions(-) create mode 100644 mobile/lib/shared/profile/profile_event_parser.dart diff --git a/mobile/lib/shared/profile/profile_event_parser.dart b/mobile/lib/shared/profile/profile_event_parser.dart new file mode 100644 index 00000000000..eec12bf9a11 --- /dev/null +++ b/mobile/lib/shared/profile/profile_event_parser.dart @@ -0,0 +1,53 @@ +import 'dart:isolate'; + +import '../crypto/nip_oa.dart'; +import '../relay/nostr_models.dart'; +import 'user_profile.dart'; + +/// A verified profile together with its NIP-01 replacement order. +typedef ParsedProfileEvent = ({ + UserProfile profile, + int createdAt, + String eventId, +}); + +/// Parses and verifies profile history without blocking the UI isolate. +/// +/// Capture only the event batch here, never a provider or its surrounding +/// context. NIP-OA verification performs synchronous elliptic-curve work. +Future> parseProfileEventBatch( + List events, +) => Isolate.run(() => _parseBatch(events)); + +List _parseBatch(List events) { + final latest = {}; + for (final event in events) { + if (event.kind != 0) continue; + final pubkey = event.pubkey.toLowerCase(); + final current = latest[pubkey]; + if (current == null || + event.createdAt > current.createdAt || + (event.createdAt == current.createdAt && + event.id.compareTo(current.id) < 0)) { + latest[pubkey] = event; + } + } + return [for (final event in latest.values) parseProfileEvent(event)]; +} + +/// Parses a single live profile, including its verified owner attribution. +ParsedProfileEvent parseProfileEvent(NostrEvent event) { + final data = ProfileData.fromEvent(event); + return ( + profile: UserProfile( + pubkey: data.pubkey.toLowerCase(), + displayName: data.displayName, + avatarUrl: data.avatarUrl, + about: data.about, + nip05Handle: data.nip05, + ownerPubkey: verifiedOaOwnerPubkey(event.tags, event.pubkey), + ), + createdAt: event.createdAt, + eventId: event.id, + ); +} diff --git a/mobile/lib/shared/profile/user_cache_provider.dart b/mobile/lib/shared/profile/user_cache_provider.dart index c974b4a055f..3240e00caba 100644 --- a/mobile/lib/shared/profile/user_cache_provider.dart +++ b/mobile/lib/shared/profile/user_cache_provider.dart @@ -3,10 +3,10 @@ import 'dart:async'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../community/community_provider.dart'; -import '../crypto/nip_oa.dart'; import '../push/push_presentation_cache.dart'; import '../relay/relay.dart'; import 'user_profile.dart'; +import 'profile_event_parser.dart'; /// In-memory cache of user profiles, fetched in batches from the relay. /// @@ -15,14 +15,20 @@ import 'user_profile.dart'; class UserCacheNotifier extends Notifier> { final Set _pending = {}; final Map _profileEventOrders = {}; + int _generation = 0; + bool _flushInFlight = false; + Future _verificationQueue = Future.value(); Timer? _batchTimer; Completer? _batchCompleter; @override Map build() { ref.watch(relayConfigProvider); + _generation++; _profileEventOrders.clear(); ref.onDispose(() { + _generation++; + _pending.clear(); _batchTimer?.cancel(); _batchTimer = null; _batchCompleter?.complete(false); @@ -57,7 +63,7 @@ class UserCacheNotifier extends Notifier> { if (uncached.isEmpty && !alreadyPending) return Future.value(true); _pending.addAll(uncached); final completer = _batchCompleter ??= Completer(); - _batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending); + _scheduleBatch(); return completer.future; } @@ -72,23 +78,14 @@ class UserCacheNotifier extends Notifier> { .toSet() .toList(); if (normalized.isEmpty) return true; + final generation = _generation; try { final session = ref.read(relaySessionProvider.notifier); final events = await session.fetchHistory( NostrFilters.profilesBatch(normalized), ); - final updated = Map.from(state); - final updatedOrders = Map.from( - _profileEventOrders, - ); - for (final event in events) { - _cacheProfileEvent(event, updated, updatedOrders); - } - _profileEventOrders - ..clear() - ..addAll(updatedOrders); - state = updated; - return true; + if (!_isCurrent(generation)) return false; + return await _verifyAndMerge(events, generation); } catch (_) { return false; } @@ -108,7 +105,7 @@ class UserCacheNotifier extends Notifier> { if (state.containsKey(pubkey) || _pending.contains(pubkey)) return; _pending.add(pubkey); _batchCompleter ??= Completer(); - _batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending); + _scheduleBatch(); } Future _flushPending() async { @@ -120,6 +117,8 @@ class UserCacheNotifier extends Notifier> { final completer = _batchCompleter; _batchCompleter = null; + _flushInFlight = true; + final generation = _generation; var succeeded = false; try { final communityID = ref.read(activeCommunityProvider).value?.id; @@ -128,18 +127,8 @@ class UserCacheNotifier extends Notifier> { NostrFilters.profilesBatch(pubkeys), ); - final updated = Map.from(state); - final updatedOrders = Map.from( - _profileEventOrders, - ); - for (final event in events) { - _cacheProfileEvent(event, updated, updatedOrders); - } - - _profileEventOrders - ..clear() - ..addAll(updatedOrders); - state = updated; + if (!_isCurrent(generation)) return; + if (!await _verifyAndMerge(events, generation)) return; if (communityID != null) { unawaited(cacheBuzzPushProfileEvents(communityID, events)); } @@ -148,43 +137,92 @@ class UserCacheNotifier extends Notifier> { // Silently fail — non-gating callers will just show pubkeys. } finally { completer?.complete(succeeded); + _flushInFlight = false; + if (ref.mounted && _pending.isNotEmpty) _scheduleBatch(); } } - bool _cacheProfileEvent( - NostrEvent event, - Map profiles, [ - Map? orders, - ]) { + void _scheduleBatch() { + if (_flushInFlight) return; + _batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending); + } + + Future _verifyAndMerge(List events, int generation) { + // Both refresh and preload share one worker slot. Keep this queue across + // community changes so an old worker cannot overlap a new community's job. + final result = _verificationQueue.then((_) async { + if (!_isCurrent(generation)) return false; + // Match live-cache admission before parsing. In particular, a malformed + // stale profile must not fail a refresh that previously ignored it. + final pending = [ + for (final event in events) + if (event.kind == 0 && + _isNewer(event.pubkey.toLowerCase(), event.createdAt, event.id)) + event, + ]; + if (pending.isEmpty) return true; + final profiles = await ref.read(profileEventBatchParserProvider)(pending); + if (!_isCurrent(generation)) return false; + _mergeParsedProfiles(profiles); + return true; + }); + // Callers receive the original error; only the sequencing tail recovers so + // one rejected batch does not poison all future refreshes. + _verificationQueue = result.then( + (_) {}, + onError: (Object error, StackTrace stack) {}, + ); + return result; + } + + bool _isCurrent(int generation) => ref.mounted && generation == _generation; + + void _mergeParsedProfiles(List profiles) { + // Read the current state after the isolate completes: live events and other + // batches may have installed newer profiles while this batch was parsing. + final updated = Map.from(state); + var changed = false; + for (final parsed in profiles) { + if (!_isNewer(parsed.profile.pubkey, parsed.createdAt, parsed.eventId)) { + continue; + } + updated[parsed.profile.pubkey] = parsed.profile; + _profileEventOrders[parsed.profile.pubkey] = ( + createdAt: parsed.createdAt, + eventId: parsed.eventId, + ); + changed = true; + } + if (changed) state = updated; + } + + bool _isNewer(String pubkey, int createdAt, String eventId) { + final current = _profileEventOrders[pubkey]; + return current == null || + createdAt > current.createdAt || + (createdAt == current.createdAt && + eventId.compareTo(current.eventId) < 0); + } + + bool _cacheProfileEvent(NostrEvent event, Map profiles) { if (event.kind != 0) return false; - final eventOrders = orders ?? _profileEventOrders; final pubkey = event.pubkey.toLowerCase(); - final current = eventOrders[pubkey]; - final isNewer = - current == null || - event.createdAt > current.createdAt || - (event.createdAt == current.createdAt && - event.id.compareTo(current.eventId) < 0); - if (!isNewer) return false; - - profiles[pubkey] = _profileFromEvent(event); - eventOrders[pubkey] = (createdAt: event.createdAt, eventId: event.id); + if (!_isNewer(pubkey, event.createdAt, event.id)) return false; + profiles[pubkey] = parseProfileEvent(event).profile; + _profileEventOrders[pubkey] = ( + createdAt: event.createdAt, + eventId: event.id, + ); return true; } +} - UserProfile _profileFromEvent(NostrEvent event) { - final data = ProfileData.fromEvent(event); - final pubkey = data.pubkey.toLowerCase(); - return UserProfile( - pubkey: pubkey, - displayName: data.displayName, - avatarUrl: data.avatarUrl, - about: data.about, - nip05Handle: data.nip05, - ownerPubkey: verifiedOaOwnerPubkey(event.tags, event.pubkey), +/// Profile history parser, isolated from provider state so batch completion can +/// be delayed independently of relay delivery. +final profileEventBatchParserProvider = + Provider> Function(List)>( + (ref) => parseProfileEventBatch, ); - } -} final userCacheProvider = NotifierProvider>( diff --git a/mobile/test/shared/profile/user_cache_provider_test.dart b/mobile/test/shared/profile/user_cache_provider_test.dart index 9a270861076..2871514ad9d 100644 --- a/mobile/test/shared/profile/user_cache_provider_test.dart +++ b/mobile/test/shared/profile/user_cache_provider_test.dart @@ -3,6 +3,7 @@ import 'dart:convert'; import 'dart:typed_data'; import 'package:buzz/shared/profile/user_cache_provider.dart'; +import 'package:buzz/shared/profile/profile_event_parser.dart'; import 'package:buzz/shared/relay/relay.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -140,6 +141,297 @@ void main() { expect(cache.state['agent']?.displayName, 'Valid'); }); + for (final preload in [false, true]) { + test( + '${preload ? 'preload' : 'refresh'} merges after parsing against newer live state', + () async { + final started = Completer(); + final release = Completer(); + final session = _RecordingProfileSession( + result: Future.value([ + _profileEvent(id: 'old', createdAt: 1, name: 'Old'), + ]), + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + profileEventBatchParserProvider.overrideWithValue((events) async { + final parsed = await parseProfileEventBatch(events); + started.complete(); + await release.future; + return parsed; + }), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + final result = preload + ? cache.preload(['agent']) + : cache.refresh(['agent']); + await started.future; + cache.cacheProfileEvent( + _profileEvent(id: 'new', createdAt: 2, name: 'Live'), + ); + cache.cacheProfileEvent( + _profileEvent( + id: 'other', + pubkey: 'other', + createdAt: 1, + name: 'Other', + ), + ); + release.complete(); + expect(await result, isTrue); + expect(cache.state['agent']?.displayName, 'Live'); + expect(cache.state['other']?.displayName, 'Other'); + }, + ); + + test( + '${preload ? 'preload' : 'refresh'} rejects a retired parsing completion', + () async { + final started = Completer(); + final release = Completer(); + final session = _RecordingProfileSession( + result: Future.value([ + _profileEvent(id: 'old', createdAt: 1, name: 'Previous community'), + ]), + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + profileEventBatchParserProvider.overrideWithValue((events) async { + final parsed = await parseProfileEventBatch(events); + started.complete(); + await release.future; + return parsed; + }), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + final result = preload + ? cache.preload(['agent']) + : cache.refresh(['agent']); + await started.future; + // This is the same dependency invalidation caused by community config. + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://next-community.invalid'); + container.read(userCacheProvider); + release.complete(); + expect(await result, isFalse); + expect(container.read(userCacheProvider), isEmpty); + }, + ); + } + + test( + 'community change rejects an old network response before parsing', + () async { + final response = Completer>(); + var parsed = false; + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith( + () => _RecordingProfileSession(result: response.future), + ), + profileEventBatchParserProvider.overrideWithValue((events) async { + parsed = true; + return parseProfileEventBatch(events); + }), + ], + ); + addTearDown(container.dispose); + final result = container.read(userCacheProvider.notifier).refresh([ + 'agent', + ]); + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://next-community.invalid'); + container.read(userCacheProvider); + response.complete([_profileEvent(id: 'old', createdAt: 1, name: 'Old')]); + expect(await result, isFalse); + expect(parsed, isFalse); + expect(container.read(userCacheProvider), isEmpty); + }, + ); + + test( + 'parser failure reports refresh failure without modifying cache', + () async { + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith( + () => _RecordingProfileSession( + result: Future.value([ + _profileEvent(id: 'next', createdAt: 2, name: 'Next'), + ]), + ), + ), + profileEventBatchParserProvider.overrideWithValue( + (_) => Future.error(StateError('parse failed')), + ), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + cache.cacheProfileEvent( + _profileEvent(id: 'cached', createdAt: 1, name: 'Cached'), + ); + expect(await cache.refresh(['agent']), isFalse); + expect(cache.state['agent']?.displayName, 'Cached'); + }, + ); + + test('later refresh completion cannot roll back a newer refresh', () async { + final oldResponse = Completer>(); + final session = _RecordingProfileSession( + results: [ + oldResponse.future, + Future.value([_profileEvent(id: 'new', createdAt: 2, name: 'New')]), + ], + ); + final container = ProviderContainer( + overrides: [relaySessionProvider.overrideWith(() => session)], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + final old = cache.refresh(['agent']); + expect(await cache.refresh(['agent']), isTrue); + oldResponse.complete([_profileEvent(id: 'old', createdAt: 1, name: 'Old')]); + expect(await old, isTrue); + expect(cache.state['agent']?.displayName, 'New'); + }); + + test('stale malformed profile is skipped before parsing', () async { + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith( + () => _RecordingProfileSession( + result: Future.value([ + NostrEvent( + id: 'old', + pubkey: 'agent', + createdAt: 1, + kind: 0, + tags: [], + content: '{"name":42}', + sig: 'sig', + ), + ]), + ), + ), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + cache.cacheProfileEvent( + _profileEvent(id: 'new', createdAt: 2, name: 'New'), + ); + expect(await cache.refresh(['agent']), isTrue); + expect(cache.state['agent']?.displayName, 'New'); + }); + + test( + 'refresh and preload share one verification worker and coalesce pending lookups', + () async { + final started = Completer(); + final release = Completer(); + var running = 0; + var maxRunning = 0; + var calls = 0; + final session = _RecordingProfileSession( + results: [ + Future.value([ + _profileEvent(id: 'a', pubkey: 'a', createdAt: 1, name: 'A'), + ]), + Future.value([ + _profileEvent(id: 'b', pubkey: 'b', createdAt: 1, name: 'B'), + ]), + Future.value([ + _profileEvent(id: 'c', pubkey: 'c', createdAt: 1, name: 'C'), + ]), + ], + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + profileEventBatchParserProvider.overrideWithValue((events) async { + running++; + if (running > maxRunning) maxRunning = running; + if (calls++ == 0) { + started.complete(); + await release.future; + } + final parsed = await parseProfileEventBatch(events); + running--; + return parsed; + }), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + final first = cache.preload(['a']); + await started.future; + final refresh = cache.refresh(['b']); + final next = cache.preload(['c']); + final duplicate = cache.preload(['c']); + // Let the refresh reach the worker queue before releasing its predecessor. + await Future.delayed(Duration.zero); + expect(calls, 1); + release.complete(); + expect( + await Future.wait([first, refresh, next, duplicate]), + everyElement(isTrue), + ); + expect(maxRunning, 1); + expect(calls, 3); + expect(session.requestCount, 3); + }, + ); + + test( + 'refresh verifies real OA batch while allowing main-isolate timers', + () async { + final owner = nostr.Keys.generate(); + final agents = List.generate(16, (_) => nostr.Keys.generate()); + final events = [ + for (final agent in agents) + _profileEvent( + id: agent.public, + pubkey: agent.public, + createdAt: 1, + name: 'Synthetic', + tags: [_authTag(owner, agent.public)], + ), + ]; + var timerRanBeforeParsingCompleted = false; + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith( + () => _RecordingProfileSession(result: Future.value(events)), + ), + profileEventBatchParserProvider.overrideWithValue((events) async { + var timerRan = false; + final timer = Timer(Duration.zero, () => timerRan = true); + final parsed = await parseProfileEventBatch(events); + timerRanBeforeParsingCompleted = timerRan; + timer.cancel(); + return parsed; + }), + ], + ); + addTearDown(container.dispose); + final cache = container.read(userCacheProvider.notifier); + expect(await cache.refresh(agents.map((a) => a.public).toList()), isTrue); + expect(timerRanBeforeParsingCompleted, isTrue); + for (final agent in agents) { + expect(cache.state[agent.public]?.ownerPubkey, owner.public); + } + }, + ); + test('same-second profile tie keeps the lowest event id', () { final container = ProviderContainer(); addTearDown(container.dispose); @@ -200,6 +492,7 @@ class _RecordingProfileSession extends RelaySessionNotifier { final List>> _results; NostrFilter? requestedFilter; + int requestCount = 0; @override SessionState build() => const SessionState(status: SessionStatus.connected); @@ -209,6 +502,7 @@ class _RecordingProfileSession extends RelaySessionNotifier { NostrFilter filter, { Duration timeout = const Duration(seconds: 8), }) async { + requestCount++; requestedFilter = filter; return _results.isEmpty ? const [] : _results.removeAt(0); } From f50e9fd8128a2bc0c7655bda2faad43ee2a448be Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Tue, 15 Sep 2026 11:34:17 -0700 Subject: [PATCH 2/2] Preserve confirmed profile saves across pending batches Signed-off-by: Tom Brow --- .../features/profile/profile_provider.dart | 4 +- .../profile/profile_provider_test.dart | 71 +++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/mobile/lib/features/profile/profile_provider.dart b/mobile/lib/features/profile/profile_provider.dart index 93621f7608a..cedda770051 100644 --- a/mobile/lib/features/profile/profile_provider.dart +++ b/mobile/lib/features/profile/profile_provider.dart @@ -236,7 +236,9 @@ class ProfileNotifier extends AsyncNotifier { ownerPubkey: verifiedOaOwnerPubkey(submittedEvent.tags, pubkey), ); state = AsyncData(profile); - ref.read(userCacheProvider.notifier).put(profile); + // Record the confirmed event order with the profile so an older batch + // completing after this save cannot replace it. + ref.read(userCacheProvider.notifier).cacheProfileEvent(submittedEvent); } void _requireCurrentWriteContext(_ProfileWriteContext context) { diff --git a/mobile/test/features/profile/profile_provider_test.dart b/mobile/test/features/profile/profile_provider_test.dart index 93c68cfa202..daa6c88cab6 100644 --- a/mobile/test/features/profile/profile_provider_test.dart +++ b/mobile/test/features/profile/profile_provider_test.dart @@ -3,6 +3,7 @@ import 'dart:convert'; import 'dart:typed_data'; import 'package:buzz/features/profile/profile_provider.dart'; +import 'package:buzz/shared/profile/profile_event_parser.dart'; import 'package:buzz/shared/profile/user_cache_provider.dart'; import 'package:buzz/shared/profile/user_profile.dart'; import 'package:buzz/shared/relay/relay.dart'; @@ -78,6 +79,76 @@ void main() { ); }); + test('confirmed profile save survives an older in-flight batch', () async { + final keys = nostr.Keys.generate(); + final owner = nostr.Keys.generate(); + NostrEvent event(String id, int createdAt, {bool owned = false}) => + NostrEvent( + id: id, + pubkey: keys.public, + createdAt: createdAt, + kind: EventKind.profile, + tags: owned ? [_authTag(owner, keys.public)] : const [], + content: jsonEncode({ + 'display_name': owned ? 'Current' : 'Old relay profile', + 'picture': owned ? 'https://example.com/new.png' : 'old.png', + 'about': owned ? 'Current about' : 'Old about', + 'nip05': owned ? 'agent@example.com' : 'old@example.com', + }), + sig: 'sig', + ); + var history = [event('old', 1)]; + final relaySession = _ControlledProfileRelaySession( + fetch: () async => history, + ); + final parserStarted = Completer(); + final releaseParser = Completer(); + final container = ProviderContainer( + overrides: [ + relayConfigProvider.overrideWith( + () => _FixedRelayConfigNotifier(keys.nsec), + ), + relaySessionProvider.overrideWith(() => relaySession), + profileEventBatchParserProvider.overrideWithValue((events) async { + final parsed = await parseProfileEventBatch(events); + parserStarted.complete(); + await releaseParser.future; + return parsed; + }), + ], + ); + addTearDown(container.dispose); + + await container.read(profileProvider.future); + final cache = container.read(userCacheProvider.notifier); + final refresh = cache.refresh([keys.public]); + await parserStarted.future; + history = [event('current', 2, owned: true)]; + await container + .read(profileProvider.notifier) + .updateDisplayName('Saved locally'); + final saved = container.read(userCacheProvider)[keys.public]!; + expect(saved.displayName, 'Saved locally'); + expect(saved.avatarUrl, 'https://example.com/new.png'); + expect(saved.about, 'Current about'); + expect(saved.nip05Handle, 'agent@example.com'); + expect(saved.ownerPubkey, owner.public.toLowerCase()); + expect(relaySession.published, hasLength(1)); + + releaseParser.complete(); + expect(await refresh, isTrue); + final afterBatch = container.read(userCacheProvider)[keys.public]!; + expect(afterBatch.displayName, 'Saved locally'); + expect(afterBatch.avatarUrl, saved.avatarUrl); + expect(afterBatch.about, saved.about); + expect(afterBatch.nip05Handle, saved.nip05Handle); + expect(afterBatch.ownerPubkey, owner.public.toLowerCase()); + expect( + container.read(profileProvider).requireValue?.ownerPubkey, + owner.public.toLowerCase(), + ); + }); + test('clearing a display name restores the pubkey label fallback', () async { final keys = nostr.Keys.generate(); final relaySession = _ProfileRelaySession(