fix: let the parsed time decide when claim_next finds a lease lost

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,8 @@ 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,
         a_task_closed_by_its_continuation_opens_what_waited_on_it,
@@ -2135,6 +2150,62 @@ 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));
+    }
+
+    /// 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<T>(r: &Result<T, Error>, given: u64, current: u64) -> bool {
         matches!(r, Err(Error::Fenced { given: g, current: c, .. }) if *g == given && *c == current)
     }

crates/ledger/src/postgres.rs

@@ -583,54 +583,50 @@ 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(
-                &format!(
-                    "SELECT {TASK_COLUMNS} FROM {s}.tasks WHERE {claimable_now}
-                     ORDER BY priority DESC, created"
-                ),
-                &[&now_text],
+                &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::<Result<Vec<_>, _>>()?;
+        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 = $2 AND {claimable_now}
-                         FOR UPDATE SKIP LOCKED"
-                    ),
-                    &[&now_text, &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<bool, Error> {
+        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,
@@ -840,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<C: GenericClient>(
+    tx: &C,
+    s: &str,
+    id: &str,
+    now: OffsetDateTime,
+    who: (&str, OffsetDateTime),
+    manifest: &[String],
+    handling: Label,
+) -> Result<bool, Error> {
+    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<C: GenericClient>(conn: &C, domain: &str) -> Result<(), Error> {
     conn.execute(

crates/ledger/src/sqlite.rs

@@ -367,14 +367,16 @@ impl Db {
         let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
         let candidates: Vec<Task> = {
             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))
-                 ORDER BY priority DESC, created"
+                 WHERE domain = ?1 AND status IN ('open', 'blocked', 'claimed')"
             ))?;
-            let rows = stmt.query_map(params![domain, crate::fmt_time(now)], |r| {
-                Ok(task_from_row(r))
-            })?;
-            rows.map(|r| r?).collect::<Result<_, _>>()?
+            let rows = stmt.query_map(params![domain], |r| Ok(task_from_row(r)))?;
+            let mut found: Vec<Task> = rows.map(|r| r?).collect::<Result<_, _>>()?;
+            crate::in_queue_order(&mut found);
+            found
         };
         for task in candidates {
             if !claimable(&task, now, manifest, handling)