feat: add OBS recording engine and validation probe

clipbin-obs starts the pinned libobs on a dedicated thread, opens exactly the bundled plugin allowlist, and records official capture sources through one hardware encoder into keyframe-aligned Matroska segments. Status reaches Recording only once the source has a picture and the segment grows on disk. clipbin-probe lists and verifies hardware encoders and records from a real capture source, printing status, health and segment events for validation.
2026-10-08 14:12:06 +02:00
parent 80b6abdb32
commit f2eb7139da
14 changed files with 2470 additions and 3 deletions
@@ -0,0 +1,509 @@
+use std::{
+    fmt,
+    path::{Path, PathBuf},
+    sync::{
+        Arc, Mutex,
+        mpsc::{self, Receiver},
+    },
+    time::{Duration, Instant},
+};
+
+use clipbin_core::{RecordingStatus, VideoCodec};
+
+use crate::{
+    RuntimeLayout,
+    capture::{SettingValue, SourceSpec},
+    encoders::registered_candidates,
+    ffi::{self, Pipeline, PipelineSpec, Session, StartupSpec},
+    runtime::PLUGINS,
+    thread::ObsThread,
+};
+
+/// Segments are Matroska: a segment cut short by a crash stays readable, and
+/// export remuxes to MP4 without re-encoding.
+pub const SEGMENT_EXTENSION: &str = "mkv";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct VideoFormat {
+    pub width: u32,
+    pub height: u32,
+    pub frames_per_second: u32,
+}
+
+/// The single libobs instance. Dropping it shuts OBS down.
+pub struct Engine {
+    obs: Arc<ObsThread>,
+    session: Option<Session>,
+    format: VideoFormat,
+    #[cfg(target_os = "linux")]
+    _glib: ffi::GlibLoop,
+}
+
+impl Engine {
+    pub fn start(layout: &RuntimeLayout, format: VideoFormat) -> Result<Self, EngineError> {
+        let path = |path: PathBuf| path.to_string_lossy().into_owned();
+        let spec = StartupSpec {
+            libobs_data_dir: path(layout.libobs_data_dir()),
+            plugins: PLUGINS
+                .iter()
+                .map(|plugin| {
+                    (
+                        path(layout.plugin_library(plugin)),
+                        path(layout.plugin_data_dir(plugin)),
+                    )
+                })
+                .collect(),
+            width: format.width,
+            height: format.height,
+            frames_per_second: format.frames_per_second,
+        };
+        #[cfg(target_os = "linux")]
+        let glib = ffi::GlibLoop::start();
+        let obs = Arc::new(ObsThread::spawn());
+        let session = obs
+            .run(move || ffi::startup(&spec))
+            .map_err(EngineError::Startup)?;
+        Ok(Self {
+            obs,
+            session: Some(session),
+            format,
+            #[cfg(target_os = "linux")]
+            _glib: glib,
+        })
+    }
+
+    pub fn obs_version(&self) -> &str {
+        &self.session().version
+    }
+
+    /// Bundled plugins that declined to initialise, such as NVENC without an
+    /// NVIDIA GPU. Their encoders are simply not registered.
+    pub fn plugins_failed(&self) -> &[String] {
+        &self.session().plugins_failed
+    }
+
+    fn session(&self) -> &Session {
+        self.session.as_ref().expect("the session lives until drop")
+    }
+
+    /// Registered video encoders as `(id, codec)` pairs, such as `("ffmpeg_vaapi_tex", "h264")`.
+    pub fn registered_video_encoders(&self) -> Vec<(String, String)> {
+        self.obs.run(ffi::registered_video_encoders)
+    }
+
+    /// Registered hardware encoders for `codec`, most preferred first.
+    pub fn hardware_encoders(&self, codec: VideoCodec) -> Vec<String> {
+        let registered: Vec<String> = self
+            .registered_video_encoders()
+            .into_iter()
+            .map(|(id, _)| id)
+            .collect();
+        registered_candidates(codec, &registered)
+            .into_iter()
+            .map(str::to_owned)
+            .collect()
+    }
+
+    pub fn start_recording(&self, config: RecordingConfig) -> Result<Recording, EngineError> {
+        std::fs::create_dir_all(&config.segment_dir).map_err(EngineError::Io)?;
+        let spec = PipelineSpec {
+            canvas: (self.format.width, self.format.height),
+            video_source: config.video_source,
+            system_audio: config.system_audio,
+            microphone: config.microphone,
+            video_encoder: config.video_encoder,
+            video_encoder_settings: vec![
+                ("rate_control", SettingValue::String("CBR".into())),
+                ("bitrate", SettingValue::Int(config.video_bitrate_kbps)),
+                (
+                    "keyint_sec",
+                    SettingValue::Int(config.keyframe_interval_seconds),
+                ),
+            ],
+            audio_bitrate_kbps: 160,
+            segment_dir: forward_slashes(&config.segment_dir),
+            segment_extension: SEGMENT_EXTENSION,
+            segment_seconds: config.segment_seconds,
+        };
+        let (sender, events) = mpsc::channel();
+        let current_segment = Arc::new(Mutex::new(None));
+        let segment = Arc::clone(&current_segment);
+        let pipeline = self
+            .obs
+            .run(move || Pipeline::build(&spec, sender, segment).map(Arc::new))
+            .map_err(EngineError::Pipeline)?;
+        let recording = Recording {
+            obs: Arc::clone(&self.obs),
+            pipeline: Some(pipeline),
+            events,
+            current_segment,
+        };
+        recording
+            .with_pipeline(|pipeline| pipeline.start())
+            .map_err(EngineError::Start)?;
+        Ok(recording)
+    }
+}
+
+impl Drop for Engine {
+    fn drop(&mut self) {
+        if let Some(session) = self.session.take() {
+            self.obs.run(move || ffi::shutdown(session));
+        }
+    }
+}
+
+/// Paths in OBS settings use forward slashes on every platform.
+fn forward_slashes(path: &Path) -> String {
+    path.to_string_lossy().replace('\\', "/")
+}
+
+pub struct RecordingConfig {
+    pub video_source: Option<SourceSpec>,
+    pub system_audio: Option<SourceSpec>,
+    pub microphone: Option<SourceSpec>,
+    pub video_encoder: String,
+    pub video_bitrate_kbps: i64,
+    pub keyframe_interval_seconds: i64,
+    /// Segments split at the first keyframe after this many seconds.
+    pub segment_seconds: i64,
+    pub segment_dir: PathBuf,
+}
+
+/// An active recording. Dropping it stops the output and releases its OBS
+/// objects on the OBS thread. Drop recordings before their `Engine`.
+pub struct Recording {
+    obs: Arc<ObsThread>,
+    pipeline: Option<Arc<Pipeline>>,
+    events: Receiver<RecorderEvent>,
+    current_segment: Arc<Mutex<Option<String>>>,
+}
+
+impl Recording {
+    fn with_pipeline<T: Send + 'static>(
+        &self,
+        operation: impl FnOnce(&Pipeline) -> T + Send + 'static,
+    ) -> T {
+        let pipeline = Arc::clone(
+            self.pipeline
+                .as_ref()
+                .expect("the pipeline lives until drop"),
+        );
+        self.obs.run(move || operation(&pipeline))
+    }
+
+    pub fn events(&self) -> &Receiver<RecorderEvent> {
+        &self.events
+    }
+
+    /// OBS's counters plus the size of the segment being written. OBS counts
+    /// bytes handed to its muxer process, which keeps rising even if that
+    /// process cannot write, so only growth on disk proves footage is saved.
+    pub fn health(&self) -> Health {
+        let mut health = self.with_pipeline(Pipeline::health);
+        let segment = self
+            .current_segment
+            .lock()
+            .ok()
+            .and_then(|current| current.clone());
+        if let Some(path) = segment {
+            health.current_segment_bytes =
+                std::fs::metadata(&path).map_or(0, |metadata| metadata.len());
+            health.current_segment = Some(path);
+        }
+        health
+    }
+
+    pub fn restore_token(&self) -> Option<String> {
+        self.with_pipeline(Pipeline::restore_token)
+    }
+
+    /// Requests a stop; `OutputStopped` follows once the last segment is finalised.
+    pub fn stop(&self) {
+        self.with_pipeline(Pipeline::stop)
+    }
+}
+
+impl Drop for Recording {
+    fn drop(&mut self) {
+        if let Some(pipeline) = self.pipeline.take() {
+            // OBS objects must be released on the OBS thread.
+            self.obs.run(move || drop(pipeline));
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RecorderEvent {
+    /// OBS reports the output as started. Not yet proof that footage is written.
+    OutputStarted,
+    /// The muxer opened a new segment file; the previous one is complete.
+    SegmentStarted { path: String },
+    OutputStopped {
+        code: i64,
+        last_error: Option<String>,
+    },
+}
+
+/// Counters for honest recording status: OBS accepting a start request is not
+/// enough, frames must reach disk and the source must have a picture.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Health {
+    pub source_width: u32,
+    pub source_height: u32,
+    pub output_active: bool,
+    pub total_frames: u64,
+    pub dropped_frames: u64,
+    pub total_bytes: u64,
+    pub lagged_frames: u64,
+    pub current_segment: Option<String>,
+    pub current_segment_bytes: u64,
+}
+
+/// Derives the status shown to the user from health samples and output events.
+///
+/// Recording is reported only while the source has a picture and encoded data
+/// keeps reaching the muxer. A stall after recording began is a failure, not a
+/// silent pause.
+#[derive(Debug)]
+pub struct StatusTracker {
+    status: RecordingStatus,
+    last_written: (Option<String>, u64),
+    last_progress: Instant,
+}
+
+impl StatusTracker {
+    /// How long footage may stop arriving before recording counts as failed.
+    pub const STALL_LIMIT: Duration = Duration::from_secs(3);
+
+    pub fn new(now: Instant) -> Self {
+        Self {
+            status: RecordingStatus::Starting,
+            last_written: (None, 0),
+            last_progress: now,
+        }
+    }
+
+    pub fn status(&self) -> &RecordingStatus {
+        &self.status
+    }
+
+    pub fn sample(&mut self, health: &Health, now: Instant) -> &RecordingStatus {
+        if matches!(self.status, RecordingStatus::Failed { .. }) {
+            return &self.status;
+        }
+        // Progress means the segment on disk grew or a new one began.
+        let written = (health.current_segment.clone(), health.current_segment_bytes);
+        let progressed = written.1 > 0 && written != self.last_written;
+        if progressed {
+            self.last_written = written;
+            self.last_progress = now;
+        }
+        let has_picture = health.source_width > 0 && health.source_height > 0;
+        let stalled = now.duration_since(self.last_progress) >= Self::STALL_LIMIT;
+
+        self.status = match (&self.status, has_picture, progressed || !stalled) {
+            (_, true, true)
+                if health.output_active && health.total_frames > 0 && self.last_written.1 > 0 =>
+            {
+                RecordingStatus::Recording
+            }
+            (RecordingStatus::Recording, false, _) => RecordingStatus::Failed {
+                reason: "the capture source stopped providing a picture".into(),
+            },
+            (RecordingStatus::Recording, true, false) => RecordingStatus::Failed {
+                reason: format!(
+                    "no footage was written for {} seconds",
+                    Self::STALL_LIMIT.as_secs()
+                ),
+            },
+            (_, false, _) => RecordingStatus::WaitingForCapture,
+            (status, _, _) => status.clone(),
+        };
+        &self.status
+    }
+
+    pub fn stopped(&mut self, reason: StopReason, last_error: Option<&str>) -> &RecordingStatus {
+        self.status = match reason {
+            StopReason::Success => RecordingStatus::Idle,
+            reason => RecordingStatus::Failed {
+                reason: match last_error {
+                    Some(error) => format!("{reason:?}: {error}"),
+                    None => format!("{reason:?}"),
+                },
+            },
+        };
+        &self.status
+    }
+}
+
+/// Why an output stopped, from `obs-output.h`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum StopReason {
+    Success,
+    BadPath,
+    Unsupported,
+    NoSpace,
+    EncodeError,
+    Disconnected,
+    Error,
+    Other(i64),
+}
+
+impl StopReason {
+    pub fn from_code(code: i64) -> Self {
+        match code as i32 {
+            code if code == libobs::OBS_OUTPUT_SUCCESS as i32 => Self::Success,
+            libobs::OBS_OUTPUT_BAD_PATH => Self::BadPath,
+            libobs::OBS_OUTPUT_UNSUPPORTED => Self::Unsupported,
+            libobs::OBS_OUTPUT_NO_SPACE => Self::NoSpace,
+            libobs::OBS_OUTPUT_ENCODE_ERROR => Self::EncodeError,
+            libobs::OBS_OUTPUT_DISCONNECTED => Self::Disconnected,
+            libobs::OBS_OUTPUT_ERROR => Self::Error,
+            _ => Self::Other(code),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum EngineError {
+    Startup(String),
+    Io(std::io::Error),
+    Pipeline(String),
+    Start(String),
+}
+
+impl fmt::Display for EngineError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Startup(error) => write!(formatter, "OBS did not start: {error}"),
+            Self::Io(error) => write!(formatter, "{error}"),
+            Self::Pipeline(error) => write!(formatter, "could not set up recording: {error}"),
+            Self::Start(error) => write!(formatter, "recording did not start: {error}"),
+        }
+    }
+}
+
+impl std::error::Error for EngineError {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn healthy(bytes: u64) -> Health {
+        Health {
+            source_width: 1920,
+            source_height: 1080,
+            output_active: true,
+            total_frames: bytes / 1000,
+            dropped_frames: 0,
+            total_bytes: bytes,
+            lagged_frames: 0,
+            current_segment: Some("segment-000000.mkv".into()),
+            current_segment_bytes: bytes,
+        }
+    }
+
+    #[test]
+    fn footage_must_reach_disk_not_just_the_muxer() {
+        let start = Instant::now();
+        let mut tracker = StatusTracker::new(start);
+        let unwritten = Health {
+            current_segment_bytes: 0,
+            ..healthy(5000)
+        };
+        assert_ne!(
+            tracker.sample(&unwritten, start),
+            &RecordingStatus::Recording
+        );
+        assert_eq!(
+            tracker.sample(&healthy(5000), start),
+            &RecordingStatus::Recording
+        );
+        // OBS keeps counting, but the file on disk stopped growing.
+        let stuck = Health {
+            total_bytes: 90_000,
+            ..healthy(5000)
+        };
+        let later = start + StatusTracker::STALL_LIMIT;
+        assert!(matches!(
+            tracker.sample(&stuck, later),
+            RecordingStatus::Failed { .. }
+        ));
+    }
+
+    #[test]
+    fn recording_needs_a_picture_and_written_footage() {
+        let start = Instant::now();
+        let mut tracker = StatusTracker::new(start);
+        let no_picture = Health {
+            source_width: 0,
+            source_height: 0,
+            ..healthy(5000)
+        };
+        assert_eq!(
+            tracker.sample(&no_picture, start),
+            &RecordingStatus::WaitingForCapture
+        );
+        assert_eq!(
+            tracker.sample(&healthy(0), start),
+            &RecordingStatus::WaitingForCapture
+        );
+        assert_eq!(
+            tracker.sample(&healthy(5000), start),
+            &RecordingStatus::Recording
+        );
+    }
+
+    #[test]
+    fn a_stall_or_lost_source_fails_instead_of_pausing() {
+        let start = Instant::now();
+        let mut tracker = StatusTracker::new(start);
+        tracker.sample(&healthy(5000), start);
+        let later = start + StatusTracker::STALL_LIMIT;
+        assert!(matches!(
+            tracker.sample(&healthy(5000), later),
+            RecordingStatus::Failed { .. }
+        ));
+
+        let mut tracker = StatusTracker::new(start);
+        tracker.sample(&healthy(5000), start);
+        let lost = Health {
+            source_width: 0,
+            ..healthy(6000)
+        };
+        assert!(matches!(
+            tracker.sample(&lost, start),
+            RecordingStatus::Failed { .. }
+        ));
+        // Failure persists until a new recording starts.
+        assert!(matches!(
+            tracker.sample(&healthy(9000), start),
+            RecordingStatus::Failed { .. }
+        ));
+    }
+
+    #[test]
+    fn output_errors_fail_and_clean_stops_go_idle() {
+        let mut tracker = StatusTracker::new(Instant::now());
+        assert!(matches!(
+            tracker.stopped(StopReason::NoSpace, Some("disk full")),
+            RecordingStatus::Failed { reason } if reason.contains("disk full")
+        ));
+        let mut tracker = StatusTracker::new(Instant::now());
+        assert_eq!(
+            tracker.stopped(StopReason::Success, None),
+            &RecordingStatus::Idle
+        );
+    }
+
+    #[test]
+    fn stop_codes_map_to_reasons() {
+        assert_eq!(StopReason::from_code(0), StopReason::Success);
+        assert_eq!(
+            StopReason::from_code(libobs::OBS_OUTPUT_NO_SPACE.into()),
+            StopReason::NoSpace
+        );
+        assert_eq!(StopReason::from_code(-99), StopReason::Other(-99));
+    }
+}