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

feat: queue only targets still waiting for a file

Browse Source 
The needs-a-decision queue had no liveness condition on the season lane
and none at all in the API reader, so a season pack that hard-failed
twice, fell back to per-episode grabbing exactly as §6.2 intends, and was
then fully acquired kept notifying for 30 days, and
`GET /api/queues/attention` listed titles the daemon never notified on.

DESIGN.md §5.7 now states the third face of the same rule alongside the
count and the window: a movie or an episode is queued while `wanted` and
not `available`; a season, holding no intent of its own (§4.1), while at
least one of its episodes is. Both readers apply it on all three lanes.

`just ci` passed through the gate.
This commit is contained in:
Miguel Palhas
2026-08-25 10:49:02 +01:00
parent 50056a2bd9
commit 6847d25cf5
8 changed files with 365 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-api/src/movies.rs
+241 -54
View File
@@ -570,11 +570,7 @@ async fn remove_library_files(state: &AppState, id: i64) -> Result<(), ApiError>
Ok(metadata) => metadata,
// Already gone is the state we wanted.
Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
Err(error) => {
return Err(ApiError::Filesystem(format!(
"files not removed: {error}"
)))
}
Err(error) => return Err(ApiError::Filesystem(format!("files not removed: {error}"))),
};
let removed = if metadata.is_dir() {
tokio::fs::remove_dir_all(&target).await
@@ -584,11 +580,7 @@ async fn remove_library_files(state: &AppState, id: i64) -> Result<(), ApiError>
match removed {
Ok(()) => tracing::info!(target = %target.display(), "removed library files"),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(ApiError::Filesystem(format!(
"files not removed: {error}"
)))
}
Err(error) => return Err(ApiError::Filesystem(format!("files not removed: {error}"))),
}
}
Ok(())
@@ -778,6 +770,12 @@ pub async fn grab(
Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}
// The four attention lanes (§9.5). The hard-fail lanes carry §5.7's bar in
// full — two failures on *different* releases, both inside
// `ATTENTION_WINDOW`, against a target still waiting for a file — so this
// endpoint and the daemon's notifier report the same queue. Deliberately not
// a doc comment: utoipa would fold it into the OpenAPI description and drift
// the committed document.
#[utoipa::path(
get, path = "/api/queues/attention", tag = "movies",
responses(
@@ -790,7 +788,7 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
let no_pt_source = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, poster_path, vote_average, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE id IN (SELECT m.id FROM movies m JOIN roots root ON root.id = m.root_id WHERE root.audience = 'kids' AND m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing' AND m.search_attempts > 0 AND NOT EXISTS (SELECT 1 FROM movie_releases mr JOIN releases r ON r.id = mr.release_id WHERE mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived'))) ORDER BY title"#)
.fetch_all(pool(&state)?)
.await?;
let needs_decision = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, poster_path, vote_average, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE (SELECT count(DISTINCT g.release_id) FROM grabs g WHERE g.target_kind = 'movie' AND g.target_id = movies.id AND g.state = 'failed' AND g.grabbed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)) >= 2 ORDER BY title"#, arr_db::ATTENTION_WINDOW)
let needs_decision = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, poster_path, vote_average, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE movies.wanted = 1 AND movies.state != 'available' 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' AND g.grabbed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)) >= 2 ORDER BY title"#, arr_db::ATTENTION_WINDOW)
.fetch_all(pool(&state)?)
.await?;
let (tv_no_pt_source, tv_needs_decision) = tv_attention(&state).await?;
@@ -808,37 +806,73 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
/// share a lane; a series arriving through both is merged into one entry.
///
/// Both hard-fail branches hold to §5.7's bar: two failures on *different*
/// releases, both inside `ATTENTION_WINDOW`. One bad torrent is not a
/// decision, and a failure the operator already dealt with ages out instead
/// of sitting in the queue forever (#226).
/// releases, both inside `ATTENTION_WINDOW`, against a target still waiting
/// for a file. One bad torrent is not a decision (#226), a failure the
/// operator already dealt with ages out instead of sitting in the queue
/// forever (#226), and a target since acquired leaves at once (#238). Seasons
/// hold no intent of their own (§4.1), so the season branch reads liveness
/// off its episodes: it is queued while any of them is still wanted and still
/// without a file. The daemon's notifier filters identically.
async fn tv_attention(
state: &AppState,
) -> Result<(Vec<SeriesAttention>, Vec<SeriesAttention>), ApiError> {
let database = pool(state)?;
Ok((
tv_no_pt_source_lane(database).await?,
tv_hard_fail_lane(database).await?,
))
}
let no_pt_rows = sqlx::query!(
/// §5.2's no-PT-source lane: wanted, searched episodes on a `kids` root whose
/// every candidate release was rejected for language.
async fn tv_no_pt_source_lane(
database: &sqlx::SqlitePool,
) -> Result<Vec<SeriesAttention>, ApiError> {
let rows = sqlx::query!(
r#"
SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
s.title AS "title!: String", s.year,
e.id AS "episode_id!: i64",
se.number AS "season_number!: i64", e.number AS "episode_number!: i64"
FROM episodes e
JOIN seasons se ON se.id = e.season_id
JOIN series s ON s.id = se.series_id
JOIN roots root ON root.id = s.root_id
WHERE root.audience = 'kids'
AND s.blocked = 0
AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0
AND NOT EXISTS (
SELECT 1 FROM episode_releases er
JOIN releases r ON r.id = er.release_id
WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')
)
ORDER BY se.number, e.number
"#
SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
s.title AS "title!: String", s.year,
e.id AS "episode_id!: i64",
se.number AS "season_number!: i64", e.number AS "episode_number!: i64"
FROM episodes e
JOIN seasons se ON se.id = e.season_id
JOIN series s ON s.id = se.series_id
JOIN roots root ON root.id = s.root_id
WHERE root.audience = 'kids'
AND s.blocked = 0
AND e.wanted = 1 AND e.state = 'missing' AND e.search_attempts > 0
AND NOT EXISTS (
SELECT 1 FROM episode_releases er
JOIN releases r ON r.id = er.release_id
WHERE er.episode_id = e.id AND r.verdict IN ('eligible', 'waived')
)
ORDER BY se.number, e.number
"#
)
.fetch_all(database)
.await?;
let mut entries: Vec<SeriesAttention> = Vec::new();
for row in rows {
merge_episode(
&mut entries,
row.series_id,
row.tmdb_id,
&row.title,
row.year,
row.episode_id,
row.season_number,
row.episode_number,
);
}
Ok(entries)
}
/// §5.7's hard-fail lane: episodes and seasons two *different* releases failed
/// on inside `ATTENTION_WINDOW`, restricted to targets still waiting for a
/// file. The season half reads that last condition off its episodes, which is
/// where intent lives (§4.1). Both halves merge into one entry per series.
async fn tv_hard_fail_lane(database: &sqlx::SqlitePool) -> Result<Vec<SeriesAttention>, ApiError> {
let episode_hard_fails = sqlx::query!(
r#"
SELECT s.id AS "series_id!: i64", s.tmdb_id AS "tmdb_id!: i64",
@@ -851,6 +885,7 @@ async fn tv_attention(
JOIN series s ON s.id = se.series_id
WHERE g.state = 'failed'
AND g.grabbed_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)
AND e.wanted = 1 AND e.state != 'available'
GROUP BY s.id, s.tmdb_id, s.title, s.year, e.id, se.number, e.number
HAVING count(DISTINCT g.release_id) >= 2
"#,
@@ -868,6 +903,11 @@ async fn tv_attention(
JOIN series s ON s.id = se.series_id
WHERE g.state = 'failed'
AND g.grabbed_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.tmdb_id, s.title, s.year, se.id, se.number
HAVING count(DISTINCT g.release_id) >= 2
"#,
@@ -876,24 +916,10 @@ async fn tv_attention(
.fetch_all(database)
.await?;
let mut tv_no_pt_source: Vec<SeriesAttention> = Vec::new();
for row in no_pt_rows {
merge_episode(
&mut tv_no_pt_source,
row.series_id,
row.tmdb_id,
&row.title,
row.year,
row.episode_id,
row.season_number,
row.episode_number,
);
}
let mut tv_needs_decision: Vec<SeriesAttention> = Vec::new();
let mut entries: Vec<SeriesAttention> = Vec::new();
for row in episode_hard_fails {
merge_episode(
&mut tv_needs_decision,
&mut entries,
row.series_id,
row.tmdb_id,
&row.title,
@@ -905,7 +931,7 @@ async fn tv_attention(
}
for row in season_pack_fails {
merge_season(
&mut tv_needs_decision,
&mut entries,
row.series_id,
row.tmdb_id,
&row.title,
@@ -914,8 +940,7 @@ async fn tv_attention(
row.season_number,
);
}
Ok((tv_no_pt_source, tv_needs_decision))
Ok(entries)
}
/// One more qualifying season for its series, creating the series' entry on
@@ -2014,8 +2039,9 @@ mod tests {
);
}
/// A series with one empty season, for exercising the season lane on its
/// own.
/// A series with one season holding a single wanted, missing episode — the
/// least that satisfies §5.7's liveness condition — for exercising the
/// season lane on its own.
async fn seed_bare_season(pool: &sqlx::SqlitePool, tmdb_id: i64) -> (i64, i64) {
let root_id: i64 =
sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'tv' AND audience = 'kids'")
@@ -2039,6 +2065,14 @@ mod tests {
.fetch_one(pool)
.await
.expect("season");
sqlx::query(
"INSERT INTO episodes (season_id, number, title, wanted, state)
VALUES (?, 1, 'Episode 1', 1, 'missing')",
)
.bind(season_id)
.execute(pool)
.await
.expect("episode");
(series_id, season_id)
}
@@ -2170,6 +2204,159 @@ mod tests {
);
}
/// §5.7: the queue only holds targets still waiting for a file. A season
/// whose pack failed twice, fell back to per-episode grabbing (§6.2) and
/// was then fully acquired is the system working, so it drops out at once
/// instead of sitting there for 30 days (#238).
#[tokio::test]
async fn a_fully_acquired_season_leaves_the_attention_queue() {
let (_dir, state, base) = application().await;
let pool = state.database().expect("database").pool();
let (series_id, season_id) = seed_bare_season(pool, 1).await;
sqlx::query(
"INSERT INTO episodes (season_id, number, title, wanted, state)
VALUES (?, 2, 'Episode 2', 1, 'missing')",
)
.bind(season_id)
.execute(pool)
.await
.expect("second episode");
for (guid, hash) in [("pack-one", "hash-one"), ("pack-two", "hash-two")] {
let release_id = insert_release(pool, guid).await;
insert_failed_grab(pool, release_id, "season", season_id, hash, 0).await;
}
assert_eq!(
queued_seasons(&base, series_id).await,
vec![season_id],
"two packs failed and episodes are still missing"
);
// Per-episode grabbing got one of the two. Still a gap, still queued.
sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ? AND number = 1")
.bind(season_id)
.execute(pool)
.await
.expect("first episode imported");
assert_eq!(
queued_seasons(&base, series_id).await,
vec![season_id],
"one episode still wanted and missing: still broken, still queued"
);
sqlx::query("UPDATE episodes SET state = 'available' WHERE season_id = ?")
.bind(season_id)
.execute(pool)
.await
.expect("season imported");
assert!(
queued_seasons(&base, series_id).await.is_empty(),
"every episode acquired: the fallback worked, no decision to make"
);
}
/// §5.7: the same liveness condition on the movie and episode lanes, so
/// `GET /api/queues/attention` reports exactly what the daemon notifies
/// on (#238).
#[tokio::test]
async fn an_acquired_movie_or_episode_leaves_the_attention_queue() {
let (_dir, state, base) = application().await;
let pool = state.database().expect("database").pool();
let root_id: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE audience = 'kids'")
.fetch_one(pool)
.await
.expect("kids root");
let movie = add_movie(&base, 82728, root_id).await;
let movie_id = movie["id"].as_i64().expect("id");
for (guid, hash) in [("movie-one", "hash-m1"), ("movie-two", "hash-m2")] {
let release_id = insert_release(pool, guid).await;
insert_failed_grab(pool, release_id, "movie", movie_id, hash, 0).await;
}
let (series_id, season_id) = seed_bare_season(pool, 99).await;
let episode_id: i64 =
sqlx::query_scalar("SELECT id FROM episodes WHERE season_id = ? AND number = 1")
.bind(season_id)
.fetch_one(pool)
.await
.expect("episode id");
for (guid, hash) in [("ep-one", "hash-e1"), ("ep-two", "hash-e2")] {
let release_id = insert_release(pool, guid).await;
insert_failed_grab(pool, release_id, "episode", episode_id, hash, 0).await;
}
let queues = attention_queues(&base).await;
assert_eq!(queues["needs_decision"][0]["id"], movie_id);
assert_eq!(
queued_episodes(&queues, series_id),
vec![episode_id],
"still wanted and missing: queued"
);
sqlx::query("UPDATE movies SET state = 'available' WHERE id = ?")
.bind(movie_id)
.execute(pool)
.await
.expect("movie imported");
sqlx::query("UPDATE episodes SET state = 'available' WHERE id = ?")
.bind(episode_id)
.execute(pool)
.await
.expect("episode imported");
let queues = attention_queues(&base).await;
assert_eq!(
queues["needs_decision"].as_array().map(Vec::len),
Some(0),
"imported from a third release: no decision to make"
);
assert!(
queued_episodes(&queues, series_id).is_empty(),
"imported from a third release: no decision to make"
);
// Withdrawing intent empties the lane just the same.
sqlx::query("UPDATE movies SET state = 'missing', wanted = 0 WHERE id = ?")
.bind(movie_id)
.execute(pool)
.await
.expect("movie unwanted");
assert_eq!(
attention_queues(&base).await["needs_decision"]
.as_array()
.map(Vec::len),
Some(0),
"nothing is waiting for a file"
);
}
/// The whole attention payload.
async fn attention_queues(base: &str) -> serde_json::Value {
reqwest::get(format!("{base}/api/queues/attention"))
.await
.expect("queues")
.json()
.await
.expect("queues json")
}
/// The episodes a series contributes to the hard-fail TV lane.
fn queued_episodes(queues: &serde_json::Value, series_id: i64) -> Vec<i64> {
queues["tv_needs_decision"]
.as_array()
.expect("tv lane")
.iter()
.filter(|entry| entry["series_id"] == series_id)
.flat_map(|entry| {
entry["episodes"]
.as_array()
.expect("episodes")
.iter()
.map(|episode| episode["id"].as_i64().expect("episode id"))
})
.collect()
}
/// One series hitting all three §9.5 TV entry conditions: two wanted,
/// searched episodes whose every candidate was rejected for language; a
/// season two different packs hard-failed on; and an episode two different
crates/arr-daemon/src/attention.rs
+98 -2
View File
@@ -3,8 +3,13 @@
//! 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`. One bad torrent is not a
//! decision, and a failure already dealt with ages out (#226).
//! 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
@@ -218,6 +223,11 @@ fn tv_entry(
/// 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();
@@ -280,6 +290,11 @@ async fn queue_tv(database: &Db) -> Result<Vec<(i64, String, Option<i64>, TvEntr
JOIN series s ON s.id = se.series_id
WHERE g.state = 'failed'
AND g.grabbed_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
"#,
@@ -393,6 +408,21 @@ 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`,
/// stamped `age_days` in the past, so §5.7's window can be exercised
/// without waiting a month.
@@ -687,6 +717,7 @@ mod tests {
.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;
@@ -707,6 +738,70 @@ mod tests {
);
}
/// §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]
@@ -717,6 +812,7 @@ mod tests {
.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;