fix: reopen the gap for a grab whose torrent vanished (#98)

crates/arr-daemon/src/grab.rs

@@ -475,9 +475,10 @@ impl Grabber {
         let mut outcomes = Vec::new();
         for grab in sent {
             let Some(progress) = torrents.get(&grab.infohash.to_ascii_lowercase()) else {
-                // Gone from Transmission. Deciding whether that is a failure
-                // or a manual removal is issue #86's; leaving the row alone
-                // keeps this tick from re-grabbing behind the operator.
+                outcomes.push(
+                    self.vanish(database, grab.id, &grab.target_kind, grab.target_id)
+                        .await?,
+                );
                 continue;
             };
             if *progress < 1.0 {
@@ -503,6 +504,34 @@ impl Grabber {
         Ok(outcomes)
     }
 
+    /// §86: a `sent` grab whose infohash Transmission no longer reports —
+    /// removed by hand, not a policy failure. Marked `vanished` rather than
+    /// `failed` so it does not feed the `needs_decision` queue (attention.rs),
+    /// and the gap reopens with nothing blacklisted, since the release itself
+    /// never failed policy.
+    async fn vanish(
+        &self,
+        database: &Db,
+        grab_id: i64,
+        target_kind: &str,
+        target_id: i64,
+    ) -> Result<Outcome, GrabError> {
+        sqlx::query!("UPDATE grabs SET state = 'vanished' WHERE id = ?", grab_id)
+            .execute(database.pool())
+            .await?;
+        reopen_target(database, target_kind, target_id).await?;
+        tracing::warn!(
+            grab_id,
+            target_kind,
+            target_id,
+            "torrent vanished from Transmission; gap reopened"
+        );
+        Ok(Outcome::new(
+            format!("grab {grab_id} sent, torrent vanished from Transmission"),
+            format!("reopened {target_kind} {target_id}"),
+        ))
+    }
+
     async fn record_attempt(
         &self,
         database: &Db,
@@ -561,14 +590,23 @@ impl Grabber {
             return Ok(None);
         }
 
-        // A duplicate here is the restart case: the torrent was added before
-        // the process died. `DO NOTHING` keeps the original row.
+        // `infohash` is unique, so re-grabbing the same release conflicts
+        // with its own earlier row. A `sent`/`downloaded` row is the restart
+        // case and is left alone; a `vanished` one (§86) is reclaimed, since
+        // nothing blacklisted the release.
         let target_kind = target.scope.target_kind();
         let target_id = target.scope.target_id();
         let inserted = sqlx::query!(
             r#"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
                VALUES (?, ?, ?, ?, 'sent')
-               ON CONFLICT (infohash) DO NOTHING
+               ON CONFLICT (infohash) DO UPDATE SET
+                   release_id  = excluded.release_id,
+                   target_kind = excluded.target_kind,
+                   target_id   = excluded.target_id,
+                   state       = 'sent',
+                   grabbed_at  = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
+                   imported_at = NULL
+               WHERE grabs.state = 'vanished'
                RETURNING id AS "id!: i64""#,
             winner.id,
             target_kind,
@@ -1014,6 +1052,56 @@ pub(crate) async fn record_episode_search(
     Ok(())
 }
 
+/// §86: reopen the gap behind a grab that no longer counts as in flight —
+/// a movie or episode goes back to `missing`; a season pack reopens only the
+/// episodes it was still covering (mirrors `import::hard_fail_tv`'s season
+/// case), leaving ones already imported from a partial pack alone.
+pub(crate) async fn reopen_target(
+    database: &Db,
+    target_kind: &str,
+    target_id: i64,
+) -> Result<(), sqlx::Error> {
+    match target_kind {
+        "movie" => {
+            sqlx::query!(
+                "UPDATE movies SET state = 'missing',
+                        updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+                 WHERE id = ?",
+                target_id
+            )
+            .execute(database.pool())
+            .await?;
+        }
+        "episode" => {
+            sqlx::query!(
+                "UPDATE episodes SET state = 'missing',
+                        updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+                 WHERE id = ?",
+                target_id
+            )
+            .execute(database.pool())
+            .await?;
+        }
+        "season" => {
+            sqlx::query!(
+                "UPDATE episodes
+                 SET    state = 'missing',
+                        updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+                 WHERE  season_id = ? AND state = 'downloading'
+                   AND  NOT EXISTS (
+                            SELECT 1 FROM media_files f
+                            WHERE  f.owner_kind = 'episode' AND f.owner_id = episodes.id
+                        )",
+                target_id
+            )
+            .execute(database.pool())
+            .await?;
+        }
+        other => unreachable!("grabs.target_kind CHECK constraint excludes {other:?}"),
+    }
+    Ok(())
+}
+
 /// `movies-main`, `tv-kids` (§7.1). Distinct from Radarr's own labels, so
 /// both stacks can run against one Transmission.
 fn label(loaded: &MoviePolicy) -> String {
@@ -1521,6 +1609,62 @@ mod tests {
         assert_eq!(grabs(&database).await[0].2, "downloaded");
     }
 
+    /// §86: a torrent removed by hand — gone from Transmission before it
+    /// finished — reopens the gap instead of leaving the grab stuck forever,
+    /// and is marked `vanished` rather than `failed` so it never counts
+    /// toward the `needs_decision` queue (attention.rs). A search-empty
+    /// indexer on the second tick isolates the reopen from the re-grab that
+    /// would otherwise follow it in the same tick.
+    #[tokio::test]
+    async fn a_torrent_removed_by_hand_reopens_the_gap() {
+        let (_dir, database) = wanted_movie().await;
+        let indexer = prowlarr().await;
+        let (downloader, fake) = transmission().await;
+
+        action(&indexer, &downloader).tick(&database).await.unwrap();
+        fake.torrents.lock().unwrap().clear();
+        let empty_indexer = empty_prowlarr().await;
+        let outcomes = action(&empty_indexer, &downloader)
+            .tick(&database)
+            .await
+            .unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert_eq!(grabs(&database).await[0].2, "vanished");
+        let state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(state, "missing");
+    }
+
+    /// §86 continued: nothing blacklists the vanished release, so the exact
+    /// same infohash can be grabbed again — reclaiming its own dead row
+    /// rather than silently losing the grab to the `infohash` uniqueness
+    /// constraint.
+    #[tokio::test]
+    async fn a_reopened_gap_can_regrab_the_same_infohash() {
+        let (_dir, database) = wanted_movie().await;
+        let indexer = prowlarr().await;
+        let (downloader, fake) = transmission().await;
+        let action = action(&indexer, &downloader);
+
+        action.tick(&database).await.unwrap();
+        fake.torrents.lock().unwrap().clear();
+        let outcomes = action.tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 2, "vanish, then a fresh grab");
+        assert_eq!(fake.torrents().len(), 1, "same magnet, same infohash");
+        let grabs = grabs(&database).await;
+        assert_eq!(grabs.len(), 1, "the dead row is reclaimed, not duplicated");
+        assert_eq!(grabs[0].2, "sent");
+        let state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(state, "downloading");
+    }
+
     /// §5.2: the language rules are expressed against the title's original
     /// language, and guessing it is worse than waiting for it.
     #[tokio::test]

crates/arr-daemon/src/import.rs

@@ -283,7 +283,7 @@ impl ImportAction {
             .torrent_paths(pending.grab_id, &pending.infohash)
             .await?
         else {
-            return Ok(None);
+            return self.vanish(database, pending).await.map(Some);
         };
 
         // No expected runtime yet: the movies table carries no TMDB runtime,
@@ -398,8 +398,8 @@ impl ImportAction {
         infohash: &str,
     ) -> Result<Option<Vec<PathBuf>>, ImportError> {
         let Some(content) = self.transmission.torrent_content(infohash).await? else {
-            // Gone from Transmission. Whether that is a failure or a manual
-            // removal is issue #86's call; leave the grab alone.
+            // Gone from Transmission — the caller marks the grab vanished
+            // and reopens the gap (§86).
             tracing::warn!(
                 grab_id,
                 infohash,
@@ -461,7 +461,7 @@ impl ImportAction {
             .torrent_paths(pending.grab_id, &pending.infohash)
             .await?
         else {
-            return Ok(None);
+            return self.vanish_tv(database, pending).await.map(Some);
         };
         let candidates = self.probe_all(&paths).await?;
         let assignments = assign_files(pending, &episodes, candidates);
@@ -660,6 +660,70 @@ impl ImportAction {
         ))
     }
 
+    /// §86: a `downloaded` grab whose torrent Transmission no longer reports
+    /// — removed by hand, not a policy failure. Marked `vanished` rather
+    /// than `failed` so it does not feed the `needs_decision` queue
+    /// (attention.rs), and nothing is blacklisted, since the release itself
+    /// never failed policy.
+    async fn vanish(&self, database: &Db, pending: &PendingImport) -> Result<Outcome, ImportError> {
+        sqlx::query!(
+            "UPDATE grabs SET state = 'vanished' WHERE id = ?",
+            pending.grab_id
+        )
+        .execute(database.pool())
+        .await?;
+        crate::grab::reopen_target(database, "movie", pending.movie_id).await?;
+
+        tracing::warn!(
+            grab_id = pending.grab_id,
+            movie_id = pending.movie_id,
+            title = pending.title,
+            release = pending.release_name,
+            "torrent vanished from Transmission; gap reopened"
+        );
+        Ok(Outcome::new(
+            format!(
+                "grab {} downloaded, torrent vanished from Transmission",
+                pending.grab_id
+            ),
+            format!("reopened movie {}", pending.movie_id),
+        ))
+    }
+
+    /// TV counterpart of [`Self::vanish`]: reopens the episode, or the
+    /// still-downloading episodes of a season pack.
+    async fn vanish_tv(
+        &self,
+        database: &Db,
+        pending: &PendingTvImport,
+    ) -> Result<Outcome, ImportError> {
+        sqlx::query!(
+            "UPDATE grabs SET state = 'vanished' WHERE id = ?",
+            pending.grab_id
+        )
+        .execute(database.pool())
+        .await?;
+        let (target_kind, target_id) = match pending.episode_id {
+            Some(episode_id) => ("episode", episode_id),
+            None => ("season", pending.season_id),
+        };
+        crate::grab::reopen_target(database, target_kind, target_id).await?;
+
+        tracing::warn!(
+            grab_id = pending.grab_id,
+            series = pending.series_title,
+            release = pending.release_name,
+            "torrent vanished from Transmission; gap reopened"
+        );
+        Ok(Outcome::new(
+            format!(
+                "grab {} downloaded, torrent vanished from Transmission",
+                pending.grab_id
+            ),
+            format!("reopened {target_kind} {target_id}"),
+        ))
+    }
+
     /// §7.5: the filesystem watcher misses the just-hardlinked file. A
     /// failure to reach Jellyfin must not fail the import, which has already
     /// succeeded.
@@ -1627,6 +1691,79 @@ mod tests {
         assert!(outcomes.is_empty());
     }
 
+    /// §86: a `downloaded` grab whose torrent Transmission no longer reports
+    /// — removed by hand, not a policy failure — is marked `vanished` and
+    /// reopens the movie as a gap, without touching the blacklist.
+    #[tokio::test]
+    async fn a_vanished_downloaded_grab_reopens_the_gap() {
+        let dir = tempfile::tempdir().unwrap();
+        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
+        database.migrate().await.unwrap();
+        sqlx::query(
+            "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, state)
+             SELECT 693134, 'Dune: Part Two', 2024, 'en', id, 'downloading'
+             FROM roots WHERE kind = 'movie' AND audience = 'main'",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+        let release_id = sqlx::query(
+            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
+             VALUES (7, 'good', ?, 23622320128, 'magnet:x', '{}', 'eligible')",
+        )
+        .bind(RELEASE_NAME)
+        .execute(database.pool())
+        .await
+        .unwrap()
+        .last_insert_rowid();
+        sqlx::query(
+            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
+             VALUES (?, 'movie', 1, ?, 'downloaded')",
+        )
+        .bind(release_id)
+        .bind(INFOHASH)
+        .execute(database.pool())
+        .await
+        .unwrap();
+
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "result": "success",
+                "arguments": {"torrents": []}
+            })))
+            .mount(&server)
+            .await;
+        let jellyfin_server = start_jellyfin_server().await;
+        let ntfy_server = start_ntfy_server().await;
+        let action = ImportAction::new(
+            TransmissionClient::new(&server.uri()).unwrap(),
+            Prober::new().with_binary(fake_ffprobe(dir.path(), HDR10_PROBE)),
+            JellyfinClient::new(jellyfin_server.uri(), None).unwrap(),
+            Notifier::new(ntfy_server.uri()).unwrap(),
+            Some("operator-topic".to_string()),
+        );
+
+        let outcomes = action.tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs WHERE id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(grab_state, "vanished");
+        let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(movie_state, "missing");
+        let blacklisted: i64 = sqlx::query_scalar("SELECT count(*) FROM blacklist")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(blacklisted, 0, "a vanished torrent is not a policy failure");
+    }
+
     /// Torrent-declared names are untrusted: absolute and `..`-carrying
     /// entries are skipped, and the import proceeds from what remains.
     #[tokio::test]

crates/arr-db/migrations/0011_vanished_grabs.sql

@@ -0,0 +1,25 @@
+-- #86: a grab whose torrent Transmission no longer reports (manual removal,
+-- not a policy failure) gets its own state. Distinct from 'failed' so it
+-- does not feed the needs_decision attention queue (attention.rs), which
+-- counts 'failed' as a hard-fail signal. SQLite cannot alter a CHECK, so the
+-- table is rebuilt (see 0010).
+CREATE TABLE grabs_new (
+    id          INTEGER PRIMARY KEY,
+    release_id  INTEGER NOT NULL REFERENCES releases (id),
+    target_kind TEXT NOT NULL CHECK (target_kind IN ('movie', 'episode', 'season')),
+    target_id   INTEGER NOT NULL,
+    infohash    TEXT NOT NULL UNIQUE,
+    state       TEXT NOT NULL DEFAULT 'sent'
+                CHECK (state IN ('sent', 'downloaded', 'imported', 'failed', 'vanished')),
+    grabbed_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    imported_at TEXT
+) STRICT;
+
+INSERT INTO grabs_new (id, release_id, target_kind, target_id, infohash, state, grabbed_at, imported_at)
+SELECT id, release_id, target_kind, target_id, infohash, state, grabbed_at, imported_at FROM grabs;
+
+DROP TABLE grabs;
+ALTER TABLE grabs_new RENAME TO grabs;
+
+CREATE INDEX grabs_state ON grabs (state);
+CREATE INDEX grabs_target ON grabs (target_kind, target_id);