From d30fe47876d8dd730c1a7dd71c1310ba6f0b3813 Mon Sep 17 00:00:00 2001 From: LyAhn Date: Fri, 12 Jun 2026 10:45:12 +0100 Subject: [PATCH] perf(workers): strict priority pipeline across background workers Workers now form a priority pipeline (thumbnails -> metadata -> embeddings -> tagging): a worker only claims jobs when no higher-priority tier has claimable work, so stages drain one at a time at full speed instead of all workers contending for CPU (shared rayon pool), disk, and the DB writer simultaneously. Each tier gate is a single read-only EXISTS query per idle tick using the same exclusion set the corresponding claim would (that worker''s paused folders plus actively-indexing folders), so paused or mid-scan work never blocks lower tiers, and failed jobs stop gating once they leave pending. Deferred workers wake within one backoff interval (250-750ms) of the higher tier draining. A watcher-driven trickle of new files briefly pauses lower tiers by design. A user-initiated tag run during a large embedding backlog waits for it; pausing embeddings per folder overrides if tags are wanted first. --- src-tauri/src/db.rs | 51 ++++++++++++++++++++++++++++++++++++ src-tauri/src/indexer.rs | 56 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index 2123d05..3542e5d 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -1152,6 +1152,57 @@ fn get_pending_thumbnail_jobs_excluding( Ok(rows.collect::>>()?) } +/// True if any claimable (pending, non-excluded) jobs exist in `job_table`. +/// Used by lower-priority workers to defer to higher-priority queues. +/// `extra_predicate` narrows the image join (e.g. videos only for metadata). +fn has_claimable_jobs( + conn: &Connection, + job_table: &str, + extra_predicate: &str, + excluded_folder_ids: &std::collections::HashSet, +) -> Result { + let sql = format!( + "SELECT EXISTS( + SELECT 1 + FROM {} j + JOIN images i ON i.id = j.image_id + WHERE j.status = 'pending' + {} + {} + )", + job_table, + extra_predicate, + folder_exclusion_clause("i", excluded_folder_ids) + ); + Ok(conn.query_row(&sql, [], |row| row.get::<_, i64>(0))? != 0) +} + +pub fn has_claimable_thumbnail_jobs( + conn: &Connection, + excluded_folder_ids: &std::collections::HashSet, +) -> Result { + has_claimable_jobs(conn, "thumbnail_jobs", "", excluded_folder_ids) +} + +pub fn has_claimable_metadata_jobs( + conn: &Connection, + excluded_folder_ids: &std::collections::HashSet, +) -> Result { + has_claimable_jobs( + conn, + "metadata_jobs", + "AND i.media_kind = 'video'", + excluded_folder_ids, + ) +} + +pub fn has_claimable_embedding_jobs( + conn: &Connection, + excluded_folder_ids: &std::collections::HashSet, +) -> Result { + has_claimable_jobs(conn, "embedding_jobs", "", excluded_folder_ids) +} + pub fn claim_thumbnail_jobs( conn: &mut Connection, active_folder_ids: &std::collections::HashSet, diff --git a/src-tauri/src/indexer.rs b/src-tauri/src/indexer.rs index f34ddd0..ed933c7 100644 --- a/src-tauri/src/indexer.rs +++ b/src-tauri/src/indexer.rs @@ -121,6 +121,50 @@ static FOLDER_STORAGE_PROFILES: OnceLock> = OnceLock::new(); const EMBEDDING_BATCH_SIZE: usize = 8; +/// Background workers form a strict priority pipeline: a worker only claims +/// jobs when no higher-priority claimable work exists, so stages drain one +/// at a time (thumbnails → metadata → embeddings → tagging) instead of all +/// workers contending for CPU, disk, and the DB writer at once. +#[derive(Clone, Copy, PartialEq, PartialOrd)] +enum WorkerTier { + Thumbnail = 0, + Metadata = 1, + Embedding = 2, + Tagging = 3, +} + +/// True if any tier above `own_tier` still has claimable work. Each check +/// uses the same exclusion set the corresponding claim would (that worker's +/// paused folders plus actively-indexing folders), so paused or mid-scan +/// work never gates lower tiers. +fn higher_priority_work_pending(pool: &DbPool, own_tier: WorkerTier) -> Result { + let conn = pool.get()?; + let active_folders = active_indexing_folders(); + + if own_tier > WorkerTier::Thumbnail { + let mut excluded = paused_folder_ids("thumbnail"); + excluded.extend(active_folders.iter().copied()); + if db::has_claimable_thumbnail_jobs(&conn, &excluded)? { + return Ok(true); + } + } + if own_tier > WorkerTier::Metadata { + let mut excluded = paused_folder_ids("metadata"); + excluded.extend(active_folders.iter().copied()); + if db::has_claimable_metadata_jobs(&conn, &excluded)? { + return Ok(true); + } + } + if own_tier > WorkerTier::Embedding { + let mut excluded = paused_folder_ids("embedding"); + excluded.extend(active_folders.iter().copied()); + if db::has_claimable_embedding_jobs(&conn, &excluded)? { + return Ok(true); + } + } + Ok(false) +} + #[derive(Clone, Serialize)] pub struct IndexProgress { pub folder_id: i64, @@ -702,6 +746,10 @@ fn process_metadata_batch( pool: &DbPool, media_tools: &MediaTools, ) -> Result { + if higher_priority_work_pending(pool, WorkerTier::Metadata)? { + return Ok(false); + } + let jobs = { with_db_write_lock(|| { let mut conn = pool.get()?; @@ -772,6 +820,10 @@ fn process_embedding_batch( pool: &DbPool, embedder: &mut Option, ) -> Result { + if higher_priority_work_pending(pool, WorkerTier::Embedding)? { + return Ok(false); + } + let batch_started_at = Instant::now(); let claim_started_at = Instant::now(); // Exclude folders that are actively indexing (matching the thumbnail and @@ -1032,6 +1084,10 @@ fn process_tagging_batch( return Ok(false); } + if higher_priority_work_pending(pool, WorkerTier::Tagging)? { + return Ok(false); + } + // Exclude actively-indexing folders for the same reason as the other // workers: don't compete with a running scan. let mut excluded_folders = paused_folder_ids("tagging");