diff --git a/Cargo.lock b/Cargo.lock index e0a6471..e2702b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -876,6 +876,7 @@ name = "clipbin-obs" version = "0.1.0" dependencies = [ "clipbin-core", + "fs4", "libobs", ] @@ -1842,6 +1843,16 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "fs4" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e72ed92b67c146290f88e9c89d60ca163ea417a446f61ffd7b72df3e7f1dfd5" +dependencies = [ + "rustix 1.1.5", + "windows-sys 0.61.2", +] + [[package]] name = "futf" version = "0.1.5" diff --git a/crates/clipbin-obs/Cargo.toml b/crates/clipbin-obs/Cargo.toml index 89e1181..b737bd5 100644 --- a/crates/clipbin-obs/Cargo.toml +++ b/crates/clipbin-obs/Cargo.toml @@ -10,6 +10,7 @@ publish.workspace = true [dependencies] clipbin-core.workspace = true +fs4 = { version = "1.1", default-features = false, features = ["sync"] } libobs.workspace = true # Raw libobs calls are confined to src/ffi.rs, which opts back in to unsafe code. diff --git a/crates/clipbin-obs/src/ffi.rs b/crates/clipbin-obs/src/ffi.rs index ade5720..30dba02 100644 --- a/crates/clipbin-obs/src/ffi.rs +++ b/crates/clipbin-obs/src/ffi.rs @@ -10,7 +10,7 @@ use std::{ ptr, sync::{ Arc, Mutex, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicU8, AtomicU64, Ordering}, mpsc::Sender, }, }; @@ -147,12 +147,38 @@ unsafe extern "C" fn log_to_stderr( _ => "debug", }; if let Some(text) = owned(message.array) { + if let Some(state) = pipewire_stream_state(&text) { + let code = if state == "streaming" { + STREAMING + } else { + NOT_STREAMING + }; + PIPEWIRE_STREAM.store(code, Ordering::Relaxed); + } eprintln!("obs {label}: {text}"); } // SAFETY: frees the buffer dstr_vprintf allocated (null is allowed). unsafe { sys::bfree(message.array.cast()) }; } +/// The state of the PipeWire capture stream, as last logged by OBS. +/// +/// OBS 32.0.4's PipeWire capture keeps reporting its last size and rendering +/// its last frame after the stream stops, and exposes the stream state only +/// through this log line. Clipbin records one capture source at a time. +static PIPEWIRE_STREAM: AtomicU8 = AtomicU8::new(NO_PIPEWIRE_STREAM); +const NO_PIPEWIRE_STREAM: u8 = 0; +const STREAMING: u8 = 1; +const NOT_STREAMING: u8 = 2; + +/// Parses `[pipewire] Stream 0x... state: "streaming" (error: none)` from +/// linux-pipewire's on_state_changed_cb. +fn pipewire_stream_state(message: &str) -> Option<&str> { + let rest = message.strip_prefix("[pipewire] Stream ")?; + let state = rest.split_once(" state: \"")?.1; + state.split_once('"').map(|(state, _)| state) +} + pub(crate) fn startup(spec: &StartupSpec) -> Result { // SAFETY: installs a handler with the signature libobs expects. unsafe { sys::base_set_log_handler(Some(log_to_stderr), ptr::null_mut()) }; @@ -540,6 +566,7 @@ impl Pipeline { events: Sender, current_segment: Arc>>, ) -> Result { + PIPEWIRE_STREAM.store(NO_PIPEWIRE_STREAM, Ordering::Relaxed); // OBS releases objects asynchronously, so names must not be reused. static RECORDINGS: AtomicU64 = AtomicU64::new(0); let id = RECORDINGS.fetch_add(1, Ordering::Relaxed); @@ -716,9 +743,14 @@ impl Pipeline { sys::obs_source_get_height(self.video_source), ) }; + let source_streaming = match PIPEWIRE_STREAM.load(Ordering::Relaxed) { + NO_PIPEWIRE_STREAM => None, + state => Some(state == STREAMING), + }; Health { source_width: width, source_height: height, + source_streaming, output_active: sys::obs_output_active(self.output), total_frames: sys::obs_output_get_total_frames(self.output).max(0) as u64, dropped_frames: sys::obs_output_get_frames_dropped(self.output).max(0) as u64, @@ -726,6 +758,7 @@ impl Pipeline { lagged_frames: sys::obs_get_lagged_frames() as u64, current_segment: None, current_segment_bytes: 0, + disk_available_bytes: None, } } } @@ -810,6 +843,16 @@ unsafe fn create_source(spec: &SourceSpec, name: &CStr) -> Result<*mut sys::obs_ mod tests { use super::*; + #[test] + fn pipewire_stream_state_is_read_from_the_plugin_log() { + let line = r#"[pipewire] Stream 0x7fe128037440 state: "paused" (error: none)"#; + assert_eq!(pipewire_stream_state(line), Some("paused")); + assert_eq!( + pipewire_stream_state("[pipewire] Screencast session created"), + None + ); + } + #[test] fn segment_names_round_trip() { assert_eq!( diff --git a/crates/clipbin-obs/src/recorder.rs b/crates/clipbin-obs/src/recorder.rs index 34eef70..a8ba194 100644 --- a/crates/clipbin-obs/src/recorder.rs +++ b/crates/clipbin-obs/src/recorder.rs @@ -137,6 +137,7 @@ impl Engine { pipeline: Some(pipeline), events, current_segment, + segment_dir: config.segment_dir, }; recording .with_pipeline(|pipeline| pipeline.start()) @@ -177,6 +178,7 @@ pub struct Recording { pipeline: Option>, events: Receiver, current_segment: Arc>>, + segment_dir: PathBuf, } impl Recording { @@ -201,6 +203,7 @@ impl Recording { /// 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); + health.disk_available_bytes = fs4::available_space(&self.segment_dir).ok(); let segment = self .current_segment .lock() @@ -251,6 +254,9 @@ pub enum RecorderEvent { pub struct Health { pub source_width: u32, pub source_height: u32, + /// For PipeWire capture, whether its stream is delivering frames. `None` + /// for sources that report loss through their size instead. + pub source_streaming: Option, pub output_active: bool, pub total_frames: u64, pub dropped_frames: u64, @@ -258,6 +264,8 @@ pub struct Health { pub lagged_frames: u64, pub current_segment: Option, pub current_segment_bytes: u64, + /// Free space on the recording volume, if it could be read. + pub disk_available_bytes: Option, } /// Derives the status shown to the user from health samples and output events. @@ -268,6 +276,7 @@ pub struct Health { #[derive(Debug)] pub struct StatusTracker { status: RecordingStatus, + min_free_bytes: u64, last_written: (Option, u64), last_progress: Instant, } @@ -276,9 +285,14 @@ 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 { + /// Free space below which recording stops, so the last segment can still + /// be finalised rather than failing mid-write. + pub const DEFAULT_MIN_FREE_BYTES: u64 = 1 << 30; + + pub fn new(now: Instant, min_free_bytes: u64) -> Self { Self { status: RecordingStatus::Starting, + min_free_bytes, last_written: (None, 0), last_progress: now, } @@ -299,8 +313,22 @@ impl StatusTracker { self.last_written = written; self.last_progress = now; } - let has_picture = health.source_width > 0 && health.source_height > 0; + let has_picture = health.source_width > 0 + && health.source_height > 0 + && health.source_streaming != Some(false); let stalled = now.duration_since(self.last_progress) >= Self::STALL_LIMIT; + if let Some(available) = health.disk_available_bytes + && available < self.min_free_bytes + { + self.status = RecordingStatus::Failed { + reason: format!( + "not enough disk space: {} MB free, {} MB needed", + available / 1_000_000, + self.min_free_bytes / 1_000_000 + ), + }; + return &self.status; + } self.status = match (&self.status, has_picture, progressed || !stalled) { (_, true, true) @@ -323,8 +351,13 @@ impl StatusTracker { &self.status } + /// A failure stays visible after the output stops cleanly; it clears only + /// when a new recording starts. pub fn stopped(&mut self, reason: StopReason, last_error: Option<&str>) -> &RecordingStatus { self.status = match reason { + StopReason::Success if matches!(self.status, RecordingStatus::Failed { .. }) => { + return &self.status; + } StopReason::Success => RecordingStatus::Idle, reason => RecordingStatus::Failed { reason: match last_error { @@ -394,6 +427,7 @@ mod tests { Health { source_width: 1920, source_height: 1080, + source_streaming: None, output_active: true, total_frames: bytes / 1000, dropped_frames: 0, @@ -401,13 +435,31 @@ mod tests { lagged_frames: 0, current_segment: Some("segment-000000.mkv".into()), current_segment_bytes: bytes, + disk_available_bytes: Some(u64::MAX), } } + #[test] + fn low_disk_space_fails_before_the_disk_is_full() { + let start = Instant::now(); + let mut tracker = StatusTracker::new(start, 1000); + assert_eq!( + tracker.sample(&healthy(5000), start), + &RecordingStatus::Recording + ); + let low = Health { + disk_available_bytes: Some(999), + ..healthy(9000) + }; + assert!( + matches!(tracker.sample(&low, start), RecordingStatus::Failed { reason } if reason.contains("disk space")) + ); + } + #[test] fn footage_must_reach_disk_not_just_the_muxer() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start); + let mut tracker = StatusTracker::new(start, 0); let unwritten = Health { current_segment_bytes: 0, ..healthy(5000) @@ -435,7 +487,7 @@ mod tests { #[test] fn recording_needs_a_picture_and_written_footage() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start); + let mut tracker = StatusTracker::new(start, 0); let no_picture = Health { source_width: 0, source_height: 0, @@ -458,7 +510,7 @@ mod tests { #[test] fn a_stall_or_lost_source_fails_instead_of_pausing() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start); + let mut tracker = StatusTracker::new(start, 0); tracker.sample(&healthy(5000), start); let later = start + StatusTracker::STALL_LIMIT; assert!(matches!( @@ -466,7 +518,7 @@ mod tests { RecordingStatus::Failed { .. } )); - let mut tracker = StatusTracker::new(start); + let mut tracker = StatusTracker::new(start, 0); tracker.sample(&healthy(5000), start); let lost = Health { source_width: 0, @@ -483,20 +535,58 @@ mod tests { )); } + #[test] + fn a_stopped_pipewire_stream_is_a_lost_source_despite_its_size() { + let start = Instant::now(); + let mut tracker = StatusTracker::new(start, 0); + let streaming = Health { + source_streaming: Some(true), + ..healthy(5000) + }; + assert_eq!( + tracker.sample(&streaming, start), + &RecordingStatus::Recording + ); + let paused = Health { + source_streaming: Some(false), + ..healthy(9000) + }; + assert!(matches!( + tracker.sample(&paused, start), + RecordingStatus::Failed { .. } + )); + } + #[test] fn output_errors_fail_and_clean_stops_go_idle() { - let mut tracker = StatusTracker::new(Instant::now()); + let mut tracker = StatusTracker::new(Instant::now(), 0); assert!(matches!( tracker.stopped(StopReason::NoSpace, Some("disk full")), RecordingStatus::Failed { reason } if reason.contains("disk full") )); - let mut tracker = StatusTracker::new(Instant::now()); + let mut tracker = StatusTracker::new(Instant::now(), 0); assert_eq!( tracker.stopped(StopReason::Success, None), &RecordingStatus::Idle ); } + #[test] + fn stopping_after_a_failure_keeps_the_failure() { + let start = Instant::now(); + let mut tracker = StatusTracker::new(start, 0); + tracker.sample(&healthy(5000), start); + let lost = Health { + source_width: 0, + ..healthy(6000) + }; + tracker.sample(&lost, start); + assert!(matches!( + tracker.stopped(StopReason::Success, None), + RecordingStatus::Failed { .. } + )); + } + #[test] fn stop_codes_map_to_reasons() { assert_eq!(StopReason::from_code(0), StopReason::Success); diff --git a/crates/clipbin-probe/src/main.rs b/crates/clipbin-probe/src/main.rs index c909e77..d996a6c 100644 --- a/crates/clipbin-probe/src/main.rs +++ b/crates/clipbin-probe/src/main.rs @@ -7,12 +7,17 @@ //! clipbin-probe record [--target monitor|window|game] [--window OBS_WINDOW] //! [--restore-token TOKEN] [--codec h264|hevc|av1] [--encoder ID] //! [--seconds N] [--segment-seconds N] [--keyint N] [--bitrate KBPS] -//! [--width W --height H --fps F] [--no-audio] [--mic] [--out DIR] +//! [--width W --height H --fps F] [--no-audio] [--mic] +//! [--min-free-mb MB] [--out DIR] +//! clipbin-probe check [--codec h264|hevc|av1] [--segment-seconds N] [--out DIR] +//! +//! `check` inspects the segments `record` wrote, using the ffprobe and ffmpeg +//! command-line tools (install FFmpeg; on Windows, e.g. `winget install ffmpeg`). use std::{ env, - path::PathBuf, - process::ExitCode, + path::{Path, PathBuf}, + process::{Command, ExitCode}, sync::mpsc::RecvTimeoutError, time::{Duration, Instant}, }; @@ -38,6 +43,7 @@ fn main() -> ExitCode { let result = match command.as_str() { "encoders" => encoders(&options), "record" => record(&options), + "check" => check(&options), _ => return usage("expected a command"), }; match result { @@ -51,7 +57,7 @@ fn main() -> ExitCode { } fn usage(error: &str) -> ExitCode { - eprintln!("{error}\nusage: clipbin-probe encoders|record [options] (see src/main.rs)"); + eprintln!("{error}\nusage: clipbin-probe encoders|record|check [options] (see src/main.rs)"); ExitCode::from(2) } @@ -69,6 +75,7 @@ struct Options { format: VideoFormat, audio: bool, mic: bool, + min_free_mb: u64, out: PathBuf, } @@ -92,6 +99,7 @@ impl Options { }, audio: true, mic: false, + min_free_mb: StatusTracker::DEFAULT_MIN_FREE_BYTES / 1_000_000, out: env::temp_dir().join("clipbin-probe"), }; let mut args = args.iter(); @@ -118,6 +126,7 @@ impl Options { "--fps" => options.format.frames_per_second = number(value()?)? as u32, "--no-audio" => options.audio = false, "--mic" => options.mic = true, + "--min-free-mb" => options.min_free_mb = number(value()?)? as u64, "--out" => options.out = value()?.into(), _ => return Err(format!("unknown option {flag}")), } @@ -309,7 +318,7 @@ fn record(options: &Options) -> Result> { let started = Instant::now(); let elapsed = || format!("{:.3}", started.elapsed().as_secs_f64()); let recording = engine.start_recording(config)?; - let mut tracker = StatusTracker::new(started); + let mut tracker = StatusTracker::new(started, options.min_free_mb * 1_000_000); let mut status = RecordingStatus::Starting; println!("t={} status={}", elapsed(), status.label()); @@ -323,10 +332,11 @@ fn record(options: &Options) -> Result> { next_sample += Duration::from_secs(1); let health = recording.health(); println!( - "t={} health source={}x{} active={} frames={} dropped={} lagged={} muxer_bytes={} segment_bytes={}", + "t={} health source={}x{} streaming={:?} active={} frames={} dropped={} lagged={} muxer_bytes={} segment_bytes={}", elapsed(), health.source_width, health.source_height, + health.source_streaming, health.output_active, health.total_frames, health.dropped_frames, @@ -368,7 +378,7 @@ fn record(options: &Options) -> Result> { if let Some(token) = recording.restore_token() { println!("restore_token={token}"); } - Ok(outcome == Some(StopReason::Success)) + Ok(outcome == Some(StopReason::Success) && !matches!(status, RecordingStatus::Failed { .. })) } fn report_status(shown: &mut RecordingStatus, current: &RecordingStatus, elapsed: &str) { @@ -382,3 +392,128 @@ fn report_status(shown: &mut RecordingStatus, current: &RecordingStatus, elapsed } } } + +/// Checks the recorded segments: each starts with a keyframe in the chosen +/// codec and lasts about one segment length, and consecutive segments remux +/// into one MP4 across their boundaries without re-encoding and decode cleanly. +fn check(options: &Options) -> Result> { + let dir = options.out.join("session"); + let mut segments: Vec = std::fs::read_dir(&dir)? + .filter_map(|entry| entry.ok().map(|entry| entry.path())) + .filter(|path| { + path.extension() + .is_some_and(|ext| ext == clipbin_obs::SEGMENT_EXTENSION) + }) + .collect(); + segments.sort(); + println!("segments={}", segments.len()); + if segments.len() < 3 { + println!("check=failed need at least three segments"); + return Ok(false); + } + + let expected_codec = codec_key(options.codec); + let mut ok = true; + let last = segments.len() - 1; + for (index, segment) in segments.iter().enumerate() { + let name = segment.file_name().unwrap_or_default().to_string_lossy(); + let codec = ffprobe( + segment, + &[ + "-select_streams", + "v:0", + "-show_entries", + "stream=codec_name", + ], + )?; + let audio_tracks = ffprobe( + segment, + &["-select_streams", "a", "-show_entries", "stream=index"], + )? + .lines() + .count(); + let duration: f64 = ffprobe(segment, &["-show_entries", "format=duration"])? + .parse() + .unwrap_or(0.0); + let first_flags = ffprobe( + segment, + &[ + "-select_streams", + "v:0", + "-read_intervals", + "%+#1", + "-show_entries", + "packet=flags", + ], + )?; + let starts_on_keyframe = first_flags.starts_with('K'); + // The final segment is cut short by the stop request. + let full_length = index == last || duration >= options.segment_seconds as f64 * 0.9; + let segment_ok = + codec == expected_codec && starts_on_keyframe && full_length && audio_tracks > 0; + ok &= segment_ok; + println!( + "segment={name} ok={segment_ok} codec={codec} keyframe_start={starts_on_keyframe} duration={duration:.3} audio_tracks={audio_tracks}" + ); + } + + // A clip spanning the boundaries between the first three segments. + let list = options.out.join("concat.txt"); + let entries: String = segments[..3] + .iter() + .map(|segment| { + format!( + "file '{}'\n", + segment.to_string_lossy().replace('\'', "'\\''") + ) + }) + .collect(); + std::fs::write(&list, entries)?; + let clip = options.out.join("cross-boundary-clip.mp4"); + let remux = Command::new("ffmpeg") + .args(["-v", "error", "-y", "-f", "concat", "-safe", "0", "-i"]) + .arg(&list) + .args(["-map", "0", "-c", "copy", "-movflags", "+faststart"]) + .arg(&clip) + .status()?; + let clip_codec = ffprobe( + &clip, + &[ + "-select_streams", + "v:0", + "-show_entries", + "stream=codec_name", + ], + )?; + let clip_duration: f64 = ffprobe(&clip, &["-show_entries", "format=duration"])? + .parse() + .unwrap_or(0.0); + let decode = Command::new("ffmpeg") + .args(["-v", "error", "-xerror", "-i"]) + .arg(&clip) + .args(["-f", "null", "-"]) + .output()?; + let decode_errors = String::from_utf8_lossy(&decode.stderr).lines().count(); + let clip_ok = remux.success() + && clip_codec == expected_codec + && decode.status.success() + && decode_errors == 0; + ok &= clip_ok; + println!( + "clip ok={clip_ok} codec={clip_codec} duration={clip_duration:.3} decode_errors={decode_errors} path={}", + clip.display() + ); + println!("check={}", if ok { "ok" } else { "failed" }); + Ok(ok) +} + +fn ffprobe(path: &Path, query: &[&str]) -> Result> { + let output = Command::new("ffprobe") + .args(["-v", "error"]) + .args(query) + .args(["-of", "default=noprint_wrappers=1:nokey=1"]) + .arg(path) + .output() + .map_err(|error| format!("could not run ffprobe (is FFmpeg installed?): {error}"))?; + Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned()) +} diff --git a/scripts/rig/headless-wayland.sh b/scripts/rig/headless-wayland.sh index 16a1294..2d1af75 100755 --- a/scripts/rig/headless-wayland.sh +++ b/scripts/rig/headless-wayland.sh @@ -66,6 +66,11 @@ pipewire > "${rig_dir}/pipewire.log" 2>&1 & pids+=($!) wait_for pipewire '[[ -S "${XDG_RUNTIME_DIR}/pipewire-0" ]]' wireplumber > "${rig_dir}/wireplumber.log" 2>&1 & pids+=($!) pipewire-pulse > "${rig_dir}/pipewire-pulse.log" 2>&1 & pids+=($!) +wait_for pipewire-pulse '[[ -S "${XDG_RUNTIME_DIR}/pulse/native" ]]' +# No sound card here: route playback to a virtual sink whose monitor OBS +# records as desktop audio. +pactl load-module module-null-sink sink_name=clipbin-rig-sink > /dev/null +pactl set-default-sink clipbin-rig-sink WLR_BACKENDS=headless WLR_LIBINPUT_NO_DEVICES=1 WLR_HEADLESS_OUTPUTS=1 \ sway --config "${XDG_CONFIG_HOME}/sway/config" > "${rig_dir}/sway.log" 2>&1 & pids+=($!) @@ -77,18 +82,21 @@ export SWAYSOCK dbus-update-activation-environment WAYLAND_DISPLAY XDG_CURRENT_DESKTOP XDG_RUNTIME_DIR XDG_CONFIG_HOME SWAYSOCK /usr/lib/xdg-desktop-portal-wlr > "${rig_dir}/xdpw.log" 2>&1 & pids+=($!) +echo "${pids[-1]}" > "${rig_dir}/xdpw.pid" /usr/lib/xdg-desktop-portal -r > "${rig_dir}/xdp.log" 2>&1 & pids+=($!) -# A moving 60 FPS picture with a tone, standing in for a game. +# A moving 60 FPS picture with a 440 Hz tone, standing in for a game. SDL_VIDEODRIVER=wayland ffplay -loglevel error -fs -window_title clipbin-test-game \ - -f lavfi "testsrc2=size=${width}x${height}:rate=60,format=yuv420p" \ + -f lavfi "testsrc2=size=${width}x${height}:rate=60,format=yuv420p[out0];sine=frequency=440:sample_rate=48000[out1]" \ > "${rig_dir}/game.log" 2>&1 & pids+=($!) +echo "${pids[-1]}" > "${rig_dir}/game.pid" sleep 1 +export CLIPBIN_RIG_DIR="${rig_dir}" echo "rig: ready (WAYLAND_DISPLAY=${WAYLAND_DISPLAY}, logs in ${rig_dir})" >&2 status=0 "$@" || status=$? if [[ -n "${CLIPBIN_RIG_KEEP_LOGS:-}" ]]; then - cp -r "${rig_dir}" "${CLIPBIN_RIG_KEEP_LOGS}" + mkdir -p "${CLIPBIN_RIG_KEEP_LOGS}" && cp "${rig_dir}"/*.log "${CLIPBIN_RIG_KEEP_LOGS}/" fi exit "${status}"