Hard/soft fail handling and release blacklist (#87)

crates/arr-daemon/src/grab.rs

@@ -13,13 +13,13 @@
 //! - the `grabs` row is written from that response, so a crash between the add
 //!   and the insert heals on the next tick instead of leaving an orphan.
 
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::path::PathBuf;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use arr_core::policy::{evaluate, Candidate};
 use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict};
-use arr_db::{Db, MoviePolicy};
+use arr_db::{blacklist, Blacklist, Db, MoviePolicy};
 use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
 use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
 
@@ -167,7 +167,7 @@ impl GrabAction {
         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 #24's; leaving the row alone
+                // or a manual removal is issue #86's; leaving the row alone
                 // keeps this tick from re-grabbing behind the operator.
                 continue;
             };
@@ -252,6 +252,7 @@ impl GrabAction {
         indexers: &[i64],
         loaded: &MoviePolicy,
         original_language: &Language,
+        blacklist: &Blacklist,
     ) -> Result<Vec<Eligible>, GrabError> {
         let request = SearchRequest::Text {
             query: search_query(movie),
@@ -274,6 +275,7 @@ impl GrabAction {
                 &loaded.policy,
                 &loaded.overrides,
                 original_language,
+                blacklist,
             )
             .await?;
             if let Some(candidate) = stored {
@@ -314,17 +316,24 @@ impl GrabAction {
         };
         let original_language = arr_db::policy::language(original_language);
 
-        let candidates = self
-            .search(database, movie, indexers, &loaded, &original_language)
-            .await?;
         // §6.3: anything that hard-failed post-ffprobe is never grabbed
-        // again. The same release reappears under new infohashes, so the key
-        // is the normalised name.
-        let blacklisted = blacklisted_names(database).await?;
-        let Some(winner) = candidates
-            .into_iter()
-            .find(|candidate| !blacklisted.contains(&arr_parse::normalise(&candidate.name)))
-        else {
+        // again. `store_release` already rejects a blacklisted candidate as
+        // it classifies it, so this second pass only catches a row that was
+        // classified before the blacklist entry existed.
+        let blacklist = Blacklist::load(database.pool()).await?;
+        let candidates = self
+            .search(
+                database,
+                movie,
+                indexers,
+                &loaded,
+                &original_language,
+                &blacklist,
+            )
+            .await?;
+        let Some(winner) = candidates.into_iter().find(|candidate| {
+            !blacklist.blocks_candidate(&candidate.name, &candidate.download_url)
+        }) else {
             tracing::info!(
                 movie_id = movie.id,
                 title = movie.title,
@@ -345,6 +354,15 @@ impl GrabAction {
             .await?;
         let infohash = added.hash.to_ascii_lowercase();
 
+        // §6.3's second key. A `.torrent` link hides its infohash until
+        // Transmission has fetched it, so the same blacklisted torrent can
+        // reach here under a new name.
+        if blacklist.blocks_infohash(&infohash) {
+            self.drop_blacklisted_torrent(database, movie, &winner, &added)
+                .await?;
+            return Ok(None);
+        }
+
         // A duplicate here is the restart case: the torrent was added before
         // the process died. `DO NOTHING` keeps the original row.
         let inserted = sqlx::query!(
@@ -390,6 +408,60 @@ impl GrabAction {
             format!("grabbed {} as grab {}", winner.name, inserted.id),
         )))
     }
+
+    /// Undo a grab whose infohash turned out to be blacklisted (§6.3).
+    ///
+    /// The name is added to the blacklist so the next tick stops at the cheap
+    /// check instead of paying Transmission again, and no `grabs` row is
+    /// written, which leaves the title a gap for the next candidate.
+    async fn drop_blacklisted_torrent(
+        &self,
+        database: &Db,
+        movie: &PendingMovie,
+        winner: &Eligible,
+        added: &arr_dl::AddedTorrent,
+    ) -> Result<(), GrabError> {
+        let release_name = &winner.name;
+        blacklist::add(
+            database.pool(),
+            None,
+            release_name,
+            "blacklisted infohash under a new name",
+        )
+        .await?;
+        // `store_release` classified this row before the infohash was known,
+        // so it still reads eligible. Correct it here rather than waiting for
+        // the next search to overwrite it: until then §9.3's manual view
+        // would keep offering a release this tick just refused.
+        sqlx::query!(
+            "UPDATE releases SET verdict = 'rejected', rejected_rule = ? WHERE id = ?",
+            blacklist::RULE,
+            winner.id
+        )
+        .execute(database.pool())
+        .await?;
+        if added.was_duplicate {
+            // The earlier grab's torrent, still working off its seeding
+            // obligation (§7.3). Nothing here deletes a torrent.
+            tracing::warn!(
+                movie_id = movie.id,
+                release = release_name,
+                infohash = added.hash,
+                "blacklisted torrent re-listed under a new name; left seeding"
+            );
+        } else {
+            // This tick added it seconds ago, so it carries no seeding
+            // obligation and has nothing on disk worth keeping.
+            self.transmission.remove_torrent(added.id, true).await?;
+            tracing::warn!(
+                movie_id = movie.id,
+                release = release_name,
+                infohash = added.hash,
+                "blacklisted torrent re-listed under a new name; removed"
+            );
+        }
+        Ok(())
+    }
 }
 
 impl Action for GrabAction {
@@ -471,6 +543,11 @@ async fn pending_movies(database: &Db) -> Result<Vec<PendingMovie>, GrabError> {
 ///
 /// Returns the candidate only when the release is eligible: automatic
 /// selection never takes a waiver (§9.3).
+///
+/// The blacklist is applied here rather than at selection so it reaches every
+/// trigger (§6.3: "including by RSS") and so the stored row says why — a
+/// blacklisted release is rejected under the `blacklisted` rule, which is
+/// what stops §9.3's manual view from offering it as a clean match.
 async fn store_release(
     database: &Db,
     movie_id: i64,
@@ -478,6 +555,7 @@ async fn store_release(
     policy: &Policy,
     overrides: &TitleOverrides,
     original_language: &Language,
+    blacklist: &Blacklist,
 ) -> Result<Option<Eligible>, GrabError> {
     let parsed = arr_parse::parse(&release.name);
     let evaluation = evaluate(
@@ -501,7 +579,11 @@ async fn store_release(
     } else {
         scored.source.saturating_add(scored.seeders)
     };
-    let (verdict, rule) = verdict_columns(&evaluation.verdict);
+    let (verdict, rule) = if blacklist.blocks_candidate(&release.name, &release.download_url) {
+        ("rejected", Some(blacklist::RULE.to_owned()))
+    } else {
+        verdict_columns(&evaluation.verdict)
+    };
     let parsed_json = serde_json::to_string(&parsed).map_err(|source| GrabError::Parsed {
         name: release.name.clone(),
         source,
@@ -565,17 +647,6 @@ async fn store_release(
     }))
 }
 
-/// Every blacklist key (§6.3), for filtering candidates. Household scale: a
-/// handful of rows, refetched per title rather than cached.
-async fn blacklisted_names(database: &Db) -> Result<HashSet<String>, GrabError> {
-    let names = sqlx::query_scalar!(
-        r#"SELECT normalised_name AS "normalised_name!: String" FROM blacklist"#
-    )
-    .fetch_all(database.pool())
-    .await?;
-    Ok(names.into_iter().collect())
-}
-
 /// Record that the title was searched, so the next tick takes a different one.
 async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> {
     sqlx::query!(
@@ -730,6 +801,17 @@ mod tests {
                         .collect();
                     success(&json!({"torrents": torrents}))
                 }
+                "torrent-remove" => {
+                    let removed: Vec<i64> = arguments["ids"]
+                        .as_array()
+                        .map(|ids| ids.iter().filter_map(serde_json::Value::as_i64).collect())
+                        .unwrap_or_default();
+                    self.torrents
+                        .lock()
+                        .unwrap()
+                        .retain(|torrent| !removed.contains(&torrent.id));
+                    success(&json!({}))
+                }
                 _ => success(&json!({})),
             }
         }
@@ -1022,6 +1104,75 @@ mod tests {
         );
     }
 
+    /// §6.3 reaches every trigger, so the exclusion is applied where a
+    /// release is classified rather than where one is picked: the cached row
+    /// says `blacklisted`, which is also what stops §9.3's manual view from
+    /// offering it as a clean match.
+    #[tokio::test]
+    async fn a_blacklisted_release_is_cached_as_rejected() {
+        let (_dir, database) = wanted_movie().await;
+        blacklist::add(
+            database.pool(),
+            Some("ffff"),
+            "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos",
+            "dolby_vision_profile",
+        )
+        .await
+        .unwrap();
+        let indexer = prowlarr().await;
+        let (downloader, _fake) = transmission().await;
+
+        action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        let (verdict, rule): (String, Option<String>) =
+            sqlx::query_as("SELECT verdict, rejected_rule FROM releases WHERE guid = 'good'")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(verdict, "rejected");
+        assert_eq!(rule.as_deref(), Some("blacklisted"));
+    }
+
+    /// §6.3's second key. A `.torrent` link hides its infohash until
+    /// Transmission has fetched it, so the blacklisted torrent is only
+    /// recognised after the add — and must not leave a grab behind.
+    #[tokio::test]
+    async fn a_blacklisted_infohash_never_becomes_a_grab() {
+        let (_dir, database) = wanted_movie().await;
+        // What the fake hands back for the first torrent it accepts.
+        blacklist::add(
+            database.pool(),
+            Some(&format!("{:040x}", 7)),
+            "Some.Older.Name.Of.The.Same.Torrent",
+            "required_audio",
+        )
+        .await
+        .unwrap();
+        let indexer = prowlarr().await;
+        let (downloader, fake) = transmission().await;
+
+        action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        assert!(grabs(&database).await.is_empty());
+        assert!(
+            fake.torrents().is_empty(),
+            "a torrent added this tick and then found blacklisted is removed"
+        );
+        // Recorded under its new name, so the next tick stops before paying
+        // Transmission again.
+        let blacklist = Blacklist::load(database.pool()).await.unwrap();
+        assert!(blacklist.blocks_name("Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos"));
+        // And the cached row stops reading eligible straight away, so §9.3's
+        // manual view never offers what this tick just refused.
+        let (verdict, rule): (String, Option<String>) =
+            sqlx::query_as("SELECT verdict, rejected_rule FROM releases WHERE guid = 'good'")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(verdict, "rejected");
+        assert_eq!(rule.as_deref(), Some("blacklisted"));
+    }
+
     /// §6.3: `blocked` stops targeted search for a title.
     #[tokio::test]
     async fn a_blocked_title_is_not_searched() {

crates/arr-daemon/src/import.rs

@@ -203,7 +203,7 @@ impl ImportAction {
 
         let Some(content) = self.transmission.torrent_content(&pending.infohash).await? else {
             // Gone from Transmission. Whether that is a failure or a manual
-            // removal is issue #24's call; leave the grab alone.
+            // removal is issue #86's call; leave the grab alone.
             tracing::warn!(
                 grab_id = pending.grab_id,
                 infohash = pending.infohash,
@@ -312,16 +312,12 @@ impl ImportAction {
         pending: &PendingImport,
         reason: &str,
     ) -> Result<Outcome, ImportError> {
-        let normalised = arr_parse::normalise(&pending.release_name);
-        sqlx::query!(
-            "INSERT INTO blacklist (infohash, normalised_name, reason)
-             VALUES (?, ?, ?)
-             ON CONFLICT (infohash) DO NOTHING",
-            pending.infohash,
-            normalised,
-            reason
+        arr_db::blacklist::add(
+            database.pool(),
+            Some(&pending.infohash),
+            &pending.release_name,
+            reason,
         )
-        .execute(database.pool())
         .await?;
         sqlx::query!(
             "UPDATE grabs SET state = 'failed' WHERE id = ?",
@@ -770,6 +766,60 @@ mod tests {
         );
     }
 
+    /// §5.7 soft fail: watchable but not what was asked. It imports, and the
+    /// row carries the relaxed rule — the file must never read as a clean
+    /// match. The torrent is untouched either way (§7.3).
+    #[tokio::test]
+    async fn an_english_only_kids_import_carries_a_waiver() {
+        let h = harness(HDR10_PROBE).await;
+        // The kids policy requires Portuguese audio; `allow_english_audio`
+        // turns that hard fail into a waiver (§5.2, §5.7).
+        let kids_library = h.library.join("kids");
+        std::fs::create_dir_all(&kids_library).unwrap();
+        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'movie' AND audience = 'kids'")
+            .bind(kids_library.to_string_lossy().into_owned())
+            .execute(h.database.pool())
+            .await
+            .unwrap();
+        sqlx::query(
+            r#"UPDATE movies
+               SET root_id   = (SELECT id FROM roots
+                                WHERE kind = 'movie' AND audience = 'kids'),
+                   overrides = '{"allow_english_audio":true}'
+               WHERE id = 1"#,
+        )
+        .execute(h.database.pool())
+        .await
+        .unwrap();
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let waiver: Option<String> = sqlx::query_scalar(
+            "SELECT json_extract(waiver, '$.rule') FROM media_files
+             WHERE owner_kind = 'movie' AND owner_id = 1",
+        )
+        .fetch_one(h.database.pool())
+        .await
+        .unwrap();
+        assert_eq!(waiver.as_deref(), Some("required_audio"));
+
+        let blacklisted: i64 = sqlx::query_scalar("SELECT count(*) FROM blacklist")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(blacklisted, 0, "a soft fail blacklists nothing");
+        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(grab_state, "imported");
+        assert!(
+            h.downloads.join("Dune/Dune.mkv").is_file(),
+            "§7.3: neither failure mode deletes the torrent"
+        );
+    }
+
     /// §8: killed between the hardlink and the bookkeeping, a restart
     /// converges instead of failing on the existing destination.
     #[tokio::test]