From 0e3fca9967cf5f4275eaeece8c6560efb67ab128 Mon Sep 17 00:00:00 2001 From: zarzet Date: Wed, 26 Aug 2026 18:37:56 +0700 Subject: [PATCH] perf(downloads): stream concurrent native queue work --- .../com/zarz/spotiflac/DownloadService.kt | 630 +++++++++++++++++- .../kotlin/com/zarz/spotiflac/MainActivity.kt | 22 + go_backend/exports_download.go | 1 + lib/providers/download_queue_provider.dart | 13 + ...download_queue_provider_native_worker.dart | 103 ++- lib/providers/extension_models.dart | 42 ++ lib/services/download_request_payload.dart | 4 + lib/services/platform_bridge.dart | 28 + test/models_and_utils_test.dart | 41 ++ 9 files changed, 840 insertions(+), 44 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 236f2c89..6066df70 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt @@ -24,11 +24,19 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay +import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.sync.withPermit import org.json.JSONArray import org.json.JSONObject import java.io.File +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicLong /** @@ -53,6 +61,9 @@ class DownloadService : Service() { const val ACTION_STOP = "com.zarz.spotiflac.action.STOP_DOWNLOAD" const val ACTION_UPDATE_PROGRESS = "com.zarz.spotiflac.action.UPDATE_PROGRESS" const val ACTION_START_NATIVE_QUEUE = "com.zarz.spotiflac.action.START_NATIVE_QUEUE" + 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_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" @@ -67,6 +78,7 @@ class DownloadService : Service() { const val EXTRA_SETTINGS_JSON = "settings_json" const val EXTRA_REQUESTS_PATH = "requests_path" const val EXTRA_SETTINGS_PATH = "settings_path" + const val EXTRA_RUN_ID = "run_id" 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" @@ -139,6 +151,23 @@ class DownloadService : Service() { } } + fun appendNativeQueueFromFile(context: Context, requestsPath: String, runId: String) { + val intent = Intent(context, DownloadService::class.java).apply { + action = ACTION_APPEND_NATIVE_QUEUE + putExtra(EXTRA_REQUESTS_PATH, requestsPath) + putExtra(EXTRA_RUN_ID, runId) + } + context.startService(intent) + } + + fun finishNativeQueuePreparation(context: Context, runId: String) { + val intent = Intent(context, DownloadService::class.java).apply { + action = ACTION_FINISH_NATIVE_QUEUE_PREPARATION + putExtra(EXTRA_RUN_ID, runId) + } + context.startService(intent) + } + fun pauseNativeQueue(context: Context) { val intent = Intent(context, DownloadService::class.java).apply { action = ACTION_PAUSE_NATIVE_QUEUE @@ -283,6 +312,8 @@ class DownloadService : Service() { internal val serviceScope = CoroutineScope(SupervisorJob() + Dispatchers.IO) internal var nativeWorkerJob: Job? = null + private var nativeWorkerRequestChannel: Channel? = null + @Volatile private var nativeWorkerPreparationComplete = true private var wakeLock: PowerManager.WakeLock? = null private var currentTrackName = "" private var currentArtistName = "" @@ -363,6 +394,23 @@ class DownloadService : Service() { ) startNativeWorker(requestsJson, settingsJson) } + ACTION_APPEND_NATIVE_QUEUE -> { + val requestsJson = readNativeQueuePayload( + intent, + EXTRA_REQUESTS_JSON, + EXTRA_REQUESTS_PATH, + "[]" + ) + appendNativeWorkerRequests( + requestsJson, + intent.getStringExtra(EXTRA_RUN_ID).orEmpty(), + ) + } + ACTION_FINISH_NATIVE_QUEUE_PREPARATION -> { + finishNativeWorkerPreparation( + intent.getStringExtra(EXTRA_RUN_ID).orEmpty(), + ) + } ACTION_PAUSE_NATIVE_QUEUE -> { nativeWorkerPaused = true cancelActiveNativeItemForPause() @@ -388,6 +436,8 @@ class DownloadService : Service() { ACTION_CANCEL_NATIVE_QUEUE -> { nativeWorkerCancelRequested = true nativeWorkerVerificationPaused = false + nativeWorkerPreparationComplete = true + nativeWorkerRequestChannel?.close() cancelNativeVerificationNotification() synchronized(nativeWorkerItems) { for (item in nativeWorkerItems) { @@ -477,6 +527,8 @@ class DownloadService : Service() { return } nativeWorkerCancelRequested = true + nativeWorkerPreparationComplete = true + nativeWorkerRequestChannel?.close() // Supersede the coroutine before cancelling it. Its catch/finally // blocks must not publish a skipped/finished state over the recovery // snapshot written below. @@ -602,6 +654,7 @@ class DownloadService : Service() { } } NativeDownloadFinalizer.cancelActiveWork() + nativeWorkerRequestChannel?.close() nativeWorkerGeneration++ val generation = nativeWorkerGeneration nativeWorkerJob?.cancel(CancellationException("Native queue replaced")) @@ -610,7 +663,19 @@ class DownloadService : Service() { nativeWorkerVerificationPaused = false nativeWorkerCancelRequested = false unregisterNativeWorkerNetworkCallback() - queueCount = requests.size + val workerSettings = try { + JSONObject(settingsJson) + } catch (_: Exception) { + JSONObject() + } + val streamingPreparation = workerSettings.optBoolean( + "preparation_streaming", + false, + ) + nativeWorkerPreparationComplete = !streamingPreparation + queueCount = workerSettings + .optInt("expected_total_items", requests.size) + .coerceAtLeast(requests.size) synchronized(nativeReplayGainEntries) { nativeReplayGainEntries.clear() } @@ -659,8 +724,16 @@ class DownloadService : Service() { includeItems = true ) + val requestChannel = Channel(Channel.UNLIMITED) + nativeWorkerRequestChannel = requestChannel + for (request in requests) { + requestChannel.trySend(request) + } + if (!streamingPreparation) { + requestChannel.close() + } nativeWorkerJob = serviceScope.launch { - runNativeWorker(requests, settingsJson, generation) + runNativeWorkerConcurrent(requestChannel, settingsJson, generation) } } @@ -672,6 +745,75 @@ class DownloadService : Service() { } } + private fun appendNativeWorkerRequests(requestsJson: String, runId: String) { + if (runId.isBlank() || runId != nativeWorkerRunId || nativeWorkerPreparationComplete) { + return + } + val channel = nativeWorkerRequestChannel ?: return + val requests = try { + parseNativeDownloadRequests(requestsJson) + } catch (e: Exception) { + android.util.Log.w( + "DownloadService", + "Ignoring invalid native queue append: ${e.message}", + ) + return + } + if (requests.isEmpty()) return + + val knownIds = synchronized(nativeWorkerItems) { + nativeWorkerItems.mapTo(mutableSetOf()) { it.itemId } + } + val additions = requests.filter { knownIds.add(it.itemId) } + if (additions.isEmpty()) return + + synchronized(nativeReplayGainRequestAlbumKeys) { + for (request in additions) { + try { + val key = NativeDownloadFinalizer.replayGainAlbumKey( + request.requestJson, + request.itemJson, + ) + if (key.isNotBlank()) { + nativeReplayGainRequestAlbumKeys[request.itemId] = key + } + } catch (_: Exception) { + } + } + } + synchronized(nativeWorkerItems) { + nativeWorkerItems.addAll( + additions.map { + NativeWorkerItem( + itemId = it.itemId, + trackName = it.trackName, + artistName = it.artistName, + itemJson = it.itemJson, + ) + }, + ) + queueCount = maxOf(queueCount, nativeWorkerItems.size) + } + for (request in additions) { + channel.trySend(request) + } + writeNativeReplayGainJournal() + writeNativeWorkerSnapshotAsync( + isRunning = nativeWorkerJob?.isActive == true, + isPaused = isNativeWorkerPaused(), + currentItemId = nativeWorkerCurrentItemId, + message = "Preparing queue", + includeItems = true, + ) + } + + private fun finishNativeWorkerPreparation(runId: String) { + if (runId.isBlank() || runId != nativeWorkerRunId) return + nativeWorkerPreparationComplete = true + nativeWorkerRequestChannel?.close() + writeNativeAlbumReplayGainIfComplete() + } + internal fun isNativeWorkerPaused(): Boolean = nativeWorkerPaused || nativeWorkerNetworkPaused || @@ -684,29 +826,7 @@ class DownloadService : Service() { } internal fun cancelActiveNativeItemForPause() { - var itemIdToCancel = "" - synchronized(nativeWorkerItems) { - val activeItem = nativeWorkerItems.firstOrNull { - it.status == "downloading" || it.status == "finalizing" - } ?: nativeWorkerItems.firstOrNull { - it.itemId == nativeWorkerCurrentItemId && it.status == "queued" - } - activeItem?.let { - it.status = "queued" - it.progress = 0.0 - it.bytesReceived = 0L - it.bytesTotal = 0L - itemIdToCancel = it.itemId - } - } - if (itemIdToCancel.isBlank()) itemIdToCancel = nativeWorkerCurrentItemId - if (itemIdToCancel.isNotBlank()) { - try { - Gobackend.cancelDownload(itemIdToCancel) - } catch (_: Exception) { - } - } - NativeDownloadFinalizer.cancelActiveWork() + cancelConcurrentNativeDownloadsForPause() } private fun parseNativeDownloadRequests(requestsJson: String): List { @@ -763,6 +883,462 @@ class DownloadService : Service() { } } + private fun nativeWorkerConcurrency(settingsJson: String): Int { + return try { + JSONObject(settingsJson).optInt("concurrent_downloads", 1).coerceIn(1, 3) + } catch (_: Exception) { + 1 + } + } + + private fun nativeRequestProviderKey(request: NativeDownloadRequest): String { + return try { + val payload = JSONObject(request.requestJson) + payload.optString("download_provider", "") + .ifBlank { payload.optString("service", "") } + .trim() + .lowercase() + .ifBlank { "default" } + } catch (_: Exception) { + "default" + } + } + + private fun nativeRequestProviderConcurrency(request: NativeDownloadRequest): Int { + return try { + JSONObject(request.requestJson) + .optInt("network_concurrency_limit", 3) + .coerceIn(1, 3) + } catch (_: Exception) { + 3 + } + } + + private fun cancelConcurrentNativeDownloadsForPause(excludeItemId: String = "") { + val ids = synchronized(nativeWorkerItems) { + nativeWorkerItems + .filter { + it.itemId != excludeItemId && + (it.status == "downloading" || + it.status == "finalizing") + } + .map { item -> + item.status = "queued" + item.progress = 0.0 + item.bytesReceived = 0L + item.bytesTotal = 0L + item.error = "" + item.itemId + } + } + for (itemId in ids) { + try { + Gobackend.cancelDownload(itemId) + } catch (_: Exception) { + } + } + if (ids.isNotEmpty()) { + NativeDownloadFinalizer.cancelActiveWork() + } + } + + private suspend fun processConcurrentNativeRequest( + request: NativeDownloadRequest, + settingsJson: String, + generation: Long, + networkSemaphore: Semaphore, + providerSemaphore: Semaphore, + finalizerMutex: Mutex, + rateLimitAttempts: ConcurrentHashMap, + ) { + while (!nativeWorkerCancelRequested && generation == nativeWorkerGeneration) { + while (isNativeWorkerPaused() && + !nativeWorkerCancelRequested && + generation == nativeWorkerGeneration + ) { + delay(500) + } + if (nativeWorkerCancelRequested || generation != nativeWorkerGeneration) return + + var progressJob: Job? = null + var progressInitialized = false + var retryCurrentRequest = false + try { + // Acquire the provider permit first. If several requests from + // one provider are queued, they must not occupy every global + // network slot while waiting for that provider's lower limit. + val response = providerSemaphore.withPermit { + networkSemaphore.withPermit { + if (isNativeWorkerPaused() || + nativeWorkerCancelRequested || + generation != nativeWorkerGeneration + ) { + throw CancellationException("Native queue paused") + } + nativeWorkerCurrentItemId = request.itemId + currentTrackName = request.trackName + currentArtistName = request.artistName + currentStatus = "preparing" + lastProgress = 0L + lastTotal = 0L + updateNotification(0L, 0L) + updateNativeWorkerItem(request.itemId) { + it.status = "preparing" + it.progress = 0.0 + it.bytesReceived = 0L + it.bytesTotal = 0L + it.error = "" + it.resultJson = null + } + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = false, + currentItemId = request.itemId, + message = "Preparing", + settingsJson = settingsJson, + includeItems = true, + ) + Gobackend.initItemProgress(request.itemId) + progressInitialized = true + progressJob = serviceScope.launch { + var lastSignature: String? = null + while (true) { + updateNativeWorkerItemProgress(request.itemId) + val signature = synchronized(nativeWorkerItems) { + nativeWorkerItems + .firstOrNull { it.itemId == request.itemId } + ?.let { + "${it.status}:${it.bytesReceived}:" + + "${it.bytesTotal}:${it.progress}" + } + } + if (signature != lastSignature) { + lastSignature = signature + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = false, + currentItemId = request.itemId, + message = "Downloading", + settingsJson = settingsJson, + ) + } + delay(1000) + } + } + currentStatus = "downloading" + updateNativeWorkerItem(request.itemId) { + it.status = "downloading" + } + try { + SafDownloadHandler.handle(this, request.requestJson) { json -> + Gobackend.downloadByStrategy(json) + } + } finally { + progressJob?.cancel() + progressJob = null + updateNativeWorkerItemProgress(request.itemId) + try { + Gobackend.clearItemProgress(request.itemId) + } catch (_: Exception) { + } + progressInitialized = false + } + } + } + if (generation != nativeWorkerGeneration) return + + var result = JSONObject(response) + if (result.optBoolean("success", false)) { + currentStatus = "finalizing" + updateNativeWorkerItem(request.itemId) { + it.status = "finalizing" + it.progress = 0.95 + it.error = "" + } + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = false, + currentItemId = request.itemId, + message = "Finalizing", + settingsJson = settingsJson, + ) + // Finalization is intentionally independent from the + // network semaphore: the next transfer can begin while + // metadata/FFmpeg/SAF work remains serialized here. + result = finalizerMutex.withLock { + NativeDownloadFinalizer.finalize( + this, + request.itemId, + request.requestJson, + request.itemJson, + result, + settingsJson, + ) { + nativeWorkerCancelRequested || + isNativeWorkerPaused() || + generation != nativeWorkerGeneration + } + } + } + + if (result.optBoolean("success", false)) { + result.optJSONObject("replaygain")?.let { replayGain -> + synchronized(nativeReplayGainEntries) { + nativeReplayGainEntries.add(JSONObject(replayGain.toString())) + } + } + updateNativeWorkerItem(request.itemId) { + it.status = "completed" + it.progress = 1.0 + it.error = "" + it.resultJson = result + } + writeNativeReplayGainJournal() + if (nativeWorkerPreparationComplete) { + writeNativeAlbumReplayGainIfComplete() + } + } else { + val errorType = result.optString("error_type") + val errorMessage = result.optString("error") + if (errorType == "cancelled" && + !isNativeWorkerPaused() && + !nativeWorkerCancelRequested && + generation == nativeWorkerGeneration + ) { + var waitedMs = 0L + while (waitedMs < 1500 && + !isNativeWorkerPaused() && + !nativeWorkerCancelRequested && + generation == nativeWorkerGeneration + ) { + delay(100) + waitedMs += 100 + } + } + + if (errorType == "cancelled" && + isNativeWorkerPaused() && + !nativeWorkerCancelRequested + ) { + updateNativeWorkerItem(request.itemId) { + it.status = "queued" + it.progress = 0.0 + it.bytesReceived = 0L + it.bytesTotal = 0L + it.error = "" + it.resultJson = null + } + retryCurrentRequest = true + } else if (NativeWorkerPolicy.shouldRetryRateLimit( + errorType = errorType, + errorMessage = errorMessage, + attempts = rateLimitAttempts[request.itemId] ?: 0, + ) + ) { + rateLimitAttempts.compute(request.itemId) { _, value -> + (value ?: 0) + 1 + } + val delaySeconds = NativeWorkerPolicy.rateLimitDelaySeconds( + retryAfterSeconds = result + .optInt("retry_after_seconds", 0) + .takeIf { it > 0 }, + errorMessage = errorMessage, + ) + currentStatus = "rate_limited" + updateNativeWorkerItem(request.itemId) { + it.status = "queued" + it.progress = 0.0 + it.bytesReceived = 0L + it.bytesTotal = 0L + it.error = "Rate limited, retrying in ${delaySeconds}s" + it.resultJson = null + } + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = isNativeWorkerPaused(), + currentItemId = request.itemId, + message = "Rate limited, retrying in ${delaySeconds}s", + settingsJson = settingsJson, + includeItems = true, + ) + delay(delaySeconds * 1000L) + retryCurrentRequest = true + } else if (NativeWorkerPolicy.isVerificationRequired( + errorType = errorType, + errorMessage = errorMessage, + ) + ) { + nativeWorkerVerificationPaused = true + currentStatus = "verification_required" + updateNativeWorkerItem(request.itemId) { + it.status = "failed" + it.error = errorMessage + it.resultJson = result + } + cancelConcurrentNativeDownloadsForPause(request.itemId) + writeNativeReplayGainJournal() + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = true, + currentItemId = request.itemId, + message = "Verification required", + lastResult = result, + settingsJson = settingsJson, + includeItems = true, + ) + showNativeVerificationRequired() + updateNotification(0L, 0L) + retryCurrentRequest = true + } else { + updateNativeWorkerItem(request.itemId) { + it.status = if (errorType == "cancelled") "skipped" else "failed" + it.error = errorMessage + it.resultJson = result + } + writeNativeReplayGainJournal() + } + } + + if (!retryCurrentRequest) { + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = false, + currentItemId = request.itemId, + message = if (result.optBoolean("success", false)) "Completed" else "Failed", + lastResult = result, + settingsJson = settingsJson, + includeItems = true, + ) + } + } catch (e: CancellationException) { + if (nativeWorkerCancelRequested && generation == nativeWorkerGeneration) { + updateNativeWorkerItem(request.itemId) { + it.status = "skipped" + it.error = "Cancelled" + } + throw e + } + if (isNativeWorkerPaused() && !nativeWorkerCancelRequested) { + updateNativeWorkerItem(request.itemId) { + it.status = "queued" + it.progress = 0.0 + it.bytesReceived = 0L + it.bytesTotal = 0L + it.error = "" + it.resultJson = null + } + retryCurrentRequest = true + } else { + throw e + } + } catch (e: Exception) { + updateNativeWorkerItem(request.itemId) { + it.status = "failed" + it.error = e.message ?: "Native download failed" + } + writeNativeReplayGainJournal() + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = false, + currentItemId = request.itemId, + message = e.message ?: "Native download failed", + settingsJson = settingsJson, + includeItems = true, + ) + } finally { + progressJob?.cancel() + if (progressInitialized) { + updateNativeWorkerItemProgress(request.itemId) + try { + Gobackend.clearItemProgress(request.itemId) + } catch (_: Exception) { + } + } + } + + if (!retryCurrentRequest) { + if (nativeWorkerCurrentItemId == request.itemId) { + nativeWorkerCurrentItemId = "" + } + return + } + } + } + + private suspend fun runNativeWorkerConcurrent( + requests: Channel, + settingsJson: String, + generation: Long, + ) { + val concurrency = nativeWorkerConcurrency(settingsJson) + val networkSemaphore = Semaphore(concurrency) + val finalizerMutex = Mutex() + val providerSemaphores = ConcurrentHashMap() + val rateLimitAttempts = ConcurrentHashMap() + + 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) + } + processConcurrentNativeRequest( + request = request, + settingsJson = settingsJson, + generation = generation, + networkSemaphore = networkSemaphore, + providerSemaphore = providerSemaphore, + finalizerMutex = finalizerMutex, + rateLimitAttempts = rateLimitAttempts, + ) + } + } + itemJobs.joinAll() + } + } finally { + if (generation == nativeWorkerGeneration) { + nativeWorkerRequestChannel = null + nativeWorkerPreparationComplete = true + if (!nativeWorkerCancelRequested) { + flushNativeAlbumReplayGainJournalIfComplete() + } + val counts = nativeWorkerCounts() + val shouldNotifyCompletion = NativeWorkerPolicy.shouldNotifyQueueComplete( + cancelRequested = nativeWorkerCancelRequested, + completed = counts.completed, + failed = counts.failed, + ) + currentStatus = "finalizing" + writeNativeWorkerSnapshot( + isRunning = false, + isPaused = false, + currentItemId = "", + message = if (nativeWorkerCancelRequested) "Cancelled" else "Finished", + settingsJson = settingsJson, + includeItems = true, + ) + stopForegroundService(cancelNativeWorker = false) + if (shouldNotifyCompletion) { + showNativeQueueComplete(counts) + } + } + } + } + private suspend fun runNativeWorker( requests: List, settingsJson: String, @@ -1135,6 +1711,8 @@ class DownloadService : Service() { private fun stopForegroundService(cancelNativeWorker: Boolean = true) { if (cancelNativeWorker) { nativeWorkerCancelRequested = true + nativeWorkerPreparationComplete = true + nativeWorkerRequestChannel?.close() NativeDownloadFinalizer.cancelActiveWork() nativeWorkerJob?.cancel(CancellationException("Download service stopped")) nativeWorkerPaused = false @@ -1363,6 +1941,8 @@ class DownloadService : Service() { override fun onDestroy() { unregisterNativeWorkerNetworkCallback() nativeWorkerCancelRequested = true + nativeWorkerPreparationComplete = true + nativeWorkerRequestChannel?.close() NativeDownloadFinalizer.cancelActiveWork() nativeWorkerJob?.cancel(CancellationException("Download service destroyed")) if (hasNativeWorkerState()) { 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 e97e119f..84c08627 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt @@ -1755,6 +1755,28 @@ class MainActivity: FlutterFragmentActivity() { } result.success(null) } + "appendNativeDownloadWorkerRequests" -> { + val requestsPath = call.argument("requests_path") ?: "" + val runId = call.argument("run_id") ?: "" + if (requestsPath.isNotBlank() && runId.isNotBlank()) { + DownloadService.appendNativeQueueFromFile( + this@MainActivity, + requestsPath, + runId, + ) + } + result.success(null) + } + "finishNativeDownloadWorkerPreparation" -> { + val runId = call.argument("run_id") ?: "" + if (runId.isNotBlank()) { + DownloadService.finishNativeQueuePreparation( + this@MainActivity, + runId, + ) + } + result.success(null) + } "pauseNativeDownloadWorker" -> { DownloadService.pauseNativeQueue(this@MainActivity) result.success(null) diff --git a/go_backend/exports_download.go b/go_backend/exports_download.go index abe7a56b..d6442257 100644 --- a/go_backend/exports_download.go +++ b/go_backend/exports_download.go @@ -58,6 +58,7 @@ type DownloadRequest struct { AllowQualityVariant bool `json:"allow_quality_variant,omitempty"` QualityVariant string `json:"quality_variant,omitempty"` SongLinkRegion string `json:"songlink_region,omitempty"` + NetworkConcurrencyLimit int `json:"network_concurrency_limit,omitempty"` } type DownloadResponse struct { diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index d4b0ede2..985b0fe1 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -660,6 +660,14 @@ class DownloadQueueNotifier extends Notifier { final postProcessingEnabled = settings.useExtensionProviders && extensionState.extensions.any((e) => e.enabled && e.hasPostProcessing); + final selectedDownloadExtension = extensionState.extensions + .where( + (extension) => + extension.enabled && + extension.hasDownloadProvider && + extension.id.toLowerCase() == item.service.toLowerCase(), + ) + .firstOrNull; final normalizedTrackNumber = (track.trackNumber != null && track.trackNumber! > 0) ? track.trackNumber! @@ -752,6 +760,11 @@ class DownloadQueueNotifier extends Notifier { : '', qualityVariantCollisionOnly: qualityVariantCollisionOnly, songLinkRegion: settings.songLinkRegion, + networkConcurrencyLimit: + selectedDownloadExtension + ?.downloadTransferPolicy + .maxConcurrentDownloads ?? + 3, ); } diff --git a/lib/providers/download_queue_provider_native_worker.dart b/lib/providers/download_queue_provider_native_worker.dart index 956457ef..ef0a738b 100644 --- a/lib/providers/download_queue_provider_native_worker.dart +++ b/lib/providers/download_queue_provider_native_worker.dart @@ -181,12 +181,6 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { if (!Platform.isAndroid || !settings.nativeDownloadWorkerEnabled) { return false; } - if (settings.concurrentDownloads > 1) { - // The native worker downloads strictly sequentially, so - // prefer the Dart queue when the user enabled concurrent downloads. - _log.i('Concurrent downloads enabled; skipping native worker'); - return false; - } if (!settings.useExtensionProviders) { return false; } @@ -563,26 +557,36 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { } final contexts = {}; - final requests = >[]; - for (final item in queuedItems) { - final context = await _buildAndroidNativeWorkerRequest(item, settings); - if (context == null) { - _log.w( - 'Native worker gate rejected ${item.track.name}; falling back to Dart queue', - ); - return false; - } - contexts[item.id] = context; - requests.add({ + Map encodeRequest( + DownloadItem item, + _NativeWorkerRequestContext context, + ) { + return { 'contract_version': DownloadRequestPayload.nativeWorkerContractVersion, 'item_id': item.id, 'track_name': item.track.name, 'artist_name': item.track.artistName, 'item_json': jsonEncode(item.toJson()), 'request_json': context.requestJson, - }); + }; } + // Only the first request blocks startup. The rest are prepared with a + // small metadata pipeline and appended to the already-running native + // worker, reducing time-to-first-byte for large albums/playlists. + final firstItem = queuedItems.first; + final firstContext = await _buildAndroidNativeWorkerRequest( + firstItem, + settings, + ); + if (firstContext == null) { + _log.w( + 'Native worker gate rejected ${firstItem.track.name}; falling back to Dart queue', + ); + return false; + } + contexts[firstItem.id] = firstContext; + if (!canStartForegroundDownloadForLifecycle( WidgetsBinding.instance.lifecycleState, )) { @@ -601,9 +605,10 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { final runId = _newNativeWorkerRunId(); await _persistNativeWorkerRunId(runId); final reconciledIds = {}; + Future? preparationFuture; try { await PlatformBridge.startNativeDownloadWorker( - requests: requests, + requests: [encodeRequest(firstItem, firstContext)], settings: { 'worker': 'android_native', 'version': 1, @@ -613,9 +618,62 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { 'created_at': DateTime.now().toIso8601String(), 'save_download_history': settings.saveDownloadHistory, 'download_network_mode': settings.downloadNetworkMode, + 'concurrent_downloads': settings.concurrentDownloads.clamp(1, 3), + 'finalizer_concurrency': 1, + 'preparation_streaming': true, + 'expected_total_items': queuedItems.length, }, ); + 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', + ); + 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, + ); + } + } + }), + ); + } finally { + try { + await PlatformBridge.finishNativeDownloadWorkerPreparation( + runId: runId, + ); + } catch (_) { + // Do not leave the foreground worker waiting forever on an open + // preparation channel if the final hand-off fails. + await PlatformBridge.cancelNativeDownloadWorker(); + rethrow; + } + } + }(); + final runStartWait = Stopwatch()..start(); var lastStateSerial = 0; while (true) { @@ -694,6 +752,13 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { _failedInSession++; } } finally { + if (preparationFuture != null) { + try { + await preparationFuture; + } catch (e) { + _log.w('Native worker preparation pipeline stopped: $e'); + } + } state = state.copyWith(isProcessing: false, currentDownload: null); _stopConnectivityMonitoring(); try { diff --git a/lib/providers/extension_models.dart b/lib/providers/extension_models.dart index 55d54efd..b993b1d8 100644 --- a/lib/providers/extension_models.dart +++ b/lib/providers/extension_models.dart @@ -50,6 +50,46 @@ List? _tryDecodeStringListPreference(String rawJson, String key) { } } +class ExtensionDownloadTransferPolicy { + final int maxAttempts; + final String resumePolicy; + final bool persistentCheckpoint; + final int maxParallelSegments; + final int maxConcurrentDownloads; + + const ExtensionDownloadTransferPolicy({ + this.maxAttempts = 3, + this.resumePolicy = 'none', + this.persistentCheckpoint = false, + this.maxParallelSegments = 3, + this.maxConcurrentDownloads = 3, + }); + + factory ExtensionDownloadTransferPolicy.fromCapabilities( + Map capabilities, + ) { + final raw = capabilities['downloadTransfer']; + if (raw is! Map) return const ExtensionDownloadTransferPolicy(); + final values = Map.from(raw); + int boundedInt(String key, int fallback, int min, int max) { + final value = values[key]; + final parsed = value is num ? value.round() : fallback; + return parsed.clamp(min, max).toInt(); + } + + final requestedResume = values['resumePolicy']?.toString().trim(); + final resumePolicy = requestedResume == 'validated' ? 'validated' : 'none'; + return ExtensionDownloadTransferPolicy( + maxAttempts: boundedInt('maxAttempts', 3, 1, 8), + resumePolicy: resumePolicy, + persistentCheckpoint: + resumePolicy == 'validated' && values['persistentCheckpoint'] == true, + maxParallelSegments: boundedInt('maxParallelSegments', 3, 1, 8), + maxConcurrentDownloads: boundedInt('maxConcurrentDownloads', 3, 1, 3), + ); + } +} + /// First enabled custom-search extension, preferring ones marked primary. Extension? defaultSearchExtension(List extensions) { return extensions @@ -239,6 +279,8 @@ class Extension { bool get hasPostProcessing => postProcessing?.enabled ?? false; bool get hasServiceHealth => serviceHealth.isNotEmpty; bool get hasHomeFeed => capabilities['homeFeed'] == true; + ExtensionDownloadTransferPolicy get downloadTransferPolicy => + ExtensionDownloadTransferPolicy.fromCapabilities(capabilities); bool get requiresNativeContainerConversion => capabilities['requiresContainerConversion'] == true || capabilities['requiresNativeContainerConversion'] == true; diff --git a/lib/services/download_request_payload.dart b/lib/services/download_request_payload.dart index 32318138..c5ce5fc2 100644 --- a/lib/services/download_request_payload.dart +++ b/lib/services/download_request_payload.dart @@ -60,6 +60,7 @@ class DownloadRequestPayload { final String qualityVariant; final bool qualityVariantCollisionOnly; final String songLinkRegion; + final int networkConcurrencyLimit; const DownloadRequestPayload({ this.contractVersion = nativeWorkerContractVersion, @@ -121,6 +122,7 @@ class DownloadRequestPayload { this.qualityVariant = '', this.qualityVariantCollisionOnly = false, this.songLinkRegion = 'US', + this.networkConcurrencyLimit = 3, }); Map toJson() { @@ -184,6 +186,7 @@ class DownloadRequestPayload { 'quality_variant': qualityVariant, 'quality_variant_collision_only': qualityVariantCollisionOnly, 'songlink_region': songLinkRegion, + 'network_concurrency_limit': networkConcurrencyLimit, }; } @@ -251,6 +254,7 @@ class DownloadRequestPayload { qualityVariant: qualityVariant, qualityVariantCollisionOnly: qualityVariantCollisionOnly, songLinkRegion: songLinkRegion, + networkConcurrencyLimit: networkConcurrencyLimit, ); } } diff --git a/lib/services/platform_bridge.dart b/lib/services/platform_bridge.dart index 183688a0..7dd3bc56 100644 --- a/lib/services/platform_bridge.dart +++ b/lib/services/platform_bridge.dart @@ -1192,6 +1192,34 @@ class PlatformBridge { } } + static Future appendNativeDownloadWorkerRequests({ + required String runId, + required List> requests, + }) async { + if (requests.isEmpty) return; + final payloadDir = await _nativeWorkerPayloadDir(); + final stamp = DateTime.now().microsecondsSinceEpoch; + final requestPath = '${payloadDir.path}/append_$stamp.json'; + await File(requestPath).writeAsString(jsonEncode(requests), flush: true); + try { + await _channel.invokeMethod('appendNativeDownloadWorkerRequests', { + 'run_id': runId, + 'requests_path': requestPath, + }); + } catch (_) { + unawaited(_deleteFileIfExists(requestPath)); + rethrow; + } + } + + static Future finishNativeDownloadWorkerPreparation({ + required String runId, + }) async { + await _channel.invokeMethod('finishNativeDownloadWorkerPreparation', { + 'run_id': runId, + }); + } + 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 b8750486..1845b855 100644 --- a/test/models_and_utils_test.dart +++ b/test/models_and_utils_test.dart @@ -8,6 +8,7 @@ import 'package:spotiflac_android/models/theme_settings.dart'; import 'package:spotiflac_android/models/track.dart'; import 'package:spotiflac_android/providers/library_collections_provider.dart'; import 'package:spotiflac_android/providers/download_queue_provider.dart'; +import 'package:spotiflac_android/providers/extension_provider.dart'; import 'package:spotiflac_android/services/app_remote_config_service.dart'; import 'package:spotiflac_android/services/download_request_payload.dart'; import 'package:spotiflac_android/services/history_database.dart'; @@ -927,6 +928,7 @@ void main() { qualityVariant: 'qv_12345678', qualityVariantCollisionOnly: true, songLinkRegion: 'ID', + networkConcurrencyLimit: 2, ); expect(payload.toJson(), { @@ -989,6 +991,7 @@ void main() { 'quality_variant': 'qv_12345678', 'quality_variant_collision_only': true, 'songlink_region': 'ID', + 'network_concurrency_limit': 2, }); }); @@ -1026,6 +1029,44 @@ void main() { }); }); + group('extension download transfer policy', () { + test('parses and bounds the generic manifest capability', () { + final extension = Extension.fromJson({ + 'id': 'provider', + 'name': 'provider', + 'capabilities': { + 'downloadTransfer': { + 'maxAttempts': 99, + 'resumePolicy': 'validated', + 'persistentCheckpoint': true, + 'maxParallelSegments': 99, + 'maxConcurrentDownloads': 99, + }, + }, + }); + + final policy = extension.downloadTransferPolicy; + expect(policy.maxAttempts, 8); + expect(policy.resumePolicy, 'validated'); + expect(policy.persistentCheckpoint, isTrue); + expect(policy.maxParallelSegments, 8); + expect(policy.maxConcurrentDownloads, 3); + }); + + test('disables checkpoints unless validated resume is selected', () { + final extension = Extension.fromJson({ + 'id': 'provider', + 'name': 'provider', + 'capabilities': { + 'downloadTransfer': {'persistentCheckpoint': true}, + }, + }); + + expect(extension.downloadTransferPolicy.resumePolicy, 'none'); + expect(extension.downloadTransferPolicy.persistentCheckpoint, isFalse); + }); + }); + group('artist utils', () { test('splits common artist separators and removes duplicates for tags', () { expect(splitArtistNames(' A, B & C feat. D x E with F '), [