2.1.5 bug fixes: reconnect, offline playback, downloads

app/src/main/java/com/ninelivesaudio/app/data/repository/ProgressRepository.kt

@@ -14,6 +14,7 @@ import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
 import javax.inject.Inject
 import javax.inject.Singleton
 import kotlin.time.Duration
@@ -320,6 +321,7 @@ class ProgressRepository @Inject constructor(
         isFinished: Boolean,
         duration: Double,
         pushToServer: Boolean,
+        pushTimeout: Duration = Duration.INFINITE,
         onPersisted: suspend () -> Unit = {},
     ): Boolean = pendingProgressQueueOwner.withItemLock(itemId) {
         persistProgressAndEnqueueLocked(
@@ -330,12 +332,15 @@ class ProgressRepository @Inject constructor(
             onPersisted = onPersisted,
         )
         val pushed = if (pushToServer && progressCanBeDelivered(isFinished, duration)) {
-            try {
-                apiService.updateProgress(itemId, currentTime, isFinished, duration)
-            } catch (cancellation: kotlinx.coroutines.CancellationException) {
-                throw cancellation
-            } catch (_: Exception) {
-                false
+            // The row queued above delivers it later if this gives up.
+            pushWithin(pushTimeout) {
+                try {
+                    apiService.updateProgress(itemId, currentTime, isFinished, duration)
+                } catch (cancellation: kotlinx.coroutines.CancellationException) {
+                    throw cancellation
+                } catch (_: Exception) {
+                    false
+                }
             }
         } else {
             false
@@ -619,6 +624,10 @@ internal fun serverProgressMayReplaceLocal(hasPendingProgress: Boolean): Boolean
  * from their durable source of truth by [legacyProgressSnapshot], so this check
  * only decides the fate of rows with proven atomic provenance.
  */
+/** A push that has not answered within [timeout] counts as not delivered. */
+internal suspend fun pushWithin(timeout: Duration, push: suspend () -> Boolean): Boolean =
+    withTimeoutOrNull(timeout) { push() } ?: false
+
 internal fun queuedRowsAreSuperseded(
     rows: List<PendingProgressEntity>,
     durableProgress: PlaybackProgressEntity?,

app/src/main/java/com/ninelivesaudio/app/service/DownloadManager.kt

@@ -18,10 +18,15 @@ import com.ninelivesaudio.app.service.download.DownloadEngine
 import com.ninelivesaudio.app.service.download.DownloadNotifications
 import com.ninelivesaudio.app.service.download.DownloadQueueWorker
 import com.ninelivesaudio.app.service.download.estimateTotalBytes
+import com.ninelivesaudio.app.service.download.finishInOwnerScope
+import com.ninelivesaudio.app.service.download.pauseKeepsCompletedRow
 import com.ninelivesaudio.app.service.download.DownloadSlotStore
 import com.ninelivesaudio.app.service.download.selectNextDownload
+import com.ninelivesaudio.app.service.download.writeAfterEngineStops
 import dagger.hilt.android.qualifiers.ApplicationContext
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
@@ -70,6 +75,13 @@ class DownloadManager @Inject constructor(
     private val slotMutex = Mutex()
     private val workManager: WorkManager by lazy { WorkManager.getInstance(context) }
 
+    /**
+     * Where user pause, cancel and delete actually run. Their callers are screen
+     * coroutines, and leaving the screen between the worker stop and the row
+     * write used to leave the book Downloading with the queue stopped.
+     */
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+
     private companion object {
         const val WORKER_STOP_TIMEOUT_MS = 10_000L
         const val WORKER_STOP_POLL_MS = 100L
@@ -386,8 +398,7 @@ class DownloadManager @Inject constructor(
         }
         if (!hasLosingDownload) return
 
-        workManager.cancelUniqueWork(DOWNLOAD_WORK_NAME)
-        awaitDrainStopped()
+        stopDrainAndAwait()
 
         slotMutex.withLock {
             downloadItemDao.getAll()
@@ -412,6 +423,11 @@ class DownloadManager @Inject constructor(
      * gives up and proceeds, which is the same risk profile as before this
      * function existed.
      */
+    private suspend fun stopDrainAndAwait() {
+        workManager.cancelUniqueWork(DOWNLOAD_WORK_NAME)
+        awaitDrainStopped()
+    }
+
     private suspend fun awaitDrainStopped() {
         val deadline = System.currentTimeMillis() + WORKER_STOP_TIMEOUT_MS
         while (System.currentTimeMillis() < deadline) {
@@ -450,12 +466,23 @@ class DownloadManager @Inject constructor(
     }
 
     /** Pause a download: mark it Paused; restart the drain if it was the active one. */
-    suspend fun pauseDownload(downloadId: String) {
-        val entity = downloadItemDao.getById(downloadId) ?: return
+    suspend fun pauseDownload(downloadId: String): Unit = finishInOwnerScope(scope) {
+        val entity = downloadItemDao.getById(downloadId) ?: return@finishInOwnerScope
         val wasDownloading = entity.status == DownloadStatus.Downloading.ordinal
-        downloadItemDao.upsert(entity.copy(status = DownloadStatus.Paused.ordinal))
+        writeAfterEngineStops(wasDownloading, { stopDrainAndAwait() }) {
+            // Re-read after the stop: the engine's last tick moved the bytes on,
+            // and it may have finished the book in the meantime. A Completed row
+            // only counts once the book itself is marked downloaded, since the
+            // stop can land between those two writes.
+            val latest = downloadItemDao.getById(downloadId) ?: return@writeAfterEngineStops
+            val book = audioBookDao.getById(latest.audioBookId)
+            if (pauseKeepsCompletedRow(latest.status, book?.isDownloaded, book?.localPath)) {
+                return@writeAfterEngineStops
+            }
+            downloadItemDao.upsert(latest.copy(status = DownloadStatus.Paused.ordinal))
+        }
         if (wasDownloading) {
-            // Stop the engine on this book and let the drain continue with the rest.
+            // The engine was stopped above; let the drain continue with the rest.
             enqueueDrain(replace = true)
         }
     }
@@ -489,29 +516,43 @@ class DownloadManager @Inject constructor(
     }
 
     /** Cancel a download and clean up; restart the drain if it was the active one. */
-    suspend fun cancelDownload(downloadId: String) {
+    suspend fun cancelDownload(downloadId: String): Unit = finishInOwnerScope(scope) {
         val entity = downloadItemDao.getById(downloadId)
         val wasDownloading = entity?.status == DownloadStatus.Downloading.ordinal
-        downloadItemDao.deleteById(downloadId)
+        writeAfterEngineStops(wasDownloading, { stopDrainAndAwait() }) {
+            downloadItemDao.deleteById(downloadId)
+        }
 
         if (downloadItemDao.getDownloadable().isEmpty()) {
             // Nothing left to download: stop the drain and clear the notification
             // so it doesn't linger after cancelling the last item.
             workManager.cancelUniqueWork(DOWNLOAD_WORK_NAME)
             DownloadNotifications.clearAll(context)
         } else if (wasDownloading) {
-            // More queued: stop the engine on the cancelled book and continue.
+            // More queued: the engine was stopped above, so continue with the rest.
             enqueueDrain(replace = true)
         }
     }
 
     /** Delete a download's files and DB record. */
-    suspend fun deleteDownload(audioBookId: String) {
+    suspend fun deleteDownload(audioBookId: String): Unit = finishInOwnerScope(scope) {
+        val downloadEntity = downloadItemDao.getByAudioBookId(audioBookId)
+        val wasDownloading = downloadEntity?.status == DownloadStatus.Downloading.ordinal
+        writeAfterEngineStops(wasDownloading, { stopDrainAndAwait() }) {
+            deleteDownloadedFilesAndRows(audioBookId, downloadEntity?.id)
+        }
+        if (wasDownloading) {
+            // Restart the drain for whatever else is queued.
+            enqueueDrain(replace = true)
+        }
+    }
+
+    private suspend fun deleteDownloadedFilesAndRows(audioBookId: String, downloadId: String?) {
+        // Read the book only now, after any engine on it has stopped, so a
+        // completion that landed during the stop is the state being undone.
         // Use the actual localPath stored on the audiobook — this matches the path
         // set by the engine (basePath/Author - Title), not basePath/audioBookId.
         val bookEntity = audioBookDao.getById(audioBookId)
-        val downloadEntity = downloadItemDao.getByAudioBookId(audioBookId)
-        val wasDownloading = downloadEntity?.status == DownloadStatus.Downloading.ordinal
 
         withContext(Dispatchers.IO) {
             val localPath = bookEntity?.localPath
@@ -525,12 +566,8 @@ class DownloadManager @Inject constructor(
             // above, so drop its reference too.
             audioBookDao.upsert(bookEntity.copy(isDownloaded = 0, localPath = null, localCoverPath = null))
         }
-        if (downloadEntity != null) {
-            downloadItemDao.deleteById(downloadEntity.id)
-        }
-        if (wasDownloading) {
-            // Stop the engine if it was mid-download on this book.
-            enqueueDrain(replace = true)
+        if (downloadId != null) {
+            downloadItemDao.deleteById(downloadId)
         }
     }
 

app/src/main/java/com/ninelivesaudio/app/service/PlaybackManager.kt

@@ -559,6 +559,31 @@ internal fun listeningSessionToOpen(
     else -> ListeningSessionKind.NONE
 }
 
+internal enum class LoadSessionOpen { AWAIT, BOUNDED, DEFER }
+
+/** How long a downloaded book's load waits on its server listening session. */
+internal val DOWNLOADED_SESSION_OPEN_TIMEOUT = 5.seconds
+
+/**
+ * How a load opens its server listening session.
+ *
+ * A streaming book waits for it, because its tracks and position feed the
+ * load. A downloaded book plays from disk, so it skips the open while the
+ * server is not reachable and waits at most [DOWNLOADED_SESSION_OPEN_TIMEOUT]
+ * while it is. Either way ensureListeningSessionStarted opens the session in
+ * the background once playback starts. SYNCING counts as reachable, as it
+ * does everywhere else.
+ */
+internal fun loadSessionOpen(
+    usesLocalTracks: Boolean,
+    connectionStatus: ConnectivityMonitor.ConnectionStatus,
+): LoadSessionOpen = when {
+    !usesLocalTracks -> LoadSessionOpen.AWAIT
+    connectionStatus == ConnectivityMonitor.ConnectionStatus.CONNECTED ||
+        connectionStatus == ConnectivityMonitor.ConnectionStatus.SYNCING -> LoadSessionOpen.BOUNDED
+    else -> LoadSessionOpen.DEFER
+}
+
 internal enum class PlaybackItemPersistence { KEEP, SAVE, CLEAR }
 
 internal fun playbackItemPersistenceAction(
@@ -1273,7 +1298,13 @@ class PlaybackManager @Inject constructor(
                     hasLocalSession = synchronized(sessionLock) { currentLocalSessionId != null },
                 ) == ListeningSessionKind.SERVER
             ) {
-                openServerListeningSession(book, requestedGeneration, loadRequest)
+                when (loadSessionOpen(remoteAccess.usesLocalTracks(), connectivityMonitor.connectionStatus.value)) {
+                    LoadSessionOpen.AWAIT -> openServerListeningSession(book, requestedGeneration, loadRequest)
+                    LoadSessionOpen.BOUNDED -> withTimeoutOrNull(DOWNLOADED_SESSION_OPEN_TIMEOUT) {
+                        openServerListeningSession(book, requestedGeneration, loadRequest)
+                    }
+                    LoadSessionOpen.DEFER -> Unit
+                }
                 if (!playbackLoadOwner.isCurrent(loadRequest)) return false
             }
 
@@ -2024,7 +2055,7 @@ class PlaybackManager @Inject constructor(
                         try {
                             syncTerminalSession(terminal)
                         } finally {
-                            closeSession(terminal.serverSessionId, terminal.serverListening)
+                            closeSession(terminal.serverSessionId, terminal.serverListening, terminal = true)
                             if (localSession.sessionId != null) {
                                 try {
                                     sessionRepository.updateLocalSession(
@@ -2798,16 +2829,20 @@ class PlaybackManager @Inject constructor(
     private suspend fun syncTerminalSession(snapshot: TerminalPlaybackSnapshot) {
         val posSec = snapshot.position.toDouble(kotlin.time.DurationUnit.SECONDS)
         val durSec = snapshot.duration.toDouble(kotlin.time.DurationUnit.SECONDS)
-        if (snapshot.serverSessionId != null) {
+        if (snapshot.serverSessionId != null && terminalServerLive()) {
             try {
                 snapshot.serverListening?.deliver(snapshot.timeListened) { delta ->
-                    progressRepository.syncSessionProgress(
-                        itemId = snapshot.bookId,
-                        sessionId = snapshot.serverSessionId,
-                        currentTime = posSec,
-                        duration = durSec,
-                        timeListened = delta,
-                    )
+                    // Bounded inside deliver, whose own NonCancellable would
+                    // otherwise outlast any timeout placed around it.
+                    terminalServerCall(serverLive = true) {
+                        progressRepository.syncSessionProgress(
+                            itemId = snapshot.bookId,
+                            sessionId = snapshot.serverSessionId,
+                            currentTime = posSec,
+                            duration = durSec,
+                            timeListened = delta,
+                        )
+                    } ?: false
                 }
             } catch (cancellation: CancellationException) {
                 throw cancellation
@@ -2897,11 +2932,27 @@ class PlaybackManager @Inject constructor(
         }
     }
 
-    private suspend fun closeSession(sessionId: String?, listening: ServerSessionListening? = null) {
+    private fun terminalServerLive(): Boolean = terminalProgressPushAllowed(
+        isOnline = connectivityMonitor.isOnline.value,
+        connectionStatus = connectivityMonitor.connectionStatus.value,
+    )
+
+    private suspend fun closeSession(
+        sessionId: String?,
+        listening: ServerSessionListening? = null,
+        terminal: Boolean = false,
+    ) {
         if (sessionId == null) return
-        try {
-            if (listening != null) listening.close { apiService.closeSession(sessionId) }
+        // A stopped book's close holds up the next load, so it gets the terminal
+        // policy. A skipped or timed out close leaves the server session open
+        // until the server expires it; local state is cleared below either way.
+        val closeRemote: suspend () -> Unit = {
+            if (terminal) terminalServerCall(terminalServerLive()) { apiService.closeSession(sessionId) }
             else apiService.closeSession(sessionId)
+        }
+        try {
+            if (listening != null) listening.close(closeRemote)
+            else closeRemote()
         } catch (cancellation: CancellationException) {
             throw cancellation
         } catch (_: Exception) {}