perf: use single plot prefetch worker

Replace per-batch plot prefetch thread spawning with a persistent background worker that drains superseded queued navigation jobs before rendering.

Keep prefetch results cache-only, preserve foreground ownership of terminal writes, and add a regression test for cancelling stale pending prefetch keys.

Author: siriusctrl

CHANGELOG.md

@@ -8,6 +8,12 @@ matching `## [X.Y.Z]` section before the release tag is pushed.
 
 ## [Unreleased]
 
+### Changed
+
+- Replace per-batch plot prefetch threads with a single background prefetch
+  worker that keeps the newest queued navigation job and cancels superseded
+  pending prefetch keys before rendering them.
+
 ## [0.2.5] - 2026-06-18
 
 ### Added

docs/architecture.md

@@ -111,13 +111,16 @@ Interactive plot viewing keeps terminal input ahead of expensive protocol
 payload work. The event loop drains pending key and resize events before drawing
 so burst input renders the latest state instead of every intermediate state. It
 keeps a bounded cache of encoded frames by protocol, plot kind, viewport, and
-terminal size. After user navigation, a small background prefetcher warms likely
-next frames without blocking the foreground draw. Prefetch results are cached
-as future candidates only; they must not replace the current visible frame just
-because the background worker finished late. For repeated pan actions on large
-scenes, the prefetcher can render a transparent marks atlas once and crop
-future same-zoom pan frames, then composite those marks over the current
-grid/frame layer so axis labels and grid lines stay correct.
+terminal size. After user navigation, a single background prefetch worker warms
+likely next frames without blocking the foreground draw. Before rendering a
+prefetch job, the worker waits briefly and keeps only the newest queued
+navigation job, cancelling older pending keys that have been superseded by
+later input. Prefetch results are cached as future candidates only; they must
+not replace the current visible frame just because the background worker
+finished late. For repeated pan actions on large scenes, the prefetcher can
+render a transparent marks atlas once and crop future same-zoom pan frames, then
+composite those marks over the current grid/frame layer so axis labels and grid
+lines stay correct.
 
 Mouse input in the plot viewer is read-only. Hover events remember the latest
 terminal cell, snap to the nearest visible plot point, and render a separate
@@ -146,7 +149,8 @@ share the same stable z-index, avoiding broad z-index or full-screen deletes
 that can blank the plot during fast navigation. The prefetch list stays
 intentionally small: more candidates increase background raster work and hidden
 terminal bytes, so newer directional batches suppress stale,
-not-yet-transmitted candidates.
+not-yet-transmitted candidates. The worker caches results and pretransmit
+payloads only; terminal writes remain owned by the foreground viewer loop.
 
 Kitty frames request the full terminal cell area while rendering normal terminal
 windows at the full terminal pixel estimate. Very large windows use a bounded

docs/testing.md

@@ -83,6 +83,12 @@ navigation hit should show `payload_bytes_delta` near zero because the
 foreground update is an image placement command for an idle-transmitted image,
 not a new image transfer.
 
+The plot prefetch path uses one background worker. Tests should keep covering
+two invariants: late worker results must not replace the current visible frame,
+and a newer navigation job must cancel older pending prefetch keys before they
+become cache entries. The worker may prepare cached frames and Kitty transmit
+payloads, but terminal writes must remain in the foreground viewer loop.
+
 ## Selector Tests
 
 Selector tests live in `src/render/terminal.rs`.

src/viewer/plot/cache.rs

@@ -1,11 +1,7 @@
 use std::{
     borrow::Cow,
     collections::VecDeque,
-    sync::{
-        Arc,
-        atomic::{AtomicUsize, Ordering},
-        mpsc::{self, Receiver, Sender},
-    },
+    sync::mpsc::{self, Receiver, Sender},
     thread,
     time::Duration,
 };
@@ -33,7 +29,6 @@ use super::{
 };
 
 const MAX_CACHED_FRAMES: usize = 48;
