M4.6: cancel frame jobs playback no longer needs (stale in-flight jobs, prefetch after Stop)

- filmcraft_media::cancel: a per-job cancellation flag passed through a thread-local, so the
  MediaSource API is unchanged. Frame workers run each job inside it.
- FrameServer::retain_queue (called by schedule_playback every refresh) now also cancels running
  jobs for frames the playhead has passed. Before, a worker blocked on a source's decoder lock
  went on to decode its frame after playback had moved 50–100 frames on, found it evicted and
  restarted the decoder at the keyframe (decoder trace: "seek i=32 key=0 next=142").
- GopCache gives up (CodecError/MediaError::Cancelled) before seeking or decoding for a cancelled
  job and between samples; decoder state stays consistent, so the next request carries on.
- The CPU compositor checks the flag between tracks and effects; cancelled results are never
  cached.
- Stop drops queued prefetch jobs and cancels running ones (FrameServer::stop_prefetch). They
  used to run to completion after Stop, e.g. 14 frames of live effects rendering competing with
  a Render In to Out started right after.

bench, same-session A/B (GPU path, 8 s, load avg 35–41):
  after-preview Full live play (1080p + Lumetri/Sharpen/Levels/Tint, CPU-bound):
    38/154 → 98/94 shown/dropped, 297 → 267 ms CPU/frame, 84 jobs cancelled
  after-preview plays from previews, h264-1080, stack3, h264-2160: unchanged within noise

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/codecs/src/gop.rs

@@ -172,6 +172,11 @@ impl GopCache {
             st.decoder = Some(d);
             st.next = usize::MAX;
         }
+        // Nothing cached: the work ahead (possibly a seek and a GOP of decoding) is only worth it
+        // while someone still wants the frame.
+        if filmcraft_media::cancel::cancelled() {
+            return Err(CodecError::Cancelled);
+        }
         if st.intra {
             return self.intra_frame(st, s, i, want_pts);
         }
@@ -196,6 +201,10 @@ impl GopCache {
         }
         let limit = (i.max(st.next) + 64).min(n);
         while st.next < limit {
+            if filmcraft_media::cancel::cancelled() {
+                // The decoder state stays consistent (`next`, `out_max`): a later request continues.
+                return Err(CodecError::Cancelled);
+            }
             let k = st.next;
             let data = s.read(k)?;
             let out = st.decoder.as_mut().expect("decoder").decode(&data, s.pts(k))?;
@@ -385,4 +394,20 @@ mod tests {
         assert!(s.decodes.load(Ordering::Relaxed) <= 64 + 4 * 4);
         assert_eq!(s.resets.load(Ordering::Relaxed), 0);
     }
+
+    #[test]
+    fn cancelled_request_stops_decoding_and_keeps_state() {
+        let s = samples(300, 250, 0, false);
+        let c = GopCache::new(None);
+        assert_eq!(index_of(&c.frame(&s, 5_000).expect("frame")), 5);
+        let flag = Arc::new(std::sync::atomic::AtomicBool::new(true));
+        let r = filmcraft_media::cancel::with_cancel(&flag, || c.frame(&s, 200_000));
+        assert!(matches!(r, Err(CodecError::Cancelled)));
+        assert_eq!(s.decodes.load(Ordering::Relaxed), 6);
+        // cached frames are still served, and the decoder carries on from where it was
+        let r = filmcraft_media::cancel::with_cancel(&flag, || c.frame(&s, 3_000));
+        assert_eq!(index_of(&r.expect("cached")), 3);
+        assert_eq!(index_of(&c.frame(&s, 7_000).expect("frame")), 7);
+        assert_eq!(s.resets.load(Ordering::Relaxed), 1);
+    }
 }

crates/codecs/src/lib.rs

@@ -32,6 +32,9 @@ pub enum CodecError {
     Decode(String),
     #[error("container: {0}")]
     Container(String),
+    /// The frame is no longer wanted (`filmcraft_media::cancel`).
+    #[error("cancelled")]
+    Cancelled,
 }
 
 pub type Result<T> = std::result::Result<T, CodecError>;
