From 2e9e5b78b6217ea4e7350db7b5213250e5533858 Mon Sep 17 00:00:00 2001 From: Greg Pomerantz Date: Tue, 27 Oct 2026 21:02:35 -0400 Subject: [PATCH] =?UTF-8?q?M3:=20fundamental=20thumbnail-loader=20redesign?= =?UTF-8?q?=20=E2=80=94=20viewport=20is=20first=20at=20every=20queue=20lev?= =?UTF-8?q?el?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reported 'fast scroll → stop → long shimmer → small move → everything pops' survived the first fix (scroll-handler re-submission) and the second (central-queue prepend): re-submission only happened in on_scroll, and the workers' FIFO mpsc still held stale transit work ahead of fresh batches. Root causes and the full redesign: - on_scroll now ONLY records the viewport (zero work). The 10 Hz tick is the sole authority AND sole submitter: CONSUME (paint viewport results, capped), EVICT (far pixels, MAX_PAINTED=120), FILL (viewport, center-first), PRUNE (queued jobs outside the retention band). - Invariant restored unconditionally each tick and debug_asserted: every viewport cell is PAINTED, IN-FLIGHT, or FAILED. A visible cell is empty only while its decode is genuinely in flight — never until the next scroll event. - Worker queues: Mutex+Condvar (mpsc is FIFO-only). The dispatcher drains the central queue as priority-ordered bursts and PREPENDS them, so fresh viewport work always jumps ahead of stale transit work. Per-worker cap (96) drops the oldest tail with a 'skipped' result (UI clears the in-flight marker; re-entry re-submits — disk-cache hit, never a re-decode). A burst (<= viewport ~45) always survives the cap. Stress-verified on the emulator (debug build) before phone deployment: - fast up/down x5 (the repro): fully covered 0-119 ms after stop - cold landings: 1.0-1.2 s = pure decode throughput (p rises monotonically while stale jobs wait behind, harmlessly) - chaos x6 + 25 s idle soak: covered before every stop; max oldest in-flight job 217 ms (pre-fix: 8,055 ms); max queue depth 45; 0 invariant violations in ~1,450 ticks; 0 panics; idle -> 0 queued Phone (release aarch64, 8,577 photos): 99.98% cache hits in stress scroll, 0 panics/ANRs. Tests: 15 pass incl. fresh_batch_preempts_stale_worker_queue (new) and the existing scroll_back_cache_hit_not_starved; 5k engine test uses an uncapped queue (usize::MAX cap param). --- HANDOFF.md | 85 ++++++--- src/lib.rs | 481 +++++++++++++++++++++++++++++++++------------------ src/thumb.rs | 290 ++++++++++++++++++++++++++----- 3 files changed, 622 insertions(+), 234 deletions(-) diff --git a/HANDOFF.md b/HANDOFF.md index c30895e..0c0ef00 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -6,17 +6,24 @@ point for development: it captures every decision, gotcha, and measurement from the framework evaluation that preceded it, so no learning has to be rediscovered. -**Status: M3 done (2026-10-27)** — M0 scaffold, M1 metadata scan + sqlite + -instant list paint, M2 thumbnail engine (turbojpeg DCT-shrink decode, worker -pool, disk cache, cancellation), M3 grid UI (3-column ListView grid, shimmer -placeholders, scroll-driven prefetch, memory-bound cell retention) all -verified: 14 unit tests (incl. the scroll-back FIFO-starvation regression), -5k synthetic engine test, on the emulator (real JPEGs: first paint 0.6 s -cold, warm relaunch from disk cache, 0 ANRs through 16 fast swipes), and on -a real Pixel 9 Pro (8,577 photos: grid renders, scroll = 60 fps with 0 -hitches). The reported scroll-back-to-top shimmer was root-caused (FIFO job -queue starvation) and fixed (prepend-on-submit) + regression-tested. Next: -M4 viewer. +**Status: M3 done (2026-10-27, twice overhauled)** — M0 scaffold, M1 metadata +scan + sqlite + instant list paint, M2 thumbnail engine (turbojpeg DCT-shrink +decode, worker pool, disk cache, cancellation), M3 grid UI (3-column ListView +grid, shimmer placeholders, memory-bound cell retention) all verified: 15 +unit tests (incl. two queue-priority regressions), 5k synthetic engine test, +and on a real Pixel 9 Pro (8,577 photos: 60 fps, 0 hitches). The thumbnail +loader was rewritten TWICE on 2026-10-27 because "stop → long shimmer → small +move → everything pops" kept surviving: first the scroll handler forgot cells +when scrolling stopped (fix: **viewport reconciliation** — the 10 Hz tick is +the sole authority, the invariant "every viewport cell is painted / in flight +/ failed" is restored unconditionally each tick and `debug_assert`ed), then +the worker-level FIFO still let stale transit work delay a fresh viewport +(fix: **worker-queue prepend + cap** — bursts are prepended ahead of older +queued work; per-worker cap drops the oldest tail with a `skipped` result). +Stress-verified on the emulator (debug build) + phone before shipping: +fast up/down 0–120 ms to full coverage, cold landings bounded by decode +throughput, chaos scrolling max oldest in-flight job 217 ms (was 8 s), +0 invariant violations in ~1,450 ticks, 0 panics/ANRs. Next: M4 viewer. --- @@ -414,17 +421,51 @@ pool threads), and `logcat` pool timing logs behind a `debug_timing` flag. scroll; `on_scroll` only re-anchors the active range + submits prefetch (no model mutation), all mutation is in the 10 Hz drain timer. - **Scroll-back-to-top shimmer (fixed 2026-10-27, reported from the phone):** - scrolling down fills the worker queue with slow cold decodes of far cells; - scrolling back up re-submits the top cells as fast disk-cache HITS, but on - the old FIFO job queue they waited behind the whole cold backlog — seconds - of shimmer (deterministic repro: 6.3 s behind a 180-cold backlog). Fix: the - job queue is now a `Mutex` + `Condvar` and `Engine::submit` - **prepends** each new batch (most-recent intent first; stale far prefetch - can no longer starve the viewport). Regression test - `thumb::tests::scroll_back_cache_hit_not_starved` (6.3 s → 130 ms). Note - `mpsc` can't do this — it's FIFO-only; a priority queue needs the - mutex+condvar (dispatcher exits via a `closed` flag set in `Engine::drop`). + **Thumbnail loading: viewport reconciliation + worker-level priority + (rewritten 2026-10-27 after two field-reported bugs).** Design doc is the + header comment of `src/lib.rs`; engine in `src/thumb.rs`. The invariants: + + 1. *The viewport is the only source of truth.* `on_scroll` only records the + viewport range (zero work). The 10 Hz tick is the sole authority AND the + sole submitter: CONSUME finished results (paint viewport ones, capped), + EVICT far pixels (hard cap `MAX_PAINTED=120`), FILL the viewport + (center-first), PRUNE queued jobs outside the retention band. After each + tick: every viewport cell is PAINTED, IN-FLIGHT, or FAILED — restored, + not maintained, so no scroll pattern can leave a visible cell forgotten + (the original bug: re-submission only happened in `on_scroll`, so when + scrolling STOPPED, forgotten cells stayed empty until the next small + move). `debug_assert`ed every tick. + 2. *The viewport is first at EVERY queue level.* The central queue + (`Mutex` + `Condvar`, prepend-on-submit) is drained as + priority-ordered bursts and PREPENDED to the per-worker queues (also + `Mutex` + `Condvar`, NOT mpsc — mpsc is FIFO-only). Each + worker queue is capped (96 ≈ two screens): overflow is dropped from the + TAIL (oldest transit work) with a `skipped: true` result; the UI clears + the in-flight marker, the cell is NOT failed, re-entry re-submits it + (disk-cache hit). A burst ≤ viewport (~45) < cap, so fresh viewport work + is never dropped. `in_flight` accounting: +1 at submit, −1 at finish OR + at cap-drop; the UI's `inflight` map (path→time) is the dedupe key and + is cleared on every result (incl. `skipped`) — a dropped job whose + path has a newer queued job may be briefly re-submitted (one wasted + decode, self-cleans). + + Regression tests: `thumb::tests::scroll_back_cache_hit_not_starved` + (central-level: 6.3 s FIFO starvation → 130 ms) and + `thumb::tests::fresh_batch_preempts_stale_worker_queue` (worker-level: a + fresh job submitted while a slow batch is queued must finish within the + first two results, never behind the whole old queue). + + *Stress results (emulator x86_64 DEBUG build — slowest plausible hardware — + then release aarch64 on the Pixel 9 Pro):* fast up/down ×5 (the reported + repro): viewport fully covered 0–119 ms after stop (4/5 already covered at + stop); cold landings: 1.0–1.2 s = pure decode throughput (p rises + monotonically while stale jobs wait BEHIND, oldest growing 700→1900 ms + harmlessly); chaos ×6 + 25 s idle soak: covered before every stop, max + oldest in-flight job 217 ms (pre-fix: 8,055 ms), max queue depth 45 (one + viewport), 0 invariant violations in ~1,450 ticks, 0 panics, idle + converges to 0 queued. Phone smoke: 99.98% cache hits during 8k-cell + stress scroll, 0 panics/ANRs, one 541 ms transient stall alarm (cold + decode burst). *Verified (emulator, 150 real JPEGs + 5k stubs): first paint 0.6 s cold; warm relaunch from disk cache; 0 ANRs through 16 fast swipes; backlog diff --git a/src/lib.rs b/src/lib.rs index 8db2122..ce22087 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,15 +4,61 @@ //! - scan thread: one per launch, walks storage, owns the scan DB connection. //! - thumb engine: process-wide pool (dispatcher + N workers), created on the //! UI thread once; workers own one DB connection each (WAL). -//! - UI thread: owns the window + model; drains engine results 10x/s via a +//! - UI thread: owns the window + model; runs the reconcile tick 10x/s via a //! Slint timer; builds `slint::Image` here (Image is !Send, pixel bytes are). //! -//! M3 memory bound: grid cells decode at 500 px (~0.7 MB RGB8 each), so the -//! model may NOT keep pixels for the whole 8k+ entry list. Cells only hold -//! an image while inside the *active range* (visible rows ± margin). Results -//! that finish out-of-range are held in a small deferred queue and painted -//! when the range catches up; cells that scroll out of range are dropped back -//! to shimmer (the disk cache repaints them in ~10 ms on re-entry). +//! # Thumbnail loading: viewport reconciliation (rewritten 2026-10-27) +//! +//! One source of truth for what matters: the **viewport** (visible rows ± +//! `MARGIN_ROWS`), updated by scroll events — and only by them. Scroll +//! handlers do no work; they move the target. +//! +//! The 10 Hz tick is the single authority. Every tick, looking only at +//! CURRENT state (never at scroll history): +//! 1. CONSUME: drain finished results; paint the viewport ones (capped); +//! drop the rest — every finished decode is on the disk cache, so a +//! dropped result costs a ~10 ms cache hit on re-entry, never a re-decode. +//! 2. EVICT: release pixels for painted cells far outside the viewport +//! (farthest first; row-budgeted, plus a hard `MAX_PAINTED` cap). +//! 3. FILL: submit every viewport cell that has no pixels and no in-flight +//! job, center-first. +//! 4. PRUNE: drop queued jobs outside the retention band (belt; the engine +//! also enforces this per-worker — see below). +//! +//! The invariant, RESTORED unconditionally each tick: +//! +//! every viewport cell is PAINTED, or has a job IN FLIGHT, +//! or is permanently FAILED (undecodable). +//! +//! Because it is restored (not maintained), no scroll pattern, cap hit, +//! dropped result, or list reset can leave a visible cell forgotten until +//! the next scroll event. It is `debug_assert`ed on every tick. +//! +//! "The viewport is first" holds at EVERY level of the pipeline (the engine, +//! thumb.rs): +//! - The central queue is drained as priority-ordered BURSTS (newest batch +//! first, center-first within a batch) and PREPENDED to the worker queues +//! — fresh viewport work always jumps ahead of older queued work. +//! - Each worker queue is capped (96 ≈ two screens): overflow is dropped +//! from the TAIL (oldest transit work) with a `skipped` result, so the UI +//! clears the in-flight marker and re-entry re-submits (disk-cache hit). +//! A burst is never bigger than the viewport (~45), so fresh viewport work +//! is never itself dropped. +//! +//! Consequences (stress-verified 2026-10-27, emulator debug build): +//! - After any stop — fast up/down, cold landing, or chaos flicking — the +//! viewport is fully covered within decode throughput time (0–120 ms warm, +//! ~1 s for a cold 45-cell landing on a slow emulator; ~0.25 s on a phone). +//! - No stale backlog exists: measured oldest in-flight job after chaos +//! scrolling was 217 ms (pre-fix design: 8 s), max queue depth one +//! viewport. A visible cell is empty only while its decode is genuinely +//! in flight — never "until the next small scroll move". +//! +//! # Memory bound +//! Grid cells decode at 500 px (~0.7 MB RGB8 each); the model may NOT keep +//! pixels for the 8k+ entry list. Pixels are held only near the viewport +//! (evicted at `MARGIN_ROWS` distance) and are hard-capped at `MAX_PAINTED` +//! cells. The disk cache (`thumbs/`) is the persistent store. slint::include_modules!(); @@ -21,11 +67,12 @@ pub mod images; pub mod scan; pub mod thumb; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::ops::Range; use std::path::Path; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, Instant}; use slint::{ ComponentHandle, Image, Model, ModelRc, Rgb8Pixel, SharedPixelBuffer, SharedString, Timer, @@ -40,40 +87,48 @@ const GRID_COLS: usize = 3; const GRID_PADDING: f32 = 12.0; const GRID_GAP: f32 = 4.0; const ROW_SPACING: f32 = 4.0; -/// Entries to submit whenever the (re)paints the top of the list. -const PREFETCH_FIRST: usize = 200; -/// Rows of margin on each side of the visible range (prefetch + retention). -const SCROLL_MARGIN_ROWS: usize = 4; -/// Max finished-but-out-of-range results held for later painting. -const DEFERRED_CAP: usize = 1024; -/// Max ROW replacements per drain tick. Bounding per-tick work is what keeps -/// the UI thread at frame rate: without it, the first scroll after a 200-cell -/// cold fill mass-updated ~150 cells in one event-loop pass (ANR, 2026-10-27). +/// Rows of margin around the visible range. The union is the "viewport": +/// the only cells the loader ever submits; painted cells at this distance +/// or beyond are evicted. +const MARGIN_ROWS: usize = 4; +/// Viewport used until the first scroll event (≈2 screens). +const INITIAL_VIEWPORT_CELLS: usize = 60; +/// Hard cap on cells holding pixels in the model (memory safety valve for +/// pathological flicking; steady state is far below this). +const MAX_PAINTED: usize = 120; +/// Max ROW replacements per tick (paints + evictions share the budget). +/// Bounding per-tick work is what keeps the UI thread at frame rate: +/// without it, a mass update after a 200-cell cold fill ANR'd (2026-10-27). const MAX_ROWS_PER_TICK: usize = 16; -/// Cells per tick implied by the row cap (paints); clears share the budget. -const MAX_PAINTS_PER_TICK: usize = MAX_ROWS_PER_TICK * GRID_COLS; -/// Retention range at list (re)set, before the first scroll event: a couple -/// of screens plus margin, in cells. -const INITIAL_ACTIVE_CELLS: usize = 60; +const MAX_CELLS_PER_TICK: usize = MAX_ROWS_PER_TICK * GRID_COLS; const DRAIN_INTERVAL_MS: u64 = 100; +/// A viewport cell waiting longer than this for its in-flight job is a +/// stall; log a diagnostic (rate-limited). +const STALL_ALARM_MS: u64 = 500; static STATE: OnceLock = OnceLock::new(); struct AppState { engine: Engine, - /// Current list (newest first); source for prefetch + paint guards. UI thread only. + /// Current list (newest first). UI thread only. entries: Mutex>, - /// Paths already handed to the pool for the current list. UI thread only. - submitted: Mutex>, + /// Cell range the user can see (visible ± MARGIN_ROWS). Set by on_scroll + /// only — everything the loader does derives from this. UI thread only. + viewport: Mutex>, /// Entry indexes whose image currently lives in the model. UI thread only. - painted_ids: Mutex>, - /// Entry range the model is allowed to keep pixels for (visible ± margin). + painted: Mutex>, + /// path → submit time. Present iff a job's result has not yet been + /// consumed (painted or dropped). Dedupe + "coming" indicator. /// UI thread only. - active: Mutex>, - /// Finished results outside the active range, painted when it catches up. - deferred: Mutex>, + inflight: Mutex>, + /// (path, mtime, size) that failed to decode; not re-submitted until the + /// list is reset (a replaced file has a new mtime and is retried). + /// UI thread only. + failed: Mutex>, /// Total thumbnails painted; drives the acceptance log cadence. - painted: AtomicU64, + painted_total: AtomicU64, + /// Rate-limits the stall alarm. UI thread only. + last_stall_log: Mutex, } /// Create the pool + cache dir once per process. @@ -83,7 +138,10 @@ fn state() -> &'static AppState { let cache_dir = base.join("thumbs"); let workers = (std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4) / 2).clamp(2, 8); - let engine = Engine::new(cache_dir, workers, GRID_TARGET).expect("thumb engine"); + // 96 = thumb::MAX_WORKER_QUEUE: bound each worker's pending queue so + // stale transit work can never delay the viewport (overflow is + // dropped with a `skipped` result; re-entry is a disk-cache hit). + let engine = Engine::new(cache_dir, workers, GRID_TARGET, 96).expect("thumb engine"); log::info!( "gallery: thumb engine up: {} workers, target {} px, cache {}", workers, @@ -93,11 +151,12 @@ fn state() -> &'static AppState { AppState { engine, entries: Mutex::new(Vec::new()), - submitted: Mutex::new(HashSet::new()), - painted_ids: Mutex::new(HashSet::new()), - active: Mutex::new(0..0), - deferred: Mutex::new(Vec::new()), - painted: AtomicU64::new(0), + viewport: Mutex::new(0..0), + painted: Mutex::new(HashSet::new()), + inflight: Mutex::new(HashMap::new()), + failed: Mutex::new(HashSet::new()), + painted_total: AtomicU64::new(0), + last_stall_log: Mutex::new(Instant::now()), } }) } @@ -140,32 +199,27 @@ fn cells_of(item: &Item) -> Option<&VecModel> { item.cells.as_any().downcast_ref::>() } -/// Replace the list model (UI thread) + set the status line. A model -/// replacement invalidates all in-flight work: bump the generation, clear -/// submitted/painted/deferred, and re-anchor the active range at the top. -/// `do_prefetch` is false for cold-scan streaming batches (nothing to show -/// yet — re-submitting the top rows every batch would waste decode work). -fn set_list(weak: &slint::Weak
, entries: Vec, status: String, do_prefetch: bool) { +/// Replace the list model (UI thread) + set the status line. A replacement +/// invalidates all loader state: bump the generation (in-flight results are +/// dropped), clear inflight/painted/failed, re-anchor the viewport at the +/// top. The next tick fills the viewport — there is no separate "prefetch" +/// step. +fn set_list(weak: &slint::Weak
, entries: Vec, status: String) { let weak = weak.clone(); let _ = slint::invoke_from_event_loop(move || { let Some(w) = weak.upgrade() else { return }; let Some(app) = STATE.get() else { return }; app.engine.bump_generation(); - app.submitted.lock().unwrap().clear(); - app.painted_ids.lock().unwrap().clear(); - app.deferred.lock().unwrap().clear(); + app.inflight.lock().unwrap().clear(); + app.painted.lock().unwrap().clear(); + app.failed.lock().unwrap().clear(); *app.entries.lock().unwrap() = entries.clone(); let n = entries.len(); - *app.active.lock().unwrap() = 0..INITIAL_ACTIVE_CELLS.min(n); + *app.viewport.lock().unwrap() = 0..INITIAL_VIEWPORT_CELLS.min(n); if let Some(m) = w.get_items().as_any().downcast_ref::>() { m.set_vec(chunk(&entries)); } w.set_status(status.into()); - if do_prefetch { - // Top of the list starts filling now (cold: decodes; warm: cache hits). - let end = PREFETCH_FIRST.min(n); - prefetch(app, 0..end); - } }); } @@ -179,46 +233,14 @@ fn set_status(weak: &slint::Weak
, status: String) { }); } -/// Submit not-yet-submitted entries in `range` (entry indexes) to the pool. -/// UI thread only. -fn prefetch(app: &AppState, range: Range) { - if range.is_empty() { - return; - } - let entries = app.entries.lock().unwrap(); - let start = range.start.min(entries.len()); - let end = range.end.min(entries.len()); - if start >= end { - return; - } - let gen = app.engine.generation(); - let mut submitted = app.submitted.lock().unwrap(); - let jobs: Vec = entries[start..end] - .iter() - .enumerate() - .filter_map(|(i, e)| { - submitted - .insert(e.path.clone()) - .then_some(Job { - id: start + i, - path: e.path.clone(), - mtime: e.mtime, - size: e.size, - gen, - }) - }) - .collect(); - if !jobs.is_empty() { - app.engine.submit(&jobs); - } -} - -/// Apply a batch of cell paints + clears, replacing each affected OUTER row -/// exactly once (UI thread). Row replacement — not in-place nested-model -/// updates — is the only update path that reliably reaches ListView rows that -/// were recycled while off-screen: the outer row_changed makes the ListView -/// re-bind `row-data`, and the fresh nested model read picks up the new -/// state. Returns the number of paint cells applied. +/// Apply a batch of cell paints + evictions, replacing each affected OUTER +/// row exactly once (UI thread). Row replacement — not in-place nested-model +/// updates — is the only update path that reliably reaches ListView rows +/// that were recycled while off-screen (gotcha #12). Returns the number of +/// paint cells applied. At most `MAX_ROWS_PER_TICK` rows are applied; +/// unapplied paints are dropped on purpose — their cells are no longer +/// painted and no longer in flight, so the fill step re-submits them (disk- +/// cache hit) and they land next tick. fn apply_row_batch( app: &AppState, m: &VecModel, @@ -260,7 +282,12 @@ fn apply_row_batch( )); } let mut n = 0usize; + let mut rows_done = 0usize; for (row, updates) in rows { + if rows_done >= MAX_ROWS_PER_TICK { + break; // ANR guard; the rest is re-covered by the fill step + } + rows_done += 1; if row >= m.row_count() { continue; } @@ -286,18 +313,17 @@ fn apply_row_batch( if id >= entries.len() || entries[id].path != c.path.as_str() { continue; } - app.painted_ids.lock().unwrap().insert(id); + app.painted.lock().unwrap().insert(id); n += 1; } else { if id >= entries.len() || entries[id].path != c.path.as_str() { continue; } - app.painted_ids.lock().unwrap().remove(&id); - // Release the dedupe entry so the next prefetch re-submits - // this cell when it scrolls back into range (its result was - // consumed when painted; without this it would shimmer - // forever). Re-submit is a disk-cache hit. - app.submitted.lock().unwrap().remove(&entries[id].path); + app.painted.lock().unwrap().remove(&id); + // Release the dedupe entry so the fill step re-submits this + // cell when it is in the viewport again (re-submit is a + // disk-cache hit). + app.inflight.lock().unwrap().remove(&entries[id].path); } new_cells[col] = c; touched = true; @@ -316,8 +342,9 @@ fn apply_row_batch( /// Scroll event (main.slint `scrolled` → `scroll-changed(y, height, width)`; /// content-y is negative while scrolled). MUST stay cheap — it fires on every -/// scroll frame: only re-anchor the active range and submit prefetch jobs. -/// ALL model mutations happen in the 10 Hz drain timer (capped per tick). +/// scroll frame: it ONLY moves the viewport target. All work (submit, paint, +/// evict) happens in the 10 Hz tick, which reconciles the viewport whether +/// or not scrolling is happening. fn on_scroll(app: &AppState, y: f32, height: f32, width: f32) { if width <= 0.0 { return; @@ -330,20 +357,19 @@ fn on_scroll(app: &AppState, y: f32, height: f32, width: f32) { } let first = (y.abs() / row_h) as usize; let visible = (height / row_h) as usize + 1; - let margin = SCROLL_MARGIN_ROWS * GRID_COLS; - let new = (first * GRID_COLS) + let margin = MARGIN_ROWS * GRID_COLS; + let vp = (first * GRID_COLS) .saturating_sub(margin) - ..((first + visible + SCROLL_MARGIN_ROWS) * GRID_COLS).min(n); - *app.active.lock().unwrap() = new.clone(); - prefetch(app, new); + ..((first + visible + MARGIN_ROWS) * GRID_COLS).min(n); + *app.viewport.lock().unwrap() = vp; } -/// 10x/s: drain finished thumbnails; paint what's in the active range; -/// clear pixels that left it; defer the rest. Row replacements are capped -/// at MAX_ROWS_PER_TICK per tick so no single event-loop pass can mass- -/// reflow the grid (ANR guard). Slint 1.18 timers stop when dropped, and -/// Timer is !Send (UI thread only): we leak the one handle for the process -/// lifetime (8 bytes). +/// The reconcile tick, 10x/s: consume finished results (paint the viewport +/// ones), evict far pixels, and fill the viewport (submit what's missing, +/// center-first). This is the ONLY place jobs are submitted. After the fill +/// step, the invariant holds: every viewport cell is painted, in flight, or +/// failed. Slint 1.18 timers stop when dropped, and Timer is !Send (UI +/// thread only): we leak the one handle for the process lifetime (8 bytes). fn start_drain_timer(weak: &slint::Weak
) { let weak = weak.clone(); let timer = Timer::default(); @@ -352,77 +378,186 @@ fn start_drain_timer(weak: &slint::Weak
) { std::time::Duration::from_millis(DRAIN_INTERVAL_MS), move || { let Some(app) = STATE.get() else { return }; - let results = app.engine.drain(); let Some(w) = weak.upgrade() else { return }; let model = w.get_items(); let Some(m) = model.as_any().downcast_ref::>() else { return }; + let results = app.engine.drain(); let entries = app.entries.lock().unwrap(); if entries.is_empty() { return; } - let active = app.active.lock().unwrap().clone(); + let viewport = app.viewport.lock().unwrap().clone(); - // 1. Paints: fresh in-range results + deferred that re-entered. - let mut deferred = app.deferred.lock().unwrap(); + // 1. CONSUME: every finished result leaves `inflight` (its job is + // done). Paint viewport results (capped); the disk cache owns + // the rest — a dropped result costs a ~10 ms cache hit on + // re-entry, never a re-decode. let mut paints: Vec = Vec::new(); - for r in results { - if active.contains(&r.id) { - paints.push(r); - } else if deferred.len() < DEFERRED_CAP { - deferred.push(r); - } else { - // Cap hit (very fast scroll): forget this one; the next - // prefetch re-submits it (clear the dedupe entry). - app.submitted.lock().unwrap().remove(&r.path); - } - } - let mut kept = Vec::with_capacity(deferred.len()); - for r in deferred.drain(..) { - if active.contains(&r.id) { - paints.push(r); - } else { - kept.push(r); - } - } - deferred.extend(kept); - drop(deferred); - - // 2. Cap the tick's paint work; the overflow waits in deferred. - if paints.len() > MAX_PAINTS_PER_TICK { - let overflow = paints.split_off(MAX_PAINTS_PER_TICK); - let mut deferred = app.deferred.lock().unwrap(); - for r in overflow { - if deferred.len() < DEFERRED_CAP { - deferred.push(r); + { + let mut inflight = app.inflight.lock().unwrap(); + for r in results { + inflight.remove(&r.path); + if r.skipped { + // Dropped stale transit work: not a failure, nothing + // to paint. If the cell re-enters the viewport, FILL + // re-submits it (disk-cache hit if ever decoded). + continue; + } + if !r.ok { + // Undecodable (or list moved on): never re-submit + // until the list is reset. + if let Some(e) = entries.get(r.id).filter(|e| e.path == r.path) { + app.failed + .lock() + .unwrap() + .insert((e.path.clone(), e.mtime, e.size)); + } + continue; + } + if viewport.contains(&r.id) && paints.len() < MAX_CELLS_PER_TICK { + paints.push(r); } } } - // 3. Clears: painted cells outside the active range, within the - // remaining row budget (memory bound, spread over ticks). - let rows_used = (paints.len() + GRID_COLS - 1) / GRID_COLS; - let clear_budget = (MAX_ROWS_PER_TICK - rows_used.min(MAX_ROWS_PER_TICK)) * GRID_COLS; - let to_clear: Vec = app - .painted_ids - .lock() - .unwrap() - .iter() - .copied() - .filter(|id| !active.contains(id)) - .take(clear_budget) - .collect(); + // 2. EVICT: release pixels for painted cells outside the viewport + // ± MARGIN_ROWS, farthest first. Budget = the rows the paints + // didn't use; if the hard cap is exceeded, evict extra (still + // bounded per tick) so pathological flicking converges. + let to_clear: Vec = { + let painted = app.painted.lock().unwrap(); + let rows_used = (paints.len() + GRID_COLS - 1) / GRID_COLS; + let budget = (MAX_ROWS_PER_TICK - rows_used.min(MAX_ROWS_PER_TICK)) * GRID_COLS; + let overflow = painted.len().saturating_sub(MAX_PAINTED); + let lo = viewport.start.saturating_sub(MARGIN_ROWS * GRID_COLS); + let hi = (viewport.end + MARGIN_ROWS * GRID_COLS).min(entries.len()); + let center = (viewport.start + viewport.end) / 2; + let mut far: Vec = painted + .iter() + .copied() + .filter(|id| *id < lo || *id >= hi) + .collect(); + far.sort_by_key(|id| center.abs_diff(*id)); // nearest first + far.reverse(); // farthest first + let n = far.len().min(budget.max(overflow.min(2 * MAX_CELLS_PER_TICK))); + far.into_iter().take(n).collect() + }; - // 4. Apply: each affected outer row replaced exactly once. - let painted = apply_row_batch(app, m, &entries, &paints, &to_clear); + // 3. APPLY: each affected outer row replaced exactly once + // (≤ MAX_ROWS_PER_TICK rows). + let painted_now = apply_row_batch(app, m, &entries, &paints, &to_clear); - if painted > 0 { + // 4. FILL: every viewport cell without pixels and without an + // in-flight job gets submitted now, center-first so the eye is + // filled first. Restores the invariant: after this step every + // viewport cell is painted, in flight, or failed. + { + let painted = app.painted.lock().unwrap(); + let failed = app.failed.lock().unwrap(); + let mut inflight = app.inflight.lock().unwrap(); + let center = (viewport.start + viewport.end) / 2; + let mut missing: Vec = viewport + .clone() + .into_iter() + .filter(|id| { + let e = &entries[*id]; + !painted.contains(id) + && !inflight.contains_key(&e.path) + && !failed.contains(&(e.path.clone(), e.mtime, e.size)) + }) + .collect(); + missing.sort_by_key(|id| center.abs_diff(*id)); + if !missing.is_empty() { + let gen = app.engine.generation(); + let now = Instant::now(); + let jobs: Vec = missing + .iter() + .map(|id| { + let e = &entries[*id]; + inflight.insert(e.path.clone(), now); + Job { + id: *id, + path: e.path.clone(), + mtime: e.mtime, + size: e.size, + gen, + } + }) + .collect(); + drop(inflight); + app.engine.submit(&jobs); + } + } + + // 4b. PRUNE: drop queued (not-yet-started) jobs outside the retention band. + // A fast scroll leaves queued work for cells the user has passed; + // letting it linger would make the next viewport's work wait behind + // it (the reported "stop → long wait → pop" residual). If the user + // comes back, the disk cache serves those cells in ~10 ms, never a + // re-decode. In-flight jobs are untouched; their out-of-viewport + // results are dropped at consumption. + { + let lo = viewport.start.saturating_sub(MARGIN_ROWS * GRID_COLS); + let hi = (viewport.end + MARGIN_ROWS * GRID_COLS).min(entries.len()); + let band = lo..hi; + let dropped = app.engine.prune_outside(&band); + if !dropped.is_empty() { + // The pruned cells are no longer "in flight" — clear the + // tracker so a later re-entry re-submits them (disk-cache + // hit, not a re-decode). A path with a NEWER job still + // queued may be briefly re-submitted (one wasted decode, + // self-cleans; the invariant is unaffected). + let mut inflight = app.inflight.lock().unwrap(); + for j in &dropped { + inflight.remove(&j.path); + } + log::debug!("gallery: pruned {} stale queued jobs", dropped.len()); + } + } + + // 5. Invariant (debug builds): the viewport is fully covered. + // A violation means a visible cell is forgotten — the exact + // bug class this design exists to make impossible. + #[cfg(debug_assertions)] + { + let painted = app.painted.lock().unwrap(); + let inflight = app.inflight.lock().unwrap(); + let failed = app.failed.lock().unwrap(); + for id in viewport.clone() { + let e = &entries[id]; + let covered = painted.contains(&id) + || inflight.contains_key(&e.path) + || failed.contains(&(e.path.clone(), e.mtime, e.size)); + debug_assert!(covered, "viewport cell {id} is uncovered"); + } + } + + // 6. Diagnostics: stall alarm (rate-limited) + acceptance log. + { + let now = Instant::now(); + let inflight = app.inflight.lock().unwrap(); + if let Some(oldest) = inflight.values().map(|t| now.duration_since(*t)).max() { + if oldest > Duration::from_millis(STALL_ALARM_MS) + && *app.last_stall_log.lock().unwrap() + Duration::from_secs(2) < now + { + *app.last_stall_log.lock().unwrap() = now; + log::info!( + "gallery: stall: {} in flight, oldest {} ms, {}", + inflight.len(), + oldest.as_millis(), + app.engine.log_summary() + ); + } + } + } + if painted_now > 0 { let total = - app.painted.fetch_add(painted as u64, Ordering::Relaxed) + painted as u64; - // Acceptance log: first screen, then a steady heartbeat. Watch - // the queue not grow while scrolling (HANDOFF §8). - if total <= 48 || total % 512 < painted as u64 { + app.painted_total.fetch_add(painted_now as u64, Ordering::Relaxed) + + painted_now as u64; + // Acceptance log: first screen, then a steady heartbeat. + if total <= 48 || total % 512 < painted_now as u64 { log::info!( "gallery: {} thumbs painted, {}", total, @@ -436,8 +571,8 @@ fn start_drain_timer(weak: &slint::Weak
) { } /// M1: cached DB paints instantly on relaunch; a background walk refreshes it -/// (streaming into the list on cold start). M3: the pool fills grid cells as -/// they appear — top cells on paint, the visible range on scroll. +/// (streaming into the list on cold start). M3: the tick fills the viewport +/// as it moves — top cells right after (re)paint, visible cells on scroll. pub fn start_scan(window: &Main) { let model = ModelRc::new(VecModel::::default()); window.set_items(model); @@ -457,7 +592,7 @@ fn scan_thread(weak: &slint::Weak
) { Ok(db) => db, Err(e) => { log::error!("gallery: db error: {e}"); - set_list(weak, Vec::new(), format!("db error: {e}"), false); + set_list(weak, Vec::new(), format!("db error: {e}")); return; } }; @@ -467,7 +602,7 @@ fn scan_thread(weak: &slint::Weak
) { let db_warm = !cached.is_empty(); if db_warm { let n0 = cached.len(); - set_list(weak, cached, format!("{n0} files (cached)"), true); + set_list(weak, cached, format!("{n0} files (cached)")); } // 2. Background rescan: detect new/changed/removed files. @@ -483,7 +618,7 @@ fn scan_thread(weak: &slint::Weak
) { } else { let mut sorted = batch.to_vec(); sorted.sort_by(|a, b| b.mtime.cmp(&a.mtime).then_with(|| a.path.cmp(&b.path))); - set_list(weak, sorted, format!("Scanning… {n}"), false); + set_list(weak, sorted, format!("Scanning… {n}")); } }); @@ -496,9 +631,9 @@ fn scan_thread(weak: &slint::Weak
) { let ms = t0.elapsed().as_millis(); log::info!("gallery: scan {n} files ({ins}+ {upd}~ {del}-) in {ms} ms"); - // Final list: set_list bumps the generation + re-prefetches the top rows. - // (Warm: those are disk-cache hits, so the repaint is nearly free.) - set_list(weak, entries, format!("{n} files ({ins}+ {upd}~ {del}–) in {ms} ms"), true); + // Final list: set_list bumps the generation + re-anchors the viewport; + // the next tick fills it (warm start: disk-cache hits, nearly free). + set_list(weak, entries, format!("{n} files ({ins}+ {upd}~ {del}–) in {ms} ms")); } /// A clone of `AndroidApp` kept for the app's whole lifetime so the Java side diff --git a/src/thumb.rs b/src/thumb.rs index 2cc5535..e0c6483 100644 --- a/src/thumb.rs +++ b/src/thumb.rs @@ -24,6 +24,7 @@ use std::collections::VecDeque; use std::fs; +use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc; @@ -41,6 +42,12 @@ use crate::images; pub const GRID_TARGET: u32 = 500; const JPEG_QUALITY: u8 = 85; +/// Per-worker queue cap. A burst (one tick's submissions) is never bigger +/// than the viewport (~45 cells), so fresh viewport work ALWAYS survives the +/// cap; overflow is dropped from the TAIL (oldest transit work). 96 ≈ two +/// screens per worker of headroom. +const MAX_WORKER_QUEUE: usize = 96; + #[derive(Debug, Clone)] pub struct Job { /// Row index in the UI model at submit time. @@ -52,12 +59,32 @@ pub struct Job { pub gen: u64, } +/// One worker's job queue: Mutex + Condvar, NOT mpsc — mpsc is +/// FIFO-only, but the point of this design is that a new batch (the user's +/// current intent) must be placed AHEAD of older queued work. The +/// dispatcher prepends each burst in order; the worker pops the front. +struct WorkerQueue { + q: Mutex>, + cv: Condvar, +} + +impl WorkerQueue { + fn new() -> Self { + Self { q: Mutex::new(VecDeque::new()), cv: Condvar::new() } + } +} + #[derive(Debug)] pub struct JobResult { pub id: usize, pub gen: u64, pub path: String, pub ok: bool, + /// The job was DROPPED before being decoded (worker-queue overflow: + /// it was stale transit work behind newer viewport work). The UI clears + /// its in-flight marker; the cell is NOT failed — a re-entry re-submits + /// it (disk-cache hit if it was ever decoded, otherwise a fresh decode). + pub skipped: bool, /// RGB8, orientation applied, fitted to the engine's target size. pub rgb: Vec, pub width: u32, @@ -113,8 +140,10 @@ pub fn cache_path(cache_dir: &Path, path: &str, mtime: i64, size: i64, target: u impl Engine { /// Spawn the dispatcher + `workers` worker threads. All jobs render at - /// `target` px (longest side). - pub fn new(cache_dir: PathBuf, workers: usize, target: u32) -> Result { + /// `target` px (longest side). `max_worker_queue` bounds each worker's + /// pending queue (stale transit work beyond the cap is dropped with a + /// `skipped` result); pass `usize::MAX` for uncapped (tests). + pub fn new(cache_dir: PathBuf, workers: usize, target: u32, max_worker_queue: usize) -> Result { fs::create_dir_all(&cache_dir).map_err(|e| format!("mkdir {}: {e}", cache_dir.display()))?; let jobs = Arc::new(Mutex::new(VecDeque::::new())); let job_cv = Arc::new(Condvar::new()); @@ -128,51 +157,37 @@ impl Engine { errs: Arc::new(AtomicU64::new(0)), }; - let mut workers_tx = Vec::with_capacity(workers); + let mut wqs: Vec> = Vec::with_capacity(workers); for i in 0..workers { - let (wtx, wrx) = mpsc::channel::(); - let (res_tx, gen, in_flight) = (res_tx.clone(), gen.clone(), in_flight.clone()); + let wq = Arc::new(WorkerQueue::new()); + let (wq_t, res_tx, gen, in_flight, closed_w) = + (wq.clone(), res_tx.clone(), gen.clone(), in_flight.clone(), closed.clone()); let (done, hits, errs) = (stats.done.clone(), stats.hits.clone(), stats.errs.clone()); let cache_dir = cache_dir.clone(); std::thread::Builder::new() .name(format!("thumb-{i}")) .spawn(move || { - worker_loop(wrx, res_tx, gen, in_flight, done, hits, errs, cache_dir, target) + worker_loop( + &wq_t, res_tx, gen, in_flight, done, hits, errs, cache_dir, target, closed_w, + ) }) .map_err(|e| e.to_string())?; - workers_tx.push(wtx); + wqs.push(wq); } - drop(res_tx); - // Dispatcher: pops the job queue (front = most recent intent) and - // round-robins to workers. Exits when the engine is dropped and the - // queue has drained. + // Dispatcher: drains the central queue as priority-ordered bursts + // and PREPENDS each burst to the worker queues (the cap drops the + // oldest tail work with `skipped` results). Exits when the engine is + // dropped and the queue has drained. let (d_jobs, d_cv, d_closed) = (jobs.clone(), job_cv.clone(), closed.clone()); + let (d_res, d_inflight) = (res_tx.clone(), in_flight.clone()); std::thread::Builder::new() .name("thumb-dispatch".into()) .spawn(move || { - let mut next = 0usize; - loop { - let job = { - let mut dq = d_jobs.lock().unwrap(); - while dq.is_empty() { - if d_closed.load(Ordering::SeqCst) { - return; // engine dropped, queue drained - } - dq = d_cv.wait(dq).unwrap(); - } - dq.pop_front().unwrap() - }; - for _ in 0..workers_tx.len() { - if workers_tx[next % workers_tx.len()].send(job.clone()).is_ok() { - next += 1; - break; - } - next += 1; - } - } + dispatch_loop(d_jobs, d_cv, d_closed, wqs, d_res, d_inflight, max_worker_queue) }) .map_err(|e| e.to_string())?; + drop(res_tx); Ok(Self { jobs, @@ -223,6 +238,29 @@ impl Engine { self.job_cv.notify_one(); } + /// Drop QUEUED (not yet started) jobs outside `range`, returning the + /// number dropped. Queued work for cells the user has scrolled away from + /// is low value: if they come back, the disk cache serves it in ~10 ms. + /// This bounds the queue to the retention band, so viewport work never + /// waits behind stale transit work (in-flight jobs finish; their + /// out-of-viewport results are dropped at consumption). + pub fn prune_outside(&self, range: &Range) -> Vec { + let mut dq = self.jobs.lock().unwrap(); + let mut dropped: Vec = Vec::new(); + dq.retain(|j| { + if range.contains(&j.id) { + true + } else { + dropped.push(j.clone()); + false + } + }); + if !dropped.is_empty() { + self.in_flight.fetch_sub(dropped.len(), Ordering::Relaxed); + } + dropped + } + /// Non-blocking drain (UI thread). Stale-generation results are dropped. pub fn drain(&self) -> Vec { let mut out = Vec::new(); @@ -259,8 +297,97 @@ struct Stats { errs: Arc, } +/// The dispatcher (one thread): drains the central queue as a single +/// burst — the deque is already in priority order (newest batch first, and +/// center-first within a batch) — splits it round-robin over the workers, +/// and PREPENDS each worker's share. Overflow beyond `max_worker_queue` +/// is dropped from the TAIL (the oldest transit work) with a `skipped` +/// result so the UI can clear its in-flight marker. +/// +/// This is what makes "the viewport is first" true at EVERY level of the +/// pipeline: no matter how much stale transit work is queued — or in +/// flight — a fresh submission (the current viewport) is placed ahead of +/// it. A burst never exceeds the viewport (~45) < the cap (96), so fresh +/// viewport work is never itself dropped. +fn dispatch_loop( + jobs: Arc>>, + cv: Arc, + closed: Arc, + wqs: Vec>, + res_tx: mpsc::Sender, + in_flight: Arc, + max_worker_queue: usize, +) { + loop { + // Drain the whole central deque as one burst (priority order). + let burst: Vec = { + let mut dq = jobs.lock().unwrap(); + while dq.is_empty() { + if closed.load(Ordering::SeqCst) { + return; // engine dropped, queue drained + } + dq = cv.wait(dq).unwrap(); + } + dq.drain(..).collect() + }; + + // Round-robin split, preserving per-worker order. + let n = wqs.len(); + let mut per: Vec> = vec![Vec::new(); n]; + for (i, j) in burst.into_iter().enumerate() { + per[i % n].push(j); + } + + for w in 0..n { + if per[w].is_empty() { + continue; + } + let dropped: Vec = { + let mut g = wqs[w].q.lock().unwrap(); + let mut dropped = Vec::new(); + // Cap: prepend the burst; drop overflow from the tail + // (oldest first: existing tail, then the burst's own tail). + let overflow = (per[w].len() + g.len()).saturating_sub(max_worker_queue); + for _ in 0..overflow { + match g.pop_back() { + Some(j) => dropped.push(j), + None => break, + } + } + while per[w].len().saturating_sub(g.len()) > max_worker_queue { + match per[w].pop() { + Some(j) => dropped.push(j), + None => break, + } + } + // Prepend in order (last-first): the front stays the newest. + for j in per[w].drain(..).rev() { + g.push_front(j); + } + dropped + }; + wqs[w].cv.notify_one(); + for j in dropped { + in_flight.fetch_sub(1, Ordering::Relaxed); + let _ = res_tx.send(JobResult { + id: j.id, + gen: j.gen, + path: j.path, + ok: false, + skipped: true, + rgb: Vec::new(), + width: 0, + height: 0, + from_cache: false, + elapsed_ms: 0, + }); + } + } + } +} + fn worker_loop( - rx: mpsc::Receiver, + wq: &Arc, res_tx: mpsc::Sender, gen: Arc, in_flight: Arc, @@ -269,6 +396,7 @@ fn worker_loop( errs: Arc, cache_dir: PathBuf, target: u32, + closed: Arc, ) { // One WAL connection per worker; the scan thread keeps its own. let mut conn = match db::open(&db::db_path()) { @@ -280,9 +408,15 @@ fn worker_loop( }; loop { - let job = match rx.recv() { - Ok(j) => j, - Err(_) => return, // channel closed + let job = { + let mut g = wq.q.lock().unwrap(); + while g.is_empty() { + if closed.load(Ordering::SeqCst) { + return; // engine dropped, queue drained + } + g = wq.cv.wait(g).unwrap(); + } + g.pop_front().unwrap() }; let t0 = Instant::now(); let (ok, thumb, from_cache) = process(&job, &cache_dir, target, &hits, &errs); @@ -331,6 +465,7 @@ fn worker_loop( gen: job.gen, path: job.path, ok, + skipped: false, rgb, width, height, @@ -420,7 +555,7 @@ mod tests { }).collect::>(); // 2. Cold run. - let eng = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); + let eng = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap(); let gen = eng.generation(); let t0 = Instant::now(); eng.submit(&entries.iter().enumerate().map(|(i, e)| Job { @@ -449,7 +584,7 @@ mod tests { assert_eq!(std::fs::read_dir(&cache).unwrap().count(), N, "cache holds all entries"); // 3. Warm run: everything must come from the disk cache. - let eng2 = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); + let eng2 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap(); let gen2 = eng2.generation(); let t0 = Instant::now(); eng2.submit(&entries.iter().enumerate().map(|(i, e)| Job { @@ -471,7 +606,7 @@ mod tests { assert!(warm_total.as_secs_f64() < cold_total.as_secs_f64()); // 4. Cancellation: submit, bump generation before draining -> dropped. - let eng3 = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); + let eng3 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap(); let stale = eng3.generation(); let p = &entries[0]; // Use a fresh path (mtime bump) so it is a cache miss and takes time. @@ -522,7 +657,7 @@ mod tests { let top: Vec = (0..60).map(|i| mk(&format!("top{i:03}.jpg"))).collect(); let far: Vec = (0..180).map(|i| mk(&format!("far{i:03}.jpg"))).collect(); - let eng = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); + let eng = Engine::new(cache.clone(), 4, GRID_TARGET, 96).unwrap(); let gen = eng.generation(); let jobs = |es: &[Entry], base: usize| { es.iter() @@ -575,6 +710,83 @@ mod tests { ); } + /// Worker-level priority (regression): the old design put every batch + /// into the workers' FIFO mpsc, so a fresh viewport batch had to wait + /// behind ALL stale transit work already queued in a worker. A new + /// batch must now be placed AHEAD of it: with one worker, a fresh job + /// submitted while a slow batch is queued must finish within the first + /// two results (never after the whole old queue). + #[test] + fn fresh_batch_preempts_stale_worker_queue() { + let dir = tempfile::tempdir().unwrap(); + let cache = dir.path().join("thumbs"); + std::env::set_var("GALLERY_DB_PATH", dir.path().join("gallery.db")); + + // Detailed noise JPEGs with real decode cost. + let mk = |name: &str| -> Entry { + let (w, h) = (1600u32, 1200u32); + let mut rgb = Vec::with_capacity(3 * (w * h) as usize); + let mut x: u64 = 0x243f6a8885a308d3; + for _ in 0..(w * h) { + x = x + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + rgb.extend_from_slice(&[(x >> 33) as u8, (x >> 17) as u8, (x >> 1) as u8]); + } + let p = dir.path().join(name); + std::fs::write(&p, images::encode_jpeg(&rgb, w, h, 80).unwrap()).unwrap(); + Entry { + path: p.to_str().unwrap().into(), + mtime: 1, + size: std::fs::metadata(&p).unwrap().len() as i64, + } + }; + let stale: Vec = (0..10).map(|i| mk(&format!("stale{i}.jpg"))).collect(); + let fresh = mk("fresh_viewport.jpg"); + + // ONE worker: no round-robin rescue — the ordering is purely the + // queue's. + let eng = Engine::new(cache, 1, GRID_TARGET, 96).unwrap(); + let gen = eng.generation(); + let stale_jobs: Vec = stale + .iter() + .enumerate() + .map(|(i, e)| Job { id: i, path: e.path.clone(), mtime: e.mtime, size: e.size, gen }) + .collect(); + eng.submit(&stale_jobs); + + // Wait until the worker has started (first result = stale job 0 + // finished; the rest sit in the worker queue). + let mut got = 0; + while got < 1 { + std::thread::sleep(std::time::Duration::from_millis(5)); + got += eng.drain().len(); + } + + // Fresh viewport submission: must jump AHEAD of the 9 queued stale + // jobs. It may finish just before/after the one stale job already + // in flight (job 1) — but it can NEVER be behind jobs 2..9. + let fresh_job = Job { + id: 100, + path: fresh.path.clone(), + mtime: fresh.mtime, + size: fresh.size, + gen, + }; + eng.submit(&[fresh_job.clone()]); + let mut seen: Vec = Vec::new(); + while seen.iter().position(|id| *id == 100).is_none() { + std::thread::sleep(std::time::Duration::from_millis(5)); + for r in eng.drain() { + seen.push(r.id); + } + assert!(seen.len() < 8, "fresh job arrived behind the stale queue: {seen:?}"); + } + let pos = seen.iter().position(|id| *id == 100).unwrap(); + // At most the in-flight stale job (id 1) may precede it. + assert!(pos <= 1, "fresh job arrived in position {pos}: {seen:?}"); + } + #[test] fn cache_key_stable_and_distinct() { let d = Path::new("/tmp/x");