Protect a show's Tier 3 pending episode from retention reclaim

Peter's call (option 1 of the three in docs/detection-findings.md): exempt
an episode retention would otherwise reclaim while it is a show's pending
comparison basis, rather than extending the grace period or accepting the
gap.

RetentionSweeper gains a pendingEpisodeIds seam (suspend () -> Set<String>,
defaulting to empty) and filters its reclaim candidates against it before
the grace-period policy ever sees them -- every existing caller and test
was unaffected by the default. RetentionWorker wires the real thing: the
union of two new, narrow DAO queries --
ShowFingerprintDao.getAllPendingEpisodeIds() (intro/outro) and
MidrollDao.getAllMidrollPendingEpisodeIds() (midroll, reading
SubscriptionEntity.midrollPendingEpisodeId).

Applied to both fingerprinting mechanisms, not just the one that got
diagnosed: MidrollUpdater's discovery has the identical rolling-pending-
episode shape and would race retention the same way for a fast show, just
silently, with no frozen row to reveal it.

Not yet re-verified on a device -- the next live test is two Up First
downloads close enough together to actually compare, confirming
pendingEpisodeId advances to a real second episode instead of restarting.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Sj3CoWEedM56bebByKgNxA

5 files changed, 80 insertions, 3 deletions

diff --git a/app/src/main/java/org/terst/listen/data/Daos.kt b/app/src/main/java/org/terst/listen/data/Daos.kt
index dec1ba0..2ca227c 100644
--- a/app/src/main/java/org/terst/listen/data/Daos.kt
+++ b/app/src/main/java/org/terst/listen/data/Daos.kt
@@ -354,6 +354,14 @@ interface ShowFingerprintDao {
 
     @Query("UPDATE episodes SET detectedOutroMs = :ms WHERE id = :id")
     suspend fun setDetectedOutroMs(id: String, ms: Long)
+
+    /**
+     * Every episode id any show currently relies on as its pending intro/outro comparison basis,
+     * across both kinds. [org.terst.listen.work.RetentionSweeper] reads this to avoid reclaiming
+     * one out from under detection -- see its own doc comment for why that race matters.
+     */
+    @Query("SELECT DISTINCT pendingEpisodeId FROM show_fingerprints WHERE pendingEpisodeId IS NOT NULL")
+    suspend fun getAllPendingEpisodeIds(): List<String>
 }
 
 /**
@@ -394,6 +402,15 @@ interface MidrollDao {
 
     @Query("UPDATE episodes SET midrollCheckedAtMs = :ms WHERE id = :id")
     suspend fun setMidrollCheckedAtMs(id: String, ms: Long)
+
+    /**
+     * Every episode id any show currently relies on as its midroll-discovery pending basis.
+     * [org.terst.listen.work.RetentionSweeper] reads this for the same reason it reads
+     * [ShowFingerprintDao.getAllPendingEpisodeIds] -- reclaiming this episode's file defeats
+     * discovery the same way, silently, before the next download ever gets compared against it.
+     */
+    @Query("SELECT DISTINCT midrollPendingEpisodeId FROM subscriptions WHERE midrollPendingEpisodeId IS NOT NULL")
+    suspend fun getAllMidrollPendingEpisodeIds(): List<String>
 }
 
 /**
diff --git a/app/src/main/java/org/terst/listen/work/RetentionSweeper.kt b/app/src/main/java/org/terst/listen/work/RetentionSweeper.kt
index cd329b3..fd80379 100644
--- a/app/src/main/java/org/terst/listen/work/RetentionSweeper.kt
+++ b/app/src/main/java/org/terst/listen/work/RetentionSweeper.kt
@@ -16,13 +16,26 @@ class RetentionSweeper(
     private val policy: RetentionPolicy,
     /** Injected so the undeletable-file path can be exercised without a read-only filesystem. */
     private val deleteFile: (File) -> Boolean = { it.delete() },
