From 8ed964ced26e386361e953568d30b7325e45788b Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Thu, 12 Nov 2026 08:43:33 +0100 Subject: [PATCH 1/2] fix: let the parsed time decide when claim_next finds a lease lost `claim_next` selected the claimed tasks whose `lease_until` was before the present time, and compared the two times as text. A time renders with as many digits of a second as it has. A lease that ended at `10:05:00Z` was thus not lost at `10:05:00.5Z`, and `claim_next` gave no task for up to one second. The named `claim` gave the task at the same time, because it parses the times. The error was on the safe side: a task never had two holders. But a lost lease did not go back to the pool at its time. The two backends now select each open, blocked and claimed task. `claimable` decides with the parsed times, as it did after the selection. The adversarial pass on the taint change found this. The new assertion failed on each backend before the change: `called Option::unwrap() on a None value`, where `claim_next` gave no task. Co-Authored-By: Claude Fable 5.2 --- crates/ledger/src/lib.rs | 14 ++++++++++++++ crates/ledger/src/postgres.rs | 13 +++++++------ crates/ledger/src/sqlite.rs | 9 +++++---- 3 files changed, 26 insertions(+), 10 deletions(-) diff --git a/crates/ledger/src/lib.rs b/crates/ledger/src/lib.rs index 5b2e4d14..a25d601c 100644 --- a/crates/ledger/src/lib.rs +++ b/crates/ledger/src/lib.rs @@ -1731,6 +1731,7 @@ mod tests { } on_both! { + a_lease_that_ran_out_half_a_second_ago_is_lost_to_claim_next, a_dependent_in_another_domain_is_opened_and_retriaged_too, a_blocked_task_whose_dependencies_are_done_is_claimable_whatever_its_state_says, a_task_closed_by_its_continuation_opens_what_waited_on_it, @@ -2135,6 +2136,19 @@ mod tests { ); } + /// The lease ends on a whole second and the clock stands a half second after: the two + /// times, as text, sort the other way. + async fn a_lease_that_ran_out_half_a_second_ago_is_lost_to_claim_next(b: &Site) { + let (mut l, now) = b.ledger().await; + let t = admitted(&mut l, spec("Long job")).await; + claim(&mut l, &t.id, "s1", mins(5)).await.unwrap(); + *now.lock().unwrap() = T0 + mins(5) - Duration::milliseconds(500); + assert!(next(&mut l, "s2", mins(5)).await.is_none(), "still held"); + *now.lock().unwrap() = T0 + mins(5) + Duration::milliseconds(500); + let c = next(&mut l, "s2", mins(5)).await.unwrap(); + assert_eq!((c.task.id.as_str(), c.generation), (t.id.as_str(), 2)); + } + fn is_fenced(r: &Result, given: u64, current: u64) -> bool { matches!(r, Err(Error::Fenced { given: g, current: c, .. }) if *g == given && *c == current) } diff --git a/crates/ledger/src/postgres.rs b/crates/ledger/src/postgres.rs index bf65e7d1..e9e90136 100644 --- a/crates/ledger/src/postgres.rs +++ b/crates/ledger/src/postgres.rs @@ -583,9 +583,10 @@ impl Db { let Some(s) = self.schema(domain).await? else { return Ok(None); }; - let claimable_now = - "(status IN ('open', 'blocked') OR (status = 'claimed' AND lease_until < $1))"; - let now_text = crate::fmt_time(now); + // every task that may be claimable. Whether a claimed one's lease has run out is + // asked of the parsed time: as text, a time with fewer digits of a second sorts + // after a later one with more + let claimable_now = "status IN ('open', 'blocked', 'claimed')"; let tx = self.client.transaction().await?; let rows = tx .query( @@ -593,7 +594,7 @@ impl Db { "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE {claimable_now} ORDER BY priority DESC, created" ), - &[&now_text], + &[], ) .await?; for row in &rows { @@ -604,10 +605,10 @@ impl Db { let Some(locked) = tx .query_opt( &format!( - "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE id = $2 AND {claimable_now} + "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE id = $1 AND {claimable_now} FOR UPDATE SKIP LOCKED" ), - &[&now_text, &seen.id], + &[&seen.id], ) .await? else { diff --git a/crates/ledger/src/sqlite.rs b/crates/ledger/src/sqlite.rs index 8684dd0c..55476eae 100644 --- a/crates/ledger/src/sqlite.rs +++ b/crates/ledger/src/sqlite.rs @@ -367,13 +367,14 @@ impl Db { let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let candidates: Vec = { let mut stmt = tx.prepare(&format!( + // every task that may be claimable. Whether a claimed one's lease has run + // out is asked of the parsed time below: as text, a time with fewer + // digits of a second sorts after a later one with more "SELECT {TASK_COLUMNS} FROM tasks - WHERE domain = ?1 AND (status IN ('open', 'blocked') OR (status = 'claimed' AND lease_until < ?2)) + WHERE domain = ?1 AND status IN ('open', 'blocked', 'claimed') ORDER BY priority DESC, created" ))?; - let rows = stmt.query_map(params![domain, crate::fmt_time(now)], |r| { - Ok(task_from_row(r)) - })?; + let rows = stmt.query_map(params![domain], |r| Ok(task_from_row(r)))?; rows.map(|r| r?).collect::>()? }; for task in candidates { From 12bcc507fcc7e6800f10deb13d6c8a21a5145b2d Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Thu, 12 Nov 2026 08:59:45 +0100 Subject: [PATCH 2/2] fix: order the queue by time, and test the check that stops a second holder The adversarial pass ran on 8ed964c and confirmed two findings. - On PostgreSQL, `claim_next` reads the queue and then locks one row. A different connection can claim that task and commit between the two. The check after the lock then stops a second holder. On `main`, the time term in the query was a second guard, and 8ed964c removed it. With the check off, all ledger tests passed. The check is now in a function of its own, `take`, and a test gives it a task that a different holder claimed. - The queue was in the order of `created` as text. Of two tasks of one priority, filed at `10:00:00Z` and `10:00:00.5Z`, `claim_next` gave the second one first. The two backends now put the candidates in order by the parsed time. New assertions that failed before their fix: - the queue order: the pass ran it on 8ed964c, and `claim_next` gave "Late" on the two backends; - the check after the lock, with the check off: `assertion failed: !db.take_alone(..)`. Co-Authored-By: Claude Fable 5.2 --- crates/ledger/src/lib.rs | 57 +++++++++++++++++++++ crates/ledger/src/postgres.rs | 95 +++++++++++++++++++++++------------ crates/ledger/src/sqlite.rs | 7 +-- 3 files changed, 125 insertions(+), 34 deletions(-) diff --git a/crates/ledger/src/lib.rs b/crates/ledger/src/lib.rs index a25d601c..d597c464 100644 --- a/crates/ledger/src/lib.rs +++ b/crates/ledger/src/lib.rs @@ -677,6 +677,19 @@ fn claimable(task: &Task, now: OffsetDateTime, manifest: &[String], handling: La } /// Whether `by` acts under the claim in force: its generation first, since a claim made +/// The queue's order: the highest priority first, and of equal priorities the task +/// filed first. The times are compared as times; as text, one with fewer digits of a +/// second sorts after a later one with more. +pub(crate) fn in_queue_order(tasks: &mut [Task]) { + tasks.sort_by(|a, b| { + b.spec + .priority + .cmp(&a.spec.priority) + .then(a.created.cmp(&b.created)) + .then_with(|| a.id.cmp(&b.id)) + }); +} + /// since is the reason to stop whoever the caller is, then its name and the lease. pub(crate) fn check_holder(task: &Task, by: Held<'_>, now: OffsetDateTime) -> Result<(), Error> { if by.generation != task.generation { @@ -1731,6 +1744,7 @@ mod tests { } on_both! { + of_two_tasks_of_one_priority_the_one_filed_first_is_claimed_first, a_lease_that_ran_out_half_a_second_ago_is_lost_to_claim_next, a_dependent_in_another_domain_is_opened_and_retriaged_too, a_blocked_task_whose_dependencies_are_done_is_claimable_whatever_its_state_says, @@ -2149,6 +2163,49 @@ mod tests { assert_eq!((c.task.id.as_str(), c.generation), (t.id.as_str(), 2)); } + /// Filed half a second apart: the earlier time has fewer digits of a second, and + /// as text sorts after the later one. + async fn of_two_tasks_of_one_priority_the_one_filed_first_is_claimed_first(b: &Site) { + let (mut l, now) = b.ledger().await; + let early = admitted(&mut l, spec("Early")).await; + *now.lock().unwrap() = T0 + Duration::milliseconds(500); + let late = admitted(&mut l, spec("Late")).await; + assert_eq!(next(&mut l, "s", mins(5)).await.unwrap().task.id, early.id); + assert_eq!(next(&mut l, "s", mins(5)).await.unwrap().task.id, late.id); + } + + /// The candidate was claimed by another connection, and that committed, between the + /// read of the queue and the lock of its row. + #[tokio::test] + async fn on_postgres_a_candidate_claimed_meanwhile_is_not_taken() { + let Some(server) = PgServer::start().await.unwrap() else { + return; + }; + let mut l = Ledger::connect(server.url()).await.unwrap(); + let t = admitted(&mut l, spec("One holder")).await; + let other = admitted(&mut l, spec("Free")).await; + let now = l.now(); + claim(&mut l, &t.id, "first", mins(30)).await.unwrap(); + let Backend::Postgres(db) = &mut l.backend else { + unreachable!("a ledger connected to PostgreSQL"); + }; + let until = now + mins(30); + assert!(!db.take_alone(&t.id, now, ("second", until)).await.unwrap()); + assert!(db + .take_alone(&other.id, now, ("second", until)) + .await + .unwrap()); + let held = l.get(&t.id).await.unwrap(); + assert_eq!( + (held.holder.as_deref(), held.generation), + (Some("first"), 1) + ); + assert_eq!( + l.get(&other.id).await.unwrap().holder.as_deref(), + Some("second") + ); + } + fn is_fenced(r: &Result, given: u64, current: u64) -> bool { matches!(r, Err(Error::Fenced { given: g, current: c, .. }) if *g == given && *c == current) } diff --git a/crates/ledger/src/postgres.rs b/crates/ledger/src/postgres.rs index e9e90136..df8c7a80 100644 --- a/crates/ledger/src/postgres.rs +++ b/crates/ledger/src/postgres.rs @@ -590,48 +590,43 @@ impl Db { let tx = self.client.transaction().await?; let rows = tx .query( - &format!( - "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE {claimable_now} - ORDER BY priority DESC, created" - ), + &format!("SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE {claimable_now}"), &[], ) .await?; - for row in &rows { - let seen = task_from_row(row)?; + let mut candidates = rows + .iter() + .map(task_from_row) + .collect::, _>>()?; + crate::in_queue_order(&mut candidates); + for seen in candidates { if !claimable(&seen, now, manifest, handling) { continue; } - let Some(locked) = tx - .query_opt( - &format!( - "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE id = $1 AND {claimable_now} - FOR UPDATE SKIP LOCKED" - ), - &[&seen.id], - ) - .await? - else { - continue; - }; - let task = task_from_row(&locked)?; - if !claimable(&task, now, manifest, handling) - || !deps_complete(&tx, &task.spec.depends_on).await? - || awaits_continuation(&tx, &s, &task.id).await? - { - continue; + let who = (holder, until); + if take(&tx, &s, &seen.id, now, who, manifest, handling).await? { + tx.commit().await?; + return Ok(Some(seen.id)); } - let claimed = Event::Claimed { - holder: holder.to_string(), - until, - }; - apply(&tx, &s, &task.id, now, &claimed).await?; - tx.commit().await?; - return Ok(Some(task.id)); } Ok(None) } + /// [`take`] in a transaction of its own, for a test of what it refuses. + #[cfg(test)] + pub(crate) async fn take_alone( + &mut self, + id: &str, + now: OffsetDateTime, + who: (&str, OffsetDateTime), + ) -> Result { + let s = self.schema_of_task(id).await?; + let tx = self.client.transaction().await?; + let taken = take(&tx, &s, id, now, who, &[], Label::Restricted).await?; + tx.commit().await?; + Ok(taken) + } + pub(crate) async fn claim( &mut self, id: &str, @@ -841,6 +836,44 @@ impl Db { } } +/// Lock one candidate's row and claim it for `who`, a holder and the end of its lease, if +/// the task is still claimable now that the row is held. The candidate was read before +/// the lock: another connection may have claimed it and committed since, and this check +/// is then all that stands between the task and a second holder. A row another +/// connection holds is passed over. +async fn take( + tx: &C, + s: &str, + id: &str, + now: OffsetDateTime, + who: (&str, OffsetDateTime), + manifest: &[String], + handling: Label, +) -> Result { + let locked = tx + .query_opt( + &format!("SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE id = $1 FOR UPDATE SKIP LOCKED"), + &[&id], + ) + .await?; + let Some(locked) = locked else { + return Ok(false); + }; + let task = task_from_row(&locked)?; + if !claimable(&task, now, manifest, handling) + || !deps_complete(tx, &task.spec.depends_on).await? + || awaits_continuation(tx, s, &task.id).await? + { + return Ok(false); + } + let claimed = Event::Claimed { + holder: who.0.to_string(), + until: who.1, + }; + apply(tx, s, &task.id, now, &claimed).await?; + Ok(true) +} + /// Take a domain's lock until the transaction ends. async fn lock_domain(conn: &C, domain: &str) -> Result<(), Error> { conn.execute( diff --git a/crates/ledger/src/sqlite.rs b/crates/ledger/src/sqlite.rs index 55476eae..42ddf9e6 100644 --- a/crates/ledger/src/sqlite.rs +++ b/crates/ledger/src/sqlite.rs @@ -371,11 +371,12 @@ impl Db { // out is asked of the parsed time below: as text, a time with fewer // digits of a second sorts after a later one with more "SELECT {TASK_COLUMNS} FROM tasks - WHERE domain = ?1 AND status IN ('open', 'blocked', 'claimed') - ORDER BY priority DESC, created" + WHERE domain = ?1 AND status IN ('open', 'blocked', 'claimed')" ))?; let rows = stmt.query_map(params![domain], |r| Ok(task_from_row(r)))?; - rows.map(|r| r?).collect::>()? + let mut found: Vec = rows.map(|r| r?).collect::>()?; + crate::in_queue_order(&mut found); + found }; for task in candidates { if !claimable(&task, now, manifest, handling)