From 8c5613b2471a9d6342f85b9b5a004e7220b25bbd Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Tue, 25 Aug 2026 05:54:19 +0100 Subject: [PATCH 1/5] feat(arr): probe methods for subtitle providers and engines --- crates/arr-subs/src/command.rs | 33 ++++++++++- crates/arr-subs/src/deepl.rs | 56 +++++++++++++++++- crates/arr-subs/src/google.rs | 66 ++++++++++++++++++++- crates/arr-subs/src/lib.rs | 18 +++++- crates/arr-subs/src/openai.rs | 80 +++++++++++++++++++++++++- crates/arr-subs/src/opensubtitles.rs | 54 ++++++++++++++++- crates/arr-subs/src/podnapisi.rs | 8 +++ crates/arr-subs/src/sync.rs | 73 ++++++++++++++++++++++- crates/arr-subs/src/translate.rs | 20 ++++++- crates/arr-subs/tests/command.rs | 24 +++++++- crates/arr-subs/tests/opensubtitles.rs | 39 +++++++++++++ crates/arr-subs/tests/podnapisi.rs | 23 ++++++++ 12 files changed, 480 insertions(+), 14 deletions(-) diff --git a/crates/arr-subs/src/command.rs b/crates/arr-subs/src/command.rs index 465fc7a..b20a333 100644 --- a/crates/arr-subs/src/command.rs +++ b/crates/arr-subs/src/command.rs @@ -35,8 +35,8 @@ use tokio::{ use arr_core::Language; use crate::translate::{ - strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture, - TranslatedCue, + strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, ProbeFuture, Result, + TranslateFuture, TranslatedCue, }; const BACKEND_NAME: &str = "command"; @@ -210,6 +210,35 @@ impl Backend for Command { fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> { Box::pin(async move { self.translate_inner(batch).await }) } + + /// Runs the template once over an empty batch (#200): the lamp is the + /// command starting and exiting cleanly, not what it says. The same + /// timeout cell bounds it, so a wedged template cannot pile up probes. + fn probe(&self) -> ProbeFuture<'_> { + Box::pin(async move { + let payload = format!( + "{}\n[]\n", + system_prompt(&Language::Other("en".to_owned()), &Language::PortuguesePortugal) + ); + let mut tokens = argv(&self.config.template); + for token in &mut tokens { + *token = token + .replace("{source}", language_tag(&Language::Other("en".to_owned()))) + .replace("{target}", language_tag(&Language::PortuguesePortugal)); + } + let (program, args) = tokens.split_first().expect("the constructor rejects empty"); + + let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed)); + match time::timeout(timeout, run(program, args, payload, self.id.clone())).await { + Err(_) => Err(Error::Transport { + backend: self.id(), + source: format!("timed out after {timeout:?}").into(), + }), + Ok(Err(err)) => Err(err), + Ok(Ok(_)) => Ok(()), + } + }) + } } /// Spawn the command once, feed it `payload`, and collect its stdout. diff --git a/crates/arr-subs/src/deepl.rs b/crates/arr-subs/src/deepl.rs index e18c81f..f4c7217 100644 --- a/crates/arr-subs/src/deepl.rs +++ b/crates/arr-subs/src/deepl.rs @@ -24,7 +24,9 @@ use serde::{Deserialize, Serialize}; use arr_core::Language; -use crate::translate::{Backend, BackendId, Batch, Error, TranslateFuture, TranslatedCue}; +use crate::translate::{ + Backend, BackendId, Batch, Error, ProbeFuture, TranslateFuture, TranslatedCue, +}; use crate::{Error as SubsError, ProviderId, Result}; /// The `DeepL` API endpoint for keys ending in `:fx` (the free tier). @@ -282,6 +284,31 @@ impl Backend for DeepL { .collect()) }) } + + /// `/v2/usage` is the documented key check: one cheap authenticated GET + /// that spends none of the character quota (#200). + fn probe(&self) -> ProbeFuture<'_> { + Box::pin(async move { + let reply = self + .http + .get(self.join("usage")) + .header("DeepL-Auth-Key", &self.config.auth_key) + .send() + .await + .map_err(|error| Error::Transport { + backend: self.id(), + source: Box::new(error), + })?; + match reply.status() { + StatusCode::OK => Ok(()), + StatusCode::FORBIDDEN => Err(Error::Unauthorized { backend: self.id() }), + status => Err(Error::Malformed { + backend: self.id(), + detail: format!("status {}", status.as_u16()), + }), + } + }) + } } fn retry_after(headers: &reqwest::header::HeaderMap) -> Option { @@ -534,4 +561,31 @@ mod tests { }; assert_eq!(retry_after, Some(Duration::from_secs(30))); } + + #[tokio::test] + async fn a_probe_asks_usage_and_judges_the_key() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v2/usage")) + .and(header("DeepL-Auth-Key", KEY)) + .respond_with(ResponseTemplate::new(200).set_body_string("{}")) + .mount(&server) + .await; + deepl(format!("{}/v2", server.uri())) + .probe() + .await + .expect("a 200 usage answer is a lit lamp"); + + server.reset().await; + Mock::given(method("GET")) + .and(path("/v2/usage")) + .respond_with(ResponseTemplate::new(403)) + .mount(&server) + .await; + let error = deepl(format!("{}/v2", server.uri())) + .probe() + .await + .expect_err("bad key"); + assert!(matches!(error, Error::Unauthorized { .. }), "got {error:?}"); + } } diff --git a/crates/arr-subs/src/google.rs b/crates/arr-subs/src/google.rs index 244e1dc..55f4d25 100644 --- a/crates/arr-subs/src/google.rs +++ b/crates/arr-subs/src/google.rs @@ -26,7 +26,9 @@ use serde::{Deserialize, Serialize}; use arr_core::Language; -use crate::translate::{Backend, BackendId, Batch, Error, TranslateFuture, TranslatedCue}; +use crate::translate::{ + Backend, BackendId, Batch, Error, ProbeFuture, TranslateFuture, TranslatedCue, +}; use crate::{Error as SubsError, ProviderId, Result}; /// The public Google Cloud Translation v2 endpoint. @@ -291,6 +293,39 @@ impl Backend for Google { .collect()) }) } + + /// `languages` is the cheapest authenticated v2 call and spends none of + /// the translation quota (#200). The request itself is fixed, so a 400 + /// can only mean the key was refused — Google reports bad keys as 400, + /// not 401. + fn probe(&self) -> ProbeFuture<'_> { + Box::pin(async move { + let mut url = self.base_url.clone(); + url.set_path(&format!("{}/languages", url.path().trim_end_matches('/'))); + url.query_pairs_mut() + .append_pair("key", &self.config.api_key); + + let reply = self + .http + .get(url) + .send() + .await + .map_err(|error| Error::Transport { + backend: self.id(), + source: Box::new(error), + })?; + match reply.status() { + StatusCode::OK => Ok(()), + StatusCode::BAD_REQUEST | StatusCode::FORBIDDEN => { + Err(Error::Unauthorized { backend: self.id() }) + } + status => Err(Error::Malformed { + backend: self.id(), + detail: format!("status {}", status.as_u16()), + }), + } + }) + } } fn api_error_detail(status: StatusCode, body: &[u8]) -> String { @@ -367,7 +402,7 @@ fn numeric_entity(entity: &str) -> Option { #[cfg(test)] mod tests { use arr_core::Language; - use wiremock::matchers::method; + use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; use super::{Google, GoogleConfig}; @@ -584,4 +619,31 @@ mod tests { // A stray ampersand that is not an entity stays one. assert_eq!(unescape_html("fish & chips"), "fish & chips"); } + + #[tokio::test] + async fn a_probe_lists_languages_and_judges_the_key() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v2/languages")) + .and(query_param("key", KEY)) + .respond_with(ResponseTemplate::new(200).set_body_string("{}")) + .mount(&server) + .await; + google(format!("{}/v2", server.uri())) + .probe() + .await + .expect("a languages answer is a lit lamp"); + + server.reset().await; + // Google reports a bad key as a 400, not a 401. + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(400)) + .mount(&server) + .await; + let error = google(format!("{}/v2", server.uri())) + .probe() + .await + .expect_err("bad key"); + assert!(matches!(error, Error::Unauthorized { .. }), "got {error:?}"); + } } diff --git a/crates/arr-subs/src/lib.rs b/crates/arr-subs/src/lib.rs index b68c4e0..7e3f855 100644 --- a/crates/arr-subs/src/lib.rs +++ b/crates/arr-subs/src/lib.rs @@ -55,7 +55,8 @@ pub use opensubtitles::{moviehash, OpenSubtitles, OpenSubtitlesConfig}; pub use podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL}; pub use srt::Cue; pub use sync::{ - Outcome, Rejection, Settled, SyncState, Syncer, DEFAULT_BINARY as ALASS_DEFAULT_BINARY, + binary_present, Outcome, Rejection, Settled, SyncState, Syncer, + DEFAULT_BINARY as ALASS_DEFAULT_BINARY, }; pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue}; @@ -65,6 +66,9 @@ pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, Transl #[cfg(test)] use wiremock as _; +/// One health probe (#200): reachability and, where they exist, credentials. +pub type ProbeFuture<'a> = Pin> + Send + 'a>>; + /// The candidates one search turned up. pub type SearchFuture<'a> = Pin>> + Send + 'a>>; @@ -131,6 +135,14 @@ pub trait Provider: fmt::Debug + Send + Sync { /// Anything in [`Error`], and [`Error::NotFound`] when the candidate has /// gone away between the search and the download. fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a>; + + /// One health probe (#200): reachable, and credentials accepted. + /// + /// Cheap by contract — the endpoint carrying it is polled. It must cost + /// no search or download quota: a probe that spends budget is a lamp + /// that drains the account it watches. [`Error::Unauthorized`] means the + /// credentials were refused; anything else is an outage. + fn probe(&self) -> ProbeFuture<'_>; } #[cfg(test)] @@ -190,6 +202,10 @@ mod tests { }) }) } + + fn probe(&self) -> super::ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } fn request() -> SearchRequest { diff --git a/crates/arr-subs/src/openai.rs b/crates/arr-subs/src/openai.rs index cc5b17c..5453997 100644 --- a/crates/arr-subs/src/openai.rs +++ b/crates/arr-subs/src/openai.rs @@ -26,8 +26,8 @@ use serde::{Deserialize, Serialize}; use arr_core::Language; use crate::translate::{ - strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture, - TranslatedCue, + strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, ProbeFuture, Result, + TranslateFuture, TranslatedCue, }; const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1/"; @@ -262,6 +262,36 @@ impl Backend for OpenAi { fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> { Box::pin(async move { self.translate_inner(batch).await }) } + + /// `models` is a read-only listing: reachable and key accepted, nothing + /// spent (#200). Endpoints that need no key are probed the same way, + /// just without the header. + fn probe(&self) -> ProbeFuture<'_> { + Box::pin(async move { + let mut url = self.base_url.clone(); + url.set_path(&format!("{}/models", url.path().trim_end_matches('/'))); + let mut request = self.http.get(url); + if let Some(key) = &self.config.api_key { + request = request.bearer_auth(key); + } + match request.send().await { + Ok(response) => match response.status() { + StatusCode::OK => Ok(()), + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => { + Err(Error::Unauthorized { backend: self.id() }) + } + status => Err(Error::Malformed { + backend: self.id(), + detail: format!("status {}", status.as_u16()), + }), + }, + Err(error) => Err(Error::Transport { + backend: self.id(), + source: Box::new(error), + }), + } + }) + } } fn retry_after(response: &reqwest::Response) -> Option { @@ -382,4 +412,50 @@ mod tests { assert_eq!(strip_code_fence("```\n[1,2]\n```"), "[1,2]"); assert_eq!(strip_code_fence("[1,2]"), "[1,2]"); } + + #[tokio::test] + async fn a_probe_lists_models_and_judges_the_key() { + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/models")) + .and(header("Authorization", "Bearer test-key")) + .respond_with(ResponseTemplate::new(200).set_body_string("{}")) + .mount(&server) + .await; + OpenAi::with_base_url( + "gpt-4o-mini", + super::OpenAiConfig { + api_key: Some("test-key".to_owned()), + }, + &format!("{}/v1", server.uri()), + ) + .expect("client builds") + .probe() + .await + .expect("a models answer is a lit lamp"); + + server.reset().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + let error = OpenAi::with_base_url( + "gpt-4o-mini", + super::OpenAiConfig { + api_key: Some("test-key".to_owned()), + }, + &format!("{}/v1", server.uri()), + ) + .expect("client builds") + .probe() + .await + .expect_err("bad key"); + assert!( + matches!(error, crate::translate::Error::Unauthorized { .. }), + "got {error:?}" + ); + } } diff --git a/crates/arr-subs/src/opensubtitles.rs b/crates/arr-subs/src/opensubtitles.rs index a456111..d4b601b 100644 --- a/crates/arr-subs/src/opensubtitles.rs +++ b/crates/arr-subs/src/opensubtitles.rs @@ -222,6 +222,23 @@ impl OpenSubtitles { query: Option<&[(String, String)]>, json: Option<&serde_json::Value>, bearer: Option<&str>, + ) -> Result { + let response = self.send_raw(method, path, query, json, bearer).await?; + self.check_status(response).await + } + + /// Send a request and hand back the raw response, whatever its status. + /// + /// The health probe (#200) needs statuses `check_status` folds away: a + /// 400 there means "the API refused the request *after* accepting the + /// key", which is exactly what a lamp wants to hear. + async fn send_raw( + &self, + method: Method, + path: &str, + query: Option<&[(String, String)]>, + json: Option<&serde_json::Value>, + bearer: Option<&str>, ) -> Result { let url = self.base_url.join(path).map_err(|err| Error::Malformed { provider: self.id.clone(), @@ -244,11 +261,10 @@ impl OpenSubtitles { } tracing::debug!(provider = %self.id, path, "OpenSubtitles request"); - let response = request.send().await.map_err(|err| Error::Transport { + request.send().await.map_err(|err| Error::Transport { provider: self.id.clone(), source: Box::new(err), - })?; - self.check_status(response).await + }) } /// Fetch (or refetch) the user token downloads travel under. @@ -292,6 +308,34 @@ impl OpenSubtitles { self.login().await } + /// One health probe (#200): reachable, and the API key accepted. + /// + /// Deliberately not a real search — searches spend quota, and health is + /// polled. An unparseable candidate id is refused with a 4xx *after* + /// authentication, so anything short of an auth refusal or a 5xx is + /// proof the key was accepted, at no cost. + async fn probe_inner(&self) -> Result<()> { + let response = self + .send_raw( + Method::GET, + "subtitles", + Some(&[("id".to_owned(), "not-a-number".to_owned())]), + None, + None, + ) + .await?; + match response.status() { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => Err(Error::Unauthorized { + provider: self.id.clone(), + }), + status if status.is_server_error() => Err(Error::Malformed { + provider: self.id.clone(), + detail: format!("status {}", status.as_u16()), + }), + _ => Ok(()), + } + } + async fn search_inner(&self, request: &SearchRequest) -> Result> { let hash = moviehash(&request.file.path, request.file.size).map_err(|source| Error::Io { @@ -432,6 +476,10 @@ impl Provider for OpenSubtitles { fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> { Box::pin(async move { self.download_inner(id).await }) } + + fn probe(&self) -> crate::ProbeFuture<'_> { + Box::pin(async move { self.probe_inner().await }) + } } fn retry_after(response: &reqwest::Response) -> Option { diff --git a/crates/arr-subs/src/podnapisi.rs b/crates/arr-subs/src/podnapisi.rs index d6206d7..78e3bba 100644 --- a/crates/arr-subs/src/podnapisi.rs +++ b/crates/arr-subs/src/podnapisi.rs @@ -255,6 +255,14 @@ impl Provider for Podnapisi { fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> { Box::pin(async move { self.download_candidate(id).await }) } + + /// Anonymous, so the lamp is reachability alone (#200): the site root + /// answers, no quota spent. + fn probe(&self) -> crate::ProbeFuture<'_> { + Box::pin(async move { + self.send(self.base_url.clone(), None).await.map(|_| ()) + }) + } } /// Configuration for a [`Podnapisi`]. diff --git a/crates/arr-subs/src/sync.rs b/crates/arr-subs/src/sync.rs index e47b575..4c2d4bf 100644 --- a/crates/arr-subs/src/sync.rs +++ b/crates/arr-subs/src/sync.rs @@ -14,7 +14,7 @@ //! reference is always the media file itself; subtitle and video are both on //! disk by the time this runs. -use std::{ffi::OsString, path::Path, process::Stdio, time::Duration}; +use std::{ffi::OsStr, ffi::OsString, path::Path, process::Stdio, time::Duration}; use tokio::process::Command; @@ -37,6 +37,38 @@ pub const MAX_SHIFT: Duration = Duration::from_secs(60); /// How much of `alass`'s stderr is kept in an error. const STDERR_LIMIT: usize = 512; +/// Whether a configured external binary resolves to an executable. +/// +/// A name with any path component (`/usr/local/bin/alass`, `./alass`) must +/// exist and be executable exactly where it points; a bare name (`alass`) is +/// resolved through `PATH`, the way spawning it would (#200). No process is +/// started: the health endpoint is polled, and launching the real thing per +/// poll would be neither cheap nor side-effect-free. +#[must_use] +pub fn binary_present(binary: &OsStr) -> bool { + let path = Path::new(binary); + if path.components().count() > 1 { + return is_executable_file(path); + } + std::env::var_os("PATH").is_some_and(|paths| { + std::env::split_paths(&paths) + .map(|dir| dir.join(path)) + .any(|candidate| is_executable_file(&candidate)) + }) +} + +#[cfg(unix)] +fn is_executable_file(path: &Path) -> bool { + use std::os::unix::fs::PermissionsExt; + path.is_file() + && std::fs::metadata(path).is_ok_and(|meta| meta.permissions().mode() & 0o111 != 0) +} + +#[cfg(not(unix))] +fn is_executable_file(path: &Path) -> bool { + path.is_file() +} + /// Runs `alass` over one subtitle against its video. #[derive(Clone, Debug)] pub struct Syncer { @@ -74,6 +106,18 @@ impl Syncer { self } + /// The configured binary, as health probes (#200) report it. + #[must_use] + pub fn binary_path(&self) -> &OsStr { + &self.binary + } + + /// Whether [`Self::binary_path`] resolves to an executable (#200). + #[must_use] + pub fn present(&self) -> bool { + binary_present(&self.binary) + } + /// Sync `subtitle` against `video`, in place on disk. /// /// Returns the synced SRT text on acceptance; the input file is never @@ -637,4 +681,31 @@ mod tests { "1 of 2 cues survived" ); } + + /// The health lamp (#200) is made of this, so it must agree with what + /// spawning would do: an explicit path is judged where it points, a bare + /// name through `PATH`, and a missing or non-executable file is absent. + #[test] + fn binary_presence_follows_the_same_rules_spawning_does() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().expect("tempdir"); + let executable = dir.path().join("alass"); + std::fs::write(&executable, "#!/bin/sh\n").expect("write"); + std::fs::set_permissions(&executable, std::fs::Permissions::from_mode(0o755)) + .expect("chmod"); + let plain = dir.path().join("not-executable"); + std::fs::write(&plain, "text").expect("write"); + + assert!(super::binary_present(executable.as_os_str())); + assert!(!super::binary_present(plain.as_os_str())); + assert!(!super::binary_present(std::ffi::OsStr::new( + "/nowhere/alass" + ))); + // `sh` is on PATH of every machine that runs these tests. + assert!(super::binary_present(std::ffi::OsStr::new("sh"))); + assert!(Syncer::new() + .with_binary(executable.as_os_str()) + .present()); + } } diff --git a/crates/arr-subs/src/translate.rs b/crates/arr-subs/src/translate.rs index 7b9423b..2c1efe7 100644 --- a/crates/arr-subs/src/translate.rs +++ b/crates/arr-subs/src/translate.rs @@ -105,6 +105,9 @@ pub struct TranslatedCue { pub type TranslateFuture<'a> = Pin>> + Send + 'a>>; +/// One health probe (#200): reachable and credentials accepted. +pub type ProbeFuture<'a> = Pin> + Send + 'a>>; + /// One translation backend (DESIGN.md §15). /// /// Boxed futures rather than `async fn` for the same reason as @@ -146,6 +149,15 @@ pub trait Backend: fmt::Debug + Send + Sync { /// /// Anything in [`Error`]. fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a>; + + /// One health probe (#200): reachable, and credentials accepted. + /// + /// For the remote-command backend this means the configured command + /// runs and exits cleanly. Cheap by contract — the endpoint carrying it + /// is polled, and it must spend none of the quota translation spends. + /// [`Error::Unauthorized`] means the credentials were refused; anything + /// else is an outage. + fn probe(&self) -> ProbeFuture<'_>; } /// Result alias for translation. @@ -424,7 +436,9 @@ mod tests { use arr_core::Language; - use super::{translate, Backend, BackendId, Batch, Error, TranslateFuture, TranslatedCue}; + use super::{ + translate, Backend, BackendId, Batch, Error, ProbeFuture, TranslateFuture, TranslatedCue, + }; use crate::srt::Cue; /// Uppercases every cue and records how it was called; misbehaves on cue. @@ -466,6 +480,10 @@ mod tests { Ok(reply) }) } + + fn probe(&self) -> ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } fn cues(texts: &[&str]) -> Vec { diff --git a/crates/arr-subs/tests/command.rs b/crates/arr-subs/tests/command.rs index 49ef69b..15dcde2 100644 --- a/crates/arr-subs/tests/command.rs +++ b/crates/arr-subs/tests/command.rs @@ -21,7 +21,7 @@ use std::{ use arr_core::Language; use arr_subs::translate::{translate, Error}; -use arr_subs::{Command, CommandConfig}; +use arr_subs::{Backend as _, Command, CommandConfig}; fn cues(texts: &[&str]) -> Vec { texts @@ -270,3 +270,25 @@ async fn an_edited_timeout_reaches_the_next_call() { } const DEFAULT: Duration = arr_subs::COMMAND_DEFAULT_TIMEOUT; + +/// The lamp (#200) is the command starting and exiting cleanly — not what it +/// says. An empty batch goes in; only the exit status is judged. +#[tokio::test] +async fn a_probe_runs_the_template_and_wants_a_clean_exit() { + let dir = tempfile::tempdir().expect("tempdir"); + + let ok = stub(dir.path(), "ok.sh", "cat > /dev/null\n"); + Command::new(config(&ok.display().to_string(), DEFAULT)) + .expect("backend constructs") + .probe() + .await + .expect("a clean exit is a lit lamp"); + + let failing = stub(dir.path(), "fail.sh", "echo nope >&2\nexit 3\n"); + let error = Command::new(config(&failing.display().to_string(), DEFAULT)) + .expect("backend constructs") + .probe() + .await + .expect_err("non-zero exit"); + assert!(matches!(error, Error::Transport { .. }), "got {error:?}"); +} diff --git a/crates/arr-subs/tests/opensubtitles.rs b/crates/arr-subs/tests/opensubtitles.rs index 6e53a89..babb568 100644 --- a/crates/arr-subs/tests/opensubtitles.rs +++ b/crates/arr-subs/tests/opensubtitles.rs @@ -424,3 +424,42 @@ async fn an_id_this_provider_never_offered_is_not_found() { other => panic!("expected NotFound, got {other:?}"), } } + +#[tokio::test] +async fn a_probe_passes_when_the_key_is_accepted_even_on_a_refused_request() { + let server = MockServer::start().await; + // An unparseable candidate id is rejected *after* authentication: any + // 4xx short of 401/403 is proof the key was accepted, and spends no + // search quota (#200). + Mock::given(method("GET")) + .and(path("/api/v1/subtitles")) + .respond_with(ResponseTemplate::new(400)) + .mount(&server) + .await; + + client(&server) + .probe() + .await + .expect("a refused request still proves the key"); +} + +#[tokio::test] +async fn a_probe_reports_refused_credentials_and_outages() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/v1/subtitles")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + let err = client(&server).probe().await.expect_err("bad key"); + assert!(matches!(err, Error::Unauthorized { .. })); + + server.reset().await; + Mock::given(method("GET")) + .and(path("/api/v1/subtitles")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + let err = client(&server).probe().await.expect_err("outage"); + assert!(matches!(err, Error::Malformed { .. })); +} diff --git a/crates/arr-subs/tests/podnapisi.rs b/crates/arr-subs/tests/podnapisi.rs index dc11213..e684957 100644 --- a/crates/arr-subs/tests/podnapisi.rs +++ b/crates/arr-subs/tests/podnapisi.rs @@ -282,3 +282,26 @@ async fn an_id_this_provider_never_issued_is_not_found() { assert!(matches!(error, Error::NotFound { .. }), "got {error:?}"); } + +#[tokio::test] +async fn a_probe_is_reachability_alone() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/subtitles/")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + provider(&server) + .probe() + .await + .expect("the anonymous provider is up when the site answers"); + + server.reset().await; + Mock::given(method("GET")) + .and(path("/subtitles/")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + let error = provider(&server).probe().await.expect_err("site down"); + assert!(matches!(error, Error::Malformed { .. }), "got {error:?}"); +} From be7fa87e74ee080046e8ff0e650ec38292e652bc Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Tue, 25 Aug 2026 06:04:30 +0100 Subject: [PATCH 2/5] feat(api): subtitle lamps in the health report --- crates/arr-api/src/health.rs | 373 +++++++++++++++++++++++- crates/arr-api/src/lib.rs | 6 +- crates/arr-api/src/state.rs | 21 ++ crates/arr-api/src/subtitle_settings.rs | 4 +- crates/arr-api/src/subtitles.rs | 16 + 5 files changed, 408 insertions(+), 12 deletions(-) diff --git a/crates/arr-api/src/health.rs b/crates/arr-api/src/health.rs index 53832eb..a660390 100644 --- a/crates/arr-api/src/health.rs +++ b/crates/arr-api/src/health.rs @@ -6,11 +6,15 @@ //! the body carries the verdict, so a degraded service can still explain //! itself to the UI instead of looking like a fourth outage. +use std::ffi::OsStr; + use axum::extract::State; use axum::Json; use serde::Serialize; use utoipa::ToSchema; +use arr_subs::{binary_present, translate as backend_error}; + use crate::state::AppState; /// Whether the service as a whole can do its job. @@ -68,6 +72,44 @@ impl Check { } } +/// One enabled subtitle provider's verdict (#200). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)] +pub struct ProviderCheck { + /// The provider's id as settings name it. + pub id: String, + pub status: Status, + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +/// The subtitle lane's verdicts (DESIGN.md §15, #200). Only what is actually +/// in use appears here: providers nobody enabled and engines nobody selected +/// cannot be broken. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)] +pub struct SubtitleHealth { + /// One entry per provider the settings enable, in that order. + pub providers: Vec, + /// The selected translation engine, or `None` when none is chosen — + /// which is a configuration state, not an outage. + #[serde(skip_serializing_if = "Option::is_none")] + pub translation: Option, + /// The `alass` sync binary, present at its configured path. + pub alass: Check, + /// The `ffmpeg` extraction binary, present at its configured path. + pub ffmpeg: Check, +} + +impl SubtitleHealth { + /// Every verdict the lane carries, for the overall status. + fn statuses(&self) -> impl Iterator + '_ { + self.providers + .iter() + .map(|provider| provider.status) + .chain(self.translation.iter().map(|check| check.status)) + .chain([self.alass.status, self.ffmpeg.status]) + } +} + /// The body of `GET /api/health`. #[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)] pub struct HealthReport { @@ -78,9 +120,11 @@ pub struct HealthReport { pub prowlarr: Check, pub transmission: Check, pub tmdb: Check, + pub subtitles: SubtitleHealth, } -/// Report reachability of Prowlarr, Transmission and TMDB. +/// Report reachability of Prowlarr, Transmission and TMDB, plus the +/// subtitle upstreams (#200). #[utoipa::path( get, path = "/api/health", @@ -90,17 +134,15 @@ pub struct HealthReport { ), )] pub async fn health(State(state): State) -> Json { - // Three independent network probes; serialising them would make the - // endpoint as slow as the sum of the timeouts. - let (prowlarr, transmission, tmdb) = tokio::join!( - probe_prowlarr(&state), - probe_transmission(&state), - probe_tmdb(&state), - ); + // Independent network probes; serialising them would make the endpoint + // as slow as the sum of the timeouts. + let (prowlarr, transmission, tmdb, subtitles) = + tokio::join!(probe_prowlarr(&state), probe_transmission(&state), probe_tmdb(&state), probe_subtitles(&state)); let status = if [prowlarr.status, transmission.status, tmdb.status] - .iter() - .all(|s| *s == Status::Ok) + .into_iter() + .chain(subtitles.statuses()) + .all(|check| check == Status::Ok) { Health::Ok } else { @@ -113,6 +155,7 @@ pub async fn health(State(state): State) -> Json { prowlarr, transmission, tmdb, + subtitles, }) } @@ -181,3 +224,313 @@ async fn probe_tmdb(state: &AppState) -> Check { fn describe(err: reqwest::Error) -> String { err.without_url().to_string() } + +// ---- the subtitle lane (DESIGN.md §15, #200) ----------------------------- + +/// Probe every subtitle lamp: enabled providers, the selected engine, and +/// the two binaries. The settings row decides what is *in use* — a provider +/// nobody enabled or an engine nobody selected is not probed at all, so it +/// cannot fail a lamp. +async fn probe_subtitles(state: &AppState) -> SubtitleHealth { + let settings = match state.database() { + Some(_) => crate::subtitle_settings::load(state).await.ok(), + None => None, + }; + // No readable row means nothing is known to be in use; the lamps stay + // quiet rather than failing on data the operator has not entered yet. + let enabled = settings + .as_ref() + .map(|settings| settings.providers_enabled.as_slice()) + .unwrap_or_default(); + let engine = settings + .as_ref() + .and_then(|settings| settings.translation_engine.as_deref()); + + let mut probes = Vec::with_capacity(enabled.len()); + for id in enabled { + probes.push(probe_provider(state, id).await); + } + let translation = match engine { + Some(engine) => Some(probe_engine(state, engine).await), + None => None, + }; + + SubtitleHealth { + providers: probes, + translation, + alass: binary_check("alass", state.syncer().binary_path()), + ffmpeg: binary_check("ffmpeg", state.ffmpeg_binary()), + } +} + +/// Reachable and credentials accepted, per enabled provider (#200). +/// +/// A provider that is enabled but was never attached is unconfigured, not +/// unreachable: its credentials are bootstrap config that this deployment +/// simply does not have. +async fn probe_provider(state: &AppState, id: &str) -> ProviderCheck { + let check = match state.subtitle_provider(id) { + Some(provider) => check_from(provider.probe().await), + None => Check::unconfigured(format!( + "{id} is enabled but not configured — credentials are bootstrap config" + )), + }; + ProviderCheck { + id: id.to_owned(), + status: check.status, + detail: check.detail, + } +} + +/// Reachable, credentials accepted — for the remote-command backend, +/// "reachable" means the command ran and exited cleanly (#200). +async fn probe_engine(state: &AppState, engine: &str) -> Check { + match state.translation_backend(engine) { + Some(backend) => backend_check(backend.probe().await), + None => Check::unconfigured(format!( + "'{engine}' is selected but unavailable in this build" + )), + } +} + +/// One probe verdict, whatever kind of upstream produced it. A refused key +/// gets its own wording because it never resolves by retrying. The two error +/// enums — providers' and backends' — carry the same shape for this purpose. +fn check_from(result: Result<(), arr_subs::Error>) -> Check { + match result { + Ok(()) => Check::ok(), + Err(arr_subs::Error::Unauthorized { .. }) => Check::unreachable("credentials refused"), + Err(error) => Check::unreachable(error.to_string()), + } +} + +fn backend_check(result: Result<(), backend_error::Error>) -> Check { + match result { + Ok(()) => Check::ok(), + Err(backend_error::Error::Unauthorized { .. }) => { + Check::unreachable("credentials refused") + } + Err(error) => Check::unreachable(error.to_string()), + } +} + +/// A binary lamp (#200): present and executable at its configured path. +/// Nothing is spawned — the endpoint is polled, and starting `ffmpeg` per +/// poll would be neither cheap nor side-effect-free. +fn binary_check(name: &str, binary: &OsStr) -> Check { + if binary_present(binary) { + Check::ok() + } else { + Check::unreachable(format!("{name} not found at {}", binary.to_string_lossy())) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arr_db::Db; + use arr_subs::{ + Backend, CandidateId, DownloadFuture, Provider, ProviderId, SearchFuture, SearchRequest, + Syncer, + }; + use arr_subs::translate as backend; + + use crate::{router, AppState, Upstreams}; + + /// A provider whose lamp is what `lamp` says — the probe under test is + /// ours, so the stub never touches the network. + #[derive(Debug)] + struct StubProvider { + id: &'static str, + up: bool, + } + + impl Provider for StubProvider { + fn id(&self) -> ProviderId { + ProviderId::new(self.id) + } + + fn search<'a>(&'a self, _request: &'a SearchRequest) -> SearchFuture<'a> { + Box::pin(async move { Ok(Vec::new()) }) + } + + fn download<'a>(&'a self, _id: &'a CandidateId) -> DownloadFuture<'a> { + unreachable!("health probes never download"); + } + + fn probe(&self) -> arr_subs::ProbeFuture<'_> { + let result = if self.up { + Ok(()) + } else { + Err(arr_subs::Error::Unauthorized { + provider: ProviderId::new(self.id), + }) + }; + Box::pin(async move { result }) + } + } + + /// Same idea for the selected engine; the closure rebuilds its error per + /// call because probing borrows. + #[derive(Debug)] + struct StubBackend { + #[allow(dead_code)] + detail: &'static str, + } + + impl Backend for StubBackend { + fn id(&self) -> backend::BackendId { + backend::BackendId::new("openai") + } + + fn supports(&self, _target: &arr_core::Language) -> bool { + true + } + + fn translate<'a>(&'a self, _batch: &'a backend::Batch) -> backend::TranslateFuture<'a> { + unreachable!("health probes never translate"); + } + + fn probe(&self) -> backend::ProbeFuture<'_> { + Box::pin(async move { + Err(backend::Error::Transport { + backend: backend::BackendId::new("openai"), + source: "connection refused".into(), + }) + }) + } + } + + /// Serve the app on an ephemeral port and return its base URL. + async fn serve(state: AppState) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let address = listener.local_addr().expect("address"); + tokio::spawn(async move { axum::serve(listener, router(state)).await.expect("serve") }); + format!("http://{address}") + } + + /// An app with a migrated database and both binaries somewhere findable. + /// The three classic upstreams point at a dead port; these tests read the + /// subtitle lamps, not theirs. + async fn application() -> (tempfile::TempDir, AppState) { + let dir = tempfile::tempdir().expect("tempdir"); + let database = Db::connect(dir.path().join("arr.db")) + .await + .expect("connect"); + database.migrate().await.expect("migrate"); + 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_syncer(Syncer::new().with_binary("sh")) + .with_ffmpeg_binary("sh"); + (dir, state) + } + + async fn report(state: AppState) -> serde_json::Value { + let base = serve(state).await; + reqwest::get(format!("{base}/api/health")) + .await + .expect("request health") + .json() + .await + .expect("health body") + } + + /// Replace the seeded enabled set. Runtime-checked rather than going + /// through the settings API, whose validation refuses engines this test + /// binary has not compiled in. + async fn set_enabled(state: &AppState, providers: &str, engine: Option<&str>) { + sqlx::query("UPDATE subtitle_settings SET providers_enabled = ?, translation_engine = ?") + .bind(providers) + .bind(engine) + .execute(state.database().expect("database").pool()) + .await + .expect("settings update"); + } + + #[tokio::test] + async fn without_a_settings_row_the_lane_is_quiet_and_binaries_still_checked() { + // No database attached: nothing is known to be in use, so nothing + // may fail a lamp — but a missing binary is a fact about the host. + let state = AppState::new(Upstreams::new( + "http://127.0.0.1:1".into(), + "http://127.0.0.1:1".into(), + )) + .expect("state") + .with_syncer(Syncer::new().with_binary("sh")) + .with_ffmpeg_binary("/nowhere/ffmpeg"); + + let body = report(state).await; + assert_eq!(body["subtitles"]["providers"], serde_json::json!([])); + assert!(body["subtitles"]["translation"].is_null()); + assert_eq!(body["subtitles"]["alass"]["status"], "ok"); + assert_eq!(body["subtitles"]["ffmpeg"]["status"], "unreachable"); + assert_eq!(body["status"], "degraded"); + } + + #[tokio::test] + async fn an_enabled_provider_without_credentials_is_unconfigured() { + let (_dir, state) = application().await; + // The seed row enables opensubtitles and podnapisi; none are attached. + let body = report(state).await; + assert_eq!( + body["subtitles"]["providers"], + serde_json::json!([ + { "id": "opensubtitles", "status": "unconfigured", + "detail": "opensubtitles is enabled but not configured — credentials are bootstrap config" }, + { "id": "podnapisi", "status": "unconfigured", + "detail": "podnapisi is enabled but not configured — credentials are bootstrap config" }, + ]) + ); + assert_eq!(body["status"], "degraded"); + } + + #[tokio::test] + async fn an_attached_provider_lamp_follows_its_probe() { + let (_dir, state) = application().await; + set_enabled(&state, r#"["ok","bad"]"#, None).await; + let state = state.with_subtitle_providers(vec![ + Arc::new(StubProvider { id: "ok", up: true }), + Arc::new(StubProvider { id: "bad", up: false }), + ]); + + let body = report(state).await; + assert_eq!(body["subtitles"]["providers"][0]["id"], "ok"); + assert_eq!(body["subtitles"]["providers"][0]["status"], "ok"); + assert_eq!(body["subtitles"]["providers"][1]["id"], "bad"); + assert_eq!(body["subtitles"]["providers"][1]["status"], "unreachable"); + assert_eq!(body["subtitles"]["providers"][1]["detail"], "credentials refused"); + } + + #[tokio::test] + async fn a_selected_engine_is_probed_only_when_selected() { + let (_dir, state) = application().await; + let body = report(state.clone()).await; + assert!(body["subtitles"]["translation"].is_null(), "{body}"); + + set_enabled(&state, "[]", Some("openai")).await; + // Selected but never attached: unavailable in this deployment. + let body = report(state).await; + assert_eq!(body["subtitles"]["translation"]["status"], "unconfigured"); + } + + #[tokio::test] + async fn an_unreachable_engine_fails_the_whole_report() { + let (_dir, state) = application().await; + set_enabled(&state, "[]", Some("openai")).await; + let state = state + .with_translation_backends(vec![Arc::new(StubBackend { + detail: "connection refused", + })]); + + let body = report(state).await; + assert_eq!(body["subtitles"]["translation"]["status"], "unreachable"); + assert_eq!(body["status"], "degraded"); + } +} diff --git a/crates/arr-api/src/lib.rs b/crates/arr-api/src/lib.rs index be82dd8..e1dd341 100644 --- a/crates/arr-api/src/lib.rs +++ b/crates/arr-api/src/lib.rs @@ -233,7 +233,11 @@ mod tests { .with_tmdb_url(tmdb.uri()) .with_tmdb_api_key(Some("key".into())), ) - .expect("state"); + .expect("state") + // The subtitle binaries default to `PATH`; pin them to something + // every machine has so this test stays about the classic upstreams. + .with_syncer(arr_subs::Syncer::new().with_binary("sh")) + .with_ffmpeg_binary("sh"); let body = report(state).await; assert_eq!(body["status"], "ok"); diff --git a/crates/arr-api/src/state.rs b/crates/arr-api/src/state.rs index cc5be15..beae631 100644 --- a/crates/arr-api/src/state.rs +++ b/crates/arr-api/src/state.rs @@ -1,6 +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::ffi::{OsStr, OsString}; use std::sync::{atomic::AtomicU64, Arc}; use std::time::Duration; @@ -14,6 +15,10 @@ use crate::jellyfin::JellyfinClient; /// is configurable, so this is a constant that tests point elsewhere. pub const DEFAULT_TMDB_URL: &str = "https://api.themoviedb.org/3"; +/// The `ffmpeg` invoked when nothing else is configured — same default as +/// `arr-probe`'s extractor, which is where it is actually run. +pub const DEFAULT_FFMPEG_BINARY: &str = "ffmpeg"; + /// How long an upstream has to answer a probe before it counts as /// unreachable. Health is polled by a human waiting on a page. const PROBE_TIMEOUT: Duration = Duration::from_secs(3); @@ -85,6 +90,8 @@ pub struct AppState { /// `None` unless the daemon compiled and configured that backend; the /// settings API writes it on every edit of the row. command_timeout: Option>, + /// The configured `ffmpeg` binary, for the health lamps (#200). + ffmpeg_binary: OsString, jellyfin: Option, syncer: Syncer, } @@ -160,6 +167,7 @@ impl AppState { subtitle_providers: Arc::new(Vec::new()), translation_backends: Arc::new(Vec::new()), command_timeout: None, + ffmpeg_binary: DEFAULT_FFMPEG_BINARY.into(), jellyfin: None, syncer: Syncer::default(), }) @@ -245,6 +253,19 @@ impl AppState { self } + /// Attach the configured `ffmpeg` binary, for the health lamps (#200). + /// Defaults to resolving `ffmpeg` from `PATH`. + #[must_use] + pub fn with_ffmpeg_binary(mut self, binary: impl Into) -> Self { + self.ffmpeg_binary = binary.into(); + self + } + + /// The configured `ffmpeg` binary. + pub(crate) fn ffmpeg_binary(&self) -> &OsStr { + &self.ffmpeg_binary + } + pub(crate) fn syncer(&self) -> &Syncer { &self.syncer } diff --git a/crates/arr-api/src/subtitle_settings.rs b/crates/arr-api/src/subtitle_settings.rs index 70409a6..e0b671f 100644 --- a/crates/arr-api/src/subtitle_settings.rs +++ b/crates/arr-api/src/subtitle_settings.rs @@ -153,7 +153,9 @@ impl SettingsColumns { } } -async fn load(state: &AppState) -> Result { +/// Read the row and parse it. Shared with the health lamps (#200), which +/// need the enabled set and the chosen engine but not the budgets. +pub(crate) async fn load(state: &AppState) -> Result { let row = sqlx::query_as!( SettingsColumns, r#"SELECT wanted_languages AS "wanted_languages!: String", diff --git a/crates/arr-api/src/subtitles.rs b/crates/arr-api/src/subtitles.rs index b7503af..a111059 100644 --- a/crates/arr-api/src/subtitles.rs +++ b/crates/arr-api/src/subtitles.rs @@ -1705,6 +1705,10 @@ mod tests { }) }) } + + fn probe(&self) -> arr_subs::ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } /// A provider that is configured but never answers. @@ -1732,6 +1736,14 @@ mod tests { }) }) } + + fn probe(&self) -> arr_subs::ProbeFuture<'_> { + Box::pin(async move { + Err(arr_subs::Error::Unauthorized { + provider: ProviderId::new("dead"), + }) + }) + } } /// Uppercases every cue. Enough to prove the pipeline, and it keeps cue @@ -1760,6 +1772,10 @@ mod tests { .collect()) }) } + + fn probe(&self) -> arr_subs::translate::ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } struct Fixture { From 8c6d4ca577e8e8e5073d33ee808d8e8cfec8abe2 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Tue, 25 Aug 2026 06:12:22 +0100 Subject: [PATCH 3/5] feat(daemon): fold subtitle lamps into broken notifications --- crates/arr-daemon/src/broken.rs | 207 +++++++++++++++++++++++++++-- crates/arr-daemon/src/main.rs | 58 +++++--- crates/arr-daemon/src/subtitles.rs | 18 ++- 3 files changed, 251 insertions(+), 32 deletions(-) diff --git a/crates/arr-daemon/src/broken.rs b/crates/arr-daemon/src/broken.rs index 015d86a..afafed0 100644 --- a/crates/arr-daemon/src/broken.rs +++ b/crates/arr-daemon/src/broken.rs @@ -1,5 +1,5 @@ //! §9.5 *broken* → the operator alone: Prowlarr, Transmission or TMDB -//! unreachable. +//! unreachable, or a subtitle lamp failing (#200). //! //! Edge-triggered: notifies once when an upstream stops answering, and //! silently re-arms once it answers again. There is no "fixed" notification — @@ -29,52 +29,117 @@ pub struct Upstreams { pub tmdb_api_key: Option, } +/// The subtitle upstreams (#200): the providers and engines this deployment +/// has credentials for, and the two binaries with their configured paths. +/// +/// What is *in use* is read from the settings row per tick; this only holds +/// what could ever answer. +#[derive(Debug, Clone)] +pub struct SubtitleUpstreams { + pub providers: Vec>, + pub backends: Vec>, + pub alass_path: std::path::PathBuf, + pub ffmpeg_path: std::path::PathBuf, +} + +impl SubtitleUpstreams { + /// The lamps as `(name, reachable)` pairs. Only what the settings row + /// has in use is probed — a provider nobody enabled cannot be broken. + async fn probe(&self, database: &Db) -> Vec<(String, bool)> { + let (providers_enabled, engine) = match crate::subtitles::load_settings(database).await { + Ok(settings) => (settings.providers_enabled, settings.translation_engine), + // An unreadable row says nothing about any upstream; skipping the + // whole lane beats notifying on our own database. + Err(error) => { + tracing::warn!(%error, "subtitle settings unreadable; subtitle lamps skipped"); + return Vec::new(); + } + }; + + let mut lamps = Vec::new(); + for id in &providers_enabled { + let reachable = match self.providers.iter().find(|p| p.id().as_str() == id.as_str()) { + Some(provider) => provider.probe().await.is_ok(), + None => false, + }; + lamps.push((id.clone(), reachable)); + } + if let Some(engine) = engine { + let reachable = match self.backends.iter().find(|b| b.id().as_str() == engine) { + Some(backend) => backend.probe().await.is_ok(), + None => false, + }; + lamps.push((engine, reachable)); + } + lamps.push(( + "alass".to_owned(), + arr_subs::binary_present(std::ffi::OsStr::new(&self.alass_path)), + )); + lamps.push(( + "ffmpeg".to_owned(), + arr_subs::binary_present(std::ffi::OsStr::new(&self.ffmpeg_path)), + )); + lamps + } +} + #[derive(Debug)] pub struct BrokenAction { http: Client, upstreams: Upstreams, + subtitles: SubtitleUpstreams, notifier: Notifier, operator_topic: String, /// Which upstreams are currently notified as broken. Transient — a /// restart re-probes and re-notifies whatever is still down. - broken: Arc>>, + broken: Arc>>, } impl BrokenAction { #[must_use] - pub fn new(upstreams: Upstreams, notifier: Notifier, operator_topic: String) -> Self { + pub fn new( + upstreams: Upstreams, + subtitles: SubtitleUpstreams, + notifier: Notifier, + operator_topic: String, + ) -> Self { Self { http: Client::new(), upstreams, + subtitles, notifier, operator_topic, broken: Arc::new(Mutex::new(HashSet::new())), } } - async fn tick(&self) -> Vec { - let (prowlarr, transmission, tmdb) = tokio::join!( + async fn tick(&self, database: &Db) -> Vec { + let (prowlarr, transmission, tmdb, subtitles) = tokio::join!( self.probe_prowlarr(), self.probe_transmission(), self.probe_tmdb(), + self.subtitles.probe(database), ); let mut outcomes = Vec::new(); outcomes.extend(self.notify_transition("prowlarr", prowlarr).await); outcomes.extend(self.notify_transition("transmission", transmission).await); outcomes.extend(self.notify_transition("tmdb", tmdb).await); + for (name, reachable) in subtitles { + outcomes.extend(self.notify_transition(&name, reachable).await); + } outcomes } /// `reachable` is `true` when the upstream answered, or when it needs no /// key and none is configured (not an outage — see `probe_tmdb`). - async fn notify_transition(&self, name: &'static str, reachable: bool) -> Option { + async fn notify_transition(&self, name: &str, reachable: bool) -> Option { let mut broken = self.broken.lock().await; if reachable { broken.remove(name); return None; } - if !broken.insert(name) { + if !broken.insert(name.to_owned()) { return None; } match self @@ -153,8 +218,8 @@ impl Action for BrokenAction { "broken" } - fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> { - Box::pin(async move { Ok(self.tick().await) }) + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { Ok(self.tick(database).await) }) } } @@ -176,6 +241,30 @@ mod tests { } } + /// A migrated database whose settings enable nothing — the classic + /// upstreams under test here must not share the tick with subtitle + /// lamps the seed row would otherwise put in use. + async fn database() -> (tempfile::TempDir, Db) { + let dir = tempfile::tempdir().unwrap(); + let db = Db::connect(dir.path().join("broken-test.db")).await.unwrap(); + db.migrate().await.unwrap(); + sqlx::query("UPDATE subtitle_settings SET providers_enabled = '[]'") + .execute(db.pool()) + .await + .unwrap(); + (dir, db) + } + + fn subtitles() -> SubtitleUpstreams { + SubtitleUpstreams { + providers: Vec::new(), + backends: Vec::new(), + // Present on every machine that runs these tests. + alass_path: "sh".into(), + ffmpeg_path: "sh".into(), + } + } + #[tokio::test] async fn an_unreachable_upstream_notifies_once_then_re_arms() { let prowlarr = MockServer::start().await; @@ -192,14 +281,16 @@ mod tests { .mount(&ntfy) .await; + let (_dir, db) = database().await; let action = BrokenAction::new( upstreams(prowlarr.uri(), transmission.uri()), + subtitles(), Notifier::new(ntfy.uri()).unwrap(), "operator-topic".to_string(), ); - let first = action.tick().await; - let second = action.tick().await; + let first = action.tick(&db).await; + let second = action.tick(&db).await; assert_eq!(first.len(), 1, "notifies on the tick it goes unreachable"); assert_eq!(second.len(), 0, "does not repeat while still broken"); @@ -209,12 +300,102 @@ mod tests { .respond_with(ResponseTemplate::new(200)) .mount(&prowlarr) .await; - let recovered = action.tick().await; + let recovered = action.tick(&db).await; assert_eq!(recovered.len(), 0, "recovery is silent, no fourth event"); // Take it down again: a fresh outage re-arms and notifies again. prowlarr.reset().await; - let broken_again = action.tick().await; + let broken_again = action.tick(&db).await; assert_eq!(broken_again.len(), 1, "re-arms after recovering"); } + + /// #200: an enabled provider nobody configured is a broken lamp like any + /// other, and it notifies once — then stays quiet while it stays broken. + #[tokio::test] + async fn an_enabled_but_missing_subtitle_provider_notifies_once() { + let ntfy = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&ntfy) + .await; + + let dir = tempfile::tempdir().unwrap(); + let db = Db::connect(dir.path().join("broken-subs.db")).await.unwrap(); + db.migrate().await.unwrap(); + // The seed row already enables opensubtitles and podnapisi; none are + // attached to this action. + let prowlarr = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/ping")) + .respond_with(ResponseTemplate::new(200)) + .mount(&prowlarr) + .await; + let transmission = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(409)) + .mount(&transmission) + .await; + + let action = BrokenAction::new( + upstreams(prowlarr.uri(), transmission.uri()), + subtitles(), + Notifier::new(ntfy.uri()).unwrap(), + "operator-topic".to_string(), + ); + + let first = action.tick(&db).await; + let second = action.tick(&db).await; + + assert_eq!(first.len(), 2, "one lamp per missing provider"); + assert_eq!(second.len(), 0, "does not repeat while still broken"); + + // Attaching nothing but disabling them silences the lamps. + sqlx::query("UPDATE subtitle_settings SET providers_enabled = '[]'") + .execute(db.pool()) + .await + .unwrap(); + let third = action.tick(&db).await; + assert_eq!(third.len(), 0, "a disabled provider cannot be broken"); + } + + /// #200: binaries are judged at their configured paths. + #[tokio::test] + async fn a_missing_binary_is_a_broken_lamp() { + let ntfy = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&ntfy) + .await; + + let (_dir, db) = database().await; + let prowlarr = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/ping")) + .respond_with(ResponseTemplate::new(200)) + .mount(&prowlarr) + .await; + let transmission = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(409)) + .mount(&transmission) + .await; + + let subs = SubtitleUpstreams { + alass_path: "/nowhere/alass".into(), + ..subtitles() + }; + let action = BrokenAction::new( + upstreams(prowlarr.uri(), transmission.uri()), + subs, + Notifier::new(ntfy.uri()).unwrap(), + "operator-topic".to_string(), + ); + + let first = action.tick(&db).await; + assert_eq!(first.len(), 1, "only the missing binary notifies"); + assert!( + first.iter().any(|outcome| format!("{outcome:?}").contains("alass")), + "{first:?}" + ); + } } diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 6a73f75..54ef2a4 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -303,8 +303,7 @@ fn reconcile_loop( tmdb: Option<&Arc>, notifier: &Notifier, translators: &Translators, -) -> Result<(ReconcileLoop, Option, Option), Error> { - let reconcile = ReconcileLoop::new(database.clone()); +) -> Result<(ReconcileLoop, Option, Option), Error> { let reconcile = ReconcileLoop::new(database.clone()); let seeding = SeedingRules::new( SeedingLimits { ratio: config.seed_ratio_limit, @@ -387,19 +386,12 @@ fn reconcile_loop( Tick::Reconcile, AttentionAction::new(notifier.clone(), operator_topic.clone()), ); - let broken_upstreams = broken::Upstreams { - prowlarr_url: config.prowlarr_url.clone(), - prowlarr_api_key: config.prowlarr_api_key.clone(), - transmission_url: config.transmission_url.clone(), - tmdb_url: config - .tmdb_url - .clone() - .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()), - tmdb_api_key: config.tmdb_api_key.clone(), - }; - reconcile = reconcile.register( - Tick::Reconcile, - BrokenAction::new(broken_upstreams, notifier.clone(), operator_topic.clone()), + reconcile = register_broken( + reconcile, + config, + translators, + notifier, + operator_topic.clone(), ); } else { tracing::warn!( @@ -411,6 +403,42 @@ fn reconcile_loop( Ok((reconcile, manual_grab, manual_tv)) } +/// The *broken* lane (#200 included): Prowlarr, Transmission and TMDB, plus +/// the subtitle lamps — an enabled provider, the selected engine, or a +/// missing `alass`/`ffmpeg` all fold into the same operator message. +fn register_broken( + reconcile: ReconcileLoop, + config: &Config, + translators: &Translators, + notifier: &Notifier, + operator_topic: String, +) -> ReconcileLoop { + let broken_upstreams = broken::Upstreams { + prowlarr_url: config.prowlarr_url.clone(), + prowlarr_api_key: config.prowlarr_api_key.clone(), + transmission_url: config.transmission_url.clone(), + tmdb_url: config + .tmdb_url + .clone() + .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()), + tmdb_api_key: config.tmdb_api_key.clone(), + }; + let broken_subtitles = broken::SubtitleUpstreams { + providers: subtitle_providers( + config.opensubtitles_api_key.clone(), + config.opensubtitles_username.clone(), + config.opensubtitles_password.clone(), + ), + backends: translators.backends.clone(), + alass_path: config.alass_path.clone(), + ffmpeg_path: config.ffmpeg_path.clone(), + }; + reconcile.register( + Tick::Reconcile, + BrokenAction::new(broken_upstreams, broken_subtitles, notifier.clone(), operator_topic), + ) +} + /// Register the TV grab lane on `reconcile` and hand back a second, /// independent instance for `manual::run` (issue #132). `None` when Prowlarr /// is not configured. TV grabbing needs no TMDB at grab time: air dates are diff --git a/crates/arr-daemon/src/subtitles.rs b/crates/arr-daemon/src/subtitles.rs index f014e27..630e0f6 100644 --- a/crates/arr-daemon/src/subtitles.rs +++ b/crates/arr-daemon/src/subtitles.rs @@ -57,10 +57,10 @@ pub enum SubtitleError { /// The runtime-editable half of §15's configuration, read once per tick. #[derive(Debug, Clone)] -struct Settings { +pub(crate) struct Settings { wanted: Vec, - providers_enabled: BTreeSet, - translation_engine: Option, + pub(crate) providers_enabled: BTreeSet, + pub(crate) translation_engine: Option, /// Provider id -> daily download allowance (§15, #197). A name absent /// from the map has no cap configured, which is unlimited (§10), not /// zero. @@ -80,7 +80,9 @@ impl Settings { } } -async fn load_settings(database: &Db) -> Result { +/// Shared with the broken-upstream action (#200), which needs the enabled +/// set and the chosen engine per tick. +pub(crate) async fn load_settings(database: &Db) -> Result { let row = sqlx::query!( r#"SELECT wanted_languages AS "wanted_languages!: String", providers_enabled AS "providers_enabled!: String", @@ -1273,6 +1275,10 @@ mod tests { }) }) } + + fn probe(&self) -> arr_subs::ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } #[derive(Debug)] @@ -1299,6 +1305,10 @@ mod tests { .collect()) }) } + + fn probe(&self) -> arr_subs::translate::ProbeFuture<'_> { + Box::pin(async move { Ok(()) }) + } } fn candidate(provider: &str, id: &str, language: Language) -> Candidate { From a836967e32bea9f5a3f16486fd1c08d4b51feb73 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Tue, 25 Aug 2026 06:24:32 +0100 Subject: [PATCH 4/5] feat(web): subtitle lane under the signal chain --- web/index.html | 2 ++ web/src/health.ts | 16 +++++++++++++++ web/src/main.ts | 52 ++++++++++++++++++++++++++++++++++++++++++++++- web/src/style.css | 11 ++++++++++ 4 files changed, 80 insertions(+), 1 deletion(-) diff --git a/web/index.html b/web/index.html index dda56e6..3e88a44 100644 --- a/web/index.html +++ b/web/index.html @@ -446,6 +446,8 @@

rpc · grab and seed

+ +