fix(native-worker): recover from dead runs and make queue replacement safe

Stale snapshot (queue wedged forever): the service is START_NOT_STICKY, so a process kill or reboot mid-batch froze the state file at is_running:true with nothing ever rewriting it. Adoption then set isProcessing and polled that flag indefinitely; restored items sat at downloading/queued until the user intervened. Adoption and the reconcile poll now check DownloadService.isServiceRunning() (three consecutive dead polls in the loop), reconcile whatever the worker finished, requeue in-flight items, clear the run id, and hand the batch back to the Dart queue. Adoption gate race (duplicate executors): adoption was gated on _canUseAndroidNativeWorker, which reads extension state that loads asynchronously after startup and settings toggles the user may have flipped mid-run. A background batch surviving task removal was then orphaned while the Dart queue re-downloaded the same items. Adoption now only requires Android; request contexts that cannot be built yet (extensions still loading) are retried on each reconcile poll. Queue replacement race (superseded worker kept running): startNativeWorker only cancelled the coroutine, which cannot interrupt the blocking gomobile call, and it reset the shared cancel flag the old loop checks - so a replaced worker kept downloading its old batch, and its finally block wrote is_running:false over the new run's snapshot and stopSelf'd the service underneath it. Replacement now cancels the old run's Go downloads and FFmpeg work first, and workers carry a generation token: a superseded generation exits at the next checkpoint and skips the final snapshot write and service teardown. An invalid replacement payload no longer stops the service while a run is active.
2026-07-25 21:40:57 +02:00 · 2026-07-10 07:03:40 +07:00
parent fb3fad279c
commit 22151b0988
2 changed files with 185 additions and 29 deletions
@@ -263,6 +263,12 @@ class DownloadService : Service() {
    private var latestCommittedProgressSnapshotSerial = 0L
    @Volatile private var nativeWorkerPaused = false
    @Volatile private var nativeWorkerCancelRequested = false
+    // Bumped every time a new native queue replaces the current one. A worker
+    // coroutine that observes a different generation than its own must stop
+    // without touching the snapshot or the service lifecycle: cancel() alone
+    // cannot interrupt the blocking gomobile call it may be sitting in, and
+    // the shared pause/cancel flags get reset for the new run.
+    @Volatile private var nativeWorkerGeneration = 0L
    
    override fun onCreate() {
        super.onCreate()
@@ -473,17 +479,38 @@ class DownloadService : Service() {
        val requests = try {
            parseNativeDownloadRequests(requestsJson)
        } catch (e: Exception) {
-            writeNativeWorkerSnapshot(
-                isRunning = false,
-                isPaused = false,
-                currentItemId = "",
-                message = "Invalid native queue payload: ${e.message}",
-                settingsJson = settingsJson,
-                includeItems = true
-            )
-            stopForegroundService(cancelNativeWorker = false)
+            if (nativeWorkerJob?.isActive != true) {
+                writeNativeWorkerSnapshot(
+                    isRunning = false,
+                    isPaused = false,
+                    currentItemId = "",
+                    message = "Invalid native queue payload: ${e.message}",
+                    settingsJson = settingsJson,
+                    includeItems = true
+                )
+                stopForegroundService(cancelNativeWorker = false)
+            }
            return
        }
+        // Abort the previous run's in-flight work before the shared flags are
+        // reset for the new run: the coroutine cancel below cannot interrupt a
+        // blocking gomobile download by itself.
+        synchronized(nativeWorkerItems) {
+            for (item in nativeWorkerItems) {
+                if (item.status == "preparing" ||
+                    item.status == "downloading" ||
+                    item.status == "finalizing"
+                ) {
+                    try {
+                        Gobackend.cancelDownload(item.itemId)
+                    } catch (_: Exception) {
+                    }
+                }
+            }
+        }
+        NativeDownloadFinalizer.cancelActiveWork()
+        nativeWorkerGeneration++
+        val generation = nativeWorkerGeneration
        nativeWorkerJob?.cancel(CancellationException("Native queue replaced"))
        nativeWorkerPaused = false
        nativeWorkerCancelRequested = false
@@ -536,7 +563,7 @@ class DownloadService : Service() {
        )

        nativeWorkerJob = serviceScope.launch {
-            runNativeWorker(requests, settingsJson)
+            runNativeWorker(requests, settingsJson, generation)
        }
    }

@@ -602,14 +629,21 @@ class DownloadService : Service() {
        }
    }

-    private suspend fun runNativeWorker(requests: List<NativeDownloadRequest>, settingsJson: String) {
+    private suspend fun runNativeWorker(
+        requests: List<NativeDownloadRequest>,
+        settingsJson: String,
+        generation: Long
+    ) {
        var completed = 0
        var failed = 0
        try {
            var requestIndex = 0
            while (requestIndex < requests.size) {
                val request = requests[requestIndex]
-                while (nativeWorkerPaused && !nativeWorkerCancelRequested) {
+                while (nativeWorkerPaused &&
+                    !nativeWorkerCancelRequested &&
+                    generation == nativeWorkerGeneration
+                ) {
                    writeNativeWorkerSnapshot(
                        isRunning = true,
                        isPaused = true,
@@ -620,7 +654,7 @@ class DownloadService : Service() {
                    )
                    delay(500)
                }
-                if (nativeWorkerCancelRequested) {
+                if (nativeWorkerCancelRequested || generation != nativeWorkerGeneration) {
                    break
                }

@@ -670,6 +704,11 @@ class DownloadService : Service() {
                    }
                    progressJob.cancel()
                    progressJob = null
+                    if (generation != nativeWorkerGeneration) {
+                        // Superseded while blocked in the download call; the
+                        // new run owns the shared state now.
+                        break
+                    }
                    var result = JSONObject(response)
                    if (result.optBoolean("success", false)) {
                        currentStatus = "finalizing"
@@ -695,6 +734,7 @@ class DownloadService : Service() {
                        ) {
                            nativeWorkerCancelRequested ||
                                nativeWorkerPaused ||
+                                generation != nativeWorkerGeneration ||
                                nativeWorkerJob?.isActive == false
                        }
                    }
@@ -797,19 +837,21 @@ class DownloadService : Service() {
                }
            }
        } finally {
-            if (!nativeWorkerCancelRequested) {
-                flushNativeAlbumReplayGainJournalIfComplete()
+            if (generation == nativeWorkerGeneration) {
+                if (!nativeWorkerCancelRequested) {
+                    flushNativeAlbumReplayGainJournalIfComplete()
+                }
+                currentStatus = "finalizing"
+                writeNativeWorkerSnapshot(
+                    isRunning = false,
+                    isPaused = false,
+                    currentItemId = "",
+                    message = if (nativeWorkerCancelRequested) "Cancelled" else "Finished",
+                    settingsJson = settingsJson,
+                    includeItems = true
+                )
+                stopForegroundService(cancelNativeWorker = false)
            }
-            currentStatus = "finalizing"
-            writeNativeWorkerSnapshot(
-                isRunning = false,
-                isPaused = false,
-                currentItemId = "",
-                message = if (nativeWorkerCancelRequested) "Cancelled" else "Finished",
-                settingsJson = settingsJson,
-                includeItems = true
-            )
-            stopForegroundService(cancelNativeWorker = false)
        }
    }

@@ -3833,7 +3833,12 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
    List<DownloadItem> restoredItems,
  ) async {
    final settings = ref.read(settingsProvider);
-    if (!_canUseAndroidNativeWorker(settings)) {
+    // Adoption deliberately skips _canUseAndroidNativeWorker: that gate reads
+    // extension state that loads asynchronously after startup, and toggles the
+    // user may have flipped mid-run. If a native batch is already running we
+    // must reconcile with it regardless, or the Dart queue would download the
+    // same items a second time alongside it.
+    if (!Platform.isAndroid) {
      return false;
    }

@@ -3872,20 +3877,47 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
    }

    final contexts = <String, _NativeWorkerRequestContext>{};
+    final pendingContextIds = <String>{};
    for (final item in restoredItems) {
      if (!snapshotIds.contains(item.id)) continue;
      final context = await _buildAndroidNativeWorkerRequest(item, settings);
      if (context != null) {
        contexts[item.id] = context;
+      } else {
+        // Extensions may still be loading this early in startup; remember the
+        // item and retry building its context on later reconcile polls.
+        pendingContextIds.add(item.id);
      }
    }
-    if (contexts.isEmpty) {
+
+    final snapshotClaimsRunning = snapshot['is_running'] == true;
+    if (snapshotClaimsRunning && !await _isNativeWorkerServiceAlive()) {
+      // The service died mid-batch (process kill/reboot): the snapshot is
+      // frozen at is_running=true and nothing will ever update it. Reconcile
+      // what the worker managed to finish, requeue the rest, and let the
+      // caller fall back to the Dart queue.
+      _log.w(
+        'Native worker snapshot claims running but the service is dead; reclaiming run $runId',
+      );
+      if (contexts.isNotEmpty) {
+        await _applyAndroidNativeWorkerSnapshot(
+          snapshot,
+          contexts,
+          <String>{},
+          settings,
+        );
+      }
+      _requeueInFlightNativeWorkerItems(snapshotIds);
+      await _clearNativeWorkerRunId(runId);
+      return false;
+    }
+    if (contexts.isEmpty && pendingContextIds.isEmpty) {
      return false;
    }

    _log.i('Adopting Android native worker snapshot');
    final reconciledIds = <String>{};
-    _totalQueuedAtStart = contexts.length;
+    _totalQueuedAtStart = contexts.length + pendingContextIds.length;
    _completedInSession = 0;
    _failedInSession = 0;
    state = state.copyWith(
@@ -3903,6 +3935,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
      unawaited(
        _continueAndroidNativeWorkerAdoption(
          contexts,
+          pendingContextIds,
          reconciledIds,
          settings,
          runId,
@@ -3920,19 +3953,100 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
    return true;
  }

+  Future<bool> _isNativeWorkerServiceAlive() async {
+    try {
+      return await PlatformBridge.isDownloadServiceRunning();
+    } catch (_) {
+      return false;
+    }
+  }
+
+  /// Puts items the dead native worker left mid-flight back into the queue so
+  /// the Dart queue can pick them up.
+  void _requeueInFlightNativeWorkerItems(Set<String> itemIds) {
+    for (final id in itemIds) {
+      final current = _findItemById(id);
+      if (current == null) continue;
+      if (current.status == DownloadStatus.downloading ||
+          current.status == DownloadStatus.finalizing) {
+        updateItemStatus(id, DownloadStatus.queued, progress: 0.0);
+      }
+    }
+  }
+
+  Future<void> _rebuildPendingNativeWorkerContexts(
+    Map<String, _NativeWorkerRequestContext> contexts,
+    Set<String> pendingContextIds,
+    AppSettings settings,
+  ) async {
+    if (pendingContextIds.isEmpty) return;
+    final resolved = <String>{};
+    for (final id in pendingContextIds) {
+      final item = _findItemById(id);
+      if (item == null) {
+        resolved.add(id);
+        continue;
+      }
+      final context = await _buildAndroidNativeWorkerRequest(item, settings);
+      if (context != null) {
+        contexts[id] = context;
+        resolved.add(id);
+      }
+    }
+    pendingContextIds.removeAll(resolved);
+  }
+
  Future<void> _continueAndroidNativeWorkerAdoption(
    Map<String, _NativeWorkerRequestContext> contexts,
+    Set<String> pendingContextIds,
    Set<String> reconciledIds,
    AppSettings settings,
    String runId,
  ) async {
+    var deadServicePolls = 0;
    try {
      while (true) {
        final snapshot = await PlatformBridge.getNativeDownloadWorkerSnapshot();
-        if (!_isNativeWorkerSnapshotForRun(snapshot, runId)) {
+        final matchesRun = _isNativeWorkerSnapshotForRun(snapshot, runId);
+        if (!matchesRun || snapshot['is_running'] == true) {
+          if (await _isNativeWorkerServiceAlive()) {
+            deadServicePolls = 0;
+          } else {
+            deadServicePolls++;
+            if (deadServicePolls >= 3) {
+              _log.w(
+                'Native worker run $runId died without a final snapshot; reclaiming queue',
+              );
+              if (matchesRun) {
+                await _applyAndroidNativeWorkerSnapshot(
+                  snapshot,
+                  contexts,
+                  reconciledIds,
+                  settings,
+                );
+              }
+              _requeueInFlightNativeWorkerItems(
+                {...contexts.keys, ...pendingContextIds},
+              );
+              await _clearNativeWorkerRunId(runId);
+              if (state.items.any(
+                (item) => item.status == DownloadStatus.queued,
+              )) {
+                Future.microtask(() => _processQueue());
+              }
+              return;
+            }
+          }
+        }
+        if (!matchesRun) {
          await Future<void>.delayed(const Duration(seconds: 1));
          continue;
        }
+        await _rebuildPendingNativeWorkerContexts(
+          contexts,
+          pendingContextIds,
+          settings,
+        );
        await _applyAndroidNativeWorkerSnapshot(
          snapshot,
          contexts,