From d1f1a361f0a896178bb19c731e70e55f6cbfa0f0 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:33:24 +0200 Subject: [PATCH] =?UTF-8?q?feat(#2949):=20claim=5Fnext=20=E2=80=94=20the?= =?UTF-8?q?=20seam=20that=20cannot=20be=20half-used?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The caller supplies how to run a node and spawns what it gets back; it never touches claiming or completion. The returned future runs the node *and completes it*, so "forgot to finish the node" stops being something a caller can do — completion is inside the thing they spawn. The `Option` is answered synchronously, before anything is awaited, so the run loop learns whether there was work without waiting on the node it just started. That is what lets it choose between claiming again immediately and backing off; an id alone cannot express that choice. Locking: taken twice, briefly, and never held across the await — once to claim, once inside the future to complete. A guard alive across an await point would make the future non-`Send` and unspawnable, which is also why the node itself runs unlocked for however long it takes. `Arc` + `std::sync::Mutex` keep this runtime-agnostic: no tokio in this crate. `run` receives an owned payload rather than a borrow for the same reason a `&Job` could not be threaded through the executors: a reference parameter is live for the whole future, borrowing the graph across the await and poisoning `Send`. The output carries the insert result instead of swallowing it. This crate has no logger by design, so a malformed grown job is reported to the caller, who can log it. The node completes either way — its own work already happened. --- hive-jobq/src/scheduler.rs | 56 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 3fd46c3d..8b1e0413 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -29,7 +29,9 @@ //! is a deferred optimization — unsafe under dynamically-appended subnodes.) use std::collections::HashMap; +use std::future::Future; use std::hash::Hash; +use std::sync::{Arc, Mutex}; use crate::builder::{BuildError, JobBuilder, NodeGuid}; use crate::resources::ResourceTable; @@ -160,6 +162,60 @@ impl Scheduler { .find(|&id| self.node_deps_satisfied(id) && self.try_start(id)) } + /// Claim one runnable node and return **the work that runs it**, or `None` + /// when nothing is runnable right now. + /// + /// This is the seam: the caller supplies how to execute a node and spawns + /// the returned future, but never touches claiming or completion. The + /// future runs the node **and completes it**, so "forgot to finish the + /// node" is not expressible — completion is inside the thing you spawn. + /// + /// The `Option` is answered *synchronously*, before anything is awaited, so + /// the caller can decide "claim again immediately" vs "back off" without + /// waiting on the node it just started. + /// + /// ## Locking + /// The lock is taken twice, briefly, and **never held across the await**: + /// once here to claim, once inside the future to complete. That is what + /// keeps the returned future `Send` — a guard alive across an await point + /// would poison it — and it is why the node itself runs unlocked, for + /// however many minutes it needs. + /// + /// ## Why the payload is cloned + /// `run` gets an owned `N` rather than a borrow: a `&N` parameter is live + /// for the whole future, which both borrows the graph across the await and + /// makes the future non-`Send`. + /// + /// The output carries the insert result rather than swallowing it — this + /// crate has no logger, so a malformed grown job is reported to the caller, + /// who is the one that can log it. The node is completed either way: its + /// own work already happened. + pub fn claim_next( + sched: &Arc>, + run: F, + ) -> Option)> + use> + where + N: Clone, + F: FnOnce(NodeId, N, JobBuilder) -> Fut, + Fut: Future, Outcome)>, + { + let (id, payload) = { + let mut guard = sched.lock().expect("jobq scheduler mutex poisoned"); + let id = guard.claim_one()?; + let payload = guard.graph.node(id)?.payload.clone(); + (id, payload) + }; + let sched = Arc::clone(sched); + Some(async move { + let (grown, outcome) = run(id, payload, JobBuilder::new()).await; + let grew = sched + .lock() + .expect("jobq scheduler mutex poisoned") + .complete_growing(id, outcome, grown); + (id, grew) + }) + } + /// Claim every currently-runnable pending node. Equivalent to calling /// [`Self::claim_one`] until it yields `None`: a node started by an earlier /// iteration is `Running`, not terminal, so it cannot satisfy another