fix(prism): autonomous Lium rent pool with 429 requeue

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" }

bins/prism-challenge/src/main.rs

@@ -362,6 +362,7 @@ fn build_topmodel() -> Option<Arc<prism_registry::TopModelPublisher>> {
     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<AppState>) {
     });
 }
 
-fn spawn_orchestrator(cli: &Cli, orchestrator: &Arc<Orchestrator<chain_live::LiveChainClient>>) {
+fn spawn_orchestrator(
+    cli: &Cli,
+    orchestrator: &Arc<Orchestrator<chain_live::LiveChainClient>>,
+    store: Arc<dyn PrismStore>,
+    gating: Option<Arc<dyn GatingStore>>,
+) {
     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<Orchestrator<chain_live::Liv
     tokio::spawn(async move { o.run_sweeper().await });
     let o = Arc::clone(orchestrator);
     tokio::spawn(async move { o.run_emitter().await });
+    spawn_rate_limit_recovery(store, gating);
+}
+
+/// Re-queue last-6h failed rows that died on Lium HTTP 429 (no `retry_count` burn).
+async fn recover_rate_limited(
+    store: &Arc<dyn PrismStore>,
+    gating: Option<&Arc<dyn GatingStore>>,
+) -> 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<dyn PrismStore>, gating: Option<Arc<dyn GatingStore>>) {
+    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;
+        }
+    });
 }

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<PrismSubmissionRow, DbError> {
     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)

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