fix(daemon): consume the MovieCommand channel

Nothing drained AppState's movie_commands mpsc, so manual search and
one-click grab were accepted with 202 and then did nothing until the
64-slot buffer filled and the endpoint started 503ing.

A new daemon task drains it: Search resets the movie's backoff and
runs the targeted-search + grab lane immediately; Grab sends the
already-chosen release straight to Transmission.

Closes #107

crates/arr-daemon/src/grab.rs

@@ -373,6 +373,64 @@ impl GrabAction {
             )
             .await
     }
+
+    /// The manual trigger (§6.2, §9.3, issue #107): reset the backoff so
+    /// `search_due` cannot skip the title, then run the same search-and-grab
+    /// path as a tick, scoped to this one movie instead of the tick's
+    /// `MOVIES_PER_TICK` batch.
+    pub(crate) async fn search_now(
+        &self,
+        database: &Db,
+        movie_id: i64,
+    ) -> Result<Option<Outcome>, GrabError> {
+        reset_search_backoff(database, movie_id).await?;
+        let Some(movie) = pending_movie(database, movie_id).await? else {
+            return Ok(None);
+        };
+        let Some(movie) = self.refresh_metadata(database, movie).await? else {
+            return Ok(None);
+        };
+        let searchable = self.indexers.searchable().await?;
+        if searchable.is_empty() {
+            tracing::warn!("no indexer advertises a text search; nothing can be grabbed");
+            return Ok(None);
+        }
+        self.grab_one(database, &movie, &searchable).await
+    }
+
+    /// The manual one-click grab (§9.3, issue #107): the release is already
+    /// chosen, so this skips search and scoring and sends it straight to
+    /// Transmission.
+    pub(crate) async fn grab_release_now(
+        &self,
+        database: &Db,
+        movie_id: i64,
+        release_id: i64,
+    ) -> Result<Option<Outcome>, GrabError> {
+        let Some(loaded) = database.movie_policy(movie_id).await? else {
+            return Ok(None);
+        };
+        let Some(title) = movie_title(database, movie_id).await? else {
+            return Ok(None);
+        };
+        let Some(release) = load_release(database, movie_id, release_id).await? else {
+            return Ok(None);
+        };
+        let blacklist = Blacklist::load(database.pool()).await?;
+        self.grabber
+            .send_winner(
+                database,
+                &GrabTarget {
+                    scope: GrabScope::Movie { movie_id },
+                    title: &title,
+                    counts_as_attempt: false,
+                },
+                &loaded,
+                &blacklist,
+                release,
+            )
+            .await
+    }
 }
 
 /// Sending a chosen release to Transmission and recording the grab.
@@ -842,6 +900,109 @@ fn search_due(movie: &PendingMovie) -> bool {
     backoff_elapsed(movie.search_attempts, movie.last_searched_at.as_deref())
 }
 
