From 83ba5259d96dc6890afd4283254cc7892f6722c6 Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 15:57:53 -0500 Subject: [PATCH 1/6] Add constants for cursor move event managementperf(recording): deduplicate cursor moves and cap in-memory buffer (#109) Added constants to limit in-memory cursor move events and prevent redundant moves. --- crates/recording/src/cursor.rs | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index e8aa1c71aed..fd3b1c7cfdf 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,23 @@ pub struct CursorActorResponse { pub keyboard_presses: Vec, } +impl CursorActorResponse { + #[inline] + pub fn push_move_deduped(&mut self, ev: CursorMoveEvent) { + if let Some(last) = self.moves.last() { + 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 && self.moves.capacity() > MAX_IN_MEMORY_CURSOR_MOVES * 2 { + self.moves.shrink_to(MAX_IN_MEMORY_CURSOR_MOVES); + } + self.moves.push(ev); + } +} + pub struct CursorActor { stop: Option, stop_wakeup: Option>, From 8a8df72f0b933acae0be1dfd9716d584974b297e Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 16:18:23 -0500 Subject: [PATCH 2/6] Replace push with push_move_deduped for mouse events --- crates/recording/src/cursor.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index fd3b1c7cfdf..fa8c056f626 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -905,7 +905,7 @@ pub fn spawn_cursor_recorder( x, y, }; - response.moves.push(mouse_event); + response.push_move_deduped(mouse_event); } } From 1cd836e5b4b94ee098d16175e3faccf5ad74c60c Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 17:57:49 -0500 Subject: [PATCH 3/6] Update cursor.rs --- crates/recording/src/cursor.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index fa8c056f626..bf9551a3a4a 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -51,9 +51,6 @@ impl CursorActorResponse { return; } } - if self.moves.len() >= MAX_IN_MEMORY_CURSOR_MOVES && self.moves.capacity() > MAX_IN_MEMORY_CURSOR_MOVES * 2 { - self.moves.shrink_to(MAX_IN_MEMORY_CURSOR_MOVES); - } self.moves.push(ev); } } @@ -939,7 +936,11 @@ pub fn spawn_cursor_recorder( ); last_keys = current_keys; - if has_incremental_outputs && last_flush.elapsed() >= flush_interval { + if has_incremental_outputs + && (last_flush.elapsed() >= flush_interval + || response.moves.len().saturating_sub(flushed.moves) + >= MAX_IN_MEMORY_CURSOR_MOVES) + { let delta = flushed.take(&incremental_outputs, &response); match &writer { Some(writer) => writer.flush(delta), From 2896da3f02bc10aac582f9200b0bcf1a18684a59 Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Sat, 7 Nov 2026 20:26:37 -0500 Subject: [PATCH 4/6] perf(recording): drain flushed cursor moves and recover unpersisted deltas --- crates/recording/src/cursor.rs | 111 ++++++++++++++++++++++++++------- 1 file changed, 87 insertions(+), 24 deletions(-) diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index bf9551a3a4a..7f177535bf1 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -43,14 +43,27 @@ pub struct CursorActorResponse { impl CursorActorResponse { #[inline] - pub fn push_move_deduped(&mut self, ev: CursorMoveEvent) { - if let Some(last) = self.moves.last() { + 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); } } @@ -290,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 } } @@ -399,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) { @@ -430,6 +465,7 @@ enum WriterMessage { struct CursorFileWriter { tx: std::sync::mpsc::Sender, pending_deltas: Arc, + unpersisted: Arc>, thread: std::thread::JoinHandle>, } @@ -440,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 || { @@ -475,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, }) } @@ -902,7 +947,7 @@ pub fn spawn_cursor_recorder( x, y, }; - response.push_move_deduped(mouse_event); + response.push_move_deduped(mouse_event, flushed.last_move.as_ref()); } } @@ -936,12 +981,12 @@ pub fn spawn_cursor_recorder( ); last_keys = current_keys; + 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 - || response.moves.len().saturating_sub(flushed.moves) - >= MAX_IN_MEMORY_CURSOR_MOVES) + && (last_flush.elapsed() >= flush_interval || cursor_buffer_full) { - let delta = flushed.take(&incremental_outputs, &response); + let delta = flushed.take(&incremental_outputs, &mut response); match &writer { Some(writer) => writer.flush(delta), None => inline_files.flush(delta), @@ -952,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); }); @@ -1979,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); @@ -1989,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(), 95); } #[test] From c17da4cfaae2d75fc60d2135f4d2440fe26a2d7a Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Sat, 7 Nov 2026 20:31:33 -0500 Subject: [PATCH 5/6] perf(recording): preserve incrementally flushed cursor.json when response is drained --- crates/recording/src/studio_recording.rs | 28 +++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) 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( From 214375f6d3e3d6ccf26ec296d28faa3aff8a0ab1 Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Sat, 7 Nov 2026 20:37:35 -0500 Subject: [PATCH 6/6] test(recording): compare retained keyboard presses against expected count --- crates/recording/src/cursor.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index 7f177535bf1..7e57593da6b 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -2061,7 +2061,7 @@ mod incremental_writer_tests { assert_eq!(taken.moves.len(), 95); assert!(response.moves.is_empty()); assert!(taken.keyboard_presses.is_empty()); - assert_eq!(response.keyboard_presses.len(), 95); + assert_eq!(response.keyboard_presses.len(), expected.keyboard_presses.len()); } #[test]