perf: harden download and persistence lifecycle

2026-09-02 16:20:57 +02:00 · 2026-08-28 12:51:03 +07:00
parent 51f61c9c95
commit 45d9bbd7b2
22 changed files with 1656 additions and 231 deletions
@@ -15,6 +15,7 @@ import android.net.NetworkRequest
 import android.os.Build
 import android.os.IBinder
 import android.os.PowerManager
+import android.os.SystemClock
 import android.util.AtomicFile
 import androidx.core.app.NotificationCompat
 import gobackend.Gobackend
@@ -56,6 +57,7 @@ class DownloadService : Service() {
        private const val VERIFICATION_REQUIRED_NOTIFICATION_ID = 4
        private const val WAKELOCK_TAG = "SpotiFLAC:DownloadWakeLock"
        private const val WAKELOCK_RENEW_MS = 30 * 60 * 1000L
+        private const val WAKELOCK_RENEW_INTERVAL_MS = 15 * 60 * 1000L
        
        const val ACTION_START = "com.zarz.spotiflac.action.START_DOWNLOAD"
        const val ACTION_STOP = "com.zarz.spotiflac.action.STOP_DOWNLOAD"
@@ -294,6 +296,9 @@ class DownloadService : Service() {
            if (progress.has("item_delta")) {
                state.put("item_delta", progress.get("item_delta"))
            }
+            if (progress.has("item_deltas")) {
+                state.put("item_deltas", progress.get("item_deltas"))
+            }
            state.put("snapshot_mode", "compact_with_delta")
        }
    }
