diff --git a/crates/arr-api/src/lib.rs b/crates/arr-api/src/lib.rs index ad787b0..06184f7 100644 --- a/crates/arr-api/src/lib.rs +++ b/crates/arr-api/src/lib.rs @@ -49,8 +49,8 @@ pub use series::{ UpdateSeason, UpdateSeries, }; pub use state::{ - AppState, EpisodeCommand, MetadataCommand, MovieCommand, SeasonCommand, Upstreams, - DEFAULT_TMDB_URL, + AppState, EpisodeCommand, GrabResult, MetadataCommand, MovieCommand, Request, SeasonCommand, + Upstreams, DEFAULT_TMDB_URL, }; pub use subtitle_settings::{SubtitleSettings, SubtitleSettingsInput}; pub use subtitles::{ diff --git a/crates/arr-api/src/movies.rs b/crates/arr-api/src/movies.rs index 6134997..c48eb21 100644 --- a/crates/arr-api/src/movies.rs +++ b/crates/arr-api/src/movies.rs @@ -217,6 +217,10 @@ pub enum ApiError { NoTrailer, Conflict(String), Invalid(String), + /// A manual grab the daemon tried and could not send to qBittorrent. + /// Carries the reason, since the release usually needs replacing rather + /// than retrying. + GrabFailed(String), /// The SQLite handle is gone. Distinct from [`Self::Upstream`] so a /// TMDB or Prowlarr fault never reads as a database fault (issue #168). Unavailable, @@ -258,6 +262,7 @@ impl IntoResponse for ApiError { ), Self::NoTrailer => (StatusCode::NOT_FOUND, "no trailer".to_string(), None), Self::Conflict(error) => (StatusCode::CONFLICT, error, None), + Self::GrabFailed(error) => (StatusCode::BAD_GATEWAY, error, None), Self::Invalid(error) => (StatusCode::UNPROCESSABLE_ENTITY, error, None), Self::Unavailable => ( StatusCode::SERVICE_UNAVAILABLE, @@ -838,6 +843,7 @@ pub async fn files( (status = 202, body = Accepted), (status = 404, body = ErrorBody), (status = 500, body = ErrorBody), + (status = 502, body = ErrorBody), (status = 503, body = ErrorBody) ) )] @@ -851,13 +857,34 @@ pub async fn grab( if !exists { return Err(ApiError::NotFound); } - state - .send_movie_command(MovieCommand::Grab { + let outcome = state + .request_movie_grab(MovieCommand::Grab { movie_id, release_id, }) .map_err(|_| ApiError::Upstream("daemon"))?; - Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) + await_grab(outcome).await +} + +/// How long a grab request waits for the daemon before answering 202 and +/// leaving the outcome to the deck. Resolving the link and adding the torrent +/// take seconds; a queue of manual searches ahead of it can take longer. +const GRAB_WAIT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Answer a manual grab with what the daemon did. A failure used to be +/// logged and nothing else, so the button read "sent" for a grab that never +/// reached qBittorrent. +pub(crate) async fn await_grab( + outcome: tokio::sync::oneshot::Receiver, +) -> Result<(StatusCode, Json), ApiError> { + match tokio::time::timeout(GRAB_WAIT, outcome).await { + Ok(Ok(Err(reason))) => Err(ApiError::GrabFailed(reason)), + // Sent, timed out, or the reply was dropped unanswered: still queued + // or done as far as this request can tell. + Ok(Ok(Ok(())) | Err(_)) | Err(_) => { + Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) + } + } } // The four attention lanes (§9.5). The hard-fail lanes carry §5.7's bar in @@ -2127,14 +2154,14 @@ mod tests { .expect("release json"); assert_eq!(releases[0]["verdict"], "eligible"); - let response = reqwest::Client::new() - .post(format!( - "{base}/api/movies/{movie_id}/releases/{release_id}/grab" - )) - .send() - .await - .expect("grab"); - assert_eq!(response.status(), StatusCode::ACCEPTED); + // The handler waits on the daemon's answer, so drain concurrently. + let grab = tokio::spawn( + reqwest::Client::new() + .post(format!( + "{base}/api/movies/{movie_id}/releases/{release_id}/grab" + )) + .send(), + ); assert_eq!( state.next_movie_command().await.expect("command"), MovieCommand::Grab { @@ -2142,6 +2169,43 @@ mod tests { release_id } ); + let response = grab.await.expect("grab task").expect("grab"); + assert_eq!(response.status(), StatusCode::ACCEPTED); + } + + /// A grab the daemon could not send must fail the request with the + /// daemon's reason, not answer 202 for a torrent qBittorrent never got. + #[tokio::test] + async fn a_grab_the_daemon_could_not_send_fails_with_its_reason() { + let (_dir, state, base) = application().await; + let movie = add_movie(&base, 693_134, 2).await; + let movie_id = movie["id"].as_i64().expect("id"); + let pool = state.database().expect("database").pool(); + let release_id = sqlx::query("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict) VALUES (7, 'guid', 'release', 1000, 'url', '{}', 42, 'waived')") + .execute(pool).await.expect("release").last_insert_rowid(); + sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)") + .bind(movie_id) + .bind(release_id) + .execute(pool) + .await + .expect("association"); + + let grab = tokio::spawn( + reqwest::Client::new() + .post(format!( + "{base}/api/movies/{movie_id}/releases/{release_id}/grab" + )) + .send(), + ); + let (_, reply) = state.next_movie_request().await.expect("command"); + reply + .expect("a grab carries a reply") + .send(Err("download selectors didn't match".to_owned())) + .expect("handler waiting"); + let response = grab.await.expect("grab task").expect("grab"); + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + let body: serde_json::Value = response.json().await.expect("error body"); + assert_eq!(body["error"], "download selectors didn't match"); } /// #114: a deck must score against the current policy, not whatever the diff --git a/crates/arr-api/src/series.rs b/crates/arr-api/src/series.rs index 17d4d4a..ad1330d 100644 --- a/crates/arr-api/src/series.rs +++ b/crates/arr-api/src/series.rs @@ -1520,6 +1520,7 @@ pub async fn episode_releases( (status = 202, body = Accepted), (status = 404, body = ErrorBody), (status = 500, body = ErrorBody), + (status = 502, body = ErrorBody), (status = 503, body = ErrorBody) ) )] @@ -1533,13 +1534,13 @@ pub async fn grab_episode( if !exists { return Err(ApiError::EpisodeNotFound); } - state - .send_episode_command(EpisodeCommand::Grab { + let outcome = state + .request_episode_grab(EpisodeCommand::Grab { episode_id, release_id, }) .map_err(|_| ApiError::Unavailable)?; - Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) + crate::movies::await_grab(outcome).await } /// Resolves a season by its number within one series, so the deck is @@ -1920,6 +1921,7 @@ pub async fn season_pack_state( (status = 202, body = Accepted), (status = 404, body = ErrorBody), (status = 500, body = ErrorBody), + (status = 502, body = ErrorBody), (status = 503, body = ErrorBody) ) )] @@ -1935,13 +1937,13 @@ pub async fn grab_season_release( if !exists { return Err(ApiError::SeasonNotFound); } - state - .send_season_command(SeasonCommand::Grab { + let outcome = state + .request_season_grab(SeasonCommand::Grab { season_id, release_id, }) .map_err(|_| ApiError::Unavailable)?; - Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true }))) + crate::movies::await_grab(outcome).await } /// One imported episode file, keyed to its episode so the detail view can @@ -3322,14 +3324,14 @@ mod tests { .expect("releases json"); assert_eq!(releases[0]["verdict"], "eligible"); - let response = reqwest::Client::new() - .post(format!( - "{base}/api/episodes/{episode_id}/releases/{release_id}/grab" - )) - .send() - .await - .expect("grab"); - assert_eq!(response.status(), StatusCode::ACCEPTED); + // The handler waits on the daemon's answer, so drain concurrently. + let grab = tokio::spawn( + reqwest::Client::new() + .post(format!( + "{base}/api/episodes/{episode_id}/releases/{release_id}/grab" + )) + .send(), + ); assert_eq!( state.next_episode_command().await.expect("command"), EpisodeCommand::Grab { @@ -3337,6 +3339,8 @@ mod tests { release_id } ); + let response = grab.await.expect("grab task").expect("grab"); + assert_eq!(response.status(), StatusCode::ACCEPTED); // A release belonging to another episode is not grabbable through // this one. @@ -3433,14 +3437,14 @@ mod tests { "the deck reflects the current policy, not the stored score" ); - let response = reqwest::Client::new() - .post(format!( - "{base}/api/series/{series_id}/seasons/1/releases/{release_id}/grab" - )) - .send() - .await - .expect("grab"); - assert_eq!(response.status(), StatusCode::ACCEPTED); + // The handler waits on the daemon's answer, so drain concurrently. + let grab = tokio::spawn( + reqwest::Client::new() + .post(format!( + "{base}/api/series/{series_id}/seasons/1/releases/{release_id}/grab" + )) + .send(), + ); assert_eq!( state.next_season_command().await.expect("command"), SeasonCommand::Grab { @@ -3448,6 +3452,8 @@ mod tests { release_id } ); + let response = grab.await.expect("grab task").expect("grab"); + assert_eq!(response.status(), StatusCode::ACCEPTED); // A release belonging to another season is not grabbable through // this one, and an unknown season number is not another season. diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index 07f55b7..431d98e 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -9,8 +9,8 @@ use arr_db::Db; use arr_dl::QbitClient; use arr_probe::Extractor; use arr_subs::{Backend, OpenAiEndpoint, Provider, Syncer}; -use tokio::sync::mpsc; use tokio::sync::RwLock; +use tokio::sync::{mpsc, oneshot}; use crate::jellyfin::JellyfinClient; @@ -81,12 +81,12 @@ pub struct AppState { http: reqwest::Client, upstreams: Arc, database: Option, - movie_commands: mpsc::Sender, - pending_movie_commands: Arc>>, - episode_commands: mpsc::Sender, - pending_episode_commands: Arc>>, - season_commands: mpsc::Sender, - pending_season_commands: Arc>>, + movie_commands: mpsc::Sender>, + pending_movie_commands: Arc>>>, + episode_commands: mpsc::Sender>, + pending_episode_commands: Arc>>>, + season_commands: mpsc::Sender>, + pending_season_commands: Arc>>>, metadata_commands: mpsc::Sender, pending_metadata_commands: Arc>>, subtitle_providers: Arc>>, @@ -113,6 +113,18 @@ pub struct AppState { download_snapshot: DownloadSnapshot, } +/// What the daemon reports back for a manual grab: `Err` carries the reason +/// the release never reached qBittorrent. +pub type GrabResult = Result<(), String>; + +/// A manual command, plus where to report a grab's outcome. Searches carry +/// no reply; their results land on the deck. +pub type Request = (C, Option>); + +/// The command channel was full or nothing is draining it. +#[derive(Debug)] +pub(crate) struct CommandRefused; + /// Work explicitly requested through the movie API. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum MovieCommand { @@ -325,6 +337,12 @@ impl AppState { /// /// If every sender has been dropped. pub async fn next_movie_command(&self) -> Option { + self.next_movie_request().await.map(|(command, _)| command) + } + + /// [`Self::next_movie_command`], with the channel a grab's outcome goes + /// back on. + pub async fn next_movie_request(&self) -> Option> { self.pending_movie_commands.lock().await.recv().await } @@ -334,6 +352,14 @@ impl AppState { /// /// If every sender has been dropped. pub async fn next_episode_command(&self) -> Option { + self.next_episode_request() + .await + .map(|(command, _)| command) + } + + /// [`Self::next_episode_command`], with the channel a grab's outcome goes + /// back on. + pub async fn next_episode_request(&self) -> Option> { self.pending_episode_commands.lock().await.recv().await } @@ -343,6 +369,12 @@ impl AppState { /// /// If every sender has been dropped. pub async fn next_season_command(&self) -> Option { + self.next_season_request().await.map(|(command, _)| command) + } + + /// [`Self::next_season_command`], with the channel a grab's outcome goes + /// back on. + pub async fn next_season_request(&self) -> Option> { self.pending_season_commands.lock().await.recv().await } @@ -376,25 +408,61 @@ impl AppState { &self.download_snapshot } - pub(crate) fn send_movie_command( + pub(crate) fn send_movie_command(&self, command: MovieCommand) -> Result<(), CommandRefused> { + self.movie_commands + .try_send((command, None)) + .map_err(|_| CommandRefused) + } + + /// Queue a grab and hand back where its outcome will arrive. + pub(crate) fn request_movie_grab( &self, command: MovieCommand, - ) -> Result<(), mpsc::error::TrySendError> { - self.movie_commands.try_send(command) + ) -> Result, CommandRefused> { + let (reply, outcome) = oneshot::channel(); + self.movie_commands + .try_send((command, Some(reply))) + .map_err(|_| CommandRefused)?; + Ok(outcome) } pub(crate) fn send_episode_command( &self, command: EpisodeCommand, - ) -> Result<(), mpsc::error::TrySendError> { - self.episode_commands.try_send(command) + ) -> Result<(), CommandRefused> { + self.episode_commands + .try_send((command, None)) + .map_err(|_| CommandRefused) } - pub(crate) fn send_season_command( + /// Queue a grab and hand back where its outcome will arrive. + pub(crate) fn request_episode_grab( + &self, + command: EpisodeCommand, + ) -> Result, CommandRefused> { + let (reply, outcome) = oneshot::channel(); + self.episode_commands + .try_send((command, Some(reply))) + .map_err(|_| CommandRefused)?; + Ok(outcome) + } + + pub(crate) fn send_season_command(&self, command: SeasonCommand) -> Result<(), CommandRefused> { + self.season_commands + .try_send((command, None)) + .map_err(|_| CommandRefused) + } + + /// Queue a grab and hand back where its outcome will arrive. + pub(crate) fn request_season_grab( &self, command: SeasonCommand, - ) -> Result<(), mpsc::error::TrySendError> { - self.season_commands.try_send(command) + ) -> Result, CommandRefused> { + let (reply, outcome) = oneshot::channel(); + self.season_commands + .try_send((command, Some(reply))) + .map_err(|_| CommandRefused)?; + Ok(outcome) } /// Ask the metadata lane to refresh one title now. Best effort by diff --git a/crates/arr-daemon/src/manual.rs b/crates/arr-daemon/src/manual.rs index 4b7c168..d1968c8 100644 --- a/crates/arr-daemon/src/manual.rs +++ b/crates/arr-daemon/src/manual.rs @@ -5,9 +5,9 @@ //! search. Movies, episodes and seasons share one lane because they share //! one operator waiting on a 202. -use arr_api::{AppState, EpisodeCommand, MovieCommand, SeasonCommand}; +use arr_api::{AppState, EpisodeCommand, GrabResult, MovieCommand, SeasonCommand}; use arr_db::Db; -use tokio::sync::watch; +use tokio::sync::{oneshot, watch}; use crate::grab::{GrabAction, GrabError}; use crate::tv_grab::TvGrabAction; @@ -39,11 +39,11 @@ pub async fn run( } continue; } - 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), + request = state.next_movie_request() => request.map(|(command, reply)| (Command::Movie(command), reply)), + request = state.next_episode_request() => request.map(|(command, reply)| (Command::Episode(command), reply)), + request = state.next_season_request() => request.map(|(command, reply)| (Command::Season(command), reply)), }; - let Some(command) = command else { + let Some((command, reply)) = command else { return; }; match command { @@ -52,36 +52,53 @@ pub async fn run( tracing::warn!( "manual movie command received but Prowlarr or TMDB is not configured" ); + answer(reply, Err("Prowlarr or TMDB is not configured".to_owned())); continue; }; - if let Err(error) = handle_movie(movies, &database, command).await { + let outcome = handle_movie(movies, &database, command).await; + if let Err(error) = &outcome { tracing::error!(%error, "manual movie command failed"); } + answer(reply, outcome.map_err(|error| error.to_string())); } Command::Episode(command) => { let Some(tv) = &tv else { tracing::warn!( "manual episode command received but Prowlarr is not configured" ); + answer(reply, Err("Prowlarr is not configured".to_owned())); continue; }; - if let Err(error) = handle_episode(tv, &database, command).await { + let outcome = handle_episode(tv, &database, command).await; + if let Err(error) = &outcome { tracing::error!(%error, "manual episode command failed"); } + answer(reply, outcome.map_err(|error| error.to_string())); } Command::Season(command) => { let Some(tv) = &tv else { tracing::warn!("manual season command received but Prowlarr is not configured"); + answer(reply, Err("Prowlarr is not configured".to_owned())); continue; }; - if let Err(error) = handle_season(tv, &database, command).await { + let outcome = handle_season(tv, &database, command).await; + if let Err(error) = &outcome { tracing::error!(%error, "manual season command failed"); } + answer(reply, outcome.map_err(|error| error.to_string())); } } } } +/// Report a grab's outcome to the request waiting on it. The request may +/// have given up already, which is not an error here. +fn answer(reply: Option>, outcome: GrabResult) { + if let Some(reply) = reply { + let _ = reply.send(outcome); + } +} + async fn handle_movie( grab: &GrabAction, database: &Db, diff --git a/crates/arr-indexer/src/lib.rs b/crates/arr-indexer/src/lib.rs index e55ddf3..a9aaa52 100644 --- a/crates/arr-indexer/src/lib.rs +++ b/crates/arr-indexer/src/lib.rs @@ -209,11 +209,23 @@ impl ProwlarrClient { continue; } + let status = response.status(); + if !status.is_success() { + // Prowlarr explains a failed download as a Torznab error + // body ("Download selectors didn't match ..."); without it + // the operator sees a bare 500. + let body = response.text().await.unwrap_or_default(); + return Err(match torznab_error(&body) { + Some(description) => DownloadError::Rejected { + status, + description, + }, + None => DownloadError::Request { + status: Some(status), + }, + }); + } let body = response - .error_for_status() - .map_err(|error| DownloadError::Request { - status: error.status(), - })? .bytes() .await .map_err(|error| DownloadError::Request { @@ -273,6 +285,11 @@ pub enum Error { pub enum DownloadError { #[error("download link request failed (status: {status:?})")] Request { status: Option }, + #[error("indexer refused the download ({status}): {description}")] + Rejected { + status: StatusCode, + description: String, + }, #[error("download link is not a valid URL")] InvalidUrl, #[error("download link redirected to a location that cannot be used")] @@ -337,6 +354,15 @@ struct TorznabSearchMode { supported_parameters: Option, } +/// The description of a Torznab error body, if that is what `body` is. +fn torznab_error(body: &str) -> Option { + if response_root(body).ok()? != "error" { + return None; + } + let error: TorznabError = quick_xml::de::from_str(body).ok()?; + (!error.description.is_empty()).then_some(error.description) +} + fn parse_capabilities(body: &str) -> Result { if response_root(body)? == "error" { let error: TorznabError = @@ -599,6 +625,57 @@ mod tests { assert!(!sent_key); } + /// Prowlarr explains a failed download in a Torznab error body; that + /// reason is what tells the operator whether to retry or pick another + /// release, so it must survive into the error. + #[tokio::test] + async fn a_refused_download_carries_prowlarrs_reason() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/download")) + .respond_with(ResponseTemplate::new(500).set_body_string( + r#" +"#, + )) + .mount(&server) + .await; + + let error = client(&server) + .download(&format!("{}/download", server.uri())) + .await + .expect_err("a 500 is not a torrent"); + + assert_eq!( + error, + DownloadError::Rejected { + status: StatusCode::INTERNAL_SERVER_ERROR, + description: "Download selectors didn't match".to_owned(), + } + ); + } + + #[tokio::test] + async fn a_failed_download_without_a_reason_keeps_its_status() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/download")) + .respond_with(ResponseTemplate::new(502).set_body_string("bad gateway")) + .mount(&server) + .await; + + let error = client(&server) + .download(&format!("{}/download", server.uri())) + .await + .expect_err("a 502 is not a torrent"); + + assert_eq!( + error, + DownloadError::Request { + status: Some(StatusCode::BAD_GATEWAY), + } + ); + } + /// An HTML error page or a login form is not a torrent, and passing it on /// as metainfo would only fail later and less clearly. #[tokio::test]