From d233966606ad6aaeac0c5d8661ecd93b7353c4fd Mon Sep 17 00:00:00 2001 From: PonceGL Date: Sun, 9 Aug 2026 11:53:48 -0600 Subject: [PATCH 1/3] fix(wear): make the transfer idle watchdog actually cancel the stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found on real hardware: the watchdog firing only flipped bookkeeping state (activeTransfers -> FAILED, pendingMetadata cleared) while the coroutine actually reading the ChannelClient InputStream kept running, completely unaware anything had happened. Two ways that went wrong in one 6-song transfer over Bluetooth with a connected BT headset: - A song reported 'timed out' but the read loop kept going anyway and finished successfully seconds later — a false alarm, but the watchdog re-armed on every remaining read, so the same requestId could fire repeatedly (observed 2, 3, and 4 times on one transfer). - Worse: firing cleared pendingMetadata out from under the still-live loop, so when it finally tried to resolve metadata to finish writing the file, it found nothing — 'Transfer metadata missing', the file got deleted, and the song was lost for real despite every byte having arrived over the wire. Root cause: no link between the watchdog and the actual I/O. Now armTransferWatchdog closes the live InputStream (tracked per requestId in openAudioStreams) when it fires, which unblocks the loop's read() with an IOException and routes it through onAudioChannelOpened's own catch block for one consistent cleanup path — instead of the watchdog declaring failure independently. watchdogTimedOutRequestIds lets that catch block report 'Transfer timed out' instead of a generic stream-closed message. Not unit-testable as-is: the watchdog is a real delay() with no injected clock, and onAudioChannelOpened is private. Verified by code review against the exact failure sequence from hardware; needs re-verification on-device (same scenario: BT headphones connected, transfer running) before calling it confirmed fixed. --- .../pixelplay/data/WearTransferRepository.kt | 37 ++++++++++++++++++- 1 file changed, 35 insertions(+), 2 deletions(-) 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") + } } } } From 51ffdef5047b21f9dbc624f0a76753bcda2ff5f9 Mon Sep 17 00:00:00 2001 From: PonceGL Date: Sun, 9 Aug 2026 11:53:54 -0600 Subject: [PATCH 2/3] fix(wear): stop counting failed transfers as still receiving MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found on real hardware: once a song's transfer failed, its playlist kept showing the 'Receiving…' badge forever. playlistIdsReceiving treated mere presence in WearTransferRepository.activeTransfers as 'in progress' — but a failed entry deliberately stays in that map (DownloadsScreen lists it under 'Transfer issues'), it's just no longer active. Now only STATUS_TRANSFERRING counts. --- .../viewmodel/WearLocalPlaylistViewModel.kt | 18 ++++++++-- .../WearLocalPlaylistViewModelTest.kt | 35 +++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) 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))) From e4b13fc26735e0fd8b29ee224ad156ea64926938 Mon Sep 17 00:00:00 2001 From: PonceGL Date: Sun, 9 Aug 2026 11:54:01 -0600 Subject: [PATCH 3/3] feat(app): retry a song once after a transient transfer failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The plan's own risk table already called for this ('reintento con backoff') but it was never implemented. Real hardware testing showed why it matters: 2 of 6 songs in one batch failed outright, both consistent with a Bluetooth stall from the watch's radio being shared with a connected BT headset — a real, non-theoretical condition, not a broken link. transferSongToAllNodesWithRetry re-attempts once, after a short fixed backoff, before the coordinator gives up on a song. Re-transcodes on the retry rather than caching the first attempt's output — simpler, and cheap enough on a modern phone's hardware encoder to not be worth the extra bookkeeping. --- .../wear/PlaylistWatchTransferCoordinator.kt | 37 +++++++- .../PlaylistWatchTransferCoordinatorTest.kt | 88 ++++++++++++++++++- 2 files changed, 123 insertions(+), 2 deletions(-) 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")