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 {