From 04c42bd5119910413fb88dbef08e3163135d83d6 Mon Sep 17 00:00:00 2001 From: zarzet Date: Thu, 27 Aug 2026 00:17:25 +0700 Subject: [PATCH] perf(downloads): bound native queue state --- .../com/zarz/spotiflac/DownloadService.kt | 146 +++++++++++++---- .../spotiflac/DownloadServiceReplayGain.kt | 18 +- .../zarz/spotiflac/DownloadServiceSnapshot.kt | 54 +++++- .../kotlin/com/zarz/spotiflac/MainActivity.kt | 12 ++ .../com/zarz/spotiflac/NativeWorkerPolicy.kt | 3 + .../zarz/spotiflac/NativeWorkerPolicyTest.kt | 10 ++ lib/providers/download_queue_provider.dart | 154 ++++++++++++++---- ...download_queue_provider_native_worker.dart | 143 ++++++++++++---- lib/services/platform_bridge.dart | 11 ++ test/models_and_utils_test.dart | 40 +++++ 10 files changed, 483 insertions(+), 108 deletions(-) diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt index 6066df70..419eb7aa 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt @@ -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? = 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() + internal val nativeWorkerTerminalStatuses = mutableMapOf() internal val nativeReplayGainEntries = mutableListOf() internal val nativeReplayGainRequestAlbumKeys = mutableMapOf() 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().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() - 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 diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceReplayGain.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceReplayGain.kt index 947a5349..98e69bd1 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceReplayGain.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceReplayGain.kt @@ -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 { + 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 diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt index 8921ec11..bf8add5b 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt @@ -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 } - diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt index 84c08627..3544579c 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt @@ -1777,6 +1777,18 @@ class MainActivity: FlutterFragmentActivity() { } result.success(null) } + "acknowledgeNativeDownloadWorkerItems" -> { + val runId = call.argument("run_id") ?: "" + val itemIdsJson = call.argument("item_ids_json") ?: "[]" + if (runId.isNotBlank()) { + DownloadService.acknowledgeNativeQueueItems( + this@MainActivity, + runId, + itemIdsJson, + ) + } + result.success(null) + } "pauseNativeDownloadWorker" -> { DownloadService.pauseNativeQueue(this@MainActivity) result.success(null) diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeWorkerPolicy.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeWorkerPolicy.kt index ccedbb3f..e95f0937 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeWorkerPolicy.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeWorkerPolicy.kt @@ -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 diff --git a/android/app/src/test/kotlin/com/zarz/spotiflac/NativeWorkerPolicyTest.kt b/android/app/src/test/kotlin/com/zarz/spotiflac/NativeWorkerPolicyTest.kt index 969e4ca4..a0470a12 100644 --- a/android/app/src/test/kotlin/com/zarz/spotiflac/NativeWorkerPolicyTest.kt +++ b/android/app/src/test/kotlin/com/zarz/spotiflac/NativeWorkerPolicyTest.kt @@ -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)) + } + } } diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index 985b0fe1..0768c35d 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -53,6 +53,52 @@ part 'download_queue_provider_single_item.dart'; final _log = AppLogger('DownloadQueue'); +typedef _PersistedQueueItemCache = Map; + +/// 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 { Timer? _queuePersistDebounce; Future _queuePersistenceWrite = Future.value(); Future _queuePausePersistenceWrite = Future.value(); - final Map _persistedQueueJsonById = {}; + final _PersistedQueueItemCache _persistedQueueItemById = {}; + final Set _nonCanonicalPersistedQueueIds = {}; + final QueueProcessingGate _queueProcessingGate = QueueProcessingGate(); StreamSubscription>? _connectivitySub; int _downloadCount = 0; static const _cleanupInterval = 50; @@ -273,6 +321,8 @@ class DownloadQueueNotifier extends Notifier { 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 { 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 { final pendingItems = []; 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.from(decoded)); + final persistedItem = DownloadItem.fromJson( + Map.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 { 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 { 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 = {}; + final currentItemsById = {}; final upserts = >[]; - 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 { } Future _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 _processQueueSingleFlight() async { if (Platform.isAndroid && state.items.any((item) => item.status == DownloadStatus.queued) && !canStartForegroundDownloadForLifecycle( diff --git a/lib/providers/download_queue_provider_native_worker.dart b/lib/providers/download_queue_provider_native_worker.dart index ef0a738b..4c088af6 100644 --- a/lib/providers/download_queue_provider_native_worker.dart +++ b/lib/providers/download_queue_provider_native_worker.dart @@ -606,6 +606,7 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { await _persistNativeWorkerRunId(runId); final reconciledIds = {}; Future? 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.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 = >[]; + final appendedIds = []; + 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.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 _completeAndroidNativeWorkerItem( diff --git a/lib/services/platform_bridge.dart b/lib/services/platform_bridge.dart index 7dd3bc56..5a7887f3 100644 --- a/lib/services/platform_bridge.dart +++ b/lib/services/platform_bridge.dart @@ -1220,6 +1220,17 @@ class PlatformBridge { }); } + static Future acknowledgeNativeDownloadWorkerItems({ + required String runId, + required List itemIds, + }) async { + if (runId.isEmpty || itemIds.isEmpty) return; + await _channel.invokeMethod('acknowledgeNativeDownloadWorkerItems', { + 'run_id': runId, + 'item_ids_json': jsonEncode(itemIds), + }); + } + static Future _deleteFileIfExists(String path) async { try { final file = File(path); diff --git a/test/models_and_utils_test.dart b/test/models_and_utils_test.dart index 1845b855..e18f2e92 100644 --- a/test/models_and_utils_test.dart +++ b/test/models_and_utils_test.dart @@ -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; + + 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', () {