Cancel runs if no progress is made in the manifest

Cargo.toml

@@ -43,6 +43,7 @@ tokio = { version = "1.26.0", features = [
 ] }
 tokio-rustls = "0.23.4"
 futures = "0.3.27"
+pin-project-lite = "0.2.9"
 
 parking_lot = "0.12.1"
 moka = { version = "0.10.0", features = ["future"] }

crates/abq_cli/src/instance.rs

@@ -1,6 +1,7 @@
 use abq_queue::persistence;
 use abq_queue::persistence::remote::NoopPersister;
 use abq_queue::queue::{Abq, QueueConfig};
+use abq_queue::RunTimeoutStrategy;
 use abq_utils::auth::{AdminToken, ServerAuthStrategy, UserToken};
 use abq_utils::exit::ExitCode;
 use abq_utils::net_opt::ServerOptions;
@@ -57,6 +58,8 @@ pub async fn start_abq_forever(
         RESULTS_PERSISTENCE_LRU_CAPACITY,
     );
 
+    let run_timeout_strategy = RunTimeoutStrategy::RUN_BASED;
+
     let queue_config = QueueConfig {
         public_ip,
         bind_ip,
@@ -66,6 +69,7 @@ pub async fn start_abq_forever(
         server_options,
         persist_manifest,
         persist_results,
+        run_timeout_strategy,
     };
     let mut abq = Abq::start(queue_config).await;
 

crates/abq_queue/Cargo.toml

@@ -23,6 +23,7 @@ tracing.workspace = true
 tokio.workspace = true
 futures.workspace = true
 async-trait.workspace = true
+pin-project-lite.workspace = true
 
 anyhow.workspace = true
 

crates/abq_queue/src/job_queue.rs

@@ -120,6 +120,11 @@ impl JobQueue {
 
         persistence::manifest::ManifestView::new(queue, assigned_entities)
     }
+
+    /// Reads the current index at a given point in time. Not atomic.
+    pub fn read_index(&self) -> usize {
+        self.ptr.load(atomic::ORDERING)
+    }
 }
 
 #[cfg(test)]

crates/abq_queue/src/lib.rs

@@ -5,5 +5,8 @@ mod prelude;
 mod job_queue;
 pub mod persistence;
 pub mod queue;
+mod timeout;
 mod worker_timings;
 mod worker_tracking;
+
+pub use timeout::{RunTimeoutStrategy, TimeoutReason};