+/// Same shape as [`pending_movies`], scoped to one id and without the tick's
+/// batch limit — the manual trigger already named which title to search.
+async fn pending_movie(database: &Db, movie_id: i64) -> Result<Option<PendingMovie>, GrabError> {
+    let row = sqlx::query!(
+        r#"
+        SELECT   m.id    AS "id!: i64",
+                 m.tmdb_id AS "tmdb_id!: i64",
+                 m.title AS "title!: String",
+                 m.year,
+                 m.original_language,
+                 m.search_attempts AS "search_attempts!: i64",
+                 m.last_searched_at,
+                 m.digital_release,
+                 m.metadata_refreshed_at
+        FROM     movies m
+        WHERE    m.id = ?
+          AND    m.wanted = 1
+          AND    m.blocked = 0
+          AND    NOT EXISTS (
+                     SELECT 1 FROM media_files f
+                     WHERE  f.owner_kind = 'movie' AND f.owner_id = m.id
+                 )
+          AND    NOT EXISTS (
+                     SELECT 1 FROM grabs g
+                     WHERE  g.target_kind = 'movie' AND g.target_id = m.id
+                       AND  g.state IN ('sent', 'downloaded', 'imported')
+                 )
+        "#,
+        movie_id
+    )
+    .fetch_optional(database.pool())
+    .await?;
+
+    Ok(row.map(|row| PendingMovie {
+        id: row.id,
+        tmdb_id: row.tmdb_id,
+        title: row.title,
+        year: row.year,
+        original_language: row.original_language,
+        search_attempts: row.search_attempts,
+        last_searched_at: row.last_searched_at,
+        digital_release: row.digital_release,
+        metadata_refreshed_at: row.metadata_refreshed_at,
+    }))
+}
+
+/// Unconditional, unlike [`record_search`]: the manual trigger's whole point
+/// is to ignore the exponential backoff (§6.2).
+async fn reset_search_backoff(database: &Db, movie_id: i64) -> Result<(), GrabError> {
+    sqlx::query!(
+        "UPDATE movies SET search_attempts = 0, last_searched_at = NULL,
+                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+         WHERE id = ?",
+        movie_id
+    )
+    .execute(database.pool())
+    .await?;
+    Ok(())
+}
+
+async fn movie_title(database: &Db, movie_id: i64) -> Result<Option<String>, GrabError> {
+    Ok(
+        sqlx::query_scalar!("SELECT title FROM movies WHERE id = ?", movie_id)
+            .fetch_optional(database.pool())
+            .await?,
+    )
+}
+
+/// The chosen release, still associated with the movie and not hard-failed —
+/// a manual grab may take a `waived` release (§9.3), never a `rejected` one.
+async fn load_release(
+    database: &Db,
+    movie_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     movie_releases mr ON mr.release_id = r.id
+        WHERE    mr.movie_id = ? AND r.id = ? AND r.verdict IN ('eligible', 'waived')
+        "#,
+        movie_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 §6.2 targeted-search backoff, shared by movie and episode search.
 pub(crate) fn backoff_elapsed(search_attempts: i64, last_searched_at: Option<&str>) -> bool {
     let Some(last_searched_at) = last_searched_at else {

crates/arr-daemon/src/main.rs

@@ -7,6 +7,7 @@ mod grab;
 mod import;
 mod indexers;
 mod jellyfin;
+mod manual;
 mod notify;
 mod reaper;
 pub mod reconcile;
@@ -100,8 +101,8 @@ enum Error {
     },
     #[error("serve: {0}")]
     Serve(std::io::Error),
-    #[error("reconcile task: {0}")]
-    ReconcileTask(#[from] tokio::task::JoinError),
+    #[error("background task: {0}")]
+    BackgroundTask(#[from] tokio::task::JoinError),
 }
 
 async fn run() -> Result<(), Error> {
@@ -120,7 +121,8 @@ async fn run() -> Result<(), Error> {
         None
     };
     let notifier = Notifier::new(config.ntfy_url.clone())?;
-    let reconcile = reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
+    let (reconcile, manual_grab) =
+        reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
 
     // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
     // needs its own TMDB client for `movie/lookup`.
@@ -135,9 +137,9 @@ async fn run() -> Result<(), Error> {
     if let Some(tmdb_url) = config.tmdb_url {
         upstreams = upstreams.with_tmdb_url(tmdb_url);
     }
-    let state = AppState::new(upstreams)?.with_database(database);
+    let state = AppState::new(upstreams)?.with_database(database.clone());
 
-    let app = arr_api::router(state)
+    let app = arr_api::router(state.clone())
         .merge(arr_compat::router(compat))
         .fallback(web::serve)
         .layer(TraceLayer::new_for_http());
@@ -152,7 +154,11 @@ async fn run() -> Result<(), Error> {
     tracing::info!(addr = %config.bind_addr, docs = arr_api::DOCS_PATH, "listening");
 
     let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
-    let mut reconcile_task = tokio::spawn(reconcile.run(shutdown_rx));
+    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));
     let signal_tx = shutdown_tx.clone();
     let server = async move {
         axum::serve(listener, app)
@@ -168,10 +174,19 @@ async fn run() -> Result<(), Error> {
         result = &mut server => {
             let _ = shutdown_tx.send(true);
             reconcile_task.await?;
+            manual_task.await?;
             result.map_err(Error::Serve)
         }
         result = &mut reconcile_task => {
             result?;
+            let _ = shutdown_tx.send(true);
+            manual_task.await?;
+            server.await.map_err(Error::Serve)
+        }
+        result = &mut manual_task => {
+            result?;
+            let _ = shutdown_tx.send(true);
+            reconcile_task.await?;
             server.await.map_err(Error::Serve)
         }
     }
@@ -180,14 +195,19 @@ async fn run() -> Result<(), Error> {
 /// Wire the reconcile lanes (DESIGN.md §8). Grab and RSS both need a
 /// 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.
 fn reconcile_loop(
     database: &Db,
     config: &Config,
     transmission: &arr_dl::TransmissionClient,
     tmdb: Option<&Arc<TmdbClient>>,
     notifier: &Notifier,
-) -> Result<ReconcileLoop, Error> {
-    let mut reconcile = ReconcileLoop::new(database.clone());
+) -> Result<(ReconcileLoop, Option<GrabAction>), Error> {
+    let reconcile = ReconcileLoop::new(database.clone());
     let seeding = SeedingRules::new(
         SeedingLimits {
             ratio: config.seed_ratio_limit,
@@ -213,20 +233,14 @@ fn reconcile_loop(
         .map(|key| arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key))
         .transpose()?;
 
-    if let (Some(prowlarr), Some(tmdb)) = (prowlarr.as_ref(), tmdb) {
-        reconcile = reconcile.register(
-            Tick::Reconcile,
-            GrabAction::new(
-                prowlarr.clone(),
-                transmission.clone(),
-                config.download_dir.clone(),
-                seeding.clone(),
-            )
-            .with_tmdb(Arc::clone(tmdb)),
-        );
-    } else {
-        tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed");
-    }
+    let (mut reconcile, manual_grab) = register_movie_grab(
+        reconcile,
+        prowlarr.as_ref(),
+        transmission,
+        config,
+        &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).
@@ -298,7 +312,57 @@ fn reconcile_loop(
         );
     }
 
-    Ok(reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone())))
+    let reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone()));
+    Ok((reconcile, manual_grab))
+}
+
+/// Register the movie grab lane on `reconcile` and hand back a second,
+/// independent instance for `manual::run` (issue #107) — the manual trigger
+/// needs the same search-and-grab path on demand rather than on the tick's
+/// schedule, and `ReconcileLoop::register` takes ownership of the one it
+/// ticks. `None` when Prowlarr or TMDB is not configured; nothing can be
+/// grabbed either way.
+fn register_movie_grab(
+    mut reconcile: ReconcileLoop,
+    prowlarr: Option<&arr_indexer::ProwlarrClient>,
+    transmission: &arr_dl::TransmissionClient,
+    config: &Config,
+    seeding: &SeedingRules,
+    tmdb: Option<&Arc<TmdbClient>>,
+) -> (ReconcileLoop, Option<GrabAction>) {
+    let Some((prowlarr, tmdb)) = prowlarr.zip(tmdb) else {
+        tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed");
+        return (reconcile, None);
+    };
+    reconcile = reconcile.register(
+        Tick::Reconcile,
+        movie_grab_action(prowlarr, transmission, config, seeding, tmdb),
+    );
+    let manual_grab = Some(movie_grab_action(
+        prowlarr,
+        transmission,
+        config,
+        seeding,
+        tmdb,
+    ));
+    (reconcile, manual_grab)
+}
+
+/// The movie grab lane, built fresh for each caller.
+fn movie_grab_action(
+    prowlarr: &arr_indexer::ProwlarrClient,
+    transmission: &arr_dl::TransmissionClient,
+    config: &Config,
+    seeding: &SeedingRules,
+    tmdb: &Arc<TmdbClient>,
+) -> GrabAction {
+    GrabAction::new(
+        prowlarr.clone(),
+        transmission.clone(),
+        config.download_dir.clone(),
+        seeding.clone(),
+    )
+    .with_tmdb(Arc::clone(tmdb))
 }
 
 /// Stop accepting on Ctrl-C, or on the SIGTERM a service manager sends.

crates/arr-daemon/src/manual.rs

@@ -0,0 +1,250 @@
+//! 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;
+//! `Grab` sends an already-chosen release straight to Transmission, skipping
+//! search.
+
+use arr_api::{AppState, MovieCommand};
+use arr_db::Db;
+use tokio::sync::watch;
+
+use crate::grab::{GrabAction, GrabError};
+
+/// 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.
+pub async fn run(
+    state: AppState,
+    database: Db,
+    grab: Option<GrabAction>,
+    mut shutdown: watch::Receiver<bool>,
+) {
+    loop {
+        let command = tokio::select! {
+            biased;
+            changed = shutdown.changed() => {
+                if changed.is_err() || *shutdown.borrow() {
+                    return;
+                }
+                continue;
+            }
+            command = state.next_movie_command() => command,
+        };
+        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");
+        }
+    }
+}
+
+async fn handle(grab: &GrabAction, database: &Db, command: MovieCommand) -> Result<(), GrabError> {
+    match command {
+        MovieCommand::Search { movie_id } => {
+            grab.search_now(database, movie_id).await?;
+        }
+        MovieCommand::Grab {
+            movie_id,
+            release_id,
+        } => {
+            grab.grab_release_now(database, movie_id, release_id)
+                .await?;
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use std::collections::HashMap;
+    use std::path::PathBuf;
+
+    use arr_db::Db;
+    use arr_dl::TransmissionClient;
+    use arr_indexer::ProwlarrClient;
+    use wiremock::matchers::{method, path, path_regex, query_param};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+    use crate::grab::{SeedingLimits, SeedingRules};
+
+    async fn database_with_wanted_movie() -> (tempfile::TempDir, Db) {
+        let dir = tempfile::tempdir().unwrap();
+        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
+        database.migrate().await.unwrap();
+        sqlx::query(
+            "INSERT INTO movies (tmdb_id, title, year, original_language, root_id)
+             SELECT 693134, 'Dune Part Two', 2024, 'en', id
+             FROM roots WHERE kind = 'movie' AND audience = 'main'",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+        (dir, database)
+    }
+
+    async fn transmission() -> MockServer {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "result": "success",
+                "arguments": {"torrent-added": {"id": 1, "name": "x", "hashString": "aaaa"}}
+            })))
+            .mount(&server)
+            .await;
+        server
+    }
+
+    fn grab_action(indexer: &MockServer, transmission: &MockServer) -> GrabAction {
+        GrabAction::new(
+            ProwlarrClient::new(indexer.uri(), "key").unwrap(),
+            TransmissionClient::new(&transmission.uri()).unwrap(),
+            PathBuf::from("/mnt/media/transmission/complete"),
+            SeedingRules::new(
+                SeedingLimits {
+                    ratio: 1.5,
+                    idle_minutes: 60,
+                },
+                HashMap::new(),
+            ),
+        )
+    }
+
+    /// Issue #107: a `Search` command must reset the backoff, not merely be
+    /// accepted onto the channel. `search_attempts` starts at 5 (a 7-day
+    /// backoff, nowhere near elapsed) so a normal tick would skip this movie;
+    /// ending at 1 rather than 6 proves the reset happened, not just a
+    /// one-off bypass of the gate.
+    #[tokio::test]
+    async fn a_search_command_resets_backoff_and_searches_now() {
+        let (_dir, database) = database_with_wanted_movie().await;
+        sqlx::query(
+            "UPDATE movies SET search_attempts = 5,
+                    last_searched_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-6 days')",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+
+        let indexer = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/api/v1/indexer"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
+                {"id": 7, "name": "tracker", "enable": true}
+            ])))
+            .mount(&indexer)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/7/api"))
+            .and(query_param("t", "caps"))
+            .respond_with(ResponseTemplate::new(200).set_body_string(
+                r#"<caps><searching><search available="yes" supportedParams="q"/></searching></caps>"#,
+            ))
+            .mount(&indexer)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/7/api"))
+            .and(query_param("t", "search"))
+            .respond_with(
+                ResponseTemplate::new(200).set_body_string("<rss><channel></channel></rss>"),
+            )
+            .mount(&indexer)
+            .await;
+        let transmission = transmission().await;
+        let grab = grab_action(&indexer, &transmission);
+
+        handle(&grab, &database, MovieCommand::Search { movie_id: 1 })
+            .await
+            .unwrap();
+
+        let searched = indexer
+            .received_requests()
+            .await
+            .unwrap()
+            .into_iter()
+            .any(|request| {
+                request.url.path() == "/7/api"
+                    && request
+                        .url
+                        .query_pairs()
+                        .any(|(name, value)| name == "t" && value == "search")
+            });
+        assert!(searched, "the reset backoff must let the search run now");
+        let (attempts, last_searched_at): (i64, Option<String>) =
+            sqlx::query_as("SELECT search_attempts, last_searched_at FROM movies WHERE id = 1")
+                .fetch_one(database.pool())
+                .await
+                .unwrap();
+        assert_eq!(attempts, 1, "reset to 0, then one real attempt — not 6");
+        assert!(last_searched_at.is_some());
+    }
+
+    /// Issue #107: a `Grab` command sends the already-chosen release straight
+    /// to Transmission — no indexer search at all.
+    #[tokio::test]
+    async fn a_grab_command_sends_the_chosen_release_without_searching() {
+        let (_dir, database) = database_with_wanted_movie().await;
+        let indexer = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path_regex("^/dl/"))
+            .respond_with(ResponseTemplate::new(302).insert_header(
+                "location",
+                "magnet:?xt=urn:btih:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+            ))
+            .mount(&indexer)
+            .await;
+        let transmission = transmission().await;
+        let grab = grab_action(&indexer, &transmission);
+
+        let release_id = sqlx::query(
+            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict)
+             VALUES (7, 'chosen', 'Dune.Part.Two.2024.2160p.WEB-DL', 1, ?, '{}', 900, 'eligible')",
+        )
+        .bind(format!("{}/dl/chosen", indexer.uri()))
+        .execute(database.pool())
+        .await
+        .unwrap()
+        .last_insert_rowid();
+        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (1, ?)")
+            .bind(release_id)
+            .execute(database.pool())
+            .await
+            .unwrap();
+
+        handle(
+            &grab,
+            &database,
+            MovieCommand::Grab {
+                movie_id: 1,
+                release_id,
+            },
+        )
+        .await
+        .unwrap();
+
+        let grabbed: i64 = sqlx::query_scalar("SELECT count(*) FROM grabs WHERE target_id = 1")
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+        assert_eq!(grabbed, 1);
+        let searched = indexer
+            .received_requests()
+            .await
+            .unwrap()
+            .into_iter()
+            .any(|request| {
+                request
+                    .url
+                    .query_pairs()
+                    .any(|(name, value)| name == "t" && value == "search")
+            });
+        assert!(!searched, "a chosen release must not trigger a search");
+    }
+}