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

Merge main into blitz/subtitles

Browse Source 
Feedback pass 2 and the size-band work landed on main while this branch
was finishing. Brings them in ahead of the merge back.

# Conflicts:
#	crates/arr-api/src/movies.rs
#	crates/arr-api/src/state.rs
#	crates/arr-daemon/src/main.rs
#	web/src/main.ts
This commit is contained in:
Miguel Palhas
2026-08-25 17:53:18 +01:00
parent d7e9f710e0 8947febbff
commit 34e5663920
76 changed files with 7184 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-daemon/src/attention.rs
+251 -16
View File
@@ -2,6 +2,15 @@
//! the no-PT-source queue, or hard-failed twice on different releases (the
//! same queues `GET /api/queues/attention` reports, §9.3).
//!
//! §5.7 sets the bar for the hard-fail side: two failures on *different*
//! releases, both inside `arr_db::ATTENTION_WINDOW`, against a target still
//! waiting for a file. One bad torrent is not a decision, a failure already
//! dealt with ages out (#226), and a target that has since been acquired
//! leaves at once (#238). The season lane reads that last condition off its
//! episodes, which is where intent lives (§4.1). `GET /api/queues/attention`
//! filters identically, or the two channels tell the operator different
//! stories.
//!
//! Edge-triggered per title: it notifies once when the title enters either
//! queue, and is forgotten once it leaves both, so a future re-entry notifies
//! again. A series notifies as its series, never per episode — a broken
@@ -34,8 +43,8 @@ struct TvEntry {
no_pt_source: Vec<i64>,
/// Episodes two different releases hard-failed post-probe (§5.7).
hard_failed_episodes: Vec<i64>,
/// Seasons whose pack grab hard-failed, sending the season back to
/// per-episode grabbing.
/// Seasons two different pack releases hard-failed on (§5.7), sending the
/// season back to per-episode grabbing.
failed_season_packs: Vec<i64>,
}
@@ -59,8 +68,8 @@ impl TvEntry {
if !self.failed_season_packs.is_empty() {
parts.push(plural(
self.failed_season_packs.len(),
"season pack hard-failed",
"season packs hard-failed",
"season hard-failed twice on different packs",
"seasons hard-failed twice on different packs",
));
}
parts.join("; ")
@@ -134,8 +143,10 @@ impl AttentionAction {
AND (SELECT count(DISTINCT g.release_id)
FROM grabs g
WHERE g.target_kind = 'movie' AND g.target_id = movies.id
AND g.state = 'failed') >= 2
"#
AND g.state = 'failed'
AND g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)) >= 2
"#,
arr_db::ATTENTION_WINDOW
)
.fetch_all(database.pool())
.await?;
@@ -209,9 +220,14 @@ fn tv_entry(
/// TV roll-up (§9.5): every queued series with what put it there — wanted
/// episodes whose every candidate was rejected for language, episodes two
/// different releases hard-failed post-probe, and seasons whose pack grab
/// hard-failed. One entry per series, so the notification can be one per
/// different releases hard-failed post-probe, and seasons two different packs
/// hard-failed on. One entry per series, so the notification can be one per
/// series however long the broken season is.
///
/// Both hard-fail lanes carry §5.7's liveness condition: an episode is queued
/// only while `wanted` and not `available`, and a season only while at least
/// one of its episodes is. A season pack that failed twice and then fell back
/// to per-episode grabbing (§6.2) drops out as those episodes land.
async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntry)>, sqlx::Error> {
let mut tv = HashMap::new();
@@ -250,10 +266,12 @@ async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntr
JOIN seasons se ON se.id = e.season_id
JOIN series s ON s.id = se.series_id
WHERE g.state = 'failed'
AND g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)
AND e.wanted = 1 AND e.state != 'available'
GROUP BY s.id, s.title, s.year, e.id
HAVING count(DISTINCT g.release_id) >= 2
"#
"#,
arr_db::ATTENTION_WINDOW
)
.fetch_all(database.pool())
.await?;
@@ -271,8 +289,16 @@ async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntr
JOIN seasons se ON g.target_kind = 'season' AND se.id = g.target_id
JOIN series s ON s.id = se.series_id
WHERE g.state = 'failed'
AND g.failed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)
AND EXISTS (
SELECT 1 FROM episodes e
WHERE e.season_id = se.id
AND e.wanted = 1 AND e.state != 'available'
)
GROUP BY s.id, s.title, s.year, se.id
"#
HAVING count(DISTINCT g.release_id) >= 2
"#,
arr_db::ATTENTION_WINDOW
)
.fetch_all(database.pool())
.await?;
@@ -382,13 +408,32 @@ mod tests {
series_id
}
/// A wanted, missing episode: the least that keeps its season live for
/// §5.7's liveness condition.
async fn insert_wanted_episode(database: &Db, season_id: i64, number: i64) -> i64 {
sqlx::query_scalar(
"INSERT INTO episodes (season_id, number, title, wanted, state)
VALUES (?, ?, ?, 1, 'missing') RETURNING id",
)
.bind(season_id)
.bind(number)
.bind(format!("Episode {number}"))
.fetch_one(database.pool())
.await
.unwrap()
}
/// A failed grab by `release_guid` against `target_kind`/`target_id`,
/// standing in for what the import tick leaves behind post-probe.
async fn insert_failed_grab(
/// grabbed `grab_age_days` ago and failed `fail_age_days` ago, so §5.7's
/// window — which runs from the failure — can be exercised without
/// waiting a month.
async fn insert_dated_failed_grab(
database: &Db,
target_kind: &str,
target_id: i64,
release_guid: &str,
grab_age_days: i64,
fail_age_days: i64,
) {
let release_id: i64 = sqlx::query_scalar(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
@@ -400,18 +445,53 @@ mod tests {
.await
.unwrap();
sqlx::query(
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
VALUES (?, ?, ?, ?, 'failed')",
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state, grabbed_at, failed_at)
VALUES (?, ?, ?, ?, 'failed',
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?),
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?))",
)
.bind(release_id)
.bind(target_kind)
.bind(target_id)
.bind(format!("hash-{release_guid}"))
.bind(format!("-{grab_age_days} days"))
.bind(format!("-{fail_age_days} days"))
.execute(database.pool())
.await
.unwrap();
}
/// A failed grab by `release_guid` against `target_kind`/`target_id`,
/// failed `age_days` in the past.
async fn insert_aged_failed_grab(
database: &Db,
target_kind: &str,
target_id: i64,
release_guid: &str,
age_days: i64,
) {
insert_dated_failed_grab(
database,
target_kind,
target_id,
release_guid,
age_days,
age_days,
)
.await;
}
/// A failed grab by `release_guid` against `target_kind`/`target_id`,
/// standing in for what the import tick leaves behind post-probe.
async fn insert_failed_grab(
database: &Db,
target_kind: &str,
target_id: i64,
release_guid: &str,
) {
insert_dated_failed_grab(database, target_kind, target_id, release_guid, 0, 0).await;
}
async fn action(server: &MockServer) -> AttentionAction {
Mock::given(method("POST"))
.respond_with(ResponseTemplate::new(200))
@@ -548,9 +628,10 @@ mod tests {
.unwrap();
insert_failed_grab(&database, "episode", episode_id, "first").await;
insert_failed_grab(&database, "episode", episode_id, "second").await;
// The pack's failure sent this season back to per-episode grabbing;
// it queues the same series, so it must not double the message.
// The packs' failures sent this season back to per-episode grabbing;
// they queue the same series, so it must not double the message.
insert_failed_grab(&database, "season", season_id, "pack").await;
insert_failed_grab(&database, "season", season_id, "pack-two").await;
let server = MockServer::start().await;
let action = action(&server).await;
@@ -631,4 +712,158 @@ mod tests {
assert_eq!(second.len(), 0, "leaves the queue once imported");
assert_eq!(server.received_requests().await.unwrap().len(), 1);
}
/// §5.7: the season lane holds to the same two-distinct-releases bar the
/// episode lane does, so one bad pack does not notify (#226).
#[tokio::test]
async fn one_failed_season_pack_does_not_notify() {
let (_dir, database) = seeded_database().await;
insert_no_pt_source_series(&database, 1, 0).await;
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
insert_wanted_episode(&database, season_id, 1).await;
insert_failed_grab(&database, "season", season_id, "pack").await;
let server = MockServer::start().await;
let action = action(&server).await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
0,
"one failed pack is the blacklist working, not a decision"
);
insert_failed_grab(&database, "season", season_id, "pack-two").await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
1,
"two distinct packs hard-failed: the operator decides"
);
}
/// §5.7: the queue only holds targets still waiting for a file. A season
/// whose packs both hard-failed falls back to per-episode grabbing (§6.2);
/// once every episode has landed the system worked, so the season leaves
/// the queue at once rather than notifying for 30 days (#238).
#[tokio::test]
async fn a_fully_acquired_season_leaves_the_queue() {
let (_dir, database) = seeded_database().await;
insert_no_pt_source_series(&database, 1, 0).await;
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
insert_wanted_episode(&database, season_id, 1).await;
insert_wanted_episode(&database, season_id, 2).await;
insert_failed_grab(&database, "season", season_id, "pack").await;
insert_failed_grab(&database, "season", season_id, "pack-two").await;
let server = MockServer::start().await;
let action = action(&server).await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
1,
"two packs failed and the season still has episodes missing"
);
// Per-episode grabbing got the first one. Still a gap, still queued.
sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ? AND number = 1")
.bind(season_id)
.execute(database.pool())
.await
.unwrap();
assert_eq!(
action.tick(&database).await.unwrap().len(),
0,
"already notified, and still queued"
);
sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ?")
.bind(season_id)
.execute(database.pool())
.await
.unwrap();
assert_eq!(
action.tick(&database).await.unwrap().len(),
0,
"every episode acquired: nothing left to decide"
);
// Proof it actually left rather than merely staying quiet: a season
// still queued would not notify again on re-entry.
sqlx::query("UPDATE episodes SET state = 'missing' WHERE season_id = ? AND number = 2")
.bind(season_id)
.execute(database.pool())
.await
.unwrap();
assert_eq!(
action.tick(&database).await.unwrap().len(),
1,
"broken again: re-enters the queue and notifies"
);
assert_eq!(server.received_requests().await.unwrap().len(), 2);
}
/// §5.7: a failure counts for 30 days, so a season dealt with leaves the
/// queue instead of sitting in it forever (#226).
#[tokio::test]
async fn season_failures_older_than_the_window_do_not_notify() {
let (_dir, database) = seeded_database().await;
insert_no_pt_source_series(&database, 1, 0).await;
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
insert_wanted_episode(&database, season_id, 1).await;
insert_aged_failed_grab(&database, "season", season_id, "old-one", 40).await;
insert_aged_failed_grab(&database, "season", season_id, "old-two", 35).await;
let server = MockServer::start().await;
let action = action(&server).await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
0,
"failures older than the window are history, not attention"
);
insert_aged_failed_grab(&database, "season", season_id, "new-one", 0).await;
insert_aged_failed_grab(&database, "season", season_id, "new-two", 0).await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
1,
"still breaking: back in the queue"
);
}
/// §5.7/#239: the window runs from the failure, not the grab. A torrent
/// that stalls on a slow swarm for five weeks and then hard-fails at
/// import is fresh evidence the target is broken, however old the grab.
#[tokio::test]
async fn a_grab_stalled_past_the_window_before_failing_still_counts() {
let (_dir, database) = seeded_database().await;
insert_no_pt_source_series(&database, 1, 0).await;
let season_id: i64 = sqlx::query_scalar("SELECT id FROM seasons WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
insert_wanted_episode(&database, season_id, 1).await;
// Both grabbed 40 days ago — outside the window — but failed today.
insert_dated_failed_grab(&database, "season", season_id, "stalled-one", 40, 0).await;
insert_dated_failed_grab(&database, "season", season_id, "stalled-two", 40, 0).await;
let server = MockServer::start().await;
let action = action(&server).await;
assert_eq!(
action.tick(&database).await.unwrap().len(),
1,
"grab age is irrelevant: two fresh failures queue the target"
);
}
}
crates/arr-daemon/src/grab.rs
+55 -1
View File
@@ -1192,6 +1192,7 @@ pub(crate) async fn store_release(
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
// A movie is one episode's worth and is never runtime-scaled (§5.5).
let (release_id, eligible) = classify_and_store(
database,
release,
@@ -1200,6 +1201,7 @@ pub(crate) async fn store_release(
original_language,
blacklist,
1,
0,
)
.await?;
sqlx::query!(
@@ -1241,6 +1243,11 @@ pub(crate) async fn store_episode_release(
_ => BTreeMap::new(),
};
let episode_count = claimed_episode_count(claim.as_ref(), &season_lengths);
// §5.5: the size bands scale by the series' minutes per episode.
let runtime_minutes = match episode_ids.first() {
Some(&episode_id) => series_runtime_of(database, episode_id).await?,
None => 0,
};
let (release_id, eligible) = classify_and_store(
database,
release,
@@ -1249,6 +1256,7 @@ pub(crate) async fn store_episode_release(
original_language,
blacklist,
episode_count,
runtime_minutes,
)
.await?;
for episode_id in episode_ids {
@@ -1293,6 +1301,26 @@ async fn season_lengths_of(
.collect())
}
/// The minutes-per-episode of the series one covered episode belongs to
/// (`DESIGN.md` §5.5): the scale factor for its size bands. Zero when the
/// series has no known runtime, which applies the bands unscaled.
async fn series_runtime_of(database: &Db, episode_id: i64) -> Result<u32, GrabError> {
let minutes = sqlx::query_scalar!(
r#"SELECT s.runtime_minutes FROM series s
WHERE s.id = (SELECT s2.series_id FROM episodes e
JOIN seasons s2 ON s2.id = e.season_id
WHERE e.id = ?)"#,
episode_id
)
.fetch_optional(database.pool())
.await?
.flatten();
Ok(minutes
.and_then(|minutes| u32::try_from(minutes).ok())
.unwrap_or(0))
}
#[allow(clippy::too_many_arguments)]
async fn classify_and_store(
database: &Db,
release: &SearchRelease,
@@ -1301,6 +1329,7 @@ async fn classify_and_store(
original_language: &Language,
blacklist: &Blacklist,
episode_count: u32,
runtime_minutes: u32,
) -> Result<(i64, Option<Eligible>), GrabError> {
let parsed = arr_parse::parse(&release.name);
let evaluation = evaluate(
@@ -1310,6 +1339,7 @@ async fn classify_and_store(
Candidate::PreGrab(&parsed),
release.size,
episode_count,
runtime_minutes,
);
let scored = score(
policy,
@@ -1317,6 +1347,7 @@ async fn classify_and_store(
release.size.unwrap_or_default(),
release.seeders.unwrap_or_default(),
episode_count,
runtime_minutes,
);
// A release that did not say its size is not a tiny one: scoring it
// against the band's floor would bury it. Same treatment as the manual
@@ -1503,10 +1534,17 @@ fn search_query(movie: &PendingMovie) -> String {
)
}
/// The `verdict` and `rejected_rule` columns for a verdict.
///
/// A waiver names the rule it relaxed (#211). §5.7 calls a soft fail
/// "watchable but not what was asked", and which rule was relaxed is the
/// whole content of that sentence, so §9.3's deck can name it the way it
/// names a rejection. Rows written before 0032 hold `NULL` there and stay
/// readable.
fn verdict_columns(verdict: &Verdict) -> (&'static str, Option<String>) {
match verdict {
Verdict::Eligible => ("eligible", None),
Verdict::Waived(_) => ("waived", None),
Verdict::Waived(rule) => ("waived", Some(rule.name())),
Verdict::Rejected(rule) => ("rejected", Some(rule.name())),
}
}
@@ -1998,6 +2036,22 @@ mod tests {
);
}
/// #211: a waiver names the rule it relaxed, the same as a rejection, so
/// §9.3's deck shows what was given up instead of a bare `waived`.
/// Migration 0032 relaxed the constraint that forbade it.
#[test]
fn a_waiver_records_the_rule_it_relaxed() {
assert_eq!(
verdict_columns(&Verdict::Waived(arr_core::Rule::Size)),
("waived", Some("size".to_owned()))
);
assert_eq!(verdict_columns(&Verdict::Eligible), ("eligible", None));
assert_eq!(
verdict_columns(&Verdict::Rejected(arr_core::Rule::RequiredAudio)),
("rejected", Some("required_audio".to_owned()))
);
}
/// Every candidate is cached with its verdict, which is what the manual
/// search view and the attention queues read (§9.3).
#[tokio::test]
crates/arr-daemon/src/import.rs
+40 -10
View File
@@ -308,6 +308,7 @@ impl ImportAction {
Candidate::PostDownload(&feature.media),
Some(feature.size),
1,
0,
);
let waiver: Option<Rule> = match evaluation.verdict {
Verdict::Rejected(rule) => {
@@ -477,6 +478,10 @@ impl ImportAction {
// §5.6 second phase of truth, over every file that would be
// imported, before anything is placed: one hard failure condemns
// the whole release (§5.7), not the episodes.
let runtime_minutes = pending
.runtime_minutes
.and_then(|minutes| u32::try_from(minutes).ok())
.unwrap_or(0);
let mut imports = Vec::new();
for assignment in assignments {
if assignment.episode.has_file {
@@ -496,6 +501,7 @@ impl ImportAction {
Candidate::PostDownload(&assignment.file.media),
Some(assignment.file.size),
1,
runtime_minutes,
);
let waiver = match evaluation.verdict {
Verdict::Rejected(rule) => {
@@ -615,7 +621,10 @@ impl ImportAction {
)
.await?;
sqlx::query!(
"UPDATE grabs SET state = 'failed' WHERE id = ?",
"UPDATE grabs
SET state = 'failed',
failed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
pending.grab_id
)
.execute(database.pool())
@@ -753,7 +762,10 @@ impl ImportAction {
)
.await?;
sqlx::query!(
"UPDATE grabs SET state = 'failed' WHERE id = ?",
"UPDATE grabs
SET state = 'failed',
failed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
pending.grab_id
)
.execute(database.pool())
@@ -938,6 +950,10 @@ struct PendingTvImport {
series_title: String,
series_year: Option<i64>,
original_language: Option<String>,
/// §5.5: the series' minutes per episode, scaling the size bands the
/// same way the pre-grab verdict scaled them. `None` applies them
/// unscaled.
runtime_minutes: Option<i64>,
release_name: String,
}
@@ -974,6 +990,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
s.title AS "series_title!: String",
s.year AS "series_year",
s.original_language,
s.runtime_minutes,
r.name AS "release_name!: String"
FROM grabs g
JOIN episodes e ON e.id = g.target_id
@@ -997,6 +1014,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
series_title: row.series_title,
series_year: row.series_year,
original_language: row.original_language,
runtime_minutes: row.runtime_minutes,
release_name: row.release_name,
}));
@@ -1011,6 +1029,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
s.title AS "series_title!: String",
s.year AS "series_year",
s.original_language,
s.runtime_minutes,
r.name AS "release_name!: String"
FROM grabs g
JOIN seasons se ON se.id = g.target_id
@@ -1033,6 +1052,7 @@ async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, Impor
series_title: row.series_title,
series_year: row.series_year,
original_language: row.original_language,
runtime_minutes: row.runtime_minutes,
release_name: row.release_name,
}));
@@ -1557,11 +1577,16 @@ mod tests {
assert_eq!(normalised, arr_parse::normalise(RELEASE_NAME));
assert_eq!(reason, "dolby_vision_profile");
let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
.fetch_one(h.database.pool())
.await
.unwrap();
let (grab_state, failed_at): (String, Option<String>) =
sqlx::query_as("SELECT state, failed_at FROM grabs")
.fetch_one(h.database.pool())
.await
.unwrap();
assert_eq!(grab_state, "failed");
assert!(
failed_at.is_some(),
"§5.7's window runs from the failure, so the failure is stamped"
);
let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
.fetch_one(h.database.pool())
.await
@@ -2195,11 +2220,16 @@ mod tests {
)],
"only the release is blacklisted, never the season"
);
let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
.fetch_one(h.database.pool())
.await
.unwrap();
let (grab_state, failed_at): (String, Option<String>) =
sqlx::query_as("SELECT state, failed_at FROM grabs")
.fetch_one(h.database.pool())
.await
.unwrap();
assert_eq!(grab_state, "failed");
assert!(
failed_at.is_some(),
"§5.7's window runs from the failure, so the failure is stamped"
);
let states: Vec<(String, bool)> =
sqlx::query_as("SELECT state, wanted FROM episodes ORDER BY number")
.fetch_all(h.database.pool())
crates/arr-daemon/src/rss.rs
+58 -2
View File
@@ -625,9 +625,11 @@ async fn pack_allowed(database: &Db, season_id: i64) -> Result<bool, GrabError>
)
.fetch_all(database.pool())
.await?;
// §6.2's ladder runs from the failure, not the grab (#245), with
// `grabbed_at` as the fallback for rows older than #239's column.
let failed_packs = sqlx::query!(
r#"SELECT count(*) AS "failures!: i64",
max(grabbed_at) AS "last_failed_at?: String"
r#"SELECT count(*) AS "failures!: i64",
max(coalesce(failed_at, grabbed_at)) AS "last_failed_at?: String"
FROM grabs
WHERE target_kind = 'season' AND target_id = ? AND state = 'failed'"#,
season_id
@@ -1164,6 +1166,60 @@ mod tests {
);
}
/// §6.2, issue #245: the RSS lane reads the same ladder, anchored on the
/// failure. A pack sent five weeks ago and condemned at import ten
/// minutes ago holds the lane shut, where anchoring on the grab would
/// have handed it the very release class that just failed.
#[tokio::test]
async fn a_pack_that_stalled_for_weeks_stays_backed_off_on_rss() {
let (_dir, database) = wanted(&[]).await;
let (season_id, episodes) =
wanted_series(&database, &["2024-04-11", "2024-04-18", "2024-04-25"]).await;
let release_id: i64 = sqlx::query_scalar(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
VALUES (7, 'oldpack', 'Fallout.S01.2160p.WEB-DL.OLD', 85899345920,
'https://tracker/oldpack.torrent', '{}', 'eligible')
RETURNING id",
)
.fetch_one(database.pool())
.await
.unwrap();
sqlx::query(
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state,
grabbed_at, failed_at)
VALUES (?, 'season', ?, 'dead', 'failed',
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-35 days'),
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-10 minutes'))",
)
.bind(release_id)
.bind(season_id)
.execute(database.pool())
.await
.unwrap();
let indexer = prowlarr(TV_FEED).await;
let (downloader, _fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let sent: Vec<(String, i64, String)> = tv_grabs(&database)
.await
.into_iter()
.filter(|(_, _, state)| state == "sent")
.collect();
assert!(
!sent.iter().any(|(kind, _, _)| kind == "season"),
"the failure is ten minutes old, so the pack lane is shut: {sent:?}"
);
assert_eq!(
sent,
vec![
("episode".to_owned(), episodes[0], "sent".to_owned()),
("episode".to_owned(), episodes[1], "sent".to_owned()),
("episode".to_owned(), episodes[2], "sent".to_owned()),
]
);
}
/// §6.2 with #117's guard: an episode already on disk keeps the season
/// per-episode here too — the pack is skipped and the open gaps take
/// their singles.
crates/arr-daemon/src/series_refresh.rs
+66 -2
View File
@@ -75,7 +75,7 @@ impl SeriesRefreshAction {
title AS "title!: String", year, original_language,
root_id AS "root_id!: i64", auto_track AS "auto_track!: bool",
upstream_ended AS "upstream_ended!: bool", metadata_refreshed_at,
poster_path, backdrop_path, vote_average
poster_path, backdrop_path, vote_average, runtime_minutes
FROM series
ORDER BY metadata_refreshed_at IS NOT NULL, metadata_refreshed_at, id"#
)
@@ -118,7 +118,7 @@ impl SeriesRefreshAction {
title AS "title!: String", year, original_language,
root_id AS "root_id!: i64", auto_track AS "auto_track!: bool",
upstream_ended AS "upstream_ended!: bool", metadata_refreshed_at,
poster_path, backdrop_path, vote_average
poster_path, backdrop_path, vote_average, runtime_minutes
FROM series WHERE id = ?"#,
series_id
)
@@ -250,6 +250,20 @@ impl SeriesRefreshAction {
changed = true;
}
}
// §5.5: the size bands scale by this. TMDB's `episode_run_time` is
// frequently empty; a known value is never overwritten by a missing
// one, so a series keeps its runtime across TMDB's blank spells.
let runtime = metadata.episode_runtime.map(i64::from);
if runtime.is_some() && runtime != stale.runtime_minutes {
sqlx::query!(
"UPDATE series SET runtime_minutes = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
runtime,
stale.id
)
.execute(&mut *executor)
.await?;
changed = true;
}
let ended = is_upstream_ended(&metadata.status);
if ended != stale.upstream_ended {
sqlx::query!(
@@ -547,6 +561,7 @@ struct DueSeries {
poster_path: Option<String>,
backdrop_path: Option<String>,
vote_average: Option<f64>,
runtime_minutes: Option<i64>,
}
/// TMDB numbers are unbounded; ours are `u16` (`CHECK (number >= 0)`,
/// STRICT). A number past `u16::MAX` cannot match anything real and would
@@ -817,6 +832,55 @@ mod tests {
assert_eq!(vote, Some(8.417));
}
/// §5.5: the refresh stores the minutes-per-episode the size bands scale
/// by, and a later refresh with TMDB's frequently-empty
/// `episode_run_time` never blanks a known value.
#[tokio::test]
async fn refresh_stores_the_episode_runtime_and_keeps_it_over_blanks() {
let (_dir, database) = seeded_series(false).await;
let server = MockServer::start().await;
let body = |episode_run_time: serde_json::Value| {
json!({
"id": 82_728,
"name": "Bluey",
"original_language": "en",
"first_air_date": "2018-10-01",
"status": "Returning Series",
"episode_run_time": episode_run_time,
"seasons": []
})
};
Mock::given(method("GET"))
.and(path("/tv/82728"))
.respond_with(ResponseTemplate::new(200).set_body_json(body(json!([7]))))
.mount(&server)
.await;
action(&server).tick(&database).await.unwrap();
let runtime: Option<i64> =
sqlx::query_scalar("SELECT runtime_minutes FROM series WHERE tmdb_id = 82728")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(runtime, Some(7));
server.reset().await;
Mock::given(method("GET"))
.and(path("/tv/82728"))
.respond_with(ResponseTemplate::new(200).set_body_json(body(json!([]))))
.mount(&server)
.await;
expire_refresh(&database).await;
action(&server).tick(&database).await.unwrap();
let runtime: Option<i64> =
sqlx::query_scalar("SELECT runtime_minutes FROM series WHERE tmdb_id = 82728")
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(runtime, Some(7));
}
/// #160. A series' first refresh reveals its back catalogue, but §4.1
/// never tracks what was already there at add time: nothing is tracked,
/// nothing arrives wanted.
crates/arr-daemon/src/tv_grab.rs
+51 -7
View File
@@ -1139,11 +1139,15 @@ async fn record_pack_search(database: &Db, season_id: i64) -> Result<(), GrabErr
/// Whether failed season-pack grabs still hold this season off the pack
/// lane. §6.2: a failure quiets the pack search on the shared backoff curve
/// (each failed grab is one attempt), it never disables it. Anchored on the
/// latest failed grab's `grabbed_at` — failure time itself is not recorded.
/// latest `failed_at` (#245), the same anchor §5.7's window uses: a torrent
/// can stall for weeks before `ffprobe` condemns it, and measured from the
/// grab the whole ladder would already have elapsed when the failure lands.
/// `grabbed_at` is the fallback for rows written before #239 added the
/// column.
async fn pack_backoff_active(database: &Db, season_id: i64) -> Result<bool, GrabError> {
let row = sqlx::query!(
r#"SELECT count(*) AS "failures!: i64",
max(grabbed_at) AS "last_failed_at?: String"
r#"SELECT count(*) AS "failures!: i64",
max(coalesce(failed_at, grabbed_at)) AS "last_failed_at?: String"
FROM grabs
WHERE target_kind = 'season' AND target_id = ? AND state = 'failed'"#,
season_id
@@ -1496,8 +1500,17 @@ mod tests {
/// release on the blacklist and the season falls back to per-episode —
/// the pack is not tried again and the episodes are not written off.
/// Seed what the import tick leaves behind after a pack fails: one
/// `failed` season grab per (infohash, age) pair.
/// `failed` season grab per (infohash, age) pair, grabbed and failed at
/// the same age, which is the usual case — the two are minutes apart.
async fn failed_packs(database: &Db, season_id: i64, ages: &[&str]) {
let pairs: Vec<(&str, &str)> = ages.iter().map(|age| (*age, *age)).collect();
stalled_failed_packs(database, season_id, &pairs).await;
}
/// The same seed, but with the grab and the failure at different ages —
/// the #245 case, where a torrent stalls for weeks before `ffprobe`
/// condemns it at import.
async fn stalled_failed_packs(database: &Db, season_id: i64, ages: &[(&str, &str)]) {
let release_id: i64 = sqlx::query_scalar(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
VALUES (7, 'oldpack', 'Fallout.S01.2160p.WEB-DL.OLD', 85899345920,
@@ -1507,16 +1520,19 @@ mod tests {
.fetch_one(database.pool())
.await
.unwrap();
for (index, age) in ages.iter().enumerate() {
for (index, (grabbed_age, failed_age)) in ages.iter().enumerate() {
sqlx::query(
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state, grabbed_at)
"INSERT INTO grabs (release_id, target_kind, target_id, infohash, state,
grabbed_at, failed_at)
VALUES (?, 'season', ?, ?, 'failed',
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?),
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?))",
)
.bind(release_id)
.bind(season_id)
.bind(format!("dead{index}"))
.bind(age)
.bind(grabbed_age)
.bind(failed_age)
.execute(database.pool())
.await
.unwrap();
@@ -1575,6 +1591,34 @@ mod tests {
);
}
/// §6.2, issue #245: the ladder runs from the failure, not the grab. A
/// pack sent five weeks ago and condemned by `ffprobe` ten minutes ago
/// is one minute into a 1h window, not five weeks past it — the lane
/// stays quiet and the episodes carry the season instead.
#[tokio::test]
async fn a_pack_that_stalled_for_weeks_backs_off_from_the_failure() {
let (_dir, database, season_id) =
wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await;
stalled_failed_packs(&database, season_id, &[("-35 days", "-10 minutes")]).await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader).tick(&database).await.unwrap();
let sources: Vec<String> = fake
.torrents()
.into_iter()
.map(|torrent| torrent.source)
.collect();
assert!(
sources
.iter()
.all(|source| !source.ends_with("pack.torrent")),
"the grab is five weeks old but the failure is ten minutes old: {sources:?}"
);
assert_eq!(sources.len(), 3, "{sources:?}");
}
/// Repeated failures ride the capped curve: five failed packs mean a 7d
/// window — still closed at 6d, open at 8d. Quiet, never off.
#[tokio::test]