feat(gateway): bench non-delivering winners; 6-min image receipt timeout; v0.2.68

Cargo.toml

@@ -23,7 +23,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.2.67"
+version = "0.2.68"
 edition = "2024"
 license = "MIT"
 repository = "https://github.com/profullstack/c0mpute"

apps/tui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@c0mpute/tui",
-  "version": "0.2.67",
+  "version": "0.2.68",
   "private": true,
   "type": "module",
   "bin": {

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "c0mpute",
-  "version": "0.2.67",
+  "version": "0.2.68",
   "private": true,
   "type": "module",
   "scripts": {

apps/web/src/app/releases/latest.json/route.ts

@@ -28,7 +28,7 @@ export const dynamic = "force-dynamic";
  * this in sync with the workspace version in `Cargo.toml` on each release so a
  * freshly built CLI reports "already latest" rather than a phantom upgrade.
  */
-const FALLBACK_VERSION = "0.2.67";
+const FALLBACK_VERSION = "0.2.68";
 const FALLBACK_MIN_REQUIRED = "0.0.1";
 
 type Channel = "stable" | "beta" | "nightly";

node/crates/c0mpute-core/src/lib.rs

@@ -14,6 +14,7 @@ pub(crate) mod gateway_util;
 pub mod jobs_gateway;
 pub mod net_guard;
 pub mod opendisk;
+pub mod penalty_box;
 pub mod pricing;
 pub mod reachability;
 pub mod private;

node/crates/c0mpute-core/src/penalty_box.rs

@@ -0,0 +1,89 @@
+//! Buyer-side memory of workers that won and did not deliver.
+//!
+//! A worker that wins an auction and then fails the job, never sends a
+//! receipt, or cannot be fetched from is benched for this gateway's
+//! auctions: 30 minutes for the first strike, doubling per strike up to
+//! 6 hours. A delivered job clears the record. Without this, a fast-bidding
+//! worker that cannot deliver (behind NAT, too slow, broken runtime) wins
+//! every auction and every caller waits out the full receipt timeout.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+pub const FIRST_BENCH: Duration = Duration::from_secs(30 * 60);
+pub const MAX_BENCH: Duration = Duration::from_secs(6 * 60 * 60);
+
+#[derive(Clone, Default)]
+pub struct PenaltyBox {
+    inner: Arc<Mutex<HashMap<String, (Instant, u32)>>>,
+}
+
+/// How long the `strikes`-th strike benches a worker.
+pub fn bench_for(strikes: u32) -> Duration {
+    let factor = 1u32.checked_shl(strikes.saturating_sub(1)).unwrap_or(u32::MAX);
+    FIRST_BENCH.saturating_mul(factor).min(MAX_BENCH)
+}
+
+impl PenaltyBox {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Peers benched at `now`.
+    pub fn benched_at(&self, now: Instant) -> Vec<String> {
+        let map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
+        map.iter()
+            .filter(|(_, (until, _))| *until > now)
+            .map(|(peer, _)| peer.clone())
+            .collect()
+    }
+
+    pub fn benched(&self) -> Vec<String> {
+        self.benched_at(Instant::now())
+    }
+
+    /// Record a non-delivery; returns how long the peer is benched.
+    pub fn strike_at(&self, peer: &str, now: Instant) -> Duration {
+        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
+        let strikes = map.get(peer).map(|(_, s)| s + 1).unwrap_or(1);
+        let bench = bench_for(strikes);
+        map.insert(peer.to_string(), (now + bench, strikes));
+        bench
+    }
+
+    pub fn strike(&self, peer: &str) -> Duration {
+        self.strike_at(peer, Instant::now())
+    }
+
+    /// A delivered job wipes the record.
+    pub fn clear(&self, peer: &str) {
+        self.inner.lock().unwrap_or_else(|e| e.into_inner()).remove(peer);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn bench_doubles_and_caps() {
+        assert_eq!(bench_for(1), Duration::from_secs(30 * 60));
+        assert_eq!(bench_for(2), Duration::from_secs(60 * 60));
+        assert_eq!(bench_for(3), Duration::from_secs(120 * 60));
+        assert_eq!(bench_for(10), MAX_BENCH);
+        assert_eq!(bench_for(100), MAX_BENCH);
+    }
+
+    #[test]
+    fn struck_peers_are_benched_until_expiry_and_cleared_on_delivery() {
+        let pb = PenaltyBox::new();
+        let t0 = Instant::now();
+        pb.strike_at("a", t0);
+        assert_eq!(pb.benched_at(t0 + Duration::from_secs(60)), vec!["a".to_string()]);
+        assert!(pb.benched_at(t0 + FIRST_BENCH + Duration::from_secs(1)).is_empty());
+        assert_eq!(pb.strike_at("a", t0), Duration::from_secs(60 * 60), "second strike doubles");
+        pb.clear("a");
+        assert!(pb.benched_at(t0).is_empty());
+    }
+}

node/crates/c0mpute-core/src/speech_gateway.rs

@@ -27,6 +27,9 @@ const MAX_AUDIO_BYTES: usize = 25 * 1024 * 1024;
 const MAX_WAV_BYTES: usize = 20 * 1024 * 1024;
 const MAX_SPEECH_WORKER_ATTEMPTS: usize = 3;
 const MAX_CONCURRENT_IMAGE_JOBS: usize = 4;
+/// A healthy CPU worker draws an image in ~2 minutes. One that has not
+/// answered in 6 is benched and the job goes to the next worker.
+const IMAGE_RECEIPT_TIMEOUT: Duration = Duration::from_secs(6 * 60);
 
 #[derive(Clone)]
 struct GatewayState {
@@ -40,6 +43,8 @@ struct GatewayState {
     image_slots: Arc<Semaphore>,
     /// Warm, pinned sessions this node is holding (see `speech_session`).
     sessions: crate::speech_session::SessionStore,
+    /// Workers that won and did not deliver, benched for a while.
+    penalties: crate::penalty_box::PenaltyBox,
 }
 
 #[derive(Debug)]
@@ -130,6 +135,7 @@ pub async fn serve(bind: std::net::SocketAddr, token: String, max_price_usd: f64
         stt_slots: Arc::new(Semaphore::new(2)),
         image_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_IMAGE_JOBS)),
         sessions: crate::speech_session::SessionStore::new(),
+        penalties: crate::penalty_box::PenaltyBox::new(),
     };
     let app = Router::new()
         .route("/healthz", get(healthz))
@@ -440,29 +446,41 @@ async fn run_speech_auction(
     spec: serde_json::Value,
     required: Vec<String>,
     excluded_bidders: &[String],
-) -> ApiResult<crate::AuctionOutcome> {
+) -> Result<crate::AuctionOutcome, (ApiError, Option<String>)> {
+    // Learn the winner as soon as it is chosen, so a winner that then goes
+    // silent can be benched and the job re-auctioned.
+    let (winner_tx, mut winner_rx) = tokio::sync::mpsc::unbounded_channel();
     let auction = JobAuction::new(workload, spec)
+        .with_accept_notify(winner_tx)
         .with_required_capabilities(required)
         .with_excluded_bidders(excluded_bidders.to_vec())
         .with_max_price_usd(state.max_price_usd)
         .with_bid_window(Duration::from_secs(4))
         .with_deadline(if is_video(workload) {
             crate::video::JOB_TIMEOUT + Duration::from_secs(5 * 60)
+        } else if workload == crate::diffusion::WORKLOAD_TXT2IMG {
+            IMAGE_RECEIPT_TIMEOUT + Duration::from_secs(60)
         } else {
             Duration::from_secs(20 * 60)
         })
         .with_receipt_timeout(if is_video(workload) {
             crate::video::JOB_TIMEOUT
+        } else if workload == crate::diffusion::WORKLOAD_TXT2IMG {
+            IMAGE_RECEIPT_TIMEOUT
         } else {
             Duration::from_secs(15 * 60)
         });
     run_auction(state.net.clone(), auction)
         .await
         .map_err(|error| {
-            tracing::warn!(workload, err = %error, "speech auction failed");
-            ApiError(
-                StatusCode::SERVICE_UNAVAILABLE,
-                "no c0mpute worker completed the speech job".into(),
+            let winner = winner_rx.try_recv().ok().map(|(_, bid)| bid.bidder_peer_id);
+            tracing::warn!(workload, err = %error, winner = ?winner, "speech auction failed");
+            (
+                ApiError(
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "no c0mpute worker completed the speech job".into(),
+                ),
+                winner,
             )
         })
 }
@@ -478,7 +496,8 @@ async fn run_speech_job(
     required: Vec<String>,
     max_bytes: usize,
 ) -> ApiResult<(crate::AuctionOutcome, Vec<u8>)> {
-    let mut excluded_bidders = Vec::new();
+    // Workers benched for not delivering earlier stay out of the auction.
+    let mut excluded_bidders = state.penalties.benched();
     let mut last_error = None;
 
     for attempt in 1..=MAX_SPEECH_WORKER_ATTEMPTS {
@@ -492,7 +511,17 @@ async fn run_speech_job(
         .await
         {
             Ok(outcome) => outcome,
-            Err(error) => return Err(last_error.unwrap_or(error)),
+            // A winner was chosen and then never answered: bench it and try
+            // the next worker instead of failing the caller.
+            Err((error, Some(winner))) => {
+                let bench = state.penalties.strike(&winner);
+                tracing::warn!(workload, attempt, worker = %winner, bench_secs = bench.as_secs(),
+                    "winner never delivered; benched, trying another worker");
+                excluded_bidders.push(winner);
+                last_error = Some(error);
+                continue;
+            }
+            Err((error, None)) => return Err(last_error.unwrap_or(error)),
         };
         let worker = outcome.accepted_bid.bidder_peer_id.clone();
         let result = if outcome.receipt.status != c0mpute_net::topics::JobStatus::Completed {
@@ -517,8 +546,12 @@ async fn run_speech_job(
         };
 
         match result {
-            Ok(bytes) => return Ok((outcome, bytes)),
+            Ok(bytes) => {
+                state.penalties.clear(&worker);
+                return Ok((outcome, bytes));
+            }
             Err(error) => {
+                state.penalties.strike(&worker);
                 tracing::warn!(
                     workload,
                     attempt,