Merge #194: sync subtitles with alass

Closes #194

crates/arr-subs/Cargo.toml

@@ -25,12 +25,13 @@ encoding_rs.workspace = true
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+tempfile.workspace = true
 thiserror.workspace = true
+tokio.workspace = true
 tracing.workspace = true
 zip.workspace = true
 
 [dev-dependencies]
-tempfile.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 wiremock.workspace = true
 

crates/arr-subs/src/error.rs

@@ -111,6 +111,53 @@ pub enum Error {
         /// What went wrong, short enough to log.
         detail: String,
     },
+
+    /// `alass` could not be started at all — the binary is not installed
+    /// where the configuration points (#194). A health condition (#200),
+    /// folded into the operator message §9.5 defines; never a panic.
+    #[error("could not run {binary}")]
+    Spawn {
+        /// The binary that was attempted.
+        binary: String,
+        /// The underlying IO failure.
+        #[source]
+        source: std::io::Error,
+    },
+
+    /// One `alass` run took longer than its timeout.
+    #[error("syncing {path} timed out")]
+    Timeout {
+        /// The subtitle being synced.
+        path: PathBuf,
+    },
+
+    /// `alass` ran and exited non-zero.
+    ///
+    /// Its own variant rather than [`Error::Io`]: the process answered, so
+    /// this is about the sync, and `status` plus a trimmed stderr is what an
+    /// operator needs to see.
+    #[error("syncing {path} failed (status {status:?}): {stderr}")]
+    SyncFailed {
+        /// The subtitle being synced.
+        path: PathBuf,
+        /// The exit code, when it exited rather than died on a signal.
+        status: Option<i32>,
+        /// The first part of what it wrote to stderr.
+        stderr: String,
+    },
+
+    /// A file that should be SRT — the subtitle going into `alass`, or what
+    /// came out of it — does not parse as SRT.
+    ///
+    /// An implausible output is *not* here: that is [`crate::sync::Outcome::Rejected`],
+    /// a decision, while a file nothing can parse is a fact worth failing on.
+    #[error("{path} is not SRT: {detail}")]
+    NotSrt {
+        /// The file involved.
+        path: PathBuf,
+        /// What was wrong with it, short enough to log.
+        detail: String,
+    },
 }
 
 impl Error {
@@ -120,7 +167,10 @@ impl Error {
     /// response do not fix themselves, and retrying them only burns budget.
     #[must_use]
     pub const fn is_transient(&self) -> bool {
-        matches!(self, Self::Transport { .. } | Self::RateLimited { .. })
+        matches!(
+            self,
+            Self::Transport { .. } | Self::RateLimited { .. } | Self::Timeout { .. }
+        )
     }
 }
 
@@ -139,4 +189,24 @@ mod tests {
         .is_transient());
         assert!(!Error::Unauthorized { provider }.is_transient());
     }
+
+    #[test]
+    fn a_timeout_is_worth_retrying_and_a_missing_binary_is_not() {
+        let timeout = Error::Timeout {
+            path: std::path::PathBuf::from("/m/a.srt"),
+        };
+        let spawn = Error::Spawn {
+            binary: "alass".to_owned(),
+            source: std::io::Error::new(std::io::ErrorKind::NotFound, "no such file"),
+        };
+        let not_srt = Error::NotSrt {
+            path: std::path::PathBuf::from("/m/a.srt"),
+            detail: "line 1: expected a timing line".to_owned(),
+        };
+        assert!(timeout.is_transient());
+        // A missing binary needs installing, and a broken subtitle does not
+        // heal; neither resolves by retrying.
+        assert!(!spawn.is_transient());
+        assert!(!not_srt.is_transient());
+    }
 }

crates/arr-subs/src/lib.rs

@@ -33,6 +33,7 @@ pub mod model;
 pub mod opensubtitles;
 pub mod podnapisi;
 pub mod srt;
+pub mod sync;
 pub mod translate;
 
 pub use error::{Error, Result};
