From db15e230fab72282ea32592203fd25f30263c956 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Mon, 10 Aug 2026 07:50:59 +0000 Subject: [PATCH] fix(prism): autonomous Lium rent pool with 429 requeue Serialize rents against Lium's 3/5s and 60/h caps, cool down from 429 bodies, requeue without burning retry budget, and recover last-6h rate-limited failures into the queue automatically. --- Cargo.lock | 12 + bins/prism-challenge/Cargo.toml | 2 + bins/prism-challenge/src/main.rs | 63 ++++- crates/db/src/prism_store.rs | 17 +- crates/lium-rent-pool/Cargo.toml | 19 ++ crates/lium-rent-pool/src/lib.rs | 260 +++++++++++++++++++++ crates/prism-challenge/Cargo.toml | 1 + crates/prism-challenge/src/api.rs | 2 +- crates/prism-challenge/src/orchestrator.rs | 46 ++-- crates/prism-emit/tests/epoch_semantics.rs | 2 +- crates/prism-lium/Cargo.toml | 1 + crates/prism-lium/src/client.rs | 43 ++-- crates/prism-store/src/dbprism.rs | 12 +- crates/prism-store/src/store.rs | 24 +- docs/PRISM.md | 6 +- 15 files changed, 451 insertions(+), 59 deletions(-) create mode 100644 crates/lium-rent-pool/Cargo.toml create mode 100644 crates/lium-rent-pool/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index d73a856ad..b1c19a0ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3000,6 +3000,14 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "lium-rent-pool" +version = "0.1.0" +dependencies = [ + "tokio", + "tracing", +] + [[package]] name = "lock_api" version = "0.4.14" @@ -3747,6 +3755,7 @@ dependencies = [ "db", "hex", "http-body-util", + "lium-rent-pool", "prism-challenge-task", "prism-emit", "prism-lium", @@ -3781,9 +3790,11 @@ dependencies = [ "clap", "crypto", "db", + "lium-rent-pool", "predicates", "prism-challenge", "prism-lium", + "prism-pipeline", "prism-recipe", "prism-registry", "prism-review", @@ -3821,6 +3832,7 @@ version = "0.1.0" dependencies = [ "async-trait", "hex", + "lium-rent-pool", "prism-recipe", "reqwest 0.12.28", "serde", diff --git a/bins/prism-challenge/Cargo.toml b/bins/prism-challenge/Cargo.toml index 40462eaf1..e297a29ad 100644 --- a/bins/prism-challenge/Cargo.toml +++ b/bins/prism-challenge/Cargo.toml @@ -18,8 +18,10 @@ challenge-keys = { path = "../../crates/challenge-keys" } chain = { path = "../../crates/chain" } chain-live = { path = "../../crates/chain-live" } clap = { version = "4", features = ["derive", "env"] } +lium-rent-pool = { path = "../../crates/lium-rent-pool" } prism-challenge = { path = "../../crates/prism-challenge" } prism-lium = { path = "../../crates/prism-lium" } +prism-pipeline = { path = "../../crates/prism-pipeline" } prism-recipe = { path = "../../crates/prism-recipe" } prism-registry = { path = "../../crates/prism-registry" } prism-review = { path = "../../crates/prism-review" } diff --git a/bins/prism-challenge/src/main.rs b/bins/prism-challenge/src/main.rs index 05da7cbca..9b6420baa 100644 --- a/bins/prism-challenge/src/main.rs +++ b/bins/prism-challenge/src/main.rs @@ -362,6 +362,7 @@ fn build_topmodel() -> Option> { p } +#[allow(clippy::too_many_lines)] async fn cmd_serve(cli: Cli) -> Result<(), String> { let path = resolve_sk_path(cli.challenge_sk_file.as_ref())?; if !path.is_file() { @@ -457,8 +458,12 @@ async fn cmd_serve(cli: Cli) -> Result<(), String> { ); } let orchestrator = Arc::new(orchestrator); - spawn_orchestrator(&cli, &orchestrator); - + spawn_orchestrator( + &cli, + &orchestrator, + Arc::clone(&state.store), + gating_enabled.then_some(gating), + ); let listener = TcpListener::bind(cli.bind) .await .map_err(|e| format!("bind {}: {e}", cli.bind))?; @@ -502,7 +507,12 @@ fn spawn_epoch_feed(chain_ep: &str, state: &Arc) { }); } -fn spawn_orchestrator(cli: &Cli, orchestrator: &Arc>) { +fn spawn_orchestrator( + cli: &Cli, + orchestrator: &Arc>, + store: Arc, + gating: Option>, +) { let permits = cli.max_concurrent_evals.max(1) as usize; let sem = Arc::new(Semaphore::new(permits)); for i in 0..permits { @@ -518,4 +528,51 @@ fn spawn_orchestrator(cli: &Cli, orchestrator: &Arc, + gating: Option<&Arc>, +) -> u32 { + #[allow(clippy::cast_possible_truncation)] + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_millis() as u64); + let Ok(failed) = store.list(Some("failed"), None, 500).await else { + return 0; + }; + let mut n = 0u32; + for row in failed { + let Some(err) = row.error_detail.as_deref() else { + continue; + }; + if !lium_rent_pool::should_recover(err, row.updated_at_ms, now) { + continue; + } + if store.reset_for_retry(&row.id, false).await.is_err() { + continue; + } + n = n.saturating_add(1); + tracing::info!(submission_id = %row.id, "requeued rate-limited submission"); + if let Some(g) = gating { + let key = prism_pipeline::gating_key(row.arch_id.as_deref()); + let _ = g.reset_open(&key, &row.miner_hotkey).await; + let _ = g.mark_registered(&key, &row.miner_hotkey, None).await; + } + } + n +} + +fn spawn_rate_limit_recovery(store: Arc, gating: Option>) { + tokio::spawn(async move { + loop { + let n = recover_rate_limited(&store, gating.as_ref()).await; + if n > 0 { + tracing::info!(requeued = n, "lium 429 recovery tick"); + } + tokio::time::sleep(Duration::from_mins(2)).await; + } + }); } diff --git a/crates/db/src/prism_store.rs b/crates/db/src/prism_store.rs index 32fb39886..f74334809 100644 --- a/crates/db/src/prism_store.rs +++ b/crates/db/src/prism_store.rs @@ -221,20 +221,31 @@ pub async fn update_prism_submission( Ok(row) } -/// Reset a failed row for a retry: clears all execution/score fields and -/// re-queues it. `retry_count` is bumped (policy enforced by the caller). +/// Reset a row for retry: clears exec/score fields and re-queues. +/// When `bump_retry`, increments `retry_count` (manual/auto infra). When +/// false, keeps attempts (Lium 429 autonomous requeue — do not burn budget). /// /// # Errors /// SQL error / 0 rows for id. pub async fn reset_prism_submission_for_retry( pool: &PgPool, id: &str, + bump_retry: bool, ) -> Result { let q = format!( - "UPDATE prism_submission SET status = 'queued', pod_id = NULL, pod_provider = NULL, receipt_json = NULL, metrics_json = NULL, bpb = NULL, review_json = NULL, similarity_json = NULL, kind = NULL, score = NULL, absence_reason = NULL, emitted_epoch = NULL, error_detail = NULL, retry_count = retry_count + 1, updated_at = now() WHERE id = $1 RETURNING {COLS}" + "UPDATE prism_submission SET \ + status = 'queued', pod_id = NULL, pod_provider = NULL, \ + receipt_json = NULL, metrics_json = NULL, bpb = NULL, \ + review_json = NULL, similarity_json = NULL, \ + kind = NULL, score = NULL, absence_reason = NULL, emitted_epoch = NULL, \ + error_detail = NULL, \ + retry_count = CASE WHEN $2 THEN retry_count + 1 ELSE retry_count END, \ + updated_at = now() \ + WHERE id = $1 RETURNING {COLS}" ); let row = sqlx::query_as::<_, PrismSubmissionRow>(&q) .bind(id) + .bind(bump_retry) .fetch_one(pool) .await?; Ok(row) diff --git a/crates/lium-rent-pool/Cargo.toml b/crates/lium-rent-pool/Cargo.toml new file mode 100644 index 000000000..fb4253f27 --- /dev/null +++ b/crates/lium-rent-pool/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "lium-rent-pool" +description = "Autonomous Lium rent rate-limit pool (3/5s + 60/h) with 429-aware backoff" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +publish = false + +[dependencies] +tokio = { version = "1", features = ["sync", "time", "macros"] } +tracing = "0.1" + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } + +[lints] +workspace = true diff --git a/crates/lium-rent-pool/src/lib.rs b/crates/lium-rent-pool/src/lib.rs new file mode 100644 index 000000000..14bc437e2 --- /dev/null +++ b/crates/lium-rent-pool/src/lib.rs @@ -0,0 +1,260 @@ +//! Autonomous Lium rent budget: **3 / 5s** burst + **60 / hour**, serialized +//! rents, and cooldown from `429` bodies (`Please try again in N seconds`). + +#![forbid(unsafe_code)] +#![allow(clippy::missing_errors_doc)] + +use std::collections::VecDeque; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use tokio::sync::Semaphore; +use tokio::time::sleep; +use tracing::{info, warn}; + +/// Lium burst ceiling on `POST /executors/{id}/rent`. +pub const BURST_MAX: usize = 3; +/// Burst window. +pub const BURST_WINDOW: Duration = Duration::from_secs(5); +/// Lium hourly rent ceiling. +pub const HOUR_MAX: usize = 60; +/// Hourly window. +pub const HOUR_WINDOW: Duration = Duration::from_hours(1); +/// Cap a single acquire sleep (hourly cooldowns are long). +pub const MAX_ACQUIRE_WAIT: Duration = Duration::from_hours(1); +/// Autonomous recovery looks back this far for failed 429 submissions. +pub const RECOVERY_WINDOW_MS: u64 = 6 * 60 * 60 * 1000; + +/// True when a failed row should re-enter the rent queue (429 within window). +#[must_use] +pub fn should_recover(error_detail: &str, updated_at_ms: u64, now_ms: u64) -> bool { + is_rate_limited(error_detail) && now_ms.saturating_sub(updated_at_ms) <= RECOVERY_WINDOW_MS +} + +/// Parse Lium / Retry-After wait hints from a 429 body or header value. +#[must_use] +pub fn parse_retry_secs(text: &str) -> Option { + if let Some(i) = text.find("try again in ") { + let rest = &text[i + "try again in ".len()..]; + let digits: String = rest.chars().take_while(char::is_ascii_digit).collect(); + if let Ok(n) = digits.parse::() { + if n > 0 { + return Some(n.min(7200)); + } + } + } + let t = text.trim(); + if !t.is_empty() && t.chars().all(|c| c.is_ascii_digit()) { + if let Ok(n) = t.parse::() { + if n > 0 { + return Some(n.min(7200)); + } + } + } + if text.contains("per 1 hour") { + return Some(120); + } + if text.contains("per 5 seconds") { + return Some(5); + } + None +} + +/// True when an error string is a Lium rate-limit (HTTP 429). +#[must_use] +pub fn is_rate_limited(msg: &str) -> bool { + let l = msg.to_ascii_lowercase(); + l.contains("429") || l.contains("too many requests") || l.contains("rate limit") +} + +struct Inner { + burst: VecDeque, + hour: VecDeque, + cooldown_until: Option, +} + +impl Inner { + fn purge(&mut self, now: Instant) { + while self + .burst + .front() + .is_some_and(|t| now.duration_since(*t) >= BURST_WINDOW) + { + self.burst.pop_front(); + } + while self + .hour + .front() + .is_some_and(|t| now.duration_since(*t) >= HOUR_WINDOW) + { + self.hour.pop_front(); + } + if self.cooldown_until.is_some_and(|u| now >= u) { + self.cooldown_until = None; + } + } + + fn wait_hint(&self, now: Instant) -> Option { + if let Some(until) = self.cooldown_until { + if now < until { + return Some(until.saturating_duration_since(now)); + } + } + if self.burst.len() >= BURST_MAX { + if let Some(oldest) = self.burst.front() { + let elapsed = now.duration_since(*oldest); + if elapsed < BURST_WINDOW { + return BURST_WINDOW.checked_sub(elapsed); + } + } + } + if self.hour.len() >= HOUR_MAX { + if let Some(oldest) = self.hour.front() { + let elapsed = now.duration_since(*oldest); + if elapsed < HOUR_WINDOW { + return HOUR_WINDOW + .checked_sub(elapsed) + .map(|d| d.min(MAX_ACQUIRE_WAIT)); + } + } + } + None + } + + fn record(&mut self, now: Instant) { + self.burst.push_back(now); + self.hour.push_back(now); + } +} + +/// Process-wide rent gate: one in-flight rent + sliding-window budgets. +pub struct RentPool { + inner: Mutex, + gate: Semaphore, +} + +impl Default for RentPool { + fn default() -> Self { + Self::new() + } +} + +impl RentPool { + /// Empty pool (budgets open). + #[must_use] + pub fn new() -> Self { + Self { + inner: Mutex::new(Inner { + burst: VecDeque::with_capacity(BURST_MAX + 1), + hour: VecDeque::with_capacity(HOUR_MAX + 1), + cooldown_until: None, + }), + gate: Semaphore::new(1), + } + } + + /// Wait until a rent is allowed; hold the serialize lock until permit drop. + pub async fn take(&self) -> RentPermit<'_> { + let owned = loop { + match self.gate.acquire().await { + Ok(p) => break p, + Err(_) => sleep(Duration::from_secs(1)).await, + } + }; + loop { + let wait = match self.inner.lock() { + Ok(mut g) => { + let now = Instant::now(); + g.purge(now); + g.wait_hint(now) + } + Err(_) => Some(Duration::from_millis(50)), + }; + if let Some(w) = wait { + let w = w.max(Duration::from_millis(200)).min(MAX_ACQUIRE_WAIT); + info!(wait_ms = w.as_millis(), "lium rent pool waiting"); + sleep(w).await; + continue; + } + if let Ok(mut g) = self.inner.lock() { + g.record(Instant::now()); + } + return RentPermit { + pool: self, + _permit: owned, + }; + } + } + + /// Apply a 429 cooldown (from body / Retry-After). + pub fn note_429(&self, retry_secs: u64) { + let secs = retry_secs.clamp(1, 7200); + if let Ok(mut g) = self.inner.lock() { + let until = Instant::now() + Duration::from_secs(secs); + g.cooldown_until = Some(match g.cooldown_until { + Some(prev) if prev > until => prev, + _ => until, + }); + } + warn!(retry_secs = secs, "lium rent pool cooldown from 429"); + } + + /// Snapshot `(burst_used, hour_used, cooldown_secs)`. + #[must_use] + pub fn stats(&self) -> (usize, usize, Option) { + let Ok(mut g) = self.inner.lock() else { + return (0, 0, None); + }; + let now = Instant::now(); + g.purge(now); + let cool = g + .cooldown_until + .map(|u| u.saturating_duration_since(now).as_secs()); + (g.burst.len(), g.hour.len(), cool) + } +} + +/// RAII permit: drops the serialize lock when the rent HTTP call finishes. +pub struct RentPermit<'a> { + pool: &'a RentPool, + _permit: tokio::sync::SemaphorePermit<'a>, +} + +impl RentPermit<'_> { + /// Forward a 429 into the pool cooldown. + pub fn rate_limited(&self, msg: &str) { + self.pool.note_429(parse_retry_secs(msg).unwrap_or(5)); + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used)] + use super::*; + + #[test] + fn parses_try_again_in_seconds() { + let s = r#"{"message":"Too many requests. You can make 60 requests per 1 hour. Please try again in 1845 seconds."}"#; + assert_eq!(parse_retry_secs(s), Some(1845)); + assert_eq!( + parse_retry_secs("Too many requests. You can make 3 requests per 5 seconds."), + Some(5) + ); + assert!(is_rate_limited( + "lium api: POST /rent -> 429 Too Many Requests" + )); + assert!(should_recover("provision: 429 rate limit", 100, 100 + 1)); + assert!(!should_recover("provision: 429", 0, RECOVERY_WINDOW_MS + 1)); + } + + #[tokio::test] + async fn burst_forces_wait() { + let pool = RentPool::new(); + for _ in 0..BURST_MAX { + drop(pool.take().await); + } + let start = Instant::now(); + drop(pool.take().await); + assert!(start.elapsed() >= Duration::from_millis(200)); + } +} diff --git a/crates/prism-challenge/Cargo.toml b/crates/prism-challenge/Cargo.toml index b8de9fb86..abc93a263 100644 --- a/crates/prism-challenge/Cargo.toml +++ b/crates/prism-challenge/Cargo.toml @@ -19,6 +19,7 @@ challenge-common = { path = "../challenge-common" } crypto = { path = "../crypto" } db = { path = "../db" } hex = "0.4" +lium-rent-pool = { path = "../lium-rent-pool" } prism-challenge-task = { path = "../prism-challenge-task" } prism-emit = { path = "../prism-emit" } prism-lium = { path = "../prism-lium" } diff --git a/crates/prism-challenge/src/api.rs b/crates/prism-challenge/src/api.rs index 6bb5f3080..5e95bde16 100644 --- a/crates/prism-challenge/src/api.rs +++ b/crates/prism-challenge/src/api.rs @@ -451,7 +451,7 @@ async fn post_retry(State(st): State>, Path(id): Path) -> let _ = g.mark_registered(&gate_key, &row.miner_hotkey, None).await; } } - match st.store.reset_for_retry(&id).await { + match st.store.reset_for_retry(&id, true).await { Ok(_) => ( StatusCode::ACCEPTED, Json(json!({"submission_id": id, "status": "queued"})), diff --git a/crates/prism-challenge/src/orchestrator.rs b/crates/prism-challenge/src/orchestrator.rs index 263e51f43..17e8e2f18 100644 --- a/crates/prism-challenge/src/orchestrator.rs +++ b/crates/prism-challenge/src/orchestrator.rs @@ -1,18 +1,5 @@ -//! Lium job orchestrator: DB-backed state machine, recovery, epoch emitter. -//! -//! Workers claim `queued` rows, run cheap source screens (copy gate, static -//! cheat patterns, AST similarity) **before** renting a Lium pod, then run -//! the recipe + master-side LLM review + agentic anti-cheat, and compute the -//! chain-facing score. Leaf emission is decoupled from finalizes: the -//! epoch-close emitter ([`prism_emit::EpochEmitter`], driven by -//! [`Orchestrator::run_emitter`]) assigns every newly-finalized row to the -//! next chain-epoch boundary's D24 set via the emission outbox -//! (`emitted_epoch` watermark + emit cursor), so independent same-epoch -//! scorers all land and each scoring run is assigned exactly once. Positive -//! scores then carry into later epochs' competition sets until superseded; -//! leaf emission applies WTA so only the single best hotkey gets Prism's -//! share. All state lives in the store, so the API is a pure projection and -//! restarts sweep orphans. +//! Lium job orchestrator: claim→screen→pod→review→score; epoch emitter via +//! [`Orchestrator::run_emitter`]. State in store; API is a projection. use std::sync::Arc; use std::time::Duration; @@ -242,21 +229,23 @@ impl Orchestrator { Ok(true) } - /// Requeue on infra-class failures while the auto-retry budget lasts - /// (`false` → caller finalizes terminal). Records the gating attempt. + /// Requeue on infra failures while auto-retry budget lasts (`false` → + /// terminal). Lium **429** requeues without burning `retry_count` / gating. async fn maybe_auto_retry(&self, row: &SubmissionState, class: &str, msg: &str) -> bool { - if row.retry_count >= self.cfg.auto_retry_max { + let rate = lium_rent_pool::is_rate_limited(msg); + if !rate && row.retry_count >= self.cfg.auto_retry_max { return false; } warn!( submission_id = %row.id, class, + rate_limited = rate, attempt = row.retry_count + 1, max = self.cfg.auto_retry_max, error = %msg, "auto-retrying submission after infra failure" ); - let _ = self.store.reset_for_retry(&row.id).await; + let _ = self.store.reset_for_retry(&row.id, !rate).await; let _ = self .store .apply( @@ -267,6 +256,7 @@ impl Orchestrator { detail: Some(serde_json::json!({ "auto_retry": true, "class": class, + "rate_limited": rate, "attempt": row.retry_count + 1, "error": msg, })), @@ -274,14 +264,16 @@ impl Orchestrator { }), ) .await; - if let Some(g) = &self.gating { - let _ = g - .bump_attempt( - &gating_key(row.arch_id.as_deref()), - &row.miner_hotkey, - class, - ) - .await; + if !rate { + if let Some(g) = &self.gating { + let _ = g + .bump_attempt( + &gating_key(row.arch_id.as_deref()), + &row.miner_hotkey, + class, + ) + .await; + } } true } diff --git a/crates/prism-emit/tests/epoch_semantics.rs b/crates/prism-emit/tests/epoch_semantics.rs index 3f6b4c372..0c9438049 100644 --- a/crates/prism-emit/tests/epoch_semantics.rs +++ b/crates/prism-emit/tests/epoch_semantics.rs @@ -403,7 +403,7 @@ async fn retry_reenters_outbox() { } )); - store.reset_for_retry("sub-a").await.unwrap(); + store.reset_for_retry("sub-a", true).await.unwrap(); let row = store.get("sub-a").await.unwrap().expect("row"); assert!(row.final_score.is_none()); // Re-score via apply (mirrors the orchestrator's finalize write). diff --git a/crates/prism-lium/Cargo.toml b/crates/prism-lium/Cargo.toml index 4faeb4b92..89e931018 100644 --- a/crates/prism-lium/Cargo.toml +++ b/crates/prism-lium/Cargo.toml @@ -9,6 +9,7 @@ rust-version.workspace = true publish = false [dependencies] +lium-rent-pool = { path = "../lium-rent-pool" } prism-recipe = { path = "../prism-recipe" } async-trait = "0.1" hex = "0.4" diff --git a/crates/prism-lium/src/client.rs b/crates/prism-lium/src/client.rs index 7ba275848..b2a0438ef 100644 --- a/crates/prism-lium/src/client.rs +++ b/crates/prism-lium/src/client.rs @@ -1,14 +1,21 @@ //! Real Lium HTTPS client + SSH-backed live eval. use std::path::{Path, PathBuf}; +use std::sync::OnceLock; use std::time::{Duration, Instant}; use async_trait::async_trait; +use lium_rent_pool::RentPool; use reqwest::header::{HeaderMap, HeaderValue}; use serde_json::Value; use tokio::time::sleep; use tracing::{debug, info, warn}; +fn rent_pool() -> &'static RentPool { + static POOL: OnceLock = OnceLock::new(); + POOL.get_or_init(RentPool::new) +} + use crate::error::{CostGuardrailError, LiumError}; use crate::ssh::{ parse_ssh_target, resolve_private_key, ssh_exec, ssh_exec_allow_fail, truncate_tail, SshTarget, @@ -157,15 +164,22 @@ impl LiumClient { .text() .await .map_err(|e| LiumError::Transport(sanitize_err(&e.to_string(), &self.api_key)))?; - if status.as_u16() == 429 && attempt < RATE_LIMIT_RETRIES { - attempt = attempt.saturating_add(1); - let wait_ms = retry_after.map_or( - RATE_LIMIT_BASE_MS << (attempt - 1).min(3), - |s| s.saturating_mul(1000).max(50), - ); - warn!(%path, attempt, wait_ms, "lium 429; backing off"); - sleep(Duration::from_millis(wait_ms)).await; - continue; + // Rent POSTs are gated by `RentPool` — do not multi-retry (each + // attempt burns Lium's 60/h budget). Other endpoints keep backoff. + let is_rent = path.contains("/rent"); + if status.as_u16() == 429 { + let secs = retry_after.or_else(|| lium_rent_pool::parse_retry_secs(&text)); + if is_rent { + rent_pool().note_429(secs.unwrap_or(5)); + } else if attempt < RATE_LIMIT_RETRIES { + attempt = attempt.saturating_add(1); + let wait_ms = secs.map_or(RATE_LIMIT_BASE_MS << (attempt - 1).min(3), |s| { + s.saturating_mul(1000).max(50) + }); + warn!(%path, attempt, wait_ms, "lium 429; backing off"); + sleep(Duration::from_millis(wait_ms)).await; + continue; + } } if !status.is_success() { return Err(LiumError::Api(format!( @@ -769,6 +783,7 @@ impl EvalJobBackend for LiumClient { let mut rented: Result = Err(LiumError::Api("unrented".into())); for gcount in split_choices { let body = make_body(gcount); + let permit = rent_pool().take().await; rented = self .request( reqwest::Method::POST, @@ -777,7 +792,12 @@ impl EvalJobBackend for LiumClient { ) .await; if let Err(e) = &rented { - if e.to_string().contains("splitting") { + let es = e.to_string(); + if lium_rent_pool::is_rate_limited(&es) { + permit.rate_limited(&es); + return Err(LiumError::Api(es)); + } + if es.contains("splitting") { continue; } } @@ -819,9 +839,6 @@ impl EvalJobBackend for LiumClient { } Err(e) => { last_err = e.to_string(); - if last_err.contains("429") { - sleep(Duration::from_millis(RATE_LIMIT_BASE_MS)).await; - } self.cleanup_after_rent(pod_id.as_deref()).await; } } diff --git a/crates/prism-store/src/dbprism.rs b/crates/prism-store/src/dbprism.rs index 8a6cd7ab5..acaecde9d 100644 --- a/crates/prism-store/src/dbprism.rs +++ b/crates/prism-store/src/dbprism.rs @@ -281,8 +281,12 @@ impl PrismStore for DbPrismStore { .ok_or(StoreError::NotFound)?) } - async fn reset_for_retry(&self, id: &str) -> Result { - let row = dbs::reset_prism_submission_for_retry(&self.pool, id) + async fn reset_for_retry( + &self, + id: &str, + bump_retry: bool, + ) -> Result { + let row = dbs::reset_prism_submission_for_retry(&self.pool, id, bump_retry) .await .map_err(|e| StoreError::Backend(e.to_string()))?; if let Err(e) = crate::telemetry::delete_telemetry(&self.pool, id).await { @@ -293,7 +297,9 @@ impl PrismStore for DbPrismStore { &dbs::NewPrismStageEvent { submission_id: id, stage: "queued", - detail: Some(serde_json::json!({"op": "retry"})), + detail: Some(serde_json::json!({ + "op": if bump_retry { "retry" } else { "rate_limit_requeue" }, + })), }, ) .await diff --git a/crates/prism-store/src/store.rs b/crates/prism-store/src/store.rs index 075f42cd5..ccf815fd5 100644 --- a/crates/prism-store/src/store.rs +++ b/crates/prism-store/src/store.rs @@ -279,10 +279,14 @@ pub trait PrismStore: Send + Sync + std::fmt::Debug { event: Option<&StageEvent>, ) -> Result; - /// Retry reset: clears exec/score fields and re-queues a failed row. - /// Implementations MUST actually null the pod/receipt/score columns - /// (SQL) or reset the in-memory row mirror-equivalently. - async fn reset_for_retry(&self, id: &str) -> Result; + /// Retry reset: clears exec/score fields and re-queues. + /// `bump_retry` increments `retry_count` (manual/auto infra). Pass + /// `false` for Lium 429 autonomous requeue (do not burn attempt budget). + async fn reset_for_retry( + &self, + id: &str, + bump_retry: bool, + ) -> Result; /// Newsfeed listing for the API (`status` / `miner` optional filters). async fn list( @@ -528,7 +532,11 @@ impl PrismStore for MemoryPrismStore { Ok(out) } - async fn reset_for_retry(&self, id: &str) -> Result { + async fn reset_for_retry( + &self, + id: &str, + bump_retry: bool, + ) -> Result { let mut rows = self .rows .lock() @@ -547,7 +555,9 @@ impl PrismStore for MemoryPrismStore { row.similarity = None; row.final_score = None; row.error_detail = None; - row.retry_count = row.retry_count.saturating_add(1); + if bump_retry { + row.retry_count = row.retry_count.saturating_add(1); + } row.updated_at_ms = now_ms(); let out = row.clone(); drop(rows); @@ -1023,7 +1033,7 @@ mod tests { ) .await .unwrap(); - s.reset_for_retry("a").await.unwrap(); + s.reset_for_retry("a", true).await.unwrap(); assert!(s.telemetry("a").await.unwrap().is_empty()); assert!(s.get("a").await.unwrap().unwrap().metrics_json.is_none()); } diff --git a/docs/PRISM.md b/docs/PRISM.md index eeb7657db..e2adc5aed 100644 --- a/docs/PRISM.md +++ b/docs/PRISM.md @@ -79,7 +79,11 @@ enforces **one accepted submission per `(prism, hotkey)`** unknown hotkey → `403 hotkey_not_in_metagraph`. Infra-class failures (`install` = Lium/pod, `ast_infra` = similarity, `llm_infra` = review/agentic) **auto-retry up to 3 times** before a terminal `blocked`; cheat / suspicious -verdicts are terminal `rejected` (no retry). After an infra `blocked`, the +verdicts are terminal `rejected` (no retry). Lium **HTTP 429** on rent is +special: `lium-rent-pool` serializes rents (≤**3 / 5s**, ≤**60 / hour**), +waits on `Retry-After` / body hints, and the orchestrator **requeues without +burning** `retry_count` / gating attempts. A background tick re-queues +failed 429 rows from the last **6 hours**. After an infra `blocked`, the miner may **resubmit for up to 30 minutes** (new `POST /v1/submissions` or `POST /v1/submissions/{id}/retry` for `ChallengeInternal`); after the window the slot stays blocked until the metagraph watcher reopens it (hotkey left /