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

feat(daemon): drain all manual command channels

Browse Source
This commit is contained in:
Miguel Palhas
2026-08-23 18:24:50 +01:00
parent 39e6bce2a0
commit 8a77049fdb
2 changed files with 151 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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 {