From 0585811a0c4d9c083e937037ea250543fdf57277 Mon Sep 17 00:00:00 2001 From: "naps62-yolo (agent)" Date: Sun, 23 Aug 2026 00:03:30 +0100 Subject: [PATCH] Per-tracker seeding rules and torrent reaper (#88) --- crates/arr-daemon/src/config.rs | 48 +++++++++++ crates/arr-daemon/src/grab.rs | 68 ++++++++++++--- crates/arr-daemon/src/main.rs | 30 +++++-- crates/arr-daemon/src/reaper.rs | 148 ++++++++++++++++++++++++++++++++ crates/arr-dl/src/lib.rs | 8 +- crates/arr-e2e/tests/e2e.rs | 38 ++++++++ 6 files changed, 319 insertions(+), 21 deletions(-) create mode 100644 crates/arr-daemon/src/reaper.rs diff --git a/crates/arr-daemon/src/config.rs b/crates/arr-daemon/src/config.rs index ba96884..7a99cbf 100644 --- a/crates/arr-daemon/src/config.rs +++ b/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, #[serde(default)] + tracker_seeding: HashMap, + #[serde(default)] jellyfin_url: Option, #[serde(default)] ntfy_url: Option, } +#[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 { 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, pub tmdb_api_key: Option, /// 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::() + .map(|id| (id, rule)) + .map_err(|_| ConfigError::InvalidTrackerId(id.clone())) + }) + .collect::, _>>()?; 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 { diff --git a/crates/arr-daemon/src/grab.rs b/crates/arr-daemon/src/grab.rs index fcac3f6..aedd744 100644 --- a/crates/arr-daemon/src/grab.rs +++ b/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, +} + +impl SeedingRules { + #[must_use] + pub fn new(default: SeedingLimits, trackers: HashMap) -> 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, /// [`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(); @@ -665,12 +686,12 @@ async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> { /// `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 { - let kind = if loaded.root_kind == "movie" { - "movies" - } else { - &loaded.root_kind - }; - format!("{kind}-{}", loaded.root_audience) + label_for_root(&loaded.root_kind, &loaded.root_audience) +} + +pub(crate) fn label_for_root(kind: &str, audience: &str) -> String { + let kind = if kind == "movie" { "movies" } else { kind }; + format!("{kind}-{audience}") } /// Indexers hand out magnets and `.torrent` links interchangeably; @@ -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", diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 0ae001c..031e51a 100644 --- a/crates/arr-daemon/src/main.rs +++ b/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`. diff --git a/crates/arr-daemon/src/reaper.rs b/crates/arr-daemon/src/reaper.rs new file mode 100644 index 0000000..f710091 --- /dev/null +++ b/crates/arr-daemon/src/reaper.rs @@ -0,0 +1,148 @@ +//! Removes arr torrents only after Transmission says their seeding obligation +//! is complete. The library/import state is deliberately not consulted (§7.3). + +use std::collections::HashSet; + +use arr_db::Db; +use arr_dl::TransmissionClient; + +use crate::grab::label_for_root; +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, labels: &HashSet) -> Result, 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| labels.contains(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 { + let labels = sqlx::query_as::<_, (String, String)>("SELECT kind, audience FROM roots") + .fetch_all(_database.pool()) + .await + .map_err(|error| Box::new(error) as crate::reconcile::ActionError)? + .into_iter() + .map(|(kind, audience)| label_for_root(&kind, &audience)) + .collect::>(); + self.tick(&labels) + .await + .map_err(|error| Box::new(error) as crate::reconcile::ActionError) + }) + } +} + +#[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>>, + } + + 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 labels = HashSet::from(["movies-main".to_owned(), "movies-kids".to_owned()]); + let outcomes = action.tick(&labels).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()); + + let labels = HashSet::from(["movies-kids".to_owned()]); + action.tick(&labels).await.unwrap(); + + assert_eq!(removed.lock().unwrap().len(), 1); + } +} diff --git a/crates/arr-dl/src/lib.rs b/crates/arr-dl/src/lib.rs index 644b4db..2edb7c5 100644 --- a/crates/arr-dl/src/lib.rs +++ b/crates/arr-dl/src/lib.rs @@ -68,6 +68,9 @@ pub struct Torrent { pub progress: f64, pub download_dir: PathBuf, pub labels: Vec, + /// 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, + #[serde(rename = "isFinished", default)] + is_finished: bool, } impl From for Torrent { @@ -406,6 +411,7 @@ impl From for Torrent { progress: value.progress, download_dir: value.download_dir, labels: value.labels, + is_finished: value.is_finished, } } } diff --git a/crates/arr-e2e/tests/e2e.rs b/crates/arr-e2e/tests/e2e.rs index ba135ba..47376d6 100644 --- a/crates/arr-e2e/tests/e2e.rs +++ b/crates/arr-e2e/tests/e2e.rs @@ -162,6 +162,44 @@ 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. The +/// real service must report an unfinished torrent before its seed limit clears. +#[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("/tmp/arr-e2e"); + 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 + .remove_torrent(added.id, true) + .await + .expect("cleanup"); +} + fn torrent_with_name(name: &str) -> Vec { let piece_hash = [0_u8; 20]; let mut bytes = format!("d4:infod6:lengthi1e4:name{}:{name}", name.len()).into_bytes();