From c784816615fb93e2ef8b982f823c3638d8337737 Mon Sep 17 00:00:00 2001 From: Skyler Lehmkuhl Date: Wed, 17 Jun 2026 14:30:32 -0400 Subject: [PATCH] Phase 2: bound video frame cache + stream the export mux - VideoManager.frame_cache: unbounded HashMap (grew per distinct frame during playback) -> LruCache evicted by a 256MB byte budget. Byte-budget rather than frame count is robust across resolutions (a 4K frame is ~33MB vs ~2MB at 800x600). unload_video pops per-clip keys (LruCache has no retain). - mux_video_and_audio: stream-merge the two inputs by PTS with one pending packet per stream (O(1) memory) instead of collecting every packet into Vecs first (O(duration)). Output is byte-identical. - export AAC: sanitize the planar-f32 path (non-finite -> 0, finite clamped to [-1,1]) like the integer paths, with a one-time warning. A stray NaN/Inf render sample no longer fails the whole export. Co-Authored-By: Claude Opus 4.8 (1M context) --- daw-backend/src/audio/export.rs | 29 +++- .../lightningbeam-core/src/video.rs | 61 +++++++-- .../lightningbeam-editor/src/export/mod.rs | 126 +++++++++--------- 3 files changed, 142 insertions(+), 74 deletions(-) diff --git a/daw-backend/src/audio/export.rs b/daw-backend/src/audio/export.rs index bdc940b..83d3858 100644 --- a/daw-backend/src/audio/export.rs +++ b/daw-backend/src/audio/export.rs @@ -671,14 +671,39 @@ fn convert_chunk_to_planar_i16(interleaved: &[f32], channels: u32) -> Vec Vec> { let num_frames = interleaved.len() / channels as usize; let mut planar = vec![vec![0.0f32; num_frames]; channels as usize]; + let mut non_finite = 0u64; for (i, chunk) in interleaved.chunks(channels as usize).enumerate() { for (ch, &sample) in chunk.iter().enumerate() { - planar[ch][i] = sample; + planar[ch][i] = if sample.is_finite() { + sample.clamp(-1.0, 1.0) + } else { + non_finite += 1; + 0.0 + }; + } + } + if non_finite > 0 { + // One-time warning: we sanitized rather than failed, but a non-finite + // sample reaching here means something upstream (an effect, automation, + // or a source decode) produced NaN/Inf — worth chasing if audio is wrong. + use std::sync::atomic::{AtomicBool, Ordering}; + static WARNED: AtomicBool = AtomicBool::new(false); + if !WARNED.swap(true, Ordering::Relaxed) { + eprintln!( + "⚠️ [EXPORT] sanitized {} non-finite (NaN/Inf) audio sample(s) in a chunk — \ + check effects/automation/source decode", + non_finite + ); } } diff --git a/lightningbeam-ui/lightningbeam-core/src/video.rs b/lightningbeam-ui/lightningbeam-core/src/video.rs index b2ad7cc..df1469c 100644 --- a/lightningbeam-ui/lightningbeam-core/src/video.rs +++ b/lightningbeam-ui/lightningbeam-core/src/video.rs @@ -458,9 +458,14 @@ pub struct VideoManager { /// Pool of video decoders, one per clip decoders: HashMap>>, - /// Frame cache: (clip_id, timestamp_ms) -> frame - /// Stores raw RGBA data for zero-copy rendering - frame_cache: HashMap<(Uuid, i64), Arc>, + /// Frame cache: (clip_id, timestamp_ms) -> frame. Stores decoded RGBA for + /// zero-copy rendering. Bounded by a **byte budget** (not a frame count, which + /// would be unsafe across resolutions — a 4K frame is ~33MB vs ~2MB at 800x600) + /// so playback of arbitrarily long video never grows unbounded. + frame_cache: LruCache<(Uuid, i64), Arc>, + /// Running total of bytes held in `frame_cache` (sum of each frame's RGBA len), + /// kept in sync on insert/evict/remove so eviction is O(1) per frame. + frame_cache_bytes: usize, /// Thumbnail cache: clip_id -> Vec of (timestamp, rgba_data) /// Low-resolution (64px width) thumbnails for scrubbing @@ -470,6 +475,11 @@ pub struct VideoManager { cache_size: usize, } +/// Byte budget for [`VideoManager::frame_cache`] (decoded full-resolution frames). +/// At ~2MB/frame (800x600) this holds ~128 frames; at ~33MB/frame (4K) ~8 — in +/// both cases enough for the current frame plus a scrub window, while bounding RAM. +const FRAME_CACHE_BYTE_BUDGET: usize = 256 * 1024 * 1024; + impl VideoManager { /// Create a new video manager with default cache size pub fn new() -> Self { @@ -480,7 +490,8 @@ impl VideoManager { pub fn with_cache_size(cache_size: usize) -> Self { Self { decoders: HashMap::new(), - frame_cache: HashMap::new(), + frame_cache: LruCache::unbounded(), + frame_cache_bytes: 0, thumbnail_cache: HashMap::new(), cache_size, } @@ -533,14 +544,16 @@ impl VideoManager { return Some(Arc::clone(cached_frame)); } - // Get decoder for this clip - let decoder_arc = self.decoders.get(clip_id)?; + // Get decoder for this clip. Clone the Arc so we don't hold a borrow of + // `self.decoders` across the `&mut self` cache insert below. + let decoder_arc = Arc::clone(self.decoders.get(clip_id)?); let mut decoder = decoder_arc.lock().ok()?; // Decode the frame let rgba_data = decoder.get_frame(timestamp).ok()?; let width = decoder.output_width; let height = decoder.output_height; + drop(decoder); // release the lock before touching `self` // Create VideoFrame and cache it let frame = Arc::new(VideoFrame { @@ -550,11 +563,29 @@ impl VideoManager { timestamp, }); - self.frame_cache.insert(cache_key, Arc::clone(&frame)); + self.cache_frame(cache_key, Arc::clone(&frame)); Some(frame) } + /// Insert a frame into the byte-budgeted cache, evicting least-recently-used + /// frames until the total is within [`FRAME_CACHE_BYTE_BUDGET`]. + fn cache_frame(&mut self, key: (Uuid, i64), frame: Arc) { + let bytes = frame.rgba_data.len(); + if let Some(old) = self.frame_cache.put(key, frame) { + self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(old.rgba_data.len()); + } + self.frame_cache_bytes += bytes; + // Keep at least one frame resident even if it alone exceeds the budget. + while self.frame_cache_bytes > FRAME_CACHE_BYTE_BUDGET && self.frame_cache.len() > 1 { + if let Some((_, evicted)) = self.frame_cache.pop_lru() { + self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(evicted.rgba_data.len()); + } else { + break; + } + } + } + /// Get the decoder Arc for a clip (for external thumbnail generation) /// This allows external code to decode frames without holding the VideoManager lock pub fn get_decoder(&self, clip_id: &Uuid) -> Option>> { @@ -614,8 +645,19 @@ impl VideoManager { pub fn unload_video(&mut self, clip_id: &Uuid) { self.decoders.remove(clip_id); - // Remove all cached frames for this clip - self.frame_cache.retain(|(id, _), _| id != clip_id); + // Remove all cached frames for this clip (LruCache has no retain; collect + // matching keys, then pop each, keeping the byte total in sync). + let keys: Vec<(Uuid, i64)> = self + .frame_cache + .iter() + .filter(|((id, _), _)| id == clip_id) + .map(|(k, _)| *k) + .collect(); + for key in keys { + if let Some(frame) = self.frame_cache.pop(&key) { + self.frame_cache_bytes = self.frame_cache_bytes.saturating_sub(frame.rgba_data.len()); + } + } // Remove thumbnails self.thumbnail_cache.remove(clip_id); @@ -624,6 +666,7 @@ impl VideoManager { /// Clear all frame caches (useful for memory management) pub fn clear_frame_cache(&mut self) { self.frame_cache.clear(); + self.frame_cache_bytes = 0; } } diff --git a/lightningbeam-ui/lightningbeam-editor/src/export/mod.rs b/lightningbeam-ui/lightningbeam-editor/src/export/mod.rs index c164e8f..a8fdc19 100644 --- a/lightningbeam-ui/lightningbeam-editor/src/export/mod.rs +++ b/lightningbeam-ui/lightningbeam-editor/src/export/mod.rs @@ -384,90 +384,90 @@ impl ExportOrchestrator { println!("🎵 [MUX] Audio stream - Input TB: {}/{}, Output TB: {}/{}", audio_input_tb.0, audio_input_tb.1, audio_output_tb.0, audio_output_tb.1); - // Collect all packets with their stream info and timestamps - let mut video_packets = Vec::new(); - for (stream, packet) in video_input.packets() { - if stream.index() == video_stream_index { - video_packets.push(packet); + // Stream-merge the two inputs by PTS, writing each packet as it's read — + // O(1) memory (one pending packet per stream) instead of collecting every + // packet first, so muxing a long export never grows unbounded. + let video_idx = video_stream_index; + let audio_idx = audio_stream_index; + let mut v_iter = video_input.packets(); + let mut a_iter = audio_input.packets(); + + // Pull the next packet belonging to the desired stream from each input. + let mut next_video = move || -> Option { + loop { + match v_iter.next() { + Some((stream, packet)) => { + if stream.index() == video_idx { + return Some(packet); + } + } + None => return None, + } } - } - - let mut audio_packets = Vec::new(); - for (stream, packet) in audio_input.packets() { - if stream.index() == audio_stream_index { - audio_packets.push(packet); + }; + let mut next_audio = move || -> Option { + loop { + match a_iter.next() { + Some((stream, packet)) => { + if stream.index() == audio_idx { + return Some(packet); + } + } + None => return None, + } } - } + }; - println!("🎬 [MUX] Collected {} video packets, {} audio packets", - video_packets.len(), audio_packets.len()); + let mut pending_v = next_video(); + let mut pending_a = next_audio(); + let mut v_count = 0usize; + let mut a_count = 0usize; + let mut log_count = 0; - // Report first and last timestamps - if !video_packets.is_empty() { - println!("🎬 [MUX] Video PTS range: {} to {}", - video_packets[0].pts().unwrap_or(0), - video_packets[video_packets.len()-1].pts().unwrap_or(0)); - } - if !audio_packets.is_empty() { - println!("🎵 [MUX] Audio PTS range: {} to {}", - audio_packets[0].pts().unwrap_or(0), - audio_packets[audio_packets.len()-1].pts().unwrap_or(0)); - } - - // Interleave packets by comparing timestamps in a common time base (use microseconds) - let mut v_idx = 0; - let mut a_idx = 0; - let mut interleave_log_count = 0; - - while v_idx < video_packets.len() || a_idx < audio_packets.len() { - let write_video = if v_idx >= video_packets.len() { - false // No more video - } else if a_idx >= audio_packets.len() { - true // No more audio, write video - } else { - // Compare timestamps - convert both to microseconds - let v_pts = video_packets[v_idx].pts().unwrap_or(0); - let a_pts = audio_packets[a_idx].pts().unwrap_or(0); - - // Convert to microseconds: pts * 1000000 * tb.num / tb.den - let v_us = v_pts * 1_000_000 * video_input_tb.0 as i64 / video_input_tb.1 as i64; - let a_us = a_pts * 1_000_000 * audio_input_tb.0 as i64 / audio_input_tb.1 as i64; - - v_us <= a_us // Write video if it comes before or at same time as audio + loop { + // Write whichever pending packet has the earlier PTS (in a common + // microsecond base); when one stream is exhausted, drain the other. + let write_video = match (&pending_v, &pending_a) { + (None, None) => break, + (Some(_), None) => true, + (None, Some(_)) => false, + (Some(v), Some(a)) => { + let v_us = v.pts().unwrap_or(0) * 1_000_000 * video_input_tb.0 as i64 + / video_input_tb.1 as i64; + let a_us = a.pts().unwrap_or(0) * 1_000_000 * audio_input_tb.0 as i64 + / audio_input_tb.1 as i64; + v_us <= a_us + } }; if write_video { - let mut packet = video_packets[v_idx].clone(); + let mut packet = pending_v.take().unwrap(); packet.set_stream(0); packet.rescale_ts(video_input_tb, video_output_tb); - - if interleave_log_count < 10 { - println!("🎬 [MUX] Writing V packet {} - PTS={:?}, DTS={:?}, Duration={:?}", - v_idx, packet.pts(), packet.dts(), packet.duration()); - interleave_log_count += 1; + if log_count < 10 { + println!("🎬 [MUX] Writing V packet - PTS={:?}, DTS={:?}", packet.pts(), packet.dts()); + log_count += 1; } - packet.write_interleaved(&mut output) .map_err(|e| format!("Failed to write video packet: {}", e))?; - v_idx += 1; + v_count += 1; + pending_v = next_video(); } else { - let mut packet = audio_packets[a_idx].clone(); + let mut packet = pending_a.take().unwrap(); packet.set_stream(1); packet.rescale_ts(audio_input_tb, audio_output_tb); - - if interleave_log_count < 10 { - println!("🎵 [MUX] Writing A packet {} - PTS={:?}, DTS={:?}, Duration={:?}", - a_idx, packet.pts(), packet.dts(), packet.duration()); - interleave_log_count += 1; + if log_count < 10 { + println!("🎵 [MUX] Writing A packet - PTS={:?}, DTS={:?}", packet.pts(), packet.dts()); + log_count += 1; } - packet.write_interleaved(&mut output) .map_err(|e| format!("Failed to write audio packet: {}", e))?; - a_idx += 1; + a_count += 1; + pending_a = next_audio(); } } - println!("🎬 [MUX] Wrote {} video packets, {} audio packets", v_idx, a_idx); + println!("🎬 [MUX] Wrote {} video packets, {} audio packets", v_count, a_count); // Write trailer output.write_trailer().map_err(|e| format!("Failed to write trailer: {}", e))?;