Merge #200: lamp the subtitle upstreams

Closes #200
This commit is contained in:
Miguel Palhas
2026-08-25 06:30:34 +01:00
24 changed files with 1243 additions and 53 deletions
+366 -6
View File
@@ -6,11 +6,15 @@
//! the body carries the verdict, so a degraded service can still explain //! the body carries the verdict, so a degraded service can still explain
//! itself to the UI instead of looking like a fourth outage. //! itself to the UI instead of looking like a fourth outage.
use std::ffi::OsStr;
use axum::extract::State; use axum::extract::State;
use axum::Json; use axum::Json;
use serde::Serialize; use serde::Serialize;
use utoipa::ToSchema; use utoipa::ToSchema;
use arr_subs::{binary_present, translate as backend_error};
use crate::state::AppState; use crate::state::AppState;
/// Whether the service as a whole can do its job. /// 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<String>,
}
/// 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<ProviderCheck>,
/// 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<Check>,
/// 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<Item = Status> + '_ {
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`. /// The body of `GET /api/health`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)]
pub struct HealthReport { pub struct HealthReport {
@@ -78,9 +120,11 @@ pub struct HealthReport {
pub prowlarr: Check, pub prowlarr: Check,
pub transmission: Check, pub transmission: Check,
pub tmdb: 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( #[utoipa::path(
get, get,
path = "/api/health", path = "/api/health",
@@ -90,17 +134,19 @@ pub struct HealthReport {
), ),
)] )]
pub async fn health(State(state): State<AppState>) -> Json<HealthReport> { pub async fn health(State(state): State<AppState>) -> Json<HealthReport> {
// Three independent network probes; serialising them would make the // Independent network probes; serialising them would make the endpoint
// endpoint as slow as the sum of the timeouts. // as slow as the sum of the timeouts.
let (prowlarr, transmission, tmdb) = tokio::join!( let (prowlarr, transmission, tmdb, subtitles) = tokio::join!(
probe_prowlarr(&state), probe_prowlarr(&state),
probe_transmission(&state), probe_transmission(&state),
probe_tmdb(&state), probe_tmdb(&state),
probe_subtitles(&state)
); );
let status = if [prowlarr.status, transmission.status, tmdb.status] let status = if [prowlarr.status, transmission.status, tmdb.status]
.iter() .into_iter()
.all(|s| *s == Status::Ok) .chain(subtitles.statuses())
.all(|check| check == Status::Ok)
{ {
Health::Ok Health::Ok
} else { } else {
@@ -113,6 +159,7 @@ pub async fn health(State(state): State<AppState>) -> Json<HealthReport> {
prowlarr, prowlarr,
transmission, transmission,
tmdb, tmdb,
subtitles,
}) })
} }
@@ -181,3 +228,316 @@ async fn probe_tmdb(state: &AppState) -> Check {
fn describe(err: reqwest::Error) -> String { fn describe(err: reqwest::Error) -> String {
err.without_url().to_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::translate as backend;
use arr_subs::{
Backend, CandidateId, DownloadFuture, Provider, ProviderId, SearchFuture, SearchRequest,
Syncer,
};
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");
}
}
+5 -1
View File
@@ -233,7 +233,11 @@ mod tests {
.with_tmdb_url(tmdb.uri()) .with_tmdb_url(tmdb.uri())
.with_tmdb_api_key(Some("key".into())), .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; let body = report(state).await;
assert_eq!(body["status"], "ok"); assert_eq!(body["status"], "ok");
+21
View File
@@ -1,6 +1,7 @@
//! What the API needs to answer a request: one HTTP client and the addresses //! 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). //! 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::sync::{atomic::AtomicU64, Arc};
use std::time::Duration; use std::time::Duration;
@@ -14,6 +15,10 @@ use crate::jellyfin::JellyfinClient;
/// is configurable, so this is a constant that tests point elsewhere. /// is configurable, so this is a constant that tests point elsewhere.
pub const DEFAULT_TMDB_URL: &str = "https://api.themoviedb.org/3"; 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 /// How long an upstream has to answer a probe before it counts as
/// unreachable. Health is polled by a human waiting on a page. /// unreachable. Health is polled by a human waiting on a page.
const PROBE_TIMEOUT: Duration = Duration::from_secs(3); 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 /// `None` unless the daemon compiled and configured that backend; the
/// settings API writes it on every edit of the row. /// settings API writes it on every edit of the row.
command_timeout: Option<Arc<AtomicU64>>, command_timeout: Option<Arc<AtomicU64>>,
/// The configured `ffmpeg` binary, for the health lamps (#200).
ffmpeg_binary: OsString,
jellyfin: Option<JellyfinClient>, jellyfin: Option<JellyfinClient>,
syncer: Syncer, syncer: Syncer,
} }
@@ -160,6 +167,7 @@ impl AppState {
subtitle_providers: Arc::new(Vec::new()), subtitle_providers: Arc::new(Vec::new()),
translation_backends: Arc::new(Vec::new()), translation_backends: Arc::new(Vec::new()),
command_timeout: None, command_timeout: None,
ffmpeg_binary: DEFAULT_FFMPEG_BINARY.into(),
jellyfin: None, jellyfin: None,
syncer: Syncer::default(), syncer: Syncer::default(),
}) })
@@ -245,6 +253,19 @@ impl AppState {
self 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<OsString>) -> 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 { pub(crate) fn syncer(&self) -> &Syncer {
&self.syncer &self.syncer
} }
+3 -1
View File
@@ -153,7 +153,9 @@ impl SettingsColumns {
} }
} }
async fn load(state: &AppState) -> Result<SubtitleSettings, ApiError> { /// 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<SubtitleSettings, ApiError> {
let row = sqlx::query_as!( let row = sqlx::query_as!(
SettingsColumns, SettingsColumns,
r#"SELECT wanted_languages AS "wanted_languages!: String", r#"SELECT wanted_languages AS "wanted_languages!: String",
+16
View File
@@ -1714,6 +1714,10 @@ mod tests {
}) })
}) })
} }
fn probe(&self) -> arr_subs::ProbeFuture<'_> {
Box::pin(async move { Ok(()) })
}
} }
/// A provider that is configured but never answers. /// A provider that is configured but never answers.
@@ -1741,6 +1745,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 /// Uppercases every cue. Enough to prove the pipeline, and it keeps cue
@@ -1769,6 +1781,10 @@ mod tests {
.collect()) .collect())
}) })
} }
fn probe(&self) -> arr_subs::translate::ProbeFuture<'_> {
Box::pin(async move { Ok(()) })
}
} }
struct Fixture { struct Fixture {
+204 -13
View File
@@ -1,5 +1,5 @@
//! §9.5 *broken* → the operator alone: Prowlarr, Transmission or TMDB //! §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 //! Edge-triggered: notifies once when an upstream stops answering, and
//! silently re-arms once it answers again. There is no "fixed" notification — //! silently re-arms once it answers again. There is no "fixed" notification —
@@ -29,52 +29,121 @@ pub struct Upstreams {
pub tmdb_api_key: Option<String>, 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)] #[derive(Debug)]
pub struct BrokenAction { pub struct BrokenAction {
http: Client, http: Client,
upstreams: Upstreams, upstreams: Upstreams,
subtitles: SubtitleUpstreams,
notifier: Notifier, notifier: Notifier,
operator_topic: String, operator_topic: String,
/// Which upstreams are currently notified as broken. Transient — a /// Which upstreams are currently notified as broken. Transient — a
/// restart re-probes and re-notifies whatever is still down. /// restart re-probes and re-notifies whatever is still down.
broken: Arc<Mutex<HashSet<&'static str>>>, broken: Arc<Mutex<HashSet<String>>>,
} }
impl BrokenAction { impl BrokenAction {
#[must_use] #[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 { Self {
http: Client::new(), http: Client::new(),
upstreams, upstreams,
subtitles,
notifier, notifier,
operator_topic, operator_topic,
broken: Arc::new(Mutex::new(HashSet::new())), 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_prowlarr(),
self.probe_transmission(), self.probe_transmission(),
self.probe_tmdb(), self.probe_tmdb(),
self.subtitles.probe(database),
); );
let mut outcomes = Vec::new(); let mut outcomes = Vec::new();
outcomes.extend(self.notify_transition("prowlarr", prowlarr).await); outcomes.extend(self.notify_transition("prowlarr", prowlarr).await);
outcomes.extend(self.notify_transition("transmission", transmission).await); outcomes.extend(self.notify_transition("transmission", transmission).await);
outcomes.extend(self.notify_transition("tmdb", tmdb).await); outcomes.extend(self.notify_transition("tmdb", tmdb).await);
for (name, reachable) in subtitles {
outcomes.extend(self.notify_transition(&name, reachable).await);
}
outcomes outcomes
} }
/// `reachable` is `true` when the upstream answered, or when it needs no /// `reachable` is `true` when the upstream answered, or when it needs no
/// key and none is configured (not an outage — see `probe_tmdb`). /// 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; let mut broken = self.broken.lock().await;
if reachable { if reachable {
broken.remove(name); broken.remove(name);
return None; return None;
} }
if !broken.insert(name) { if !broken.insert(name.to_owned()) {
return None; return None;
} }
match self match self
@@ -153,8 +222,8 @@ impl Action for BrokenAction {
"broken" "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 +245,32 @@ 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] #[tokio::test]
async fn an_unreachable_upstream_notifies_once_then_re_arms() { async fn an_unreachable_upstream_notifies_once_then_re_arms() {
let prowlarr = MockServer::start().await; let prowlarr = MockServer::start().await;
@@ -192,14 +287,16 @@ mod tests {
.mount(&ntfy) .mount(&ntfy)
.await; .await;
let (_dir, db) = database().await;
let action = BrokenAction::new( let action = BrokenAction::new(
upstreams(prowlarr.uri(), transmission.uri()), upstreams(prowlarr.uri(), transmission.uri()),
subtitles(),
Notifier::new(ntfy.uri()).unwrap(), Notifier::new(ntfy.uri()).unwrap(),
"operator-topic".to_string(), "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!(first.len(), 1, "notifies on the tick it goes unreachable");
assert_eq!(second.len(), 0, "does not repeat while still broken"); assert_eq!(second.len(), 0, "does not repeat while still broken");
@@ -209,12 +306,106 @@ mod tests {
.respond_with(ResponseTemplate::new(200)) .respond_with(ResponseTemplate::new(200))
.mount(&prowlarr) .mount(&prowlarr)
.await; .await;
let recovered = action.tick().await; let recovered = action.tick(&db).await;
assert_eq!(recovered.len(), 0, "recovery is silent, no fourth event"); assert_eq!(recovered.len(), 0, "recovery is silent, no fourth event");
// Take it down again: a fresh outage re-arms and notifies again. // Take it down again: a fresh outage re-arms and notifies again.
prowlarr.reset().await; 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"); 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:?}"
);
}
} }
+45 -11
View File
@@ -388,6 +388,33 @@ fn reconcile_loop(
Tick::Reconcile, Tick::Reconcile,
AttentionAction::new(notifier.clone(), operator_topic.clone()), 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 { let broken_upstreams = broken::Upstreams {
prowlarr_url: config.prowlarr_url.clone(), prowlarr_url: config.prowlarr_url.clone(),
prowlarr_api_key: config.prowlarr_api_key.clone(), prowlarr_api_key: config.prowlarr_api_key.clone(),
@@ -398,18 +425,25 @@ fn reconcile_loop(
.unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()), .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()),
tmdb_api_key: config.tmdb_api_key.clone(), 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, Tick::Reconcile,
BrokenAction::new(broken_upstreams, notifier.clone(), operator_topic.clone()), BrokenAction::new(
); broken_upstreams,
} else { broken_subtitles,
tracing::warn!( notifier.clone(),
"ARR_NTFY_OPERATOR_TOPIC is not configured: needs-a-decision and broken notifications are disabled" operator_topic,
); ),
} )
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, /// Register the TV grab lane on `reconcile` and hand back a second,
+14 -4
View File
@@ -57,10 +57,10 @@ pub enum SubtitleError {
/// The runtime-editable half of §15's configuration, read once per tick. /// The runtime-editable half of §15's configuration, read once per tick.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
struct Settings { pub(crate) struct Settings {
wanted: Vec<String>, 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 /// Provider id -> daily download allowance (§15, #197). A name absent
/// from the map has no cap configured, which is unlimited (§10), not /// from the map has no cap configured, which is unlimited (§10), not
/// zero. /// 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!( let row = sqlx::query!(
r#"SELECT wanted_languages AS "wanted_languages!: String", r#"SELECT wanted_languages AS "wanted_languages!: String",
providers_enabled AS "providers_enabled!: 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)] #[derive(Debug)]
@@ -1299,6 +1305,10 @@ mod tests {
.collect()) .collect())
}) })
} }
fn probe(&self) -> arr_subs::translate::ProbeFuture<'_> {
Box::pin(async move { Ok(()) })
}
} }
fn candidate(provider: &str, id: &str, language: Language) -> Candidate { fn candidate(provider: &str, id: &str, language: Language) -> Candidate {
+34 -2
View File
@@ -35,8 +35,8 @@ use tokio::{
use arr_core::Language; use arr_core::Language;
use crate::translate::{ use crate::translate::{
strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture, strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, ProbeFuture, Result,
TranslatedCue, TranslateFuture, TranslatedCue,
}; };
const BACKEND_NAME: &str = "command"; const BACKEND_NAME: &str = "command";
@@ -210,6 +210,38 @@ impl Backend for Command {
fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> { fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> {
Box::pin(async move { self.translate_inner(batch).await }) 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. /// Spawn the command once, feed it `payload`, and collect its stdout.
+55 -1
View File
@@ -24,7 +24,9 @@ use serde::{Deserialize, Serialize};
use arr_core::Language; 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}; use crate::{Error as SubsError, ProviderId, Result};
/// The `DeepL` API endpoint for keys ending in `:fx` (the free tier). /// The `DeepL` API endpoint for keys ending in `:fx` (the free tier).
@@ -282,6 +284,31 @@ impl Backend for DeepL {
.collect()) .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<Duration> { fn retry_after(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
@@ -534,4 +561,31 @@ mod tests {
}; };
assert_eq!(retry_after, Some(Duration::from_secs(30))); 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:?}");
}
} }
+64 -2
View File
@@ -26,7 +26,9 @@ use serde::{Deserialize, Serialize};
use arr_core::Language; 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}; use crate::{Error as SubsError, ProviderId, Result};
/// The public Google Cloud Translation v2 endpoint. /// The public Google Cloud Translation v2 endpoint.
@@ -291,6 +293,39 @@ impl Backend for Google {
.collect()) .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 { fn api_error_detail(status: StatusCode, body: &[u8]) -> String {
@@ -367,7 +402,7 @@ fn numeric_entity(entity: &str) -> Option<char> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use arr_core::Language; use arr_core::Language;
use wiremock::matchers::method; use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate}; use wiremock::{Mock, MockServer, ResponseTemplate};
use super::{Google, GoogleConfig}; use super::{Google, GoogleConfig};
@@ -584,4 +619,31 @@ mod tests {
// A stray ampersand that is not an entity stays one. // A stray ampersand that is not an entity stays one.
assert_eq!(unescape_html("fish & chips"), "fish & chips"); 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:?}");
}
} }
+17 -1
View File
@@ -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 podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL};
pub use srt::Cue; pub use srt::Cue;
pub use sync::{ 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}; pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue};
@@ -65,6 +66,9 @@ pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, Transl
#[cfg(test)] #[cfg(test)]
use wiremock as _; use wiremock as _;
/// One health probe (#200): reachability and, where they exist, credentials.
pub type ProbeFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
/// The candidates one search turned up. /// The candidates one search turned up.
pub type SearchFuture<'a> = Pin<Box<dyn Future<Output = Result<Vec<Candidate>>> + Send + 'a>>; pub type SearchFuture<'a> = Pin<Box<dyn Future<Output = Result<Vec<Candidate>>> + Send + 'a>>;
@@ -131,6 +135,14 @@ pub trait Provider: fmt::Debug + Send + Sync {
/// Anything in [`Error`], and [`Error::NotFound`] when the candidate has /// Anything in [`Error`], and [`Error::NotFound`] when the candidate has
/// gone away between the search and the download. /// gone away between the search and the download.
fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a>; 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)] #[cfg(test)]
@@ -190,6 +202,10 @@ mod tests {
}) })
}) })
} }
fn probe(&self) -> super::ProbeFuture<'_> {
Box::pin(async move { Ok(()) })
}
} }
fn request() -> SearchRequest { fn request() -> SearchRequest {
+78 -2
View File
@@ -26,8 +26,8 @@ use serde::{Deserialize, Serialize};
use arr_core::Language; use arr_core::Language;
use crate::translate::{ use crate::translate::{
strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture, strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, ProbeFuture, Result,
TranslatedCue, TranslateFuture, TranslatedCue,
}; };
const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1/"; 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> { fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> {
Box::pin(async move { self.translate_inner(batch).await }) 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<Duration> { fn retry_after(response: &reqwest::Response) -> Option<Duration> {
@@ -382,4 +412,50 @@ mod tests {
assert_eq!(strip_code_fence("```\n[1,2]\n```"), "[1,2]"); assert_eq!(strip_code_fence("```\n[1,2]\n```"), "[1,2]");
assert_eq!(strip_code_fence("[1,2]"), "[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:?}"
);
}
} }
+51 -3
View File
@@ -222,6 +222,23 @@ impl OpenSubtitles {
query: Option<&[(String, String)]>, query: Option<&[(String, String)]>,
json: Option<&serde_json::Value>, json: Option<&serde_json::Value>,
bearer: Option<&str>, bearer: Option<&str>,
) -> Result<reqwest::Response> {
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<reqwest::Response> { ) -> Result<reqwest::Response> {
let url = self.base_url.join(path).map_err(|err| Error::Malformed { let url = self.base_url.join(path).map_err(|err| Error::Malformed {
provider: self.id.clone(), provider: self.id.clone(),
@@ -244,11 +261,10 @@ impl OpenSubtitles {
} }
tracing::debug!(provider = %self.id, path, "OpenSubtitles request"); 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(), provider: self.id.clone(),
source: Box::new(err), source: Box::new(err),
})?; })
self.check_status(response).await
} }
/// Fetch (or refetch) the user token downloads travel under. /// Fetch (or refetch) the user token downloads travel under.
@@ -292,6 +308,34 @@ impl OpenSubtitles {
self.login().await 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<Vec<Candidate>> { async fn search_inner(&self, request: &SearchRequest) -> Result<Vec<Candidate>> {
let hash = let hash =
moviehash(&request.file.path, request.file.size).map_err(|source| Error::Io { 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> { fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> {
Box::pin(async move { self.download_inner(id).await }) 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<Duration> { fn retry_after(response: &reqwest::Response) -> Option<Duration> {
+6
View File
@@ -255,6 +255,12 @@ impl Provider for Podnapisi {
fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> { fn download<'a>(&'a self, id: &'a CandidateId) -> DownloadFuture<'a> {
Box::pin(async move { self.download_candidate(id).await }) 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`]. /// Configuration for a [`Podnapisi`].
+70 -1
View File
@@ -14,7 +14,7 @@
//! reference is always the media file itself; subtitle and video are both on //! reference is always the media file itself; subtitle and video are both on
//! disk by the time this runs. //! 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; 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. /// How much of `alass`'s stderr is kept in an error.
const STDERR_LIMIT: usize = 512; 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. /// Runs `alass` over one subtitle against its video.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Syncer { pub struct Syncer {
@@ -74,6 +106,18 @@ impl Syncer {
self 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. /// Sync `subtitle` against `video`, in place on disk.
/// ///
/// Returns the synced SRT text on acceptance; the input file is never /// Returns the synced SRT text on acceptance; the input file is never
@@ -639,4 +683,29 @@ mod tests {
"1 of 2 cues survived" "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());
}
} }
+19 -1
View File
@@ -105,6 +105,9 @@ pub struct TranslatedCue {
pub type TranslateFuture<'a> = pub type TranslateFuture<'a> =
Pin<Box<dyn Future<Output = Result<Vec<TranslatedCue>>> + Send + 'a>>; Pin<Box<dyn Future<Output = Result<Vec<TranslatedCue>>> + Send + 'a>>;
/// One health probe (#200): reachable and credentials accepted.
pub type ProbeFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
/// One translation backend (DESIGN.md §15). /// One translation backend (DESIGN.md §15).
/// ///
/// Boxed futures rather than `async fn` for the same reason as /// 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`]. /// Anything in [`Error`].
fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a>; 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. /// Result alias for translation.
@@ -424,7 +436,9 @@ mod tests {
use arr_core::Language; 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; use crate::srt::Cue;
/// Uppercases every cue and records how it was called; misbehaves on cue. /// Uppercases every cue and records how it was called; misbehaves on cue.
@@ -466,6 +480,10 @@ mod tests {
Ok(reply) Ok(reply)
}) })
} }
fn probe(&self) -> ProbeFuture<'_> {
Box::pin(async move { Ok(()) })
}
} }
fn cues(texts: &[&str]) -> Vec<Cue> { fn cues(texts: &[&str]) -> Vec<Cue> {
+23 -1
View File
@@ -21,7 +21,7 @@ use std::{
use arr_core::Language; use arr_core::Language;
use arr_subs::translate::{translate, Error}; use arr_subs::translate::{translate, Error};
use arr_subs::{Command, CommandConfig}; use arr_subs::{Backend as _, Command, CommandConfig};
fn cues(texts: &[&str]) -> Vec<arr_subs::Cue> { fn cues(texts: &[&str]) -> Vec<arr_subs::Cue> {
texts texts
@@ -270,3 +270,25 @@ async fn an_edited_timeout_reaches_the_next_call() {
} }
const DEFAULT: Duration = arr_subs::COMMAND_DEFAULT_TIMEOUT; 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:?}");
}
+39
View File
@@ -424,3 +424,42 @@ async fn an_id_this_provider_never_offered_is_not_found() {
other => panic!("expected NotFound, got {other:?}"), 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 { .. }));
}
+23
View File
@@ -282,3 +282,26 @@ async fn an_id_this_provider_never_issued_is_not_found() {
assert!(matches!(error, Error::NotFound { .. }), "got {error:?}"); 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:?}");
}
+2
View File
@@ -446,6 +446,8 @@
<p class="module-detail readout" data-role="detail">rpc · grab and seed</p> <p class="module-detail readout" data-role="detail">rpc · grab and seed</p>
</article> </article>
</section> </section>
<section class="chain-lane" id="chain-subs" aria-label="subtitle lane" hidden></section>
</section> </section>
<section class="deck-group" id="group-roots" hidden aria-labelledby="label-roots"> <section class="deck-group" id="group-roots" hidden aria-labelledby="label-roots">
+16
View File
@@ -9,12 +9,28 @@ export interface Check {
detail?: string; detail?: string;
} }
/** One enabled subtitle provider (#200). */
export interface ProviderCheck {
id: string;
status: CheckStatus;
detail?: string;
}
/** The subtitle lane: providers in use, the selected engine, the binaries. */
export interface SubtitleHealth {
providers: ProviderCheck[];
translation?: Check | null;
alass: Check;
ffmpeg: Check;
}
export interface HealthReport { export interface HealthReport {
status: "ok" | "degraded"; status: "ok" | "degraded";
version: string; version: string;
prowlarr: Check; prowlarr: Check;
transmission: Check; transmission: Check;
tmdb: Check; tmdb: Check;
subtitles: SubtitleHealth;
} }
export type Probe = export type Probe =
+59 -1
View File
@@ -1,4 +1,10 @@
import { type CheckStatus, type Probe, probeHealth } from "./health"; import {
type Check,
type CheckStatus,
type Probe,
probeHealth,
type SubtitleHealth,
} from "./health";
import { import {
fetchLibrary, fetchLibrary,
type LibrarySeries, type LibrarySeries,
@@ -171,6 +177,54 @@ function setModule(refs: ModuleRefs, state: string, tone: Tone, word: string, de
refs.detail.textContent = detail ?? refs.defaultDetail; refs.detail.textContent = detail ?? refs.defaultDetail;
} }
/** One lamp of the subtitle lane (#200): same module shape as the chain,
* rendered per poll which providers are in use is a settings row, not
* markup. */
function subtitleModule(name: string, role: string, check: Check): HTMLElement {
const article = document.createElement("article");
article.className = "module";
const head = document.createElement("header");
head.className = "module-head";
const lamp = document.createElement("span");
lamp.className = "lamp";
lamp.dataset.state = check.status;
const title = document.createElement("h2");
title.className = "module-name";
title.textContent = name;
const status = document.createElement("span");
status.className = "module-status readout";
status.dataset.tone = TONE_BY_STATUS[check.status];
status.textContent = check.status;
head.append(lamp, title, status);
const roleLine = document.createElement("p");
roleLine.className = "module-role";
roleLine.textContent = role;
const detail = document.createElement("p");
detail.className = "module-detail readout";
detail.textContent = check.detail ?? "";
article.append(head, roleLine, detail);
return article;
}
function renderSubtitles(subLane: HTMLElement, subs: SubtitleHealth) {
const entries: { name: string; role: string; check: Check }[] = subs.providers.map(
(provider) => ({ name: provider.id, role: "subtitles in", check: provider }),
);
if (subs.translation) {
entries.push({ name: "translate", role: "engine out", check: subs.translation });
}
entries.push({ name: "alass", role: "timing sync", check: subs.alass });
entries.push({ name: "ffmpeg", role: "track extraction", check: subs.ffmpeg });
subLane.replaceChildren(
...entries.map((entry) => subtitleModule(entry.name, entry.role, entry.check)),
);
subLane.hidden = false;
}
function must<T extends Element>(selector: string): T { function must<T extends Element>(selector: string): T {
const element = document.querySelector<T>(selector); const element = document.querySelector<T>(selector);
if (!element) { if (!element) {
@@ -181,6 +235,7 @@ function must<T extends Element>(selector: string): T {
function main() { function main() {
const chain = must<HTMLElement>(".chain"); const chain = must<HTMLElement>(".chain");
const subLane = must<HTMLElement>("#chain-subs");
const masterLamp = must<HTMLElement>("#master-lamp"); const masterLamp = must<HTMLElement>("#master-lamp");
const version = must<HTMLElement>("#version"); const version = must<HTMLElement>("#version");
@@ -235,6 +290,8 @@ function main() {
for (const name of ["tmdb", "prowlarr", "master"] as const) { for (const name of ["tmdb", "prowlarr", "master"] as const) {
setTrace(name, null); setTrace(name, null);
} }
subLane.hidden = true;
subLane.replaceChildren();
return; return;
} }
@@ -259,6 +316,7 @@ function main() {
setTrace(name, TONE_BY_STATUS[check.status]); setTrace(name, TONE_BY_STATUS[check.status]);
} }
} }
renderSubtitles(subLane, report.subtitles);
} }
let inFlight = false; let inFlight = false;
+11
View File
@@ -315,6 +315,17 @@ body {
/* ---- modules --------------------------------------------------------- */ /* ---- modules --------------------------------------------------------- */
/* The subtitle lane (#200): one lamp per upstream actually in use, rendered
by the health poll under the main chain. Same module vocabulary, no bus
these feed the reconcile loop, not each other. */
.chain-lane {
margin-top: var(--space-6);
display: grid;
grid-template-columns: repeat(auto-fit, minmax(12rem, 1fr));
gap: var(--space-4);
align-items: stretch;
}
.module { .module {
background: var(--panel); background: var(--panel);
border: 1px solid var(--line); border: 1px solid var(--line);