diff --git a/mobile/lib/features/channels/channel_mutes/channel_mutes_manager.dart b/mobile/lib/features/channels/channel_mutes/channel_mutes_manager.dart index 32c163a1bb..683b1b5d41 100644 --- a/mobile/lib/features/channels/channel_mutes/channel_mutes_manager.dart +++ b/mobile/lib/features/channels/channel_mutes/channel_mutes_manager.dart @@ -45,6 +45,17 @@ class ChannelMutesManager { void Function()? _unsubscribe; bool _disposed = false; + /// Base delay for the recovery-read retry backoff. Overridable in tests. + final Duration _startupRetryBaseDelay; + Timer? _startupRetryTimer; + int _startupRetryAttempt = 0; + bool _headApplied = false; + bool _liveSettled = false; + int _localRevision = 0; + bool _publishPending = false; + Future? _syncInFlight; + bool _syncAgain = false; + ChannelMutesManager({ required this.pubkey, required SharedPreferences prefs, @@ -53,12 +64,15 @@ class ChannelMutesManager { required SignedEventRelay? signedEventRelay, required bool remoteEnabled, required VoidCallback onChanged, + @visibleForTesting + Duration startupRetryBaseDelay = const Duration(seconds: 2), }) : _storage = ChannelMutesStorage(prefs), _crypto = crypto, _relaySession = relaySession, _signedEventRelay = signedEventRelay, _remoteEnabled = remoteEnabled, _onChanged = onChanged, + _startupRetryBaseDelay = startupRetryBaseDelay, _store = ChannelMutesStorage(prefs).read(pubkey); ChannelMuteStore get store => _store; @@ -71,14 +85,84 @@ class ChannelMutesManager { return; } - await _fetchAndMerge(); - await _startLiveSubscription(); + await _syncWithRelay(); _onChanged(); } + /// Re-reads the retained head once, e.g. on app foreground resume, to catch + /// an EVENT a healthy socket never delivered. One shot: skipped while an + /// edit is pending, not retried on failure, and leaves startup state alone. + void refreshFromRelay() { + if (_disposed || !_remoteEnabled || _relaySession == null) return; + unawaited(_recoverHead()); + } + + /// Applies the retained head and opens the live subscription, retrying with + /// bounded backoff (2s base, 30s ceiling) until both are done. Transient + /// live-subscription CLOSED recovery belongs to [RelaySessionNotifier]. + Future _syncWithRelay() { + if (_disposed) return Future.value(); + final inFlight = _syncInFlight; + if (inFlight != null) { + _syncAgain = true; + return inFlight; + } + final sync = _runSyncWithRelay(); + _syncInFlight = sync; + return sync.whenComplete(() { + _syncInFlight = null; + if (_disposed || !_syncAgain) return; + _syncAgain = false; + unawaited(_syncWithRelay()); + }); + } + + Future _runSyncWithRelay() async { + if (!_headApplied) { + final applied = await _recoverHead(); + if (_disposed) return; + _headApplied = applied; + } + + final subscribed = _liveSettled || await _startLiveSubscription(); + if (_disposed) return; + + if (!_headApplied || !subscribed) { + _scheduleStartupRetry(); + } else { + _startupRetryAttempt = 0; + } + } + + void _scheduleStartupRetry() { + if (_disposed) return; + _startupRetryTimer?.cancel(); + final delayMs = min( + _startupRetryBaseDelay.inMilliseconds << min(_startupRetryAttempt, 5), + 30000, + ); + _startupRetryAttempt++; + debugPrint( + '[ChannelMutesManager] relay sync incomplete; ' + 'retrying in ${delayMs}ms (attempt $_startupRetryAttempt)', + ); + _startupRetryTimer = Timer(Duration(milliseconds: delayMs), () { + _startupRetryTimer = null; + unawaited( + _syncWithRelay().then((_) { + if (!_disposed) _onChanged(); + }), + ); + }); + } + void dispose({bool flushPending = true}) { if (_disposed) return; _disposed = true; + _syncAgain = false; + + _startupRetryTimer?.cancel(); + _startupRetryTimer = null; final hadPending = _publishDebounce != null; _publishDebounce?.cancel(); @@ -99,7 +183,7 @@ class ChannelMutesManager { updatedAt: currentUnixSeconds(), ); _store = ChannelMuteStore(channels: {..._store.channels, channelId: entry}); - _persist(); + unawaited(_persist()); _onChanged(); markDirty(); } @@ -111,101 +195,126 @@ class ChannelMutesManager { updatedAt: currentUnixSeconds(), ); _store = ChannelMuteStore(channels: {..._store.channels, channelId: entry}); - _persist(); + unawaited(_persist()); _onChanged(); markDirty(); } void markDirty() { if (!_remoteEnabled || _disposed) return; + _localRevision++; + _publishPending = true; _publishDebounce?.cancel(); _publishDebounce = Timer(const Duration(seconds: 5), () { _publishDebounce = null; - unawaited(_publish()); + unawaited( + _publish().whenComplete( + () => _publishPending = _publishDebounce != null, + ), + ); }); } - Future _fetchAndMerge() async { - if (_relaySession == null) return; + /// A recovery read that loses to a local edit is deferred, never applied: + /// a peer entry stamped by a skewed clock could otherwise overwrite it. + Future _recoverHead() async { + if (_publishPending) return false; + final revision = _localRevision; + final events = await _fetchHead(); + if (events == null || _disposed) return false; + if (revision != _localRevision || _publishPending) return false; + final applied = await _applyEvents(events); + if (!_disposed) _onChanged(); + return applied; + } + + Future?> _fetchHead() async { try { - final events = await _relaySession.fetchHistory( - NostrFilter( - kinds: const [EventKind.readState], - authors: [pubkey], - tags: const { - '#d': ['channel-mutes'], - }, - limit: 1, - ), - ); - _mergeEvents(events); - _persist(); - if (!_disposed) _onChanged(); - } catch (_) { - // Local state remains usable when relay is unavailable. + return await _relaySession!.fetchHistory(_filter()); + } catch (error) { + debugPrint('[ChannelMutesManager] fetch failed: $error'); + return null; } } - Future _startLiveSubscription() async { - if (_relaySession == null) return; + Future _startLiveSubscription() async { try { - _unsubscribe = await _relaySession.subscribe( - NostrFilter( - kinds: const [EventKind.readState], - authors: [pubkey], - tags: const { - '#d': ['channel-mutes'], - }, - limit: 1, - ), + final unsubscribe = await _relaySession!.subscribe( + _filter(), _handleIncomingEvent, + // Only terminal rejections reach here; re-sending the same REQ would + // be rejected again, so the live lane stays settled. + onClosed: (message) { + _liveSettled = true; + debugPrint( + '[ChannelMutesManager] live subscription closed: $message', + ); + }, ); - } catch (_) { - // Non-fatal — local state and history still work. + if (_disposed) { + unsubscribe(); + return false; + } + _unsubscribe = unsubscribe; + return _liveSettled = true; + } catch (error) { + debugPrint('[ChannelMutesManager] live subscription failed: $error'); + return _liveSettled; } } - void _mergeEvents(List events) { + NostrFilter _filter() => NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: const { + '#d': ['channel-mutes'], + }, + limit: 1, + ); + + /// Relay retains `ORDER BY created_at DESC, id ASC`: at equal second the + /// lower event ID wins. + bool _isAfterCursor(int createdAt, String id) => + createdAt > _lastRemoteCreatedAt || + (createdAt == _lastRemoteCreatedAt && + id.compareTo(_lastRemoteEventId ?? '') < 0); + + /// Returns false only when a newer head could not be persisted; the cursor + /// then stays put so the same head is retried. An undecodable head is + /// skipped without holding startup open. + Future _applyEvents(List events) async { + var applied = true; for (final event in events) { - if (event.pubkey != pubkey) continue; - _mergeEvent(event); - } - } - - void _mergeEvent(NostrEvent event) { - // Only process channel-mutes d-tag events. - final dTag = event.getTagValue('d'); - if (dTag != 'channel-mutes') return; - - try { - final plaintext = _crypto.decrypt(event.content); - final parsed = jsonDecode(plaintext); - if (parsed is! Map) return; - - final incoming = ChannelMuteStore.fromJson(parsed); - - // Gate on createdAt: ignore events older than what we've already seen. - final isNewer = - event.createdAt > _lastRemoteCreatedAt || - (event.createdAt == _lastRemoteCreatedAt && - event.id.compareTo(_lastRemoteEventId ?? '') > 0); - - if (isNewer) { + if (event.pubkey != pubkey || event.getTagValue('d') != 'channel-mutes') { + continue; + } + if (!_isAfterCursor(event.createdAt, event.id)) continue; + try { + final parsed = jsonDecode(_crypto.decrypt(event.content)); + if (parsed is! Map) throw const FormatException(); + // Per-channel merge: keep the entry with the highest updatedAt. + _store = mergeStores(_store, ChannelMuteStore.fromJson(parsed)); + } catch (error) { + debugPrint('[ChannelMutesManager] undecodable head: $error'); + continue; + } + if (!await _persist()) { + applied = false; + continue; + } + if (_isAfterCursor(event.createdAt, event.id)) { _lastRemoteCreatedAt = event.createdAt; _lastRemoteEventId = event.id; - // Per-channel merge: keep the entry with the highest updatedAt for each channel. - _store = mergeStores(_store, incoming); - _persist(); } - } catch (_) { - // Decryption failure or parse error — keep existing state. } + return applied; } void _handleIncomingEvent(NostrEvent event) { if (_disposed) return; - _mergeEvent(event); - if (!_disposed) _onChanged(); + unawaited( + _applyEvents([event]).then((_) => _disposed ? null : _onChanged()), + ); } bool _isIdenticalToLastPublished() { @@ -232,7 +341,9 @@ class ChannelMutesManager { } // Read-before-write: merge remote state before publishing - await _fetchAndMerge(); + final events = await _fetchHead(); + if (events != null) await _applyEvents(events); + if (!_disposed) _onChanged(); // No-op suppression: skip if nothing changed if (_isIdenticalToLastPublished()) return; @@ -242,6 +353,7 @@ class ChannelMutesManager { final ciphertext = _crypto.encrypt(payload); final createdAt = max(currentUnixSeconds(), _lastRemoteCreatedAt + 1); + String? signedId; await _signedEventRelay.submit( kind: EventKind.readState, content: ciphertext, @@ -250,16 +362,29 @@ class ChannelMutesManager { ['t', 'channel-mutes'], ], createdAt: createdAt, + onSigned: (event) => signedId = event.id, ); - _lastRemoteCreatedAt = max(_lastRemoteCreatedAt, createdAt); + // Keep the cursor a coherent (created_at, id) pair so an OK that beats + // its own echo cannot make later same-second heads compare against a + // stale ID. A newer head learned during the await stays. + if (signedId != null && _isAfterCursor(createdAt, signedId!)) { + _lastRemoteCreatedAt = createdAt; + _lastRemoteEventId = signedId; + } _lastPublishedStore = ChannelMuteStore(channels: Map.of(_store.channels)); } catch (error) { debugPrint('[ChannelMutesManager] publish failed: $error'); } } - void _persist() { - _storage.write(pubkey, _store); + Future _persist() async { + try { + if (await _storage.write(pubkey, _store)) return true; + debugPrint('[ChannelMutesManager] persist returned false'); + } catch (error) { + debugPrint('[ChannelMutesManager] persist failed: $error'); + } + return false; } } diff --git a/mobile/lib/features/channels/channel_mutes/channel_mutes_provider.dart b/mobile/lib/features/channels/channel_mutes/channel_mutes_provider.dart index cd5e41da1a..f58eabcacb 100644 --- a/mobile/lib/features/channels/channel_mutes/channel_mutes_provider.dart +++ b/mobile/lib/features/channels/channel_mutes/channel_mutes_provider.dart @@ -1,3 +1,4 @@ +import 'package:flutter/widgets.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:nostr/nostr.dart' as nostr; @@ -69,6 +70,11 @@ class ChannelMutesNotifier extends Notifier { ); _manager = manager; + // Resume re-read catches an EVENT a healthy socket never delivered. + ref.listen(appLifecycleProvider, (_, next) { + if (next == AppLifecycleState.resumed) manager.refreshFromRelay(); + }); + ref.onDispose(() { manager.dispose(); if (_manager == manager) { diff --git a/mobile/lib/features/channels/channel_mutes/channel_mutes_storage.dart b/mobile/lib/features/channels/channel_mutes/channel_mutes_storage.dart index 28e7f50e19..c37669c3eb 100644 --- a/mobile/lib/features/channels/channel_mutes/channel_mutes_storage.dart +++ b/mobile/lib/features/channels/channel_mutes/channel_mutes_storage.dart @@ -85,7 +85,6 @@ class ChannelMutesStorage { } } - void write(String pubkey, ChannelMuteStore store) { - _prefs.setString(channelMutesKey(pubkey), jsonEncode(store.toJson())); - } + Future write(String pubkey, ChannelMuteStore store) => + _prefs.setString(channelMutesKey(pubkey), jsonEncode(store.toJson())); } diff --git a/mobile/lib/features/channels/channel_sections/channel_sections_manager.dart b/mobile/lib/features/channels/channel_sections/channel_sections_manager.dart index e36a514ff6..2e4967da0f 100644 --- a/mobile/lib/features/channels/channel_sections/channel_sections_manager.dart +++ b/mobile/lib/features/channels/channel_sections/channel_sections_manager.dart @@ -56,6 +56,8 @@ class ChannelSectionsManager { Future? _syncInFlight; bool _syncAgain = false; int _subscriptionGeneration = 0; + int _localRevision = 0; + bool _publishPending = false; ChannelSectionsManager({ required this.pubkey, @@ -90,6 +92,19 @@ class ChannelSectionsManager { _onChanged(); } + /// Re-reads the retained head once, e.g. on app foreground resume, to catch + /// an EVENT a healthy socket never delivered. Skipped while a local edit is + /// pending and discarded if one lands before the read returns. + void refreshFromRelay() { + if (_disposed || !_remoteEnabled || _publishPending) return; + final revision = _localRevision; + unawaited( + _fetchAndMerge( + isCurrent: () => revision == _localRevision && !_publishPending, + ), + ); + } + /// One startup-sync attempt: fetch the remote blob, then start the live /// subscription. Either step can lose a transient race on cold start (the /// relay rate-limits the burst of per-channel subscriptions and rejects @@ -287,16 +302,25 @@ class ChannelSectionsManager { void markDirty() { if (!_remoteEnabled || _disposed) return; + _localRevision++; + _publishPending = true; _publishDebounce?.cancel(); _publishDebounce = Timer(const Duration(seconds: 5), () { _publishDebounce = null; - unawaited(_publish()); + unawaited( + _publish().whenComplete( + () => _publishPending = _publishDebounce != null, + ), + ); }); } /// Returns whether the fetch reached the relay (regardless of whether a /// remote blob exists). - Future _fetchAndMerge({bool allowDisposed = false}) async { + Future _fetchAndMerge({ + bool allowDisposed = false, + bool Function()? isCurrent, + }) async { if (_relaySession == null) return false; try { final events = await _relaySession.fetchHistory( @@ -310,6 +334,7 @@ class ChannelSectionsManager { ), ); if (_disposed && !allowDisposed) return false; + if (isCurrent != null && !isCurrent()) return false; _mergeEvents(events); _persist(); if (!_disposed) _onChanged(); @@ -386,13 +411,7 @@ class ChannelSectionsManager { final incoming = ChannelSectionStore.fromJson(parsed); - // Last-write-wins: newer createdAt wins; tie-break by event ID. - final isNewer = - event.createdAt > _lastRemoteCreatedAt || - (event.createdAt == _lastRemoteCreatedAt && - event.id.compareTo(_lastRemoteEventId ?? '') > 0); - - if (isNewer) { + if (_isAfterCursor(event.createdAt, event.id)) { _lastRemoteCreatedAt = event.createdAt; _lastRemoteEventId = event.id; _store = incoming; @@ -403,6 +422,13 @@ class ChannelSectionsManager { } } + /// Last-write-wins. Relay retains `ORDER BY created_at DESC, id ASC`: at + /// equal second the lower event ID wins. + bool _isAfterCursor(int createdAt, String id) => + createdAt > _lastRemoteCreatedAt || + (createdAt == _lastRemoteCreatedAt && + id.compareTo(_lastRemoteEventId ?? '') < 0); + void _handleIncomingEvent(NostrEvent event) { if (_disposed) return; _mergeEvent(event); @@ -443,11 +469,14 @@ class ChannelSectionsManager { // No-op suppression: skip if nothing changed if (_isIdenticalToLastPublished()) return; + final submitted = _store; + final revision = _localRevision; try { - final payload = jsonEncode(_store.toJson()); + final payload = jsonEncode(submitted.toJson()); final ciphertext = _crypto.encrypt(payload); final createdAt = max(currentUnixSeconds(), _lastRemoteCreatedAt + 1); + String? signedId; await _signedEventRelay.submit( kind: EventKind.readState, content: ciphertext, @@ -456,12 +485,26 @@ class ChannelSectionsManager { ['t', 'channel-sections'], ], createdAt: createdAt, + onSigned: (event) => signedId = event.id, ); - _lastRemoteCreatedAt = max(_lastRemoteCreatedAt, createdAt); + // Keep the cursor a coherent (created_at, id) pair so an OK that beats + // its own echo cannot make later same-second heads compare against a + // stale ID. A newer head learned during the await stays. + if (signedId != null && _isAfterCursor(createdAt, signedId!)) { + _lastRemoteCreatedAt = createdAt; + _lastRemoteEventId = signedId; + // A same-second loser merged during the await must not strand the + // device on content the relay did not keep. + if (revision == _localRevision && !identical(_store, submitted)) { + _store = submitted; + _persist(); + if (!_disposed) _onChanged(); + } + } _lastPublishedStore = ChannelSectionStore( - sections: List.of(_store.sections), - assignments: Map.of(_store.assignments), + sections: List.of(submitted.sections), + assignments: Map.of(submitted.assignments), ); } catch (error) { debugPrint('[ChannelSectionsManager] publish failed: $error'); diff --git a/mobile/lib/features/channels/channel_sections/channel_sections_provider.dart b/mobile/lib/features/channels/channel_sections/channel_sections_provider.dart index 2caa35c25f..6466e6a4d4 100644 --- a/mobile/lib/features/channels/channel_sections/channel_sections_provider.dart +++ b/mobile/lib/features/channels/channel_sections/channel_sections_provider.dart @@ -1,3 +1,4 @@ +import 'package:flutter/widgets.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:nostr/nostr.dart' as nostr; @@ -69,6 +70,11 @@ class ChannelSectionsNotifier extends Notifier { ); _manager = manager; + // Resume re-read catches an EVENT a healthy socket never delivered. + ref.listen(appLifecycleProvider, (_, next) { + if (next == AppLifecycleState.resumed) manager.refreshFromRelay(); + }); + ref.onDispose(() { manager.dispose(); if (_manager == manager) { diff --git a/mobile/lib/features/channels/channel_sort/channel_sort_manager.dart b/mobile/lib/features/channels/channel_sort/channel_sort_manager.dart index a434fec17b..9f222967f3 100644 --- a/mobile/lib/features/channels/channel_sort/channel_sort_manager.dart +++ b/mobile/lib/features/channels/channel_sort/channel_sort_manager.dart @@ -113,6 +113,23 @@ class ChannelSortManager { } } + /// Re-reads the retained head once, e.g. on app foreground resume, to catch + /// an EVENT a healthy socket never delivered. Skipped while a local edit is + /// pending and discarded if a mutation lands before the read returns. + void refreshFromRelay() { + if (_disposed || !_remoteEnabled || _syncState.hasPendingLocalChanges) { + return; + } + final generation = _generation; + unawaited( + _fetchAndApply( + isCurrent: () => !_disposed && generation == _generation, + ).then((_) { + if (!_disposed && generation != _generation) _onChanged(); + }), + ); + } + void _scheduleStartupRetry() { if (_disposed) return; _startupRetryTimer?.cancel(); @@ -169,10 +186,11 @@ class ChannelSortManager { }); } - Future _fetchAndApply() async { + Future _fetchAndApply({bool Function()? isCurrent}) async { if (_relaySession == null) return null; try { final events = await _relaySession.fetchHistory(_filter()); + if (isCurrent != null && !isCurrent()) return null; var found = false; for (final event in events) { if (event.pubkey != pubkey || event.getTagValue('d') != _dTag) continue; diff --git a/mobile/lib/features/channels/channel_sort/channel_sort_provider.dart b/mobile/lib/features/channels/channel_sort/channel_sort_provider.dart index 5a5d144660..4fdbbe00f3 100644 --- a/mobile/lib/features/channels/channel_sort/channel_sort_provider.dart +++ b/mobile/lib/features/channels/channel_sort/channel_sort_provider.dart @@ -1,3 +1,4 @@ +import 'package:flutter/widgets.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:nostr/nostr.dart' as nostr; @@ -77,6 +78,11 @@ class ChannelSortNotifier extends Notifier { ); _manager = manager; + // Resume re-read catches an EVENT a healthy socket never delivered. + ref.listen(appLifecycleProvider, (_, next) { + if (next == AppLifecycleState.resumed) manager.refreshFromRelay(); + }); + ref.onDispose(() { manager.dispose(); if (_manager == manager) { diff --git a/mobile/lib/features/channels/channel_stars/channel_stars_manager.dart b/mobile/lib/features/channels/channel_stars/channel_stars_manager.dart index 94fba0acb9..c98c7da2d6 100644 --- a/mobile/lib/features/channels/channel_stars/channel_stars_manager.dart +++ b/mobile/lib/features/channels/channel_stars/channel_stars_manager.dart @@ -45,6 +45,17 @@ class ChannelStarsManager { void Function()? _unsubscribe; bool _disposed = false; + /// Base delay for the recovery-read retry backoff. Overridable in tests. + final Duration _startupRetryBaseDelay; + Timer? _startupRetryTimer; + int _startupRetryAttempt = 0; + bool _headApplied = false; + bool _liveSettled = false; + int _localRevision = 0; + bool _publishPending = false; + Future? _syncInFlight; + bool _syncAgain = false; + ChannelStarsManager({ required this.pubkey, required SharedPreferences prefs, @@ -53,12 +64,15 @@ class ChannelStarsManager { required SignedEventRelay? signedEventRelay, required bool remoteEnabled, required VoidCallback onChanged, + @visibleForTesting + Duration startupRetryBaseDelay = const Duration(seconds: 2), }) : _storage = ChannelStarsStorage(prefs), _crypto = crypto, _relaySession = relaySession, _signedEventRelay = signedEventRelay, _remoteEnabled = remoteEnabled, _onChanged = onChanged, + _startupRetryBaseDelay = startupRetryBaseDelay, _store = ChannelStarsStorage(prefs).read(pubkey); ChannelStarStore get store => _store; @@ -71,14 +85,84 @@ class ChannelStarsManager { return; } - await _fetchAndMerge(); - await _startLiveSubscription(); + await _syncWithRelay(); _onChanged(); } + /// Re-reads the retained head once, e.g. on app foreground resume, to catch + /// an EVENT a healthy socket never delivered. One shot: skipped while an + /// edit is pending, not retried on failure, and leaves startup state alone. + void refreshFromRelay() { + if (_disposed || !_remoteEnabled || _relaySession == null) return; + unawaited(_recoverHead()); + } + + /// Applies the retained head and opens the live subscription, retrying with + /// bounded backoff (2s base, 30s ceiling) until both are done. Transient + /// live-subscription CLOSED recovery belongs to [RelaySessionNotifier]. + Future _syncWithRelay() { + if (_disposed) return Future.value(); + final inFlight = _syncInFlight; + if (inFlight != null) { + _syncAgain = true; + return inFlight; + } + final sync = _runSyncWithRelay(); + _syncInFlight = sync; + return sync.whenComplete(() { + _syncInFlight = null; + if (_disposed || !_syncAgain) return; + _syncAgain = false; + unawaited(_syncWithRelay()); + }); + } + + Future _runSyncWithRelay() async { + if (!_headApplied) { + final applied = await _recoverHead(); + if (_disposed) return; + _headApplied = applied; + } + + final subscribed = _liveSettled || await _startLiveSubscription(); + if (_disposed) return; + + if (!_headApplied || !subscribed) { + _scheduleStartupRetry(); + } else { + _startupRetryAttempt = 0; + } + } + + void _scheduleStartupRetry() { + if (_disposed) return; + _startupRetryTimer?.cancel(); + final delayMs = min( + _startupRetryBaseDelay.inMilliseconds << min(_startupRetryAttempt, 5), + 30000, + ); + _startupRetryAttempt++; + debugPrint( + '[ChannelStarsManager] relay sync incomplete; ' + 'retrying in ${delayMs}ms (attempt $_startupRetryAttempt)', + ); + _startupRetryTimer = Timer(Duration(milliseconds: delayMs), () { + _startupRetryTimer = null; + unawaited( + _syncWithRelay().then((_) { + if (!_disposed) _onChanged(); + }), + ); + }); + } + void dispose({bool flushPending = true}) { if (_disposed) return; _disposed = true; + _syncAgain = false; + + _startupRetryTimer?.cancel(); + _startupRetryTimer = null; final hadPending = _publishDebounce != null; _publishDebounce?.cancel(); @@ -99,7 +183,7 @@ class ChannelStarsManager { updatedAt: currentUnixSeconds(), ); _store = ChannelStarStore(channels: {..._store.channels, channelId: entry}); - _persist(); + unawaited(_persist()); _onChanged(); markDirty(); } @@ -111,101 +195,126 @@ class ChannelStarsManager { updatedAt: currentUnixSeconds(), ); _store = ChannelStarStore(channels: {..._store.channels, channelId: entry}); - _persist(); + unawaited(_persist()); _onChanged(); markDirty(); } void markDirty() { if (!_remoteEnabled || _disposed) return; + _localRevision++; + _publishPending = true; _publishDebounce?.cancel(); _publishDebounce = Timer(const Duration(seconds: 5), () { _publishDebounce = null; - unawaited(_publish()); + unawaited( + _publish().whenComplete( + () => _publishPending = _publishDebounce != null, + ), + ); }); } - Future _fetchAndMerge() async { - if (_relaySession == null) return; + /// A recovery read that loses to a local edit is deferred, never applied: + /// a peer entry stamped by a skewed clock could otherwise overwrite it. + Future _recoverHead() async { + if (_publishPending) return false; + final revision = _localRevision; + final events = await _fetchHead(); + if (events == null || _disposed) return false; + if (revision != _localRevision || _publishPending) return false; + final applied = await _applyEvents(events); + if (!_disposed) _onChanged(); + return applied; + } + + Future?> _fetchHead() async { try { - final events = await _relaySession.fetchHistory( - NostrFilter( - kinds: const [EventKind.readState], - authors: [pubkey], - tags: const { - '#d': ['channel-stars'], - }, - limit: 1, - ), - ); - _mergeEvents(events); - _persist(); - if (!_disposed) _onChanged(); - } catch (_) { - // Local state remains usable when relay is unavailable. + return await _relaySession!.fetchHistory(_filter()); + } catch (error) { + debugPrint('[ChannelStarsManager] fetch failed: $error'); + return null; } } - Future _startLiveSubscription() async { - if (_relaySession == null) return; + Future _startLiveSubscription() async { try { - _unsubscribe = await _relaySession.subscribe( - NostrFilter( - kinds: const [EventKind.readState], - authors: [pubkey], - tags: const { - '#d': ['channel-stars'], - }, - limit: 1, - ), + final unsubscribe = await _relaySession!.subscribe( + _filter(), _handleIncomingEvent, + // Only terminal rejections reach here; re-sending the same REQ would + // be rejected again, so the live lane stays settled. + onClosed: (message) { + _liveSettled = true; + debugPrint( + '[ChannelStarsManager] live subscription closed: $message', + ); + }, ); - } catch (_) { - // Non-fatal — local state and history still work. + if (_disposed) { + unsubscribe(); + return false; + } + _unsubscribe = unsubscribe; + return _liveSettled = true; + } catch (error) { + debugPrint('[ChannelStarsManager] live subscription failed: $error'); + return _liveSettled; } } - void _mergeEvents(List events) { + NostrFilter _filter() => NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: const { + '#d': ['channel-stars'], + }, + limit: 1, + ); + + /// Relay retains `ORDER BY created_at DESC, id ASC`: at equal second the + /// lower event ID wins. + bool _isAfterCursor(int createdAt, String id) => + createdAt > _lastRemoteCreatedAt || + (createdAt == _lastRemoteCreatedAt && + id.compareTo(_lastRemoteEventId ?? '') < 0); + + /// Returns false only when a newer head could not be persisted; the cursor + /// then stays put so the same head is retried. An undecodable head is + /// skipped without holding startup open. + Future _applyEvents(List events) async { + var applied = true; for (final event in events) { - if (event.pubkey != pubkey) continue; - _mergeEvent(event); - } - } - - void _mergeEvent(NostrEvent event) { - // Only process channel-stars d-tag events. - final dTag = event.getTagValue('d'); - if (dTag != 'channel-stars') return; - - try { - final plaintext = _crypto.decrypt(event.content); - final parsed = jsonDecode(plaintext); - if (parsed is! Map) return; - - final incoming = ChannelStarStore.fromJson(parsed); - - // Gate on createdAt: ignore events older than what we've already seen. - final isNewer = - event.createdAt > _lastRemoteCreatedAt || - (event.createdAt == _lastRemoteCreatedAt && - event.id.compareTo(_lastRemoteEventId ?? '') > 0); - - if (isNewer) { + if (event.pubkey != pubkey || event.getTagValue('d') != 'channel-stars') { + continue; + } + if (!_isAfterCursor(event.createdAt, event.id)) continue; + try { + final parsed = jsonDecode(_crypto.decrypt(event.content)); + if (parsed is! Map) throw const FormatException(); + // Per-channel merge: keep the entry with the highest updatedAt. + _store = mergeStores(_store, ChannelStarStore.fromJson(parsed)); + } catch (error) { + debugPrint('[ChannelStarsManager] undecodable head: $error'); + continue; + } + if (!await _persist()) { + applied = false; + continue; + } + if (_isAfterCursor(event.createdAt, event.id)) { _lastRemoteCreatedAt = event.createdAt; _lastRemoteEventId = event.id; - // Per-channel merge: keep the entry with the highest updatedAt for each channel. - _store = mergeStores(_store, incoming); - _persist(); } - } catch (_) { - // Decryption failure or parse error — keep existing state. } + return applied; } void _handleIncomingEvent(NostrEvent event) { if (_disposed) return; - _mergeEvent(event); - if (!_disposed) _onChanged(); + unawaited( + _applyEvents([event]).then((_) => _disposed ? null : _onChanged()), + ); } bool _isIdenticalToLastPublished() { @@ -232,7 +341,9 @@ class ChannelStarsManager { } // Read-before-write: merge remote state before publishing - await _fetchAndMerge(); + final events = await _fetchHead(); + if (events != null) await _applyEvents(events); + if (!_disposed) _onChanged(); // No-op suppression: skip if nothing changed if (_isIdenticalToLastPublished()) return; @@ -242,6 +353,7 @@ class ChannelStarsManager { final ciphertext = _crypto.encrypt(payload); final createdAt = max(currentUnixSeconds(), _lastRemoteCreatedAt + 1); + String? signedId; await _signedEventRelay.submit( kind: EventKind.readState, content: ciphertext, @@ -250,16 +362,29 @@ class ChannelStarsManager { ['t', 'channel-stars'], ], createdAt: createdAt, + onSigned: (event) => signedId = event.id, ); - _lastRemoteCreatedAt = max(_lastRemoteCreatedAt, createdAt); + // Keep the cursor a coherent (created_at, id) pair so an OK that beats + // its own echo cannot make later same-second heads compare against a + // stale ID. A newer head learned during the await stays. + if (signedId != null && _isAfterCursor(createdAt, signedId!)) { + _lastRemoteCreatedAt = createdAt; + _lastRemoteEventId = signedId; + } _lastPublishedStore = ChannelStarStore(channels: Map.of(_store.channels)); } catch (error) { debugPrint('[ChannelStarsManager] publish failed: $error'); } } - void _persist() { - _storage.write(pubkey, _store); + Future _persist() async { + try { + if (await _storage.write(pubkey, _store)) return true; + debugPrint('[ChannelStarsManager] persist returned false'); + } catch (error) { + debugPrint('[ChannelStarsManager] persist failed: $error'); + } + return false; } } diff --git a/mobile/lib/features/channels/channel_stars/channel_stars_provider.dart b/mobile/lib/features/channels/channel_stars/channel_stars_provider.dart index 4f6fb305b4..a341417837 100644 --- a/mobile/lib/features/channels/channel_stars/channel_stars_provider.dart +++ b/mobile/lib/features/channels/channel_stars/channel_stars_provider.dart @@ -1,3 +1,4 @@ +import 'package:flutter/widgets.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:nostr/nostr.dart' as nostr; @@ -69,6 +70,11 @@ class ChannelStarsNotifier extends Notifier { ); _manager = manager; + // Resume re-read catches an EVENT a healthy socket never delivered. + ref.listen(appLifecycleProvider, (_, next) { + if (next == AppLifecycleState.resumed) manager.refreshFromRelay(); + }); + ref.onDispose(() { manager.dispose(); if (_manager == manager) { diff --git a/mobile/lib/features/channels/channel_stars/channel_stars_storage.dart b/mobile/lib/features/channels/channel_stars/channel_stars_storage.dart index 07d19bbd01..cf6809cbd0 100644 --- a/mobile/lib/features/channels/channel_stars/channel_stars_storage.dart +++ b/mobile/lib/features/channels/channel_stars/channel_stars_storage.dart @@ -85,7 +85,6 @@ class ChannelStarsStorage { } } - void write(String pubkey, ChannelStarStore store) { - _prefs.setString(channelStarsKey(pubkey), jsonEncode(store.toJson())); - } + Future write(String pubkey, ChannelStarStore store) => + _prefs.setString(channelStarsKey(pubkey), jsonEncode(store.toJson())); } diff --git a/mobile/test/features/channels/channel_reader_recovery_test.dart b/mobile/test/features/channels/channel_reader_recovery_test.dart new file mode 100644 index 0000000000..406788e3b2 --- /dev/null +++ b/mobile/test/features/channels/channel_reader_recovery_test.dart @@ -0,0 +1,497 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:buzz/features/channels/channel_mutes/channel_mutes_manager.dart'; +import 'package:buzz/features/channels/channel_mutes/channel_mutes_provider.dart'; +import 'package:buzz/features/channels/channel_sections/channel_sections_provider.dart'; +import 'package:buzz/features/channels/channel_sort/channel_sort_provider.dart'; +import 'package:buzz/features/channels/channel_stars/channel_stars_provider.dart'; +import 'package:buzz/shared/community/community.dart'; +import 'package:buzz/shared/community/community_provider.dart'; +import 'package:buzz/shared/theme/theme_provider.dart'; +import 'package:flutter/widgets.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:buzz/features/channels/channel_sections/channel_sections_manager.dart'; +import 'package:buzz/features/channels/channel_sort/channel_sort_manager.dart'; +import 'package:buzz/features/channels/channel_sort/channel_sort_storage.dart'; +import 'package:buzz/features/channels/channel_stars/channel_stars_manager.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +import 'sidebar_sync_fixture.dart'; + +/// The per-entry lanes share one manager shape; each case runs on both. +class _Lane { + const _Lane(this.dTag, this.field, this.build); + + final String dTag; + final String field; + final _Subject Function(SharedPreferences prefs, SidebarRelay relay) build; +} + +class _Subject { + _Subject({ + required this.init, + required this.values, + required this.set, + required this.refresh, + required this.dispose, + }); + + final Future Function() init; + final Map Function() values; + final void Function(String channel, bool value) set; + final void Function() refresh; + final void Function() dispose; +} + +final _lanes = [ + _Lane('channel-stars', 'starred', (prefs, relay) { + final m = ChannelStarsManager( + pubkey: relay.pubkey, + prefs: prefs, + crypto: ChannelStarsCrypto(relay.keys.nsec, relay.pubkey), + relaySession: relay.session, + signedEventRelay: relay.signer, + remoteEnabled: true, + onChanged: () {}, + ); + return _Subject( + init: m.initialize, + values: () => { + for (final e in m.store.channels.entries) e.key: e.value.starred, + }, + set: (c, v) => v ? m.starChannel(c) : m.unstarChannel(c), + refresh: m.refreshFromRelay, + dispose: () => m.dispose(flushPending: false), + ); + }), + _Lane('channel-mutes', 'muted', (prefs, relay) { + final m = ChannelMutesManager( + pubkey: relay.pubkey, + prefs: prefs, + crypto: ChannelMutesCrypto(relay.keys.nsec, relay.pubkey), + relaySession: relay.session, + signedEventRelay: relay.signer, + remoteEnabled: true, + onChanged: () {}, + ); + return _Subject( + init: m.initialize, + values: () => { + for (final e in m.store.channels.entries) e.key: e.value.muted, + }, + set: (c, v) => v ? m.muteChannel(c) : m.unmuteChannel(c), + refresh: m.refreshFromRelay, + dispose: () => m.dispose(flushPending: false), + ); + }), +]; + +class _Lifecycle extends AppLifecycleNotifier { + @override + AppLifecycleState build() => AppLifecycleState.paused; + void resume() => state = AppLifecycleState.resumed; +} + +class _Config extends RelayConfigNotifier { + _Config(this.nsec); + final String nsec; + @override + RelayConfig build() => + RelayConfig(baseUrl: 'https://relay.example', nsec: nsec); +} + +/// Missed heads, one per lane, each marked with `zz` so the persisted +/// cache shows whether that lane's provider re-read on resume. +const _missedHeads = { + 'channel-stars': + '{"version":1,"channels":{"zz":{"starred":true,"updatedAt":1}}}', + 'channel-mutes': + '{"version":1,"channels":{"zz":{"muted":true,"updatedAt":1}}}', + 'channel-sections': + '{"version":1,"sections":[{"id":"zz","name":"zz","order":0}],"assignments":{}}', + 'channel-sort': '{"version":1,"groups":{"zz":"recent"}}', +}; + +void main() { + late SharedPreferences prefs; + setUp(() async => prefs = await freshPrefs()); + wholeBlobLanes(() => prefs); + + for (final MapEntry(key: dTag, value: head) in _missedHeads.entries) { + fakeAsyncTest('$dTag provider re-reads a missed head on resume', (clock) { + final relay = SidebarRelay(); + final c = ProviderContainer( + overrides: [ + savedPrefsProvider.overrideWithValue(prefs), + relayConfigProvider.overrideWith(() => _Config(relay.keys.nsec)), + relaySessionProvider.overrideWith(() => relay.session), + activeCommunityProvider.overrideWith( + (ref) async => Community( + id: 'c', + name: 'c', + relayUrl: 'wss://relay.example', + addedAt: DateTime(2026), + ), + ), + appLifecycleProvider.overrideWith(_Lifecycle.new), + ], + ); + for (final p in [ + channelStarsProvider, + channelMutesProvider, + channelSectionsProvider, + channelSortProvider, + ]) { + c.listen(p, (_, _) {}); + } + bool adopted() => + prefs.getKeys().any((k) => '${prefs.get(k)}'.contains('zz')); + clock.elapse(const Duration(seconds: 1)); + expect(adopted(), isFalse); + relay.stored.add(relay.event(dTag, jsonDecode(head), nowSeconds() + 60)); + (c.read(appLifecycleProvider.notifier) as _Lifecycle).resume(); + clock.elapse(const Duration(seconds: 1)); + expect(adopted(), isTrue); + expect(relay.reqsFor(dTag, 'l-'), hasLength(1)); + c.dispose(); + }); + } + + for (final lane in _lanes) { + Map blob(Map entries) => { + 'version': 1, + 'channels': { + for (final e in entries.entries) + e.key: {lane.field: e.value.$1, 'updatedAt': e.value.$2}, + }, + }; + + group(lane.dTag, () { + late SidebarRelay relay; + late _Subject subject; + late int t; + setUp(() { + relay = SidebarRelay(); + t = nowSeconds(); + }); + tearDown(() => subject.dispose()); + + _Subject start(FakeAsync clock, {SharedPreferences? store}) { + subject = lane.build(store ?? prefs, relay); + subject.init(); + clock.flushMicrotasks(); + return subject; + } + + for (final priorHead in [false, true]) { + fakeAsyncTest('OK before echo keeps a coherent cursor ' + '(${priorHead ? 'stale prior id' : 'no prior id'})', (clock) { + const low = + '0000000000000000000000000000000000000000000000000000000000000000'; + if (priorHead) { + relay.stored.add(relay.event(lane.dTag, blob({}), t + 30, id: low)); + } + start(clock).set('mine', true); + clock.elapse(const Duration(seconds: 5)); + final own = relay.published.single; + expect( + own.createdAt, + priorHead ? t + 31 : inInclusiveRange(t, nowSeconds()), + ); + + // The OK beat the (never delivered) echo. A same-second peer that + // the relay retains over our event must still be adopted. + // A same-second peer with a higher ID must still lose. + final peerId = '${low.substring(1)}1'; + expect(peerId.compareTo(own.id), lessThan(0)); + for (final (name, id) in [ + ('loser', ''.padLeft(64, 'f')), + ('peer', peerId), + ]) { + relay.emit( + relay.event( + lane.dTag, + blob({name: (true, own.createdAt)}), + own.createdAt, + id: id, + ), + ); + } + clock.elapse(const Duration(milliseconds: 20)); + expect(subject.values(), {'mine': true, 'peer': true}); + }); + } + + for (final afterFallback in [false, true]) { + fakeAsyncTest( + 'terminal CLOSED ${afterFallback ? 'after' : 'before'} the readiness ' + 'fallback is not re-sent; history still recovers', + (clock) { + relay + ..historyFailures = 1 + ..stored.add(relay.event(lane.dTag, blob({'a': (true, t)}), t)); + if (afterFallback) { + relay.withholdEose = true; + } else { + relay.rejectLive = 'restricted: not allowed'; + } + start(clock); + clock.elapse(const Duration(milliseconds: 600)); + if (afterFallback) { + relay.closeLive(lane.dTag, 'error: too many subscriptions'); + } + clock.elapse(const Duration(minutes: 5)); + expect(relay.reqsFor(lane.dTag, 'l-'), hasLength(1)); + expect(relay.reqsFor(lane.dTag, 'h-'), hasLength(2)); + expect(subject.values(), {'a': true}); + }, + ); + } + + for (final error in [null, StateError('disk full')]) { + fakeAsyncTest( + 'failed persist (${error == null ? 'false' : 'throws'}) retries the ' + 'same head until it is durable', + (clock) { + final flaky = FlakyPrefs(prefs) + ..failures = 1 + ..error = error; + relay.stored.add(relay.event(lane.dTag, blob({'a': (true, t)}), t)); + start(clock, store: flaky); + expect(prefs.getKeys(), isEmpty); + clock.elapse(const Duration(seconds: 2)); + expect(relay.reqsFor(lane.dTag, 'h-'), hasLength(2)); + expect(prefs.getKeys(), hasLength(1)); + expect(subject.values(), {'a': true}); + }, + ); + } + + fakeAsyncTest('an undecodable head does not hold startup open', (clock) { + relay.stored.add(relay.event(lane.dTag, 'garbage', t)); + start(clock); + clock.elapse(const Duration(seconds: 70)); + expect(relay.reqsFor(lane.dTag, 'h-'), hasLength(1)); + expect(relay.reqsFor(lane.dTag, 'l-'), hasLength(1)); + }); + + fakeAsyncTest('resume adopts a head the healthy socket missed', (clock) { + relay.stored.add(relay.event(lane.dTag, blob({'a': (true, t)}), t)); + start(clock); + expect(subject.values(), {'a': true}); + relay.stored.add( + relay.event( + lane.dTag, + blob({'a': (true, t), 'b': (true, t + 1)}), + t + 1, + ), + ); + subject.refresh(); + clock.flushMicrotasks(); + expect(subject.values(), {'a': true, 'b': true}); + expect(prefs.getString(prefs.getKeys().single), contains('"b"')); + expect(relay.reqsFor(lane.dTag, 'l-'), hasLength(1)); + + // A failed resume read is one shot, not a retry loop. + relay.historyFailures = 1; + subject.refresh(); + clock.elapse(const Duration(seconds: 70)); + expect(relay.reqsFor(lane.dTag, 'h-'), hasLength(3)); + }); + + fakeAsyncTest( + 'a read held across a local edit defers instead of overwriting it', + (clock) { + start(clock); + // The peer's clock runs one second ahead of ours on the same channel. + relay.stored.add( + relay.event(lane.dTag, blob({'c': (false, t + 1)}), t + 1), + ); + final held = relay.holdHistory = Completer(); + subject.refresh(); + clock.flushMicrotasks(); + subject.set('c', true); + held.complete(); + relay.holdHistory = null; + clock.flushMicrotasks(); + expect(subject.values(), {'c': true}); + + // Resume while the edit is still pending leaves it alone too. + subject.refresh(); + clock.flushMicrotasks(); + expect(subject.values(), {'c': true}); + expect(relay.reqsFor(lane.dTag, 'h-'), hasLength(2)); + }, + ); + }); + } +} + +/// Whole-blob lanes: resume re-read, and sections' publication cursor. +void wholeBlobLanes(SharedPreferences Function() prefs) { + group('sections', () { + late SidebarRelay relay; + late ChannelSectionsManager m; + setUp(() => relay = SidebarRelay()); + tearDown(() => m.dispose(flushPending: false)); + + Map blob(String name) => { + 'version': 1, + 'sections': [ + {'id': name, 'name': name, 'order': 0}, + ], + 'assignments': {}, + }; + + ChannelSectionsManager start(FakeAsync clock) { + m = ChannelSectionsManager( + pubkey: relay.pubkey, + prefs: prefs(), + crypto: ChannelSectionsCrypto(relay.keys.nsec, relay.pubkey), + relaySession: relay.session, + signedEventRelay: relay.signer, + remoteEnabled: true, + onChanged: () {}, + )..initialize(); + clock.flushMicrotasks(); + return m; + } + + List names() => [for (final s in m.store.sections) s.name]; + + fakeAsyncTest('resume adopts a head the healthy socket missed', (clock) { + final t = nowSeconds(); + start(clock); + relay.stored.add(relay.event('channel-sections', blob('later'), t)); + m.refreshFromRelay(); + clock.flushMicrotasks(); + expect(names(), ['later']); + expect(relay.reqsFor('channel-sections', 'l-'), hasLength(1)); + }); + + fakeAsyncTest('resume leaves a pending edit alone', (clock) { + final t = nowSeconds(); + start(clock).createSection('mine'); + relay.stored.add(relay.event('channel-sections', blob('peer'), t + 9)); + m.refreshFromRelay(); + clock.flushMicrotasks(); + expect(names(), ['mine']); + }); + + fakeAsyncTest('OK before echo keeps a coherent cursor', (clock) { + start(clock).createSection('mine'); + clock.elapse(const Duration(seconds: 5)); + final own = relay.published.single; + // A same-second higher ID loses to our own event; a lower one wins. + for (final (name, pad, want) in [ + ('loser', 'f', 'mine'), + ('peer', '0', 'peer'), + ]) { + relay.emit( + relay.event( + 'channel-sections', + blob(name), + own.createdAt, + id: ''.padLeft(64, pad), + ), + ); + clock.elapse(const Duration(milliseconds: 20)); + expect(names(), [want]); + } + }); + + fakeAsyncTest('a same-second loser merged during the OK wait is undone', ( + clock, + ) { + relay.holdOk = Completer(); + start(clock).createSection('mine'); + clock.elapse(const Duration(seconds: 5)); + final own = relay.published.single; + final loser = relay.event( + 'channel-sections', + blob('loser'), + own.createdAt, + id: ''.padLeft(64, 'f'), + ); + relay + ..stored.add(loser) + ..emit(loser); + clock.elapse(const Duration(milliseconds: 20)); + expect(names(), ['loser']); + relay.holdOk!.complete(); + clock.flushMicrotasks(); + relay.emit(own); + m.refreshFromRelay(); + clock.elapse(const Duration(milliseconds: 20)); + expect(names(), ['mine']); + expect(prefs().getString(prefs().getKeys().single), contains('"mine"')); + }); + }); + + group('sort', () { + late SidebarRelay relay; + late ChannelSortManager m; + setUp(() => relay = SidebarRelay()); + tearDown(() => m.dispose()); + + ChannelSortManager start(FakeAsync clock, {bool remote = true}) { + m = ChannelSortManager( + pubkey: relay.pubkey, + relayUrl: 'wss://relay.example', + prefs: prefs(), + crypto: ChannelSortCrypto(relay.keys.nsec, relay.pubkey), + relaySession: relay.session, + signedEventRelay: relay.signer, + remoteEnabled: remote, + onChanged: () {}, + )..initialize(); + clock.flushMicrotasks(); + return m; + } + + NostrEvent recent(int t) => relay.event('channel-sort', { + 'version': 1, + 'groups': {'dms': 'recent'}, + }, t); + + fakeAsyncTest('resume adopts a head the healthy socket missed', (clock) { + start(clock); + relay.stored.add(recent(nowSeconds())); + m.refreshFromRelay(); + clock.flushMicrotasks(); + expect(m.sortModeFor('dms'), ChannelSortMode.recent); + expect(relay.reqsFor('channel-sort', 'l-'), hasLength(1)); + }); + + fakeAsyncTest('a retired resume read cannot overwrite its successor', ( + clock, + ) { + start(clock); + relay.stored.add(recent(nowSeconds() + 9)); + final held = relay.holdHistory = Completer(); + m.refreshFromRelay(); + clock.flushMicrotasks(); + m.dispose(); + start(clock, remote: false).setSortModeFor('dms', ChannelSortMode.alpha); + clock.flushMicrotasks(); + final before = {for (final k in prefs().getKeys()) k: prefs().get(k)}; + held.complete(); + clock.flushMicrotasks(); + expect({for (final k in prefs().getKeys()) k: prefs().get(k)}, before); + }); + + fakeAsyncTest('resume leaves a pending edit alone', (clock) { + start(clock).setSortModeFor('dms', ChannelSortMode.alpha); + relay.stored.add(recent(nowSeconds() + 9)); + final reads = relay.reqsFor('channel-sort', 'h-').length; + m.refreshFromRelay(); + clock.flushMicrotasks(); + expect(m.sortModeFor('dms'), ChannelSortMode.alpha); + expect(relay.reqsFor('channel-sort', 'h-'), hasLength(reads)); + }); + }); +} diff --git a/mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart b/mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart index 16d1936d39..a8520d6107 100644 --- a/mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart +++ b/mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart @@ -240,6 +240,56 @@ void main() { }, ); + test('tie_break_lower_id_wins_at_equal_second', () async { + // Relay retains `ORDER BY created_at DESC, id ASC` — the lower event ID + // wins at equal second. Pre-fix the manager kept the higher ID, which + // permanently disagreed with desktop post-#7805. + await setUpEnv(); + final relay = _RateLimitedRelaySession(failuresBeforeSuccess: 0); + final manager = buildManager(relaySession: relay); + await manager.initialize(); + + // Deliver two equal-second events in higher-then-lower ID order via the + // live subscription and verify the lower-ID blob is the final state. + final higherIdEvent = sectionsEvent( + sections: [ + {'id': 's-high', 'name': 'From Higher ID', 'order': 0}, + ], + createdAt: 100, + id: 'zzzz', + ); + final lowerIdEvent = sectionsEvent( + sections: [ + {'id': 's-low', 'name': 'From Lower ID', 'order': 0}, + ], + createdAt: 100, + id: 'aaaa', + ); + + relay.emit(higherIdEvent); + await Future.delayed(const Duration(milliseconds: 20)); + expect(manager.store.sections.single.name, 'From Higher ID'); + + relay.emit(lowerIdEvent); + await Future.delayed(const Duration(milliseconds: 20)); + expect( + manager.store.sections.single.name, + 'From Lower ID', + reason: 'lower-ID event must win at equal second (relay id ASC order)', + ); + + // Re-delivery of the higher-ID event must NOT roll back. + relay.emit(higherIdEvent); + await Future.delayed(const Duration(milliseconds: 20)); + expect( + manager.store.sections.single.name, + 'From Lower ID', + reason: 'higher-ID event re-delivery must be rejected by tie-break guard', + ); + + manager.dispose(flushPending: false); + }); + test('backoff resets after full recovery so later failures start from the ' 'base delay', () async { await setUpEnv(); diff --git a/mobile/test/features/channels/sidebar_sync_fixture.dart b/mobile/test/features/channels/sidebar_sync_fixture.dart new file mode 100644 index 0000000000..99ca53cc24 --- /dev/null +++ b/mobile/test/features/channels/sidebar_sync_fixture.dart @@ -0,0 +1,171 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:buzz/shared/crypto/nip44.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; +import 'package:shared_preferences/shared_preferences.dart'; + +/// A scripted relay behind the real [RelaySessionNotifier], so REQ/EOSE, +/// CLOSED classification, event batching and OK all run production code. +/// Retains replaceable heads the way the relay does: `created_at DESC, id ASC`. +class SidebarRelay { + SidebarRelay() { + session.debugAttachSocketForTest(_ScriptedSocket(_onSend)); + } + + final keys = nostr.Keys.generate(); + final session = ConnectedRelaySession(); + late final signer = SignedEventRelay(session: session, nsec: keys.nsec); + final stored = []; + final published = []; + final reqs = >[]; + final _liveSubIds = {}; + int historyFailures = 0; + Completer? holdHistory; + Completer? holdOk; + String? rejectLive; + bool withholdEose = false; + + String get pubkey => keys.public; + + NostrEvent event(String dTag, Object payload, int createdAt, {String? id}) => + NostrEvent( + id: id ?? 'e$createdAt'.padRight(64, 'f'), + pubkey: pubkey, + createdAt: createdAt, + kind: EventKind.readState, + tags: [ + ['d', dTag], + ], + content: nip44Encrypt(_key, jsonEncode(payload)), + sig: 'sig', + ); + + Object? decrypt(NostrEvent event) => + jsonDecode(nip44Decrypt(_key, event.content)); + + late final _key = getConversationKey( + nostr.Nip19.decode(payload: keys.nsec).data, + keys.public, + ); + + List> reqsFor(String dTag, String prefix) => [ + for (final req in reqs) + if ((req[1] as String).startsWith(prefix) && _dTagOf(req) == dTag) req, + ]; + + void emit(NostrEvent event) => session.debugHandleMessage([ + 'EVENT', + _liveSubIds[event.getTagValue('d')], + event.toJson(), + ]); + + void closeLive(String dTag, String message) => + session.debugHandleMessage(['CLOSED', _liveSubIds[dTag], message]); + + void _onSend(List message) { + if (message.first == 'EVENT') { + final event = NostrEvent.fromJson(message[1] as Map); + published.add(event); + stored.add(event); + final ok = ['OK', event.id, true, '']; + final hold = holdOk; + hold == null + ? scheduleMicrotask(() => _reply(ok)) + : hold.future.then((_) => _reply(ok)); + } + if (message.first != 'REQ') return; + reqs.add(message); + final subId = message[1] as String; + if (subId.startsWith('h-')) { + unawaited(_answerHistory(subId, _dTagOf(message))); + return; + } + _liveSubIds[_dTagOf(message)] = subId; + final rejection = rejectLive; + if (rejection == null && withholdEose) return; + scheduleMicrotask( + () => _reply( + rejection == null ? ['EOSE', subId] : ['CLOSED', subId, rejection], + ), + ); + } + + Future _answerHistory(String subId, String dTag) async { + await holdHistory?.future; + if (historyFailures > 0) { + historyFailures--; + return _reply(['CLOSED', subId, 'error: overloaded']); + } + final heads = stored.where((e) => e.getTagValue('d') == dTag).toList() + ..sort( + (a, b) => a.createdAt != b.createdAt + ? b.createdAt.compareTo(a.createdAt) + : a.id.compareTo(b.id), + ); + if (heads.isNotEmpty) _reply(['EVENT', subId, heads.first.toJson()]); + _reply(['EOSE', subId]); + } + + void _reply(List message) => session.debugHandleMessage(message); + + static String _dTagOf(List req) => + ((req[2] as Map)['#d'] as List).single as String; +} + +class ConnectedRelaySession extends RelaySessionNotifier { + @override + SessionState build() => const SessionState(status: SessionStatus.connected); +} + +class _ScriptedSocket extends RelaySocket { + _ScriptedSocket(this._send) + : super( + wsUrl: 'wss://relay.example', + nsec: null, + onMessage: (_) {}, + onConnected: () {}, + onDisconnected: (_) {}, + ); + + final void Function(List) _send; + + @override + void send(List payload) => _send(payload); + + @override + void dispose() {} +} + +/// Preferences whose next [failures] writes return false or throw [error]. +class FlakyPrefs extends Fake implements SharedPreferences { + FlakyPrefs(this._inner); + + final SharedPreferences _inner; + int failures = 0; + Object? error; + + @override + String? getString(String key) => _inner.getString(key); + + @override + Future setString(String key, String value) async { + if (failures == 0) return _inner.setString(key, value); + failures--; + if (error == null) return false; + throw error!; + } +} + +Future freshPrefs() async { + SharedPreferences.setMockInitialValues({}); + return SharedPreferences.getInstance(); +} + +int nowSeconds() => DateTime.now().millisecondsSinceEpoch ~/ 1000; + +void fakeAsyncTest(String name, void Function(FakeAsync clock) body) => + test(name, () => fakeAsync(body));