-const MAX_IN_FLIGHT_PREFETCHES: usize = 3;
 const PREFETCH_FORWARD_STEPS: usize = 3;
 const PREFETCH_GRACE_PERIOD: Duration = Duration::from_millis(12);
 const PAN_ATLAS_MIN_POINTS: usize = 2_000;
@@ -50,9 +45,8 @@ pub(super) struct PlotFrameCacheKey {
 pub(super) struct PlotFrameCache {
     entries: VecDeque<CachedPlotFrame>,
     pub(super) last: Option<CachedPlotFrame>,
-    prefetch_tx: Sender<PrefetchResult>,
-    prefetch_rx: Receiver<PrefetchResult>,
-    in_flight: Arc<AtomicUsize>,
+    prefetch_tx: Sender<PrefetchJob>,
+    prefetch_rx: Receiver<PrefetchEvent>,
     queued: Vec<PlotFrameCacheKey>,
     next_image_id: u32,
     next_transmit_batch: u64,
@@ -98,15 +92,41 @@ struct PrefetchResult {
     frame: Result<CachedPlotFrame, String>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PrefetchMode {
+    Batch,
+    Pan,
+}
+
+#[derive(Debug)]
+struct PrefetchJob {
+    scene: PlotScene,
+    requests: Vec<PrefetchRequest>,
+    mode: PrefetchMode,
+}
+
+impl PrefetchJob {
+    fn keys(&self) -> Vec<PlotFrameCacheKey> {
+        self.requests.iter().map(|request| request.key).collect()
+    }
+}
+
+#[derive(Debug)]
+enum PrefetchEvent {
+    Cancelled(Vec<PlotFrameCacheKey>),
+    Result(PrefetchResult),
+}
+
 impl Default for PlotFrameCache {
     fn default() -> Self {
-        let (prefetch_tx, prefetch_rx) = mpsc::channel();
+        let (prefetch_tx, prefetch_job_rx) = mpsc::channel();
+        let (prefetch_event_tx, prefetch_rx) = mpsc::channel();
+        spawn_prefetch_worker(prefetch_job_rx, prefetch_event_tx);
         Self {
             entries: VecDeque::new(),
             last: None,
             prefetch_tx,
             prefetch_rx,
-            in_flight: Arc::new(AtomicUsize::new(0)),
             queued: Vec::new(),
             next_image_id: 1,
             next_transmit_batch: 0,
@@ -161,9 +181,6 @@ impl PlotFrameCache {
         if protocol == Protocol::Blocks {
             return;
         }
-        if self.in_flight.load(Ordering::Relaxed) >= MAX_IN_FLIGHT_PREFETCHES {
-            return;
-        }
 
         let Some(action) = recent_action else {
             return;
@@ -208,13 +225,13 @@ impl PlotFrameCache {
         ) {
             let keys = self.keys_with_image_ids(keys);
             if scene.total_points() >= PAN_ATLAS_MIN_POINTS {
-                self.spawn_pan_prefetch(scene.clone(), keys);
+                self.schedule_prefetch(scene.clone(), keys, PrefetchMode::Pan);
             } else {
-                self.spawn_batch_prefetch(scene.clone(), keys);
+                self.schedule_prefetch(scene.clone(), keys, PrefetchMode::Batch);
             }
         } else {
             let keys = self.keys_with_image_ids(keys);
-            self.spawn_batch_prefetch(scene.clone(), keys);
+            self.schedule_prefetch(scene.clone(), keys, PrefetchMode::Batch);
         }
     }
 
@@ -223,6 +240,11 @@ impl PlotFrameCache {
             || self.entries.iter().any(|cached| cached.key == key)
     }
 
+    #[cfg(test)]
+    pub(super) fn has_cached_or_queued_key(&self, key: PlotFrameCacheKey) -> bool {
+        self.contains_key(key) || self.queued.contains(&key)
+    }
+
     pub(super) fn drain_transmit_payloads(&mut self, max_count: usize) -> Vec<String> {
         self.collect_prefetches();
         let mut payloads = Vec::new();
@@ -291,11 +313,20 @@ impl PlotFrameCache {
     }
 
     fn collect_prefetches(&mut self) {
-        while let Ok(result) = self.prefetch_rx.try_recv() {
-            self.queued.retain(|key| *key != result.key);
-            if let Ok(mut frame) = result.frame {
-                discard_stale_transmit_payload(&mut frame, self.min_transmit_priority);
-                self.insert_prefetched(frame);
+        while let Ok(event) = self.prefetch_rx.try_recv() {
+            match event {
+                PrefetchEvent::Cancelled(keys) => {
+                    for key in keys {
+                        self.queued.retain(|queued| *queued != key);
+                    }
+                }
+                PrefetchEvent::Result(result) => {
+                    self.queued.retain(|key| *key != result.key);
+                    if let Ok(mut frame) = result.frame {
+                        discard_stale_transmit_payload(&mut frame, self.min_transmit_priority);
+                        self.insert_prefetched(frame);
+                    }
+                }
             }
         }
     }
@@ -320,42 +351,32 @@ impl PlotFrameCache {
         image_id
     }
 
-    fn spawn_batch_prefetch(&mut self, scene: PlotScene, keys: Vec<PrefetchRequest>) {
-        self.discard_stale_transmit_payloads(&keys);
-        self.queued.extend(keys.iter().map(|request| request.key));
-        self.in_flight.fetch_add(1, Ordering::Relaxed);
-        let tx = self.prefetch_tx.clone();
-        let in_flight = Arc::clone(&self.in_flight);
-        thread::spawn(move || {
-            thread::sleep(PREFETCH_GRACE_PERIOD);
-            for request in keys {
-                let frame = render_plot_frame_for_key(
-                    &scene,
-                    request.key,
-                    Some(request.image_id),
-                    request.transmit_priority,
-                )
-                .map_err(|error| error.to_string());
-                let key = request.key;
-                let _ = tx.send(PrefetchResult { key, frame });
-            }
-            in_flight.fetch_sub(1, Ordering::Relaxed);
-        });
-    }
-
-    fn spawn_pan_prefetch(&mut self, scene: PlotScene, keys: Vec<PrefetchRequest>) {
-        self.discard_stale_transmit_payloads(&keys);
-        self.queued.extend(keys.iter().map(|request| request.key));
-        self.in_flight.fetch_add(1, Ordering::Relaxed);
-        let tx = self.prefetch_tx.clone();
-        let in_flight = Arc::clone(&self.in_flight);
-        thread::spawn(move || {
-            thread::sleep(PREFETCH_GRACE_PERIOD);
-            for (key, frame) in render_pan_prefetch_frames(&scene, &keys) {
-                let _ = tx.send(PrefetchResult { key, frame });
+    fn schedule_prefetch(
+        &mut self,
+        scene: PlotScene,
+        requests: Vec<PrefetchRequest>,
+        mode: PrefetchMode,
+    ) {
+        self.discard_stale_transmit_payloads(&requests);
+        self.queued
+            .extend(requests.iter().map(|request| request.key));
+        let queued_keys = requests
+            .iter()
+            .map(|request| request.key)
+            .collect::<Vec<_>>();
+        if self
+            .prefetch_tx
+            .send(PrefetchJob {
+                scene,
+                requests,
+                mode,
+            })
+            .is_err()
+        {
+            for key in queued_keys {
+                self.queued.retain(|queued| *queued != key);
             }
-            in_flight.fetch_sub(1, Ordering::Relaxed);
-        });
+        }
     }
 
     fn render_plot_frame(
@@ -405,6 +426,55 @@ fn discard_stale_transmit_payload(frame: &mut CachedPlotFrame, min_priority: u64
     }
 }
 
+fn spawn_prefetch_worker(job_rx: Receiver<PrefetchJob>, event_tx: Sender<PrefetchEvent>) {
+    thread::spawn(move || {
+        while let Ok(job) = job_rx.recv() {
+            let job = newest_prefetch_job_after_grace(job, &job_rx, &event_tx);
+            render_prefetch_job(job, &event_tx);
+        }
+    });
+}
+
+fn newest_prefetch_job_after_grace(
+    mut job: PrefetchJob,
+    job_rx: &Receiver<PrefetchJob>,
+    event_tx: &Sender<PrefetchEvent>,
+) -> PrefetchJob {
+    thread::sleep(PREFETCH_GRACE_PERIOD);
+    let mut cancelled = Vec::new();
+    while let Ok(next) = job_rx.try_recv() {
+        cancelled.extend(job.keys());
+        job = next;
+    }
+    if !cancelled.is_empty() {
+        let _ = event_tx.send(PrefetchEvent::Cancelled(cancelled));
+    }
+    job
+}
+
+fn render_prefetch_job(job: PrefetchJob, event_tx: &Sender<PrefetchEvent>) {
+    match job.mode {
+        PrefetchMode::Batch => {
+            for request in job.requests {
+                let frame = render_plot_frame_for_key(
+                    &job.scene,
+                    request.key,
+                    Some(request.image_id),
+                    request.transmit_priority,
+                )
+                .map_err(|error| error.to_string());
+                let key = request.key;
+                let _ = event_tx.send(PrefetchEvent::Result(PrefetchResult { key, frame }));
+            }
+        }
+        PrefetchMode::Pan => {
+            for (key, frame) in render_pan_prefetch_frames(&job.scene, &job.requests) {
+                let _ = event_tx.send(PrefetchEvent::Result(PrefetchResult { key, frame }));
+            }
+        }
+    }
+}
+
 pub(super) fn render_plot_frame(
     scene: &PlotScene,
     kind: PlotKind,

src/viewer/plot/tests.rs

@@ -474,6 +474,67 @@ fn late_prefetch_results_do_not_replace_visible_plot_frame() {
     assert_eq!(cache.last.as_ref().unwrap().key, visible_key);
 }
 
+#[test]
+fn prefetch_worker_cancels_superseded_job_before_rendering() {
+    let scene = PlotScene {
+        title: Some("latency".to_owned()),
+        series: vec![PlotSeries {
+            name: "api".to_owned(),
+            points: (0..80)
+                .map(|index| PlotPoint {
+                    x: index as f64,
+                    y: (index % 17) as f64,
+                })
+                .collect(),
+        }],
+    };
+    let size = TerminalSize {
+        width: 120,
+        height: 32,
+    };
+    let mut state = PlotViewState::new(scene.bounds().unwrap().normalized());
+    state.zoom_in();
+    let mut cache = PlotFrameCache::default();
+
+    cache.prefetch_neighbors(
+        &scene,
+        PlotKind::Line,
+        &state,
+        Protocol::Kitty,
+        size,
+        Some(PlotNavAction::ZoomIn),
+    );
+    let mut stale_zoom = state;
+    stale_zoom.zoom_in();
+    let stale_key = cache::PlotFrameCacheKey {
+        kind: PlotKind::Line,
+        protocol: Protocol::Kitty,
+        visible: stale_zoom.visible,
+        size,
+        hover_x: None,
+    };
+
+    cache.prefetch_neighbors(
+        &scene,
+        PlotKind::Line,
+        &state,
+        Protocol::Kitty,
+        size,
+        Some(PlotNavAction::PanRight),
+    );
+    let transmit_payloads = wait_for_transmit_payloads(&mut cache, 4);
+
+    assert!(
+        transmit_payloads
+            .iter()
+            .any(|payload| payload.contains("a=t"))
+    );
+    assert!(
+        !cache.has_cached_or_queued_key(stale_key),
+        "worker should drop an older pending prefetch job when a newer navigation job arrives"
+    );
+}
+
 #[test]
 fn stale_prefetched_frame_still_replaces_previous_visible_image() {
     let scene = PlotScene {