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

feat: season-pack vs per-episode grab selection (#94)
ci / web (push) Successful in 24s
Details
e2e / e2e (push) Successful in 54s
Details
ci / rust (push) Successful in 2m11s
Details

Browse Source
This commit was merged in pull request #94.
This commit is contained in:
naps62-yolo (agent)
2026-08-23 01:40:53 +01:00
parent 35391f02fc
commit 9f812fac11
28 changed files with 2855 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-daemon/src/grab.rs
+249 -96
View File
@@ -118,7 +118,7 @@ impl GrabAction {
}
async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let mut outcomes = self.track_sent_grabs(database).await?;
let mut outcomes = self.grabber.track_sent_grabs(database).await?;
let gaps = pending_movies(database).await?;
if gaps.is_empty() {
return Ok(outcomes);
@@ -247,61 +247,6 @@ impl GrabAction {
}))
}
/// Move grabs Transmission reports as complete out of `sent`.
///
/// Transmission is authoritative and its view is rebuilt on every tick
/// rather than cached (§8), so this is also what reconstructs in-flight
/// state after a restart.
async fn track_sent_grabs(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let sent = sqlx::query!(
r#"SELECT id AS "id!: i64", infohash AS "infohash!: String", target_id AS "target_id!: i64"
FROM grabs WHERE state = 'sent'"#
)
.fetch_all(database.pool())
.await?;
if sent.is_empty() {
return Ok(Vec::new());
}
let torrents: HashMap<String, f64> = self
.grabber
.transmission
.list_torrents()
.await?
.into_iter()
.map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent.progress))
.collect();
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.
continue;
};
if *progress < 1.0 {
continue;
}
sqlx::query!(
"UPDATE grabs SET state = 'downloaded' WHERE id = ?",
grab.id
)
.execute(database.pool())
.await?;
outcomes.push(Outcome::new(
format!("grab {} downloaded, still marked sent", grab.id),
format!("marked grab {} downloaded", grab.id),
));
tracing::info!(
grab_id = grab.id,
movie_id = grab.target_id,
"download complete"
);
}
Ok(outcomes)
}
/// Search every indexer for one title, cache each candidate with its
/// verdict and score (§9.3), and return the eligible ones best first.
///
@@ -409,7 +354,7 @@ impl GrabAction {
.send_winner(
database,
&GrabTarget {
movie_id: movie.id,
scope: GrabScope::Movie { movie_id: movie.id },
title: &movie.title,
counts_as_attempt: true,
},
@@ -436,13 +381,57 @@ pub(crate) struct Grabber {
/// The title a winning release is being grabbed for.
#[derive(Debug)]
pub(crate) struct GrabTarget<'a> {
pub(crate) movie_id: i64,
pub(crate) scope: GrabScope,
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,
}
/// What a grab targets: the `grabs` row's kind and id, plus the episodes the
/// torrent covers — attempts and state changes land on those leaves (§4.1).
#[derive(Debug)]
pub(crate) enum GrabScope {
Movie {
movie_id: i64,
},
Episode {
episode_id: i64,
},
/// A season pack: one torrent, one `grabs` row on the season, every
/// missing wanted episode it covers flipped to downloading.
Season {
season_id: i64,
episode_ids: Vec<i64>,
},
}
impl GrabScope {
fn target_kind(&self) -> &'static str {
match self {
Self::Movie { .. } => "movie",
Self::Episode { .. } => "episode",
Self::Season { .. } => "season",
}
}
fn target_id(&self) -> i64 {
match self {
Self::Movie { movie_id } => *movie_id,
Self::Episode { episode_id } => *episode_id,
Self::Season { season_id, .. } => *season_id,
}
}
fn episode_ids(&self) -> &[i64] {
match self {
Self::Movie { .. } => &[],
Self::Episode { episode_id } => std::slice::from_ref(episode_id),
Self::Season { episode_ids, .. } => episode_ids,
}
}
}
impl Grabber {
pub(crate) fn new(
transmission: TransmissionClient,
@@ -456,15 +445,76 @@ impl Grabber {
}
}
/// Move grabs Transmission reports as complete out of `sent`, whatever
/// they target.
///
/// Transmission is authoritative and its view is rebuilt on every tick
/// rather than cached (§8), so this is also what reconstructs in-flight
/// state after a restart. Both grab actions call it; whichever runs first
/// does the work and the other finds nothing.
pub(crate) async fn track_sent_grabs(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
let sent = sqlx::query!(
r#"SELECT id AS "id!: i64", infohash AS "infohash!: String",
target_kind AS "target_kind!: String", target_id AS "target_id!: i64"
FROM grabs WHERE state = 'sent'"#
)
.fetch_all(database.pool())
.await?;
if sent.is_empty() {
return Ok(Vec::new());
}
let torrents: HashMap<String, f64> = self
.transmission
.list_torrents()
.await?
.into_iter()
.map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent.progress))
.collect();
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.
continue;
};
if *progress < 1.0 {
continue;
}
sqlx::query!(
"UPDATE grabs SET state = 'downloaded' WHERE id = ?",
grab.id
)
.execute(database.pool())
.await?;
outcomes.push(Outcome::new(
format!("grab {} downloaded, still marked sent", grab.id),
format!("marked grab {} downloaded", grab.id),
));
tracing::info!(
grab_id = grab.id,
target_kind = grab.target_kind,
target_id = grab.target_id,
"download complete"
);
}
Ok(outcomes)
}
async fn record_attempt(
&self,
database: &Db,
target: &GrabTarget<'_>,
) -> Result<(), GrabError> {
if target.counts_as_attempt {
record_search(database, target.movie_id).await?;
if !target.counts_as_attempt {
return Ok(());
}
match &target.scope {
GrabScope::Movie { movie_id } => record_search(database, *movie_id).await,
scope => record_episode_search(database, scope.episode_ids()).await,
}
Ok(())
}
/// Add the winning release to Transmission and record the grab.
@@ -513,29 +563,49 @@ impl Grabber {
// A duplicate here is the restart case: the torrent was added before
// the process died. `DO NOTHING` keeps the original row.
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 (?, 'movie', ?, ?, 'sent')
VALUES (?, ?, ?, ?, 'sent')
ON CONFLICT (infohash) DO NOTHING
RETURNING id AS "id!: i64""#,
winner.id,
target.movie_id,
target_kind,
target_id,
infohash
)
.fetch_optional(database.pool())
.await?;
sqlx::query!(
"UPDATE movies SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
target.movie_id
)
.execute(database.pool())
.await?;
match &target.scope {
GrabScope::Movie { movie_id } => {
sqlx::query!(
"UPDATE movies SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
movie_id
)
.execute(database.pool())
.await?;
}
scope => {
for episode_id in scope.episode_ids() {
sqlx::query!(
"UPDATE episodes SET state = 'downloading',
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
episode_id
)
.execute(database.pool())
.await?;
}
}
}
let Some(inserted) = inserted else {
tracing::info!(
movie_id = target.movie_id,
target_kind,
target_id,
infohash,
"grab already recorded for this torrent"
);
@@ -543,7 +613,8 @@ impl Grabber {
};
tracing::info!(
movie_id = target.movie_id,
target_kind,
target_id,
title = target.title,
release = winner.name,
score = winner.score,
@@ -552,7 +623,7 @@ impl Grabber {
"grabbed"
);
Ok(Some(Outcome::new(
format!("movie {} wanted with no file", target.movie_id),
format!("{target_kind} {target_id} wanted with no file"),
format!("grabbed {} as grab {}", winner.name, inserted.id),
)))
}
@@ -592,7 +663,7 @@ impl Grabber {
// The earlier grab's torrent, still working off its seeding
// obligation (§7.3). Nothing here deletes a torrent.
tracing::warn!(
movie_id = target.movie_id,
title = target.title,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; left seeding"
@@ -602,7 +673,7 @@ impl Grabber {
// obligation and has nothing on disk worth keeping.
self.transmission.remove_torrent(added.id, true).await?;
tracing::warn!(
movie_id = target.movie_id,
title = target.title,
release = release_name,
infohash = added.hash,
"blacklisted torrent re-listed under a new name; removed"
@@ -645,7 +716,7 @@ pub(crate) struct Eligible {
pub(crate) indexer_id: i64,
pub(crate) guid: String,
pub(crate) name: String,
download_url: String,
pub(crate) download_url: String,
pub(crate) score: i64,
}
@@ -703,13 +774,18 @@ async fn pending_movies(database: &Db) -> Result<Vec<PendingMovie>, GrabError> {
}
fn search_due(movie: &PendingMovie) -> bool {
let Some(last_searched_at) = &movie.last_searched_at else {
backoff_elapsed(movie.search_attempts, movie.last_searched_at.as_deref())
}
/// The §6.2 targeted-search backoff, shared by movie and episode search.
pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&str>) -> bool {
let Some(last_searched_at) = last_searched_at else {
return true;
};
let Ok(last_searched_at) = chrono::DateTime::parse_from_rfc3339(last_searched_at) else {
return true;
};
let backoff = match movie.search_attempts {
let backoff = match search_attempts {
1 => chrono::TimeDelta::hours(1),
2 => chrono::TimeDelta::hours(6),
3 => chrono::TimeDelta::days(1),
@@ -756,6 +832,67 @@ pub(crate) async fn store_release(
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
let (release_id, eligible) = classify_and_store(
database,
release,
policy,
overrides,
original_language,
blacklist,
)
.await?;
sqlx::query!(
"INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
movie_id,
release_id
)
.execute(database.pool())
.await?;
Ok(eligible)
}
/// The TV counterpart: one release row, associated with every episode the
/// claim covers — a season pack matches the whole season.
pub(crate) async fn store_episode_release(
database: &Db,
episode_ids: &[i64],
release: &SearchRelease,
policy: &Policy,
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
let (release_id, eligible) = classify_and_store(
database,
release,
policy,
overrides,
original_language,
blacklist,
)
.await?;
for episode_id in episode_ids {
sqlx::query!(
"INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
episode_id,
release_id
)
.execute(database.pool())
.await?;
}
Ok(eligible)
}
async fn classify_and_store(
database: &Db,
release: &SearchRelease,
policy: &Policy,
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<(i64, Option<Eligible>), GrabError> {
let parsed = arr_parse::parse(&release.name);
let evaluation = evaluate(
policy,
@@ -824,26 +961,20 @@ pub(crate) async fn store_release(
.fetch_one(database.pool())
.await?;
sqlx::query!(
"INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
movie_id,
id
)
.execute(database.pool())
.await?;
if verdict != "eligible" {
return Ok(None);
return Ok((id, None));
}
Ok(Some(Eligible {
Ok((
id,
indexer_id: release.indexer_id,
guid: release.guid.clone(),
name: release.name.clone(),
download_url: release.download_url.clone(),
score,
}))
Some(Eligible {
id,
indexer_id: release.indexer_id,
guid: release.guid.clone(),
name: release.name.clone(),
download_url: release.download_url.clone(),
score,
}),
))
}
/// Record that the title was searched, so the next tick takes a different one.
@@ -861,6 +992,28 @@ async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> {
Ok(())
}
/// The episode-level backoff counter (§6.2), one bump per searched episode. A
/// season-pack search touches every episode it was trying to satisfy, so the
/// whole season backs off together.
pub(crate) async fn record_episode_search(
database: &Db,
episode_ids: &[i64],
) -> Result<(), GrabError> {
for episode_id in episode_ids {
sqlx::query!(
"UPDATE episodes
SET search_attempts = search_attempts + 1,
last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
episode_id
)
.execute(database.pool())
.await?;
}
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 {