yolo/arr
3
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: RSS sync worker (#91)
ci / web (push) Successful in 26s
Details
e2e / e2e (push) Successful in 53s
Details
ci / rust (push) Successful in 2m13s
Details

Browse Source
This commit was merged in pull request #91.
This commit is contained in:
naps62-yolo (agent)
2026-08-23 01:04:15 +01:00
parent e01505bd53
commit 225be33b55
11 changed files with 1285 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-daemon/src/grab.rs
+95 -103
View File
@@ -16,7 +16,7 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use std::time::{SystemTime, UNIX_EPOCH};
use arr_core::policy::{evaluate, Candidate};
use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict};
@@ -25,6 +25,7 @@ use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
use arr_meta::TmdbClient;
use crate::indexers::{DiscoveryError, IndexerDirectory};
use crate::reconcile::{Action, ActionFuture, Outcome};
/// How many titles one tick may search. The reconcile lane has a 25 s budget
@@ -33,16 +34,6 @@ use crate::reconcile::{Action, ActionFuture, Outcome};
/// and release-date gating are issue #26.
const MOVIES_PER_TICK: i64 = 5;
/// How long a discovered indexer list is reused. Prowlarr enumerates one
/// indexer per call plus a `t=caps` probe each, all sequential, so paying for
/// it every 30 s would leave the tick no room to search.
const INDEXER_CACHE_TTL: Duration = Duration::from_mins(15);
/// The share of the reconcile lane's 25 s budget discovery may spend. One
/// unresponsive tracker must not cancel the whole action before a single
/// targeted search has run; past this the last known list is used instead.
const INDEXER_DISCOVERY_TIMEOUT: Duration = Duration::from_secs(8);
/// Seeding obligations, per tracker in principle (§7.3) and per install in
/// practice until issue #25 gives them a home. Both are set on the torrent at
/// add time and enforced by Transmission.
@@ -79,8 +70,8 @@ pub enum GrabError {
Database(#[from] sqlx::Error),
#[error("policy: {0}")]
Policy(#[from] arr_db::PolicyError),
#[error("prowlarr: {0}")]
Prowlarr(#[from] arr_indexer::Error),
#[error("{0}")]
Discovery(#[from] DiscoveryError),
#[error("TMDB: {0}")]
Metadata(#[from] arr_meta::Error),
#[error("movie {0} has an invalid TMDB id")]
@@ -92,15 +83,6 @@ pub enum GrabError {
name: String,
source: serde_json::Error,
},
#[error("indexer discovery timed out after {0:?} and none is known yet")]
IndexerDiscoveryTimeout(Duration),
}
/// The searchable indexer ids, and when they were last discovered.
#[derive(Debug, Default)]
struct IndexerCache {
ids: Vec<i64>,
refreshed_at: Option<Instant>,
}
/// Sends the best eligible release for every wanted movie that has neither a
@@ -108,14 +90,9 @@ struct IndexerCache {
#[derive(Debug)]
pub struct GrabAction {
prowlarr: ProwlarrClient,
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
grabber: Grabber,
tmdb: Option<Arc<TmdbClient>>,
indexers: tokio::sync::RwLock<IndexerCache>,
/// [`INDEXER_DISCOVERY_TIMEOUT`], overridden by tests that cannot wait
/// out the real one. Mirrors `ReconcileLoop`'s action timeout override.
discovery_timeout: Duration,
indexers: IndexerDirectory,
}
impl GrabAction {
@@ -127,13 +104,10 @@ impl GrabAction {
seeding: SeedingRules,
) -> Self {
Self {
indexers: IndexerDirectory::new(prowlarr.clone()),
prowlarr,
transmission,
download_dir,
seeding,
grabber: Grabber::new(transmission, download_dir, seeding),
tmdb: None,
indexers: tokio::sync::RwLock::new(IndexerCache::default()),
discovery_timeout: INDEXER_DISCOVERY_TIMEOUT,
}
}
@@ -165,7 +139,7 @@ impl GrabAction {
if !search_due(&movie) {
continue;
}
let searchable = self.searchable_indexers().await?;
let searchable = self.indexers.searchable().await?;
if searchable.is_empty() {
tracing::warn!("no indexer advertises a text search; nothing can be grabbed");
return Ok(outcomes);
@@ -290,6 +264,7 @@ impl GrabAction {
}
let torrents: HashMap<String, f64> = self
.grabber
.transmission
.list_torrents()
.await?
@@ -327,51 +302,6 @@ impl GrabAction {
Ok(outcomes)
}
/// The indexers that accept a text search, cached across ticks.
///
/// Discovery is bounded and its result is reused, because Prowlarr probes
/// capabilities one indexer at a time: a single slow tracker would
/// otherwise burn the whole reconcile budget before any search runs. A
/// refresh that fails or times out keeps the previous list rather than
/// stopping the tick, and only an empty cache turns that into an error.
async fn searchable_indexers(&self) -> Result<Vec<i64>, GrabError> {
{
let cache = self.indexers.read().await;
if cache
.refreshed_at
.is_some_and(|at| at.elapsed() < INDEXER_CACHE_TTL)
{
return Ok(cache.ids.clone());
}
}
let discovered =
tokio::time::timeout(self.discovery_timeout, self.prowlarr.indexers()).await;
let mut cache = self.indexers.write().await;
match discovered {
Ok(Ok(indexers)) => {
cache.ids = indexers
.iter()
.filter(|indexer| indexer.capabilities.search.available)
.map(|indexer| indexer.id)
.collect();
cache.refreshed_at = Some(Instant::now());
}
Ok(Err(error)) if cache.ids.is_empty() => return Err(error.into()),
Err(_) if cache.ids.is_empty() => {
return Err(GrabError::IndexerDiscoveryTimeout(self.discovery_timeout))
}
Ok(Err(error)) => {
tracing::warn!(%error, "indexer discovery failed; using the last known list");
}
Err(_) => tracing::warn!(
timeout_seconds = self.discovery_timeout.as_secs(),
"indexer discovery timed out; using the last known list"
),
}
Ok(cache.ids.clone())
}
/// Search every indexer for one title, cache each candidate with its
/// verdict and score (§9.3), and return the eligible ones best first.
///
@@ -475,9 +405,67 @@ impl GrabAction {
return Ok(None);
};
self.send_winner(database, movie, &loaded, &blacklist, winner)
self.grabber
.send_winner(
database,
&GrabTarget {
movie_id: movie.id,
title: &movie.title,
counts_as_attempt: true,
},
&loaded,
&blacklist,
winner,
)
.await
}
}
/// Sending a chosen release to Transmission and recording the grab.
///
/// Targeted search and RSS (§6.2) differ in how a title is chosen and in
/// whether a failed attempt counts toward a backoff; from the winning
/// release onward they are the same writes, so they share this.
#[derive(Debug)]
pub(crate) struct Grabber {
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
}
/// The title a winning release is being grabbed for.
#[derive(Debug)]
pub(crate) struct GrabTarget<'a> {
pub(crate) movie_id: i64,
pub(crate) title: &'a str,
/// Whether a grab that does not complete counts toward the targeted
/// search backoff (§6.2). RSS never backs off, so it passes `false`.
pub(crate) counts_as_attempt: bool,
}
impl Grabber {
pub(crate) fn new(
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
) -> Self {
Self {
transmission,
download_dir,
seeding,
}
}
async fn record_attempt(
&self,
database: &Db,
target: &GrabTarget<'_>,
) -> Result<(), GrabError> {
if target.counts_as_attempt {
record_search(database, target.movie_id).await?;
}
Ok(())
}
/// Add the winning release to Transmission and record the grab.
///
@@ -485,10 +473,10 @@ impl GrabAction {
/// not a completed grab — a Transmission error or a blacklisted-infohash
/// drop must not leave the same release to repeat next tick with no
/// backoff.
async fn send_winner(
pub(crate) async fn send_winner(
&self,
database: &Db,
movie: &PendingMovie,
target: &GrabTarget<'_>,
loaded: &MoviePolicy,
blacklist: &Blacklist,
winner: Eligible,
@@ -507,7 +495,7 @@ impl GrabAction {
{
Ok(added) => added,
Err(error) => {
record_search(database, movie.id).await?;
self.record_attempt(database, target).await?;
return Err(error.into());
}
};
@@ -517,8 +505,8 @@ impl GrabAction {
// Transmission has fetched it, so the same blacklisted torrent can
// reach here under a new name.
if blacklist.blocks_infohash(&infohash) {
record_search(database, movie.id).await?;
self.drop_blacklisted_torrent(database, movie, &winner, &added)
self.record_attempt(database, target).await?;
self.drop_blacklisted_torrent(database, target, &winner, &added)
.await?;
return Ok(None);
}
@@ -531,7 +519,7 @@ impl GrabAction {
ON CONFLICT (infohash) DO NOTHING
RETURNING id AS "id!: i64""#,
winner.id,
movie.id,
target.movie_id,
infohash
)
.fetch_optional(database.pool())
@@ -540,14 +528,14 @@ impl GrabAction {
"UPDATE movies SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie.id
target.movie_id
)
.execute(database.pool())
.await?;
let Some(inserted) = inserted else {
tracing::info!(
movie_id = movie.id,
movie_id = target.movie_id,
infohash,
"grab already recorded for this torrent"
);
@@ -555,8 +543,8 @@ impl GrabAction {
};
tracing::info!(
movie_id = movie.id,
title = movie.title,
movie_id = target.movie_id,
title = target.title,
release = winner.name,
score = winner.score,
infohash,
@@ -564,7 +552,7 @@ impl GrabAction {
"grabbed"
);
Ok(Some(Outcome::new(
format!("movie {} wanted with no file", movie.id),
format!("movie {} wanted with no file", target.movie_id),
format!("grabbed {} as grab {}", winner.name, inserted.id),
)))
}
@@ -577,7 +565,7 @@ impl GrabAction {
async fn drop_blacklisted_torrent(
&self,
database: &Db,
movie: &PendingMovie,
target: &GrabTarget<'_>,
winner: &Eligible,
added: &arr_dl::AddedTorrent,
) -> Result<(), GrabError> {
@@ -604,7 +592,7 @@ impl GrabAction {
// The earlier grab's torrent, still working off its seeding
// obligation (§7.3). Nothing here deletes a torrent.
tracing::warn!(
movie_id = movie.id,
movie_id = target.movie_id,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; left seeding"
@@ -614,7 +602,7 @@ impl GrabAction {
// obligation and has nothing on disk worth keeping.
self.transmission.remove_torrent(added.id, true).await?;
tracing::warn!(
movie_id = movie.id,
movie_id = target.movie_id,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; removed"
@@ -652,13 +640,13 @@ struct PendingMovie {
/// The eligible view of a stored release, ranked for selection.
#[derive(Debug, Clone)]
struct Eligible {
pub(crate) struct Eligible {
id: i64,
indexer_id: i64,
guid: String,
name: String,
pub(crate) indexer_id: i64,
pub(crate) guid: String,
pub(crate) name: String,
download_url: String,
score: i64,
pub(crate) score: i64,
}
/// The gap, straight out of the domain rows (§8).
@@ -759,7 +747,7 @@ fn is_digitally_released(digital_release: Option<&str>) -> bool {
/// 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(
pub(crate) async fn store_release(
database: &Db,
movie_id: i64,
release: &SearchRelease,
@@ -919,6 +907,7 @@ fn rfc3339(time: SystemTime) -> Option<String> {
#[allow(clippy::unwrap_used)]
mod tests {
use std::sync::{Arc, Mutex};
use std::time::Duration;
use serde_json::{json, Value};
use wiremock::matchers::{method, path, query_param};
@@ -1697,14 +1686,17 @@ mod tests {
let (downloader, fake) = transmission().await;
let mut action = action(&indexer, &downloader);
action.discovery_timeout = Duration::from_millis(50);
action.indexers.discovery_timeout = Duration::from_millis(50);
let error = action
.tick(&database)
.await
.expect_err("discovery cannot complete");
assert!(matches!(error, GrabError::IndexerDiscoveryTimeout(_)));
assert!(matches!(
error,
GrabError::Discovery(DiscoveryError::Timeout(_))
));
assert!(fake.torrents().is_empty());
}
crates/arr-daemon/src/indexers.rs
+98
View File
@@ -0,0 +1,98 @@
//! The searchable indexer list, discovered once and reused. See DESIGN.md
//! §6.1.
//!
//! Prowlarr enumerates one indexer per call and probes `t=caps` for each,
//! sequentially, so both the 30 s grab lane and the 10 min RSS lane would
//! spend most of their budget rediscovering the same list. Each lane keeps
//! its own directory: they run concurrently and neither should wait on the
//! other's HTTP calls.
use std::time::{Duration, Instant};
use arr_indexer::ProwlarrClient;
/// How long a discovered indexer list is reused.
const CACHE_TTL: Duration = Duration::from_mins(15);
/// The share of a tick's budget discovery may spend. One unresponsive
/// tracker must not cancel the whole action before a single search has run;
/// past this the last known list is used instead.
const DISCOVERY_TIMEOUT: Duration = Duration::from_secs(8);
/// A discovery that produced nothing usable at all.
#[derive(Debug, thiserror::Error)]
pub enum DiscoveryError {
#[error("prowlarr: {0}")]
Prowlarr(#[from] arr_indexer::Error),
#[error("indexer discovery timed out after {0:?} and none is known yet")]
Timeout(Duration),
}
#[derive(Debug, Default)]
struct Cache {
ids: Vec<i64>,
refreshed_at: Option<Instant>,
}
/// The indexers that accept a text search, cached across ticks.
#[derive(Debug)]
pub struct IndexerDirectory {
prowlarr: ProwlarrClient,
cache: tokio::sync::RwLock<Cache>,
/// [`DISCOVERY_TIMEOUT`], overridden by tests that cannot wait out the
/// real one. Mirrors `ReconcileLoop`'s action timeout override.
pub(crate) discovery_timeout: Duration,
}
impl IndexerDirectory {
pub fn new(prowlarr: ProwlarrClient) -> Self {
Self {
prowlarr,
cache: tokio::sync::RwLock::new(Cache::default()),
discovery_timeout: DISCOVERY_TIMEOUT,
}
}
/// The ids of every indexer advertising a text search — which is also
/// the RSS feed, an empty-query text search (§6.2).
///
/// A refresh that fails or times out keeps the previous list rather than
/// stopping the tick, and only an empty cache turns that into an error.
pub async fn searchable(&self) -> Result<Vec<i64>, DiscoveryError> {
{
let cache = self.cache.read().await;
if cache
.refreshed_at
.is_some_and(|at| at.elapsed() < CACHE_TTL)
{
return Ok(cache.ids.clone());
}
}
let discovered =
tokio::time::timeout(self.discovery_timeout, self.prowlarr.indexers()).await;
let mut cache = self.cache.write().await;
match discovered {
Ok(Ok(indexers)) => {
cache.ids = indexers
.iter()
.filter(|indexer| indexer.capabilities.search.available)
.map(|indexer| indexer.id)
.collect();
cache.refreshed_at = Some(Instant::now());
}
Ok(Err(error)) if cache.ids.is_empty() => return Err(error.into()),
Err(_) if cache.ids.is_empty() => {
return Err(DiscoveryError::Timeout(self.discovery_timeout))
}
Ok(Err(error)) => {
tracing::warn!(%error, "indexer discovery failed; using the last known list");
}
Err(_) => tracing::warn!(
timeout_seconds = self.discovery_timeout.as_secs(),
"indexer discovery timed out; using the last known list"
),
}
Ok(cache.ids.clone())
}
}
crates/arr-daemon/src/main.rs
+75 -43
View File
@@ -3,8 +3,10 @@
mod config;
mod grab;
mod import;
mod indexers;
mod reaper;
pub mod reconcile;
mod rss;
mod web;
use std::process::ExitCode;
@@ -19,6 +21,7 @@ use grab::{GrabAction, SeedingLimits, SeedingRules};
use import::ImportAction;
use reaper::ReaperAction;
use reconcile::{ReconcileLoop, Tick};
use rss::RssAction;
use tower_http::trace::TraceLayer;
/// Dump the `OpenAPI` document and exit, instead of serving. `just gen-client`
@@ -103,49 +106,7 @@ async fn run() -> Result<(), Error> {
} else {
None
};
let mut reconcile = ReconcileLoop::new(database.clone());
// Without a Prowlarr key nothing can be searched, so the grab lane stays
// unregistered rather than failing a tick every 30 seconds.
if let (Some(key), Some(tmdb)) = (config.prowlarr_api_key.clone(), tmdb.clone()) {
let prowlarr = arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)?;
reconcile = reconcile.register(
Tick::Reconcile,
GrabAction::new(
prowlarr,
transmission.clone(),
config.download_dir.clone(),
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(),
),
)
.with_tmdb(tmdb),
);
} else {
tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed");
}
// Grab before import, so a download that completes on this tick is
// imported on this tick.
reconcile = reconcile.register(
Tick::Reconcile,
ImportAction::new(transmission.clone(), arr_probe::Prober::new()),
);
reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission));
let reconcile = reconcile_loop(&database, &config, &transmission, tmdb.as_ref())?;
// Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
// needs its own TMDB client for `movie/lookup`.
@@ -202,6 +163,77 @@ async fn run() -> Result<(), Error> {
}
}
/// Wire the reconcile lanes (DESIGN.md §8). Grab and RSS both need a
/// Prowlarr key and grab needs TMDB as well; a lane whose upstream is not
/// configured stays unregistered rather than failing every tick.
fn reconcile_loop(
database: &Db,
config: &Config,
transmission: &arr_dl::TransmissionClient,
tmdb: Option<&Arc<TmdbClient>>,
) -> Result<ReconcileLoop, Error> {
let mut reconcile = ReconcileLoop::new(database.clone());
let seeding = 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(),
);
let prowlarr = config
.prowlarr_api_key
.clone()
.map(|key| arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key))
.transpose()?;
if let (Some(prowlarr), Some(tmdb)) = (prowlarr.as_ref(), tmdb) {
reconcile = reconcile.register(
Tick::Reconcile,
GrabAction::new(
prowlarr.clone(),
transmission.clone(),
config.download_dir.clone(),
seeding.clone(),
)
.with_tmdb(Arc::clone(tmdb)),
);
} else {
tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed");
}
// RSS needs no TMDB: it matches what the feeds already carry against the
// wanted list (§6.2).
if let Some(prowlarr) = prowlarr {
reconcile = reconcile.register(
Tick::Rss,
RssAction::new(
prowlarr,
transmission.clone(),
config.download_dir.clone(),
seeding,
),
);
}
// Grab before import, so a download that completes on this tick is
// imported on this tick.
reconcile = reconcile.register(
Tick::Reconcile,
ImportAction::new(transmission.clone(), arr_probe::Prober::new()),
);
Ok(reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone())))
}
/// Stop accepting on Ctrl-C, or on the SIGTERM a service manager sends.
async fn shutdown() {
let interrupt = async {
crates/arr-daemon/src/rss.rs
+574
View File
@@ -0,0 +1,574 @@
//! RSS sync: one empty-query Torznab call per indexer, matched against the
//! whole wanted list locally. See DESIGN.md §6.2.
//!
//! The property this lane is built around is that its cost is one call per
//! indexer, whatever the library holds. Nothing here is per-title, so every
//! wanted item is compared against every result on every pass, forever:
//! there is no backoff, no attempt counter, and no release-date gate. Those
//! belong to targeted search, which pays per title.
//!
//! `blocked` is honoured the other way round from targeted search (§6.3): it
//! stops a title being searched for, and leaves it matching RSS.
use std::collections::hash_map::Entry;
use std::collections::HashMap;
use std::path::PathBuf;
use arr_core::matching::{match_movie, MatchKind, WantedMovie};
use arr_core::{Language, MovieId};
use arr_db::{Blacklist, Db, MoviePolicy};
use arr_dl::TransmissionClient;
use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
use crate::grab::{store_release, Eligible, GrabError, GrabTarget, Grabber, SeedingRules};
use crate::indexers::IndexerDirectory;
use crate::reconcile::{Action, ActionFuture, Outcome};
/// Reads every indexer's feed and grabs what it can tie to a wanted title.
#[derive(Debug)]
pub struct RssAction {
prowlarr: ProwlarrClient,
grabber: Grabber,
indexers: IndexerDirectory,
}
impl RssAction {
#[must_use]
pub fn new(
prowlarr: ProwlarrClient,
transmission: TransmissionClient,
download_dir: PathBuf,
seeding: SeedingRules,
) -> Self {
Self {
indexers: IndexerDirectory::new(prowlarr.clone()),
prowlarr,
grabber: Grabber::new(transmission, download_dir, seeding),
}
}
async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let wanted = wanted_movies(database).await?;
if wanted.is_empty() {
return Ok(Vec::new());
}
let indexers = self.indexers.searchable().await?;
if indexers.is_empty() {
tracing::warn!("no indexer advertises a text search; no RSS feed to read");
return Ok(Vec::new());
}
let releases = self.feeds(&indexers).await;
let blacklist = Blacklist::load(database.pool()).await?;
let winners = self
.match_feeds(database, &wanted, releases, &blacklist)
.await?;
let mut outcomes = Vec::new();
for (movie, winner) in winners {
let outcome = self
.grabber
.send_winner(
database,
&GrabTarget {
movie_id: movie.id,
title: &movie.title,
// §6.2: RSS never backs off, so a failed grab here
// must not spend one of targeted search's attempts.
counts_as_attempt: false,
},
&movie.policy,
&blacklist,
winner,
)
.await?;
outcomes.extend(outcome);
}
Ok(outcomes)
}
/// One empty-query call per indexer (§6.2). A tracker that fails is
/// logged and skipped: the next pass is ten minutes away.
async fn feeds(&self, indexers: &[i64]) -> Vec<SearchRelease> {
let mut releases = Vec::new();
for search in self
.prowlarr
.search_indexers(indexers, &SearchRequest::Rss)
.await
{
if let Some(error) = search.error {
tracing::warn!(indexer_id = search.indexer_id, %error, "RSS feed failed");
}
releases.extend(search.releases);
}
releases
}
/// Tie each feed item to a wanted title, cache it with its verdict, and
/// keep the best eligible candidate per title.
///
/// One title takes at most one torrent per pass, and the ordering is the
/// same total order targeted search uses, so the two triggers agree on
/// which release wins.
async fn match_feeds(
&self,
database: &Db,
wanted: &[WantedMovie],
releases: Vec<SearchRelease>,
blacklist: &Blacklist,
) -> Result<Vec<(Grabbable, Eligible)>, GrabError> {
let mut policies: HashMap<i64, Option<Grabbable>> = HashMap::new();
let mut best: HashMap<i64, Eligible> = HashMap::new();
for release in releases {
let claims = arr_parse::parse(&release.name);
let Some(matched) = match_movie(wanted, release.tmdb_id, &claims) else {
continue;
};
let movie_id = matched.movie.0;
let grabbable = match policies.entry(movie_id) {
Entry::Occupied(entry) => entry.into_mut(),
Entry::Vacant(entry) => {
entry.insert(load_grabbable(database, wanted, movie_id).await?)
}
};
let Some(grabbable) = grabbable.as_ref() else {
continue;
};
tracing::info!(
movie_id,
release = release.name,
by = match matched.kind {
MatchKind::TmdbId => "tmdb id",
MatchKind::TitleAndYear => "title and year",
},
"RSS result matched a wanted title"
);
let stored = store_release(
database,
movie_id,
&release,
&grabbable.policy.policy,
&grabbable.policy.overrides,
&grabbable.language,
blacklist,
)
.await?;
let Some(candidate) = stored else {
continue;
};
let incumbent = best.get(&movie_id);
if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) {
best.insert(movie_id, candidate);
}
}
let mut winners: Vec<(Grabbable, Eligible)> = best
.into_iter()
.filter_map(|(movie_id, candidate)| {
let grabbable = policies.get(&movie_id).and_then(Clone::clone)?;
Some((grabbable, candidate))
})
.collect();
winners.sort_by_key(|(movie, _)| movie.id);
Ok(winners)
}
}
impl Action for RssAction {
fn name(&self) -> &'static str {
"rss"
}
fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
Box::pin(async move { self.tick(database).await.map_err(Into::into) })
}
}
/// A matched title with everything a grab needs loaded.
#[derive(Clone, Debug)]
struct Grabbable {
id: i64,
title: String,
policy: MoviePolicy,
language: Language,
}
/// Better score first, then the tiebreaks targeted search sorts by, so both
/// triggers pick the same release out of the same candidates.
fn beats(candidate: &Eligible, incumbent: &Eligible) -> bool {
candidate
.score
.cmp(&incumbent.score)
.then_with(|| incumbent.indexer_id.cmp(&candidate.indexer_id))
.then_with(|| incumbent.guid.cmp(&candidate.guid))
== std::cmp::Ordering::Greater
}
/// Every wanted title with neither a file nor a grab in flight.
///
/// Unlike targeted search's work list this is not limited or ordered by when
/// a title was last searched: matching is local, so the cost of carrying the
/// whole list is a string comparison per title per feed item (§6.2). Blocked
/// titles are here on purpose (§6.3).
async fn wanted_movies(database: &Db) -> Result<Vec<WantedMovie>, GrabError> {
let rows = sqlx::query!(
r#"
SELECT m.id AS "id!: i64",
m.tmdb_id AS "tmdb_id!: i64",
m.title AS "title!: String",
m.year
FROM movies m
WHERE m.wanted = 1
AND NOT EXISTS (
SELECT 1 FROM media_files f
WHERE f.owner_kind = 'movie' AND f.owner_id = m.id
)
AND NOT EXISTS (
SELECT 1 FROM grabs g
WHERE g.target_kind = 'movie' AND g.target_id = m.id
AND g.state IN ('sent', 'downloaded', 'imported')
)
ORDER BY m.id
"#
)
.fetch_all(database.pool())
.await?;
Ok(rows
.into_iter()
.map(|row| WantedMovie {
id: MovieId(row.id),
tmdb_id: u32::try_from(row.tmdb_id).ok(),
title: row.title,
year: row.year.and_then(|year| u16::try_from(year).ok()),
})
.collect())
}
/// The policy and original language a matched title needs, or `None` when
/// the title cannot be evaluated yet.
async fn load_grabbable(
database: &Db,
wanted: &[WantedMovie],
movie_id: i64,
) -> Result<Option<Grabbable>, GrabError> {
let Some(title) = wanted
.iter()
.find(|movie| movie.id == MovieId(movie_id))
.map(|movie| movie.title.clone())
else {
return Ok(None);
};
let Some(policy) = database.movie_policy(movie_id).await? else {
return Ok(None);
};
// §5.2: every language rule is expressed against the title's own
// original language, and guessing it is worse than not grabbing.
let language: Option<String> = sqlx::query_scalar!(
"SELECT original_language FROM movies WHERE id = ?",
movie_id
)
.fetch_optional(database.pool())
.await?
.flatten();
let Some(language) = language else {
tracing::warn!(movie_id, "no original language yet; not grabbing from RSS");
return Ok(None);
};
Ok(Some(Grabbable {
id: movie_id,
title,
policy,
language: arr_db::policy::language(&language),
}))
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use std::sync::{Arc, Mutex};
use serde_json::{json, Value};
use wiremock::matchers::{method, path, query_param, query_param_is_missing};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
use super::*;
use crate::grab::SeedingLimits;
use arr_dl::TransmissionClient;
/// The recorded feed: two releases of one wanted title, one that only a
/// TMDB id identifies, one near miss, one TV item and one film nobody
/// asked for.
const FEED: &str = include_str!("../tests/fixtures/rss.xml");
const INDEXERS: [i64; 2] = [7, 9];
/// Enough of Transmission to add a torrent and list nothing back.
#[derive(Clone, Debug, Default)]
struct FakeTransmission {
added: Arc<Mutex<Vec<String>>>,
}
impl Respond for FakeTransmission {
fn respond(&self, request: &Request) -> ResponseTemplate {
let body: Value = serde_json::from_slice(&request.body).unwrap();
match body["method"].as_str().unwrap_or_default() {
"torrent-add" => {
let source = body["arguments"]["filename"]
.as_str()
.unwrap_or_default()
.to_owned();
let mut added = self.added.lock().unwrap();
added.push(source.clone());
let id = i64::try_from(added.len()).unwrap();
success(&json!({"torrent-added": {
"id": id, "name": source, "hashString": format!("{id:040x}")
}}))
}
"torrent-get" => success(&json!({"torrents": []})),
_ => success(&json!({})),
}
}
}
fn success(arguments: &Value) -> ResponseTemplate {
ResponseTemplate::new(200).set_body_json(json!({
"result": "success", "arguments": arguments
}))
}
/// Two indexers, both advertising a text search, both serving the same
/// feed to an empty query.
async fn prowlarr() -> MockServer {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/indexer"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!([
{"id": 7, "name": "alpha", "enable": true},
{"id": 9, "name": "beta", "enable": true},
])))
.mount(&server)
.await;
for id in INDEXERS {
Mock::given(method("GET"))
.and(path(format!("/{id}/api")))
.and(query_param("t", "caps"))
.respond_with(ResponseTemplate::new(200).set_body_string(
r#"<caps><searching><search available="yes" supportedParams="q"/></searching></caps>"#,
))
.mount(&server)
.await;
// The RSS call is `t=search` with no `q` at all (§6.2).
Mock::given(method("GET"))
.and(path(format!("/{id}/api")))
.and(query_param("t", "search"))
.and(query_param_is_missing("q"))
.respond_with(ResponseTemplate::new(200).set_body_string(FEED))
.mount(&server)
.await;
}
server
}
async fn transmission() -> (MockServer, FakeTransmission) {
let server = MockServer::start().await;
let fake = FakeTransmission::default();
Mock::given(method("POST"))
.respond_with(fake.clone())
.mount(&server)
.await;
(server, fake)
}
/// `(tmdb_id, title, year, blocked)`.
async fn wanted(movies: &[(i64, &str, i64, bool)]) -> (tempfile::TempDir, Db) {
let dir = tempfile::tempdir().unwrap();
let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
database.migrate().await.unwrap();
for &(tmdb_id, title, year, blocked) in movies {
sqlx::query(
"INSERT INTO movies (tmdb_id, title, year, original_language, blocked, root_id)
SELECT ?, ?, ?, 'en', ?, id
FROM roots WHERE kind = 'movie' AND audience = 'main'",
)
.bind(tmdb_id)
.bind(title)
.bind(year)
.bind(i64::from(blocked))
.execute(database.pool())
.await
.unwrap();
}
(dir, database)
}
fn action(prowlarr: &MockServer, transmission: &MockServer) -> RssAction {
RssAction::new(
ProwlarrClient::new(prowlarr.uri(), "key").unwrap(),
TransmissionClient::new(&transmission.uri()).unwrap(),
PathBuf::from("/mnt/media/transmission/complete"),
SeedingRules::new(
SeedingLimits {
ratio: 1.0,
idle_minutes: 60,
},
HashMap::new(),
),
)
}
/// `(movie title, grabbed release name)` for every grab that was made.
async fn grabbed(database: &Db) -> Vec<(String, String)> {
sqlx::query_as::<_, (String, String)>(
"SELECT m.title, r.name
FROM grabs g
JOIN movies m ON m.id = g.target_id
JOIN releases r ON r.id = g.release_id
ORDER BY m.title",
)
.fetch_all(database.pool())
.await
.unwrap()
}
async fn stored_releases(database: &Db) -> Vec<String> {
sqlx::query_scalar::<_, String>("SELECT DISTINCT guid FROM releases ORDER BY guid")
.fetch_all(database.pool())
.await
.unwrap()
}
/// The acceptance case: a recorded feed against a seeded wanted list
/// grabs exactly the intended titles, by TMDB id where the tracker
/// supplied one.
#[tokio::test]
async fn exactly_the_intended_titles_match() {
let (_dir, database) = wanted(&[
(693_134, "Dune: Part Two", 2024, false),
(933_260, "The Substance", 2024, false),
(438_631, "Dune", 2021, false),
])
.await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 3);
assert_eq!(
grabbed(&database).await,
vec![
(
"Dune".to_owned(),
"Dune.2021.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned()
),
(
"Dune: Part Two".to_owned(),
"Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP".to_owned()
),
(
"The Substance".to_owned(),
"La.Sustancia.2024.2160p.WEB-DL.DDP5.1.H.265-GRUPO".to_owned()
),
],
"one torrent per title, and the 1080p Dune: Part Two loses to the 2160p one"
);
assert_eq!(fake.added.lock().unwrap().len(), 3);
// The TV item and the film nobody asked for are not cached against
// any title: an unmatched release belongs to nothing. Both indexers
// carry the same four, under their own indexer id.
assert_eq!(
stored_releases(&database).await,
vec!["alpha-1", "alpha-2", "alpha-3", "alpha-4"]
);
let rows: i64 = sqlx::query_scalar("SELECT count(*) FROM releases")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(rows, 8);
}
/// The near miss. "Dune: Part Three" is one word from a release that is
/// in the feed, from the same franchise, in the same year.
#[tokio::test]
async fn a_near_miss_is_not_grabbed() {
let (_dir, database) = wanted(&[(9_999_999, "Dune: Part Three", 2024, false)]).await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert!(outcomes.is_empty());
assert!(fake.added.lock().unwrap().is_empty());
assert!(stored_releases(&database).await.is_empty());
}
/// §6.2: RSS never backs off, so nothing it does may spend one of
/// targeted search's attempts.
#[tokio::test]
async fn matching_never_touches_the_search_backoff() {
let (_dir, database) = wanted(&[
(693_134, "Dune: Part Two", 2024, false),
(9_999_999, "Dune: Part Three", 2024, false),
])
.await;
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let attempts: Vec<(i64, Option<String>)> =
sqlx::query_as("SELECT search_attempts, last_searched_at FROM movies ORDER BY id")
.fetch_all(database.pool())
.await
.unwrap();
assert_eq!(attempts, vec![(0, None), (0, None)]);
}
/// §6.3: `blocked` stops targeted search and leaves RSS matching on.
#[tokio::test]
async fn a_blocked_title_still_matches_rss() {
let (_dir, database) = wanted(&[(693_134, "Dune: Part Two", 2024, true)]).await;
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
assert_eq!(outcomes.len(), 1);
}
/// The property the whole lane exists for: cost is one call per indexer,
/// whatever the wanted list holds.
#[tokio::test]
async fn one_empty_query_call_per_indexer() {
let (_dir, database) = wanted(&[
(693_134, "Dune: Part Two", 2024, false),
(933_260, "The Substance", 2024, false),
(438_631, "Dune", 2021, false),
])
.await;
let indexer = prowlarr().await;
let (downloader, _fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
for id in INDEXERS {
let feeds = indexer
.received_requests()
.await
.unwrap()
.into_iter()
.filter(|request| {
request.url.path() == format!("/{id}/api")
&& request
.url
.query_pairs()
.any(|(name, value)| name == "t" && value == "search")
})
.count();
assert_eq!(feeds, 1);
}
}
}
crates/arr-daemon/tests/fixtures/rss.xml
Vendored
+61
View File
@@ -0,0 +1,61 @@
<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:torznab="http://torznab.com/schemas/2015/feed">
<channel>
<title>Alpha Tracker</title>
<item>
<title>Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP</title>
<guid>alpha-1</guid>
<link>https://indexer.invalid/download/alpha-1</link>
<pubDate>Sat, 22 Aug 2026 09:14:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-1" length="23622320128"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="41"/>
</item>
<item>
<title>Dune.Part.Two.2024.1080p.WEB-DL.DD5.1.H.264-OTHER</title>
<guid>alpha-2</guid>
<link>https://indexer.invalid/download/alpha-2</link>
<pubDate>Sat, 22 Aug 2026 08:02:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-2" length="8589934592"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="12"/>
</item>
<item>
<title>La.Sustancia.2024.2160p.WEB-DL.DDP5.1.H.265-GRUPO</title>
<guid>alpha-3</guid>
<link>https://indexer.invalid/download/alpha-3</link>
<pubDate>Sat, 22 Aug 2026 07:40:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-3" length="21474836480"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="30"/>
<torznab:attr name="tmdbid" value="933260"/>
</item>
<item>
<title>Dune.2021.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP</title>
<guid>alpha-4</guid>
<link>https://indexer.invalid/download/alpha-4</link>
<pubDate>Sat, 22 Aug 2026 06:55:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-4" length="22548578304"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="88"/>
</item>
<item>
<title>Dune.Prophecy.S01E03.2024.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
<guid>alpha-5</guid>
<link>https://indexer.invalid/download/alpha-5</link>
<pubDate>Sat, 22 Aug 2026 06:31:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-5" length="6442450944"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="55"/>
</item>
<item>
<title>Nosferatu.2024.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP</title>
<guid>alpha-6</guid>
<link>https://indexer.invalid/download/alpha-6</link>
<pubDate>Sat, 22 Aug 2026 05:12:00 +0000</pubDate>
<enclosure url="https://indexer.invalid/download/alpha-6" length="20401094656"
type="application/x-bittorrent"/>
<torznab:attr name="seeders" value="14"/>
</item>
</channel>
</rss>