harry
435ddf2b17
Detect games with native process exit watchers, own OBS on a background worker, and persist settings and recoverable session timelines. Place segments by split frame counts, recover capture with honest gaps, and finalise on game exit or explicit Quit. Keep the current recording alive after GPUI closes its last window, then exit when the game ends. Add an end-to-end rig covering recording, recovery, exact packet-count offsets, disk library reload and clean Quit.
566 lines
22 KiB
Rust
566 lines
22 KiB
Rust
//! One worker owns OBS; the interface sends commands and reads cheap snapshots.
|
|
use clipbin_core::{
|
|
Library, RecordingStatus, Settings,
|
|
games::{self, ExitWatch, Game},
|
|
library::{Span, SpanKind},
|
|
store::{self, Metadata, Segment, Store, frame_offset, segment_offset},
|
|
};
|
|
use clipbin_obs::{
|
|
Engine, RecorderEvent, Recording, RecordingConfig, RuntimeLayout, StatusTracker, StopReason,
|
|
capture,
|
|
};
|
|
use std::{
|
|
sync::{
|
|
Arc, Mutex,
|
|
mpsc::{self, Receiver, Sender},
|
|
},
|
|
thread,
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
#[derive(Clone)]
|
|
pub struct Snapshot {
|
|
pub status: RecordingStatus,
|
|
pub game: Option<String>,
|
|
pub library: Library,
|
|
pub restore_token: Option<String>,
|
|
pub exited: bool,
|
|
}
|
|
#[derive(Clone)]
|
|
pub struct Client {
|
|
commands: Sender<Command>,
|
|
snapshot: Arc<Mutex<Snapshot>>,
|
|
}
|
|
enum Command {
|
|
Settings(Settings),
|
|
ChooseSource,
|
|
WindowClosed,
|
|
Quit,
|
|
}
|
|
impl Client {
|
|
pub fn start(
|
|
settings: Settings,
|
|
initial_error: Option<String>,
|
|
) -> (Self, thread::JoinHandle<()>) {
|
|
let (commands, receiver) = mpsc::channel();
|
|
let snapshot = Arc::new(Mutex::new(Snapshot {
|
|
status: initial_error.map_or(RecordingStatus::Idle, |reason| RecordingStatus::Failed {
|
|
reason,
|
|
}),
|
|
game: None,
|
|
library: Library::empty(settings.unprotected_session_limit_bytes),
|
|
restore_token: settings.portal_restore_token.clone(),
|
|
exited: false,
|
|
}));
|
|
let state = snapshot.clone();
|
|
let worker = thread::Builder::new()
|
|
.name("clipbin-recorder".into())
|
|
.spawn(move || {
|
|
if let Err(error) = worker(settings, receiver, &state) {
|
|
state.lock().unwrap().status = RecordingStatus::Failed { reason: error };
|
|
}
|
|
})
|
|
.expect("start recorder worker");
|
|
let client = Self { commands, snapshot };
|
|
(client, worker)
|
|
}
|
|
pub fn snapshot(&self) -> Snapshot {
|
|
self.snapshot.lock().unwrap().clone()
|
|
}
|
|
pub fn settings(&self, settings: Settings) {
|
|
let _ = self.commands.send(Command::Settings(settings));
|
|
}
|
|
pub fn choose_source(&self) {
|
|
let _ = self.commands.send(Command::ChooseSource);
|
|
}
|
|
pub fn window_closed(&self) {
|
|
let _ = self.commands.send(Command::WindowClosed);
|
|
}
|
|
pub fn quit(&self) {
|
|
if self.commands.send(Command::Quit).is_err() {
|
|
self.snapshot.lock().unwrap().exited = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
struct Active {
|
|
game: Game,
|
|
watch: ExitWatch,
|
|
metadata: Metadata,
|
|
started: Instant,
|
|
recording: Option<Recording>,
|
|
tracker: StatusTracker,
|
|
run: u32,
|
|
run_base: Option<Duration>,
|
|
span_base: usize,
|
|
last_frames: u64,
|
|
last_dropped: u64,
|
|
fps: u32,
|
|
stopping: bool,
|
|
ending: bool,
|
|
retry_at: Instant,
|
|
retries: u32,
|
|
}
|
|
impl Active {
|
|
fn start_run(
|
|
&mut self,
|
|
engine: &Engine,
|
|
settings: &Settings,
|
|
store: &Store,
|
|
) -> Result<(), String> {
|
|
let encoder = engine
|
|
.hardware_encoders(settings.codec)
|
|
.into_iter()
|
|
.next()
|
|
.ok_or_else(|| format!("No hardware {} encoder", settings.codec))?;
|
|
#[cfg(windows)]
|
|
let target = capture::CaptureTarget::Game {
|
|
window: capture::WindowMatch {
|
|
obs_window: format!("::{}", self.game.executable),
|
|
restore_token: None,
|
|
},
|
|
};
|
|
#[cfg(not(windows))]
|
|
let target = capture::CaptureTarget::Monitor {
|
|
monitor: capture::MonitorMatch {
|
|
obs_monitor: String::new(),
|
|
restore_token: settings.portal_restore_token.clone(),
|
|
},
|
|
};
|
|
#[cfg(windows)]
|
|
let video_source = target.video_source();
|
|
#[cfg(not(windows))]
|
|
let video_source = target.video_source(
|
|
capture::LinuxSession::detect().ok_or("No Wayland or X11 display for capture")?,
|
|
);
|
|
self.fps = settings.frames_per_second;
|
|
self.run += 1;
|
|
self.run_base = None;
|
|
self.span_base = self.metadata.session.spans.len();
|
|
self.last_frames = 0;
|
|
self.last_dropped = 0;
|
|
self.tracker = StatusTracker::new(
|
|
Instant::now(),
|
|
StatusTracker::DEFAULT_MIN_FREE_BYTES,
|
|
Duration::from_secs(2),
|
|
);
|
|
self.recording = Some(
|
|
engine
|
|
.start_recording(RecordingConfig {
|
|
frames_per_second: Some(settings.frames_per_second),
|
|
video_source: Some(video_source),
|
|
system_audio: settings
|
|
.system_audio_enabled
|
|
.then(capture::system_audio_source),
|
|
microphone: settings.microphone_enabled.then(capture::microphone_source),
|
|
resolution: settings.resolution,
|
|
video_encoder: encoder,
|
|
video_bitrate_kbps: 12_000,
|
|
keyframe_interval_seconds: 1,
|
|
segment_seconds: 2,
|
|
segment_dir: store
|
|
.directory(self.metadata.session.id)
|
|
.join(format!("run-{:03}", self.run)),
|
|
})
|
|
.map_err(|e| e.to_string())?,
|
|
);
|
|
self.stopping = false;
|
|
Ok(())
|
|
}
|
|
fn update_timeline(&mut self) {
|
|
let end = self.run_base.map_or(Duration::ZERO, |base| {
|
|
base + frame_offset(self.last_frames, self.fps)
|
|
});
|
|
self.metadata
|
|
.update_run(self.span_base, self.run_base, end, self.started.elapsed());
|
|
if let Some(segment) = self.metadata.segments.last_mut()
|
|
&& !segment.complete
|
|
{
|
|
segment.end = end.max(segment.start);
|
|
}
|
|
}
|
|
fn finish_run(&mut self, store: &Store) -> Result<(), String> {
|
|
let mut empty_tail = false;
|
|
if let Some(segment) = self.metadata.segments.last_mut()
|
|
&& !segment.complete
|
|
{
|
|
let path = store
|
|
.directory(self.metadata.session.id)
|
|
.join(&segment.path);
|
|
if !clipbin_core::segments::has_cluster(&path).map_err(|e| e.to_string())? {
|
|
eprintln!(
|
|
"Warning: {} has no media cluster; its tail is Missing",
|
|
path.display()
|
|
);
|
|
segment.available = false;
|
|
segment.end = segment.start;
|
|
self.last_frames = segment.frame_start;
|
|
match std::fs::remove_file(&path) {
|
|
Ok(()) => empty_tail = true,
|
|
Err(error) => {
|
|
eprintln!("Could not remove empty segment: {error}; marked unavailable")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if empty_tail {
|
|
self.metadata.segments.pop();
|
|
}
|
|
self.update_timeline();
|
|
if let Some(segment) = self.metadata.segments.last_mut()
|
|
&& !segment.complete
|
|
{
|
|
segment.complete = true;
|
|
segment.dropped_frames = self.last_dropped;
|
|
if self.last_dropped > 0 {
|
|
eprintln!(
|
|
"Warning: {} frames dropped in run {}; segment offsets may drift",
|
|
self.last_dropped, self.run
|
|
);
|
|
}
|
|
}
|
|
self.span_base = self.metadata.session.spans.len();
|
|
self.run_base = None;
|
|
self.recording = None;
|
|
self.stopping = false;
|
|
self.retry_at = Instant::now() + Duration::from_secs(5);
|
|
Ok(())
|
|
}
|
|
fn finish_session(&mut self) {
|
|
self.metadata.session.length = self.started.elapsed().max(self.metadata.session.length);
|
|
let end = self
|
|
.metadata
|
|
.session
|
|
.spans
|
|
.last()
|
|
.map_or(Duration::ZERO, |s| s.end);
|
|
if self.metadata.session.length > end {
|
|
self.metadata.session.spans.push(Span {
|
|
start: end,
|
|
end: self.metadata.session.length,
|
|
kind: SpanKind::Missing,
|
|
});
|
|
}
|
|
self.metadata.session.active = false;
|
|
}
|
|
}
|
|
|
|
fn fail(snapshot: &Arc<Mutex<Snapshot>>, reason: impl ToString) {
|
|
snapshot.lock().unwrap().status = RecordingStatus::Failed {
|
|
reason: reason.to_string(),
|
|
};
|
|
}
|
|
fn worker(
|
|
mut settings: Settings,
|
|
commands: Receiver<Command>,
|
|
snapshot: &Arc<Mutex<Snapshot>>,
|
|
) -> Result<(), String> {
|
|
let store = Store::discover().map_err(|e| e.to_string())?;
|
|
store.recover().map_err(|e| e.to_string())?;
|
|
snapshot.lock().unwrap().library = store
|
|
.library(settings.unprotected_session_limit_bytes)
|
|
.map_err(|e| e.to_string())?;
|
|
// An absent runtime is an interface-visible error; keep accepting commands.
|
|
let engine = RuntimeLayout::discover()
|
|
.map_err(|e| e.to_string())
|
|
.and_then(|layout| {
|
|
Engine::start(&layout, settings.frames_per_second).map_err(|e| e.to_string())
|
|
});
|
|
let engine = match engine {
|
|
Ok(engine) => Some(engine),
|
|
Err(e) => {
|
|
fail(snapshot, e);
|
|
None
|
|
}
|
|
};
|
|
let mut active: Option<Active> = None;
|
|
let mut quitting = false;
|
|
let mut finish_after_session = false;
|
|
let mut next_scan = Instant::now();
|
|
let mut next_poll = Instant::now();
|
|
let mut next_save = Instant::now();
|
|
let mut previous_status = None;
|
|
loop {
|
|
match commands.recv_timeout(Duration::from_millis(200)) {
|
|
Ok(Command::WindowClosed) => {
|
|
finish_after_session = true;
|
|
eprintln!("Window closed: finishing the current game session, then exiting");
|
|
}
|
|
Ok(Command::Quit) => {
|
|
quitting = true;
|
|
if let Some(a) = &mut active {
|
|
a.ending = true;
|
|
if !a.stopping
|
|
&& let Some(r) = &a.recording
|
|
{
|
|
r.stop();
|
|
a.stopping = true;
|
|
}
|
|
}
|
|
}
|
|
Ok(Command::Settings(new)) => {
|
|
settings = new;
|
|
snapshot.lock().unwrap().library.storage.budget_bytes =
|
|
settings.unprotected_session_limit_bytes;
|
|
if let Err(e) = store::save_settings(&settings) {
|
|
fail(snapshot, format!("Could not save settings: {e}"));
|
|
}
|
|
snapshot.lock().unwrap().restore_token = settings.portal_restore_token.clone();
|
|
}
|
|
Ok(Command::ChooseSource) => {
|
|
settings.portal_restore_token = None;
|
|
if let Err(e) = store::save_settings(&settings) {
|
|
fail(snapshot, e);
|
|
}
|
|
snapshot.lock().unwrap().restore_token = None;
|
|
if let Some(a) = &mut active {
|
|
a.retries = 0;
|
|
a.retry_at = Instant::now();
|
|
if !a.stopping
|
|
&& let Some(r) = &a.recording
|
|
{
|
|
r.stop();
|
|
a.stopping = true;
|
|
}
|
|
}
|
|
}
|
|
Err(mpsc::RecvTimeoutError::Disconnected) => quitting = true,
|
|
Err(mpsc::RecvTimeoutError::Timeout) => {}
|
|
}
|
|
if (quitting || finish_after_session) && active.is_none() {
|
|
break;
|
|
}
|
|
let now = Instant::now();
|
|
if active.is_none() && !quitting && now >= next_scan {
|
|
next_scan = now + Duration::from_secs(1);
|
|
match games::discover() {
|
|
Ok(games) => {
|
|
if let Some(game) = games.into_iter().next()
|
|
&& let Some(engine) = &engine
|
|
{
|
|
match ExitWatch::open(&game).and_then(|watch| {
|
|
store
|
|
.create(game.app_id, game.name.clone())
|
|
.map(|metadata| (watch, metadata))
|
|
}) {
|
|
Ok((watch, metadata)) => {
|
|
let mut a = Active {
|
|
game,
|
|
watch,
|
|
metadata,
|
|
started: now,
|
|
recording: None,
|
|
tracker: StatusTracker::new(now, 0, Duration::from_secs(2)),
|
|
run: 0,
|
|
run_base: None,
|
|
span_base: 0,
|
|
last_frames: 0,
|
|
last_dropped: 0,
|
|
fps: settings.frames_per_second,
|
|
stopping: false,
|
|
ending: false,
|
|
retry_at: now + Duration::from_secs(5),
|
|
retries: 0,
|
|
};
|
|
snapshot.lock().unwrap().game = Some(a.game.name.clone());
|
|
match a.start_run(engine, &settings, &store) {
|
|
Ok(()) => {
|
|
let mut state = snapshot.lock().unwrap();
|
|
if !matches!(state.status, RecordingStatus::Failed { .. }) {
|
|
state.status = RecordingStatus::Starting;
|
|
}
|
|
}
|
|
Err(e) => fail(snapshot, e),
|
|
}
|
|
active = Some(a);
|
|
}
|
|
Err(e) => fail(snapshot, e),
|
|
}
|
|
}
|
|
}
|
|
Err(e) => fail(snapshot, e),
|
|
}
|
|
}
|
|
if let Some(a) = &mut active {
|
|
// Process exit has priority over health: closing a game is normal.
|
|
match a.watch.exited() {
|
|
Ok(true) => a.ending = true,
|
|
Err(e) => {
|
|
fail(snapshot, format!("Could not watch game exit: {e}"));
|
|
a.ending = true;
|
|
}
|
|
Ok(false) => {}
|
|
}
|
|
if a.ending
|
|
&& !a.stopping
|
|
&& let Some(r) = &a.recording
|
|
{
|
|
r.stop();
|
|
a.stopping = true;
|
|
}
|
|
if !a.ending
|
|
&& a.recording.is_none()
|
|
&& a.retries < 3
|
|
&& now >= a.retry_at
|
|
&& let Some(engine) = &engine
|
|
{
|
|
a.retries += 1;
|
|
a.retry_at = now + Duration::from_secs(5);
|
|
if let Err(e) = a.start_run(engine, &settings, &store) {
|
|
fail(snapshot, e);
|
|
}
|
|
}
|
|
let mut stopped = false;
|
|
if let Some(r) = &a.recording {
|
|
if now >= next_poll {
|
|
next_poll = now + Duration::from_secs(1);
|
|
match r.poll() {
|
|
Ok(health) => {
|
|
a.last_frames = health.total_frames;
|
|
a.last_dropped = health.dropped_frames;
|
|
if health.output_active && a.run_base.is_none() {
|
|
a.run_base = Some(
|
|
a.started
|
|
.elapsed()
|
|
.saturating_sub(frame_offset(health.total_frames, a.fps)),
|
|
);
|
|
}
|
|
if !a.ending && !a.stopping {
|
|
let status = a.tracker.sample(&health, now).clone();
|
|
if status == RecordingStatus::Recording {
|
|
a.retries = 0;
|
|
}
|
|
// Keep a failure visible throughout recovery until footage is proven.
|
|
if status == RecordingStatus::Recording
|
|
|| !matches!(
|
|
snapshot.lock().unwrap().status,
|
|
RecordingStatus::Failed { .. }
|
|
)
|
|
{
|
|
snapshot.lock().unwrap().status = status.clone();
|
|
}
|
|
if matches!(status, RecordingStatus::Failed { .. }) {
|
|
r.stop();
|
|
a.stopping = true;
|
|
}
|
|
}
|
|
if let Some(token) = r.restore_token()
|
|
&& settings.portal_restore_token.as_ref() != Some(&token)
|
|
{
|
|
settings.portal_restore_token = Some(token.clone());
|
|
snapshot.lock().unwrap().restore_token = Some(token);
|
|
if let Err(e) = store::save_settings(&settings) {
|
|
fail(snapshot, e);
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
fail(snapshot, e);
|
|
if !a.stopping {
|
|
r.stop();
|
|
a.stopping = true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for event in r.events().try_iter() {
|
|
match event {
|
|
RecorderEvent::SegmentStarted {
|
|
path,
|
|
total_frames,
|
|
dropped_frames,
|
|
} => {
|
|
let base = *a.run_base.get_or_insert_with(|| {
|
|
a.started
|
|
.elapsed()
|
|
.saturating_sub(frame_offset(a.last_frames, a.fps))
|
|
});
|
|
let offset = base + segment_offset(total_frames, a.fps);
|
|
if let Some(previous) = a.metadata.segments.last_mut()
|
|
&& !previous.complete
|
|
{
|
|
previous.end = offset;
|
|
previous.complete = true;
|
|
previous.dropped_frames = dropped_frames;
|
|
if dropped_frames > 0 {
|
|
eprintln!(
|
|
"Warning: {dropped_frames} frames dropped in run {}; segment offsets may drift",
|
|
a.run
|
|
);
|
|
}
|
|
}
|
|
let path = std::path::PathBuf::from(path);
|
|
let relative = path
|
|
.strip_prefix(store.directory(a.metadata.session.id))
|
|
.map_err(|e| e.to_string())?
|
|
.to_path_buf();
|
|
a.metadata.segments.push(Segment {
|
|
path: relative,
|
|
start: offset,
|
|
end: offset,
|
|
complete: false,
|
|
available: true,
|
|
frame_start: total_frames.saturating_sub(1),
|
|
frames_per_second: a.fps,
|
|
dropped_frames: 0,
|
|
});
|
|
}
|
|
RecorderEvent::OutputStopped { code, last_error } => {
|
|
if !a.stopping && !a.ending {
|
|
fail(snapshot, "The recording output stopped unexpectedly");
|
|
}
|
|
if code != 0 {
|
|
snapshot.lock().unwrap().status = a
|
|
.tracker
|
|
.stopped(StopReason::from_code(code), last_error.as_deref())
|
|
.clone();
|
|
}
|
|
// Read final cumulative frame count after output finalisation.
|
|
if let Ok(health) = r.poll() {
|
|
a.last_frames = health.total_frames;
|
|
a.last_dropped = health.dropped_frames;
|
|
}
|
|
stopped = true;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
if stopped {
|
|
a.finish_run(&store)?;
|
|
}
|
|
a.update_timeline();
|
|
if now >= next_save || stopped || a.ending {
|
|
next_save = now + Duration::from_secs(1);
|
|
if a.ending && a.recording.is_none() {
|
|
a.finish_session();
|
|
store.save(&a.metadata).map_err(|e| e.to_string())?;
|
|
let mut state = snapshot.lock().unwrap();
|
|
if !matches!(state.status, RecordingStatus::Failed { .. }) {
|
|
state.status = RecordingStatus::Idle;
|
|
}
|
|
drop(state);
|
|
snapshot.lock().unwrap().game = None;
|
|
active = None;
|
|
} else {
|
|
store.save(&a.metadata).map_err(|e| e.to_string())?;
|
|
}
|
|
match store.library(settings.unprotected_session_limit_bytes) {
|
|
Ok(library) => snapshot.lock().unwrap().library = library,
|
|
Err(e) => fail(snapshot, e),
|
|
}
|
|
}
|
|
}
|
|
let status = snapshot.lock().unwrap().status.clone();
|
|
if previous_status.as_ref() != Some(&status) {
|
|
eprintln!("recorder_status={status:?}");
|
|
previous_status = Some(status);
|
|
}
|
|
}
|
|
drop(active);
|
|
drop(engine);
|
|
eprintln!("Recorder stopped; exiting");
|
|
snapshot.lock().unwrap().exited = true;
|
|
Ok(())
|
|
}
|