fix(jobs): id reuse across restarts, same-second reconcile, kill race, trim-temp orphans

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 `-<seq>` from `seq`, so clean ids stay clean.
+    /// Suffixed candidates are re-checked too: the sequence restarts at 1 each
+    /// boot, so `<base>-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 `<base>-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);

src/jobs/mod.rs

@@ -202,29 +202,45 @@ impl JobStore {
         db: crate::db::Db,
     ) -> std::io::Result<Self> {
         std::fs::create_dir_all(&dir)?;
-        let jobs = Arc::new(Mutex::new(HashMap::new()));
+        let jobs: Arc<Mutex<HashMap<JobId, Arc<Job>>>> = 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<String> = 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));

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;
-        }
-        // `<id>.log` -> `<id>`; 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") => {
+                // `<id>.log` -> `<id>`; 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;
+                }
+            }
+            // `<id>.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();