@@ -42,6 +43,7 @@ pub use model::{
 pub use opensubtitles::{moviehash, OpenSubtitles, OpenSubtitlesConfig};
 pub use podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL};
 pub use srt::Cue;
+pub use sync::{Outcome, Rejection, Syncer, DEFAULT_BINARY as ALASS_DEFAULT_BINARY};
 pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue};
 
 // The OpenSubtitles provider's HTTP paths are exercised by the integration

crates/arr-subs/src/sync.rs

@@ -0,0 +1,519 @@
+//! The `alass` wrapper — timing sync for every fetched and translated
+//! subtitle, DESIGN.md §15.
+//!
+//! The same shape as `arr-probe`'s [`crate::Prober`]: one small binary,
+//! invoked and discarded, with a configurable path and a timeout. It adds
+//! nothing at rest.
+//!
+//! `alass` reports no confidence value, so trust is bounded structurally: the
+//! output is parsed back and compared against the input it was made from. A
+//! cue that moved more than [`MAX_SHIFT`] from where it started, or cues that
+//! went missing between input and output, make the whole result implausible —
+//! and an implausible result is not an error but a [`Outcome::Rejected`], so
+//! the caller keeps the unsynced original and flags the row (#186). The
+//! reference is always the media file itself; subtitle and video are both on
+//! disk by the time this runs.
+
+use std::{ffi::OsString, path::Path, process::Stdio, time::Duration};
+
+use tokio::process::Command;
+
+use crate::{error::Error, srt, Result};
+
+/// The binary invoked when nothing else is configured.
+pub const DEFAULT_BINARY: &str = "alass";
+
+/// How long one sync may take. Generous: `alass` correlates audio against the
+/// whole file, and a feature-length remux over a network mount is slow.
+pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(180);
+
+/// How far a cue may sit from where it started and still be plausible.
+///
+/// §15's number. Wide enough for a genuine PAL-speedup or a different cut of
+/// the same film; narrow enough that a sync against the wrong file cannot
+/// sneak through.
+pub const MAX_SHIFT: Duration = Duration::from_secs(60);
+
+/// How much of `alass`'s stderr is kept in an error.
+const STDERR_LIMIT: usize = 512;
+
+/// Runs `alass` over one subtitle against its video.
+#[derive(Clone, Debug)]
+pub struct Syncer {
+    binary: OsString,
+    timeout: Duration,
+}
+
+impl Default for Syncer {
+    fn default() -> Self {
+        Self {
+            binary: DEFAULT_BINARY.into(),
+            timeout: DEFAULT_TIMEOUT,
+        }
+    }
+}
+
+impl Syncer {
+    /// A syncer that runs `alass` from `PATH`.
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Run a specific binary instead of whatever `PATH` resolves to.
+    #[must_use]
+    pub fn with_binary(mut self, binary: impl Into<OsString>) -> Self {
+        self.binary = binary.into();
+        self
+    }
+
+    /// Change how long one sync may take.
+    #[must_use]
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = timeout;
+        self
+    }
+
+    /// Sync `subtitle` against `video`, in place on disk.
+    ///
+    /// Returns the synced SRT text on acceptance; the input file is never
+    /// touched either way, so a rejection leaves the caller holding the
+    /// unsynced original it already has.
+    ///
+    /// # Errors
+    ///
+    /// [`Error::NotSrt`] when either side does not parse as SRT,
+    /// [`Error::Spawn`] when the binary is missing (a health condition,
+    /// #200), [`Error::Timeout`], [`Error::Failed`] when `alass` exits
+    /// non-zero, and [`Error::Io`] when a file cannot be read or written.
+    pub async fn sync(
+        &self,
+        video: impl AsRef<Path>,
+        subtitle: impl AsRef<Path>,
+    ) -> Result<Outcome> {
+        let video = video.as_ref();
+        let subtitle = subtitle.as_ref();
+
+        let text = tokio::fs::read_to_string(subtitle)
+            .await
+            .map_err(|source| Error::Io {
+                path: subtitle.to_path_buf(),
+                source,
+            })?;
+        let input = srt::parse(&text).map_err(|error| Error::NotSrt {
+            path: subtitle.to_path_buf(),
+            detail: error.to_string(),
+        })?;
+
+        let output_dir = tempfile::tempdir().map_err(|source| Error::Io {
+            path: std::env::temp_dir(),
+            source,
+        })?;
+        let synced_path = output_dir.path().join("synced.srt");
+        self.run(video, subtitle, &synced_path).await?;
+
+        let synced_text = tokio::fs::read_to_string(&synced_path)
+            .await
+            .map_err(|source| Error::Io {
+                path: synced_path.clone(),
+                source,
+            })?;
+        let output = srt::parse(&synced_text).map_err(|error| Error::NotSrt {
+            path: synced_path,
+            detail: error.to_string(),
+        })?;
+
+        Ok(match plausible(&input, &output) {
+            Some(rejection) => Outcome::Rejected(rejection),
+            None => Outcome::Accepted {
+                content: synced_text,
+            },
+        })
+    }
+
+    /// Spawn `alass` and wait for it to write its output.
+    async fn run(&self, video: &Path, subtitle: &Path, output: &Path) -> Result<()> {
+        let mut command = Command::new(&self.binary);
+        command
+            .args([subtitle, video])
+            .arg(output)
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true);
+
+        let child = command.spawn().map_err(|source| Error::Spawn {
+            binary: self.binary.to_string_lossy().into_owned(),
+            source,
+        })?;
+
+        let waited = tokio::time::timeout(self.timeout, child.wait_with_output()).await;
+        let output_handle = match waited {
+            Ok(output) => output.map_err(|source| Error::Io {
+                path: subtitle.to_path_buf(),
+                source,
+            })?,
+            Err(_) => {
+                return Err(Error::Timeout {
+                    path: subtitle.to_path_buf(),
+                })
+            }
+        };
+
+        if output_handle.status.success() {
+            return Ok(());
+        }
+
+        let stderr = String::from_utf8_lossy(&output_handle.stderr)
+            .trim()
+            .chars()
+            .take(STDERR_LIMIT)
+            .collect();
+        Err(Error::SyncFailed {
+            path: subtitle.to_path_buf(),
+            status: output_handle.status.code(),
+            stderr,
+        })
+    }
+}
+
+/// What one sync run decided.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Outcome {
+    /// Plausible; `content` is the synced SRT to write over the original.
+    Accepted {
+        /// The synced subtitle, as `alass` wrote it.
+        content: String,
+    },
+    /// Implausible. The unsynced original stands and the row is flagged.
+    Rejected(Rejection),
+}
+
+/// Why a sync's output was judged implausible (§15).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Rejection {
+    /// Cues disappeared between the input and the output.
+    CueLost {
+        /// How many the input had.
+        input: usize,
+        /// How many survived.
+        output: usize,
+    },
+    /// A cue moved further than [`MAX_SHIFT`] from where it started.
+    Shifted {
+        /// The cue's position in the file, 0-based.
+        index: usize,
+        /// How far it moved.
+        shift: Duration,
+    },
+}
+
+impl std::fmt::Display for Rejection {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::CueLost { input, output } => {
+                write!(formatter, "{output} of {input} cues survived")
+            }
+            Self::Shifted { index, shift } => {
+                write!(formatter, "cue {index} moved {shift:.1?}")
+            }
+        }
+    }
+}
+
+/// Compare a sync's output against its input.
+///
+/// Cues are paired positionally: `alass` shifts timings without reordering or
+/// merging, so position is the identity across the two files. Loss is checked
+/// first because a shifted comparison means nothing once counts diverge.
+fn plausible(input: &[srt::Cue], output: &[srt::Cue]) -> Option<Rejection> {
+    if output.len() < input.len() {
+        return Some(Rejection::CueLost {
+            input: input.len(),
+            output: output.len(),
+        });
+    }
+
+    input
+        .iter()
+        .zip(output)
+        .enumerate()
+        .find_map(|(index, (before, after))| {
+            let shift = after.start.abs_diff(before.start);
+            (shift > MAX_SHIFT).then_some(Rejection::Shifted { index, shift })
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{os::unix::fs::PermissionsExt, time::Duration};
+
+    use tokio::io::AsyncWriteExt;
+
+    use super::{plausible, Outcome, Rejection, Syncer, MAX_SHIFT};
+    use crate::error::Error;
+    use crate::srt::Cue;
+
+    const PLAIN: &str =
+        "1\n00:00:01,000 --> 00:00:02,500\nol\u{e1}\n\n2\n00:00:03,000 --> 00:00:04,000\ntwo lines\nof text\n";
+
+    /// Write an executable fake `alass`. Its body receives the three paths as
+    /// `$1` subtitle, `$2` video and `$3` output.
+    async fn fake_binary(dir: &std::path::Path, body: &str) -> std::path::PathBuf {
+        let path = dir.join("alass");
+        let mut file = tokio::fs::File::create(&path)
+            .await
+            .expect("test setup and fake binary succeed");
+        file.write_all(b"#!/bin/sh\n")
+            .await
+            .expect("test setup and fake binary succeed");
+        file.write_all(body.as_bytes())
+            .await
+            .expect("test setup and fake binary succeed");
+        file.sync_all()
+            .await
+            .expect("test setup and fake binary succeed");
+        drop(file);
+        tokio::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
+            .await
+            .expect("test setup and fake binary succeed");
+        path
+    }
+
+    async fn write_subtitle(
+        dir: &std::path::Path,
+        name: &str,
+        content: &str,
+    ) -> std::path::PathBuf {
+        let path = dir.join(name);
+        tokio::fs::write(&path, content)
+            .await
+            .expect("test setup and fake binary succeed");
+        path
+    }
+
+    fn cue(start: u64) -> Cue {
+        Cue {
+            start: Duration::from_millis(start * 1_000),
+            end: Duration::from_millis(start * 1_000 + 1_500),
+            text: "line".to_owned(),
+        }
+    }
+
+    #[tokio::test]
+    async fn an_unchanged_output_is_accepted_with_alass_text() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let binary = fake_binary(dir.path(), "cp \"$1\" \"$3\"\n").await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let outcome = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect("test setup and fake binary succeed");
+
+        assert_eq!(
+            outcome,
+            Outcome::Accepted {
+                content: PLAIN.to_owned()
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn a_shift_within_the_bound_is_accepted() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        // A five-second delay, the classic PAL-to-film speedup correction.
+        let binary = fake_binary(
+            dir.path(),
+            "printf '1\\n00:00:06,000 --> 00:00:07,500\\nola\\n\\n2\\n00:00:08,000 --> 00:00:09,000\\ntext\\n' > \"$3\"\n",
+        )
+        .await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let outcome = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect("test setup and fake binary succeed");
+
+        assert!(matches!(outcome, Outcome::Accepted { .. }));
+    }
+
+    #[tokio::test]
+    async fn a_shift_beyond_sixty_seconds_is_rejected_and_the_original_stands() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let binary = fake_binary(
+            dir.path(),
+            "printf '1\\n00:01:40,000 --> 00:01:41,500\\nola\\n\\n2\\n00:01:42,000 --> 00:01:43,000\\ntext\\n' > \"$3\"\n",
+        )
+        .await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let outcome = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect("test setup and fake binary succeed");
+
+        assert_eq!(
+            outcome,
+            Outcome::Rejected(Rejection::Shifted {
+                index: 0,
+                shift: Duration::from_secs(99)
+            })
+        );
+        // The rejection never touched the input file.
+        assert_eq!(
+            tokio::fs::read_to_string(&subtitle)
+                .await
+                .expect("test setup and fake binary succeed"),
+            PLAIN
+        );
+    }
+
+    #[tokio::test]
+    async fn lost_cues_are_rejected() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let binary = fake_binary(
+            dir.path(),
+            "printf '1\\n00:00:01,000 --> 00:00:02,500\\nola\\n' > \"$3\"\n",
+        )
+        .await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let outcome = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect("test setup and fake binary succeed");
+
+        assert_eq!(
+            outcome,
+            Outcome::Rejected(Rejection::CueLost {
+                input: 2,
+                output: 1
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn a_failing_run_names_the_stderr() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let binary = fake_binary(dir.path(), "echo 'no such stream' >&2\nexit 3\n").await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let error = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect_err("the run must fail");
+
+        match error {
+            Error::SyncFailed { status, stderr, .. } => {
+                assert_eq!(status, Some(3));
+                assert_eq!(stderr, "no such stream");
+            }
+            other => panic!("expected SyncFailed, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn a_missing_binary_is_a_spawn_error_not_a_panic() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let error = Syncer::new()
+            .with_binary("alass-not-installed")
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect_err("the run must fail");
+
+        assert!(matches!(error, Error::Spawn { .. }));
+    }
+
+    #[tokio::test]
+    async fn an_input_that_is_not_srt_never_reaches_alass() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        // Any run would leave this marker behind; its absence proves the
+        // binary was never invoked.
+        let marker = dir.path().join("marker");
+        let binary =
+            fake_binary(dir.path(), &format!("touch {}\nexit 0\n", marker.display())).await;
+        let subtitle = write_subtitle(dir.path(), "in.txt", "{1}{50}not srt at all").await;
+
+        let error = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect_err("the run must fail");
+
+        assert!(matches!(error, Error::NotSrt { .. }));
+        assert!(!marker.exists(), "alass must not run on unparseable input");
+    }
+
+    #[tokio::test]
+    async fn an_output_that_is_not_srt_is_an_error_rather_than_a_rejection() {
+        let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
+        let binary = fake_binary(dir.path(), "echo garbage > \"$3\"\n").await;
+        let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
+
+        let error = Syncer::new()
+            .with_binary(&binary)
+            .sync(dir.path().join("video.mkv"), &subtitle)
+            .await
+            .expect_err("the run must fail");
+
+        assert!(matches!(error, Error::NotSrt { .. }));
+    }
+
+    #[test]
+    fn exactly_sixty_seconds_is_still_plausible() {
+        let before = [cue(10)];
+        let mut after = before[0].clone();
+        after.start += MAX_SHIFT;
+        assert_eq!(plausible(&before, &[after]), None);
+    }
+
+    #[test]
+    fn a_grown_output_is_not_loss_but_a_shift_is_still_caught() {
+        let before = [cue(10), cue(20)];
+        let grown = vec![cue(10), cue(20), cue(30)];
+        assert_eq!(plausible(&before, &grown), None, "extra cues are not loss");
+
+        let mut moved = vec![cue(10), cue(20), cue(30)];
+        moved[1].start += MAX_SHIFT + Duration::from_secs(1);
+        assert_eq!(
+            plausible(&before, &moved),
+            Some(Rejection::Shifted {
+                index: 1,
+                shift: MAX_SHIFT + Duration::from_secs(1)
+            })
+        );
+    }
+
+    #[test]
+    fn loss_is_reported_before_any_shift() {
+        let before = [cue(10), cue(600)];
+        let after = [cue(700)];
+        assert_eq!(
+            plausible(&before, &after),
+            Some(Rejection::CueLost {
+                input: 2,
+                output: 1
+            })
+        );
+    }
+
+    #[test]
+    fn rejections_render_as_a_sentence() {
+        assert_eq!(
+            Rejection::CueLost {
+                input: 2,
+                output: 1
+            }
+            .to_string(),
+            "1 of 2 cues survived"
+        );
+    }
+}