Merge #219: apply the remote-command timeout setting

Closes #219

.sqlx/query-74356a3425389d243ffdf162f4adbe8ae248e7e58ad8ecf635dcbe6a9d2a0289.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT remote_command_timeout_seconds AS \"remote_command_timeout_seconds!: i64\"\n           FROM subtitle_settings WHERE id = 1",
+  "describe": {
+    "columns": [
+      {
+        "name": "remote_command_timeout_seconds!: i64",
+        "ordinal": 0,
+        "type_info": "Integer",
+        "origin": {
+          "Table": {
+            "table": "subtitle_settings",
+            "name": "remote_command_timeout_seconds"
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "74356a3425389d243ffdf162f4adbe8ae248e7e58ad8ecf635dcbe6a9d2a0289"
+}

crates/arr-api/src/state.rs

@@ -1,7 +1,7 @@
 //! What the API needs to answer a request: one HTTP client and the addresses
 //! of the three upstreams the service cannot work without (DESIGN.md §3).
 
-use std::sync::Arc;
+use std::sync::{atomic::AtomicU64, Arc};
 use std::time::Duration;
 
 use arr_db::Db;
@@ -81,6 +81,10 @@ pub struct AppState {
     pending_metadata_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MetadataCommand>>>,
     subtitle_providers: Arc<Vec<Arc<dyn Provider>>>,
     translation_backends: Arc<Vec<Arc<dyn Backend>>>,
+    /// The remote-command backend's live timeout, in milliseconds (#219).
+    /// `None` unless the daemon compiled and configured that backend; the
+    /// settings API writes it on every edit of the row.
+    command_timeout: Option<Arc<AtomicU64>>,
     jellyfin: Option<JellyfinClient>,
     syncer: Syncer,
 }
@@ -155,6 +159,7 @@ impl AppState {
             pending_metadata_commands: Arc::new(tokio::sync::Mutex::new(pending_metadata_commands)),
             subtitle_providers: Arc::new(Vec::new()),
             translation_backends: Arc::new(Vec::new()),
+            command_timeout: None,
             jellyfin: None,
             syncer: Syncer::default(),
         })
@@ -191,6 +196,15 @@ impl AppState {
         self
     }
 
+    /// Attach the cell the remote-command backend re-reads per batch (#219),
+    /// so an edit of `remote_command_timeout_seconds` reaches it without a
+    /// restart. Absent when that backend is not configured.
+    #[must_use]
+    pub fn with_command_timeout(mut self, timeout: Arc<AtomicU64>) -> Self {
+        self.command_timeout = Some(timeout);
+        self
+    }
+
     /// Attach the Jellyfin client, so a manual subtitle grab or translation
     /// can trigger the same library refresh import does (§7.5, §15).
     #[must_use]
@@ -219,6 +233,10 @@ impl AppState {
             .find(|backend| backend.id().as_str() == id)
     }
 
+    pub(crate) fn command_timeout(&self) -> Option<&Arc<AtomicU64>> {
+        self.command_timeout.as_ref()
+    }
+
     /// Attach the `alass` binary this deployment runs (§15). Defaults to
     /// resolving `alass` from `PATH`.
     #[must_use]

crates/arr-api/src/subtitle_settings.rs

@@ -197,6 +197,7 @@ pub async fn update(
     let input = parsed(body)?;
     input.validate().map_err(ApiError::Invalid)?;
     let columns = input.into_columns()?;
+    let timeout_seconds = columns.remote_command_timeout_seconds;
     sqlx::query!(
         r#"UPDATE subtitle_settings SET
                wanted_languages = ?, providers_enabled = ?, translation_engine = ?,
@@ -213,6 +214,17 @@ pub async fn update(
     )
     .execute(pool(&state)?)
     .await?;
+    // Issue #219: the row alone never reaches the running backend. Push it
+    // into the cell the remote-command translator re-reads per batch; the
+    // cell counts milliseconds, the row counts seconds.
+    if let Some(timeout) = state.command_timeout() {
+        timeout.store(
+            u64::try_from(timeout_seconds)
+                .unwrap_or(u64::MAX)
+                .saturating_mul(1_000),
+            std::sync::atomic::Ordering::Relaxed,
+        );
+    }
     Ok(Json(load(&state).await?))
 }
 
@@ -223,24 +235,39 @@ mod tests {
     use crate::{router, AppState, Upstreams};
 
     async fn application() -> (tempfile::TempDir, String) {
+        let (dir, base, _timeout) = application_with_timeout().await;
+        (dir, base)
+    }
+
+    /// The same app, with the remote-command backend's live timeout cell
+    /// attached — what the daemon wires up when that backend is configured.
+    async fn application_with_timeout() -> (
+        tempfile::TempDir,
+        String,
+        std::sync::Arc<std::sync::atomic::AtomicU64>,
+    ) {
+        use std::sync::{atomic::AtomicU64, Arc};
+
         let dir = tempfile::tempdir().expect("tempdir");
         let database = arr_db::Db::connect(dir.path().join("arr.db"))
             .await
             .expect("connect database");
         database.migrate().await.expect("migrate database");
+        let timeout = Arc::new(AtomicU64::new(30_000));
         let state = AppState::new(Upstreams::new(
             "http://127.0.0.1:1".into(),
             "http://127.0.0.1:1".into(),
         ))
         .expect("state")
-        .with_database(database);
+        .with_database(database)
+        .with_command_timeout(timeout.clone());
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
             .expect("bind");
         let address = listener.local_addr().expect("address");
         let app = router(state);
         tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
-        (dir, format!("http://{address}"))
+        (dir, format!("http://{address}"), timeout)
     }
 
     fn valid_input() -> serde_json::Value {
@@ -307,6 +334,33 @@ mod tests {
         assert_eq!(refetched, updated);
     }
 
+    /// Issue #219: the row alone never reaches the running backend, so a PUT
+    /// must push its value into the cell the command translator re-reads.
+    #[tokio::test]
+    async fn a_put_updates_the_live_command_timeout() {
+        let (_dir, base, timeout) = application_with_timeout().await;
+        assert_eq!(
+            timeout.load(std::sync::atomic::Ordering::Relaxed),
+            30_000,
+            "seeded from the row at startup"
+        );
+
+        reqwest::Client::new()
+            .put(format!("{base}/api/settings/subtitles"))
+            .json(&valid_input())
+            .send()
+            .await
+            .expect("put settings")
+            .error_for_status()
+            .expect("valid input accepted");
+
+        assert_eq!(
+            timeout.load(std::sync::atomic::Ordering::Relaxed),
+            45_000,
+            "the edited value reaches the running backend"
+        );
+    }
+
     /// Whether a known engine is selectable depends on which `translate-*`
     /// features this binary was built with, so the test asks the build rather
     /// than assuming. With no feature on, every engine is uncompiled and must

crates/arr-daemon/src/main.rs

@@ -18,7 +18,7 @@ mod tv_grab;
 mod web;
 
 use std::process::ExitCode;
-use std::sync::Arc;
+use std::sync::{atomic::AtomicU64, Arc};
 
 use arr_api::{AppState, Upstreams};
 use arr_compat::CompatState;
@@ -117,6 +117,7 @@ fn api_state(
     config: &Config,
     database: &Db,
     jellyfin: arr_api::jellyfin::JellyfinClient,
+    translators: &Translators,
 ) -> Result<AppState, Error> {
     let mut upstreams =
         Upstreams::new(config.prowlarr_url.clone(), config.transmission_url.clone())
@@ -126,16 +127,20 @@ fn api_state(
         upstreams = upstreams.with_tmdb_url(tmdb_url);
     }
 
-    Ok(AppState::new(upstreams)?
+    let mut state = AppState::new(upstreams)?
         .with_database(database.clone())
         .with_subtitle_providers(subtitle_providers(
             config.opensubtitles_api_key.clone(),
             config.opensubtitles_username.clone(),
             config.opensubtitles_password.clone(),
         ))
-        .with_translation_backends(translation_backends(config))
+        .with_translation_backends(translators.backends.clone())
         .with_jellyfin(jellyfin)
-        .with_syncer(arr_subs::Syncer::new().with_binary(config.alass_path.clone())))
+        .with_syncer(arr_subs::Syncer::new().with_binary(config.alass_path.clone()));
+    if let Some(timeout) = &translators.command_timeout {
+        state = state.with_command_timeout(Arc::clone(timeout));
+    }
+    Ok(state)
 }
 
 async fn run() -> Result<(), Error> {
@@ -155,8 +160,19 @@ async fn run() -> Result<(), Error> {
     };
     let notifier = Notifier::new(config.ntfy_url.clone())?;
     let api_jellyfin = jellyfin_client(&config)?;
-    let (reconcile, manual_grab, manual_tv) =
-        reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
+    // Built once and shared: the API's translate handler and the reconcile
+    // lane must see the same backends, or the command translator's live
+    // timeout cell (#219) would fork.
+    let translators = translation_backends(&config);
+    seed_command_timeout(&database, &translators).await?;
+    let (reconcile, manual_grab, manual_tv) = reconcile_loop(
+        &database,
+        &config,
+        &transmission,
+        tmdb.as_ref(),
+        &notifier,
+        &translators,
+    )?;
     // Issue #176: the on-demand half of the metadata lane needs its own
     // handle — the sweep's `SeriesRefreshAction` is owned by `ReconcileLoop`,
     // and the compat shim takes the other clone below.
@@ -169,7 +185,7 @@ async fn run() -> Result<(), Error> {
         compat = compat.with_tmdb(tmdb);
     }
 
-    let state = api_state(&config, &database, api_jellyfin)?;
+    let state = api_state(&config, &database, api_jellyfin, &translators)?;
 
     let app = arr_api::router(state.clone())
         .merge(arr_compat::router(compat))
@@ -247,6 +263,30 @@ async fn run() -> Result<(), Error> {
     }
 }
 
+/// Seed the command translator's live timeout from the settings row, so a
+/// restart does not fall back to the compiled-in default until the next
+/// settings edit (issue #219). The row is guaranteed to exist and to carry a
+/// positive value — migration `0025` seeds it and the column CHECK enforces
+/// it.
+async fn seed_command_timeout(database: &Db, translators: &Translators) -> Result<(), Error> {
+    let Some(cell) = &translators.command_timeout else {
+        return Ok(());
+    };
+    let seconds: i64 = sqlx::query_scalar!(
+        r#"SELECT remote_command_timeout_seconds AS "remote_command_timeout_seconds!: i64"
+           FROM subtitle_settings WHERE id = 1"#
+    )
+    .fetch_one(database.pool())
+    .await?;
+    cell.store(
+        u64::try_from(seconds)
+            .unwrap_or(u64::MAX)
+            .saturating_mul(1_000),
+        std::sync::atomic::Ordering::Relaxed,
+    );
+    Ok(())
+}
+
 /// 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.
@@ -262,6 +302,7 @@ fn reconcile_loop(
     transmission: &arr_dl::TransmissionClient,
     tmdb: Option<&Arc<TmdbClient>>,
     notifier: &Notifier,
+    translators: &Translators,
 ) -> Result<(ReconcileLoop, Option<GrabAction>, Option<TvGrabAction>), Error> {
     let reconcile = ReconcileLoop::new(database.clone());
     let seeding = SeedingRules::new(
@@ -334,7 +375,10 @@ fn reconcile_loop(
     );
 
     // §15: subtitle gaps are reconciled from the same rows the API writes.
-    reconcile = reconcile.register(Tick::Reconcile, subtitle_action(config, notifier)?);
+    reconcile = reconcile.register(
+        Tick::Reconcile,
+        subtitle_action(config, notifier, translators)?,
+    );
 
     // §9.5 *needs a decision* and *broken* both go to the operator alone;
     // without a topic configured there is nowhere to send them.
@@ -485,14 +529,18 @@ fn jellyfin_client(config: &Config) -> Result<arr_api::jellyfin::JellyfinClient,
 /// — the compiled-in backends (#191–#193) still need their bootstrap wiring
 /// (model names, base URLs) — so the translate step records "not compiled"
 /// and backs off rather than failing obscurely.
-fn subtitle_action(config: &Config, notifier: &Notifier) -> Result<SubtitleAction, Error> {
+fn subtitle_action(
+    config: &Config,
+    notifier: &Notifier,
+    translators: &Translators,
+) -> Result<SubtitleAction, Error> {
     let action = SubtitleAction::new(
         subtitle_providers(
             config.opensubtitles_api_key.clone(),
             config.opensubtitles_username.clone(),
             config.opensubtitles_password.clone(),
         ),
-        translation_backends(config),
+        translators.backends.clone(),
         arr_subs::Syncer::new().with_binary(config.alass_path.clone()),
         arr_probe::Extractor::new().with_binary(config.ffmpeg_path.clone()),
         jellyfin_client(config)?,
@@ -538,14 +586,23 @@ fn subtitle_providers(
     providers
 }
 
-/// The translation backends this deployment can offer (DESIGN.md §15,
-/// issue #216).
+/// The translation backends this deployment can offer, built once at startup
+/// and shared by the API and the reconcile lane (DESIGN.md §15, issue #216).
 ///
 /// Which cargo features this binary was built with decides what could ever
 /// be here (`compiled_engines`); credentials decide what actually is, same
 /// split `subtitle_providers` makes for search. Which one of these a
 /// translation *uses* is the `translation_engine` database setting, read per
 /// request (#198) — this only decides which ids exist to be picked.
+///
+/// When the remote-command backend is one of them, its live timeout cell
+/// rides along (#219): the API writes it on every settings edit, so the row
+/// reaches the running process without a restart.
+struct Translators {
+    backends: Vec<std::sync::Arc<dyn arr_subs::Backend>>,
+    command_timeout: Option<Arc<AtomicU64>>,
+}
+
 #[cfg_attr(
     not(any(
         feature = "translate-openai",
@@ -555,8 +612,9 @@ fn subtitle_providers(
     )),
     allow(unused_variables, unused_mut)
 )]
-fn translation_backends(config: &Config) -> Vec<std::sync::Arc<dyn arr_subs::Backend>> {
+fn translation_backends(config: &Config) -> Translators {
     let mut backends: Vec<std::sync::Arc<dyn arr_subs::Backend>> = Vec::new();
+    let mut command_timeout: Option<Arc<AtomicU64>> = None;
 
     #[cfg(feature = "translate-openai")]
     {
@@ -619,7 +677,10 @@ fn translation_backends(config: &Config) -> Vec<std::sync::Arc<dyn arr_subs::Bac
                 timeout: arr_subs::COMMAND_DEFAULT_TIMEOUT,
             };
             match arr_subs::Command::new(command_config) {
-                Ok(backend) => backends.push(std::sync::Arc::new(backend)),
+                Ok(backend) => {
+                    command_timeout = Some(backend.timeout_cell());
+                    backends.push(std::sync::Arc::new(backend));
+                }
                 Err(error) => tracing::warn!(%error, "remote-command translator not available"),
             }
         } else {
@@ -627,5 +688,8 @@ fn translation_backends(config: &Config) -> Vec<std::sync::Arc<dyn arr_subs::Bac
         }
     }
 
-    backends
+    Translators {
+        backends,
+        command_timeout,
+    }
 }

crates/arr-subs/src/command.rs

@@ -18,7 +18,12 @@
 //! exit, timeout and unparseable output fail loudly here, and a reply with
 //! the wrong cue count or numbering never becomes a subtitle.
 
-use std::{io::ErrorKind, process::Stdio, time::Duration};
+use std::{
+    io::ErrorKind,
+    process::Stdio,
+    sync::{atomic::AtomicU64, atomic::Ordering, Arc},
+    time::Duration,
+};
 
 use serde::{Deserialize, Serialize};
 use tokio::{
@@ -52,7 +57,10 @@ pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120);
 /// Settings for the remote-command backend (DESIGN.md §10, issue #198).
 ///
 /// The template is not a credential, but it names a host — it arrives from
-/// bootstrap config like the API keys do, never from the database.
+/// bootstrap config like the API keys do, never from the database. The
+/// timeout is the *initial* value only: the running backend re-reads a
+/// shared cell per batch ([`Command::timeout_cell`]), so the operator's edit
+/// of `remote_command_timeout_seconds` applies without a restart (#219).
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CommandConfig {
     /// The command to run per batch, e.g. `ssh box claude -p`. May carry
@@ -67,6 +75,10 @@ pub struct CommandConfig {
 pub struct Command {
     config: CommandConfig,
     id: BackendId,
+    /// The live per-batch timeout in milliseconds, shared with whoever wired
+    /// this backend to the settings row. Milliseconds so the tests — and an
+    /// impatient operator — can go below one second.
+    timeout_ms: Arc<AtomicU64>,
 }
 
 impl Command {
@@ -83,11 +95,25 @@ impl Command {
             });
         }
         Ok(Self {
+            timeout_ms: Arc::new(AtomicU64::new(
+                u64::try_from(config.timeout.as_millis()).unwrap_or(u64::MAX),
+            )),
             config,
             id: BackendId::new(BACKEND_NAME),
         })
     }
 
+    /// The cell this backend re-reads for every batch, in milliseconds.
+    ///
+    /// Whoever constructs the backend — the daemon, at startup — hands a
+    /// clone of this to the settings API, which stores the row's
+    /// `remote_command_timeout_seconds` into it on every edit. That is the
+    /// path from the database to the running process #219 found missing.
+    #[must_use]
+    pub fn timeout_cell(&self) -> Arc<AtomicU64> {
+        Arc::clone(&self.timeout_ms)
+    }
+
     async fn translate_inner(&self, batch: &Batch) -> Result<Vec<TranslatedCue>> {
         let request_cues: Vec<WireCue> = batch
             .cues
@@ -114,14 +140,17 @@ impl Command {
         }
         let (program, args) = tokens.split_first().expect("the constructor rejects empty");
 
+        // Re-read per call: the settings API writes this cell when the
+        // operator edits `remote_command_timeout_seconds` (#219).
+        let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed));
         let work = run(program, args, payload, self.id.clone());
 
         // Dropping `work` on timeout drops the child, whose `kill_on_drop`
         // ends the command — nothing is left resident (DESIGN.md §15).
-        match time::timeout(self.config.timeout, work).await {
+        match time::timeout(timeout, work).await {
             Err(_) => Err(Error::Transport {
                 backend: self.id.clone(),
-                source: format!("timed out after {:?}", self.config.timeout).into(),
+                source: format!("timed out after {timeout:?}").into(),
             }),
             Ok(Err(err)) => Err(err),
             Ok(Ok(out)) => Self::parse_reply(&self.id, &out),

crates/arr-subs/tests/command.rs

@@ -237,4 +237,36 @@ async fn a_hanging_command_is_killed_at_the_timeout() {
     }
 }
 
+/// Issue #219: the timeout is not frozen at construction. The settings API
+/// stores the row's value into [`Command::timeout_cell`] on every edit, and
+/// the next batch must already run under it.
+#[tokio::test]
+async fn an_edited_timeout_reaches_the_next_call() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let script = stub(dir.path(), "hang.sh", "sleep 30\n");
+    let backend =
+        Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
+
+    // The edit that a PUT of remote_command_timeout_seconds performs.
+    backend
+        .timeout_cell()
+        .store(150, std::sync::atomic::Ordering::Relaxed);
+
+    let error = translate(
+        &backend,
+        &cues(&["hello"]),
+        &Language::Other("en".to_owned()),
+        &Language::PortuguesePortugal,
+    )
+    .await
+    .expect_err("the edited timeout must apply without a restart");
+
+    match error {
+        Error::Transport { source, .. } => {
+            assert!(source.to_string().contains("150ms"), "{source}");
+        }
+        other => panic!("expected Transport, got {other:?}"),
+    }
+}
+
 const DEFAULT: Duration = arr_subs::COMMAND_DEFAULT_TIMEOUT;