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

Merge #132: drain manual episode and season commands

Browse Source 
Closes #132
This commit is contained in:
Miguel Palhas
2026-08-23 18:27:43 +01:00
parent 695f1882be 8a77049fdb
commit 127338fd4d
13 changed files with 1182 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-daemon/src/grab.rs
+6 -4
View File
@@ -908,7 +908,7 @@ struct PendingMovie {
/// The eligible view of a stored release, ranked for selection.
#[derive(Debug, Clone)]
pub(crate) struct Eligible {
id: i64,
pub(crate) id: i64,
pub(crate) indexer_id: i64,
pub(crate) guid: String,
pub(crate) name: String,
@@ -1179,7 +1179,9 @@ pub(crate) async fn store_release(
}
/// The TV counterpart: one release row, associated with every episode the
/// claim covers — a season pack matches the whole season.
/// claim covers — a season pack matches the whole season. Returns the
/// release id alongside the eligible view; the id is what links a season
/// pack into `season_releases`, which lists rejected candidates too.
pub(crate) async fn store_episode_release(
database: &Db,
episode_ids: &[i64],
@@ -1188,7 +1190,7 @@ pub(crate) async fn store_episode_release(
overrides: &TitleOverrides,
original_language: &Language,
blacklist: &Blacklist,
) -> Result<Option<Eligible>, GrabError> {
) -> Result<(i64, Option<Eligible>), GrabError> {
let (release_id, eligible) = classify_and_store(
database,
release,
@@ -1208,7 +1210,7 @@ pub(crate) async fn store_episode_release(
.execute(database.pool())
.await?;
}
Ok(eligible)
Ok((release_id, eligible))
}
async fn classify_and_store(
crates/arr-daemon/src/main.rs
+49 -26
View File
@@ -123,7 +123,7 @@ async fn run() -> Result<(), Error> {
None
};
let notifier = Notifier::new(config.ntfy_url.clone())?;
let (reconcile, manual_grab) =
let (reconcile, manual_grab, manual_tv) =
reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
// Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
@@ -157,10 +157,17 @@ async fn run() -> Result<(), Error> {
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
let mut reconcile_task = tokio::spawn(reconcile.run(shutdown_rx.clone()));
// Issue #107: nothing else drains `AppState`'s `MovieCommand` channel, so
// the manual-search and manual-grab endpoints were a no-op — the command
// sat in the 64-slot buffer forever.
let mut manual_task = tokio::spawn(manual::run(state, database, manual_grab, shutdown_rx));
// Issue #107: nothing else drains `AppState`'s command channels, so the
// manual-search and manual-grab endpoints were a no-op — the command sat
// in the 64-slot buffer forever. Issue #132: the episode and season
// channels are drained by the same lane.
let mut manual_task = tokio::spawn(manual::run(
state,
database,
manual_grab,
manual_tv,
shutdown_rx,
));
let signal_tx = shutdown_tx.clone();
let server = async move {
axum::serve(listener, app)
@@ -198,17 +205,18 @@ async fn run() -> Result<(), Error> {
/// Prowlarr key and grab needs TMDB as well; a lane whose upstream is not
/// configured stays unregistered rather than failing every tick.
///
/// Also returns a second, independent `GrabAction` for `manual::run` (issue
/// #107) — the manual trigger needs the same search-and-grab path on demand
/// rather than on the reconcile tick's schedule, and `ReconcileLoop::register`
/// takes ownership of the one it ticks.
/// Also returns independent action instances for `manual::run` (issues #107
/// and #132) — the manual trigger needs the same search-and-grab paths on
/// demand rather than on the reconcile tick's schedule, and
/// `ReconcileLoop::register` takes ownership of the ones it ticks. The movie
/// one needs TMDB too; TV grabbing does not (§6.2).
fn reconcile_loop(
database: &Db,
config: &Config,
transmission: &arr_dl::TransmissionClient,
tmdb: Option<&Arc<TmdbClient>>,
notifier: &Notifier,
) -> Result<(ReconcileLoop, Option<GrabAction>), Error> {
) -> Result<(ReconcileLoop, Option<GrabAction>, Option<TvGrabAction>), Error> {
let reconcile = ReconcileLoop::new(database.clone());
let seeding = SeedingRules::new(
SeedingLimits {
@@ -235,7 +243,7 @@ fn reconcile_loop(
.map(|key| arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key))
.transpose()?;
let (mut reconcile, manual_grab) = register_movie_grab(
let (reconcile, manual_grab) = register_movie_grab(
reconcile,
prowlarr.as_ref(),
transmission,
@@ -243,20 +251,8 @@ fn reconcile_loop(
&seeding,
tmdb,
);
// TV grabbing needs no TMDB at grab time: air dates are already on the
// episode rows, which is the same gate the digital release date is for
// movies (§6.2).
if let Some(prowlarr) = prowlarr.as_ref() {
reconcile = reconcile.register(
Tick::Reconcile,
TvGrabAction::new(
prowlarr.clone(),
transmission.clone(),
config.download_dir.clone(),
seeding.clone(),
),
);
}
let (mut reconcile, manual_tv) =
register_tv_grab(reconcile, prowlarr.as_ref(), transmission, config, &seeding);
// RSS needs no TMDB: it matches what the feeds already carry against the
// wanted list (§6.2).
if let Some(prowlarr) = prowlarr {
@@ -322,7 +318,34 @@ fn reconcile_loop(
}
let reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone()));
Ok((reconcile, manual_grab))
Ok((reconcile, manual_grab, manual_tv))
}
/// Register the TV grab lane on `reconcile` and hand back a second,
/// independent instance for `manual::run` (issue #132). `None` when Prowlarr
/// is not configured. TV grabbing needs no TMDB at grab time: air dates are
/// already on the episode rows, which is the same gate the digital release
/// date is for movies (§6.2).
fn register_tv_grab(
mut reconcile: ReconcileLoop,
prowlarr: Option<&arr_indexer::ProwlarrClient>,
transmission: &arr_dl::TransmissionClient,
config: &Config,
seeding: &SeedingRules,
) -> (ReconcileLoop, Option<TvGrabAction>) {
let Some(prowlarr) = prowlarr else {
return (reconcile, None);
};
let tv_grab_action = || {
TvGrabAction::new(
prowlarr.clone(),
transmission.clone(),
config.download_dir.clone(),
seeding.clone(),
)
};
reconcile = reconcile.register(Tick::Reconcile, tv_grab_action());
(reconcile, Some(tv_grab_action()))
}
/// Register the movie grab lane on `reconcile` and hand back a second,
crates/arr-daemon/src/manual.rs
+102 -20
View File
@@ -1,23 +1,33 @@
//! Drains `AppState`'s `MovieCommand` channel — the daemon-side consumer
//! DESIGN.md §6.2 and §9.3 assume exists (issue #107). `Search` resets the
//! movie's backoff and re-runs the targeted-search + grab lane immediately;
//! Drains `AppState`'s three command channels — the daemon-side consumer
//! DESIGN.md §6.2 and §9.3 assume exists (issues #107 and #132). A `Search`
//! resets backoff and re-runs the targeted-search + grab lane immediately;
//! `Grab` sends an already-chosen release straight to Transmission, skipping
//! search.
//! search. Movies, episodes and seasons share one lane because they share
//! one operator waiting on a 202.
use arr_api::{AppState, MovieCommand};
use arr_api::{AppState, EpisodeCommand, MovieCommand, SeasonCommand};
use arr_db::Db;
use tokio::sync::watch;
use crate::grab::{GrabAction, GrabError};
use crate::tv_grab::TvGrabAction;
/// Run until every sender is dropped or `shutdown` fires. `grab` is `None`
/// when Prowlarr or TMDB is not configured (`main.rs` warns about this
/// already for the reconcile lane) — commands are still drained so the
/// channel never fills, they just cannot be acted on.
/// Which channel a drained command came off.
enum Command {
Movie(MovieCommand),
Episode(EpisodeCommand),
Season(SeasonCommand),
}
/// Run until every sender is dropped or `shutdown` fires. Either action is
/// `None` when Prowlarr (or TMDB, for movies) is not configured (`main.rs`
/// warns about this already for the reconcile lanes) — commands are still
/// drained so the channels never fill, they just cannot be acted on.
pub async fn run(
state: AppState,
database: Db,
grab: Option<GrabAction>,
movies: Option<GrabAction>,
tv: Option<TvGrabAction>,
mut shutdown: watch::Receiver<bool>,
) {
loop {
@@ -29,22 +39,54 @@ pub async fn run(
}
continue;
}
command = state.next_movie_command() => command,
command = state.next_movie_command() => command.map(Command::Movie),
command = state.next_episode_command() => command.map(Command::Episode),
command = state.next_season_command() => command.map(Command::Season),
};
let Some(command) = command else {
return;
};
let Some(grab) = &grab else {
tracing::warn!("manual movie command received but Prowlarr or TMDB is not configured");
continue;
};
if let Err(error) = handle(grab, &database, command).await {
tracing::error!(%error, "manual movie command failed");
match command {
Command::Movie(command) => {
let Some(movies) = &movies else {
tracing::warn!(
"manual movie command received but Prowlarr or TMDB is not configured"
);
continue;
};
if let Err(error) = handle_movie(movies, &database, command).await {
tracing::error!(%error, "manual movie command failed");
}
}
Command::Episode(command) => {
let Some(tv) = &tv else {
tracing::warn!(
"manual episode command received but Prowlarr is not configured"
);
continue;
};
if let Err(error) = handle_episode(tv, &database, command).await {
tracing::error!(%error, "manual episode command failed");
}
}
Command::Season(command) => {
let Some(tv) = &tv else {
tracing::warn!("manual season command received but Prowlarr is not configured");
continue;
};
if let Err(error) = handle_season(tv, &database, command).await {
tracing::error!(%error, "manual season command failed");
}
}
}
}
}
async fn handle(grab: &GrabAction, database: &Db, command: MovieCommand) -> Result<(), GrabError> {
async fn handle_movie(
grab: &GrabAction,
database: &Db,
command: MovieCommand,
) -> Result<(), GrabError> {
match command {
MovieCommand::Search { movie_id } => {
grab.search_now(database, movie_id).await?;
@@ -60,6 +102,46 @@ async fn handle(grab: &GrabAction, database: &Db, command: MovieCommand) -> Resu
Ok(())
}
async fn handle_episode(
tv: &TvGrabAction,
database: &Db,
command: EpisodeCommand,
) -> Result<(), GrabError> {
match command {
EpisodeCommand::Search { episode_id } => {
tv.search_episode_now(database, episode_id).await?;
}
EpisodeCommand::Grab {
episode_id,
release_id,
} => {
tv.grab_episode_release_now(database, episode_id, release_id)
.await?;
}
}
Ok(())
}
async fn handle_season(
tv: &TvGrabAction,
database: &Db,
command: SeasonCommand,
) -> Result<(), GrabError> {
match command {
SeasonCommand::Search { season_id } => {
tv.search_season_now(database, season_id).await?;
}
SeasonCommand::Grab {
season_id,
release_id,
} => {
tv.grab_season_release_now(database, season_id, release_id)
.await?;
}
}
Ok(())
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
@@ -160,7 +242,7 @@ mod tests {
let transmission = transmission().await;
let grab = grab_action(&indexer, &transmission);
handle(&grab, &database, MovieCommand::Search { movie_id: 1 })
handle_movie(&grab, &database, MovieCommand::Search { movie_id: 1 })
.await
.unwrap();
@@ -218,7 +300,7 @@ mod tests {
.await
.unwrap();
handle(
handle_movie(
&grab,
&database,
MovieCommand::Grab {
crates/arr-daemon/src/tv_grab.rs
+719 -1
View File
@@ -415,7 +415,7 @@ impl TvGrabAction {
if covered.is_empty() {
continue;
}
let stored = store_episode_release(
let (stored_id, stored) = store_episode_release(
database,
&covered,
&release,
@@ -425,6 +425,22 @@ impl TvGrabAction {
blacklist,
)
.await?;
// A pack for exactly this season belongs on the season deck too
// (§9.3, issue #125) — including when it is rejected, which the
// deck shows greyed out.
if claim
.as_ref()
.is_some_and(|claim| is_this_seasons_pack(claim, season.season_number_u32()))
{
sqlx::query!(
"INSERT INTO season_releases (season_id, release_id) VALUES (?, ?)
ON CONFLICT DO NOTHING",
season.season_id,
stored_id
)
.execute(database.pool())
.await?;
}
candidates.push(TvCandidate {
claim,
eligible: stored,
@@ -432,6 +448,229 @@ impl TvGrabAction {
}
Ok(candidates)
}
/// The manual episode trigger (§6.2, §9.3, issue #132): reset the
/// episode's backoff so selection cannot skip it, then sweep the
/// indexers scoped to this one episode instead of the tick's
/// `EPISODE_SEARCHES_PER_TICK` batch.
///
/// Same shape as the movie trigger: a gap gets search and grab; an
/// already-satisfied episode gets its deck refreshed and no grab; a
/// blocked series is refused outright (§6.3). An unaired episode is not
/// a gap yet — nothing to close until it airs (§6.2).
pub(crate) async fn search_episode_now(
&self,
database: &Db,
episode_id: i64,
) -> Result<(), GrabError> {
let Some(episode) = manual_episode(database, episode_id).await? else {
tracing::info!(
episode_id,
"manual episode search refused: unknown or blocked episode"
);
return Ok(());
};
let Some(original_language) = episode.season.original_language.as_deref() else {
tracing::warn!(
series = episode.season.series_title,
"no original language yet; not searching"
);
return Ok(());
};
let original_language = arr_db::policy::language(original_language);
let Some(loaded) = database.episode_policy(episode_id).await? else {
return Ok(());
};
reset_episode_backoff(database, episode_id).await?;
let blacklist = Blacklist::load(database.pool()).await?;
let episodes = season_episodes(database, episode.season.season_id).await?;
let selector = TvSelector::Episode {
season: episode.season.season_number_u32(),
episode: u32::try_from(episode.number).unwrap_or_default(),
};
if !episode.is_gap() {
let candidates = self
.search(
database,
&episode.season,
&episodes,
selector,
&loaded,
&original_language,
&blacklist,
)
.await?;
record_episode_search(database, &[episode_id]).await?;
tracing::info!(
episode_id,
eligible = candidates.len(),
"episode release deck refreshed; not grabbing a satisfied episode"
);
return Ok(());
}
if !episode.aired() {
tracing::info!(episode_id, "episode has not aired; nothing to search for");
return Ok(());
}
let Some(this_one) = episodes.iter().find(|row| row.id == episode_id) else {
return Ok(());
};
self.grab_episodes(
database,
&episode.season,
&episodes,
std::slice::from_ref(&this_one),
&loaded,
&original_language,
&blacklist,
usize::MAX,
)
.await?;
Ok(())
}
/// The manual one-click episode grab (§9.3, issue #132): the release is
/// already chosen off the episode deck, so this skips search and scoring
/// and sends it straight to Transmission. A manual grab may take a
/// `waived` release (§9.3), never a `rejected` one.
pub(crate) async fn grab_episode_release_now(
&self,
database: &Db,
episode_id: i64,
release_id: i64,
) -> Result<Option<Outcome>, GrabError> {
let Some(loaded) = database.episode_policy(episode_id).await? else {
return Ok(None);
};
let Some(title) = episode_series_title(database, episode_id).await? else {
return Ok(None);
};
let Some(release) = load_episode_release(database, episode_id, release_id).await? else {
return Ok(None);
};
let blacklist = Blacklist::load(database.pool()).await?;
self.grabber
.send_winner(
database,
&GrabTarget {
scope: GrabScope::Episode { episode_id },
title: &title,
counts_as_attempt: false,
},
&loaded,
&blacklist,
release,
)
.await
}
/// The manual season trigger (issue #125's deck, drained per #132):
/// reset every open episode's backoff and run the same pack-or-fall-back
/// lane as the tick, scoped to this one season with no tick budgets.
///
/// A satisfied season — nothing open, or a grab already in flight —
/// gets its deck refreshed and no grab; a blocked series refuses.
pub(crate) async fn search_season_now(
&self,
database: &Db,
season_id: i64,
) -> Result<(), GrabError> {
let Some(season) = manual_season(database, season_id).await? else {
tracing::info!(
season_id,
"manual season search refused: unknown or blocked season"
);
return Ok(());
};
let Some(original_language) = season.original_language.as_deref() else {
tracing::warn!(
series = season.series_title,
"no original language yet; not searching"
);
return Ok(());
};
let original_language = arr_db::policy::language(original_language);
let Some(loaded) = database.season_policy(season_id).await? else {
return Ok(());
};
reset_season_backoff(database, season_id).await?;
let blacklist = Blacklist::load(database.pool()).await?;
if !season.is_gap {
let episodes = season_episodes(database, season_id).await?;
let open: Vec<i64> = episodes
.iter()
.filter(|episode| episode.wanted && !episode.has_file && !episode.in_flight)
.map(|episode| episode.id)
.collect();
let candidates = self
.search(
database,
&season,
&episodes,
TvSelector::Season {
season: season.season_number_u32(),
},
&loaded,
&original_language,
&blacklist,
)
.await?;
record_episode_search(database, &open).await?;
tracing::info!(
season_id,
eligible = candidates.len(),
"season release deck refreshed; not grabbing a satisfied season"
);
return Ok(());
}
self.grab_season(database, &season, &blacklist, usize::MAX)
.await?;
Ok(())
}
/// The manual one-click season grab (§9.3, issue #125): a chosen pack
/// goes straight to Transmission against the season's still-open gaps.
pub(crate) async fn grab_season_release_now(
&self,
database: &Db,
season_id: i64,
release_id: i64,
) -> Result<Option<Outcome>, GrabError> {
let Some(loaded) = database.season_policy(season_id).await? else {
return Ok(None);
};
let Some(title) = season_series_title(database, season_id).await? else {
return Ok(None);
};
let Some(release) = load_season_release(database, season_id, release_id).await? else {
return Ok(None);
};
let episode_ids: Vec<i64> = season_episodes(database, season_id)
.await?
.iter()
.filter(|episode| episode.wanted && !episode.has_file && !episode.in_flight)
.map(|episode| episode.id)
.collect();
let blacklist = Blacklist::load(database.pool()).await?;
self.grabber
.send_winner(
database,
&GrabTarget {
scope: GrabScope::Season {
season_id,
episode_ids,
},
title: &title,
counts_as_attempt: false,
},
&loaded,
&blacklist,
release,
)
.await
}
}
impl Action for TvGrabAction {
@@ -593,6 +832,280 @@ async fn season_episodes(database: &Db, season_id: i64) -> Result<Vec<SeasonEpis
.collect())
}
/// One episode named by a manual command (#132), with the series context a
/// targeted search needs. Same gap facts as [`season_episodes`] plus the
/// lane decision. Blocked refuses upstream of here.
#[derive(Debug)]
struct ManualEpisode {
number: i64,
air_date: Option<String>,
wanted: bool,
has_file: bool,
in_flight: bool,
season: PendingSeason,
}
impl ManualEpisode {
/// Wanted with neither a file nor a grab in flight — a gap the manual
/// trigger may close. Anything else only gets a deck refresh.
fn is_gap(&self) -> bool {
self.wanted && !self.has_file && !self.in_flight
}
/// Unknown is unaired (§6.2): nothing to close until it airs.
fn aired(&self) -> bool {
air_date_time(self.air_date.as_deref())
.is_some_and(|date| date <= std::time::SystemTime::now())
}
}
/// Same columns as [`pending_seasons`], scoped to one id, without the tick's
/// batch limit and without the gap filter — the manual trigger names the
/// season, so what remains decides the lane here instead of hiding it.
async fn manual_episode(
database: &Db,
episode_id: i64,
) -> Result<Option<ManualEpisode>, GrabError> {
let row = sqlx::query!(
r#"
SELECT e.id AS "id!: i64",
e.number AS "number!: i64",
e.air_date,
e.wanted AS "wanted!: bool",
EXISTS (
SELECT 1 FROM media_files f
WHERE f.owner_kind = 'episode' AND f.owner_id = e.id
) AS "has_file!: bool",
EXISTS (
SELECT 1 FROM grabs g
WHERE g.target_kind = 'episode' AND g.target_id = e.id
AND g.state IN ('sent', 'downloaded', 'imported')
) AS "in_flight!: bool",
se.id AS "season_id!: i64",
se.number AS "season_number!: i64",
s.title AS "series_title!: String",
s.tvdb_id AS series_tvdb_id,
s.original_language
FROM episodes e
JOIN seasons se ON se.id = e.season_id
JOIN series s ON s.id = se.series_id
WHERE e.id = ?
AND s.blocked = 0
"#,
episode_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| ManualEpisode {
number: row.number,
air_date: row.air_date,
wanted: row.wanted,
has_file: row.has_file,
in_flight: row.in_flight,
season: PendingSeason {
season_id: row.season_id,
season_number: row.season_number,
series_title: row.series_title,
series_tvdb_id: row.series_tvdb_id,
original_language: row.original_language,
},
}))
}
/// A season addressed by a manual command, with whether it is still a gap.
async fn manual_season(
database: &Db,
season_id: i64,
) -> Result<Option<PendingSeasonGap>, GrabError> {
let row = sqlx::query!(
r#"
SELECT se.id AS "season_id!: i64",
se.number AS "season_number!: i64",
s.title AS "series_title!: String",
s.tvdb_id AS series_tvdb_id,
s.original_language,
(
EXISTS (
SELECT 1 FROM episodes e
WHERE e.season_id = se.id
AND e.wanted = 1
AND NOT EXISTS (
SELECT 1 FROM media_files f
WHERE f.owner_kind = 'episode' AND f.owner_id = e.id
)
AND NOT EXISTS (
SELECT 1 FROM grabs g
WHERE g.target_kind = 'episode' AND g.target_id = e.id
AND g.state IN ('sent', 'downloaded', 'imported')
)
)
AND NOT EXISTS (
SELECT 1 FROM grabs g
WHERE g.target_kind = 'season' AND g.target_id = se.id
AND g.state IN ('sent', 'downloaded', 'imported')
)
) AS "is_gap!: bool"
FROM seasons se
JOIN series s ON s.id = se.series_id
WHERE se.id = ?
AND s.blocked = 0
"#,
season_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| PendingSeasonGap {
season: PendingSeason {
season_id: row.season_id,
season_number: row.season_number,
series_title: row.series_title,
series_tvdb_id: row.series_tvdb_id,
original_language: row.original_language,
},
is_gap: row.is_gap,
}))
}
/// [`PendingSeason`] plus the lane decision for the manual trigger.
struct PendingSeasonGap {
season: PendingSeason,
is_gap: bool,
}
impl std::ops::Deref for PendingSeasonGap {
type Target = PendingSeason;
fn deref(&self) -> &PendingSeason {
&self.season
}
}
/// Unconditional, unlike the tick's backoff gate: the manual trigger's whole
/// point is to ignore the exponential backoff (§6.2).
async fn reset_episode_backoff(database: &Db, episode_id: i64) -> Result<(), GrabError> {
sqlx::query!(
"UPDATE episodes SET search_attempts = 0, last_searched_at = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE id = ?",
episode_id
)
.execute(database.pool())
.await?;
Ok(())
}
/// A manual season search is one attempt against every open episode at
/// once, so they all start from zero together.
async fn reset_season_backoff(database: &Db, season_id: i64) -> Result<(), GrabError> {
sqlx::query!(
"UPDATE episodes SET search_attempts = 0, last_searched_at = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE season_id = ?",
season_id
)
.execute(database.pool())
.await?;
Ok(())
}
async fn episode_series_title(database: &Db, episode_id: i64) -> Result<Option<String>, GrabError> {
let title = sqlx::query_scalar!(
r#"SELECT s.title AS "title!: String"
FROM episodes e
JOIN seasons se ON se.id = e.season_id
JOIN series s ON s.id = se.series_id
WHERE e.id = ?"#,
episode_id
)
.fetch_optional(database.pool())
.await?;
Ok(title)
}
async fn season_series_title(database: &Db, season_id: i64) -> Result<Option<String>, GrabError> {
let title = sqlx::query_scalar!(
r#"SELECT s.title AS "title!: String"
FROM seasons se
JOIN series s ON s.id = se.series_id
WHERE se.id = ?"#,
season_id
)
.fetch_optional(database.pool())
.await?;
Ok(title)
}
/// The chosen release, still associated with the episode and not
/// hard-failed — a manual grab may take a `waived` release (§9.3).
async fn load_episode_release(
database: &Db,
episode_id: i64,
release_id: i64,
) -> Result<Option<Eligible>, GrabError> {
let row = sqlx::query!(
r#"
SELECT r.id AS "id!: i64",
r.indexer_id AS "indexer_id!: i64",
r.guid AS "guid!: String",
r.name AS "name!: String",
r.download_url AS "download_url!: String",
CAST(COALESCE(r.score, 0) AS INTEGER) AS "score!: i64"
FROM releases r
JOIN episode_releases er ON er.release_id = r.id
WHERE er.episode_id = ? AND r.id = ? AND r.verdict IN ('eligible', 'waived')
"#,
episode_id,
release_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| Eligible {
id: row.id,
indexer_id: row.indexer_id,
guid: row.guid,
name: row.name,
download_url: row.download_url,
score: row.score,
}))
}
/// The season counterpart: chosen off the season deck (issue #125).
async fn load_season_release(
database: &Db,
season_id: i64,
release_id: i64,
) -> Result<Option<Eligible>, GrabError> {
let row = sqlx::query!(
r#"
SELECT r.id AS "id!: i64",
r.indexer_id AS "indexer_id!: i64",
r.guid AS "guid!: String",
r.name AS "name!: String",
r.download_url AS "download_url!: String",
CAST(COALESCE(r.score, 0) AS INTEGER) AS "score!: i64"
FROM releases r
JOIN season_releases sr ON sr.release_id = r.id
WHERE sr.season_id = ? AND r.id = ? AND r.verdict IN ('eligible', 'waived')
"#,
season_id,
release_id
)
.fetch_optional(database.pool())
.await?;
Ok(row.map(|row| Eligible {
id: row.id,
indexer_id: row.indexer_id,
guid: row.guid,
name: row.name,
download_url: row.download_url,
score: row.score,
}))
}
/// Whether a season-pack grab for this season already hard-failed — the
/// fall-back-to-per-episode signal.
async fn pack_hard_failed(database: &Db, season_id: i64) -> Result<bool, GrabError> {
@@ -1099,4 +1612,209 @@ mod tests {
.count();
assert_eq!(searches_after_first, searches_after_second);
}
/// Issue #132: an `EpisodeCommand::Search` must do the work, not merely
/// be accepted onto the channel. `search_attempts` starts at 5 (a 7-day
/// backoff, nowhere near elapsed) so a tick would skip this episode;
/// ending at 1 rather than 6 proves the reset happened, and a torrent in
/// Transmission proves the grab did.
#[tokio::test]
async fn a_manual_episode_search_searches_and_grabs_now() {
let (_dir, database, _season_id) =
wanted_season(&["2024-04-11", "2024-04-18", "2024-04-25"]).await;
sqlx::query(
"UPDATE episodes SET search_attempts = 5,
last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-6 days')
WHERE number = 1",
)
.execute(database.pool())
.await
.unwrap();
let episode_id: i64 = sqlx::query_scalar("SELECT id FROM episodes WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader)
.search_episode_now(&database, episode_id)
.await
.unwrap();
let sources: Vec<String> = fake
.torrents()
.into_iter()
.map(|torrent| torrent.source)
.collect();
assert_eq!(sources.len(), 1, "{sources:?}");
assert!(sources[0].ends_with("e01.torrent"));
assert_eq!(
grabs(&database).await,
vec![("episode".to_owned(), episode_id, "sent".to_owned())]
);
let attempts: i64 = sqlx::query_scalar("SELECT search_attempts FROM episodes WHERE id = ?")
.bind(episode_id)
.fetch_one(database.pool())
.await
.unwrap();
assert_eq!(
attempts, 0,
"reset from 5; a completed grab records no further attempt"
);
}
/// Issue #132: an `EpisodeCommand::Grab` sends the already-chosen release
/// straight to Transmission — no indexer search at all.
#[tokio::test]
async fn a_manual_episode_grab_sends_the_chosen_release_without_searching() {
let (_dir, database, _season_id) = wanted_season(&["2024-04-11", "2024-04-18"]).await;
let episode_id: i64 = sqlx::query_scalar("SELECT id FROM episodes WHERE number = 1")
.fetch_one(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let release_id: i64 = sqlx::query_scalar(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict)
VALUES (7, 'chosen', 'Fallout.S01E01.2160p.WEB-DL.DDP5.1', 10737418240,
?, '{}', 900, 'eligible') RETURNING id",
)
.bind(format!("{}/dl/e01.torrent", indexer.uri()))
.fetch_one(database.pool())
.await
.unwrap();
sqlx::query("INSERT INTO episode_releases (episode_id, release_id) VALUES (?, ?)")
.bind(episode_id)
.bind(release_id)
.execute(database.pool())
.await
.unwrap();
action(&indexer, &downloader)
.grab_episode_release_now(&database, episode_id, release_id)
.await
.unwrap();
assert_eq!(fake.torrents().len(), 1);
assert_eq!(
grabs(&database).await,
vec![("episode".to_owned(), episode_id, "sent".to_owned())]
);
let searched = indexer
.received_requests()
.await
.unwrap()
.iter()
.any(|request| {
request
.url
.query_pairs()
.any(|(name, value)| name == "t" && value == "search")
});
assert!(!searched, "a chosen release must not trigger a search");
}
/// Issue #132: a `SeasonCommand::Search` runs the pack lane now, against
/// the season deck's own table — the backoff reset is what proves it was
/// really drained rather than dropped on the floor.
#[tokio::test]
async fn a_manual_season_search_takes_the_pack_now() {
let (_dir, database, season_id) =
wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await;
sqlx::query(
"UPDATE episodes SET search_attempts = 5,
last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-6 days')",
)
.execute(database.pool())
.await
.unwrap();
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
action(&indexer, &downloader)
.search_season_now(&database, season_id)
.await
.unwrap();
assert_eq!(fake.torrents().len(), 1);
assert!(fake.torrents()[0].source.ends_with("pack.torrent"));
assert_eq!(
grabs(&database).await,
vec![("season".to_owned(), season_id, "sent".to_owned())]
);
assert_eq!(
episode_states(&database).await,
vec!["downloading", "downloading", "downloading"]
);
let packs: Vec<String> = sqlx::query_scalar(
"SELECT r.name FROM season_releases sr JOIN releases r ON r.id = sr.release_id",
)
.fetch_all(database.pool())
.await
.unwrap();
assert!(packs.contains(&"Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos".to_owned()));
let attempts: Vec<i64> =
sqlx::query_scalar("SELECT search_attempts FROM episodes ORDER BY number")
.fetch_all(database.pool())
.await
.unwrap();
assert_eq!(
attempts,
vec![0, 0, 0],
"reset from 5; a completed grab records no further attempt"
);
}
/// Issue #132: a `SeasonCommand::Grab` sends the already-chosen pack
/// straight to Transmission — no indexer search at all.
#[tokio::test]
async fn a_manual_season_grab_sends_the_chosen_pack_without_searching() {
let (_dir, database, season_id) =
wanted_season(&["2024-04-11", "2024-04-11", "2024-04-11"]).await;
let indexer = prowlarr().await;
let (downloader, fake) = transmission().await;
let release_id: i64 = sqlx::query_scalar(
"INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict)
VALUES (7, 'chosen-pack', 'Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos', 85899345920,
?, '{}', 900, 'eligible') RETURNING id",
)
.bind(format!("{}/dl/pack.torrent", indexer.uri()))
.fetch_one(database.pool())
.await
.unwrap();
sqlx::query("INSERT INTO season_releases (season_id, release_id) VALUES (?, ?)")
.bind(season_id)
.bind(release_id)
.execute(database.pool())
.await
.unwrap();
action(&indexer, &downloader)
.grab_season_release_now(&database, season_id, release_id)
.await
.unwrap();
assert_eq!(fake.torrents().len(), 1);
assert_eq!(
grabs(&database).await,
vec![("season".to_owned(), season_id, "sent".to_owned())]
);
assert_eq!(
episode_states(&database).await,
vec!["downloading", "downloading", "downloading"]
);
let searched = indexer
.received_requests()
.await
.unwrap()
.iter()
.any(|request| {
request
.url
.query_pairs()
.any(|(name, value)| name == "t" && value == "search")
});
assert!(!searched, "a chosen release must not trigger a search");
}
}