metrale_model_layers/
video_decode_ffmpeg.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-25: Decodes video containers to RGB frames by running ffmpeg as a
4//! subprocess, for every container that is not a GIF.
5//!
6//! Owner: model-layers (vision input).
7//! Invariants:
8//! - `decode_frames` refuses before building a process unless
9//!   `FfmpegPolicy::enabled`.
10//! - The binary gets its arguments as argv, never through a shell, and the
11//!   container over stdin (`pipe:0`); no temp file is written.
12//!
13//! The payload is an untrusted request body, so a decode is bounded: the frame
14//! count by `-frames:v`, the output by a capped read that kills the child once
15//! it passes `max_output_bytes`, and the wall clock by a watchdog that kills
16//! the child after `timeout_secs`.
17
18use anyhow::{Context, Result, bail, ensure};
19use image::RgbImage;
20use std::io::{Read, Write};
21use std::process::{Command, Stdio};
22use std::sync::{Arc, Mutex};
23
24/// 2026-09-25: Operator policy for subprocess decoding. The serve builds it from
25/// the `--video-*` flags.
26#[derive(Debug, Clone)]
27pub struct FfmpegPolicy {
28    pub enabled: bool,
29    /// 2026-09-25: Binary to run. A bare name is resolved on PATH; an absolute path
30    /// is used as given.
31    pub binary: String,
32    pub max_frames: usize,
33    pub max_output_bytes: usize,
34    pub timeout_secs: u64,
35}
36
37impl Default for FfmpegPolicy {
38    fn default() -> Self {
39        Self {
40            enabled: false,
41            binary: "ffmpeg".to_string(),
42            max_frames: 768,
43            // 2026-09-25: No serve flag sets this cap; serve_load.rs takes it from
44            // this `Default`.
45            max_output_bytes: 512 * 1024 * 1024,
46            timeout_secs: 120,
47        }
48    }
49}
50
51/// 2026-09-25: The 8-byte PNG signature. Frames arrive concatenated on stdout and
52/// `split_png_stream` separates them at each occurrence.
53const PNG_MAGIC: [u8; 8] = [0x89, b'P', b'N', b'G', 0x0d, 0x0a, 0x1a, 0x0a];
54
55/// 2026-09-25: What `probe` found.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub enum Availability {
58    /// 2026-09-25: Usable, with the first line of its `-version` output.
59    Ready(String),
60    /// 2026-09-25: Enabled but not runnable, with the reason.
61    Missing(String),
62    /// 2026-09-25: `FfmpegPolicy::enabled` is false.
63    Disabled,
64}
65
66/// 2026-09-25: Check whether the configured decoder runs, with one `-version`
67/// invocation. The serve calls it at boot so a missing binary shows in the
68/// startup log rather than on the first video request.
69pub fn probe(policy: &FfmpegPolicy) -> Availability {
70    if !policy.enabled {
71        return Availability::Disabled;
72    }
73    match Command::new(&policy.binary)
74        .arg("-version")
75        .stdin(Stdio::null())
76        .stdout(Stdio::piped())
77        .stderr(Stdio::null())
78        .output()
79    {
80        Ok(out) if out.status.success() => {
81            let first = String::from_utf8_lossy(&out.stdout)
82                .lines()
83                .next()
84                .unwrap_or("unknown version")
85                .trim()
86                .to_string();
87            Availability::Ready(first)
88        }
89        Ok(out) => {
90            Availability::Missing(format!("{:?} ran but exited {}", policy.binary, out.status))
91        }
92        Err(e) => Availability::Missing(format!("{:?} could not be run: {e}", policy.binary)),
93    }
94}
95
96/// 2026-09-25: The error for a decoder that could not be started. It always
97/// carries the OS error. Only `NotFound` gets the install hint;
98/// `PermissionDenied` and `ExecutableFileBusy` (ETXTBSY) get their own hint,
99/// and other kinds get none.
100fn spawn_failure(binary: &str, err: std::io::Error) -> anyhow::Error {
101    let hint = match err.kind() {
102        std::io::ErrorKind::NotFound => {
103            " — is ffmpeg installed and on PATH? (set --video-ffmpeg-path to point at it)"
104        }
105        std::io::ErrorKind::PermissionDenied => {
106            " — not executable by this user, or its filesystem is mounted noexec"
107        }
108        std::io::ErrorKind::ExecutableFileBusy => {
109            " — another process still holds it open for writing (ETXTBSY)"
110        }
111        _ => "",
112    };
113    anyhow::anyhow!("could not run {binary:?}: {err}{hint}")
114}
115
116/// 2026-09-25: Decode `bytes` to RGB frames sampled at `target_fps`.
117///
118/// ffmpeg samples with `-vf fps=`, so the returned frames are already at
119/// `target_fps`, and the caller does not need the source frame rate. A
120/// non-finite or non-positive `target_fps` samples at 2.0.
121pub fn decode_frames(
122    bytes: &[u8],
123    target_fps: f32,
124    policy: &FfmpegPolicy,
125) -> Result<Vec<RgbImage>> {
126    ensure!(
127        policy.enabled,
128        "this container needs ffmpeg to decode and subprocess decoding is disabled; \
129         pass --video-allow-ffmpeg to enable it, or send an animated GIF"
130    );
131    let fps = if target_fps.is_finite() && target_fps > 0.0 {
132        target_fps
133    } else {
134        2.0
135    };
136
137    let mut child = Command::new(&policy.binary)
138        .args([
139            "-v",
140            "error",
141            // 2026-09-25: Stdin carries the container, so ffmpeg must not read
142            // it as keyboard input.
143            "-nostdin",
144            "-i",
145            "pipe:0",
146            "-vf",
147            &format!("fps={fps}"),
148            "-frames:v",
149            &policy.max_frames.to_string(),
150            "-f",
151            "image2pipe",
152            "-vcodec",
153            "png",
154            "pipe:1",
155        ])
156        .stdin(Stdio::piped())
157        .stdout(Stdio::piped())
158        .stderr(Stdio::piped())
159        .spawn()
160        .map_err(|err| spawn_failure(&policy.binary, err))?;
161
162    let mut stdin = child.stdin.take().context("no stdin pipe")?;
163    let mut stdout = child.stdout.take().context("no stdout pipe")?;
164    let mut stderr = child.stderr.take().context("no stderr pipe")?;
165
166    // 2026-09-25: Feed the container on a thread, concurrently with the stdout
167    // read below: ffmpeg writes output while it still consumes input, so
168    // writing everything first can deadlock once the output pipe fills.
169    let input = bytes.to_vec();
170    let writer = std::thread::spawn(move || {
171        // 2026-09-25: ffmpeg may stop reading once it has `-frames:v` frames, so
172        // a write error is expected and ignored.
173        let _ = stdin.write_all(&input);
174        drop(stdin);
175    });
176
177    let child = Arc::new(Mutex::new(child));
178    let watchdog = {
179        let child = Arc::clone(&child);
180        let secs = policy.timeout_secs.max(1);
181        std::thread::spawn(move || {
182            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs);
183            loop {
184                std::thread::sleep(std::time::Duration::from_millis(100));
185                let mut guard = match child.lock() {
186                    Ok(g) => g,
187                    Err(_) => return false,
188                };
189                match guard.try_wait() {
190                    Ok(Some(_)) => return false,
191                    Ok(None) => {}
192                    Err(_) => return false,
193                }
194                if std::time::Instant::now() >= deadline {
195                    let _ = guard.kill();
196                    return true;
197                }
198            }
199        })
200    };
201
202    // 2026-09-25: Read at most one byte past the cap; `over_cap` below detects
203    // that byte.
204    let mut out = Vec::new();
205    let read_res = (&mut stdout)
206        .take(policy.max_output_bytes as u64 + 1)
207        .read_to_end(&mut out);
208
209    // 2026-09-25: Kill before draining stderr. Past the cap nothing reads
210    // stdout, so the child blocks on a full pipe and never closes stderr, and
211    // the stderr read would wait until the watchdog fires.
212    let over_cap = out.len() > policy.max_output_bytes;
213    if over_cap && let Ok(mut g) = child.lock() {
214        let _ = g.kill();
215    }
216
217    let mut err_text = String::new();
218    let _ = stderr.read_to_string(&mut err_text);
219    let _ = writer.join();
220
221    let status = {
222        let mut g = child
223            .lock()
224            .map_err(|_| anyhow::anyhow!("decoder lock poisoned"))?;
225        g.wait().context("waiting for the decoder")?
226    };
227    let timed_out = watchdog.join().unwrap_or(false);
228
229    read_res.context("reading decoded frames")?;
230    ensure!(
231        !timed_out,
232        "decoding exceeded {}s and was stopped",
233        policy.timeout_secs
234    );
235    ensure!(
236        !over_cap,
237        "decoded output exceeded the {}-byte cap",
238        policy.max_output_bytes
239    );
240    if !status.success() {
241        let why = err_text.lines().next_back().unwrap_or("no detail").trim();
242        bail!("decoder failed: {why}");
243    }
244
245    let frames = split_png_stream(&out)?;
246    ensure!(
247        !frames.is_empty(),
248        "the container decoded to zero frames (is there a video stream?)"
249    );
250    Ok(frames)
251}
252
253/// 2026-09-25: Split a concatenated PNG stream into images, starting a new image
254/// at each `PNG_MAGIC`. An empty stream gives no images; a piece that is not a
255/// valid PNG is an error naming its index.
256fn split_png_stream(buf: &[u8]) -> Result<Vec<RgbImage>> {
257    let mut starts = Vec::new();
258    let mut i = 0usize;
259    while i + PNG_MAGIC.len() <= buf.len() {
260        if buf[i..i + PNG_MAGIC.len()] == PNG_MAGIC {
261            starts.push(i);
262            i += PNG_MAGIC.len();
263        } else {
264            i += 1;
265        }
266    }
267    let mut frames = Vec::with_capacity(starts.len());
268    for (n, &s) in starts.iter().enumerate() {
269        let e = starts.get(n + 1).copied().unwrap_or(buf.len());
270        let img = image::load_from_memory_with_format(&buf[s..e], image::ImageFormat::Png)
271            .with_context(|| format!("frame {n} did not decode as PNG"))?;
272        frames.push(img.to_rgb8());
273    }
274    Ok(frames)
275}
276
277#[cfg(test)]
278#[path = "video_decode_ffmpeg_tests.rs"]
279mod tests;
metrale_model_layers/video_decode_ffmpeg.rs

metrale_model_layers/
video_decode_ffmpeg.rs
