fix(wear): confiabilidad de la transferencia de playlists — hallazgos de la primera prueba en hardware

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<Node>,
+        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).

app/src/test/java/com/theveloper/pixelplay/data/service/wear/PlaylistWatchTransferCoordinatorTest.kt

@@ -201,6 +201,8 @@ class PlaylistWatchTransferCoordinatorTest {
             val requestId = secondArg<String>()
             val songId = thirdArg<String>()
             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<String>()
+            val songId = thirdArg<String>()
+            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<String>()
+            val songId = thirdArg<String>()
+            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<String>()
+            val songId = thirdArg<String>()
+            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")

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<String, Job>()
+    /** 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<String, InputStream>()
+    /** 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<String>()
     /** Request IDs currently receiving bytes through ChannelClient. */
     private val activeChannelRequestIds = ConcurrentHashMap.newKeySet<String>()
     /** 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")
+                }
             }
         }
     }

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<Map<String, TransferState>> = 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<Set<String>> = 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())

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)))