diff --git a/.sqlx/query-74356a3425389d243ffdf162f4adbe8ae248e7e58ad8ecf635dcbe6a9d2a0289.json b/.sqlx/query-74356a3425389d243ffdf162f4adbe8ae248e7e58ad8ecf635dcbe6a9d2a0289.json new file mode 100644 index 0000000..898e8a3 --- /dev/null +++ b/.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" +} diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index a4f6d6f..cc5be15 100644 --- a/crates/arr-api/src/state.rs +++ b/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>>, subtitle_providers: Arc>>, translation_backends: Arc>>, + /// 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>, jellyfin: Option, 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) -> 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> { + self.command_timeout.as_ref() + } + /// Attach the `alass` binary this deployment runs (§15). Defaults to /// resolving `alass` from `PATH`. #[must_use] diff --git a/crates/arr-api/src/subtitle_settings.rs b/crates/arr-api/src/subtitle_settings.rs index 5b6a28b..70409a6 100644 --- a/crates/arr-api/src/subtitle_settings.rs +++ b/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, + ) { + 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 diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index bf8c0b2..1936e53 100644 --- a/crates/arr-daemon/src/main.rs +++ b/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 { 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(), ¬ifier)?; + // 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(), + ¬ifier, + &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>, notifier: &Notifier, + translators: &Translators, ) -> Result<(ReconcileLoop, Option, Option), 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 Result { +fn subtitle_action( + config: &Config, + notifier: &Notifier, + translators: &Translators, +) -> Result { 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>, + command_timeout: Option>, +} + #[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> { +fn translation_backends(config: &Config) -> Translators { let mut backends: Vec> = Vec::new(); + let mut command_timeout: Option> = None; #[cfg(feature = "translate-openai")] { @@ -619,7 +677,10 @@ fn translation_backends(config: &Config) -> Vec 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, } 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 { + Arc::clone(&self.timeout_ms) + } + async fn translate_inner(&self, batch: &Batch) -> Result> { let request_cues: Vec = 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), diff --git a/crates/arr-subs/tests/command.rs b/crates/arr-subs/tests/command.rs index e956af4..49ef69b 100644 --- a/crates/arr-subs/tests/command.rs +++ b/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;