fix(rendering): stop exported camera frames running ahead on long recordings

crates/rendering/src/decoder/avassetreader.rs

@@ -537,7 +537,12 @@ impl AVAssetReaderDecoder {
         })
     }
 
-    fn select_best_decoder(&mut self, requested_time: f32, is_scrubbing: bool) -> (usize, bool) {
+    fn select_best_decoder(
+        &mut self,
+        requested_time: f32,
+        is_scrubbing: bool,
+        must_reset: impl Fn(&DecoderInstance) -> bool,
+    ) -> (usize, bool) {
         let decoder_count = self.decoders.len();
         let (best_id, _distance, needs_reset) = if is_scrubbing {
             self.pool_manager
@@ -552,6 +557,7 @@ impl AVAssetReaderDecoder {
         };
 
         let decoder_idx = best_id.min(decoder_count.saturating_sub(1));
+        let needs_reset = needs_reset || self.decoders.get(decoder_idx).is_some_and(must_reset);
 
         let max_decoder_count = self
             .pool_manager
@@ -660,7 +666,7 @@ impl AVAssetReaderDecoder {
             }
         };
         let mut this = None::<Self>;
-        let mut keyframe_index = None;
+        let mut keyframe_index: Option<Arc<cap_video_decode::avassetreader::KeyframeIndex>> = None;
 
         let init_result = DecoderInitResult {
             width,
@@ -694,6 +700,7 @@ impl AVAssetReaderDecoder {
             frame: u32,
             max_fallback_distance: u32,
             sender: oneshot::Sender<DecodedFrame>,
+            retried: bool,
         }
 
         let mut deferred_requests = VecDeque::<PendingRequest>::new();
@@ -751,6 +758,7 @@ impl AVAssetReaderDecoder {
                                 frame,
                                 max_fallback_distance,
                                 sender,
+                                retried: false,
                             });
                         }
                     }
@@ -769,6 +777,7 @@ impl AVAssetReaderDecoder {
                                     frame,
                                     max_fallback_distance,
                                     sender,
+                                    retried: false,
                                 });
                             }
                         }
@@ -851,6 +860,28 @@ impl AVAssetReaderDecoder {
             }
             pending_requests = still_unfulfilled;
 
+            // A track can start after the recording does. Before its first
+            // frame, that frame is the answer; serving it from the cache keeps
+            // the reader from stepping ahead on every such request.
+            let stream_start_frame = keyframe_index
+                .as_ref()
+                .and_then(|index| index.keyframes().first())
+                .map(|&(_, secs)| (secs * f64::from(fps)).round() as u32);
+            if let Some(start) = stream_start_frame
+                && let Some(first) = cache.get(&start).map(|cached| cached.data().clone())
+            {
+                let mut after_start = Vec::with_capacity(pending_requests.len());
+                for req in pending_requests.drain(..) {
+                    if req.frame < start {
+                        *last_sent_frame.borrow_mut() = Some(first.clone());
+                        let _ = req.sender.send(first.to_decoded_frame());
+                    } else {
+                        after_start.push(req);
+                    }
+                }
+                pending_requests = after_start;
+            }
+
             if pending_requests.is_empty() {
                 continue;
             }
@@ -890,7 +921,20 @@ impl AVAssetReaderDecoder {
             }
             let this = this.as_mut().expect("Video reader initialized above");
 
