diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index e8aa1c71aed..7e57593da6b 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -16,6 +16,12 @@ use std::{ use tokio::sync::oneshot; use tokio_util::sync::{CancellationToken, DropGuard}; +/// Maximum in-memory cursor move events buffered before incremental disk flush. +/// Prevents unbounded Vec growth (5GB+ RSS on 2min+ high-DPI recordings). +const MAX_IN_MEMORY_CURSOR_MOVES: usize = 4_096; +/// Minimum Euclidean distance (in normalized coordinates) required to record a redundant cursor move. +const MIN_CURSOR_MOVE_DELTA_SQ: f64 = 1e-7; + #[derive(Clone)] pub struct Cursor { pub file_name: String, @@ -35,6 +41,33 @@ pub struct CursorActorResponse { pub keyboard_presses: Vec, } +impl CursorActorResponse { + #[inline] + pub fn push_move_deduped( + &mut self, + ev: CursorMoveEvent, + last_flushed: Option<&CursorMoveEvent>, + ) { + if let Some(last) = self.moves.last().or(last_flushed) { + let dx = ev.x - last.x; + let dy = ev.y - last.y; + if last.cursor_id == ev.cursor_id && (dx * dx + dy * dy) < MIN_CURSOR_MOVE_DELTA_SQ { + return; + } + } + if self.moves.len() >= MAX_IN_MEMORY_CURSOR_MOVES { + let half = self.moves.len() / 2; + let mut idx = 0; + self.moves.retain(|_| { + let keep = idx >= half || idx % 2 == 0; + idx += 1; + keep + }); + } + self.moves.push(ev); + } +} + pub struct CursorActor { stop: Option, stop_wakeup: Option>, @@ -270,32 +303,32 @@ struct IncrementalDelta { keyboard_presses: Vec, } -/// How many of the recorder's events were already handed to the incremental -/// files, so each flush clones only the newer ones on the sampling thread. +/// Drains buffered events from `CursorActorResponse` into each incremental +/// delta and tracks the last flushed cursor move so deduplication and in-memory +/// bounds hold across flushes. #[derive(Default)] struct FlushedEvents { - moves: usize, - clicks: usize, - keyboard_presses: usize, + last_move: Option, } impl FlushedEvents { fn take( &mut self, outputs: &IncrementalCaptureOutputs, - response: &CursorActorResponse, + response: &mut CursorActorResponse, ) -> IncrementalDelta { let mut delta = IncrementalDelta::default(); if outputs.cursor.is_some() { - delta.moves = response.moves[self.moves..].to_vec(); - delta.clicks = response.clicks[self.clicks..].to_vec(); + if let Some(last) = response.moves.last().cloned() { + self.last_move = Some(last); + } + delta.moves = std::mem::take(&mut response.moves); + delta.clicks = std::mem::take(&mut response.clicks); + response.moves.shrink_to(MAX_IN_MEMORY_CURSOR_MOVES); } if outputs.keyboard.is_some() { - delta.keyboard_presses = response.keyboard_presses[self.keyboard_presses..].to_vec(); + delta.keyboard_presses = std::mem::take(&mut response.keyboard_presses); } - self.moves = response.moves.len(); - self.clicks = response.clicks.len(); - self.keyboard_presses = response.keyboard_presses.len(); delta } } @@ -379,6 +412,28 @@ impl IncrementalFiles { self.append(delta); self.write(); } + + fn take_unpersisted(&mut self) -> IncrementalDelta { + let mut delta = IncrementalDelta::default(); + if self.outputs.cursor.is_some() && self.cursor_pending { + let json = [ + b"{\"clicks\":[" as &[u8], + &self.clicks_json, + b"],\"moves\":[", + &self.moves_json, + b"]}", + ] + .concat(); + if let Ok(events) = serde_json::from_slice::(&json) { + delta.clicks = events.clicks; + delta.moves = events.moves; + } + } + if self.outputs.keyboard.is_some() && self.keyboard_pending { + delta.keyboard_presses = std::mem::take(&mut self.keyboard.presses); + } + delta + } } fn append_json(buffer: &mut Vec, event: &impl serde::Serialize) { @@ -410,6 +465,7 @@ enum WriterMessage { struct CursorFileWriter { tx: std::sync::mpsc::Sender, pending_deltas: Arc, + unpersisted: Arc>, thread: std::thread::JoinHandle>, } @@ -420,6 +476,8 @@ impl CursorFileWriter { let (tx, rx) = std::sync::mpsc::channel::(); let pending_deltas = Arc::new(AtomicUsize::new(0)); let pending = pending_deltas.clone(); + let unpersisted = Arc::new(std::sync::Mutex::new(IncrementalDelta::default())); + let unpersisted_out = unpersisted.clone(); let thread = std::thread::Builder::new() .name("cursor-writer".to_string()) .spawn(move || { @@ -455,12 +513,19 @@ impl CursorFileWriter { failed_images.retain(|(path, image)| !save_cursor_image(path, image)); } } + if files.cursor_pending || files.keyboard_pending { + files.write(); + if let Ok(mut guard) = unpersisted_out.lock() { + *guard = files.take_unpersisted(); + } + } failed_images.retain(|(path, image)| !save_cursor_image(path, image)); failed_images.into_iter().map(|(path, _)| path).collect() })?; Ok(Self { tx, pending_deltas, + unpersisted, thread, }) } @@ -882,7 +947,7 @@ pub fn spawn_cursor_recorder( x, y, }; - response.moves.push(mouse_event); + response.push_move_deduped(mouse_event, flushed.last_move.as_ref()); } } @@ -916,8 +981,12 @@ pub fn spawn_cursor_recorder( ); last_keys = current_keys; - if has_incremental_outputs && last_flush.elapsed() >= flush_interval { - let delta = flushed.take(&incremental_outputs, &response); + let cursor_buffer_full = incremental_outputs.cursor.is_some() + && response.moves.len() >= MAX_IN_MEMORY_CURSOR_MOVES; + if has_incremental_outputs + && (last_flush.elapsed() >= flush_interval || cursor_buffer_full) + { + let delta = flushed.take(&incremental_outputs, &mut response); match &writer { Some(writer) => writer.flush(delta), None => inline_files.flush(delta), @@ -928,20 +997,30 @@ pub fn spawn_cursor_recorder( info!("cursor recorder done"); - let delta = has_incremental_outputs.then(|| flushed.take(&incremental_outputs, &response)); - match writer { + let delta = + has_incremental_outputs.then(|| flushed.take(&incremental_outputs, &mut response)); + let unpersisted = match writer { Some(writer) => { + let unpersisted = writer.unpersisted.clone(); let unsaved = writer.finish(delta); response.cursors.retain(|_, cursor| { !unsaved.iter().any(|path| path.ends_with(&cursor.file_name)) }); + unpersisted + .lock() + .map(|mut guard| std::mem::take(&mut *guard)) + .unwrap_or_default() } None => { if let Some(delta) = delta { inline_files.flush(delta); } + inline_files.take_unpersisted() } - } + }; + response.moves.extend(unpersisted.moves); + response.clicks.extend(unpersisted.clicks); + response.keyboard_presses.extend(unpersisted.keyboard_presses); let _ = tx.send(response); }); @@ -1955,7 +2034,10 @@ mod incremental_writer_tests { response.moves.extend(new.moves); response.clicks.extend(new.clicks); response.keyboard_presses.extend(new.keyboard_presses); - let taken = flushed.take(&outputs, &response); + let taken = flushed.take(&outputs, &mut response); + assert!(response.moves.is_empty()); + assert!(response.clicks.is_empty()); + assert!(response.keyboard_presses.is_empty()); handed_over.moves.extend(taken.moves); handed_over.clicks.extend(taken.clicks); handed_over.keyboard_presses.extend(taken.keyboard_presses); @@ -1965,16 +2047,21 @@ mod incremental_writer_tests { assert_eq!(handed_over.clicks, expected.clicks); assert_eq!(handed_over.keyboard_presses, expected.keyboard_presses); + response.moves = expected.moves.clone(); + response.clicks = expected.clicks.clone(); + response.keyboard_presses = expected.keyboard_presses.clone(); let mut cursor_only = FlushedEvents::default(); let taken = cursor_only.take( &IncrementalCaptureOutputs { cursor: outputs.cursor.clone(), keyboard: None, }, - &response, + &mut response, ); assert_eq!(taken.moves.len(), 95); + assert!(response.moves.is_empty()); assert!(taken.keyboard_presses.is_empty()); + assert_eq!(response.keyboard_presses.len(), expected.keyboard_presses.len()); } #[test] diff --git a/crates/recording/src/studio_recording.rs b/crates/recording/src/studio_recording.rs index 93fa6ea1e26..95705f95a03 100644 --- a/crates/recording/src/studio_recording.rs +++ b/crates/recording/src/studio_recording.rs @@ -140,7 +140,9 @@ fn persist_segment_input_events( keyboard_path: Option<&Path>, failure: &mut Option, ) -> (Cursors, u32) { - if let Some(path) = cursor_path { + if let Some(path) = cursor_path + && !(path.is_file() && response.clicks.is_empty() && response.moves.is_empty()) + { let result = serde_json::to_string_pretty(&CursorEvents { clicks: response.clicks, moves: response.moves, @@ -3998,6 +4000,30 @@ mod tests { assert_eq!(std::fs::read_dir(temp.path()).unwrap().count(), 0); } + #[test] + fn sidecar_preserves_incrementally_flushed_cursor_file_when_drained() { + let temp = tempfile::tempdir().unwrap(); + let cursor = temp.path().join("cursor.json"); + let flushed = CursorEvents { + clicks: Vec::new(), + moves: vec![cap_project::CursorMoveEvent { + active_modifiers: Vec::new(), + cursor_id: "0".into(), + time_ms: 42.0, + x: 0.5, + y: 0.5, + }], + }; + std::fs::write(&cursor, serde_json::to_vec(&flushed).unwrap()).unwrap(); + let mut response = sidecar_response(); + response.keyboard_presses.clear(); + let mut failure = None; + persist_segment_input_events(response, Some(&cursor), None, &mut failure); + assert!(failure.is_none()); + let saved = CursorEvents::load_from_file(&cursor).unwrap(); + assert_eq!(saved.moves, flushed.moves); + } + #[test] fn media_failure_requires_every_producer_stop_to_be_confirmed() { let error = classify_pipeline_stop_errors(