perf: bound the orders subscription replay with a persisted since cursor

integration_test/test_helpers.dart

@@ -267,6 +267,9 @@ class FakeMostroService implements MostroService {
   @override
   final Ref ref;
 
+  @override
+  Set<String> get debugHeldEventIds => const {};
+
   @override
   void init({List<NostrKeyPairs>? keys}) {}
 

lib/features/subscriptions/subscription_manager.dart

@@ -295,6 +295,16 @@ class SubscriptionManager {
             .whereType<String>();
         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) {
@@ -352,6 +362,21 @@ class SubscriptionManager {
     }
   }
 
+  /// 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<Session> 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<Session> sessions) {
     switch (type) {
@@ -365,10 +390,30 @@ 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');
+        }
+        // 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 ??
+            _sessionsFloor(sessions)
+                .subtract(ChatCursorStore.cursorOverlap);
         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 +715,34 @@ class SubscriptionManager {
 NostrFilter buildOrdersFilter(
   Transport transport,
   List<String> 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, 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,
       );
   }
 }

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<ChatCursorStore>(
     keyPrefix: ChatCursorStore.peerKeyPrefix,
   ),
 );
+
+/// Orders (node kind-14 message) cursors, keyed by the node pubkey.
+final ordersCursorStoreProvider = Provider<ChatCursorStore>(
+  (ref) => ChatCursorStore(
+    ref.watch(sharedPreferencesProvider),
+    keyPrefix: ChatCursorStore.ordersKeyPrefix,
+  ),
+);

lib/services/mostro_service.dart

@@ -3,7 +3,9 @@ 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';
 import 'package:mostro_mobile/data/enums.dart';
 import 'package:mostro_mobile/data/models.dart';
@@ -119,11 +121,70 @@ 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<void> _markEventProcessed(NostrEvent event) {
-    return ref.read(eventStorageProvider).putItem(event.id!, {
-      'id': event.id,
+  Future<void> _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<String, DateTime> _retryableEvents = <String, DateTime>{};
+
+  /// 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;
+    _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<String> 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
+  /// (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!;
+    _pruneExpiredHolds();
+    final blocked = _retryableEvents.values.any((at) => !at.isAfter(accepted));
+    if (blocked) return;
+    unawaited(
+      ref
+          .read(ordersCursorStoreProvider)
+          .advance(_settings.mostroPublicKey, accepted),
+    );
   }
 
   Future<void> _onData(NostrEvent event) async {
@@ -149,6 +210,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;
     }
@@ -172,7 +234,10 @@ class MostroService {
         decryptedId = decryptedEvent.id;
       }
 
-      if (content == null) return;
+      if (content == null) {
+        _holdEventForRetry(event);
+        return;
+      }
 
       final result = jsonDecode(content);
 
@@ -224,6 +289,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);
     }
   }

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<void> _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<void>.delayed(const Duration(milliseconds: 10));
+  }
+}

test/features/subscriptions/orders_filter_test.dart

@@ -31,5 +31,35 @@ void main() {
       expect(filter.authors, [mostroPubkey]);
       expect(filter.p, tradeKeys);
     });
+
+    // 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,
+        tradeKeys,
+        mostroPubkey,
+        since: since,
+      );
+
+      expect(filter.since, since);
+      // 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', () {
+      final filter = buildOrdersFilter(
+        Transport.giftWrap,
+        tradeKeys,
+        mostroPubkey,
+        since: DateTime.now(),
+      );
+
+      expect(filter.since, isNull);
+    });
   });
 }