diff --git a/core/dr-catalog/src/jobs.rs b/core/dr-catalog/src/jobs.rs index 6913791..2bede77 100644 --- a/core/dr-catalog/src/jobs.rs +++ b/core/dr-catalog/src/jobs.rs @@ -10,8 +10,14 @@ //! table absorb the redundancy. //! - **Priority shared with the GPU scheduler** (ARCH §5.3), so one notion of //! urgency governs the whole app and visible work always preempts bulk work. +//! +//! Nothing in this file runs a job. [`crate::runner`] is the other half — the +//! one that claims from this table, does the work through a handler, and +//! reports back. Worth knowing because for a long time it did not exist: every +//! producer called [`enqueue`] and nothing ever called [`claim_next`], so the +//! table only ever grew. -use rusqlite::Connection; +use rusqlite::{Connection, OptionalExtension}; use crate::error::CatalogError; @@ -48,6 +54,21 @@ pub enum JobKind { } impl JobKind { + /// Every kind, so code that has to enumerate them cannot quietly miss one + /// that was added later. A `match` would catch that; a hand-written array + /// at each call site would not. + pub const ALL: [JobKind; 9] = [ + JobKind::ScanFolder, + JobKind::ExtractMetadata, + JobKind::Thumbnail, + JobKind::ReadSidecar, + JobKind::WriteSidecar, + JobKind::ContentHash, + JobKind::FetchPreview, + JobKind::FetchOriginal, + JobKind::DetectFaces, + ]; + fn from_i64(v: i64) -> Option { Some(match v { 0 => JobKind::ScanFolder, @@ -68,6 +89,16 @@ impl JobKind { pub fn is_network(self) -> bool { matches!(self, JobKind::FetchPreview | JobKind::FetchOriginal) } + + /// Whether `subject_id` names a row in `images`. + /// + /// Every kind but one is per-photograph. `ScanFolder`'s subject is a + /// *folder*, and the two id spaces are unrelated — so anything that joins + /// `subject_id` against `images` has to exclude it, or it will read one + /// table's ids as another's and act on the answer. + pub fn subject_is_image(self) -> bool { + !matches!(self, JobKind::ScanFolder) + } } /// Scheduling class, matching the GPU tile scheduler (ARCH §5.3). @@ -86,6 +117,22 @@ pub enum Priority { Interactive = 2, } +impl Priority { + /// Read back from the stored column. + /// + /// An unrecognised value reads as `Background` rather than failing: a + /// priority is a hint about ordering, and refusing to run a job because + /// its urgency is spelled oddly would be a worse answer than running it + /// last. + fn from_i64(v: i64) -> Self { + match v { + 2 => Priority::Interactive, + 1 => Priority::Prefetch, + _ => Priority::Background, + } + } +} + /// Lifecycle state. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(i64)] @@ -156,52 +203,115 @@ pub fn enqueue( /// Claim the next runnable job, highest priority first. /// /// `now` is passed rather than read from the clock so backoff is testable. -/// Claiming marks the row `Running` in the same transaction as the read, so -/// two workers cannot take the same job. pub fn claim_next(conn: &Connection, now: i64) -> Result, CatalogError> { - let tx = conn.unchecked_transaction()?; + claim(conn, now, None) +} - let job = tx - .query_row( - "SELECT id, kind, subject_id, priority, attempts, payload - FROM jobs - WHERE state = 0 AND not_before <= ?1 - ORDER BY priority DESC, id ASC - LIMIT 1", - [now], - |r| { - Ok(( - r.get::<_, i64>(0)?, - r.get::<_, i64>(1)?, - r.get::<_, Option>(2)?, - r.get::<_, i64>(3)?, - r.get::<_, i64>(4)?, - r.get::<_, Option>(5)?, - )) - }, - ) - .ok(); +/// Claim the next runnable job of one of `kinds`. +/// +/// What lets a runner take only the work it can actually do. A device with no +/// connector must leave `FetchOriginal` rows alone rather than claim them and +/// fail them five times each with backoff; a runner gated onto an unmetered +/// network (FR-NC-6) passes the local kinds only and leaves the transfers +/// where they are. Neither is expressible by filtering *after* a claim, +/// because the claim has already marked the row `Running`. +/// +/// An empty list claims nothing, which is the honest reading of "there is +/// nothing this worker can do". +pub fn claim_next_matching( + conn: &Connection, + now: i64, + kinds: &[JobKind], +) -> Result, CatalogError> { + if kinds.is_empty() { + return Ok(None); + } + claim(conn, now, Some(kinds)) +} - let Some((id, kind, subject_id, priority, attempts, payload)) = job else { +/// The claim, as one statement. +/// +/// # Why this is not a transaction around a read and a write +/// +/// It used to be, and under two connections that is not safe in the way it +/// looks. A deferred transaction takes a read lock for the `SELECT` and only +/// tries to upgrade at the `UPDATE`; with WAL, a second worker that read the +/// same snapshot gets `SQLITE_BUSY_SNAPSHOT` on its write — an error a busy +/// handler cannot retry away, because the fix is to roll back and start over. +/// So the queue was correct only in the sense that the loser failed loudly. +/// +/// `UPDATE ... WHERE id = (SELECT ...) RETURNING` is one statement, so it is +/// one implicit transaction that takes the write lock immediately. Two workers +/// serialise, the loser waits out its busy timeout rather than erroring, and +/// neither can see a row the other is already holding. +fn claim( + conn: &Connection, + now: i64, + kinds: Option<&[JobKind]>, +) -> Result, CatalogError> { + // `now` first, then the kinds, matching the order the placeholders appear + // in the text below. + let mut args: Vec = vec![now]; + let filter = match kinds { + None => String::new(), + Some(kinds) => { + // Built from the kind *count*, never from anything a user typed — + // the same discipline `collections::descendants` uses, since + // `carray` is not compiled in. + let placeholders = std::iter::repeat_n("?", kinds.len()) + .collect::>() + .join(","); + args.extend(kinds.iter().map(|k| *k as i64)); + format!(" AND kind IN ({placeholders})") + } + }; + + let sql = format!( + "UPDATE jobs + SET state = 1, attempts = attempts + 1 + WHERE id = (SELECT id + FROM jobs + WHERE state = 0 AND not_before <= ?{filter} + ORDER BY priority DESC, id ASC + LIMIT 1) + RETURNING id, kind, subject_id, priority, attempts, payload" + ); + + let claimed = conn + .query_row(&sql, rusqlite::params_from_iter(args.iter()), |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, Option>(2)?, + r.get::<_, i64>(3)?, + r.get::<_, i64>(4)?, + r.get::<_, Option>(5)?, + )) + }) + .optional()?; + + let Some((id, kind, subject_id, priority, attempts, payload)) = claimed else { return Ok(None); }; - tx.execute( - "UPDATE jobs SET state = 1, attempts = attempts + 1 WHERE id = ?1", - [id], - )?; - tx.commit()?; + let Some(kind) = JobKind::from_i64(kind) else { + // A row written by a build that knows a kind this one does not — a + // downgrade, or a catalog synced from a newer device. Running it as + // some other kind would be worse than not running it, so it is parked + // where the next claim will not see it again. + // + // Answering `None` understates what is queued for one pass. The + // alternative is a loop that keeps claiming the same unreadable row. + abandon(conn, id, &format!("unknown job kind {kind}"))?; + return Ok(None); + }; Ok(Some(Job { id, - kind: JobKind::from_i64(kind).unwrap_or(JobKind::ExtractMetadata), + kind, subject_id, - priority: match priority { - 2 => Priority::Interactive, - 1 => Priority::Prefetch, - _ => Priority::Background, - }, - attempts: attempts + 1, + priority: Priority::from_i64(priority), + attempts, payload, })) } @@ -215,16 +325,53 @@ pub fn complete(conn: &Connection, id: i64) -> Result<(), CatalogError> { /// Job failed. Reschedules with backoff, or gives up past [`MAX_ATTEMPTS`]. pub fn fail(conn: &Connection, job: &Job, now: i64, err: &str) -> Result<(), CatalogError> { if job.attempts >= MAX_ATTEMPTS { - conn.execute( - "UPDATE jobs SET state = 2, last_error = ?2 WHERE id = ?1", - rusqlite::params![job.id, err], - )?; + abandon(conn, job.id, err) } else { conn.execute( "UPDATE jobs SET state = 0, not_before = ?2, last_error = ?3 WHERE id = ?1", rusqlite::params![job.id, now + backoff_seconds(job.attempts), err], )?; + Ok(()) } +} + +/// Give up on a job now, with no further retries. +/// +/// For failures a retry cannot fix — the subject is gone, the payload is +/// unreadable, the format is one this build does not know. Walking the whole +/// retry ladder to reach a conclusion the first attempt already reached costs +/// five wakeups and five backoffs per photograph, which on a phone is the +/// difference the user notices. +/// +/// The row is kept rather than deleted, because "this file failed and here is +/// why" is something the user is entitled to see (NFR-ARCH-4). +pub fn abandon(conn: &Connection, id: i64, err: &str) -> Result<(), CatalogError> { + conn.execute( + "UPDATE jobs SET state = 2, last_error = ?2 WHERE id = ?1", + rusqlite::params![id, err], + )?; + Ok(()) +} + +/// Put a claimed job back exactly as it was found. +/// +/// For a worker that is being stopped rather than a job that is going wrong: +/// the platform revoked the slot, the user left the screen. The attempt the +/// claim consumed is given back, because nothing was learned about the file — +/// without that, five backgroundings in a row would mark good work as failed. +/// +/// Guarded on the claim still being *this* claim. There is no owner column, so +/// `attempts` stands in for one: it is bumped by every claim, so the row only +/// still reads `state = 1` with the caller's own attempt number while nobody +/// else has taken it since. A worker that comes back after the recovery pass +/// handed its job to someone else therefore changes nothing, rather than +/// releasing a job another worker is in the middle of. +pub fn release(conn: &Connection, job: &Job) -> Result<(), CatalogError> { + conn.execute( + "UPDATE jobs SET state = 0, attempts = max(0, attempts - 1) + WHERE id = ?1 AND state = 1 AND attempts = ?2", + rusqlite::params![job.id, job.attempts], + )?; Ok(()) } @@ -232,11 +379,97 @@ pub fn fail(conn: &Connection, job: &Job, now: i64, err: &str) -> Result<(), Cat /// /// A row left `Running` has no owner — the process that claimed it is gone. /// Called at startup, before any worker begins (FR-PLAT-AND-3). +/// +/// The attempt the dead claim consumed is deliberately *not* refunded. A job +/// that takes the process down with it is indistinguishable from one that +/// fails, and the attempt counter is the only evidence that survives a death — +/// without it a poison-pill job is reclaimed and re-run forever. pub fn recover_orphaned(conn: &Connection) -> Result { let n = conn.execute("UPDATE jobs SET state = 0 WHERE state = 1", [])?; Ok(n) } +/// Delete jobs whose photograph is gone. +/// +/// Coalescing keeps the table one row per unit of work, but nothing shrinks it +/// when the work stops existing: a library that has been culled carries a +/// thumbnail job for every photograph deleted since the last time anything +/// looked. Each one would be claimed, run, and failed five times. +/// +/// Only kinds whose subject really is an image ([`JobKind::subject_is_image`]) +/// are considered — `ScanFolder`'s subject is a folder id, and joining it +/// against `images` would delete jobs by coincidence of numbering. +/// +/// There is no foreign key to do this instead. `jobs.subject_id` deliberately +/// references nothing: it means different tables for different kinds, and a +/// constraint that is right for eight of nine kinds is not a constraint. +pub fn reap_orphan_subjects(conn: &Connection) -> Result { + let kinds: Vec = JobKind::ALL + .iter() + .filter(|k| k.subject_is_image()) + .map(|k| *k as i64) + .collect(); + let placeholders = std::iter::repeat_n("?", kinds.len()) + .collect::>() + .join(","); + + let n = conn.execute( + &format!( + "DELETE FROM jobs + WHERE subject_id IS NOT NULL + AND kind IN ({placeholders}) + AND NOT EXISTS (SELECT 1 FROM images WHERE images.id = jobs.subject_id)" + ), + rusqlite::params_from_iter(kinds.iter()), + )?; + Ok(n) +} + +/// How much is left, by state. +/// +/// One query rather than a listing, because the caller is a progress line: a +/// foreground service's notification has to say how much remains without +/// reading a hundred thousand rows to find out (FR-PLAT-AND-4). +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct Counts { + /// Claimable now or after a backoff. + pub pending: usize, + /// Claimed by someone. After a clean start that is a live worker; before + /// [`recover_orphaned`] it is a dead one. + pub running: usize, + /// Given up on, and kept so the user can see what failed and why. + pub failed: usize, +} + +impl Counts { + /// Work that is still going to happen. + pub fn outstanding(&self) -> usize { + self.pending + self.running + } +} + +/// Count the queue by state. +pub fn counts(conn: &Connection) -> Result { + // `sum` over no rows is NULL, not 0 — an empty queue would otherwise fail + // to convert rather than counting nothing. + let (pending, running, failed) = conn.query_row( + "SELECT sum(state = 0), sum(state = 1), sum(state = 2) FROM jobs", + [], + |r| { + Ok(( + r.get::<_, Option>(0)?, + r.get::<_, Option>(1)?, + r.get::<_, Option>(2)?, + )) + }, + )?; + Ok(Counts { + pending: pending.unwrap_or(0) as usize, + running: running.unwrap_or(0) as usize, + failed: failed.unwrap_or(0) as usize, + }) +} + #[cfg(test)] mod tests { use super::*; @@ -391,6 +624,241 @@ mod tests { assert_eq!(backoff_seconds(100), 300); } + /// An image row, so a job has a subject that exists. + fn image(c: &Connection, id: i64) { + c.execute( + "INSERT INTO roots(id, kind, label) VALUES (1, 'local', '/lib') + ON CONFLICT DO NOTHING", + [], + ) + .unwrap(); + c.execute( + "INSERT INTO images(id, root_id, source_ref, added_at) VALUES (?1, 1, ?2, 0)", + rusqlite::params![id, format!("/lib/{id}.CR3")], + ) + .unwrap(); + } + + #[test] + fn a_runner_claims_only_the_kinds_it_names() { + // The property a filtered claim exists for: work this worker cannot do + // is left untouched — not claimed, not attempted, not failed. + let c = db(); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + enqueue( + &c, + JobKind::FetchOriginal, + Some(1), + Priority::Interactive, + None, + ) + .unwrap(); + + // `FetchOriginal` is the higher priority and would be claimed first by + // an unfiltered claim. It is not this worker's to take. + let job = claim_next_matching(&c, 0, &[JobKind::Thumbnail]) + .unwrap() + .expect("the thumbnail is claimable"); + assert_eq!(job.kind, JobKind::Thumbnail); + assert!(claim_next_matching(&c, 0, &[JobKind::Thumbnail]) + .unwrap() + .is_none()); + + let (state, attempts): (i64, i64) = c + .query_row( + "SELECT state, attempts FROM jobs WHERE kind = ?1", + [JobKind::FetchOriginal as i64], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap(); + assert_eq!(state, JobState::Pending as i64); + assert_eq!(attempts, 0); + } + + #[test] + fn a_worker_that_can_do_nothing_claims_nothing() { + let c = db(); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + assert!(claim_next_matching(&c, 0, &[]).unwrap().is_none()); + } + + #[test] + fn releasing_a_claim_gives_the_attempt_back() { + // A stopped worker has learned nothing about the file, so the claim it + // is handing back must cost nothing. + let c = db(); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + + let job = claim_next(&c, 0).unwrap().unwrap(); + assert_eq!(job.attempts, 1); + release(&c, &job).unwrap(); + + let again = claim_next(&c, 0).unwrap().expect("claimable again at once"); + assert_eq!(again.attempts, 1, "the release refunded the first attempt"); + } + + #[test] + fn releasing_a_job_someone_else_has_reclaimed_does_nothing() { + // `attempts` standing in for an owner column. A worker that comes back + // after the recovery pass handed its job to someone else must not + // release a claim that is no longer its to release — which would drop + // the live worker's job back into the queue to be run twice. + let c = db(); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + + let stale = claim_next(&c, 0).unwrap().unwrap(); + recover_orphaned(&c).unwrap(); + let live = claim_next(&c, 0).unwrap().unwrap(); + assert_eq!(live.attempts, 2); + + release(&c, &stale).unwrap(); + + let (state, attempts): (i64, i64) = c + .query_row("SELECT state, attempts FROM jobs", [], |r| { + Ok((r.get(0)?, r.get(1)?)) + }) + .unwrap(); + assert_eq!( + state, + JobState::Running as i64, + "the live claim still holds" + ); + assert_eq!(attempts, 2, "and its attempt was not refunded for it"); + } + + #[test] + fn abandoning_skips_the_whole_retry_ladder() { + let c = db(); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + let job = claim_next(&c, 0).unwrap().unwrap(); + + abandon(&c, job.id, "not an image this build can read").unwrap(); + + assert!(claim_next(&c, 1_000_000).unwrap().is_none()); + let (state, attempts, err): (i64, i64, String) = c + .query_row("SELECT state, attempts, last_error FROM jobs", [], |r| { + Ok((r.get(0)?, r.get(1)?, r.get(2)?)) + }) + .unwrap(); + assert_eq!(state, JobState::Failed as i64); + assert_eq!(attempts, 1, "one attempt, not MAX_ATTEMPTS"); + // Kept, not deleted: the user is entitled to see what failed and why. + assert!(err.contains("this build can read")); + } + + #[test] + fn jobs_for_a_deleted_photograph_are_reaped() { + // Coalescing keeps the table one row per unit of work; nothing shrank + // it when the work stopped existing. + let c = db(); + image(&c, 1); + image(&c, 2); + enqueue(&c, JobKind::Thumbnail, Some(1), Priority::Background, None).unwrap(); + enqueue(&c, JobKind::Thumbnail, Some(2), Priority::Background, None).unwrap(); + enqueue( + &c, + JobKind::ExtractMetadata, + Some(2), + Priority::Background, + None, + ) + .unwrap(); + + c.execute("DELETE FROM images WHERE id = 2", []).unwrap(); + assert_eq!(reap_orphan_subjects(&c).unwrap(), 2); + + let left: i64 = c + .query_row("SELECT subject_id FROM jobs", [], |r| r.get(0)) + .unwrap(); + assert_eq!(left, 1); + } + + #[test] + fn a_folder_scan_is_not_reaped_by_image_ids() { + // `ScanFolder`'s subject is a folder. Joining it against `images` + // would delete it whenever the numbering happened not to collide — + // which, on a fresh library, is almost always. + let c = db(); + enqueue( + &c, + JobKind::ScanFolder, + Some(1), + Priority::Background, + Some("/lib/2024"), + ) + .unwrap(); + + assert_eq!(reap_orphan_subjects(&c).unwrap(), 0); + assert!(claim_next(&c, 0).unwrap().is_some()); + } + + #[test] + fn a_job_kind_from_a_newer_build_is_parked_rather_than_guessed_at() { + // A catalog synced from a device running a later build. Running an + // unknown kind as some arbitrary known one is worse than not running + // it, and the old code silently read every unknown kind as + // `ExtractMetadata`. + let c = db(); + c.execute( + "INSERT INTO jobs(kind, subject_id, priority, state) VALUES (99, 1, 0, 0)", + [], + ) + .unwrap(); + + assert!(claim_next(&c, 0).unwrap().is_none()); + + let (state, err): (i64, String) = c + .query_row("SELECT state, last_error FROM jobs", [], |r| { + Ok((r.get(0)?, r.get(1)?)) + }) + .unwrap(); + assert_eq!(state, JobState::Failed as i64); + assert!(err.contains("99"), "{err}"); + } + + #[test] + fn counts_say_what_is_left() { + // What a foreground service's notification is built from: a number, + // without reading a hundred thousand rows to find it. + let c = db(); + assert_eq!(counts(&c).unwrap(), Counts::default()); + + for id in 1..=3 { + enqueue(&c, JobKind::Thumbnail, Some(id), Priority::Background, None).unwrap(); + } + let job = claim_next(&c, 0).unwrap().unwrap(); + abandon(&c, job.id, "nope").unwrap(); + claim_next(&c, 0).unwrap().unwrap(); + + let n = counts(&c).unwrap(); + assert_eq!(n.pending, 1); + assert_eq!(n.running, 1); + assert_eq!(n.failed, 1); + assert_eq!(n.outstanding(), 2, "failed work is not outstanding work"); + } + + #[test] + fn every_kind_is_in_all() { + // `ALL` is what the reap builds its kind filter from, so a kind added + // to the enum and forgotten here would quietly stop being reaped. + for (i, kind) in JobKind::ALL.iter().enumerate() { + assert_eq!( + JobKind::from_i64(i as i64), + Some(*kind), + "ALL is out of step with the discriminants at {i}" + ); + } + assert!(JobKind::from_i64(JobKind::ALL.len() as i64).is_none()); + } + + #[test] + fn only_a_folder_scan_has_a_non_image_subject() { + assert!(!JobKind::ScanFolder.subject_is_image()); + for kind in JobKind::ALL.iter().filter(|k| **k != JobKind::ScanFolder) { + assert!(kind.subject_is_image(), "{kind:?}"); + } + } + #[test] fn network_jobs_are_identifiable_for_metered_gating() { // FR-NC-6: transfers respect unmetered-network and charging