@@ -40,6 +43,7 @@ impl From<CodecError> for filmcraft_media::MediaError {
     fn from(e: CodecError) -> Self {
         match e {
             CodecError::Unsupported(s) => filmcraft_media::MediaError::Unsupported(s),
+            CodecError::Cancelled => filmcraft_media::MediaError::Cancelled,
             other => filmcraft_media::MediaError::Decode(other.to_string()),
         }
     }

crates/media/src/cancel.rs

@@ -0,0 +1,45 @@
+//! Cooperative cancellation of frame requests.
+//!
+//! A frame worker runs a job inside [`with_cancel`]; sources that do long work for a request (a
+//! decoder seeking back to a keyframe and decoding forward) poll [`cancelled`] and give up with
+//! [`MediaError::Cancelled`](crate::MediaError::Cancelled) once the job is no longer wanted, e.g.
+//! playback has moved past its frame. The flag travels through a thread-local, so the
+//! [`MediaSource`](crate::MediaSource) API is unchanged.
+
+use std::cell::RefCell;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+thread_local! {
+    static CURRENT: RefCell<Option<Arc<AtomicBool>>> = const { RefCell::new(None) };
+}
+
+/// Run `f` with `flag` as this thread's cancellation flag (restoring the previous one after).
+pub fn with_cancel<R>(flag: &Arc<AtomicBool>, f: impl FnOnce() -> R) -> R {
+    let prev = CURRENT.with(|c| c.replace(Some(flag.clone())));
+    let r = f();
+    CURRENT.with(|c| *c.borrow_mut() = prev);
+    r
+}
+
+/// Whether the work running on this thread has been cancelled.
+pub fn cancelled() -> bool {
+    CURRENT.with(|c| c.borrow().as_ref().is_some_and(|f| f.load(Ordering::Relaxed)))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn flag_is_scoped_to_the_closure() {
+        let flag = Arc::new(AtomicBool::new(false));
+        assert!(!cancelled());
+        with_cancel(&flag, || {
+            assert!(!cancelled());
+            flag.store(true, Ordering::Relaxed);
+            assert!(cancelled());
+        });
+        assert!(!cancelled());
+    }
+}

crates/media/src/lib.rs

@@ -8,6 +8,7 @@
 //! Time passed to a source is **media time** (0 = first frame of the media), in ticks.
 
 pub mod cache;
+pub mod cancel;
 pub mod digits;
 pub mod generators;
 pub mod still;
@@ -36,6 +37,9 @@ pub enum MediaError {
     NoStream(&'static str),
     #[error("media offline: {0}")]
     Offline(String),
+    /// The request was cancelled (see [`cancel`]).
+    #[error("cancelled")]
+    Cancelled,
 }
 
 pub type Result<T> = std::result::Result<T, MediaError>;

crates/render/src/lib.rs

@@ -75,6 +75,10 @@ fn render_seq(project: &Project, seq: &Sequence, t: Tick, opts: RenderOptions, s
     }
     let tc = format_time(t, seq.settings.frame_rate, seq.settings.drop_frame, TimeDisplay::Timecode, seq.settings.sample_rate as i64);
     for track in &seq.video_tracks {
+        // A cancelled frame job (playback moved on) stops here; its result is discarded.
+        if filmcraft_media::cancel::cancelled() {
+            return canvas;
+        }
         if !track.enabled {
             continue;
         }
@@ -240,6 +244,9 @@ pub(crate) fn item_layer(
     if opts.effects {
         let cx = effects::FxCtx { t: mt, px_scale, seconds: (t - item.start).seconds(), timecode: tc, clip_name: &item.name };
         for e in item.effects.iter().filter(|e| e.def().is_some_and(|d| !d.intrinsic)) {
+            if filmcraft_media::cancel::cancelled() {
+                return None;
+            }
             effects::apply(&mut layer, e, &cx);
         }
     }

crates/ui-egui/examples/bench_playback.rs

@@ -230,6 +230,7 @@ struct PlayReport {
     ui_cpu: f64,
     jobs: usize,
     wasted_jobs: usize,
+    cancelled_jobs: usize,
     preview_jobs: usize,
     gop: filmcraft_codecs::GopStats,
     uploaded_mb: f64,
@@ -306,8 +307,9 @@ impl Bench {
         }
         meter.finish();
         let wall = anchor.elapsed().as_secs_f64();
-        // Let in-flight jobs finish so their CPU is counted, then drop the rest of the queue.
-        self.server.retain_queue(|_| false);
+        // Stop as the app does (drop and cancel prefetch), letting running jobs wind down so their
+        // CPU is counted.
+        self.server.stop_prefetch();
         std::thread::sleep(Duration::from_millis(50));
         let process_cpu = ms(process_cpu_time().unwrap_or_default().saturating_sub(p0));
         let recs: Vec<JobRecord> = self.server.take_records();
@@ -356,6 +358,9 @@ impl Bench {
             if r.preview {
                 rep.preview_jobs += 1;
             }
+            if r.cancelled {
+                rep.cancelled_jobs += 1;
+            }
         }
         rep.present = display.present.iter().map(|d| ms(*d)).collect();
         rep
@@ -422,7 +427,7 @@ impl Bench {
 
 fn print_header() {
     println!(
-        "{:<34} {:>6} {:>5} {:>5} {:>7} | {:>6} {:>6} {:>6} | {:>6} {:>6} | {:>6} {:>6} {:>6} | {:>6} {:>6} | {:>7} {:>6} | {:>5} {:>5} {:>5} | load",
+        "{:<34} {:>6} {:>5} {:>5} {:>7} | {:>6} {:>6} {:>6} | {:>6} {:>6} | {:>6} {:>6} {:>6} | {:>6} {:>6} | {:>7} {:>6} | {:>5} {:>5} {:>5} {:>4} | load",
         "scenario",
         "wall s",
         "shown",
@@ -442,7 +447,8 @@ fn print_header() {
         "cores",
         "seeks",
         "dec",
-        "waste"
+        "waste",
+        "canc"
     );
 }
 
@@ -450,7 +456,7 @@ fn print_row(r: &PlayReport, fps: f64) {
     let frames = (r.shown + r.dropped).max(1) as f64;
     let cpu_per_frame = r.process_cpu / frames;
     println!(
-        "{:<34} {:>6.2} {:>5} {:>5} {:>6.0}% | {:>6.1} {:>6.1} {:>6.1} | {:>6.1} {:>6.1} | {:>6.1} {:>6.1} {:>6.1} | {:>6.2} {:>6.2} | {:>7.1} {:>6.2} | {:>5} {:>5} {:>5} | {}",
+        "{:<34} {:>6.2} {:>5} {:>5} {:>6.0}% | {:>6.1} {:>6.1} {:>6.1} | {:>6.1} {:>6.1} | {:>6.1} {:>6.1} {:>6.1} | {:>6.2} {:>6.2} | {:>7.1} {:>6.2} | {:>5} {:>5} {:>5} {:>4} | {}",
         r.label,
         r.seconds,
         r.shown,
@@ -471,6 +477,7 @@ fn print_row(r: &PlayReport, fps: f64) {
         r.gop.seeks,
         r.gop.decoded,
         r.wasted_jobs,
+        r.cancelled_jobs,
         r.load
     );
 }
@@ -488,7 +495,7 @@ fn to_json(r: &PlayReport, scenario: &str, res: &str, path: &str) -> Value {
         "source_cpu_ms_mean": r.src_cpu.iter().sum::<f64>() / r.jobs.max(1) as f64,
         "present_ms": {"p50": pct(&r.present, 0.5), "p95": pct(&r.present, 0.95), "max": pct(&r.present, 1.0)},
         "ui_cpu_ms": r.ui_cpu, "process_cpu_ms": r.process_cpu, "cpu_ms_per_frame": r.process_cpu / frames,
-        "jobs": r.jobs, "wasted_jobs": r.wasted_jobs, "preview_jobs": r.preview_jobs, "uploaded_mb": r.uploaded_mb,
+        "jobs": r.jobs, "wasted_jobs": r.wasted_jobs, "cancelled_jobs": r.cancelled_jobs, "preview_jobs": r.preview_jobs, "uploaded_mb": r.uploaded_mb,
         "gop": {"hits": r.gop.hits, "misses": r.gop.misses, "seeks": r.gop.seeks, "decoded": r.gop.decoded, "evicted": r.gop.evicted},
         "loadavg": r.load,
     })
@@ -617,7 +624,7 @@ fn main() {
         load_avg()
     );
     println!(
-        "lat = queue→ready per job, svc = worker time per job, cpu/j = worker thread CPU, src/j = source fetch (decode) wall, srcC/j = its thread CPU,\nui = present time on the UI thread (GPU upload+draw / texture conversion), cpu ms/f = process CPU per due frame, cores = CPU cores that rate needs at the sequence fps,\nseeks/dec = decoder restarts / samples decoded, waste = finished jobs for frames never due (all times ms)\n"
+        "lat = queue→ready per job, svc = worker time per job, cpu/j = worker thread CPU, src/j = source fetch (decode) wall, srcC/j = its thread CPU,\nui = present time on the UI thread (GPU upload+draw / texture conversion), cpu ms/f = process CPU per due frame, cores = CPU cores that rate needs at the sequence fps,\nseeks/dec = decoder restarts / samples decoded, waste = finished jobs for frames never due, canc = jobs cancelled while running (all times ms)\n"
     );
     print_header();
     let mut out = Vec::new();

crates/ui-egui/src/frames.rs

@@ -46,6 +46,10 @@ pub struct Rgba {
 struct Job {
     key: FrameKey,
     queued: Instant,
+    /// Set when the job is no longer wanted; sources poll it (`filmcraft_media::cancel`).
+    cancel: Arc<std::sync::atomic::AtomicBool>,
+    /// Queued by playback prefetch (dropped when playback stops).
+    prefetch: bool,
     time: Tick,
     /// Output scale relative to the target's frame size.
     scale: f32,
@@ -58,7 +62,8 @@ struct Shared {
     cv: Condvar,
     done: Mutex<Cache>,
     plans: Mutex<HashMap<FrameKey, (Arc<filmcraft_render::plan::FramePlan>, u64)>>,
-    in_flight: Mutex<Vec<FrameKey>>,
+    /// Running jobs: key, cancel flag, prefetch.
+    in_flight: Mutex<Vec<(FrameKey, Arc<std::sync::atomic::AtomicBool>, bool)>>,
     /// Per-job timings, collected while profiling is on (benchmarks, `ui.inspect`).
     profiling: AtomicBool,
     records: Mutex<Vec<JobRecord>>,
@@ -80,6 +85,8 @@ pub struct JobRecord {
     pub source_cpu: Duration,
     /// The frame came from a render preview.
     pub preview: bool,
+    /// The job was cancelled while running (its result was discarded).
+    pub cancelled: bool,
 }
 
 /// CPU time consumed by the calling thread (None where the platform has no thread clock).
@@ -276,20 +283,25 @@ impl FrameServer {
 
     /// Queue a job unless it is cached, queued or in flight.
     pub fn request(&self, key: FrameKey, time: Tick, scale: f32, project: &Arc<Project>, prio: u32) {
+        self.request_job(key, time, scale, project, prio, false);
+    }
+
+    fn request_job(&self, key: FrameKey, time: Tick, scale: f32, project: &Arc<Project>, prio: u32, prefetch: bool) {
         if self.shared.done.lock().unwrap_or_else(|e| e.into_inner()).map.contains_key(&key)
             || self.shared.plans.lock().unwrap_or_else(|e| e.into_inner()).contains_key(&key)
         {
             return;
         }
-        if self.shared.in_flight.lock().unwrap_or_else(|e| e.into_inner()).contains(&key) {
+        if self.shared.in_flight.lock().unwrap_or_else(|e| e.into_inner()).iter().any(|(k, c, _)| *k == key && !c.load(Ordering::Relaxed)) {
             return;
         }
         let mut q = self.shared.queue.lock().unwrap_or_else(|e| e.into_inner());
         if let Some(j) = q.iter_mut().find(|j| j.key == key) {
             j.prio = j.prio.min(prio);
+            j.prefetch &= prefetch;
             return;
         }
-        q.push_back(Job { key, queued: Instant::now(), time, scale, project: project.clone(), prio });
+        q.push_back(Job { key, queued: Instant::now(), cancel: Default::default(), prefetch, time, scale, project: project.clone(), prio });
         drop(q);
         self.shared.cv.notify_one();
         #[cfg(target_arch = "wasm32")]
@@ -307,7 +319,7 @@ impl FrameServer {
         for i in 1..=ahead {
             let f = key.frame + i * dir * step;
             if f >= 0 {
-                self.request(FrameKey { frame: f, ..key }, rate.tick_of(f), scale, project, i as u32);
+                self.request_job(FrameKey { frame: f, ..key }, rate.tick_of(f), scale, project, i as u32, true);
             }
         }
         let (target, rev, cur) = (key.target, key.revision, key.frame);
@@ -315,8 +327,26 @@ impl FrameServer {
     }
 
     /// Drop queued jobs that fail `keep` (e.g. stale revisions or frames far from the playhead).
+    /// Jobs already running for such frames are cancelled: a worker blocked behind a decoder
+    /// would otherwise go on to decode a frame playback has passed, possibly from its keyframe.
     pub fn retain_queue(&self, keep: impl Fn(&FrameKey) -> bool) {
         self.shared.queue.lock().unwrap_or_else(|e| e.into_inner()).retain(|j| keep(&j.key));
+        for (k, c, _) in self.shared.in_flight.lock().unwrap_or_else(|e| e.into_inner()).iter() {
+            if !keep(k) {
+                c.store(true, Ordering::Relaxed);
+            }
+        }
+    }
+
+    /// Playback stopped: drop its queued prefetch jobs and cancel the running ones (left alone,
+    /// they would keep the workers busy, e.g. rendering effects for frames nobody will see).
+    pub fn stop_prefetch(&self) {
+        self.shared.queue.lock().unwrap_or_else(|e| e.into_inner()).retain(|j| !j.prefetch);
+        for (_, c, prefetch) in self.shared.in_flight.lock().unwrap_or_else(|e| e.into_inner()).iter() {
+            if *prefetch {
+                c.store(true, Ordering::Relaxed);
+            }
+        }
     }
 
     pub fn queue_len(&self) -> usize {
@@ -432,30 +462,37 @@ fn worker(
                 q = sh.cv.wait(q).unwrap_or_else(|e| e.into_inner());
             }
         };
-        sh.in_flight.lock().unwrap_or_else(|e| e.into_inner()).push(job.key);
+        sh.in_flight.lock().unwrap_or_else(|e| e.into_inner()).push((job.key, job.cancel.clone(), job.prefetch));
         let profiling = sh.profiling.load(Ordering::Relaxed);
         let (started, cpu0) = (Instant::now(), if profiling { thread_cpu_time().unwrap_or_default() } else { Duration::ZERO });
         SOURCE_TIME.with(|s| s.set((Duration::ZERO, Duration::ZERO)));
-        let preview;
-        if let Target::SequencePlan(seq) = job.key.target {
-            let (plan, pv) = plan_job(&job, seq, &pool, &services, &previews, profiling);
-            preview = pv;
-            let mut g = sh.plans.lock().unwrap_or_else(|e| e.into_inner());
-            let clock = g.values().map(|v| v.1).max().unwrap_or(0) + 1;
-            g.insert(job.key, (Arc::new(plan), clock));
-            if g.len() > 96 {
-                let mut v: Vec<(u64, FrameKey)> = g.iter().map(|(k, v)| (v.1, *k)).collect();
-                v.sort_unstable_by_key(|x| x.0);
-                for (_, k) in v.into_iter().take(g.len() - 96) {
-                    g.remove(&k);
+        // A cancelled job may have missed layers (its source gave up): its result is not cached.
+        let preview = filmcraft_media::cancel::with_cancel(&job.cancel, || {
+            if let Target::SequencePlan(seq) = job.key.target {
+                let (plan, pv) = plan_job(&job, seq, &pool, &services, &previews, profiling);
+                if !job.cancel.load(Ordering::Relaxed) {
+                    let mut g = sh.plans.lock().unwrap_or_else(|e| e.into_inner());
+                    let clock = g.values().map(|v| v.1).max().unwrap_or(0) + 1;
+                    g.insert(job.key, (Arc::new(plan), clock));
+                    if g.len() > 96 {
+                        let mut v: Vec<(u64, FrameKey)> = g.iter().map(|(k, v)| (v.1, *k)).collect();
+                        v.sort_unstable_by_key(|x| x.0);
+                        for (_, k) in v.into_iter().take(g.len() - 96) {
+                            g.remove(&k);
+                        }
+                    }
                 }
+                pv
+            } else {
+                let (img, pv) = render_job(&job, &pool, &services, &previews, profiling);
+                if !job.cancel.load(Ordering::Relaxed) {
+                    sh.done.lock().unwrap_or_else(|e| e.into_inner()).insert(job.key, Arc::new(img));
+                }
+                pv
             }
-        } else {
-            let (img, pv) = render_job(&job, &pool, &services, &previews, profiling);
-            preview = pv;
-            sh.done.lock().unwrap_or_else(|e| e.into_inner()).insert(job.key, Arc::new(img));
-        }
-        sh.in_flight.lock().unwrap_or_else(|e| e.into_inner()).retain(|k| *k != job.key);
+        });
+        let cancelled = job.cancel.load(Ordering::Relaxed);
+        sh.in_flight.lock().unwrap_or_else(|e| e.into_inner()).retain(|(k, c, _)| !(*k == job.key && Arc::ptr_eq(c, &job.cancel)));
         if profiling {
             let (source_wall, source_cpu) = SOURCE_TIME.with(|s| s.get());
             let rec = JobRecord {
@@ -468,6 +505,7 @@ fn worker(
                 source_wall,
                 source_cpu,
                 preview,
+                cancelled,
             };
             sh.records.lock().unwrap_or_else(|e| e.into_inner()).push(rec);
         }

crates/ui-egui/src/lib.rs

@@ -304,6 +304,7 @@ impl FilmcraftApp {
     pub fn stop(&mut self) {
         self.playback.playing = false;
         self.playback.meter.finish();
+        self.frames.stop_prefetch();
         if let Some(a) = self.audio.as_mut() {
             a.stop();
         }