From 263e7a964e515b23df252631d54a9c5fe5bf09eb Mon Sep 17 00:00:00 2001 From: Harry Date: Thu, 8 Oct 2026 14:19:57 +0200 Subject: [PATCH] feat: detect lost sources, stalled writes and low disk space PipeWire capture keeps its size and last frame after its stream stops, so the recorder follows the stream state OBS logs. Recording fails when the segment on disk stops growing (OBS kept counting bytes into a dead muxer and missed a full disk) and stops while enough space remains to finalise the last segment. Failures stay visible after the output stops. Adds a check command that verifies keyframe-aligned segments and a remux across segment boundaries, and gives the rig desktop audio. --- Cargo.lock | 11 +++ crates/clipbin-obs/Cargo.toml | 1 + crates/clipbin-obs/src/ffi.rs | 45 ++++++++- crates/clipbin-obs/src/recorder.rs | 106 ++++++++++++++++++-- crates/clipbin-probe/src/main.rs | 149 +++++++++++++++++++++++++++-- scripts/rig/headless-wayland.sh | 14 ++- 6 files changed, 307 insertions(+), 19 deletions(-) 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}"