perf(downloads): bound native queue state

2026-09-03 16:50:40 +02:00 · 2026-08-27 00:17:25 +07:00
parent 5e8608cef9
commit 04c42bd511
10 changed files with 483 additions and 108 deletions
@@ -64,6 +64,8 @@ class DownloadService : Service() {
        const val ACTION_APPEND_NATIVE_QUEUE = "com.zarz.spotiflac.action.APPEND_NATIVE_QUEUE"
        const val ACTION_FINISH_NATIVE_QUEUE_PREPARATION =
            "com.zarz.spotiflac.action.FINISH_NATIVE_QUEUE_PREPARATION"
+        const val ACTION_ACKNOWLEDGE_NATIVE_QUEUE_ITEMS =
+            "com.zarz.spotiflac.action.ACKNOWLEDGE_NATIVE_QUEUE_ITEMS"
        const val ACTION_PAUSE_NATIVE_QUEUE = "com.zarz.spotiflac.action.PAUSE_NATIVE_QUEUE"
        const val ACTION_RESUME_NATIVE_QUEUE = "com.zarz.spotiflac.action.RESUME_NATIVE_QUEUE"
        const val ACTION_CANCEL_NATIVE_QUEUE = "com.zarz.spotiflac.action.CANCEL_NATIVE_QUEUE"
@@ -79,6 +81,7 @@ class DownloadService : Service() {
        const val EXTRA_REQUESTS_PATH = "requests_path"
        const val EXTRA_SETTINGS_PATH = "settings_path"
        const val EXTRA_RUN_ID = "run_id"
+        const val EXTRA_ITEM_IDS_JSON = "item_ids_json"
        internal const val NATIVE_WORKER_STATE_FILE = "native_download_worker_state.json"
        internal const val NATIVE_WORKER_PROGRESS_FILE = "native_download_worker_progress.json"
        internal const val NATIVE_REPLAYGAIN_JOURNAL_FILE = "native_replaygain_journal.json"
@@ -168,6 +171,19 @@ class DownloadService : Service() {
            context.startService(intent)
        }

+        fun acknowledgeNativeQueueItems(
+            context: Context,
+            runId: String,
+            itemIdsJson: String,
+        ) {
+            val intent = Intent(context, DownloadService::class.java).apply {
+                action = ACTION_ACKNOWLEDGE_NATIVE_QUEUE_ITEMS
+                putExtra(EXTRA_RUN_ID, runId)
+                putExtra(EXTRA_ITEM_IDS_JSON, itemIdsJson)
+            }
+            context.startService(intent)
+        }
+
        fun pauseNativeQueue(context: Context) {
            val intent = Intent(context, DownloadService::class.java).apply {
                action = ACTION_PAUSE_NATIVE_QUEUE
@@ -294,7 +310,6 @@ class DownloadService : Service() {
        val itemId: String,
        val trackName: String,
        val artistName: String,
-        val itemJson: String = "",
        var status: String = "queued",
        var progress: Double = 0.0,
        var bytesReceived: Long = 0L,
@@ -312,13 +327,14 @@ class DownloadService : Service() {

    internal val serviceScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
    internal var nativeWorkerJob: Job? = null
+    @Volatile internal var pendingNativeItemsSnapshotJob: Job? = null
    private var nativeWorkerRequestChannel: Channel<NativeDownloadRequest>? = null
    @Volatile private var nativeWorkerPreparationComplete = true
    private var wakeLock: PowerManager.WakeLock? = null
    private var currentTrackName = ""
    private var currentArtistName = ""
    internal var currentStatus = "preparing"
-    private var queueCount = 0
+    internal var queueCount = 0
    // Signature of the last home-screen widget push; keeps widget updates
    // event-driven (track/status/queue changes, 25% steps), never per byte.
    private var widgetSignature = ""
@@ -327,6 +343,7 @@ class DownloadService : Service() {
    internal var nativeWorkerRunId = ""
    @Volatile private var nativeWorkerCurrentItemId = ""
    internal val nativeWorkerItems = mutableListOf<NativeWorkerItem>()
+    internal val nativeWorkerTerminalStatuses = mutableMapOf<String, String>()
    internal val nativeReplayGainEntries = mutableListOf<JSONObject>()
    internal val nativeReplayGainRequestAlbumKeys = mutableMapOf<String, String>()
    internal val snapshotWriteLock = Any()
@@ -411,6 +428,12 @@ class DownloadService : Service() {
                    intent.getStringExtra(EXTRA_RUN_ID).orEmpty(),
                )
            }
+            ACTION_ACKNOWLEDGE_NATIVE_QUEUE_ITEMS -> {
+                acknowledgeNativeWorkerItems(
+                    intent.getStringExtra(EXTRA_RUN_ID).orEmpty(),
+                    intent.getStringExtra(EXTRA_ITEM_IDS_JSON).orEmpty(),
+                )
+            }
            ACTION_PAUSE_NATIVE_QUEUE -> {
                nativeWorkerPaused = true
                cancelActiveNativeItemForPause()
@@ -438,6 +461,7 @@ class DownloadService : Service() {
                nativeWorkerVerificationPaused = false
                nativeWorkerPreparationComplete = true
                nativeWorkerRequestChannel?.close()
+                cancelScheduledNativeWorkerItemsSnapshot()
                cancelNativeVerificationNotification()
                synchronized(nativeWorkerItems) {
                    for (item in nativeWorkerItems) {
@@ -529,6 +553,8 @@ class DownloadService : Service() {
        nativeWorkerCancelRequested = true
        nativeWorkerPreparationComplete = true
        nativeWorkerRequestChannel?.close()
+        pendingNativeItemsSnapshotJob?.cancel()
+        pendingNativeItemsSnapshotJob = null
        // Supersede the coroutine before cancelling it. Its catch/finally
        // blocks must not publish a skipped/finished state over the recovery
        // snapshot written below.
@@ -655,6 +681,8 @@ class DownloadService : Service() {
        }
        NativeDownloadFinalizer.cancelActiveWork()
        nativeWorkerRequestChannel?.close()
+        pendingNativeItemsSnapshotJob?.cancel()
+        pendingNativeItemsSnapshotJob = null
        nativeWorkerGeneration++
        val generation = nativeWorkerGeneration
        nativeWorkerJob?.cancel(CancellationException("Native queue replaced"))
@@ -696,13 +724,13 @@ class DownloadService : Service() {
        }
        synchronized(nativeWorkerItems) {
            nativeWorkerItems.clear()
+            nativeWorkerTerminalStatuses.clear()
            nativeWorkerItems.addAll(
                requests.map {
                    NativeWorkerItem(
                        itemId = it.itemId,
                        trackName = it.trackName,
                        artistName = it.artistName,
-                        itemJson = it.itemJson
                    )
                }
            )
@@ -762,7 +790,9 @@ class DownloadService : Service() {
        if (requests.isEmpty()) return

        val knownIds = synchronized(nativeWorkerItems) {
-            nativeWorkerItems.mapTo(mutableSetOf()) { it.itemId }
+            nativeWorkerItems.mapTo(mutableSetOf()) { it.itemId }.apply {
+                addAll(nativeWorkerTerminalStatuses.keys)
+            }
        }
        val additions = requests.filter { knownIds.add(it.itemId) }
        if (additions.isEmpty()) return
@@ -788,7 +818,6 @@ class DownloadService : Service() {
                        itemId = it.itemId,
                        trackName = it.trackName,
                        artistName = it.artistName,
-                        itemJson = it.itemJson,
                    )
                },
            )
@@ -798,12 +827,11 @@ class DownloadService : Service() {
            channel.trySend(request)
        }
        writeNativeReplayGainJournal()
-        writeNativeWorkerSnapshotAsync(
+        scheduleNativeWorkerItemsSnapshot(
            isRunning = nativeWorkerJob?.isActive == true,
            isPaused = isNativeWorkerPaused(),
            currentItemId = nativeWorkerCurrentItemId,
            message = "Preparing queue",
-            includeItems = true,
        )
    }

@@ -811,9 +839,54 @@ class DownloadService : Service() {
        if (runId.isBlank() || runId != nativeWorkerRunId) return
        nativeWorkerPreparationComplete = true
        nativeWorkerRequestChannel?.close()
+        flushScheduledNativeWorkerItemsSnapshot(
+            isRunning = nativeWorkerJob?.isActive == true,
+            isPaused = isNativeWorkerPaused(),
+            currentItemId = nativeWorkerCurrentItemId,
+            message = "Queue prepared",
+        )
        writeNativeAlbumReplayGainIfComplete()
    }

+    private fun acknowledgeNativeWorkerItems(runId: String, itemIdsJson: String) {
+        if (runId.isBlank() || runId != nativeWorkerRunId || itemIdsJson.isBlank()) return
+        val itemIds = try {
+            val array = JSONArray(itemIdsJson)
+            mutableSetOf<String>().apply {
+                for (index in 0 until array.length()) {
+                    val itemId = array.optString(index, "").trim()
+                    if (itemId.isNotEmpty()) {
+                        add(itemId)
+                    }
+                }
+            }
+        } catch (_: Exception) {
+            return
+        }
+        if (itemIds.isEmpty()) return
+
+        var releasedAny = false
+        synchronized(nativeWorkerItems) {
+            val iterator = nativeWorkerItems.iterator()
+            while (iterator.hasNext()) {
+                val item = iterator.next()
+                if (item.itemId !in itemIds || !NativeWorkerPolicy.isTerminalStatus(item.status)) {
+                    continue
+                }
+                nativeWorkerTerminalStatuses[item.itemId] = item.status
+                iterator.remove()
+                releasedAny = true
+            }
+        }
+        if (!releasedAny) return
+        scheduleNativeWorkerItemsSnapshot(
+            isRunning = nativeWorkerJob?.isActive == true,
+            isPaused = isNativeWorkerPaused(),
+            currentItemId = nativeWorkerCurrentItemId,
+            message = "Queue updated",
+        )
+    }
+
    internal fun isNativeWorkerPaused(): Boolean =
        nativeWorkerPaused ||
            nativeWorkerNetworkPaused ||
@@ -1278,39 +1351,41 @@ class DownloadService : Service() {

        try {
            supervisorScope {
-                val itemJobs = mutableListOf<Job>()
-                for (request in requests) {
-                    if (nativeWorkerCancelRequested ||
-                        generation != nativeWorkerGeneration
-                    ) {
-                        break
-                    }
-                    itemJobs += launch {
-                        val providerKey = nativeRequestProviderKey(request)
-                        val providerLimit = minOf(
-                            concurrency,
-                            nativeRequestProviderConcurrency(request),
-                        )
-                        val providerSemaphore = providerSemaphores.computeIfAbsent(
-                            providerKey,
-                        ) {
-                            Semaphore(providerLimit)
+                val workers = List(concurrency) {
+                    launch {
+                        for (request in requests) {
+                            if (nativeWorkerCancelRequested ||
+                                generation != nativeWorkerGeneration
+                            ) {
+                                break
+                            }
+                            val providerKey = nativeRequestProviderKey(request)
+                            val providerLimit = minOf(
+                                concurrency,
+                                nativeRequestProviderConcurrency(request),
+                            )
+                            val providerSemaphore = providerSemaphores.computeIfAbsent(
+                                providerKey,
+                            ) {
+                                Semaphore(providerLimit)
+                            }
+                            processConcurrentNativeRequest(
+                                request = request,
+                                settingsJson = settingsJson,
+                                generation = generation,
+                                networkSemaphore = networkSemaphore,
+                                providerSemaphore = providerSemaphore,
+                                finalizerMutex = finalizerMutex,
+                                rateLimitAttempts = rateLimitAttempts,
+                            )
                        }
-                        processConcurrentNativeRequest(
-                            request = request,
-                            settingsJson = settingsJson,
-                            generation = generation,
-                            networkSemaphore = networkSemaphore,
-                            providerSemaphore = providerSemaphore,
-                            finalizerMutex = finalizerMutex,
-                            rateLimitAttempts = rateLimitAttempts,
-                        )
                    }
                }
-                itemJobs.joinAll()
+                workers.joinAll()
            }
        } finally {
            if (generation == nativeWorkerGeneration) {
+                cancelScheduledNativeWorkerItemsSnapshot()
                nativeWorkerRequestChannel = null
                nativeWorkerPreparationComplete = true
                if (!nativeWorkerCancelRequested) {
@@ -1647,6 +1722,7 @@ class DownloadService : Service() {
            }
        } finally {
            if (generation == nativeWorkerGeneration) {
+                cancelScheduledNativeWorkerItemsSnapshot()
                if (!nativeWorkerCancelRequested) {
                    flushNativeAlbumReplayGainJournalIfComplete()
                }
@@ -1709,6 +1785,7 @@ class DownloadService : Service() {

    @Synchronized
    private fun stopForegroundService(cancelNativeWorker: Boolean = true) {
+        cancelScheduledNativeWorkerItemsSnapshot()
        if (cancelNativeWorker) {
            nativeWorkerCancelRequested = true
            nativeWorkerPreparationComplete = true
@@ -1939,6 +2016,7 @@ class DownloadService : Service() {
    }
    
    override fun onDestroy() {
+        cancelScheduledNativeWorkerItemsSnapshot()
        unregisterNativeWorkerNetworkCallback()
        nativeWorkerCancelRequested = true
        nativeWorkerPreparationComplete = true
@@ -39,9 +39,7 @@ internal fun DownloadService.writeNativeAlbumReplayGainIfComplete(): Boolean {
    }
    if (entries.size <= 1) return true

-    val statuses = synchronized(nativeWorkerItems) {
-        nativeWorkerItems.associate { it.itemId to it.status }
-    }
+    val statuses = nativeWorkerStatusesSnapshot()
    val requestKeys = synchronized(nativeReplayGainRequestAlbumKeys) {
        nativeReplayGainRequestAlbumKeys.toMap()
    }
@@ -93,9 +91,7 @@ internal fun DownloadService.writeNativeReplayGainJournal() {
    val entries = synchronized(nativeReplayGainEntries) {
        nativeReplayGainEntries.map { JSONObject(it.toString()) }
    }
-    val statuses = synchronized(nativeWorkerItems) {
-        nativeWorkerItems.associate { it.itemId to it.status }
-    }
+    val statuses = nativeWorkerStatusesSnapshot()
    synchronized(DownloadService.NATIVE_REPLAYGAIN_JOURNAL_FILE_LOCK) {
        val file = AtomicFile(File(filesDir, DownloadService.NATIVE_REPLAYGAIN_JOURNAL_FILE))
        val existing = readNativeReplayGainJournalLocked(file)
@@ -134,6 +130,16 @@ internal fun DownloadService.writeNativeReplayGainJournal() {
    }
 }

+internal fun DownloadService.nativeWorkerStatusesSnapshot(): Map<String, String> {
+    return synchronized(nativeWorkerItems) {
+        val statuses = nativeWorkerTerminalStatuses.toMutableMap()
+        for (item in nativeWorkerItems) {
+            statuses[item.itemId] = item.status
+        }
+        statuses
+    }
+}
+
 internal fun DownloadService.readNativeReplayGainJournalLocked(file: AtomicFile): JSONObject? {
    return try {
        if (!file.baseFile.exists()) return null
@@ -144,6 +144,47 @@ internal fun DownloadService.writeNativeWorkerSnapshotAsync(
    }
 }

+internal fun DownloadService.scheduleNativeWorkerItemsSnapshot(
+    isRunning: Boolean,
+    isPaused: Boolean,
+    currentItemId: String,
+    message: String,
+) {
+    pendingNativeItemsSnapshotJob?.cancel()
+    pendingNativeItemsSnapshotJob = serviceScope.launch {
+        delay(250)
+        writeNativeWorkerSnapshot(
+            isRunning = isRunning,
+            isPaused = isPaused,
+            currentItemId = currentItemId,
+            message = message,
+            includeItems = true,
+        )
+    }
+}
+
+internal fun DownloadService.flushScheduledNativeWorkerItemsSnapshot(
+    isRunning: Boolean,
+    isPaused: Boolean,
+    currentItemId: String,
+    message: String,
+) {
+    pendingNativeItemsSnapshotJob?.cancel()
+    pendingNativeItemsSnapshotJob = null
+    writeNativeWorkerSnapshotAsync(
+        isRunning = isRunning,
+        isPaused = isPaused,
+        currentItemId = currentItemId,
+        message = message,
+        includeItems = true,
+    )
+}
+
+internal fun DownloadService.cancelScheduledNativeWorkerItemsSnapshot() {
+    pendingNativeItemsSnapshotJob?.cancel()
+    pendingNativeItemsSnapshotJob = null
+}
+
 internal fun DownloadService.readNativeWorkerRunIdFromSnapshotFile(): String {
    return try {
        synchronized(DownloadService.NATIVE_WORKER_STATE_FILE_LOCK) {
@@ -231,7 +272,14 @@ internal fun DownloadService.nativeWorkerCounts(): DownloadService.NativeWorkerC
    var failed = 0
    var skipped = 0
    synchronized(nativeWorkerItems) {
-        total = nativeWorkerItems.size
+        total = nativeWorkerTerminalStatuses.size + nativeWorkerItems.size
+        for (status in nativeWorkerTerminalStatuses.values) {
+            when (status) {
+                "completed" -> completed++
+                "failed" -> failed++
+                "skipped" -> skipped++
+            }
+        }
        for (item in nativeWorkerItems) {
            when (item.status) {
                "completed" -> completed++
@@ -241,7 +289,7 @@ internal fun DownloadService.nativeWorkerCounts(): DownloadService.NativeWorkerC
        }
    }
    return DownloadService.NativeWorkerCounts(
-        total = total,
+        total = maxOf(queueCount, total),
        completed = completed,
        failed = failed,
        skipped = skipped
@@ -286,7 +334,6 @@ internal fun DownloadService.nativeWorkerItemSnapshotLocked(item: DownloadServic
    if (includeStatic) {
        json.put("track_name", item.trackName)
            .put("artist_name", item.artistName)
-            .put("item_json", item.itemJson)
    }
    if (item.error.isNotBlank()) {
        json.put("error", item.error)
@@ -294,4 +341,3 @@ internal fun DownloadService.nativeWorkerItemSnapshotLocked(item: DownloadServic
    item.resultJson?.let { json.put("result", it) }
    return json
 }
-
@@ -1777,6 +1777,18 @@ class MainActivity: FlutterFragmentActivity() {
                            }
                            result.success(null)
                        }
+                        "acknowledgeNativeDownloadWorkerItems" -> {
+                            val runId = call.argument<String>("run_id") ?: ""
+                            val itemIdsJson = call.argument<String>("item_ids_json") ?: "[]"
+                            if (runId.isNotBlank()) {
+                                DownloadService.acknowledgeNativeQueueItems(
+                                    this@MainActivity,
+                                    runId,
+                                    itemIdsJson,
+                                )
+                            }
+                            result.success(null)
+                        }
                        "pauseNativeDownloadWorker" -> {
                            DownloadService.pauseNativeQueue(this@MainActivity)
                            result.success(null)
@@ -69,6 +69,9 @@ internal object NativeWorkerPolicy {
        failed: Int,
    ): Boolean = !cancelRequested && completed + failed > 0

+    fun isTerminalStatus(status: String): Boolean =
+        status == "completed" || status == "failed" || status == "skipped"
+
    fun statusAfterWorkerStop(status: String): String = when (status) {
        "preparing", "downloading", "finalizing" -> "queued"
        else -> status
@@ -160,4 +160,14 @@ class NativeWorkerPolicyTest {
        assertEquals("failed", NativeWorkerPolicy.statusAfterWorkerStop("failed"))
        assertEquals("skipped", NativeWorkerPolicy.statusAfterWorkerStop("skipped"))
    }
+
+    @Test
+    fun acknowledgedPayloadsAreReleasedOnlyForTerminalItems() {
+        listOf("completed", "failed", "skipped").forEach { status ->
+            assertTrue(NativeWorkerPolicy.isTerminalStatus(status))
+        }
+        listOf("queued", "preparing", "downloading", "finalizing").forEach { status ->
+            assertFalse(NativeWorkerPolicy.isTerminalStatus(status))
+        }
+    }
 }
@@ -53,6 +53,52 @@ part 'download_queue_provider_single_item.dart';

 final _log = AppLogger('DownloadQueue');

+typedef _PersistedQueueItemCache = Map<String, DownloadItem?>;
+
+/// Prevents asynchronous queue startup checks from overlapping before
+/// [DownloadQueueState.isProcessing] can be published.
+class QueueProcessingGate {
+  bool _active = false;
+  bool _pending = false;
+
+  bool tryEnter() {
+    if (_active) {
+      _pending = true;
+      return false;
+    }
+    _active = true;
+    return true;
+  }
+
+  bool leave() {
+    _active = false;
+    final shouldRunAgain = _pending;
+    _pending = false;
+    return shouldRunAgain;
+  }
+}
+
+/// Encodes only restart-relevant queue state. Transfer progress is delivered
+/// by the native progress stream and must not rewrite SQLite every few seconds.
+String encodeDownloadQueueItemForPersistence(DownloadItem item) {
+  final persistedStatus = downloadQueuePersistenceStatus(item.status);
+  final json = item.toJson()
+    ..['status'] = persistedStatus.name
+    ..['progress'] = 0.0
+    ..['speedMBps'] = 0.0
+    ..['bytesReceived'] = 0
+    ..['bytesTotal'] = 0
+    ..['preparationStage'] = '';
+  return jsonEncode(json);
+}
+
+DownloadStatus downloadQueuePersistenceStatus(DownloadStatus status) =>
+    switch (status) {
+      DownloadStatus.downloading ||
+      DownloadStatus.finalizing => DownloadStatus.queued,
+      _ => status,
+    };
+
 /// Set on queued items when the persisted Android SAF grant fails validation.
 /// The queue UI matches on this to offer re-selecting the download folder.
 const String safPermissionLostErrorMessage =
@@ -264,7 +310,9 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
  Timer? _queuePersistDebounce;
  Future<void> _queuePersistenceWrite = Future<void>.value();
  Future<void> _queuePausePersistenceWrite = Future<void>.value();
-  final Map<String, String> _persistedQueueJsonById = {};
+  final _PersistedQueueItemCache _persistedQueueItemById = {};
+  final Set<String> _nonCanonicalPersistedQueueIds = {};
+  final QueueProcessingGate _queueProcessingGate = QueueProcessingGate();
  StreamSubscription<List<ConnectivityResult>>? _connectivitySub;
  int _downloadCount = 0;
  static const _cleanupInterval = 50;
@@ -273,6 +321,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
  static const _progressStreamBootstrapTimeout = Duration(seconds: 3);
  static const _queueSchedulingInterval = Duration(milliseconds: 250);
  static const _queuePersistDebounceDuration = Duration(milliseconds: 350);
+  static const _nativePreparationBatchSize = 32;
+  static const _nativePreparationWindowSize = 128;
  static const _nativeWorkerRunIdPrefsKey =
      'download_queue_native_worker_run_id';
  static const _userPausedQueuePrefsKey = 'download_queue_user_paused_v1';
@@ -461,18 +511,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
      await _appStateDb.migrateQueueFromSharedPreferences();
      final restorePaused = await _loadUserPausedQueue();
      final rows = await _appStateDb.getPendingDownloadQueueRows();
-      _persistedQueueJsonById
-        ..clear()
-        ..addEntries(
-          rows
-              .map(
-                (row) => MapEntry(
-                  row['id']?.toString() ?? '',
-                  row['item_json']?.toString() ?? '',
-                ),
-              )
-              .where((entry) => entry.key.isNotEmpty),
-        );
+      _persistedQueueItemById.clear();
+      _nonCanonicalPersistedQueueIds.clear();
      if (rows.isEmpty) {
        if (restorePaused) _persistUserPausedQueue(false);
        _log.d('No queue found in storage');
@@ -481,13 +521,31 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {

      final pendingItems = <DownloadItem>[];
      for (final row in rows) {
+        final rowId = row['id']?.toString() ?? '';
+        if (rowId.isEmpty) continue;
+        // Keep a null sentinel until the payload has been decoded. If the row
+        // is corrupt, the next flush still knows that its database ID must be
+        // deleted instead of silently leaving it behind forever.
+        _persistedQueueItemById[rowId] = null;
        final itemJson = row['item_json'] as String?;
        if (itemJson == null || itemJson.isEmpty) continue;

        try {
          final decoded = jsonDecode(itemJson);
          if (decoded is! Map) continue;
-          var item = DownloadItem.fromJson(Map<String, dynamic>.from(decoded));
+          final persistedItem = DownloadItem.fromJson(
+            Map<String, dynamic>.from(decoded),
+          );
+          _persistedQueueItemById[rowId] = persistedItem;
+          final canonicalStatus = downloadQueuePersistenceStatus(
+            persistedItem.status,
+          ).name;
+          if (itemJson !=
+                  encodeDownloadQueueItemForPersistence(persistedItem) ||
+              row['status']?.toString() != canonicalStatus) {
+            _nonCanonicalPersistedQueueIds.add(rowId);
+          }
+          var item = persistedItem;
          final normalizedService = _normalizeQueuedService(item.service);
          if (normalizedService != item.service) {
            item = item.copyWith(service: normalizedService);
@@ -509,7 +567,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
        if (restorePaused) _persistUserPausedQueue(false);
        _log.d('No pending items to restore');
        await _appStateDb.replacePendingDownloadQueueRows(const []);
-        _persistedQueueJsonById.clear();
+        _persistedQueueItemById.clear();
+        _nonCanonicalPersistedQueueIds.clear();
        return;
      }

@@ -552,42 +611,54 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
    try {
      // skipped (user-cancelled) rows persist too, so a cancelled download can
      // still be retried after an app restart instead of being re-searched.
-      final pendingItems = state.items
-          .where(
-            (item) =>
-                item.status == DownloadStatus.queued ||
-                item.status == DownloadStatus.downloading ||
-                item.status == DownloadStatus.finalizing ||
-                item.status == DownloadStatus.skipped,
-          )
-          .toList(growable: false);
      final nowIso = DateTime.now().toIso8601String();
-      final currentJsonById = <String, String>{};
+      final currentItemsById = <String, DownloadItem?>{};
      final upserts = <Map<String, dynamic>>[];
-      for (final item in pendingItems) {
-        final itemJson = jsonEncode(item.toJson());
-        currentJsonById[item.id] = itemJson;
-        if (_persistedQueueJsonById[item.id] == itemJson) continue;
+      for (final item in state.items) {
+        if (item.status != DownloadStatus.queued &&
+            item.status != DownloadStatus.downloading &&
+            item.status != DownloadStatus.finalizing &&
+            item.status != DownloadStatus.skipped) {
+          continue;
+        }
+        currentItemsById[item.id] = item;
+        final previous = _persistedQueueItemById[item.id];
+        final mustRewrite = _nonCanonicalPersistedQueueIds.contains(item.id);
+        if (!mustRewrite && identical(previous, item)) continue;
+
+        final itemJson = encodeDownloadQueueItemForPersistence(item);
+        if (!mustRewrite &&
+            previous != null &&
+            encodeDownloadQueueItemForPersistence(previous) == itemJson) {
+          continue;
+        }
        upserts.add({
          'id': item.id,
          'item_json': itemJson,
-          'status': item.status.name,
+          'status': downloadQueuePersistenceStatus(item.status).name,
          'created_at': item.createdAt.toIso8601String(),
          'updated_at': nowIso,
        });
      }
-      final deletedIds = _persistedQueueJsonById.keys
-          .where((id) => !currentJsonById.containsKey(id))
+      final deletedIds = _persistedQueueItemById.keys
+          .where((id) => !currentItemsById.containsKey(id))
          .toList(growable: false);
-      if (upserts.isEmpty && deletedIds.isEmpty) return;
+      if (upserts.isEmpty && deletedIds.isEmpty) {
+        _persistedQueueItemById
+          ..clear()
+          ..addAll(currentItemsById);
+        _nonCanonicalPersistedQueueIds.clear();
+        return;
+      }

      await _appStateDb.applyPendingDownloadQueueChanges(
        upserts: upserts,
        deletedIds: deletedIds,
      );
-      _persistedQueueJsonById
+      _persistedQueueItemById
        ..clear()
-        ..addAll(currentJsonById);
+        ..addAll(currentItemsById);
+      _nonCanonicalPersistedQueueIds.clear();
      _log.d(
        'Persisted ${upserts.length} changed and removed '
        '${deletedIds.length} queue items',
@@ -1452,8 +1523,19 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
  }

  Future<void> _processQueue() async {
-    if (state.isProcessing) return;
+    if (!_queueProcessingGate.tryEnter()) return;

+    try {
+      if (state.isProcessing) return;
+      await _processQueueSingleFlight();
+    } finally {
+      if (_queueProcessingGate.leave()) {
+        Future.microtask(_processQueue);
+      }
+    }
+  }
+
+  Future<void> _processQueueSingleFlight() async {
    if (Platform.isAndroid &&
        state.items.any((item) => item.status == DownloadStatus.queued) &&
        !canStartForegroundDownloadForLifecycle(
@@ -606,6 +606,7 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier {
    await _persistNativeWorkerRunId(runId);
    final reconciledIds = <String>{};
    Future<void>? preparationFuture;
+    var preparationStopRequested = false;
    try {
      await PlatformBridge.startNativeDownloadWorker(
        requests: [encodeRequest(firstItem, firstContext)],
@@ -627,39 +628,90 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier {

      preparationFuture = () async {
        var nextIndex = 1;
-        final preparationConcurrency = min(2, queuedItems.length - 1);
        try {
-          await Future.wait(
-            List.generate(preparationConcurrency, (_) async {
-              while (nextIndex < queuedItems.length) {
-                final index = nextIndex++;
-                final item = queuedItems[index];
-                try {
-                  final context = await _buildAndroidNativeWorkerRequest(
-                    item,
-                    settings,
-                  );
-                  if (context == null) {
-                    _log.w(
-                      'Native worker gate rejected ${item.track.name}; leaving it queued for the Dart worker',
+          while (nextIndex < queuedItems.length && !preparationStopRequested) {
+            while (contexts.length >=
+                    DownloadQueueNotifier._nativePreparationWindowSize &&
+                !preparationStopRequested) {
+              await Future<void>.delayed(const Duration(milliseconds: 250));
+            }
+            if (preparationStopRequested) break;
+
+            final capacity =
+                DownloadQueueNotifier._nativePreparationWindowSize -
+                contexts.length;
+            final batchLength = min(
+              DownloadQueueNotifier._nativePreparationBatchSize,
+              min(capacity, queuedItems.length - nextIndex),
+            );
+            if (batchLength <= 0) continue;
+
+            final batchItems = queuedItems.sublist(
+              nextIndex,
+              nextIndex + batchLength,
+            );
+            nextIndex += batchLength;
+            final prepared = List<_NativeWorkerRequestContext?>.filled(
+              batchItems.length,
+              null,
+            );
+            var preparationIndex = 0;
+            final preparationConcurrency = min(2, batchItems.length);
+            await Future.wait(
+              List.generate(preparationConcurrency, (_) async {
+                while (true) {
+                  final index = preparationIndex++;
+                  if (index >= batchItems.length) return;
+                  final item = batchItems[index];
+                  try {
+                    prepared[index] = await _buildAndroidNativeWorkerRequest(
+                      item,
+                      settings,
+                    );
+                  } catch (e, stack) {
+                    _log.e(
+                      'Could not prepare native request for ${item.track.name}: $e',
+                      e,
+                      stack,
                    );
-                    continue;
                  }
-                  contexts[item.id] = context;
-                  await PlatformBridge.appendNativeDownloadWorkerRequests(
-                    runId: runId,
-                    requests: [encodeRequest(item, context)],
-                  );
-                } catch (e, stack) {
-                  _log.e(
-                    'Could not prepare native request for ${item.track.name}: $e',
-                    e,
-                    stack,
-                  );
                }
+              }),
+            );
+
+            final appendRequests = <Map<String, dynamic>>[];
+            final appendedIds = <String>[];
+            for (var index = 0; index < batchItems.length; index++) {
+              final item = batchItems[index];
+              final context = prepared[index];
+              if (context == null) {
+                _log.w(
+                  'Native worker gate rejected ${item.track.name}; leaving it queued for the Dart worker',
+                );
+                continue;
              }
-            }),
-          );
+              contexts[item.id] = context;
+              appendedIds.add(item.id);
+              appendRequests.add(encodeRequest(item, context));
+            }
+            if (appendRequests.isEmpty) continue;
+
+            try {
+              await PlatformBridge.appendNativeDownloadWorkerRequests(
+                runId: runId,
+                requests: appendRequests,
+              );
+            } catch (e, stack) {
+              for (final id in appendedIds) {
+                contexts.remove(id);
+              }
+              _log.e(
+                'Could not append ${appendRequests.length} prepared native requests: $e',
+                e,
+                stack,
+              );
+            }
+          }
        } finally {
          try {
            await PlatformBridge.finishNativeDownloadWorkerPreparation(
@@ -696,12 +748,14 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier {
        );
        lastStateSerial = _snapshotStateSerial(snapshot, lastStateSerial);
        if (snapshot['is_running'] != true) {
+          preparationStopRequested = true;
          await _clearNativeWorkerRunId(runId);
          break;
        }
        await Future<void>.delayed(const Duration(seconds: 1));
      }
    } catch (e, stack) {
+      preparationStopRequested = true;
      if (isForegroundServiceStartNotAllowed(e)) {
        _log.w(
          'Android rejected the native worker start while backgrounded; keeping the queue pending',
@@ -752,6 +806,7 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier {
        _failedInSession++;
      }
    } finally {
+      preparationStopRequested = true;
      if (preparationFuture != null) {
        try {
          await preparationFuture;
@@ -1106,6 +1161,38 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier {
        }
      }
    }
+
+    final releasableIds = contexts.keys
+        .where(reconciledIds.contains)
+        .toList(growable: false);
+    if (releasableIds.isEmpty) return;
+
+    await flushQueuePersistence();
+    if (!workerRunning) {
+      for (final itemId in releasableIds) {
+        contexts.remove(itemId);
+        reconciledIds.remove(itemId);
+      }
+      return;
+    }
+
+    final runId = _snapshotRunId(snapshot);
+    if (runId.isEmpty) return;
+    try {
+      await PlatformBridge.acknowledgeNativeDownloadWorkerItems(
+        runId: runId,
+        itemIds: releasableIds,
+      );
+      for (final itemId in releasableIds) {
+        contexts.remove(itemId);
+        reconciledIds.remove(itemId);
+      }
+    } catch (e) {
+      // Keep contexts so the acknowledgement is retried on the next poll.
+      // This also keeps the preparation window bounded while native results
+      // are still retained by the service.
+      _log.w('Could not release reconciled native queue items: $e');
+    }
  }

  Future<void> _completeAndroidNativeWorkerItem(
@@ -1220,6 +1220,17 @@ class PlatformBridge {
    });
  }

+  static Future<void> acknowledgeNativeDownloadWorkerItems({
+    required String runId,
+    required List<String> itemIds,
+  }) async {
+    if (runId.isEmpty || itemIds.isEmpty) return;
+    await _channel.invokeMethod('acknowledgeNativeDownloadWorkerItems', {
+      'run_id': runId,
+      'item_ids_json': jsonEncode(itemIds),
+    });
+  }
+
  static Future<void> _deleteFileIfExists(String path) async {
    try {
      final file = File(path);
@@ -1,3 +1,4 @@
+import 'dart:convert';
 import 'dart:io';

 import 'package:flutter/material.dart';
@@ -688,6 +689,45 @@ void main() {
      expect(item.toJson()['errorType'], 'network');
      expect(item.toJson()['preparationStage'], isEmpty);
    });
+
+    test('persists restart-safe state without transient transfer progress', () {
+      final item = DownloadItem(
+        id: 'download-active',
+        track: sampleTrack(),
+        service: 'tidal',
+        createdAt: DateTime.utc(2026),
+        status: DownloadStatus.finalizing,
+        progress: 0.97,
+        speedMBps: 3.5,
+        bytesReceived: 970,
+        bytesTotal: 1000,
+        preparationStage: 'embedding_metadata',
+      );
+
+      final persisted =
+          jsonDecode(encodeDownloadQueueItemForPersistence(item))
+              as Map<String, dynamic>;
+
+      expect(persisted['id'], item.id);
+      expect(persisted['status'], DownloadStatus.queued.name);
+      expect(persisted['progress'], 0.0);
+      expect(persisted['speedMBps'], 0.0);
+      expect(persisted['bytesReceived'], 0);
+      expect(persisted['bytesTotal'], 0);
+      expect(persisted['preparationStage'], isEmpty);
+    });
+  });
+
+  group('Download queue processing gate', () {
+    test('allows only one asynchronous startup at a time', () {
+      final gate = QueueProcessingGate();
+
+      expect(gate.tryEnter(), isTrue);
+      expect(gate.tryEnter(), isFalse);
+      expect(gate.leave(), isTrue);
+      expect(gate.tryEnter(), isTrue);
+      expect(gate.leave(), isFalse);
+    });
  });

  group('Download queue lookup', () {