address the artwork review: snapshots, atomic writes, bounded reads

- SourceRepository.coverArt and networkCoverArt read the source id and the
  source object separately, so a source switch between the two could pair one
  source's cache key with another source's URL; for the store that meant writing
  the wrong server's artwork into the per-source file for good. Both now resolve
  from a single active snapshot.
- ArtworkStore wrote through a fixed `<name>.part`, so two downloads of songs
  from one album could interleave: one publishing the file while the other still
  had the shared inode open, which leaves a torn image for Coil to decode. Each
  write gets its own partial now.
- OkHttpArtworkFetcher read the whole body into memory whatever its size. It
  refuses anything past 8 MB, checked by declared length and again while reading.
- Artwork whose library rows could not all be resolved is no longer swept:
  pruning an album while its songs remain used to delete art a download still
  held. A source with no download rows still sweeps everything.
- fetchArtworkSafely and storeArtwork no longer swallow CancellationException,
  and the poll loop fetches a resumed row's artwork off the tick instead of
  holding up the next one.
- Reconcile reads the download rows once rather than twice, and download() no
  longer repeats the check enqueueIfAbsent makes under the lock.

app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt

@@ -16,7 +16,9 @@ interface DownloadDao {
     suspend fun all(): List<SongDownload>
 
     @Query(
-        "SELECT d.sourceId AS sourceId, al.coverArt AS albumCoverArt, ar.coverArt AS artistCoverArt " +
+        "SELECT d.sourceId AS sourceId, s.albumId AS albumId, s.artistId AS artistId, " +
+            "al.id AS albumRow, al.coverArt AS albumCoverArt, " +
+            "ar.id AS artistRow, ar.coverArt AS artistCoverArt " +
             "FROM song_downloads d " +
             "LEFT JOIN songs s ON s.sourceId = d.sourceId AND s.id = d.songId " +
             "LEFT JOIN albums al ON al.sourceId = s.sourceId AND al.id = s.albumId " +

app/src/main/kotlin/com/subtracks/data/download/ArtworkStore.kt

@@ -37,9 +37,13 @@ class ArtworkStore(
         val target = file(sourceId, cacheKey)
         return runCatching {
             target.parentFile?.mkdirs()
-            val partial = File(target.parentFile, target.name + PARTIAL_SUFFIX)
-            partial.writeBytes(bytes)
-            partial.renameTo(target)
+            val partial = File.createTempFile(target.name, PARTIAL_SUFFIX, target.parentFile)
+            try {
+                partial.writeBytes(bytes)
+                partial.renameTo(target)
+            } finally {
+                partial.delete()
+            }
         }.getOrDefault(false)
     }
 
@@ -70,6 +74,7 @@ fun interface ArtworkFetcher {
 
 class OkHttpArtworkFetcher(
     http: OkHttpClient,
+    private val maxBytes: Long = MAX_BYTES,
 ) : ArtworkFetcher {
     // A blocking OkHttp call is not interrupted when the coroutine around it is cancelled, so the
     // per-call timeout is what actually bounds a stuck artwork request.
@@ -80,11 +85,17 @@ class OkHttpArtworkFetcher(
             val request = Request.Builder().url(url).build()
             http.newCall(request).execute().use { response ->
                 if (!response.isSuccessful) throw IOException("Artwork request failed with ${response.code}")
-                response.body?.bytes() ?: throw IOException("Artwork response had no body")
+                val body = response.body ?: throw IOException("Artwork response had no body")
+                if (body.contentLength() > maxBytes) throw IOException("Artwork response is larger than $maxBytes bytes")
+                val source = body.source()
+                source.request(maxBytes + 1)
+                if (source.buffer.size > maxBytes) throw IOException("Artwork response is larger than $maxBytes bytes")
+                source.readByteArray()
             }
         }
 
     private companion object {
         const val CALL_TIMEOUT_SECONDS = 15L
+        const val MAX_BYTES = 8L * 1024 * 1024
     }
 }

app/src/main/kotlin/com/subtracks/data/model/Download.kt

@@ -42,9 +42,16 @@ data class SongDownload(
 
 data class DownloadArtwork(
     val sourceId: Long,
+    val albumId: String?,
+    val artistId: String?,
+    val albumRow: String?,
+    val artistRow: String?,
     val albumCoverArt: String?,
     val artistCoverArt: String?,
-)
+) {
+    val hasMissingLibraryRow: Boolean
+        get() = (albumId != null && albumRow == null) || (artistId != null && artistRow == null)
+}
 
 class DownloadStatusConverter {
     @ColumnTypeConverter

app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt

@@ -11,6 +11,7 @@ import com.subtracks.data.download.EngineStatus
 import com.subtracks.data.model.DownloadStatus
 import com.subtracks.data.model.SongDownload
 import com.subtracks.data.model.coverArtKey
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
@@ -94,7 +95,6 @@ class DownloadRepository(
             showMessage(NOT_ACTIVE)
             return@withContext
         }
-        if (alreadyDownloadingOrDownloaded(sourceId, songId)) return@withContext
         if (sourceRepository.downloadUri(sourceId, songId) == null) {
             showMessage(NO_ADDRESS)
             return@withContext
@@ -148,8 +148,9 @@ class DownloadRepository(
         withContext(dispatcher) {
             val resumed =
                 mutex.withLock {
-                    val resumed = reconcileRows(db.downloadDao().all())
-                    sweep(db.downloadDao().all())
+                    val rows = db.downloadDao().all()
+                    val resumed = reconcileRows(rows)
+                    sweep(rows)
                     ensurePolling()
                     resumed
                 }
@@ -196,17 +197,30 @@ class DownloadRepository(
         val cacheKey = coverArtKey(sourceId, coverArt, thumbnail)
         if (artworkStore.file(sourceId, cacheKey).exists()) return
         val url = sourceRepository.networkCoverArt(sourceId, coverArt, thumbnail) ?: return
-        val bytes = runCatching { artworkFetcher.fetch(url) }.getOrNull() ?: return
+        val bytes =
+            try {
+                artworkFetcher.fetch(url)
+            } catch (e: CancellationException) {
+                throw e
+            } catch (_: Exception) {
+                return
+            }
         if (bytes.isEmpty()) return
         artworkStore.write(sourceId, cacheKey, bytes)
     }
 
-    private suspend fun neededArtwork(): Map<Long, Set<String>> =
+    /**
+     * The artwork keys every download row still needs, per source. A null entry means the source's
+     * rows could not all be resolved against the library, in which case nothing is swept for it:
+     * pruning an album or artist would otherwise delete artwork a download still holds.
+     */
+    private suspend fun artworkToKeep(): Map<Long, Set<String>?> =
         db
             .downloadDao()
             .artwork()
             .groupBy { it.sourceId }
             .mapValues { (sourceId, rows) ->
+                if (rows.any { it.hasMissingLibraryRow }) return@mapValues null
                 rows
                     .flatMap { row -> listOfNotNull(row.albumCoverArt, row.artistCoverArt) }
                     .flatMap { cover -> listOf(false, true).map { thumbnail -> coverArtKey(sourceId, cover, thumbnail) } }
@@ -272,7 +286,13 @@ class DownloadRepository(
         sourceId: Long,
         songId: String,
     ) {
-        runCatching { withTimeoutOrNull(ART_TIMEOUT_MS) { fetchArtwork(sourceId, songId) } }
+        try {
+            withTimeoutOrNull(ART_TIMEOUT_MS) { fetchArtwork(sourceId, songId) }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (_: Exception) {
+            // Artwork is best effort and must never disturb a download that is already running.
+        }
     }
 
     private suspend fun apply(
@@ -319,17 +339,23 @@ class DownloadRepository(
                             if (active.isEmpty()) return@launch
                             reconcileRows(active)
                         }
-                    resumed.forEach { fetchArtworkSafely(it.sourceId, it.songId) }
+                    // Fetching artwork here would hold up the next tick, and with it the progress
+                    // the queue is mirrored from.
+                    resumed.forEach { row -> scope.launch { fetchArtworkSafely(row.sourceId, row.songId) } }
                 }
             }
     }
 
     private suspend fun sweep(rows: List<SongDownload>) {
         val knownSongs = rows.map { it.sourceId to it.songId }.toHashSet()
-        val knownArt = neededArtwork()
+        val sourcesWithRows = rows.mapTo(HashSet()) { it.sourceId }
+        val knownArt = artworkToKeep()
         downloadsDir.listFiles().orEmpty().filter { it.isDirectory }.forEach { sourceDir ->
             val sourceId = sourceDir.name.toLongOrNull() ?: return@forEach
-            artworkStore.sweep(sourceId, knownArt[sourceId].orEmpty())
+            when {
+                sourceId !in sourcesWithRows -> artworkStore.sweep(sourceId, emptySet())
+                else -> knownArt[sourceId]?.let { artworkStore.sweep(sourceId, it) }
+            }
             sourceDir.listFiles().orEmpty().filter { it.isFile }.forEach { file ->
                 if ((sourceId to file.name) !in knownSongs) file.delete()
             }

app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt

@@ -84,11 +84,11 @@ class SourceRepository(
         coverArt: String?,
         thumbnail: Boolean = false,
     ): CoverArtRef? {
-        val sourceId = activeSourceId ?: return null
+        val source = active ?: return null
         if (coverArt == null) return null
-        val cacheKey = coverArtKey(sourceId, coverArt, thumbnail)
-        artworkStore.uri(sourceId, cacheKey)?.let { return CoverArtRef(url = it, cacheKey = cacheKey) }
-        val url = active?.coverArtUri(coverArt, thumbnail)?.toString() ?: return null
+        val cacheKey = coverArtKey(source.id, coverArt, thumbnail)
+        artworkStore.uri(source.id, cacheKey)?.let { return CoverArtRef(url = it, cacheKey = cacheKey) }
+        val url = source.coverArtUri(coverArt, thumbnail)?.toString() ?: return null
         return CoverArtRef(url = url, cacheKey = cacheKey)
     }
 
@@ -96,7 +96,11 @@ class SourceRepository(
         sourceId: Long,
         coverArt: String,
         thumbnail: Boolean,
-    ): String? = if (activeSourceId == sourceId) active?.coverArtUri(coverArt, thumbnail)?.toString() else null
+    ): String? {
+        val source = active ?: return null
+        if (source.id != sourceId) return null
+        return source.coverArtUri(coverArt, thumbnail)?.toString()
+    }
 
     fun streamUri(
         songId: String,

app/src/test/kotlin/com/subtracks/data/download/ArtworkStoreTest.kt

@@ -10,6 +10,7 @@ import org.junit.Assert.assertTrue
 import org.junit.Test
 import org.junit.runner.RunWith
 import java.io.File
+import java.util.concurrent.CountDownLatch
 
 @RunWith(AndroidJUnit4::class)
 class ArtworkStoreTest {
@@ -52,6 +53,29 @@ class ArtworkStoreTest {
         assertTrue(store.file(2, "2:other:false").exists())
     }
 
+    @Test
+    fun concurrentWritesOfTheSameKeyLeaveAWholeFile() {
+        // Unequal lengths on purpose: two writers sharing a partial file interleave into a file
+        // that matches neither payload, which equal-length payloads would hide.
+        val long = ByteArray(1 shl 20) { 1 }
+        val short = ByteArray(1 shl 19) { 2 }
+        repeat(50) {
+            val go = CountDownLatch(1)
+            val writers =
+                listOf(long, short).map { payload ->
+                    Thread {
+                        go.await()
+                        store.write(1, "1:art-1:false", payload)
+                    }.apply { start() }
+                }
+            go.countDown()
+            writers.forEach { it.join() }
+
+            val written = store.file(1, "1:art-1:false").readBytes()
+            assertTrue("torn write: ${written.size} bytes", written.contentEquals(long) || written.contentEquals(short))
+        }
+    }
+
     @Test
     fun anUnwritableArtDirectoryIsReportedRatherThanThrown() {
         File(root, "1/art").apply {

app/src/test/kotlin/com/subtracks/data/download/OkHttpArtworkFetcherTest.kt

@@ -0,0 +1,62 @@
+package com.subtracks.data.download
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import kotlinx.coroutines.runBlocking
+import okhttp3.OkHttpClient
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.MockWebServer
+import org.junit.After
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import java.io.IOException
+
+@RunWith(AndroidJUnit4::class)
+class OkHttpArtworkFetcherTest {
+    private lateinit var server: MockWebServer
+
+    @Before
+    fun setUp() {
+        server = MockWebServer()
+        server.start()
+    }
+
+    @After
+    fun tearDown() {
+        server.shutdown()
+    }
+
+    private fun url() = server.url("/rest/getCoverArt?id=art-1").toString()
+
+    @Test
+    fun theResponseBodyComesBack() {
+        server.enqueue(MockResponse().setBody("artwork"))
+
+        val bytes = runBlocking { OkHttpArtworkFetcher(OkHttpClient()).fetch(url()) }
+
+        assertArrayEquals("artwork".toByteArray(), bytes)
+    }
+
+    @Test
+    fun aBodyLargerThanTheLimitIsRefused() {
+        server.enqueue(MockResponse().setBody("0123456789"))
+
+        val fetch =
+            runBlocking {
+                runCatching { OkHttpArtworkFetcher(OkHttpClient(), maxBytes = 4).fetch(url()) }
+            }
+
+        assertTrue("expected a refusal, got $fetch", fetch.exceptionOrNull() is IOException)
+    }
+
+    @Test
+    fun anUnsuccessfulResponseIsRefused() {
+        server.enqueue(MockResponse().setResponseCode(500))
+
+        val fetch = runBlocking { runCatching { OkHttpArtworkFetcher(OkHttpClient()).fetch(url()) } }
+
+        assertTrue("expected a refusal, got $fetch", fetch.exceptionOrNull() is IOException)
+    }
+}

app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt

@@ -381,6 +381,19 @@ class DownloadRepositoryTest {
             assertEquals(1, engine.requests.size)
         }
 
+    @Test
+    fun artworkSurvivesALibraryRowGoingMissing() =
+        runTest {
+            seedLibrary()
+            runBlocking { repository.download(1, "s1") }
+            assertNotNull(artwork.uri(1, coverArtKey(1, ALBUM_ART, false)))
+
+            runBlocking { db.libraryDao().deleteAlbums(1, listOf("al1")) }
+            runBlocking { repository.reconcile() }
+
+            assertNotNull(artwork.uri(1, coverArtKey(1, ALBUM_ART, false)))
+        }
+
     @Test
     fun removingTheLastDownloadOfAnAlbumRemovesItsArtwork() =
         runTest {