feat(daemon): fold subtitle lamps into broken notifications

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<String>,
 }
 
+/// 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<Arc<dyn arr_subs::Provider>>,
+    pub backends: Vec<Arc<dyn arr_subs::Backend>>,
+    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<Mutex<HashSet<&'static str>>>,
+    broken: Arc<Mutex<HashSet<String>>>,
 }
 
 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<Outcome> {
+    async fn tick(&self, database: &Db) -> Vec<Outcome> {
-        let (prowlarr, transmission, tmdb) = tokio::join!(
+        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<Outcome> {
+    async fn notify_transition(&self, name: &str, reachable: bool) -> Option<Outcome> {
         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> {
+    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
-        Box::pin(async move { Ok(self.tick().await) })
+        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 first = action.tick(&db).await;
-        let second = action.tick().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:?}"
+        );
+    }
 }

crates/arr-daemon/src/main.rs

@@ -303,8 +303,7 @@ fn reconcile_loop(
     tmdb: Option<&Arc<TmdbClient>>,
     notifier: &Notifier,
     translators: &Translators,
-) -> Result<(ReconcileLoop, Option<GrabAction>, Option<TvGrabAction>), Error> {
+) -> Result<(ReconcileLoop, Option<GrabAction>, Option<TvGrabAction>), Error> {    let reconcile = ReconcileLoop::new(database.clone());
-    let reconcile = ReconcileLoop::new(database.clone());
     let seeding = SeedingRules::new(
         SeedingLimits {
             ratio: config.seed_ratio_limit,
@@ -387,6 +386,33 @@ fn reconcile_loop(
             Tick::Reconcile,
             AttentionAction::new(notifier.clone(), operator_topic.clone()),
         );
+        reconcile = register_broken(
+            reconcile,
+            config,
+            translators,
+            notifier,
+            operator_topic.clone(),
+        );
+    } else {
+        tracing::warn!(
+            "ARR_NTFY_OPERATOR_TOPIC is not configured: needs-a-decision and broken notifications are disabled"
+        );
+    }
+
+    let reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone()));
+    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(),
@@ -397,18 +423,20 @@ fn reconcile_loop(
             .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()),
         tmdb_api_key: config.tmdb_api_key.clone(),
     };
-        reconcile = reconcile.register(
+    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, notifier.clone(), operator_topic.clone()),
+        BrokenAction::new(broken_upstreams, broken_subtitles, notifier.clone(), operator_topic),
-        );
+    )
-    } else {
-        tracing::warn!(
-            "ARR_NTFY_OPERATOR_TOPIC is not configured: needs-a-decision and broken notifications are disabled"
-        );
-    }
-
-    let reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone()));
-    Ok((reconcile, manual_grab, manual_tv))
 }
 
 /// Register the TV grab lane on `reconcile` and hand back a second,

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<String>,
-    providers_enabled: BTreeSet<String>,
+    pub(crate) providers_enabled: BTreeSet<String>,
-    translation_engine: Option<String>,
+    pub(crate) translation_engine: Option<String>,
     /// 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<Settings, SubtitleError> {
+/// 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<Settings, SubtitleError> {
     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 {