diff --git a/crates/clipbin-obs/src/ffi.rs b/crates/clipbin-obs/src/ffi.rs index 30dba02..ba4adb6 100644 --- a/crates/clipbin-obs/src/ffi.rs +++ b/crates/clipbin-obs/src/ffi.rs @@ -129,7 +129,9 @@ unsafe extern "C" fn log_to_stderr( args: VaList, _: *mut c_void, ) { - if level >= sys::LOG_DEBUG as i32 && std::env::var_os("CLIPBIN_OBS_DEBUG").is_none() { + // LOG_* constants are u32 in the Linux bindings and i32 in the Windows ones. + let level = i64::from(level); + if level >= i64::from(sys::LOG_DEBUG) && std::env::var_os("CLIPBIN_OBS_DEBUG").is_none() { return; } let mut message = Dstr { @@ -141,9 +143,9 @@ unsafe extern "C" fn log_to_stderr( // allocates message.array with bmalloc, released with bfree. unsafe { dstr_vprintf(&mut message, format, args) }; let label = match level { - level if level <= sys::LOG_ERROR as i32 => "error", - level if level <= sys::LOG_WARNING as i32 => "warning", - level if level <= sys::LOG_INFO as i32 => "info", + level if level <= i64::from(sys::LOG_ERROR) => "error", + level if level <= i64::from(sys::LOG_WARNING) => "warning", + level if level <= i64::from(sys::LOG_INFO) => "info", _ => "debug", }; if let Some(text) = owned(message.array) { @@ -757,7 +759,7 @@ impl Pipeline { total_bytes: sys::obs_output_get_total_bytes(self.output), lagged_frames: sys::obs_get_lagged_frames() as u64, current_segment: None, - current_segment_bytes: 0, + written_bytes: 0, disk_available_bytes: None, } } @@ -824,11 +826,18 @@ impl Drop for Pipeline { /// # Safety /// Must run on the OBS thread. unsafe fn create_source(spec: &SourceSpec, name: &CStr) -> Result<*mut sys::obs_source_t, String> { + let id = cstring(spec.id); + // OBS creates a placeholder for unknown source types instead of failing. + // SAFETY: looks up a registered type by id. + if unsafe { sys::obs_source_get_display_name(id.as_ptr()) }.is_null() { + return Err(format!( + "the {} capture source is not available (its OBS plugin did not register it)", + spec.id + )); + } let settings = Data::new(&spec.settings); // SAFETY: arguments are valid; the caller owns the returned reference. - let source = unsafe { - sys::obs_source_create_private(cstring(spec.id).as_ptr(), name.as_ptr(), settings.0) - }; + let source = unsafe { sys::obs_source_create_private(id.as_ptr(), name.as_ptr(), settings.0) }; if source.is_null() { Err(format!( "OBS could not create the {} source (is its plugin loaded?)", diff --git a/crates/clipbin-obs/src/recorder.rs b/crates/clipbin-obs/src/recorder.rs index a8ba194..18951fa 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, + written: Mutex::default(), segment_dir: config.segment_dir, }; recording @@ -178,9 +179,17 @@ pub struct Recording { pipeline: Option>, events: Receiver, current_segment: Arc>>, + written: Mutex, segment_dir: PathBuf, } +/// Bytes on disk across this recording's segments. +#[derive(Default)] +struct Written { + current: Option, + finished_bytes: u64, +} + impl Recording { fn with_pipeline( &self, @@ -198,22 +207,29 @@ impl Recording { &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. + /// OBS's counters plus what reached disk. 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); health.disk_available_bytes = fs4::available_space(&self.segment_dir).ok(); - let segment = self + let current = 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); + .and_then(|path| path.clone()); + let size = |path: &str| std::fs::metadata(path).map_or(0, |metadata| metadata.len()); + if let Ok(mut written) = self.written.lock() { + if written.current != current { + // The previous segment is complete; count its final size. + if let Some(finished) = written.current.take() { + written.finished_bytes += size(&finished); + } + written.current = current.clone(); + } + health.written_bytes = written.finished_bytes + current.as_deref().map_or(0, size); } + health.current_segment = current; health } @@ -263,7 +279,8 @@ pub struct Health { pub total_bytes: u64, pub lagged_frames: u64, pub current_segment: Option, - pub current_segment_bytes: u64, + /// Bytes on disk across all of this recording's segments. + pub written_bytes: u64, /// Free space on the recording volume, if it could be read. pub disk_available_bytes: Option, } @@ -277,27 +294,34 @@ pub struct Health { pub struct StatusTracker { status: RecordingStatus, min_free_bytes: u64, - last_written: (Option, u64), + stall_limit: Duration, + last_written: 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); - /// 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 { + /// Matroska output can stay buffered until its segment closes, so footage + /// may appear on disk only once per segment. A stall is declared after two + /// segment lengths without growth. + pub fn new(now: Instant, min_free_bytes: u64, segment_length: Duration) -> Self { Self { status: RecordingStatus::Starting, min_free_bytes, - last_written: (None, 0), + stall_limit: (segment_length * 2 + Duration::from_secs(1)).max(Duration::from_secs(3)), + last_written: 0, last_progress: now, } } + /// How long footage may stop reaching disk before recording counts as failed. + pub fn stall_limit(&self) -> Duration { + self.stall_limit + } + pub fn status(&self) -> &RecordingStatus { &self.status } @@ -306,17 +330,15 @@ impl StatusTracker { 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; + let progressed = health.written_bytes > self.last_written; if progressed { - self.last_written = written; + self.last_written = health.written_bytes; self.last_progress = now; } 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; + let stalled = now.duration_since(self.last_progress) >= self.stall_limit; if let Some(available) = health.disk_available_bytes && available < self.min_free_bytes { @@ -332,7 +354,7 @@ impl StatusTracker { self.status = match (&self.status, has_picture, progressed || !stalled) { (_, true, true) - if health.output_active && health.total_frames > 0 && self.last_written.1 > 0 => + if health.output_active && health.total_frames > 0 && self.last_written > 0 => { RecordingStatus::Recording } @@ -342,7 +364,7 @@ impl StatusTracker { (RecordingStatus::Recording, true, false) => RecordingStatus::Failed { reason: format!( "no footage was written for {} seconds", - Self::STALL_LIMIT.as_secs() + self.stall_limit.as_secs() ), }, (_, false, _) => RecordingStatus::WaitingForCapture, @@ -423,6 +445,8 @@ impl std::error::Error for EngineError {} mod tests { use super::*; + const SEGMENT: Duration = Duration::from_secs(2); + fn healthy(bytes: u64) -> Health { Health { source_width: 1920, @@ -434,7 +458,7 @@ mod tests { total_bytes: bytes, lagged_frames: 0, current_segment: Some("segment-000000.mkv".into()), - current_segment_bytes: bytes, + written_bytes: bytes, disk_available_bytes: Some(u64::MAX), } } @@ -442,7 +466,7 @@ mod tests { #[test] fn low_disk_space_fails_before_the_disk_is_full() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start, 1000); + let mut tracker = StatusTracker::new(start, 1000, SEGMENT); assert_eq!( tracker.sample(&healthy(5000), start), &RecordingStatus::Recording @@ -459,9 +483,9 @@ mod tests { #[test] fn footage_must_reach_disk_not_just_the_muxer() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start, 0); + let mut tracker = StatusTracker::new(start, 0, SEGMENT); let unwritten = Health { - current_segment_bytes: 0, + written_bytes: 0, ..healthy(5000) }; assert_ne!( @@ -477,7 +501,7 @@ mod tests { total_bytes: 90_000, ..healthy(5000) }; - let later = start + StatusTracker::STALL_LIMIT; + let later = start + tracker.stall_limit(); assert!(matches!( tracker.sample(&stuck, later), RecordingStatus::Failed { .. } @@ -487,7 +511,7 @@ mod tests { #[test] fn recording_needs_a_picture_and_written_footage() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start, 0); + let mut tracker = StatusTracker::new(start, 0, SEGMENT); let no_picture = Health { source_width: 0, source_height: 0, @@ -510,15 +534,15 @@ mod tests { #[test] fn a_stall_or_lost_source_fails_instead_of_pausing() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start, 0); + let mut tracker = StatusTracker::new(start, 0, SEGMENT); tracker.sample(&healthy(5000), start); - let later = start + StatusTracker::STALL_LIMIT; + let later = start + tracker.stall_limit(); assert!(matches!( tracker.sample(&healthy(5000), later), RecordingStatus::Failed { .. } )); - let mut tracker = StatusTracker::new(start, 0); + let mut tracker = StatusTracker::new(start, 0, SEGMENT); tracker.sample(&healthy(5000), start); let lost = Health { source_width: 0, @@ -538,7 +562,7 @@ 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 mut tracker = StatusTracker::new(start, 0, SEGMENT); let streaming = Health { source_streaming: Some(true), ..healthy(5000) @@ -559,12 +583,12 @@ mod tests { #[test] fn output_errors_fail_and_clean_stops_go_idle() { - let mut tracker = StatusTracker::new(Instant::now(), 0); + let mut tracker = StatusTracker::new(Instant::now(), 0, SEGMENT); assert!(matches!( tracker.stopped(StopReason::NoSpace, Some("disk full")), RecordingStatus::Failed { reason } if reason.contains("disk full") )); - let mut tracker = StatusTracker::new(Instant::now(), 0); + let mut tracker = StatusTracker::new(Instant::now(), 0, SEGMENT); assert_eq!( tracker.stopped(StopReason::Success, None), &RecordingStatus::Idle @@ -574,7 +598,7 @@ mod tests { #[test] fn stopping_after_a_failure_keeps_the_failure() { let start = Instant::now(); - let mut tracker = StatusTracker::new(start, 0); + let mut tracker = StatusTracker::new(start, 0, SEGMENT); tracker.sample(&healthy(5000), start); let lost = Health { source_width: 0, diff --git a/crates/clipbin-probe/src/main.rs b/crates/clipbin-probe/src/main.rs index d996a6c..3982218 100644 --- a/crates/clipbin-probe/src/main.rs +++ b/crates/clipbin-probe/src/main.rs @@ -318,7 +318,11 @@ 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, options.min_free_mb * 1_000_000); + let mut tracker = StatusTracker::new( + started, + options.min_free_mb * 1_000_000, + Duration::from_secs(options.segment_seconds as u64), + ); let mut status = RecordingStatus::Starting; println!("t={} status={}", elapsed(), status.label()); @@ -332,7 +336,7 @@ fn record(options: &Options) -> Result> { next_sample += Duration::from_secs(1); let health = recording.health(); println!( - "t={} health source={}x{} streaming={:?} active={} frames={} dropped={} lagged={} muxer_bytes={} segment_bytes={}", + "t={} health source={}x{} streaming={:?} active={} frames={} dropped={} lagged={} muxer_bytes={} disk_bytes={}", elapsed(), health.source_width, health.source_height, @@ -342,7 +346,7 @@ fn record(options: &Options) -> Result> { health.dropped_frames, health.lagged_frames, health.total_bytes, - health.current_segment_bytes + health.written_bytes ); report_status(&mut status, tracker.sample(&health, now), &elapsed()); } diff --git a/scripts/rig/headless-wayland.sh b/scripts/rig/headless.sh similarity index 50% rename from scripts/rig/headless-wayland.sh rename to scripts/rig/headless.sh index 2d1af75..9023ddd 100755 --- a/scripts/rig/headless-wayland.sh +++ b/scripts/rig/headless.sh @@ -1,23 +1,38 @@ #!/usr/bin/env bash -# Run a command inside a disposable headless Wayland desktop for capture tests. +# Run a command inside a disposable headless desktop for capture tests. # -# scripts/rig/headless-wayland.sh [args...] +# scripts/rig/headless.sh [--x11] [args...] # -# Starts a private D-Bus session with PipeWire, WirePlumber, headless Sway and -# xdg-desktop-portal-wlr (auto-selecting the only output), plus a 60 FPS test -# pattern standing in for a game. Everything is torn down when the command exits. +# Starts a private D-Bus session with PipeWire, WirePlumber, a virtual audio +# sink, headless Sway on the GPU and xdg-desktop-portal-wlr (the portal +# auto-selects the only output). With --x11, the game and the command run as +# X11 clients of Sway's Xwayland instead, which has GPU rendering (DRI3) and an +# EWMH window manager like a real X11 desktop. A 60 FPS test pattern with a +# 440 Hz tone stands in for a game. Everything is torn down when the command +# exits. # -# This rig exercises the PipeWire/portal path on a machine without a display. -# It does not replace validation on real hardware (Hyprland, X11, Windows). +# The command sees CLIPBIN_RIG_DIR (logs, plus game.pid and xdpw.pid for the +# test game and the portal backend) and, with --x11, CLIPBIN_RIG_GAME_WINDOW +# (the test window's X id). Set CLIPBIN_RIG_KEEP_LOGS= to keep the logs. +# +# This exercises the capture paths on a machine without a display. It does not +# replace validation on real hardware (Hyprland, an X11 desktop, Windows). set -euo pipefail +mode=wayland +if [[ "${1:-}" == "--x11" ]]; then + mode=x11 + shift +fi if [[ $# -eq 0 ]]; then - echo "usage: $0 [args...]" >&2 + echo "usage: $0 [--x11] [args...]" >&2 exit 2 fi if [[ -z "${CLIPBIN_RIG_INNER:-}" ]]; then - exec env CLIPBIN_RIG_INNER=1 dbus-run-session -- "$0" "$@" + args=("$@") + [[ "${mode}" == x11 ]] && args=(--x11 "${args[@]}") + exec env CLIPBIN_RIG_INNER=1 dbus-run-session -- "$0" "${args[@]}" fi rig_dir="$(mktemp -d "${TMPDIR:-/tmp}/clipbin-rig.XXXXXX")" @@ -33,25 +48,14 @@ trap cleanup EXIT export XDG_RUNTIME_DIR="${rig_dir}/runtime" export XDG_CONFIG_HOME="${rig_dir}/config" -export XDG_CURRENT_DESKTOP=sway -export XDG_SESSION_TYPE=wayland -mkdir -p "${XDG_RUNTIME_DIR}" "${XDG_CONFIG_HOME}/xdg-desktop-portal-wlr" "${XDG_CONFIG_HOME}/sway" +export XDG_CURRENT_DESKTOP=sway XDG_SESSION_TYPE=wayland +mkdir -p "${XDG_RUNTIME_DIR}" "${XDG_CONFIG_HOME}/sway" "${XDG_CONFIG_HOME}/xdg-desktop-portal-wlr" chmod 700 "${XDG_RUNTIME_DIR}" +unset WAYLAND_DISPLAY DISPLAY width="${CLIPBIN_RIG_WIDTH:-1920}" height="${CLIPBIN_RIG_HEIGHT:-1080}" -cat > "${XDG_CONFIG_HOME}/sway/config" < "${XDG_CONFIG_HOME}/xdg-desktop-portal-wlr/config" < /dev/null pactl set-default-sink clipbin-rig-sink +printf '%s\n' \ + "output HEADLESS-1 resolution ${width}x${height}@60Hz" \ + "default_border none" \ + "xwayland force" \ + "exec echo \"\$DISPLAY\" > ${rig_dir}/xwayland-display" \ + > "${XDG_CONFIG_HOME}/sway/config" +printf '%s\n' \ + "[screencast]" \ + "chooser_type=none" \ + "output_name=${CLIPBIN_RIG_PORTAL_OUTPUT:-HEADLESS-1}" \ + "max_fps=60" \ + > "${XDG_CONFIG_HOME}/xdg-desktop-portal-wlr/config" + 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+=($!) wait_for sway 'compgen -G "${XDG_RUNTIME_DIR}/wayland-*" > /dev/null' WAYLAND_DISPLAY="$(basename "$(compgen -G "${XDG_RUNTIME_DIR}/wayland-*" | grep -v lock | head -1)")" -export WAYLAND_DISPLAY SWAYSOCK="$(compgen -G "${XDG_RUNTIME_DIR}/sway-ipc.*" | head -1)" -export SWAYSOCK - +export WAYLAND_DISPLAY 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+=($!) +video_driver=wayland + +if [[ "${mode}" == x11 ]]; then + wait_for Xwayland '[[ -s "${rig_dir}/xwayland-display" ]]' + DISPLAY="$(cat "${rig_dir}/xwayland-display")" + export DISPLAY XDG_SESSION_TYPE=x11 + wait_for Xwayland 'xwininfo -root > /dev/null 2>&1' + unset WAYLAND_DISPLAY + video_driver=x11 +fi # 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 \ +SDL_VIDEODRIVER="${video_driver}" ffplay -loglevel error -fs -window_title clipbin-test-game \ -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" +if [[ "${mode}" == x11 ]]; then + wait_for "the test window" 'xwininfo -name clipbin-test-game > /dev/null 2>&1' + CLIPBIN_RIG_GAME_WINDOW="$(xwininfo -name clipbin-test-game | awk '/Window id:/ { print strtonum($4) }')" + export CLIPBIN_RIG_GAME_WINDOW +fi sleep 1 export CLIPBIN_RIG_DIR="${rig_dir}" -echo "rig: ready (WAYLAND_DISPLAY=${WAYLAND_DISPLAY}, logs in ${rig_dir})" >&2 +echo "rig: ready (${mode}, logs in ${rig_dir})" >&2 status=0 "$@" || status=$? if [[ -n "${CLIPBIN_RIG_KEEP_LOGS:-}" ]]; then