From 13e0e7979f6c3bc87af3deebae2c9fdaabbeddb7 Mon Sep 17 00:00:00 2001 From: sebi Date: Thu, 2 Jul 2026 11:56:06 -0500 Subject: [PATCH] fix(jobs): id reuse across restarts, same-second reconcile, kill race, trim-temp orphans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - JobId::generate re-checks suffixed candidates and run() also treats an on-disk .log as taken: a post-restart job could mint a retained id, truncating the old job's log, losing its own row to the PK conflict, and letting persist_final clobber the old row's final state. - Startup reconcile now flips rows with started_unix <= boot (was <), excluding ids live in this process while holding the jobs lock — a crash loop restarting within the same second left a dead job reading 'running' for up to 24h. Also fixes NOT IN () syntax for the empty case. - kill_job reports by final state when the KILL finds the group already gone after a delivered TERM — that's a successful kill, not a failure. - reap_orphans sweeps aged .log.trim temps left by a crashed trim; no reaper path deleted them before. Co-Authored-By: Claude Fable 5 --- src/jobs/id.rs | 34 ++++++++++-- src/jobs/mod.rs | 127 ++++++++++++++++++++++++++++++++++++++++----- src/jobs/reaper.rs | 69 ++++++++++++++++++++---- 3 files changed, 201 insertions(+), 29 deletions(-) diff --git a/src/jobs/id.rs b/src/jobs/id.rs index af334ae..71895f7 100644 --- a/src/jobs/id.rs +++ b/src/jobs/id.rs @@ -36,6 +36,9 @@ impl JobId { /// metacharacters can reach the id or its log filename. `exists` reports /// whether a candidate is taken; only /// on a clash do we append a `-` from `seq`, so clean ids stay clean. + /// Suffixed candidates are re-checked too: the sequence restarts at 1 each + /// boot, so `-1` may itself survive from a previous run (the id + /// namespace — DB rows and log files — outlives the process). pub fn generate(seq: &AtomicU64, title: Option<&str>, exists: impl Fn(&str) -> bool) -> Self { let label = title .and_then(normalize_title) @@ -45,9 +48,15 @@ impl JobId { return Self(base); } // Same second + same title: a monotonic suffix keeps the id unique without - // sacrificing readability. - let n = seq.fetch_add(1, Ordering::Relaxed); - Self(format!("{base}-{n}")) + // sacrificing readability. The suffix space is unbounded and taken ids are + // finite, so this terminates. + loop { + let n = seq.fetch_add(1, Ordering::Relaxed); + let candidate = format!("{base}-{n}"); + if !exists(&candidate) { + return Self(candidate); + } + } } } @@ -200,14 +209,29 @@ mod tests { #[test] fn collision_appends_sequence_suffix() { let seq = AtomicU64::new(1); - // `exists` always true => the base is "taken", forcing the suffix path. - let id = JobId::generate(&seq, None, |_| true); + // `exists` true only for the bare base => one suffix step is enough. + let id = JobId::generate(&seq, None, |c| !c.ends_with("-1")); let s = id.as_ref(); assert!(valid_format(s, "job")); assert!(s.ends_with("-1"), "expected `-1` seq suffix, got {s}"); assert_eq!(seq.load(Ordering::Relaxed), 2, "clash must advance seq"); } + #[test] + fn taken_suffix_advances_to_the_next_free_one() { + // The seq restarts at 1 each boot, so `-1` can itself survive from + // a previous run. generate must skip every taken suffix, not assume the + // first one is free. + let seq = AtomicU64::new(1); + let id = JobId::generate(&seq, None, |c| !c.ends_with("-3")); + let s = id.as_ref(); + assert!(valid_format(s, "job")); + assert!( + s.ends_with("-3"), + "expected the first free suffix `-3`: {s}" + ); + } + #[test] fn display_and_as_ref_agree() { let seq = AtomicU64::new(1); diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index c4e7c74..4a6154d 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -202,29 +202,45 @@ impl JobStore { db: crate::db::Db, ) -> std::io::Result { std::fs::create_dir_all(&dir)?; - let jobs = Arc::new(Mutex::new(HashMap::new())); + let jobs: Arc>>> = Arc::new(Mutex::new(HashMap::new())); // Startup reconcile: a row left `running` by a previous process can't still // be running — that process (and its children) died with it. Flip those to - // `failed` so history is truthful. Scoped to `started_unix < boot`: only - // rows from before this process started qualify, so a job *this* process - // launches (started_unix >= boot) is never clobbered by the racing update. - // `new` is sync, so the reconcile is spawned; the reaper only touches rows - // older than its retention, so it won't race this for recent rows. + // `failed` so history is truthful. Scoped to `started_unix <= boot` minus + // the ids live in this process: `<` alone missed a dead job from a crash + // loop restarting within the same wall-clock second (it stayed `running` + // for up to 24h). The jobs lock is held across the update, and `run` + // inserts into the map (under that lock) before writing its row — so any + // row this process creates either has its id in the snapshot or appears + // only after the update ran. `new` is sync, so the reconcile is spawned. let boot = crate::db::now_unix(); { let db = db.clone(); + let jobs = jobs.clone(); tokio::spawn(async move { - if let Err(error) = db + let live = jobs.lock().await; + let live_ids: Vec = live.keys().map(|id| id.as_ref().to_string()).collect(); + let result = db .call(move |conn| { - conn.execute( + // `NOT IN ()` is a syntax error, and at boot the map is + // always empty — only add the clause when there are ids. + let exclude = if live_ids.is_empty() { + String::new() + } else { + let placeholders = vec!["?"; live_ids.len()].join(","); + format!(" AND id NOT IN ({placeholders})") + }; + let sql = format!( "UPDATE jobs SET status = 'failed', error = 'server restarted' \ - WHERE status = 'running' AND started_unix < ?1", - [boot], - ) + WHERE status = 'running' AND started_unix <= ?{exclude}" + ); + let params = std::iter::once(rusqlite::types::Value::from(boot)) + .chain(live_ids.into_iter().map(rusqlite::types::Value::from)); + conn.execute(&sql, rusqlite::params_from_iter(params)) }) - .await - { + .await; + drop(live); + if let Err(error) = result { tracing::warn!(%error, "startup job reconcile failed"); } }); @@ -266,8 +282,13 @@ impl JobStore { // reply, `job(list)`, reaper logs, and the log filename — so deriving it // from `cmd` would leak that secret. The title is normalized in `JobId`. let mut jobs = self.jobs.lock().await; + // A candidate is taken if it's live in this process OR its log file + // survives from a previous one (rows and logs outlive restarts for 24h). + // Without the disk check, a post-restart job minting a retained id would + // truncate the old job's log, lose its own row to the PK conflict, and + // clobber the old row's final state from `persist_final`. let id = JobId::generate(&self.seq, title.as_deref(), |candidate| { - jobs.contains_key(candidate) + jobs.contains_key(candidate) || self.dir.join(format!("{candidate}.log")).exists() }); let log_path = self.dir.join(format!("{id}.log")); @@ -1118,6 +1139,84 @@ mod tests { drop(store); // keep the store (its reaper task) alive across the wait } + #[tokio::test] + async fn startup_reconcile_catches_same_second_stale_row() { + let dir = tempfile::tempdir().unwrap().keep(); + let db = Db::memory(); + // A crash loop can restart within the same wall-clock second, leaving a + // row whose `started_unix` equals the new process's boot second. It's + // just as dead as an older row and must reconcile to failed. + let started = crate::db::now_unix(); + db.call(move |conn| { + conn.execute( + "INSERT INTO jobs (id, status, started_unix) VALUES ('ghost-same-sec', 'running', ?1)", + [started], + ) + }) + .await + .unwrap(); + + let store = JobStore::new(dir, Duration::from_secs(5), Shell::sh(), db.clone()).unwrap(); + let id = JobId::from("ghost-same-sec"); + let (status, _code, _tail) = await_row(&db, &id).await; + assert_eq!( + status, "failed", + "a same-second stale running row must reconcile to failed" + ); + drop(store); + } + + #[tokio::test] + async fn restart_never_reuses_a_retained_job_id() { + let dir = tempfile::tempdir().unwrap().keep(); + let db = Db::memory(); + let store = + JobStore::new(dir.clone(), Duration::from_secs(5), Shell::sh(), db.clone()).unwrap(); + let r = store + .run( + "echo first".into(), + None, + None, + false, + false, + Some("t".into()), + ) + .await + .unwrap(); + let RunResult::Inline { id: first, .. } = r else { + panic!("fast command should be inline"); + }; + await_row(&db, &first).await; + + // Restart: empty in-memory map, but the old job's log + row survive. A + // new job with the same title (often the same second) must NOT mint the + // retained id — that would truncate the old log and clobber its row. + let store2 = + JobStore::new(dir.clone(), Duration::from_secs(5), Shell::sh(), db.clone()).unwrap(); + let r = store2 + .run( + "echo second".into(), + None, + None, + false, + false, + Some("t".into()), + ) + .await + .unwrap(); + let second = match r { + RunResult::Inline { id, .. } | RunResult::Backgrounded { id } => id, + }; + assert_ne!(first, second, "retained id must never be reused"); + let old_log = tokio::fs::read_to_string(dir.join(format!("{first}.log"))) + .await + .unwrap(); + assert!( + old_log.contains("first"), + "old job's log must survive the new launch: {old_log:?}" + ); + } + #[tokio::test] async fn poll_surfaces_read_error_while_job_is_tracked() { let store = store(Duration::from_secs(5)); diff --git a/src/jobs/reaper.rs b/src/jobs/reaper.rs index ee11759..629a0fc 100644 --- a/src/jobs/reaper.rs +++ b/src/jobs/reaper.rs @@ -53,7 +53,10 @@ pub(super) async fn kill_job(job: &Job) -> bool { } // Give the group a chance to exit on TERM; force it with KILL otherwise. if !exited_within(job.done.clone(), KILL_GRACE).await && !signal_group(pgid, "KILL").await { - return false; + // The group can die between the grace expiring and the KILL — its own + // TERM worked, the KILL just found nothing. Report by final state, not + // by KILL delivery, so that success isn't misread as "nothing to kill". + return !matches!(*job.state.lock().await, JobState::Running); } true } @@ -255,15 +258,22 @@ async fn reap_orphans(db: &Db, dir: &Path, retention_secs: i64) { } }; let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some("log") { - continue; - } - // `.log` -> ``; ids never contain a `.`, so the stem is the id. - let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else { - continue; - }; - if known.contains(stem) { - continue; + match path.extension().and_then(|e| e.to_str()) { + Some("log") => { + // `.log` -> ``; ids never contain a `.`, so the stem is + // the id. + let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else { + continue; + }; + if known.contains(stem) { + continue; + } + } + // `.log.trim` — a temp a crashed `trim_log` never renamed. Always + // an orphan (a completed trim renames it away); the mtime gate below + // protects one belonging to a trim in progress right now. + Some("trim") => {} + _ => continue, } // Orphan: drop it only once it's aged past retention by mtime, so a log // whose row hasn't been written yet (a brief race) isn't deleted early. @@ -432,6 +442,45 @@ mod tests { assert!(kill_group(2_000_000_000).await); } + #[tokio::test] + async fn reap_orphans_removes_aged_trim_temps_and_keeps_known_logs() { + let dir = tempfile::tempdir().unwrap(); + let db = crate::db::Db::memory(); + db.call(|conn| { + conn.execute( + "INSERT INTO jobs (id, status, started_unix) VALUES ('known', 'exited', 0)", + [], + ) + }) + .await + .unwrap(); + + // A crashed trim's temp, an orphan log, and a known job's log. + let trim_tmp = dir.path().join("dead.log.trim"); + let orphan = dir.path().join("orphan.log"); + let known = dir.path().join("known.log"); + for p in [&trim_tmp, &orphan, &known] { + tokio::fs::write(p, "x\n").await.unwrap(); + // Age past any retention: mtime at the epoch. + let status = tokio::process::Command::new("touch") + .args(["-d", "@0"]) + .arg(p) + .status() + .await + .unwrap(); + assert!(status.success()); + } + + reap_orphans(&db, dir.path(), RETENTION_SECS).await; + + assert!( + !trim_tmp.exists(), + "aged .log.trim temp must be swept — no reaper path deleted it before" + ); + assert!(!orphan.exists(), "aged orphan log must be swept"); + assert!(known.exists(), "a log with a matching row must be kept"); + } + #[tokio::test] async fn trim_log_keeps_the_tail_with_a_marker() { let dir = tempfile::tempdir().unwrap();