[mobile][desktop] Preserve sync progress across internal retries

piatoss3612 · piatoss3612 · commit fed80c62e1fb · 2026-07-31T12:10:01.000+09:00
diff --git a/integration_test/regtest_sync_startup_stall_recovery_test.dart b/integration_test/regtest_sync_startup_stall_recovery_test.dart
@@ -1,10 +1,12 @@
 import 'package:flutter/widgets.dart';
+import 'package:flutter_riverpod/flutter_riverpod.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:integration_test/integration_test.dart';
 import 'package:zcash_wallet/app.dart';
 import 'package:zcash_wallet/src/core/config/rpc_endpoint_config.dart';
 import 'package:zcash_wallet/src/core/storage/app_secure_store.dart';
 import 'package:zcash_wallet/src/core/storage/wallet_paths.dart';
+import 'package:zcash_wallet/src/providers/sync_provider.dart';
 import 'package:zcash_wallet/src/rust/api/sync.dart' as rust_sync;
 
 import 'support/desktop_regtest_flow.dart';
@@ -17,6 +19,14 @@ const _networkFailure =
 const _endpointFailure =
     'Cannot reach the configured Zcash endpoint. Check your endpoint settings.';
 const _genericFailure = 'Sync failed. Retry sync to continue.';
+const _proxyListenPort = int.fromEnvironment(
+  'ZCASH_E2E_PROXY_PORT',
+  defaultValue: 19068,
+);
+const _lightwalletdTargetPort = int.fromEnvironment(
+  'ZCASH_E2E_LIGHTWALLETD_PORT',
+  defaultValue: 9067,
+);
 
 void main() {
   IntegrationTestWidgetsFlutterBinding.ensureInitialized();
@@ -29,7 +39,11 @@ void main() {
     'recovers from a stalled startup UTXO stream instead of staying at zero percent',
     (tester) async {
       await cleanupDesktopRegtestWallet();
-      final proxy = RegtestLightwalletdProxy(log: e2eLog);
+      final proxy = RegtestLightwalletdProxy(
+        listenPort: _proxyListenPort,
+        targetPort: _lightwalletdTargetPort,
+        log: e2eLog,
+      );
       await proxy.start();
       // Keep later scanning minimal while retaining a persisted completed tip.
       proxy.setSlowHeight(1);
@@ -117,4 +131,85 @@ void main() {
     },
     timeout: const Timeout(Duration(minutes: 3)),
   );
+
+  testWidgets(
+    'keeps logical sync progress monotonic across an internal Rust retry',
+    (tester) async {
+      await cleanupDesktopRegtestWallet();
+      final proxy = RegtestLightwalletdProxy(
+        listenPort: _proxyListenPort,
+        targetPort: _lightwalletdTargetPort,
+        log: e2eLog,
+      );
+      proxy.failBlockRangeCalls(const [2, 3]);
+      await proxy.start();
+      addTearDown(() async {
+        try {
+          await cleanupDesktopRegtestWallet();
+        } finally {
+          await proxy.stop();
+        }
+      });
+
+      final storage = AppSecureStore.instance;
+      await storage.writePlain(kRpcEndpointUrlKey, proxy.url);
+      await storage.writePlain(
+        kRpcEndpointPresetKey,
+        kCustomRpcEndpointPresetId,
+      );
+
+      await tester.pumpWidget(await buildBootstrappedZcashWalletApp());
+      final container = ProviderScope.containerOf(
+        tester.element(
+          find.byKey(const ValueKey('welcome_import_wallet_button')),
+        ),
+      );
+      final observedPercentages = <double>[];
+      final subscription = container.listen(syncProvider, (previous, next) {
+        final sync = next.value;
+        if (sync != null &&
+            (sync.isSyncing || sync.isSyncComplete) &&
+            (observedPercentages.isEmpty ||
+                (observedPercentages.last - sync.percentage).abs() >
+                    0.000001)) {
+          observedPercentages.add(sync.percentage);
+        }
+      });
+      addTearDown(subscription.close);
+
+      await importDesktopRegtestWallet(tester);
+      await pumpUntil(
+        tester,
+        () {
+          final sync = container.read(syncProvider).value;
+          return proxy.failedBlockRangeCount == 2 &&
+              !rust_sync.isSyncRunning() &&
+              sync?.isSyncComplete == true;
+        },
+        description: 'completed sync after an internal block-range retry',
+        timeout: const Duration(minutes: 2),
+      );
+
+      expect(proxy.failedBlockRangeCount, 2);
+      expect(proxy.blockRangeCallCount, greaterThanOrEqualTo(4));
+      expect(
+        observedPercentages,
+        contains(allOf(greaterThan(0.0), lessThan(1.0))),
+      );
+      for (var i = 1; i < observedPercentages.length; i++) {
+        expect(
+          observedPercentages[i] + 0.000001,
+          greaterThanOrEqualTo(observedPercentages[i - 1]),
+          reason:
+              'A retry inside one logical Rust sync must not move progress '
+              'backward: $observedPercentages',
+        );
+      }
+      e2eLog(
+        'internal retry preserved monotonic progress: '
+        '$observedPercentages',
+      );
+    },
+    timeout: const Timeout(Duration(minutes: 3)),
+  );
 }
diff --git a/integration_test/support/regtest_lightwalletd_proxy.dart b/integration_test/support/regtest_lightwalletd_proxy.dart
@@ -36,13 +36,18 @@ class RegtestLightwalletdProxy
   int? _slowHeight;
   int _sendTransactionFailuresRemaining = 0;
   int _failedSendTransactionCount = 0;
+  final Set<int> _blockRangeFailureCalls = <int>{};
+  int _blockRangeCallCount = 0;
+  int _failedBlockRangeCount = 0;
   bool _stallNextAddressUtxosAfterHeaders = false;
   Completer<void>? _addressUtxosStallRelease;
   int _addressUtxosStreamCallCount = 0;
   bool _serveEmptyGenesisTreeState = false;
 
   String get url => 'http://127.0.0.1:$listenPort';
   int get failedSendTransactionCount => _failedSendTransactionCount;
+  int get blockRangeCallCount => _blockRangeCallCount;
+  int get failedBlockRangeCount => _failedBlockRangeCount;
   int get addressUtxosStreamCallCount => _addressUtxosStreamCallCount;
 
   Future<void> start() async {
@@ -85,6 +90,22 @@ class RegtestLightwalletdProxy
     _log('primary proxy will fail next $count SendTransaction call(s)');
   }
 
+  void failBlockRangeCalls(Iterable<int> callNumbers) {
+    final calls = callNumbers.toSet();
+    if (calls.any((call) => call <= _blockRangeCallCount)) {
+      throw ArgumentError.value(
+        callNumbers,
+        'callNumbers',
+        'must contain only future GetBlockRange call numbers',
+      );
+    }
+    _blockRangeFailureCalls.addAll(calls);
+    _log(
+      'primary proxy will fail GetBlockRange call(s) '
+      '${calls.toList()..sort()}',
+    );
+  }
+
   void stallNextAddressUtxosStreamAfterHeaders() {
     _stallNextAddressUtxosAfterHeaders = true;
     _addressUtxosStallRelease = Completer<void>();
@@ -166,6 +187,17 @@ class RegtestLightwalletdProxy
     service.BlockRange request,
   ) {
     _throwIfDown();
+    _blockRangeCallCount += 1;
+    if (_blockRangeFailureCalls.remove(_blockRangeCallCount)) {
+      _failedBlockRangeCount += 1;
+      _log(
+        'primary proxy forced GetBlockRange failure '
+        '(call=$_blockRangeCallCount, failed=$_failedBlockRangeCount)',
+      );
+      return Stream.error(
+        grpc.GrpcError.unavailable('forced GetBlockRange failure'),
+      );
+    }
     return _client.getBlockRange(request);
   }
 
diff --git a/rust/src/wallet/sync_engine/mod.rs b/rust/src/wallet/sync_engine/mod.rs
@@ -1,6 +1,6 @@
 use std::collections::{BTreeSet, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 use nonempty::NonEmpty;
 use rusqlite::{params, OptionalExtension};
@@ -76,6 +76,39 @@ pub struct SyncProgressEvent {
     pub phase: String,
 }
 
+/// Keeps retry attempts inside one `run_sync_inner` call on a single logical
+/// progress scale.
+///
+/// `run_sync_impl` recalculates its work denominator from the DB at the start
+/// of every attempt. That is correct for resuming from the persisted scan
+/// frontier, but the attempt-local percentage would otherwise jump backward to
+/// zero. A fresh `RetryProgressMapper` is created for every logical sync call,
+/// so a genuinely new session (including account-growth catch-up) still starts
+/// from its own zero-based progress.
+#[derive(Debug, Default)]
+struct RetryProgressMapper {
+    retry_floor: f64,
+    last_emitted_percentage: f64,
+}
+
+impl RetryProgressMapper {
+    fn begin_retry(&mut self) {
+        self.retry_floor = self.last_emitted_percentage.clamp(0.0, 1.0);
+    }
+
+    fn map_percentage(&self, attempt_percentage: f64) -> f64 {
+        let attempt_percentage = attempt_percentage.clamp(0.0, 1.0);
+        self.retry_floor + ((1.0 - self.retry_floor) * attempt_percentage)
+    }
+
+    fn map_event(&mut self, mut event: SyncProgressEvent) -> SyncProgressEvent {
+        event.percentage = self.map_percentage(event.percentage);
+        event.display_target_percentage = self.map_percentage(event.display_target_percentage);
+        self.last_emitted_percentage = event.percentage;
+        event
+    }
+}
+
 #[cfg(any(target_os = "macos", target_os = "windows", target_os = "linux"))]
 const BATCH_SIZE_FOREGROUND: u32 = 2000;
 #[cfg(not(any(target_os = "macos", target_os = "windows", target_os = "linux")))]
@@ -1556,6 +1589,7 @@ pub async fn run_sync_inner(
 ) -> Result<(), String> {
     const MAX_RETRIES: u32 = 3;
     let mut last_err = String::new();
+    let retry_progress = Mutex::new(RetryProgressMapper::default());
     *SYNC_START.lock().unwrap() = Some(std::time::Instant::now());
 
     for attempt in 0..=MAX_RETRIES {
@@ -1582,8 +1616,19 @@ pub async fn run_sync_inner(
                     return Ok(());
                 }
             }
+            retry_progress
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .begin_retry();
         }
 
+        let mapped_progress_fn = |event| {
+            let mapped = retry_progress
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .map_event(event);
+            progress_fn(mapped);
+        };
         match run_sync_impl(
             db_data_path,
             lightwalletd_url,
@@ -1592,7 +1637,7 @@ pub async fn run_sync_inner(
             running_mode,
             desired_mode,
             allow_resubmit,
-            &progress_fn,
+            &mapped_progress_fn,
         )
         .await
         {
@@ -2904,6 +2949,20 @@ mod tests {
         );
     }
 
+    fn progress_event(percentage: f64, display_target_percentage: f64) -> SyncProgressEvent {
+        SyncProgressEvent {
+            scanned_height: 0,
+            chain_tip_height: 0,
+            percentage,
+            display_target_percentage,
+            display_target_blocks: 0,
+            is_syncing: true,
+            is_complete: false,
+            has_new_tx: false,
+            phase: "scan".into(),
+        }
+    }
+
     #[test]
     fn work_progress_matches_remaining_block_ratio() {
         let mode = ProgressDisplayMode::Work;
@@ -2915,6 +2974,48 @@ mod tests {
         );
     }
 
+    #[test]
+    fn retry_progress_resumes_from_previous_attempt_floor() {
+        let mut mapper = RetryProgressMapper::default();
+        let first_attempt = mapper.map_event(progress_event(225.0 / 1_200.0, 250.0 / 1_200.0));
+        assert_pct(first_attempt.percentage, 225.0 / 1_200.0);
+
+        mapper.begin_retry();
+        let retry_start = mapper.map_event(progress_event(0.0, 25.0 / 975.0));
+        assert_pct(retry_start.percentage, 225.0 / 1_200.0);
+        assert_pct(retry_start.display_target_percentage, 250.0 / 1_200.0);
+
+        let retry_batch_complete = mapper.map_event(progress_event(25.0 / 975.0, 50.0 / 975.0));
+        assert_pct(retry_batch_complete.percentage, 250.0 / 1_200.0);
+        assert_pct(
+            retry_batch_complete.display_target_percentage,
+            275.0 / 1_200.0,
+        );
+    }
+
+    #[test]
+    fn new_logical_sync_does_not_inherit_previous_retry_floor() {
+        let mut previous_sync = RetryProgressMapper::default();
+        previous_sync.map_event(progress_event(0.5, 0.6));
+
+        let mut new_sync = RetryProgressMapper::default();
+        let event = new_sync.map_event(progress_event(0.0, 0.1));
+
+        assert_pct(event.percentage, 0.0);
+        assert_pct(event.display_target_percentage, 0.1);
+    }
+
+    #[test]
+    fn first_attempt_preserves_existing_phase_progress_semantics() {
+        let mut mapper = RetryProgressMapper::default();
+
+        assert_pct(mapper.map_event(progress_event(0.99, 1.0)).percentage, 0.99);
+        assert_pct(
+            mapper.map_event(progress_event(0.95, 0.975)).percentage,
+            0.95,
+        );
+    }
+
     #[test]
     fn chain_window_progress_uses_session_window_not_absolute_height() {
         let mode = ProgressDisplayMode::ChainWindow {
diff --git a/scripts/e2e/flutter-macos-regtest-sync-startup-stall-recovery.sh b/scripts/e2e/flutter-macos-regtest-sync-startup-stall-recovery.sh
@@ -4,6 +4,8 @@ set -euo pipefail
 ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
 FLUTTER_DEVICE="${FLUTTER_DEVICE:-macos}"
 RESET_REGTEST="${RESET_REGTEST:-1}"
+export ZCASH_E2E_SYNC_BATCH_SIZE="${ZCASH_E2E_SYNC_BATCH_SIZE:-25}"
+export ZCASH_E2E_SYNC_BATCH_DELAY_MS="${ZCASH_E2E_SYNC_BATCH_DELAY_MS:-300}"
 
 require_cmd() {
   if ! command -v "$1" >/dev/null 2>&1; then