+    /**
+     * Episode ids Tier 3 is currently relying on as a show's pending comparison basis, across
+     * both fingerprinting mechanisms -- `ShowFingerprintEntity.pendingEpisodeId` and
+     * `SubscriptionEntity.midrollPendingEpisodeId`.
+     *
+     * A lambda rather than the DAOs themselves: this class only needs the resulting id set, not
+     * two more collaborators to know about, and a real caller composes it from wherever those
+     * ids actually live. Reclaiming one of these out from under detection is the exact race that
+     * left Up First's detector discarding its only candidate and restarting forever instead of
+     * ever comparing two episodes -- see docs/detection-findings.md.
+     */
+    private val pendingEpisodeIds: suspend () -> Set<String> = { emptySet() },
 ) {
 
     /** Returns how many episodes were reclaimed. */
     suspend fun sweep(nowMs: Long): Int {
         // The query is a coarse filter; the policy makes the decision, including cases SQL
         // cannot express such as an episode resumed after being marked played.
-        val candidates = episodes.playedDownloadsBefore(nowMs)
+        val protected = pendingEpisodeIds()
+        val candidates = episodes.playedDownloadsBefore(nowMs).filterNot { it.id in protected }
         var reclaimed = 0
 
         for (episode in policy.selectDeletable(candidates, nowMs)) {
diff --git a/app/src/main/java/org/terst/listen/work/Workers.kt b/app/src/main/java/org/terst/listen/work/Workers.kt
index 87f6814..9c48db0 100644
--- a/app/src/main/java/org/terst/listen/work/Workers.kt
+++ b/app/src/main/java/org/terst/listen/work/Workers.kt
@@ -65,8 +65,14 @@ class RetentionWorker(context: Context, params: WorkerParameters) : CoroutineWor
 
     override suspend fun doWork(): Result {
         val graph = Graph.get(applicationContext)
-        val reclaimed = RetentionSweeper(graph.episodes, graph.retention)
-            .sweep(System.currentTimeMillis())
+        val reclaimed = RetentionSweeper(
+            graph.episodes,
+            graph.retention,
+            pendingEpisodeIds = {
+                (graph.fingerprints.getAllPendingEpisodeIds() + graph.midrolls.getAllMidrollPendingEpisodeIds())
+                    .toSet()
+            },
+        ).sweep(System.currentTimeMillis())
         return Result.success(androidx.work.Data.Builder().putInt("reclaimed", reclaimed).build())
     }
 }
diff --git a/app/src/test/java/org/terst/listen/data/FakeDaos.kt b/app/src/test/java/org/terst/listen/data/FakeDaos.kt
index 8b00c3e..14d1394 100644
--- a/app/src/test/java/org/terst/listen/data/FakeDaos.kt
+++ b/app/src/test/java/org/terst/listen/data/FakeDaos.kt
@@ -310,6 +310,9 @@ class FakeShowFingerprintDao : ShowFingerprintDao {
         val index = episodeRows.indexOfFirst { it.id == id }
         if (index >= 0) episodeRows[index] = transform(episodeRows[index])
     }
+
+    override suspend fun getAllPendingEpisodeIds() =
+        rows.mapNotNull { it.pendingEpisodeId }.distinct()
 }
 
 class FakeMidrollDao : MidrollDao {
@@ -320,6 +323,9 @@ class FakeMidrollDao : MidrollDao {
      *  runs against the real `episodes` table. */
     var episodeRows: MutableList<EpisodeEntity> = mutableListOf()
 
+    /** Backs [getAllMidrollPendingEpisodeIds], which reads the real `subscriptions` table. */
+    var subscriptionRows: MutableList<SubscriptionEntity> = mutableListOf()
+
     override suspend fun getCandidates(subscriptionId: String) =
         candidateRows.filter { it.subscriptionId == subscriptionId }
 
@@ -348,6 +354,9 @@ class FakeMidrollDao : MidrollDao {
         val index = episodeRows.indexOfFirst { it.id == id }
         if (index >= 0) episodeRows[index] = episodeRows[index].copy(midrollCheckedAtMs = ms)
     }
+
+    override suspend fun getAllMidrollPendingEpisodeIds() =
+        subscriptionRows.mapNotNull { it.midrollPendingEpisodeId }.distinct()
 }
 
 class FakeQueueDao : QueueDao {
diff --git a/app/src/test/java/org/terst/listen/work/RetentionSweeperTest.kt b/app/src/test/java/org/terst/listen/work/RetentionSweeperTest.kt
index 65b739f..f587170 100644
--- a/app/src/test/java/org/terst/listen/work/RetentionSweeperTest.kt
+++ b/app/src/test/java/org/terst/listen/work/RetentionSweeperTest.kt
@@ -141,4 +141,36 @@ class RetentionSweeperTest {
         assertEquals(0, off.sweep(nowMs = 100 * day))
         assertTrue(File(episode.localPath!!).exists())
     }
+
+    /**
+     * The race that left Up First's detector restarting forever instead of ever comparing two
+     * episodes: a show's Tier 3 pending episode reclaimed before the next download arrives to
+     * compare it against. See docs/detection-findings.md.
+     */
+    @Test
+    fun `an episode still relied on as a Tier 3 pending basis is not reclaimed`() = runTest {
+        val episode = downloaded("a", playedAtMs = 0)
+        val protective = RetentionSweeper(
+            dao,
+            RetentionPolicy(gracePeriodMs = day),
+            pendingEpisodeIds = { setOf("a") },
+        )
+
+        assertEquals(0, protective.sweep(nowMs = 2 * day))
+        assertTrue(File(episode.localPath!!).exists())
+        assertEquals(DownloadState.DOWNLOADED, dao.rows.single().downloadState)
+    }
+
+    @Test
+    fun `an episode past its grace period but not pending anywhere is still reclaimed`() = runTest {
+        val episode = downloaded("a", playedAtMs = 0)
+        val protective = RetentionSweeper(
+            dao,
+            RetentionPolicy(gracePeriodMs = day),
+            pendingEpisodeIds = { setOf("some-other-episode") },
+        )
+
+        assertEquals(1, protective.sweep(nowMs = 2 * day))
+        assertTrue(!File(episode.localPath!!).exists())
+    }
 }