-            let (decoder_idx, was_reset) = this.select_best_decoder(requested_time, is_scrubbing);
+            // Pool positions are capped to the newest request, so a reader that
+            // decoded ahead can look usable for a request it has already passed.
+            // It can never vend that frame again, and reading on would answer
+            // with a stale or later frame, so it has to seek back first. Before
+            // the stream's first frame nothing better exists, so a seek is wasted.
+            let retrying = pending_requests.iter().any(|request| request.retried);
+            let before_stream_start =
+                stream_start_frame.is_some_and(|start| requested_frame < start);
+            let (decoder_idx, was_reset) =
+                this.select_best_decoder(requested_time, is_scrubbing, |decoder| {
+                    retrying
+                        || (!before_stream_start
+                            && reader_passed_request(decoder.prev_vended, requested_frame))
+                });
 
             let cache_min = if was_reset {
                 min_requested_frame.saturating_sub(FRAME_CACHE_SIZE as u32 * 2)
@@ -903,9 +947,23 @@ impl AVAssetReaderDecoder {
                 max_requested_frame + FRAME_CACHE_SIZE as u32
             };
 
+            // The last frame each other pooled reader vended is the answer for a
+            // request that lands in the gap after it, and the reader can never
+            // vend it again; keep it cached while requests alternate between
+            // readers.
+            let pinned_frames: Vec<u32> = this
+                .decoders
+                .iter()
+                .enumerate()
+                .filter(|&(index, _)| index != decoder_idx)
+                .filter_map(|(_, decoder)| decoder.prev_vended)
+                .collect();
+
             if was_reset {
                 *last_sent_frame.borrow_mut() = None;
-                cache.retain(|&f, _| f >= cache_min && f <= cache_max);
+                cache.retain(|f, _| {
+                    (*f >= cache_min && *f <= cache_max) || pinned_frames.contains(f)
+                });
                 cache_bytes = cache
                     .values()
                     .map(CachedFrame::byte_len)
@@ -923,6 +981,7 @@ impl AVAssetReaderDecoder {
             // reader re-vends from the keyframe at-or-before the request, and
             // that frame is the true VFR hold for requests inside a pts hole.
             let mut hold_candidate: Option<(u32, R<cv::ImageBuf>)> = None;
+            let mut retry_requests = Vec::new();
 
             {
                 let decoder = &mut this.decoders[decoder_idx];
@@ -950,7 +1009,8 @@ impl AVAssetReaderDecoder {
                     let current_frame =
                         pts_to_frame(frame.pts().value, Rational::new(1, frame.pts().scale), fps);
 
-                    if let Some(prev) = decoder.prev_vended
+                    let vended_before = decoder.prev_vended;
+                    if let Some(prev) = vended_before
                         && current_frame > prev + 1
                     {
                         record_pts_hole(&mut pts_holes, prev, current_frame);
@@ -1002,16 +1062,18 @@ impl AVAssetReaderDecoder {
                             )
                         {
                             if let Some(last_active) = &last_active_frame {
-                                let frame_to_remove = if requested_frame > *last_active {
-                                    *cache.keys().next().unwrap()
+                                let from_oldest = if requested_frame > *last_active {
+                                    true
                                 } else if requested_frame < *last_active {
-                                    *cache.keys().next_back().unwrap()
+                                    false
                                 } else {
-                                    let min = *cache.keys().min().unwrap();
-                                    let max = *cache.keys().max().unwrap();
-                                    if current_frame > max { min } else { max }
+                                    current_frame > *cache.keys().next_back().unwrap()
                                 };
-                                if let Some(removed) = cache.remove(&frame_to_remove) {
+                                let victim = cache_eviction_victim(&cache, from_oldest, |frame| {
+                                    Some(frame) == vended_before || pinned_frames.contains(&frame)
+                                });
+                                if let Some(removed) = victim.and_then(|frame| cache.remove(&frame))
+                                {
                                     cache_bytes = cache_bytes.saturating_sub(removed.byte_len());
                                 }
                             } else {
@@ -1065,6 +1127,26 @@ impl AVAssetReaderDecoder {
                                             .flatten()
                                             .max_by_key(|d| d.number);
 
+                                    // The reader vends every sample in pts order, so the
+                                    // frame it vended just before this one is the newest
+                                    // at or before the request. When that frame is no
+                                    // longer held anywhere, any other answer is stale or
+                                    // from the future: retry once from a fresh seek, which
+                                    // re-vends and keeps it.
+                                    if !req.retried
+                                        && answer_misses_previous_vend(
+                                            vended_before,
+                                            req.frame,
+                                            best_before.as_ref().map(|d| d.number),
+                                        )
+                                    {
+                                        retry_requests.push(PendingRequest {
+                                            retried: true,
+                                            ..req
+                                        });
+                                        continue;
+                                    }
+
                                     if let Some(data) = best_before {
                                         gap_hold = Some(data.clone());
                                         *last_sent_frame.borrow_mut() = Some(data.clone());
@@ -1105,6 +1187,10 @@ impl AVAssetReaderDecoder {
                 decoder.is_done = true;
             }
 
+            for request in retry_requests.into_iter().rev() {
+                deferred_requests.push_front(request);
+            }
+
             let last_frame_in_cache = cache.keys().max().copied().unwrap_or(0);
             if frames_iterated > 0 {
                 this.decoders[decoder_idx]
@@ -1278,6 +1364,45 @@ impl AVAssetReaderDecoder {
     }
 }
 
+fn reader_passed_request(prev_vended: Option<u32>, requested_frame: u32) -> bool {
+    prev_vended.is_some_and(|prev| prev >= requested_frame)
+}
+
+// A request the reader steps over is answered by the frame it vended just
+// before, the newest one at or before the request. Any other answer is stale.
+fn answer_misses_previous_vend(
+    vended_before: Option<u32>,
+    requested_frame: u32,
+    answer: Option<u32>,
+) -> bool {
+    vended_before
+        .filter(|&number| number <= requested_frame)
+        .is_some_and(|expected| answer != Some(expected))
+}
+
+// Drops from the end the cache is trimmed from, skipping pinned frames unless
+// every frame is pinned.
+fn cache_eviction_victim<V>(
+    cache: &BTreeMap<u32, V>,
+    from_oldest: bool,
+    is_pinned: impl Fn(u32) -> bool,
+) -> Option<u32> {
+    if from_oldest {
+        cache
+            .keys()
+            .copied()
+            .find(|&frame| !is_pinned(frame))
+            .or_else(|| cache.keys().next().copied())
+    } else {
+        cache
+            .keys()
+            .rev()
+            .copied()
+            .find(|&frame| !is_pinned(frame))
+            .or_else(|| cache.keys().next_back().copied())
+    }
+}
+
 fn should_grow_decoder_pool(
     is_scrubbing: bool,
     needs_reset: bool,
@@ -1357,6 +1482,44 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn a_reader_that_vended_the_request_cannot_serve_it_again() {
+        assert!(!reader_passed_request(None, 10));
+        assert!(!reader_passed_request(Some(9), 10));
+        assert!(reader_passed_request(Some(10), 10));
+        assert!(reader_passed_request(Some(57), 54));
+    }
+
+    #[test]
+    fn only_the_frame_vended_before_a_gap_answers_requests_inside_it() {
+        assert!(!answer_misses_previous_vend(Some(1054), 1055, Some(1054)));
+        assert!(answer_misses_previous_vend(Some(1054), 1055, Some(1052)));
+        assert!(answer_misses_previous_vend(Some(751), 752, Some(421)));
+        assert!(answer_misses_previous_vend(Some(751), 752, None));
+        assert!(!answer_misses_previous_vend(None, 752, Some(421)));
+        assert!(!answer_misses_previous_vend(Some(760), 752, Some(751)));
+    }
+
+    #[test]
+    fn cache_eviction_skips_pinned_frames() {
+        let cache: BTreeMap<u32, ()> = [415, 419, 421, 749, 751]
+            .into_iter()
+            .map(|f| (f, ()))
+            .collect();
+        assert_eq!(cache_eviction_victim(&cache, true, |_| false), Some(415));
+        assert_eq!(cache_eviction_victim(&cache, false, |_| false), Some(751));
+        assert_eq!(
+            cache_eviction_victim(&cache, false, |f| f == 751),
+            Some(749)
+        );
+        assert_eq!(cache_eviction_victim(&cache, true, |f| f < 420), Some(421));
+        assert_eq!(cache_eviction_victim(&cache, true, |_| true), Some(415));
+        assert_eq!(
+            cache_eviction_victim(&BTreeMap::<u32, ()>::new(), true, |_| false),
+            None
+        );
+    }
+
     #[test]
     fn decoder_pool_grows_only_for_unserved_non_sequential_access() {
         assert!(should_grow_decoder_pool(true, true, 1, 5));

crates/rendering/src/decoder/mod.rs

@@ -671,24 +671,30 @@ pub fn pts_to_frame(pts: i64, time_base: Rational, fps: u32) -> u32 {
 pub const FRAME_CACHE_SIZE: usize = 90;
 const DEFAULT_MAX_FALLBACK_DISTANCE: u32 = 90;
 
+/// Bounds the pts-hole map. Holes are facts about the file, and a pooled
+/// reader parked past a hole never re-vends it, so the map keeps every hole a
+/// realistic recording has and only evicts on pathological inputs.
+const MAX_TRACKED_HOLES: usize = 1 << 14;
+
 /// Records a pts hole discovered from a decode-order vend jump (frames vend
-/// in pts order, so a jump means no samples exist in between). The map stays
-/// bounded by dropping the narrowest hole — wide static-screen holds matter
-/// most.
+/// in pts order, so a jump means no samples exist in between). When the map
+/// is full it drops the hole farthest from the one just recorded: requests
+/// trail the reader, so the newest holes are the ones about to be asked for.
 pub(super) fn record_pts_hole(
     holes: &mut std::collections::BTreeMap<u32, u32>,
     start: u32,
     end: u32,
 ) {
-    const MAX_TRACKED_HOLES: usize = 64;
     holes.insert(start, end);
     if holes.len() > MAX_TRACKED_HOLES
-        && let Some(narrowest) = holes
-            .iter()
-            .min_by_key(|&(&s, &e)| e.saturating_sub(s))
-            .map(|(&s, _)| s)
+        && let (Some(&first), Some(&last)) = (holes.keys().next(), holes.keys().next_back())
     {
-        holes.remove(&narrowest);
+        let farthest = if start.abs_diff(first) >= start.abs_diff(last) {
+            first
+        } else {
+            last
+        };
+        holes.remove(&farthest);
     }
 }
 
@@ -1654,6 +1660,51 @@ pub async fn spawn_decoder(
 mod tests {
     use super::*;
 
+    #[test]
+    fn pts_hole_tracking_keeps_the_newest_holes_when_full() {
+        let mut holes = std::collections::BTreeMap::new();
+        let mut frame = 0u32;
+        for i in 0..(MAX_TRACKED_HOLES as u32 + 1000) {
+            let width = if i % 4 == 0 { 3 } else { 2 };
+            record_pts_hole(&mut holes, frame, frame + width);
+            assert_eq!(holes.get(&frame), Some(&(frame + width)));
+            frame += width;
+        }
+        assert_eq!(holes.len(), MAX_TRACKED_HOLES);
+        assert!(
+            holes
+                .keys()
+                .all(|&start| start as usize + 3 * MAX_TRACKED_HOLES >= frame as usize)
+        );
+    }
+
+    #[test]
+    fn pts_hole_tracking_drops_the_farthest_hole_after_a_backward_seek() {
+        let mut holes = std::collections::BTreeMap::new();
+        let last = MAX_TRACKED_HOLES as u32 - 1;
+        for i in 0..=last {
+            record_pts_hole(&mut holes, 1000 + i * 2, 1000 + i * 2 + 2);
+        }
+        record_pts_hole(&mut holes, 10, 40);
+        assert_eq!(holes.len(), MAX_TRACKED_HOLES);
+        assert_eq!(holes.get(&10), Some(&40));
+        assert!(holes.contains_key(&1000));
+        assert!(!holes.contains_key(&(1000 + last * 2)));
+    }
+
+    #[test]
+    fn pts_hole_tracking_keeps_an_old_static_hold_while_later_holes_arrive() {
+        let mut holes = std::collections::BTreeMap::new();
+        record_pts_hole(&mut holes, 1797, 2402);
+        let mut frame = 2402u32;
+        for i in 0..2000u32 {
+            let width = 2 + i % 7;
+            record_pts_hole(&mut holes, frame, frame + width);
+            frame += width;
+        }
+        assert_eq!(holes.get(&1797), Some(&2402));
+    }
+
     #[test]
     fn decoded_frame_storage_identity_matches_clones() {
         let frame = DecodedFrame::new_nv12(vec![0; 6], 2, 2, 2, 2);