feat(daemon): reap seeded torrents

Justfile

@@ -34,8 +34,10 @@ e2e:
 
 # Throwaway local Transmission for `just e2e`, mirroring the CI service.
 e2e-up:
+    mkdir -p target/e2e-downloads
     docker run --rm -d --name arr-e2e-transmission \
         -e PUID=1000 -e PGID=1000 -p 9091:9091 \
+        -v "${PWD}/target/e2e-downloads:${PWD}/target/e2e-downloads" \
         linuxserver/transmission:latest
 
 # Stop and discard the local Transmission container.

crates/arr-daemon/src/config.rs

@@ -8,6 +8,7 @@
 //! `deny_unknown_fields` turns an accidental key in the file into a parse
 //! error instead of silently ignoring it.
 
+use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 
@@ -58,6 +59,8 @@ pub enum ConfigError {
     },
     #[error("invalid {env} ({input:?}): expected a number")]
     InvalidNumber { env: &'static str, input: String },
+    #[error("invalid tracker_seeding key {0:?}: expected a Prowlarr indexer ID")]
+    InvalidTrackerId(String),
 }
 
 /// On-disk representation. Non-secret fields only — see the module docs.
@@ -81,11 +84,20 @@ struct ConfigFile {
     #[serde(default)]
     seed_idle_limit_minutes: Option<u64>,
     #[serde(default)]
+    tracker_seeding: HashMap<String, TrackerSeedingRule>,
+    #[serde(default)]
     jellyfin_url: Option<String>,
     #[serde(default)]
     ntfy_url: Option<String>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TrackerSeedingRule {
+    pub ratio: f64,
+    pub min_seed_time: u64,
+}
+
 impl ConfigFile {
     fn load(path: &Path) -> Result<Self, ConfigError> {
         if !path.exists() {
@@ -153,6 +165,7 @@ pub struct Config {
     pub download_dir: PathBuf,
     pub seed_ratio_limit: f64,
     pub seed_idle_limit_minutes: u64,
+    pub tracker_seeding: HashMap<i64, TrackerSeedingRule>,
     pub tmdb_api_key: Option<String>,
     /// E2E seam only, env-only. `None` means the client's built-in TMDB
     /// address; DESIGN.md §10 keeps the real URL out of configuration.
@@ -182,6 +195,15 @@ impl Config {
                 .bind_addr
                 .unwrap_or_else(|| DEFAULT_BIND_ADDR.parse().expect("valid default bind_addr")),
         };
+        let tracker_seeding = file
+            .tracker_seeding
+            .iter()
+            .map(|(id, &rule)| {
+                id.parse::<i64>()
+                    .map(|id| (id, rule))
+                    .map_err(|_| ConfigError::InvalidTrackerId(id.clone()))
+            })
+            .collect::<Result<HashMap<_, _>, _>>()?;
 
         Ok(Self {
             bind_addr,
@@ -219,6 +241,7 @@ impl Config {
                     .seed_idle_limit_minutes
                     .unwrap_or(DEFAULT_SEED_IDLE_LIMIT_MINUTES),
             },
+            tracker_seeding,
             tmdb_api_key: env.tmdb_api_key,
             tmdb_url: env.tmdb_url,
             jellyfin_url: env
@@ -269,6 +292,7 @@ mod tests {
             DEFAULT_SEED_IDLE_LIMIT_MINUTES
         );
         assert_eq!(config.jellyfin_url, DEFAULT_JELLYFIN_URL);
+        assert!(config.tracker_seeding.is_empty());
         assert_eq!(config.ntfy_url, DEFAULT_NTFY_URL);
         assert_eq!(config.prowlarr_api_key, None);
         assert_eq!(config.tmdb_api_key, None);
@@ -418,6 +442,30 @@ prowlarr_url = "http://prowlarr.internal:9696"
         ));
     }
 
+    #[test]
+    fn tracker_seeding_rules_are_keyed_by_prowlarr_id() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("arr.toml");
+        std::fs::write(
+            &path,
+            "[tracker_seeding.3]\nratio = 2.5\nmin_seed_time = 120\n",
+        )
+        .unwrap();
+        let config = Config::resolve(EnvOverrides {
+            config_file: Some(path.to_string_lossy().into_owned()),
+            ..EnvOverrides::default()
+        })
+        .unwrap();
+
+        assert_eq!(
+            config.tracker_seeding.get(&3),
+            Some(&TrackerSeedingRule {
+                ratio: 2.5,
+                min_seed_time: 120
+            })
+        );
+    }
+
     #[test]
     fn invalid_env_bind_addr_returns_diagnostic() {
         let env = EnvOverrides {

crates/arr-daemon/src/grab.rs

@@ -50,6 +50,26 @@ pub struct SeedingLimits {
     pub idle_minutes: u64,
 }
 
+#[derive(Debug, Clone)]
+pub struct SeedingRules {
+    default: SeedingLimits,
+    trackers: HashMap<i64, SeedingLimits>,
+}
+
+impl SeedingRules {
+    #[must_use]
+    pub fn new(default: SeedingLimits, trackers: HashMap<i64, SeedingLimits>) -> Self {
+        Self { default, trackers }
+    }
+
+    fn for_indexer(&self, indexer_id: i64) -> SeedingLimits {
+        self.trackers
+            .get(&indexer_id)
+            .copied()
+            .unwrap_or(self.default)
+    }
+}
+
 /// A failure during one grab tick.
 #[derive(Debug, thiserror::Error)]
 pub enum GrabError {
@@ -84,7 +104,7 @@ pub struct GrabAction {
     prowlarr: ProwlarrClient,
     transmission: TransmissionClient,
     download_dir: PathBuf,
-    seeding: SeedingLimits,
+    seeding: SeedingRules,
     indexers: tokio::sync::RwLock<IndexerCache>,
     /// [`INDEXER_DISCOVERY_TIMEOUT`], overridden by tests that cannot wait
     /// out the real one. Mirrors `ReconcileLoop`'s action timeout override.
@@ -97,7 +117,7 @@ impl GrabAction {
         prowlarr: ProwlarrClient,
         transmission: TransmissionClient,
         download_dir: PathBuf,
-        seeding: SeedingLimits,
+        seeding: SeedingRules,
     ) -> Self {
         Self {
             prowlarr,
@@ -342,14 +362,15 @@ impl GrabAction {
             return Ok(None);
         };
 
+        let seeding = self.seeding.for_indexer(winner.indexer_id);
         let added = self
             .transmission
             .add_torrent(AddTorrent {
                 source: torrent_source(&winner.download_url),
                 label: label(&loaded),
                 download_dir: self.download_dir.clone(),
-                seed_ratio_limit: self.seeding.ratio,
-                seed_idle_limit_minutes: self.seeding.idle_minutes,
+                seed_ratio_limit: seeding.ratio,
+                seed_idle_limit_minutes: seeding.idle_minutes,
             })
             .await?;
         let infohash = added.hash.to_ascii_lowercase();
@@ -901,13 +922,32 @@ mod tests {
             ProwlarrClient::new(prowlarr.uri(), "key").unwrap(),
             TransmissionClient::new(&transmission.uri()).unwrap(),
             PathBuf::from("/mnt/media/transmission/complete"),
-            SeedingLimits {
-                ratio: 1.5,
-                idle_minutes: 60,
-            },
+            SeedingRules::new(
+                SeedingLimits {
+                    ratio: 1.5,
+                    idle_minutes: 60,
+                },
+                HashMap::new(),
+            ),
         )
     }
 
+    #[test]
+    fn seeding_rules_select_by_prowlarr_indexer_id() {
+        let default = SeedingLimits {
+            ratio: 1.0,
+            idle_minutes: 60,
+        };
+        let tracker = SeedingLimits {
+            ratio: 2.5,
+            idle_minutes: 120,
+        };
+        let rules = SeedingRules::new(default, HashMap::from([(7, tracker)]));
+
+        assert_eq!(rules.for_indexer(7), tracker);
+        assert_eq!(rules.for_indexer(8), default);
+    }
+
     async fn grabs(database: &Db) -> Vec<(i64, String, String)> {
         sqlx::query_as::<_, (i64, String, String)>(
             "SELECT target_id, infohash, state FROM grabs ORDER BY id",

crates/arr-daemon/src/main.rs

@@ -3,6 +3,7 @@
 mod config;
 mod grab;
 mod import;
+mod reaper;
 pub mod reconcile;
 mod web;
 
@@ -14,8 +15,9 @@ use arr_compat::CompatState;
 use arr_db::Db;
 use arr_meta::TmdbClient;
 use config::Config;
-use grab::{GrabAction, SeedingLimits};
+use grab::{GrabAction, SeedingLimits, SeedingRules};
 use import::ImportAction;
+use reaper::ReaperAction;
 use reconcile::{ReconcileLoop, Tick};
 use tower_http::trace::TraceLayer;
 
@@ -103,10 +105,25 @@ async fn run() -> Result<(), Error> {
                 prowlarr,
                 transmission.clone(),
                 config.download_dir.clone(),
-                SeedingLimits {
-                    ratio: config.seed_ratio_limit,
-                    idle_minutes: config.seed_idle_limit_minutes,
-                },
+                SeedingRules::new(
+                    SeedingLimits {
+                        ratio: config.seed_ratio_limit,
+                        idle_minutes: config.seed_idle_limit_minutes,
+                    },
+                    config
+                        .tracker_seeding
+                        .iter()
+                        .map(|(&id, rule)| {
+                            (
+                                id,
+                                SeedingLimits {
+                                    ratio: rule.ratio,
+                                    idle_minutes: rule.min_seed_time,
+                                },
+                            )
+                        })
+                        .collect(),
+                ),
             ),
         );
     } else {
@@ -116,8 +133,9 @@ async fn run() -> Result<(), Error> {
     // imported on this tick.
     reconcile = reconcile.register(
         Tick::Reconcile,
-        ImportAction::new(transmission, arr_probe::Prober::new()),
+        ImportAction::new(transmission.clone(), arr_probe::Prober::new()),
     );
+    reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission));
 
     // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
     // needs its own TMDB client for `movie/lookup`.

crates/arr-daemon/src/reaper.rs

@@ -0,0 +1,140 @@
+//! Removes arr torrents only after Transmission says their seeding obligation
+//! is complete. The library/import state is deliberately not consulted (§7.3).
+
+use arr_db::Db;
+use arr_dl::TransmissionClient;
+
+use crate::reconcile::{Action, ActionFuture, Outcome};
+
+#[derive(Debug)]
+pub struct ReaperAction {
+    transmission: TransmissionClient,
+}
+
+impl ReaperAction {
+    #[must_use]
+    pub fn new(transmission: TransmissionClient) -> Self {
+        Self { transmission }
+    }
+
+    async fn tick(&self) -> Result<Vec<Outcome>, arr_dl::Error> {
+        let torrents = self.transmission.list_torrents().await?;
+        let mut outcomes = Vec::new();
+        for torrent in torrents {
+            if !torrent.is_finished || !torrent.labels.iter().any(|label| is_arr_label(label)) {
+                continue;
+            }
+            self.transmission.remove_torrent(torrent.id, true).await?;
+            outcomes.push(Outcome::new(
+                format!("torrent {} finished seeding", torrent.hash),
+                format!("removed torrent {} and its download data", torrent.id),
+            ));
+        }
+        Ok(outcomes)
+    }
+}
+
+impl Action for ReaperAction {
+    fn name(&self) -> &'static str {
+        "torrent-reaper"
+    }
+
+    fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> {
+        Box::pin(async move {
+            self.tick()
+                .await
+                .map_err(|error| Box::new(error) as crate::reconcile::ActionError)
+        })
+    }
+}
+
+fn is_arr_label(label: &str) -> bool {
+    matches!(label, "movies-main" | "movies-kids" | "tv-main" | "tv-kids")
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use serde_json::{json, Value};
+    use wiremock::matchers::any;
+    use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
+
+    use super::*;
+
+    #[derive(Clone)]
+    struct Transmission {
+        torrents: Value,
+        removed: Arc<Mutex<Vec<Value>>>,
+    }
+
+    impl Respond for Transmission {
+        fn respond(&self, request: &Request) -> ResponseTemplate {
+            let body: Value = serde_json::from_slice(&request.body).unwrap();
+            let arguments = match body["method"].as_str().unwrap() {
+                "torrent-get" => json!({"torrents": self.torrents}),
+                "torrent-remove" => {
+                    self.removed.lock().unwrap().push(body["arguments"].clone());
+                    json!({})
+                }
+                method => panic!("unexpected method {method}"),
+            };
+            ResponseTemplate::new(200)
+                .insert_header("x-transmission-session-id", "session")
+                .set_body_json(json!({"result": "success", "arguments": arguments}))
+        }
+    }
+
+    fn torrent(id: i64, label: &str, finished: bool) -> Value {
+        json!({
+            "id": id, "name": format!("torrent-{id}"), "hashString": format!("hash-{id}"),
+            "status": if finished { 0 } else { 6 }, "percentDone": 1.0,
+            "downloadDir": "/downloads", "labels": [label], "isFinished": finished
+        })
+    }
+
+    #[tokio::test]
+    async fn removes_only_finished_arr_torrents_with_data() {
+        let server = MockServer::start().await;
+        let removed = Arc::new(Mutex::new(Vec::new()));
+        Mock::given(any())
+            .respond_with(Transmission {
+                torrents: json!([
+                    torrent(1, "movies-main", false),
+                    torrent(2, "movies-main", true),
+                    torrent(3, "radarr", true)
+                ]),
+                removed: Arc::clone(&removed),
+            })
+            .mount(&server)
+            .await;
+        let action = ReaperAction::new(TransmissionClient::new(&server.uri()).unwrap());
+
+        let outcomes = action.tick().await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert_eq!(
+            *removed.lock().unwrap(),
+            [json!({"ids": [2], "delete-local-data": true})]
+        );
+    }
+
+    #[tokio::test]
+    async fn finished_torrent_does_not_need_a_grab_or_import_row() {
+        let server = MockServer::start().await;
+        let removed = Arc::new(Mutex::new(Vec::new()));
+        Mock::given(any())
+            .respond_with(Transmission {
+                torrents: json!([torrent(9, "movies-kids", true)]),
+                removed: Arc::clone(&removed),
+            })
+            .mount(&server)
+            .await;
+        let action = ReaperAction::new(TransmissionClient::new(&server.uri()).unwrap());
+
+        action.tick().await.unwrap();
+
+        assert_eq!(removed.lock().unwrap().len(), 1);
+    }
+}

crates/arr-dl/src/lib.rs

@@ -68,6 +68,9 @@ pub struct Torrent {
     pub progress: f64,
     pub download_dir: PathBuf,
     pub labels: Vec<String>,
+    /// Transmission has stopped this torrent because its configured seeding
+    /// ratio or idle limit was reached.
+    pub is_finished: bool,
 }
 
 /// One file inside a torrent, as Transmission reports it.
@@ -225,7 +228,7 @@ impl TransmissionClient {
                 json!({
                     "fields": [
                         "id", "name", "hashString", "status", "percentDone",
-                        "downloadDir", "labels"
+                        "downloadDir", "labels", "isFinished"
                     ]
                 }),
             )
@@ -383,6 +386,8 @@ struct RpcTorrent {
     download_dir: PathBuf,
     #[serde(default)]
     labels: Vec<String>,
+    #[serde(rename = "isFinished", default)]
+    is_finished: bool,
 }
 
 impl From<RpcTorrent> for Torrent {
@@ -406,6 +411,7 @@ impl From<RpcTorrent> for Torrent {
             progress: value.progress,
             download_dir: value.download_dir,
             labels: value.labels,
+            is_finished: value.is_finished,
         }
     }
 }

crates/arr-e2e/tests/e2e.rs

@@ -162,8 +162,77 @@ async fn transmission_add_list_and_remove() {
     assert!(torrents.iter().all(|torrent| torrent.id != second.id));
 }
 
+/// Transmission, not arr's import state, owns the done-seeding boundary. A
+/// zero ratio makes the transition quick while still exercising real RPC.
+#[tokio::test]
+async fn transmission_reports_done_only_after_its_seed_limit() {
+    let client = TransmissionClient::new(&transmission_url()).expect("valid endpoint");
+    let name = format!("arr-e2e-reaper-{}", uuid::Uuid::new_v4());
+    let download_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("../../target/e2e-downloads")
+        .canonicalize()
+        .expect("canonical download directory");
+    std::fs::write(download_dir.join(&name), [0_u8]).expect("seed torrent payload");
+    let metainfo = torrent_with_name(&name);
+    let request = |ratio| AddTorrent {
+        source: TorrentSource::Metainfo(metainfo.clone()),
+        label: "movies-main".into(),
+        download_dir: download_dir.clone(),
+        seed_ratio_limit: ratio,
+        seed_idle_limit_minutes: 60,
+    };
+    let added = client
+        .add_torrent(request(100.0))
+        .await
+        .expect("add torrent");
+
+    let before = client
+        .list_torrents()
+        .await
+        .expect("list before limit")
+        .into_iter()
+        .find(|torrent| torrent.id == added.id)
+        .expect("torrent before limit");
+    assert!(
+        !before.is_finished,
+        "the reaper must leave this torrent alone"
+    );
+
+    client
+        .add_torrent(request(0.0))
+        .await
+        .expect("lower duplicate's ratio limit");
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+    loop {
+        let torrent = client
+            .list_torrents()
+            .await
+            .expect("list after limit")
+            .into_iter()
+            .find(|torrent| torrent.id == added.id)
+            .expect("torrent after limit");
+        if torrent.is_finished {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "seed limit was not enforced: {torrent:?}"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+
+    client
+        .remove_torrent(added.id, true)
+        .await
+        .expect("cleanup");
+}
+
 fn torrent_with_name(name: &str) -> Vec<u8> {
-    let piece_hash = [0_u8; 20];
+    // SHA-1 of the single zero byte written by the seeding-boundary test.
+    let piece_hash = [
+        0x5b, 0xa9, 0x3c, 0x9d, 0xb0, 0xcf, 0xf9, 0x3f, 0x52, 0xb5, 0x21, 0xd7, 0x42, 0x0e, 0x43,
+        0xf6, 0xed, 0xa2, 0x78, 0x4f,
+    ];
     let mut bytes = format!("d4:infod6:lengthi1e4:name{}:{name}", name.len()).into_bytes();
     bytes.extend_from_slice(b"12:piece lengthi16384e6:pieces20:");
     bytes.extend_from_slice(&piece_hash);