fix: report manual grabs the daemon could not send

The grab endpoints answered 202 as soon as the command was queued, and
the daemon only logged a failure, so the UI said "grab sent" for a
torrent qBittorrent never got. The handlers now wait up to 30s for the
daemon's outcome and answer 502 with its reason. Prowlarr's Torznab
error description is kept on failed downloads so that reason is useful.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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::{

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<crate::state::GrabResult>,
+) -> Result<(StatusCode, Json<Accepted>), 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

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.

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<Upstreams>,
     database: Option<Db>,
-    movie_commands: mpsc::Sender<MovieCommand>,
-    pending_movie_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MovieCommand>>>,
-    episode_commands: mpsc::Sender<EpisodeCommand>,
-    pending_episode_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<EpisodeCommand>>>,
-    season_commands: mpsc::Sender<SeasonCommand>,
-    pending_season_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<SeasonCommand>>>,
+    movie_commands: mpsc::Sender<Request<MovieCommand>>,
+    pending_movie_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<Request<MovieCommand>>>>,
+    episode_commands: mpsc::Sender<Request<EpisodeCommand>>,
+    pending_episode_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<Request<EpisodeCommand>>>>,
+    season_commands: mpsc::Sender<Request<SeasonCommand>>,
+    pending_season_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<Request<SeasonCommand>>>>,
     metadata_commands: mpsc::Sender<MetadataCommand>,
     pending_metadata_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MetadataCommand>>>,
     subtitle_providers: Arc<Vec<Arc<dyn Provider>>>,
@@ -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> = (C, Option<oneshot::Sender<GrabResult>>);
+
+/// 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<MovieCommand> {
+        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<Request<MovieCommand>> {
         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<EpisodeCommand> {
+        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<Request<EpisodeCommand>> {
         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<SeasonCommand> {
+        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<Request<SeasonCommand>> {
         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<MovieCommand>> {
-        self.movie_commands.try_send(command)
+    ) -> Result<oneshot::Receiver<GrabResult>, 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<EpisodeCommand>> {
-        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<oneshot::Receiver<GrabResult>, 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<SeasonCommand>> {
-        self.season_commands.try_send(command)
+    ) -> Result<oneshot::Receiver<GrabResult>, 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

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<oneshot::Sender<GrabResult>>, outcome: GrabResult) {
+    if let Some(reply) = reply {
+        let _ = reply.send(outcome);
+    }
+}
+
 async fn handle_movie(
     grab: &GrabAction,
     database: &Db,

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<StatusCode> },
+    #[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<String>,
 }
 
+/// The description of a Torznab error body, if that is what `body` is.
+fn torznab_error(body: &str) -> Option<String> {
+    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<Capabilities, CapabilityError> {
     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#"<?xml version="1.0" encoding="UTF-8"?>
+<error code="500" description="Download selectors didn't match" />"#,
+            ))
+            .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]