@@ -331,6 +336,7 @@ class DownloadService : Service() {
    private var nativeWorkerRequestChannel: Channel<NativeDownloadRequest>? = null
    @Volatile private var nativeWorkerPreparationComplete = true
    private var wakeLock: PowerManager.WakeLock? = null
+    @Volatile private var wakeLockLastRenewedAt = 0L
    private var currentTrackName = ""
    private var currentArtistName = ""
    internal var currentStatus = "preparing"
@@ -338,6 +344,10 @@ class DownloadService : Service() {
    // 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 = ""
+    // NotificationManager updates are also deduplicated by the visible
+    // notification state. Progress writers can report many byte-level changes
+    // that render the same percentage/text.
+    private var lastNotificationSignature: String? = null
    internal var lastProgress = 0L
    internal var lastTotal = 0L
    internal var nativeWorkerRunId = ""
@@ -346,6 +356,14 @@ class DownloadService : Service() {
    internal val nativeWorkerTerminalStatuses = mutableMapOf<String, String>()
    internal val nativeReplayGainEntries = mutableListOf<JSONObject>()
    internal val nativeReplayGainRequestAlbumKeys = mutableMapOf<String, String>()
+    // One coordinator polls Go's delta progress stream for all native workers.
+    // Individual workers consume this cache instead of parsing the complete
+    // multi-progress JSON independently.
+    internal val nativeWorkerProgressLock = Any()
+    internal val nativeWorkerProgressItems = mutableMapOf<String, NativeBackendProgress>()
+    internal var nativeWorkerProgressSeq = 0L
+    internal val nativeWorkerProgressEpoch = AtomicLong(0L)
+    @Volatile internal var nativeWorkerProgressJob: Job? = null
    internal val snapshotWriteLock = Any()
    internal val snapshotWriteSerial = AtomicLong(0L)
    internal var latestCommittedStateSnapshotSerial = 0L
@@ -625,6 +643,7 @@ class DownloadService : Service() {
    
    private fun startForegroundService() {
        isRunning = true
+        lastNotificationSignature = null

        ensureWakeLock()

@@ -1033,7 +1052,6 @@ class DownloadService : Service() {
            }
            if (nativeWorkerCancelRequested || generation != nativeWorkerGeneration) return

-            var progressJob: Job? = null
            var progressInitialized = false
            var retryCurrentRequest = false
            try {
@@ -1073,31 +1091,6 @@ class DownloadService : Service() {
                        )
                        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"
@@ -1107,8 +1100,6 @@ class DownloadService : Service() {
                                Gobackend.downloadByStrategy(json)
                            }
                        } finally {
-                            progressJob?.cancel()
-                            progressJob = null
                            updateNativeWorkerItemProgress(request.itemId)
                            try {
                                Gobackend.clearItemProgress(request.itemId)
@@ -1319,7 +1310,6 @@ class DownloadService : Service() {
                    includeItems = true,
                )
            } finally {
-                progressJob?.cancel()
                if (progressInitialized) {
                    updateNativeWorkerItemProgress(request.itemId)
                    try {
@@ -1348,6 +1338,7 @@ class DownloadService : Service() {
        val finalizerMutex = Mutex()
        val providerSemaphores = ConcurrentHashMap<String, Semaphore>()
        val rateLimitAttempts = ConcurrentHashMap<String, Int>()
+        val progressCoordinatorJob = startNativeWorkerProgressCoordinator(generation)

        try {
            supervisorScope {
@@ -1384,6 +1375,7 @@ class DownloadService : Service() {
                workers.joinAll()
            }
        } finally {
+            stopNativeWorkerProgressCoordinator(progressCoordinatorJob)
            if (generation == nativeWorkerGeneration) {
                cancelScheduledNativeWorkerItemsSnapshot()
                nativeWorkerRequestChannel = null
@@ -1420,6 +1412,7 @@ class DownloadService : Service() {
        generation: Long
    ) {
        val rateLimitAttempts = mutableMapOf<String, Int>()
+        val progressCoordinatorJob = startNativeWorkerProgressCoordinator(generation)
        try {
            var requestIndex = 0
            while (requestIndex < requests.size) {
@@ -1467,41 +1460,11 @@ class DownloadService : Service() {
                    includeItems = true
                )

-                var progressJob: Job? = null
                try {
                    Gobackend.initItemProgress(request.itemId)
-                    progressJob = serviceScope.launch {
-                        // The snapshot write is an AtomicFile open+fsync+
-                        // rename; skip ticks where progress hasn't moved.
-                        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)
-                        }
-                    }
                    val response = SafDownloadHandler.handle(this, request.requestJson) { json ->
                        Gobackend.downloadByStrategy(json)
                    }
-                    progressJob.cancel()
-                    progressJob = null
                    if (generation != nativeWorkerGeneration) {
                        // Superseded while blocked in the download call; the
                        // new run owns the shared state now.
@@ -1706,7 +1669,6 @@ class DownloadService : Service() {
                        includeItems = true
                    )
                } finally {
-                    progressJob?.cancel()
                    updateNativeWorkerItemProgress(request.itemId)
                    try {
                        Gobackend.clearItemProgress(request.itemId)
@@ -1721,6 +1683,7 @@ class DownloadService : Service() {
                }
            }
        } finally {
+            stopNativeWorkerProgressCoordinator(progressCoordinatorJob)
            if (generation == nativeWorkerGeneration) {
                cancelScheduledNativeWorkerItemsSnapshot()
                if (!nativeWorkerCancelRequested) {
@@ -1750,10 +1713,15 @@ class DownloadService : Service() {
        }
    }

+    @Synchronized
    private fun ensureWakeLock() {
        val existingWakeLock = wakeLock
        if (existingWakeLock?.isHeld == true) {
-            existingWakeLock.acquire(WAKELOCK_RENEW_MS)
+            val now = SystemClock.elapsedRealtime()
+            if (now - wakeLockLastRenewedAt >= WAKELOCK_RENEW_INTERVAL_MS) {
+                existingWakeLock.acquire(WAKELOCK_RENEW_MS)
+                wakeLockLastRenewedAt = now
+            }
            return
        }
        if (existingWakeLock != null) {
@@ -1768,12 +1736,14 @@ class DownloadService : Service() {
            setReferenceCounted(false)
            acquire(WAKELOCK_RENEW_MS)
        }
+        wakeLockLastRenewedAt = SystemClock.elapsedRealtime()
    }

    @Synchronized
    private fun releaseWakeLock() {
        val existingWakeLock = wakeLock
        wakeLock = null
+        wakeLockLastRenewedAt = 0L
        if (existingWakeLock?.isHeld == true) {
            try {
                existingWakeLock.release()
@@ -1786,6 +1756,7 @@ class DownloadService : Service() {
    @Synchronized
    private fun stopForegroundService(cancelNativeWorker: Boolean = true) {
        cancelScheduledNativeWorkerItemsSnapshot()
+        cancelNativeWorkerProgressCoordinator()
        if (cancelNativeWorker) {
            nativeWorkerCancelRequested = true
            nativeWorkerPreparationComplete = true
@@ -1811,6 +1782,7 @@ class DownloadService : Service() {
        nativeWorkerNetworkPaused = false
        nativeWorkerJob = null
        isRunning = false
+        lastNotificationSignature = null
        widgetSignature = ""
        try {
            DownloadQueueWidgetProvider.push(this, running = false)
@@ -1828,17 +1800,49 @@ class DownloadService : Service() {
            return nativeWorkerItems.isNotEmpty()
        }
    }
+
+    internal fun isNativeWorkerProgressActive(generation: Long): Boolean =
+        generation == nativeWorkerGeneration &&
+            !nativeWorkerCancelRequested &&
+            isRunning
+
+    internal fun nativeWorkerCurrentItemIdSnapshot(): String = nativeWorkerCurrentItemId
    
+    @Synchronized
    internal fun updateNotification(progress: Long, total: Long) {
        if (!isRunning) return
+        // Keep the foreground-service wake lock alive even when the visible
+        // notification is unchanged and therefore deduplicated below.
        ensureWakeLock()

+        val visibleProgress = when {
+            total <= 0L -> "indeterminate"
+            total == NOTIFICATION_PERCENT_TOTAL ->
+                "percent:${(progress * 100 / total).toInt()}"
+            else -> {
+                // buildNotification renders one decimal place for MB and an
+                // integer percentage. Suppress byte-level updates that would
+                // produce the same visible notification.
+                val progressTenthsMb = progress / (1024L * 1024L / 10L)
+                val totalTenthsMb = total / (1024L * 1024L / 10L)
+                val percent = (progress * 100 / total).toInt()
+                "mb:${progressTenthsMb}:${totalTenthsMb}:$percent"
+            }
+        }
+        val signature = "$currentTrackName|$currentArtistName|$currentStatus|$queueCount|$visibleProgress"
+        if (signature == lastNotificationSignature) return
+        lastNotificationSignature = signature
+
        val notification = buildNotification(progress, total)
        val manager = getSystemService(NotificationManager::class.java)
        manager.notify(NOTIFICATION_ID, notification)
        pushWidgetState(progress, total)
    }

+    internal fun maintainNativeWorkerWakeLock() {
+        if (isRunning) ensureWakeLock()
+    }
+
    private fun pushWidgetState(progress: Long, total: Long) {
        val percent = if (total > 0) {
            ((progress * 100) / total).toInt().coerceIn(0, 100)
@@ -25,6 +25,7 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import org.json.JSONArray
 import org.json.JSONObject
@@ -33,6 +34,18 @@ import java.util.concurrent.atomic.AtomicLong

 // Native-worker item state snapshots for the Flutter side.

+/**
+ * The compact subset of Go progress needed by the Android worker UI. Keeping
+ * this as a Kotlin value avoids sharing mutable JSONObject instances between
+ * the polling coroutine and worker coroutines.
+ */
+internal data class NativeBackendProgress(
+    val status: String,
+    val bytesReceived: Long,
+    val bytesTotal: Long,
+    val progress: Double,
+)
+
 internal fun DownloadService.writeNativeWorkerSnapshot(
    isRunning: Boolean,
    isPaused: Boolean,
@@ -41,10 +54,20 @@ internal fun DownloadService.writeNativeWorkerSnapshot(
    lastResult: JSONObject? = null,
    settingsJson: String = "",
    includeItems: Boolean = false,
+    progressItemIds: Collection<String>? = null,
+    progressCoordinatorEpoch: Long? = null,
    snapshotSerial: Long = snapshotWriteSerial.incrementAndGet()
 ) {
    try {
        synchronized(snapshotWriteLock) {
+            // A stopped/superseded progress coordinator must not publish a
+            // newer running snapshot after the queue's terminal snapshot.
+            if (
+                progressCoordinatorEpoch != null &&
+                nativeWorkerProgressEpoch.get() != progressCoordinatorEpoch
+            ) {
+                return
+            }
            if (includeItems) {
                if (snapshotSerial < latestCommittedStateSnapshotSerial) return
            } else {
@@ -67,13 +90,23 @@ internal fun DownloadService.writeNativeWorkerSnapshot(
                .put("snapshot_serial", snapshotSerial)
                .put("state_serial", if (includeItems) snapshotSerial else latestCommittedStateSnapshotSerial)
                .put("snapshot_mode", if (includeItems) "compact_items" else "delta")
+            if (includeItems) {
+                // The queue index is structural state. Progress deltas carry
+                // only the changed item; repeating every ID here made each
+                // tick grow linearly with a large queue.
+                snapshot.put("item_ids", nativeWorkerItemIds())
+            }
            // Snapshot of the header before the per-item payload is
            // attached; served to pollers that already consumed this
            // items payload (see getNativeWorkerSnapshot).
            val headerCandidate = if (includeItems) snapshot.toString() else null
-            snapshot.put("item_ids", nativeWorkerItemIds())
            if (includeItems) {
                snapshot.put("items", nativeWorkerItemsSnapshot(includeStatic = false))
+            } else if (progressItemIds != null) {
+                snapshot.put(
+                    "item_deltas",
+                    nativeWorkerItemsSnapshot(progressItemIds, includeStatic = false),
+                )
            } else {
                nativeWorkerItemSnapshot(currentItemId, includeStatic = false)?.let {
                    snapshot.put("item_delta", it)
@@ -209,60 +242,200 @@ internal fun DownloadService.updateNativeWorkerItem(itemId: String, updater: (Do
    }
 }

-internal fun DownloadService.updateNativeWorkerItemProgress(itemId: String) {
-    try {
-        val raw = Gobackend.getAllDownloadProgress()
+/**
+ * Polls the exported Go delta API once for the whole native queue. Workers
+ * update their item state from [nativeWorkerProgressItems] instead of making
+ * independent full-payload calls. The generation check prevents a delayed
+ * gomobile call from an old queue from contaminating a replacement queue.
+ */
+internal fun DownloadService.startNativeWorkerProgressCoordinator(generation: Long): Job {
+    nativeWorkerProgressJob?.cancel()
+    val coordinatorEpoch = nativeWorkerProgressEpoch.incrementAndGet()
+    synchronized(nativeWorkerProgressLock) {
+        nativeWorkerProgressItems.clear()
+        nativeWorkerProgressSeq = 0L
+    }
+
+    val job = serviceScope.launch {
+        val lastSignatures = mutableMapOf<String, String?>()
+        while (isActive && isNativeWorkerProgressActive(generation)) {
+            maintainNativeWorkerWakeLock()
+            val changedItemIds = pollNativeWorkerProgress(generation)
+            val snapshotItemIds = mutableListOf<String>()
+            for (itemId in changedItemIds) {
+                if (!updateNativeWorkerItemProgress(itemId, emitNotification = false)) {
+                    continue
+                }
+                val signature = synchronized(nativeWorkerItems) {
+                    nativeWorkerItems.firstOrNull { it.itemId == itemId }?.let {
+                        "${it.status}:${it.bytesReceived}:${it.bytesTotal}:${it.progress}"
+                    }
+                }
+                if (signature != null && lastSignatures[itemId] != signature) {
+                    lastSignatures[itemId] = signature
+                    snapshotItemIds.add(itemId)
+                }
+            }
+
+            if (snapshotItemIds.isNotEmpty() && isNativeWorkerProgressActive(generation)) {
+                // Only the active item is visible in the foreground
+                // notification. Apply it once after all cache updates so a
+                // concurrent queue never emits one notification per worker.
+                val activeItemId = nativeWorkerCurrentItemIdSnapshot()
+                if (activeItemId.isNotBlank() && activeItemId in snapshotItemIds) {
+                    updateNativeWorkerItemProgress(activeItemId, emitNotification = true)
+                }
+                val orderedItemIds = snapshotItemIds
+                    .filter { it != activeItemId }
+                    .let { ids ->
+                        if (activeItemId in snapshotItemIds) ids + activeItemId else ids
+                    }
+                // Commit every changed worker in one AtomicFile write. Writing
+                // one item_delta per worker would overwrite the same progress
+                // file repeatedly and expose only the last delta to Flutter.
+                writeNativeWorkerSnapshot(
+                    isRunning = true,
+                    isPaused = isNativeWorkerPaused(),
+                    currentItemId = activeItemId.ifBlank { orderedItemIds.last() },
+                    message = if (isNativeWorkerPaused()) nativeWorkerPauseMessage() else "Downloading",
+                    progressItemIds = orderedItemIds,
+                    progressCoordinatorEpoch = coordinatorEpoch,
+                )
+            }
+            delay(1000)
+        }
+    }
+    nativeWorkerProgressJob = job
+    return job
+}
+
+internal fun DownloadService.stopNativeWorkerProgressCoordinator(job: Job? = nativeWorkerProgressJob) {
+    if (job == null) return
+    if (nativeWorkerProgressJob === job) {
+        nativeWorkerProgressJob = null
+        nativeWorkerProgressEpoch.incrementAndGet()
+    }
+    job.cancel()
+}
+
+internal fun DownloadService.cancelNativeWorkerProgressCoordinator() {
+    stopNativeWorkerProgressCoordinator()
+    synchronized(nativeWorkerProgressLock) {
+        nativeWorkerProgressItems.clear()
+        nativeWorkerProgressSeq = 0L
+    }
+}
+
+private fun DownloadService.pollNativeWorkerProgress(generation: Long): Set<String> {
+    val sinceSeq = synchronized(nativeWorkerProgressLock) { nativeWorkerProgressSeq }
+    val raw = try {
+        Gobackend.getAllDownloadProgressDelta(sinceSeq)
+    } catch (_: Exception) {
+        return emptySet()
+    }
+    if (raw.isBlank() || !isNativeWorkerProgressActive(generation)) return emptySet()
+
+    return try {
        val root = JSONObject(raw)
-        val items = root.optJSONObject("items") ?: return
-        val progress = items.optJSONObject(itemId) ?: return
-        val backendStatus = progress.optString("status", "downloading")
-        val bytesReceived = progress.optLong("bytes_received", 0L)
-        val bytesTotal = progress.optLong("bytes_total", 0L)
+        val nextSeq = root.optLong("seq", sinceSeq)
+        val reset = root.optBoolean("reset", false)
+        val updated = mutableMapOf<String, NativeBackendProgress>()
+        root.optJSONObject("items")?.let { items ->
+            val keys = items.keys()
+            while (keys.hasNext()) {
+                val itemId = keys.next()
+                val item = items.optJSONObject(itemId) ?: continue
+                val bytesReceived = item.optLong("bytes_received", 0L).coerceAtLeast(0L)
+                val bytesTotal = item.optLong("bytes_total", 0L).coerceAtLeast(0L)
+                val progress = item.optDouble("progress", 0.0)
+                    .takeUnless { it.isNaN() }
+                    ?.coerceIn(0.0, 1.0)
+                    ?: 0.0
+                updated[itemId] = NativeBackendProgress(
+                    status = item.optString("status", "downloading"),
+                    bytesReceived = bytesReceived,
+                    bytesTotal = bytesTotal,
+                    progress = progress,
+                )
+            }
+        }
+        val removed = mutableListOf<String>()
+        root.optJSONArray("removed")?.let { ids ->
+            for (index in 0 until ids.length()) {
+                ids.optString(index, "").takeIf { it.isNotBlank() }?.let(removed::add)
+            }
+        }
+        synchronized(nativeWorkerProgressLock) {
+            if (!isNativeWorkerProgressActive(generation)) return emptySet()
+            if (reset) nativeWorkerProgressItems.clear()
+            nativeWorkerProgressItems.putAll(updated)
+            removed.forEach { nativeWorkerProgressItems.remove(it) }
+            if (nextSeq > nativeWorkerProgressSeq) {
+                nativeWorkerProgressSeq = nextSeq
+            }
+        }
+        updated.keys
+    } catch (_: Exception) {
+        emptySet()
+    }
+}
+
+internal fun DownloadService.updateNativeWorkerItemProgress(
+    itemId: String,
+    emitNotification: Boolean = true,
+): Boolean {
+    return try {
+        val progress = synchronized(nativeWorkerProgressLock) {
+            nativeWorkerProgressItems[itemId]
+        } ?: return false
+
+        val backendStatus = progress.status
        if (backendStatus == "preparing") {
-            currentStatus = "preparing"
            updateNativeWorkerItem(itemId) {
                it.status = "preparing"
                it.progress = 0.0
                it.bytesReceived = 0L
                it.bytesTotal = 0L
            }
-            lastProgress = 0L
-            lastTotal = 0L
-            updateNotification(0L, 0L)
-            return
+            if (emitNotification) {
+                currentStatus = "preparing"
+                lastProgress = 0L
+                lastTotal = 0L
+                updateNotification(0L, 0L)
+            }
+            return true
        }
-        val progressValue = if (bytesTotal > 0L) {
-            bytesReceived.toDouble() / bytesTotal.toDouble()
+
+        val progressValue = if (progress.bytesTotal > 0L) {
+            progress.bytesReceived.toDouble() / progress.bytesTotal.toDouble()
        } else {
-            progress.optDouble("progress", 0.0)
+            progress.progress
        }.coerceIn(0.0, 1.0)
-        currentStatus = if (backendStatus == "finalizing") {
-            "finalizing"
-        } else {
-            "downloading"
-        }
+        val itemStatus = if (backendStatus == "finalizing") "finalizing" else "downloading"
        updateNativeWorkerItem(itemId) {
-            it.status = currentStatus
+            it.status = itemStatus
            it.progress = progressValue
-            it.bytesReceived = bytesReceived
-            it.bytesTotal = bytesTotal
+            it.bytesReceived = progress.bytesReceived
+            it.bytesTotal = progress.bytesTotal
        }
-        if (bytesTotal > 0L) {
-            lastProgress = bytesReceived
-            lastTotal = bytesTotal
-            updateNotification(bytesReceived, bytesTotal)
-        } else if (progressValue > 0.0) {
-            val percentProgress = (progressValue * DownloadService.NOTIFICATION_PERCENT_TOTAL).toLong()
-                .coerceIn(0L, DownloadService.NOTIFICATION_PERCENT_TOTAL)
-            lastProgress = percentProgress
-            lastTotal = DownloadService.NOTIFICATION_PERCENT_TOTAL
-            updateNotification(percentProgress, DownloadService.NOTIFICATION_PERCENT_TOTAL)
-        } else {
-            lastProgress = 0L
-            lastTotal = 0L
-            updateNotification(0L, 0L)
+        if (emitNotification) {
+            currentStatus = itemStatus
+            if (progress.bytesTotal > 0L) {
+                lastProgress = progress.bytesReceived
+                lastTotal = progress.bytesTotal
+            } else if (progressValue > 0.0) {
+                lastProgress = (progressValue * DownloadService.NOTIFICATION_PERCENT_TOTAL).toLong()
+                    .coerceIn(0L, DownloadService.NOTIFICATION_PERCENT_TOTAL)
+                lastTotal = DownloadService.NOTIFICATION_PERCENT_TOTAL
+            } else {
+                lastProgress = 0L
+                lastTotal = 0L
+            }
+            updateNotification(lastProgress, lastTotal)
        }
+        true
    } catch (_: Exception) {
+        false
    }
 }

@@ -324,6 +497,22 @@ internal fun DownloadService.nativeWorkerItemsSnapshot(includeStatic: Boolean):
    return array
 }

+internal fun DownloadService.nativeWorkerItemsSnapshot(
+    itemIds: Collection<String>,
+    includeStatic: Boolean,
+): JSONArray {
+    val requested = itemIds.toSet()
+    val array = JSONArray()
+    synchronized(nativeWorkerItems) {
+        for (item in nativeWorkerItems) {
+            if (item.itemId in requested) {
+                array.put(nativeWorkerItemSnapshotLocked(item, includeStatic))
+            }
+        }
+    }
+    return array
+}
+
 internal fun DownloadService.nativeWorkerItemSnapshotLocked(item: DownloadService.NativeWorkerItem, includeStatic: Boolean): JSONObject {
    val json = JSONObject()
        .put("item_id", item.itemId)
@@ -40,6 +40,16 @@ object NativeDownloadFinalizer {
    // Native finalizer owns background-safe history writes while Flutter may be suspended.
    // Keep this schema contract in sync with Dart HistoryDatabase before bumping either side.
    const val HISTORY_SCHEMA_VERSION = 13
+    // Keep one native connection for the process. Opening history.db and
+    // probing/migrating its schema for every finalized track was expensive,
+    // and a single guarded writer also prevents native finalizer calls from
+    // interleaving transactions on the same connection. Flutter/sqflite uses
+    // its own WAL connection, so the busy timeout remains configured once on
+    // this connection for cross-connection contention.
+    private val historyDatabaseLock = Any()
+    private var historyDatabase: SQLiteDatabase? = null
+    private var historyDatabasePath = ""
+    private var historyDatabaseSchemaVersion = 0
    internal val activeFFmpegSessionIds = mutableSetOf<Long>()
    internal val nativeFFmpegSessionIds = BoundedRegistry<Long>(maxEntries = 256)
    internal val activeFFmpegSessionLock = Any()
@@ -1346,26 +1356,18 @@ object NativeDownloadFinalizer {
        values: ContentValues,
        deduplicateTrack: Boolean = true,
    ) {
-        val dbFile = File(File(context.applicationInfo.dataDir, "app_flutter"), "history.db")
-        dbFile.parentFile?.mkdirs()
-        val db = SQLiteDatabase.openDatabase(
-            dbFile.absolutePath,
-            null,
-            SQLiteDatabase.OPEN_READWRITE or
-                SQLiteDatabase.CREATE_IF_NECESSARY or
-                SQLiteDatabase.ENABLE_WRITE_AHEAD_LOGGING,
-        )
-        try {
-	            configureHistoryDatabase(db)
-	            db.beginTransaction()
-	            try {
-	                if (db.version > HISTORY_SCHEMA_VERSION) {
-	                    throw IllegalStateException(
-	                        "history schema v${db.version} is newer than native finalizer contract v$HISTORY_SCHEMA_VERSION"
-	                    )
-	                }
-	                val needsBackfill = db.version < HISTORY_SCHEMA_VERSION
-	                db.execSQL(
+        withHistoryDatabase(context) { db ->
+            val initializeSchema = historyDatabaseSchemaVersion != HISTORY_SCHEMA_VERSION
+            db.beginTransaction()
+            try {
+                if (initializeSchema) {
+                    if (db.version > HISTORY_SCHEMA_VERSION) {
+                        throw IllegalStateException(
+                            "history schema v${db.version} is newer than native finalizer contract v$HISTORY_SCHEMA_VERSION"
+                        )
+                    }
+                    val needsBackfill = db.version < HISTORY_SCHEMA_VERSION
+                db.execSQL(
 	                    """
 	                    CREATE TABLE IF NOT EXISTS history (
                      id TEXT PRIMARY KEY,
@@ -1463,21 +1465,60 @@ object NativeDownloadFinalizer {
 	                db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_album ON history(sort_album, sort_track, id)")
 	                db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_genre ON history(sort_genre, sort_track, id)")
 	                db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_release ON history(sort_release, sort_track, id)")
-	                if (db.version < HISTORY_SCHEMA_VERSION) db.version = HISTORY_SCHEMA_VERSION
-	                if (deduplicateTrack) deleteDuplicateHistoryRows(db, values)
-	                db.insertWithOnConflict("history", null, values, SQLiteDatabase.CONFLICT_REPLACE)
-	                replaceHistoryPathKeys(db, values.getAsString("id"), values.getAsString("file_path"))
-	                db.setTransactionSuccessful()
-	            } finally {
-	                db.endTransaction()
+                    if (db.version < HISTORY_SCHEMA_VERSION) db.version = HISTORY_SCHEMA_VERSION
+                }
+                if (deduplicateTrack) deleteDuplicateHistoryRows(db, values)
+                db.insertWithOnConflict("history", null, values, SQLiteDatabase.CONFLICT_REPLACE)
+                replaceHistoryPathKeys(db, values.getAsString("id"), values.getAsString("file_path"))
+                db.setTransactionSuccessful()
+            } finally {
+                db.endTransaction()
            }
-        } finally {
-            db.close()
+            if (initializeSchema) {
+                historyDatabaseSchemaVersion = HISTORY_SCHEMA_VERSION
+            }
+        }
+    }
+
+    private inline fun <T> withHistoryDatabase(
+        context: Context,
+        block: (SQLiteDatabase) -> T,
+    ): T {
+        synchronized(historyDatabaseLock) {
+            val dbFile = File(File(context.applicationInfo.dataDir, "app_flutter"), "history.db")
+            dbFile.parentFile?.mkdirs()
+            val db = historyDatabase?.takeIf {
+                it.isOpen && historyDatabasePath == dbFile.absolutePath
+            } ?: run {
+                historyDatabase?.close()
+                val opened = SQLiteDatabase.openDatabase(
+                    dbFile.absolutePath,
+                    null,
+                    SQLiteDatabase.OPEN_READWRITE or
+                        SQLiteDatabase.CREATE_IF_NECESSARY or
+                        SQLiteDatabase.ENABLE_WRITE_AHEAD_LOGGING,
+                )
+                try {
+                    configureHistoryDatabase(opened)
+                } catch (e: Exception) {
+                    opened.close()
+                    throw e
+                }
+                historyDatabase = opened
+                historyDatabasePath = dbFile.absolutePath
+                historyDatabaseSchemaVersion = 0
+                opened
+            }
+            historyDatabase = db
+            return block(db)
        }
    }

    private fun configureHistoryDatabase(db: SQLiteDatabase) {
        runHistoryPragma(db, "PRAGMA busy_timeout = 5000", required = false)
+        // CONFLICT_REPLACE must fire the history delete trigger so the
+        // external-content FTS index does not retain the replaced rowid.
+        runHistoryPragma(db, "PRAGMA recursive_triggers = ON", required = false)
        runHistoryPragma(db, "PRAGMA synchronous = NORMAL", required = false)
        runHistoryPragma(db, "PRAGMA journal_mode = WAL", required = false)
    }