harry/clipbin-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: detect lost sources, stalled writes and low disk space

Browse Source 
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.
This commit is contained in:
harry
2026-10-08 14:19:57 +02:00
parent f2eb7139da
commit 263e7a964e
6 changed files with 307 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clipbin-obs/Cargo.toml
+1
View File
@@ -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.
crates/clipbin-obs/src/ffi.rs
+44 -1
View File
@@ -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<Session, String> {
// 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<RecorderEvent>,
current_segment: Arc<Mutex<Option<String>>>,
) -> Result<Self, String> {
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!(
crates/clipbin-obs/src/recorder.rs
+98 -8
View File
@@ -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<Arc<Pipeline>>,
events: Receiver<RecorderEvent>,
current_segment: Arc<Mutex<Option<String>>>,
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<bool>,
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<String>,
pub current_segment_bytes: u64,
/// Free space on the recording volume, if it could be read.
pub disk_available_bytes: Option<u64>,
}
/// 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<String>, 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);