From 7273323f7465d78049b9bfaa2cf38d85f98e1595 Mon Sep 17 00:00:00 2001 From: grunch Date: Mon, 31 Aug 2026 19:37:01 -0300 Subject: [PATCH 1/3] perf: bound the orders subscription replay with a persisted since cursor The orders filter (kind 14) carried no since or limit, so every (re)subscription - cold start, resume, relay recovery, node switch - replayed the node's full message history from every relay, each replayed event costing a dedup read on the UI isolate. kind 14 events carry real timestamps, so the chat cursor pattern applies: - New orders_since_ cursor namespace (keyed by node pubkey) on the existing ChatCursorStore, warmed before the filter is built. - buildOrdersFilter's nip44 branch takes since (cursor minus overlap, falling back to the default lookback on fresh installs - older history is served by the restore flow) and a limit. The legacy 1059 branch stays unbounded: gift wrap timestamps are randomized. - MostroService advances the cursor in _markEventProcessed, so events deliberately left unprocessed (no matching session yet) stay inside the cursor's overlap window and can be retried by a later resubscription. - Cursor access is best-effort: storage failures fall back to the default lookback instead of blocking the REQ. --- .../subscriptions/subscription_manager.dart | 37 +++++++++++++++++-- lib/services/chat_cursor_store.dart | 12 ++++++ lib/services/mostro_service.dart | 9 +++++ .../subscriptions/orders_filter_test.dart | 27 ++++++++++++++ 4 files changed, 82 insertions(+), 3 deletions(-) diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index cb2fcfa8..5c1f64eb 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/lib/features/subscriptions/subscription_manager.dart @@ -295,6 +295,16 @@ class SubscriptionManager { .whereType(); await ref.read(chatCursorStoreProvider).warmUp(orderIds); } + if (type == SubscriptionType.orders) { + // Best-effort: a prefs/storage failure must not prevent the REQ. + try { + await ref + .read(ordersCursorStoreProvider) + .warmUp([ref.read(settingsProvider).mostroPublicKey]); + } catch (e) { + logger.w('Orders cursor warm-up unavailable: $e'); + } + } final filter = _createFilterForType(type, sessions); if (filter == null) { @@ -365,10 +375,23 @@ class SubscriptionManager { // and re-subscribe when the node info arrives after this subscription. final transport = _resolveOrdersTransport(); _appliedOrdersTransport = transport; + final mostroPubkey = ref.read(settingsProvider).mostroPublicKey; + // Persisted cursor bounds the replay; fresh installs fall back to the + // default lookback (older history is served by the restore flow). + DateTime? cursorSince; + try { + cursorSince = + ref.read(ordersCursorStoreProvider).cachedSinceFor(mostroPubkey); + } catch (e) { + logger.w('Orders cursor unavailable, using default lookback: $e'); + } + final ordersSince = cursorSince ?? + DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback); return buildOrdersFilter( transport, tradeKeys, - ref.read(settingsProvider).mostroPublicKey, + mostroPubkey, + since: ordersSince, ); case SubscriptionType.chat: // Kind 14 chat envelope: filter by the K_sign authors derived from @@ -670,19 +693,27 @@ class SubscriptionManager { NostrFilter buildOrdersFilter( Transport transport, List tradeKeys, - String mostroPubkey, -) { + String mostroPubkey, { + DateTime? since, +}) { switch (transport) { case Transport.giftWrap: + // Legacy transport: gift wrap timestamps are randomized ±48 h, so a + // cursor since would silently drop messages. Left unbounded until the + // 1059 branch is removed. return NostrFilter( kinds: [1059], p: tradeKeys, ); case Transport.nip44: + // kind 14 carries real timestamps: bound the replay with the persisted + // cursor (minus its overlap) and a limit, mirroring the chat filters. return NostrFilter( kinds: [14], authors: [mostroPubkey], p: tradeKeys, + since: since, + limit: NostrEventExtensions.chatDefaultLimit, ); } } diff --git a/lib/services/chat_cursor_store.dart b/lib/services/chat_cursor_store.dart index 32f786f4..12bd04c0 100644 --- a/lib/services/chat_cursor_store.dart +++ b/lib/services/chat_cursor_store.dart @@ -22,6 +22,10 @@ class ChatCursorStore { /// generalization; keeping it preserves cursors stored by older builds. static const disputeKeyPrefix = 'dispute_chat_since_'; + /// Orders (node message) namespace, keyed by the node pubkey: one live + /// orders subscription exists per connected node. + static const ordersKeyPrefix = 'orders_since_'; + /// Peer (buyer-seller) chat namespace, keyed by orderId. Shared with the /// background isolate, which builds its own store without Riverpod. static const peerKeyPrefix = 'chat_since_'; @@ -120,3 +124,11 @@ final chatCursorStoreProvider = Provider( keyPrefix: ChatCursorStore.peerKeyPrefix, ), ); + +/// Orders (node kind-14 message) cursors, keyed by the node pubkey. +final ordersCursorStoreProvider = Provider( + (ref) => ChatCursorStore( + ref.watch(sharedPreferencesProvider), + keyPrefix: ChatCursorStore.ordersKeyPrefix, + ), +); diff --git a/lib/services/mostro_service.dart b/lib/services/mostro_service.dart index 3df7c3bc..3da0be6f 100644 --- a/lib/services/mostro_service.dart +++ b/lib/services/mostro_service.dart @@ -4,6 +4,7 @@ import 'package:collection/collection.dart'; import 'package:dart_nostr/dart_nostr.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:mostro_mobile/services/chat_cursor_store.dart'; import 'package:mostro_mobile/services/logger_service.dart'; import 'package:mostro_mobile/data/enums.dart'; import 'package:mostro_mobile/data/models.dart'; @@ -120,6 +121,14 @@ class MostroService { /// by every later replay, and the order sat at a stale status across /// restarts. Future _markEventProcessed(NostrEvent event) { + // Advance the orders since cursor: a processed event never needs + // replaying. Events left unmarked (e.g. no session yet) stay behind the + // cursor's overlap window so a resubscription can retry them. + unawaited( + ref + .read(ordersCursorStoreProvider) + .advance(_settings.mostroPublicKey, event.createdAt!), + ); return ref.read(eventStorageProvider).putItem(event.id!, { 'id': event.id, 'created_at': event.createdAt!.millisecondsSinceEpoch ~/ 1000, diff --git a/test/features/subscriptions/orders_filter_test.dart b/test/features/subscriptions/orders_filter_test.dart index 3d715eac..0dc66ca0 100644 --- a/test/features/subscriptions/orders_filter_test.dart +++ b/test/features/subscriptions/orders_filter_test.dart @@ -31,5 +31,32 @@ void main() { expect(filter.authors, [mostroPubkey]); expect(filter.p, tradeKeys); }); + + // Without since/limit every (re)subscription replayed the node's full + // message history from every relay; kind 14 carries real timestamps, so + // a persisted cursor can bound the replay tightly. + test('v2 (nip44) carries the cursor since and a limit', () { + final since = DateTime.fromMillisecondsSinceEpoch(1700000000 * 1000); + final filter = buildOrdersFilter( + Transport.nip44, + tradeKeys, + mostroPubkey, + since: since, + ); + + expect(filter.since, since); + expect(filter.limit, isNotNull); + }); + + test('v1 (giftWrap) ignores since: its timestamps are randomized', () { + final filter = buildOrdersFilter( + Transport.giftWrap, + tradeKeys, + mostroPubkey, + since: DateTime.now(), + ); + + expect(filter.since, isNull); + }); }); } From d4abba696013894269c984df28fa4a9c3c5476f3 Mon Sep 17 00:00:00 2001 From: grunch Date: Mon, 31 Aug 2026 21:05:38 -0300 Subject: [PATCH 2/3] fix: make the orders cursor a safe contiguous watermark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the since-cursor bounding: - Bootstrap window: with no stored cursor the filter fell back to a flat 7-day lookback, so the first launch after upgrading could silently omit responses to orders older than that (non-terminal orders live well past 30 days, and normal startup never runs the restore flow). The fallback now also reaches back to the oldest live session — no message of an active order predates its session — while a fresh install keeps the default lookback. - Marker before cursor: the cursor advanced before putItem resolved, so a failed marker write could leave an event unmarked *and* outside every later replay. The durable marker is written first. - Transport scoping: gift wrap (1059) timestamps are randomized, and this cursor feeds the kind-14 filter only. Only kind 14 advances it now. - Contiguity: the node-wide cursor could step over one trade's still unprocessed event when another trade's newer one was processed. Events left unmarked now hold the cursor back until they are retried, bounded by a 1-hour hold so an event that can never be processed cannot freeze the replay window. Also fixes dispute_chat_duplicate_envelope_test, which failed on main too: chatUnwrap now verifies and decrypts on a worker isolate, whose spawn takes real wall-clock time, so draining the event queue could return before the valid envelope was accepted. It polls for the result instead. --- .../subscriptions/subscription_manager.dart | 23 ++- lib/services/mostro_service.dart | 64 +++++- .../dispute_chat_duplicate_envelope_test.dart | 20 +- .../orders_since_bootstrap_test.dart | 144 ++++++++++++++ .../mostro_service_orders_cursor_test.dart | 183 ++++++++++++++++++ 5 files changed, 422 insertions(+), 12 deletions(-) create mode 100644 test/features/subscriptions/orders_since_bootstrap_test.dart create mode 100644 test/services/mostro_service_orders_cursor_test.dart diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index 5c1f64eb..6140a242 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/lib/features/subscriptions/subscription_manager.dart @@ -362,6 +362,20 @@ class SubscriptionManager { } } + /// Earliest start time across [sessions], capped at the default lookback so + /// a long-lived session cannot widen the window beyond it *and* a short one + /// cannot narrow it below it. Used as the bootstrap `since` when no cursor + /// is stored yet. + DateTime _sessionsFloor(List sessions) { + final defaultFloor = + DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback); + if (sessions.isEmpty) return defaultFloor; + final oldest = sessions + .map((s) => s.startTime) + .reduce((a, b) => a.isBefore(b) ? a : b); + return oldest.isBefore(defaultFloor) ? oldest : defaultFloor; + } + NostrFilter? _createFilterForType( SubscriptionType type, List sessions) { switch (type) { @@ -385,8 +399,15 @@ class SubscriptionManager { } catch (e) { logger.w('Orders cursor unavailable, using default lookback: $e'); } + // No cursor yet means a fresh install *or* the first launch after + // upgrading to the cursor build. An upgrading install can hold orders + // far older than the default lookback (non-terminal orders are kept + // well past 30 days) and normal startup does not run the restore + // flow, so the window must also reach back to the oldest live + // session: no message of an active order predates its session. final ordersSince = cursorSince ?? - DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback); + _sessionsFloor(sessions) + .subtract(ChatCursorStore.cursorOverlap); return buildOrdersFilter( transport, tradeKeys, diff --git a/lib/services/mostro_service.dart b/lib/services/mostro_service.dart index 3da0be6f..27fd860b 100644 --- a/lib/services/mostro_service.dart +++ b/lib/services/mostro_service.dart @@ -120,19 +120,58 @@ class MostroService { /// message write) — the daemon's message then never notified, was skipped /// by every later replay, and the order sat at a stale status across /// restarts. - Future _markEventProcessed(NostrEvent event) { - // Advance the orders since cursor: a processed event never needs - // replaying. Events left unmarked (e.g. no session yet) stay behind the - // cursor's overlap window so a resubscription can retry them. + Future _markEventProcessed(NostrEvent event) async { + final eventId = event.id!; + await ref.read(eventStorageProvider).putItem(eventId, { + 'id': eventId, + 'created_at': event.createdAt!.millisecondsSinceEpoch ~/ 1000, + }); + // Only once the durable marker is written may the cursor move: a failed + // putItem leaves the event unmarked, and a cursor advanced past it would + // drop it from every later replay. + _advanceOrdersCursor(event); + } + + /// Events seen but deliberately left unmarked (no matching session yet, or + /// a failure mid-processing), keyed by id. They still need replaying, so + /// the shared node cursor must not move past the oldest of them. + final Map _retryableEvents = {}; + + /// How long an unmarked event holds the cursor back. Bounded so an event + /// that can never be processed (a trade key whose session is gone) cannot + /// freeze the cursor — and with it the replay window — forever. + static const _retryHoldWindow = Duration(hours: 1); + + /// Records an event that was not marked processed, so [_advanceOrdersCursor] + /// keeps the replay window covering it. + void _holdEventForRetry(NostrEvent event) { + if (event.kind != 14 || event.id == null || event.createdAt == null) return; + _retryableEvents[event.id!] = event.createdAt!; + } + + /// Advances the orders `since` cursor for a processed event. + /// + /// Only kind 14 counts: the cursor feeds the NIP-44 filter, and gift wrap + /// (1059) timestamps are randomized, so letting them move it would push + /// `since` past kind-14 messages once the node switches transport. + /// + /// The cursor is a contiguous watermark: it never moves past an event still + /// awaiting a retry, otherwise one trade's newer response would evict + /// another trade's older, still-unprocessed one from the replay window. + void _advanceOrdersCursor(NostrEvent event) { + _retryableEvents.remove(event.id); + if (event.kind != 14) return; + final accepted = event.createdAt!; + _retryableEvents.removeWhere( + (_, at) => at.isBefore(DateTime.now().subtract(_retryHoldWindow)), + ); + final blocked = _retryableEvents.values.any((at) => !at.isAfter(accepted)); + if (blocked) return; unawaited( ref .read(ordersCursorStoreProvider) - .advance(_settings.mostroPublicKey, event.createdAt!), + .advance(_settings.mostroPublicKey, accepted), ); - return ref.read(eventStorageProvider).putItem(event.id!, { - 'id': event.id, - 'created_at': event.createdAt!.millisecondsSinceEpoch ~/ 1000, - }); } Future _onData(NostrEvent event) async { @@ -158,6 +197,7 @@ class MostroService { // Deliberately NOT marked processed: the session may simply not exist // yet (startup ordering, a child order being linked), and a later // replay must be able to retry this event. + _holdEventForRetry(event); logger.w('No matching session found for recipient: ${event.recipient}'); return; } @@ -181,7 +221,10 @@ class MostroService { decryptedId = decryptedEvent.id; } - if (content == null) return; + if (content == null) { + _holdEventForRetry(event); + return; + } final result = jsonDecode(content); @@ -233,6 +276,7 @@ class MostroService { // transient failure. A permanently undecryptable event costs one // decrypt attempt per replay, which the dedup above bounds to one // relay copy at a time. + _holdEventForRetry(event); logger.e('Error processing event', error: e); } } diff --git a/test/features/disputes/dispute_chat_duplicate_envelope_test.dart b/test/features/disputes/dispute_chat_duplicate_envelope_test.dart index 230b9708..34f08be6 100644 --- a/test/features/disputes/dispute_chat_duplicate_envelope_test.dart +++ b/test/features/disputes/dispute_chat_duplicate_envelope_test.dart @@ -174,7 +174,13 @@ void main() { nostrService.controller.add(forged); nostrService.controller.add(real); - await pumpEventQueue(times: 200); + // chatUnwrap verifies and decrypts on a worker isolate, whose spawn takes + // real wall-clock time: draining the event queue alone can return before + // the valid envelope has been accepted. Poll until it lands instead. + await _waitFor( + () => container.read(disputeChatNotifierProvider(disputeId)).messages + .isNotEmpty, + ); final messages = container.read(disputeChatNotifierProvider(disputeId)).messages; @@ -185,3 +191,15 @@ void main() { expect(notifier.mounted, isTrue); }); } + +/// Polls [condition] until it holds or [timeout] elapses, yielding to the +/// event loop between attempts so isolate results can be delivered. +Future _waitFor( + bool Function() condition, { + Duration timeout = const Duration(seconds: 10), +}) async { + final deadline = DateTime.now().add(timeout); + while (!condition() && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 10)); + } +} diff --git a/test/features/subscriptions/orders_since_bootstrap_test.dart b/test/features/subscriptions/orders_since_bootstrap_test.dart new file mode 100644 index 00000000..4f63dc9a --- /dev/null +++ b/test/features/subscriptions/orders_since_bootstrap_test.dart @@ -0,0 +1,144 @@ +import 'dart:async'; + +import 'package:dart_nostr/dart_nostr.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:mockito/mockito.dart'; +import 'package:mostro_mobile/data/models/nostr_event.dart'; +import 'package:mostro_mobile/data/models/session.dart'; +import 'package:mostro_mobile/features/settings/settings.dart'; +import 'package:mostro_mobile/features/settings/settings_notifier.dart'; +import 'package:mostro_mobile/features/settings/settings_provider.dart'; +import 'package:mostro_mobile/features/subscriptions/subscription_manager.dart'; +import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart'; +import 'package:mostro_mobile/shared/notifiers/session_notifier.dart'; +import 'package:mostro_mobile/shared/providers/nostr_service_provider.dart'; +import 'package:mostro_mobile/shared/providers/order_repository_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart'; +import 'package:shared_preferences_platform_interface/in_memory_shared_preferences_async.dart'; +import 'package:shared_preferences_platform_interface/shared_preferences_async_platform_interface.dart'; + +import '../../mocks.mocks.dart'; + +/// First launch after upgrading to the cursor build: sessions exist but no +/// `orders_since_` preference does. A flat default lookback would silently +/// drop responses to orders older than it — non-terminal orders are kept far +/// longer, and normal startup does not run the restore flow — so the +/// bootstrap window must also reach back to the oldest live session. +const _mostroPubkey = 'mostro-pubkey'; +const _tradeKey = + 'abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890'; + +void main() { + late MockNostrService nostrService; + late MockOpenOrdersRepository orderRepository; + late ProviderContainer container; + late _FakeSessionNotifier sessions; + late SubscriptionManager manager; + late List issuedRequests; + late StreamController relayGenerations; + + Session sessionStartedAt(DateTime startTime) => Session( + masterKey: NostrKeyPairs( + private: + '1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef'), + tradeKey: NostrKeyPairs(private: _tradeKey), + keyIndex: 0, + fullPrivacy: false, + startTime: startTime, + )..orderId = 'order-a'; + + setUp(() { + SharedPreferencesAsyncPlatform.instance = + InMemorySharedPreferencesAsync.empty(); + nostrService = MockNostrService(); + orderRepository = MockOpenOrdersRepository(); + issuedRequests = []; + var nextSubscriptionId = 0; + when(nostrService.subscribeToEvents(any)).thenAnswer((invocation) { + final request = invocation.positionalArguments.first as NostrRequest; + request.subscriptionId ??= 'sub-${nextSubscriptionId++}'; + issuedRequests.add(request); + return const Stream.empty(); + }); + when(nostrService.unsubscribe(any)).thenAnswer((_) async {}); + relayGenerations = StreamController.broadcast(); + when(nostrService.relayGenerationStream) + .thenAnswer((_) => relayGenerations.stream); + when(nostrService.relayGeneration).thenAnswer((_) => 0); + when(orderRepository.mostroInstanceStream) + .thenAnswer((_) => const Stream.empty()); + when(orderRepository.mostroInstance).thenReturn(null); + + container = ProviderContainer(overrides: [ + nostrServiceProvider.overrideWithValue(nostrService), + orderRepositoryProvider.overrideWithValue(orderRepository), + settingsProvider.overrideWith((ref) => _FixedSettingsNotifier()), + sessionNotifierProvider.overrideWith((ref) { + sessions = _FakeSessionNotifier(ref); + return sessions; + }), + ]); + }); + + tearDown(() { + manager.unsubscribeAll(); + container.dispose(); + relayGenerations.close(); + }); + + Future ordersSinceFor(DateTime sessionStart) async { + container.read(sessionNotifierProvider); + manager = container.read(subscriptionManagerProvider); + sessions.emit([sessionStartedAt(sessionStart)]); + await Future.delayed(const Duration(milliseconds: 50)); + + final ordersFilter = issuedRequests + .expand((r) => r.filters) + .firstWhere((f) => + (f.kinds ?? const []).contains(14) && + (f.authors ?? const []).contains(_mostroPubkey)); + expect(ordersFilter.since, isNotNull); + return ordersFilter.since!; + } + + test('with no cursor the window reaches back to the oldest session', + () async { + final start = DateTime.now().subtract(const Duration(days: 45)); + + final since = await ordersSinceFor(start); + + expect(since.isAfter(start), isFalse, + reason: 'a response to a 45-day-old order must still be replayed'); + }); + + test('a young session does not narrow the window below the lookback', + () async { + final since = await ordersSinceFor(DateTime.now()); + + final lookbackFloor = DateTime.now() + .subtract(NostrEventExtensions.chatDefaultLookback) + .subtract(const Duration(minutes: 1)); + expect(since.isAfter(lookbackFloor), isFalse, + reason: 'fresh installs keep the default lookback'); + }); +} + +class _FixedSettingsNotifier extends SettingsNotifier { + _FixedSettingsNotifier() : super(MockSharedPreferencesAsync()) { + state = Settings( + relays: const [], + fullPrivacyMode: false, + mostroPublicKey: _mostroPubkey, + ); + } +} + +class _FakeSessionNotifier extends SessionNotifier { + _FakeSessionNotifier(Ref ref) + : super(ref, MockSessionStorage(), MockSettings()) { + state = const []; + } + + void emit(List sessions) => state = sessions; +} diff --git a/test/services/mostro_service_orders_cursor_test.dart b/test/services/mostro_service_orders_cursor_test.dart new file mode 100644 index 00000000..4984be1d --- /dev/null +++ b/test/services/mostro_service_orders_cursor_test.dart @@ -0,0 +1,183 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:dart_nostr/dart_nostr.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:mockito/mockito.dart'; +import 'package:mostro_mobile/data/models/session.dart'; +import 'package:mostro_mobile/data/repositories/event_storage.dart'; +import 'package:mostro_mobile/features/settings/settings.dart'; +import 'package:mostro_mobile/features/settings/settings_notifier.dart'; +import 'package:mostro_mobile/features/settings/settings_provider.dart'; +import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart'; +import 'package:mostro_mobile/services/chat_cursor_store.dart'; +import 'package:mostro_mobile/services/mostro_service.dart'; +import 'package:mostro_mobile/shared/notifiers/session_notifier.dart'; +import 'package:mostro_mobile/shared/providers/mostro_service_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart'; +import 'package:mostro_mobile/shared/providers/storage_providers.dart'; +import 'package:mostro_mobile/shared/utils/nostr_utils.dart'; +import 'package:sembast/sembast_memory.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:shared_preferences_platform_interface/in_memory_shared_preferences_async.dart'; +import 'package:shared_preferences_platform_interface/shared_preferences_async_platform_interface.dart'; + +import '../mocks.mocks.dart'; + +/// The orders `since` cursor is shared by every session of one node, so it +/// must only ever move to a point past which nothing still needs replaying: +/// it feeds the kind-14 filter alone, and it must not step over an event +/// that was deliberately left unmarked for a later retry. +const _nodePriv = + '0000000000000000000000000000000000000000000000000000000000000003'; +const _tradePriv = + '0000000000000000000000000000000000000000000000000000000000000004'; + +void main() { + final nodeKeys = NostrKeyPairs(private: _nodePriv); + final tradeKeys = NostrKeyPairs(private: _tradePriv); + + late Database db; + late StreamController ordersController; + late MockSubscriptionManagerSpy subscriptionManager; + late _FakeSessionNotifier sessions; + late ProviderContainer container; + late ChatCursorStore cursorStore; + + /// A node message whose payload is an empty tuple: accepted and marked + /// processed by the shortest path through `_processEvent`. + Future nodeMessage({DateTime? createdAt}) async { + final encrypted = + await NostrUtils.encryptNIP44(jsonEncode([]), _nodePriv, tradeKeys.public); + return NostrEvent.fromPartialData( + kind: 14, + content: encrypted, + keyPairs: nodeKeys, + createdAt: createdAt, + tags: [ + ['p', tradeKeys.public], + ], + ); + } + + /// A kind-14 addressed to a trade key no session holds: left unmarked so a + /// later replay can retry it once the session exists. + NostrEvent orphanMessage({required DateTime createdAt}) => + NostrEvent.fromPartialData( + kind: 14, + content: 'undecryptable', + keyPairs: nodeKeys, + createdAt: createdAt, + tags: [ + ['p', 'a' * 64], + ], + ); + + setUp(() async { + SharedPreferencesAsyncPlatform.instance = + InMemorySharedPreferencesAsync.empty(); + db = await newDatabaseFactoryMemory().openDatabase('orders_cursor.db'); + ordersController = StreamController.broadcast(); + subscriptionManager = MockSubscriptionManagerSpy(); + when(subscriptionManager.orders) + .thenAnswer((_) => ordersController.stream); + + container = ProviderContainer(overrides: [ + sharedPreferencesProvider.overrideWithValue(SharedPreferencesAsync()), + eventStorageProvider.overrideWithValue(EventStorage(db: db)), + subscriptionManagerProvider.overrideWithValue(subscriptionManager), + settingsProvider + .overrideWith((ref) => _FixedSettingsNotifier(nodeKeys.public)), + sessionNotifierProvider.overrideWith((ref) { + sessions = _FakeSessionNotifier(ref); + return sessions; + }), + mostroServiceProvider.overrideWith((ref) => MostroService(ref)..init()), + ]); + container.read(sessionNotifierProvider); + container.read(mostroServiceProvider); + cursorStore = container.read(ordersCursorStoreProvider); + + sessions.emit([ + Session( + masterKey: tradeKeys, + tradeKey: tradeKeys, + keyIndex: 0, + fullPrivacy: false, + startTime: DateTime.now(), + )..orderId = 'order-a', + ]); + }); + + tearDown(() async { + await ordersController.close(); + container.dispose(); + await db.close(); + }); + + Future deliver(NostrEvent event) async { + ordersController.add(event); + await Future.delayed(const Duration(milliseconds: 100)); + } + + test('a processed kind-14 event advances the orders cursor', () async { + final event = await nodeMessage(); + + await deliver(event); + + final cursor = await cursorStore.cursorFor(nodeKeys.public); + expect(cursor, isNotNull); + expect(cursor!.millisecondsSinceEpoch ~/ 1000, + event.createdAt!.millisecondsSinceEpoch ~/ 1000); + }); + + test('a processed gift wrap does not advance the kind-14 cursor', () async { + // Gift wrap timestamps are randomized +/- 48 h; letting one move this + // cursor would push `since` past real kind-14 messages. + final wrap = await NostrUtils.createNIP59Event( + jsonEncode([]), + tradeKeys.public, + _nodePriv, + ); + + await deliver(wrap); + + expect(await cursorStore.cursorFor(nodeKeys.public), isNull, + reason: 'only the kind-14 transport owns this cursor'); + }); + + test('an older unmarked event holds the cursor back', () async { + final now = DateTime.now(); + // Arrives for a trade key whose session does not exist yet: not marked, + // so a later replay must still cover it. + await deliver(orphanMessage( + createdAt: now.subtract(const Duration(minutes: 30)))); + + // A newer message for another session must not evict it from the window. + await deliver(await nodeMessage(createdAt: now)); + + expect(await cursorStore.cursorFor(nodeKeys.public), isNull, + reason: 'the cursor is a contiguous watermark, not a high-water mark'); + }); +} + +class _FixedSettingsNotifier extends SettingsNotifier { + _FixedSettingsNotifier(String mostroPublicKey) + : super(MockSharedPreferencesAsync()) { + state = Settings( + relays: const [], + fullPrivacyMode: false, + mostroPublicKey: mostroPublicKey, + ); + } +} + +class _FakeSessionNotifier extends SessionNotifier { + _FakeSessionNotifier(Ref ref) + : super(ref, MockSessionStorage(), MockSettings()) { + state = const []; + } + + void emit(List sessions) => state = sessions; +} From 2afc638f73e2079dc7bec997b86cef517efce9c4 Mon Sep 17 00:00:00 2001 From: grunch Date: Mon, 31 Aug 2026 23:38:11 -0300 Subject: [PATCH 3/3] fix: address review on the orders since cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop the limit from the orders kind-14 filter. `since` already bounds the replay; a limit on top of it is answered with the newest n and silently drops the rest of the window. Those events are never delivered, so they never enter _retryableEvents and nothing holds the cursor back — it advances past them and they are lost for good. The window is widest exactly when it matters: the first launch after upgrading (the sessions floor deliberately reaches back to the oldest live session) and a user offline for a long stretch. main carried no limit here, so this also removes a regression in what the client receives. Replace the fixed 100 ms settle in the cursor regression test with polling on real anchors: processing runs an off-isolate NIP-44 decrypt, a Sembast write and a SharedPreferences write, none of which the test can await, so a fixed delay makes the test fail under load. A processed event settles once its durable marker is written (which _markEventProcessed writes before advancing the cursor); a held one settles once it actually holds the cursor back, via a new debugHeldEventIds test seam. Prune expired holds on the hold path too. The removeWhere ran only inside _advanceOrdersCursor, so a run in which every event is held grew the map unpruned. Fix the _sessionsFloor doc comment, whose first half stated the opposite of what the code does: an older session does widen the window past the default lookback, which is the point of it. --- integration_test/test_helpers.dart | 3 ++ .../subscriptions/subscription_manager.dart | 22 ++++++++---- lib/services/mostro_service.dart | 19 ++++++++-- .../subscriptions/orders_filter_test.dart | 13 ++++--- .../mostro_service_orders_cursor_test.dart | 36 ++++++++++++++++--- 5 files changed, 74 insertions(+), 19 deletions(-) diff --git a/integration_test/test_helpers.dart b/integration_test/test_helpers.dart index ddabe916..378cb32a 100644 --- a/integration_test/test_helpers.dart +++ b/integration_test/test_helpers.dart @@ -267,6 +267,9 @@ class FakeMostroService implements MostroService { @override final Ref ref; + @override + Set get debugHeldEventIds => const {}; + @override void init({List? keys}) {} diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index 6140a242..9494606d 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/lib/features/subscriptions/subscription_manager.dart @@ -362,10 +362,11 @@ class SubscriptionManager { } } - /// Earliest start time across [sessions], capped at the default lookback so - /// a long-lived session cannot widen the window beyond it *and* a short one - /// cannot narrow it below it. Used as the bootstrap `since` when no cursor - /// is stored yet. + /// Earliest start time across [sessions], or the default lookback when that + /// is older — whichever reaches further back. A session older than the + /// lookback widens the window to cover it, and a newer one cannot narrow it + /// below the lookback. Used as the bootstrap `since` when no cursor is + /// stored yet. DateTime _sessionsFloor(List sessions) { final defaultFloor = DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback); @@ -727,14 +728,21 @@ NostrFilter buildOrdersFilter( p: tradeKeys, ); case Transport.nip44: - // kind 14 carries real timestamps: bound the replay with the persisted - // cursor (minus its overlap) and a limit, mirroring the chat filters. + // kind 14 carries real timestamps, so the persisted cursor (minus its + // overlap) bounds the replay. Deliberately no limit on top of it: the + // relay answers a capped filter with the *newest* n and silently drops + // the rest of the window, and events that are never delivered never + // enter _retryableEvents, so nothing holds the cursor back — it + // advances past them and they are lost for good. That window is widest + // exactly when it matters (the first launch after upgrading, or a user + // offline for a long stretch). The filter is already scoped to one + // node's messages addressed to this user's trade keys, so `since` + // alone keeps it small. return NostrFilter( kinds: [14], authors: [mostroPubkey], p: tradeKeys, since: since, - limit: NostrEventExtensions.chatDefaultLimit, ); } } diff --git a/lib/services/mostro_service.dart b/lib/services/mostro_service.dart index 27fd860b..f907cdbd 100644 --- a/lib/services/mostro_service.dart +++ b/lib/services/mostro_service.dart @@ -3,6 +3,7 @@ import 'dart:convert'; import 'package:collection/collection.dart'; import 'package:dart_nostr/dart_nostr.dart'; +import 'package:flutter/foundation.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:mostro_mobile/services/chat_cursor_store.dart'; import 'package:mostro_mobile/services/logger_service.dart'; @@ -146,9 +147,23 @@ class MostroService { /// keeps the replay window covering it. void _holdEventForRetry(NostrEvent event) { if (event.kind != 14 || event.id == null || event.createdAt == null) return; + _pruneExpiredHolds(); _retryableEvents[event.id!] = event.createdAt!; } + /// Drops holds older than [_retryHoldWindow]. Runs on both the hold and the + /// advance path: pruning only when an event is accepted would let the map + /// grow unpruned through a run in which every event is held. + void _pruneExpiredHolds() { + _retryableEvents.removeWhere( + (_, at) => at.isBefore(DateTime.now().subtract(_retryHoldWindow)), + ); + } + + /// Ids currently holding the cursor back. + @visibleForTesting + Set get debugHeldEventIds => _retryableEvents.keys.toSet(); + /// Advances the orders `since` cursor for a processed event. /// /// Only kind 14 counts: the cursor feeds the NIP-44 filter, and gift wrap @@ -162,9 +177,7 @@ class MostroService { _retryableEvents.remove(event.id); if (event.kind != 14) return; final accepted = event.createdAt!; - _retryableEvents.removeWhere( - (_, at) => at.isBefore(DateTime.now().subtract(_retryHoldWindow)), - ); + _pruneExpiredHolds(); final blocked = _retryableEvents.values.any((at) => !at.isAfter(accepted)); if (blocked) return; unawaited( diff --git a/test/features/subscriptions/orders_filter_test.dart b/test/features/subscriptions/orders_filter_test.dart index 0dc66ca0..cffb82ce 100644 --- a/test/features/subscriptions/orders_filter_test.dart +++ b/test/features/subscriptions/orders_filter_test.dart @@ -32,10 +32,10 @@ void main() { expect(filter.p, tradeKeys); }); - // Without since/limit every (re)subscription replayed the node's full - // message history from every relay; kind 14 carries real timestamps, so - // a persisted cursor can bound the replay tightly. - test('v2 (nip44) carries the cursor since and a limit', () { + // Without a since every (re)subscription replayed the node's full message + // history from every relay; kind 14 carries real timestamps, so a + // persisted cursor can bound the replay tightly. + test('v2 (nip44) bounds the replay with since and no limit', () { final since = DateTime.fromMillisecondsSinceEpoch(1700000000 * 1000); final filter = buildOrdersFilter( Transport.nip44, @@ -45,7 +45,10 @@ void main() { ); expect(filter.since, since); - expect(filter.limit, isNotNull); + // A limit on top of since would be answered with the *newest* n, + // silently dropping the rest of the window. Those events are never + // delivered, so they never hold the cursor back and are lost for good. + expect(filter.limit, isNull); }); test('v1 (giftWrap) ignores since: its timestamps are randomized', () { diff --git a/test/services/mostro_service_orders_cursor_test.dart b/test/services/mostro_service_orders_cursor_test.dart index 4984be1d..b1600755 100644 --- a/test/services/mostro_service_orders_cursor_test.dart +++ b/test/services/mostro_service_orders_cursor_test.dart @@ -116,9 +116,20 @@ void main() { await db.close(); }); - Future deliver(NostrEvent event) async { + MostroService service() => container.read(mostroServiceProvider); + + /// Delivers [event] and waits for the pipeline to actually settle, rather + /// than for a fixed delay: processing runs an off-isolate NIP-44 decrypt, a + /// Sembast write and a SharedPreferences write, none of which this test can + /// await directly. A held event settles once it holds the cursor back; a + /// processed one settles once its durable marker is written (which is what + /// _markEventProcessed writes before advancing the cursor). + Future deliver(NostrEvent event, {bool held = false}) async { ordersController.add(event); - await Future.delayed(const Duration(milliseconds: 100)); + final eventStore = EventStorage(db: db); + await _waitFor(() async => held + ? service().debugHeldEventIds.contains(event.id) + : await eventStore.hasItem(event.id!)); } test('a processed kind-14 event advances the orders cursor', () async { @@ -126,6 +137,9 @@ void main() { await deliver(event); + // The advance itself is fire-and-forget after the marker is written. + await _waitFor( + () async => await cursorStore.cursorFor(nodeKeys.public) != null); final cursor = await cursorStore.cursorFor(nodeKeys.public); expect(cursor, isNotNull); expect(cursor!.millisecondsSinceEpoch ~/ 1000, @@ -151,8 +165,10 @@ void main() { final now = DateTime.now(); // Arrives for a trade key whose session does not exist yet: not marked, // so a later replay must still cover it. - await deliver(orphanMessage( - createdAt: now.subtract(const Duration(minutes: 30)))); + await deliver( + orphanMessage(createdAt: now.subtract(const Duration(minutes: 30))), + held: true, + ); // A newer message for another session must not evict it from the window. await deliver(await nodeMessage(createdAt: now)); @@ -162,6 +178,18 @@ void main() { }); } +/// Polls [condition] until it holds or [timeout] elapses, yielding to the +/// event loop between attempts. +Future _waitFor( + Future Function() condition, { + Duration timeout = const Duration(seconds: 10), +}) async { + final deadline = DateTime.now().add(timeout); + while (!await condition() && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 10)); + } +} + class _FixedSettingsNotifier extends SettingsNotifier { _FixedSettingsNotifier(String mostroPublicKey) : super(MockSharedPreferencesAsync()) {