diff --git a/app/src/main/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinator.kt b/app/src/main/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinator.kt index ece0ac012..0c67288b1 100644 --- a/app/src/main/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinator.kt +++ b/app/src/main/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinator.kt @@ -18,6 +18,7 @@ import javax.inject.Singleton import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.mapNotNull import kotlinx.coroutines.launch @@ -154,7 +155,7 @@ class PlaylistWatchTransferCoordinator @Inject constructor( continue } - val outcome = transferSongToAllNodes(batchId, nodes, song) + val outcome = transferSongToAllNodesWithRetry(batchId, nodes, song) if (outcome.completed) { transferStateStore.markBatchSongCompleted(batchId) } else { @@ -202,6 +203,35 @@ class PlaylistWatchTransferCoordinator @Inject constructor( } } + /** + * Retries [song] once after a transient failure, with a short backoff. Real hardware + * testing showed a song can legitimately fail (watch-side idle watchdog closing a live but + * slow Bluetooth stream — see WearTransferRepository) while a retry moments later succeeds + * cleanly: the watch's Bluetooth radio is shared with any connected BT headset, and a + * transfer can genuinely stall for a while under that contention without anything actually + * being broken. Doesn't retry past a cancellation, and re-transcodes on the retry rather + * than caching the first attempt's output — simpler and safe (transcoding on a modern phone + * is a few seconds, not the bottleneck), at the cost of redoing work that likely already + * succeeded once. + */ + private suspend fun transferSongToAllNodesWithRetry( + batchId: String, + nodes: List, + song: Song, + ): SongTransferResult { + val firstAttempt = transferSongToAllNodes(batchId, nodes, song) + if (firstAttempt.completed || cancelledBatchIds.contains(batchId)) return firstAttempt + + Timber.tag(TAG).w( + "Retrying transfer after failure: songId=%s errorCode=%s", + song.id, + firstAttempt.errorCode, + ) + delay(RETRY_BACKOFF_MS) + if (cancelledBatchIds.contains(batchId)) return firstAttempt + return transferSongToAllNodes(batchId, nodes, song) + } + /** Transcodes [song] once (if needed) and streams it to every reachable [nodes] in turn. */ private suspend fun transferSongToAllNodes( batchId: String, @@ -351,6 +381,11 @@ class PlaylistWatchTransferCoordinator @Inject constructor( // mark a legitimately-slow transfer as failed. private const val DEFAULT_SONG_TRANSFER_AWAIT_TIMEOUT_MS = 300_000L + // Short on purpose: a retry exists for transient stalls (radio contention with a + // connected BT headset, momentary Bluetooth hiccups), not to wait out a genuinely dead + // link — a longer backoff would just make a real failure take longer to report. + private const val RETRY_BACKOFF_MS = 3_000L + // How long resumePersistedBatchIfNeeded() waits for a fresh watch-library snapshot before // giving up and resuming anyway. Short: this only avoids some wasted duplicate-rejected // round-trips, it's not load-bearing for correctness (the watch rejects duplicates itself). diff --git a/app/src/test/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinatorTest.kt b/app/src/test/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinatorTest.kt index aea06d83f..82f034525 100644 --- a/app/src/test/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinatorTest.kt +++ b/app/src/test/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinatorTest.kt @@ -201,6 +201,8 @@ class PlaylistWatchTransferCoordinatorTest { val requestId = secondArg() val songId = thirdArg() transferredSongIdsInOrder += songId + // s2 fails on every attempt, including its retry (see the dedicated retry tests + // below) — this test is only about the batch surviving a song that never recovers. val status = if (songId == "s2") WearTransferProgress.STATUS_FAILED else WearTransferProgress.STATUS_COMPLETED transferStateStore.markProgress(requestId, songId, 0L, 0L, status) } @@ -209,13 +211,97 @@ class PlaylistWatchTransferCoordinatorTest { val batchId = coordinator.requestPlaylistTransfer("p1", "Playlist", listOf("s1", "s2", "s3")) advanceUntilIdle() - assertThat(transferredSongIdsInOrder).containsExactly("s1", "s2", "s3").inOrder() + // s2 appears twice: the first attempt and its retry. + assertThat(transferredSongIdsInOrder).containsExactly("s1", "s2", "s2", "s3").inOrder() val batch = transferStateStore.batchTransfers.value[batchId] assertThat(batch?.completedSongCount).isEqualTo(2) assertThat(batch?.failedSongCount).isEqualTo(1) assertThat(batch?.status).isEqualTo(WearTransferProgress.STATUS_COMPLETED) } + @Test + fun `a song that fails once but succeeds on retry counts as completed`() = runTest { + stubReachableNodes("node-1") + song("s1") + var attempt = 0 + every { + directTransferCoordinator.startTransferToWatch( + nodeId = any(), requestId = any(), songId = any(), + transferMode = any(), startPositionMs = any(), autoPlay = any(), audioOverride = any(), + ) + } answers { + val requestId = secondArg() + val songId = thirdArg() + attempt += 1 + val status = if (attempt == 1) WearTransferProgress.STATUS_FAILED else WearTransferProgress.STATUS_COMPLETED + transferStateStore.markProgress(requestId, songId, 0L, 0L, status) + } + val coordinator = buildCoordinator(this) + + val batchId = coordinator.requestPlaylistTransfer("p1", "Playlist", listOf("s1")) + advanceUntilIdle() + + assertThat(attempt).isEqualTo(2) + val batch = transferStateStore.batchTransfers.value[batchId] + assertThat(batch?.completedSongCount).isEqualTo(1) + assertThat(batch?.failedSongCount).isEqualTo(0) + } + + @Test + fun `a song failing twice in a row is only retried once, not indefinitely`() = runTest { + stubReachableNodes("node-1") + song("s1") + var attempts = 0 + every { + directTransferCoordinator.startTransferToWatch( + nodeId = any(), requestId = any(), songId = any(), + transferMode = any(), startPositionMs = any(), autoPlay = any(), audioOverride = any(), + ) + } answers { + val requestId = secondArg() + val songId = thirdArg() + attempts += 1 + transferStateStore.markProgress(requestId, songId, 0L, 0L, WearTransferProgress.STATUS_FAILED) + } + val coordinator = buildCoordinator(this) + + val batchId = coordinator.requestPlaylistTransfer("p1", "Playlist", listOf("s1")) + advanceUntilIdle() + + assertThat(attempts).isEqualTo(2) + val batch = transferStateStore.batchTransfers.value[batchId] + assertThat(batch?.failedSongCount).isEqualTo(1) + assertThat(batch?.completedSongCount).isEqualTo(0) + } + + @Test + fun `cancelling during the backoff window skips the retry`() = runTest { + stubReachableNodes("node-1") + song("s1") + val coordinator = buildCoordinator(this) + lateinit var batchId: String + + every { + directTransferCoordinator.startTransferToWatch( + nodeId = any(), requestId = any(), songId = any(), + transferMode = any(), startPositionMs = any(), autoPlay = any(), audioOverride = any(), + ) + } answers { + val requestId = secondArg() + val songId = thirdArg() + transferredSongIdsInOrder += songId + coordinator.cancelPlaylistTransfer(batchId) + transferStateStore.markProgress(requestId, songId, 0L, 0L, WearTransferProgress.STATUS_FAILED) + } + + batchId = coordinator.requestPlaylistTransfer("p1", "Playlist", listOf("s1")) + advanceUntilIdle() + + assertThat(transferredSongIdsInOrder).containsExactly("s1") + val batch = transferStateStore.batchTransfers.value[batchId] + assertThat(batch?.status).isEqualTo(WearTransferProgress.STATUS_CANCELLED) + } + @Test fun `cancelling a batch stops remaining songs from being transferred`() = runTest { stubReachableNodes("node-1") diff --git a/wear/src/main/java/com/theveloper/pixelplay/data/WearTransferRepository.kt b/wear/src/main/java/com/theveloper/pixelplay/data/WearTransferRepository.kt index 4821fbdb9..aaf4f0d33 100644 --- a/wear/src/main/java/com/theveloper/pixelplay/data/WearTransferRepository.kt +++ b/wear/src/main/java/com/theveloper/pixelplay/data/WearTransferRepository.kt @@ -125,6 +125,13 @@ class WearTransferRepository @Inject constructor( /** Failsafe timeout per transfer to avoid hanging states at 0%. */ private val transferWatchdogs = ConcurrentHashMap() + /** The live audio InputStream for a request, while onAudioChannelOpened is reading it — + * lets armTransferWatchdog actually interrupt a stuck read instead of just updating + * bookkeeping while the real transfer keeps running unaware. */ + private val openAudioStreams = ConcurrentHashMap() + /** Request IDs whose audio stream was closed by the watchdog, so onAudioChannelOpened's + * catch block can report "Transfer timed out" instead of a generic stream-closed message. */ + private val watchdogTimedOutRequestIds = ConcurrentHashMap.newKeySet() /** Request IDs currently receiving bytes through ChannelClient. */ private val activeChannelRequestIds = ConcurrentHashMap.newKeySet() /** Cancelled request IDs retained briefly so late metadata/progress/channel events are ignored safely. */ @@ -488,6 +495,7 @@ class WearTransferRepository @Inject constructor( if (!musicDir.exists()) musicDir.mkdirs() val tempFile = File(musicDir, "$requestId.part") var metadata: WearTransferMetadata? = pendingMetadata[requestId] + openAudioStreams[requestId] = inputStream try { if (isTransferCancelled(requestId)) { @@ -769,15 +777,18 @@ class WearTransferRepository @Inject constructor( "Transfer complete: ${resolvedMetadata.title} ($actualSize bytes) → ${localFile.absolutePath}" ) } catch (e: Exception) { + val timedOut = watchdogTimedOutRequestIds.remove(requestId) Timber.tag(TAG).e(e, "Failed to write transferred file") tempFile.delete() handleTransferError( requestId = requestId, songId = metadata?.songId ?: _activeTransfers.value[requestId]?.songId.orEmpty(), - message = e.message ?: "Write failed", + message = if (timedOut) "Transfer timed out" else (e.message ?: "Write failed"), ) } finally { activeChannelRequestIds.remove(requestId) + openAudioStreams.remove(requestId) + watchdogTimedOutRequestIds.remove(requestId) } } @@ -973,6 +984,8 @@ class WearTransferRepository @Inject constructor( pendingArtworkByRequestId.remove(requestId) clearTransferWatchdog(requestId) activeChannelRequestIds.remove(requestId) + openAudioStreams.remove(requestId) + watchdogTimedOutRequestIds.remove(requestId) } private fun handleTransferError(requestId: String, songId: String, message: String) { @@ -994,6 +1007,8 @@ class WearTransferRepository @Inject constructor( pendingArtworkByRequestId.remove(requestId) clearTransferWatchdog(requestId) activeChannelRequestIds.remove(requestId) + openAudioStreams.remove(requestId) + watchdogTimedOutRequestIds.remove(requestId) } private fun resolveTemporaryPlaybackStartPosition( @@ -1076,7 +1091,25 @@ class WearTransferRepository @Inject constructor( transferWatchdogs[requestId] = scope.launch { delay(TRANSFER_IDLE_TIMEOUT_MS) if (_activeTransfers.value.containsKey(requestId)) { - handleTransferError(requestId, songId, "Transfer timed out") + val stream = openAudioStreams[requestId] + if (stream != null) { + // A live audio stream is genuinely stuck: close it so the blocking + // read() in onAudioChannelOpened unblocks with an IOException and routes + // through that function's own catch block for cleanup — a single, + // consistent path instead of this watchdog declaring failure on its own + // while the read loop keeps running in the background, unaware anything + // happened. That's what let a "failed" transfer keep going and finish + // seconds later anyway, or worse, strip pendingMetadata out from under + // the still-running loop and turn a slow-but-fine transfer into a real + // failure ("Transfer metadata missing" from the loop's own metadata + // resolution not finding what this watchdog had just cleared). + watchdogTimedOutRequestIds.add(requestId) + runCatching { stream.close() } + } else { + // No audio stream open yet (still waiting on metadata/channel) — nothing + // to interrupt, so this is still the right place to declare failure. + handleTransferError(requestId, songId, "Transfer timed out") + } } } } diff --git a/wear/src/main/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModel.kt b/wear/src/main/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModel.kt index c681e9b20..543d48386 100644 --- a/wear/src/main/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModel.kt +++ b/wear/src/main/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModel.kt @@ -11,6 +11,7 @@ import com.theveloper.pixelplay.data.local.LocalPlaylistEntity import com.theveloper.pixelplay.data.local.LocalSongDao import com.theveloper.pixelplay.data.local.LocalSongEntity import com.theveloper.pixelplay.data.WearTransferRepository +import com.theveloper.pixelplay.shared.WearTransferProgress import dagger.hilt.android.lifecycle.HiltViewModel import javax.inject.Inject import kotlinx.coroutines.ExperimentalCoroutinesApi @@ -53,15 +54,26 @@ class WearLocalPlaylistViewModel @Inject constructor( /** In-flight song transfers from the phone, keyed by requestId — for on-screen receive feedback. */ val activeTransfers: StateFlow> = transferRepository.activeTransfers - /** Playlists that currently have at least one of their songs actively transferring. */ + /** + * Playlists that currently have at least one of their songs actively transferring. + * + * Only [WearTransferProgress.STATUS_TRANSFERRING] counts as "still receiving" — a failed or + * cancelled transfer stays in [WearTransferRepository.activeTransfers] indefinitely (so + * DownloadsScreen can list it under "Transfer issues"), but that's a terminal state, not an + * in-progress one. Treating mere presence in the map as "active" left this badge stuck on + * forever once a song failed. + */ val playlistIdsReceiving: StateFlow> = combine( localPlaylistDao.observeAllPlaylistSongCrossRefs(), transferRepository.activeTransfers, ) { crossRefs, transfers -> - if (transfers.isEmpty()) { + val activeSongIds = transfers.values + .filter { it.status == WearTransferProgress.STATUS_TRANSFERRING } + .map { it.songId } + .toSet() + if (activeSongIds.isEmpty()) { emptySet() } else { - val activeSongIds = transfers.values.map { it.songId }.toSet() crossRefs.filter { it.songId in activeSongIds }.map { it.playlistId }.toSet() } }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000L), emptySet()) diff --git a/wear/src/test/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModelTest.kt b/wear/src/test/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModelTest.kt index 3b666f93a..6458ef107 100644 --- a/wear/src/test/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModelTest.kt +++ b/wear/src/test/java/com/theveloper/pixelplay/presentation/viewmodel/WearLocalPlaylistViewModelTest.kt @@ -255,6 +255,41 @@ class WearLocalPlaylistViewModelTest { } } + @Test + fun `a failed transfer no longer counts as receiving once it reaches a terminal state`() = runTest { + allCrossRefsFlow.value = listOf(crossRef("p1", "s1", 0)) + + viewModel.playlistIdsReceiving.test { + assertThat(awaitItem()).isEmpty() // deduped placeholder, see the test above + transferRepository.onProgressReceived( + WearTransferProgress( + requestId = "r1", + songId = "s1", + bytesTransferred = 10L, + totalBytes = 100L, + status = WearTransferProgress.STATUS_TRANSFERRING, + ) + ) + assertThat(awaitItem()).containsExactly("p1") + + // The transfer fails — WearTransferRepository deliberately keeps this entry in + // activeTransfers (DownloadsScreen lists failed transfers under "Transfer issues"), + // it doesn't remove it. playlistIdsReceiving must stop counting it anyway: mere + // presence in the map isn't "still receiving" once the status is terminal. + transferRepository.onProgressReceived( + WearTransferProgress( + requestId = "r1", + songId = "s1", + bytesTransferred = 10L, + totalBytes = 100L, + status = WearTransferProgress.STATUS_FAILED, + error = "Transfer timed out", + ) + ) + assertThat(awaitItem()).isEmpty() + } + } + @Test fun `playAll switches output to watch when at least one song is available`() = expectFireAndForgetPlaybackCrash { playlistSongsFlowById["p1"] = MutableStateFlow(listOf(crossRef("p1", "s1", 0)))