From b47828aa0d159dfac9ad947609c7dacc5cb60089 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Wed, 19 Aug 2026 06:38:30 +0000 Subject: [PATCH] fix(prism): slim list queries and prefer pod resume on claim Listing submissions selected full tree_blob and scripts, which OOM-killed prism-challenge on the 16 GiB master and requeued running jobs as control_plane_restart. Claim now prefers queued rows that already have a pod_id so resume does not wait behind new submits or rent a second Lium pod. --- crates/db/src/prism_store.rs | 58 +++++++++++++++++++--- crates/prism-challenge/src/orchestrator.rs | 2 + crates/prism-orphan/src/lib.rs | 13 ++++- crates/prism-store/src/store.rs | 23 ++++++++- docs/PRISM.md | 8 ++- 5 files changed, 92 insertions(+), 12 deletions(-) diff --git a/crates/db/src/prism_store.rs b/crates/db/src/prism_store.rs index aa7489ca3..9008ee000 100644 --- a/crates/db/src/prism_store.rs +++ b/crates/db/src/prism_store.rs @@ -92,11 +92,53 @@ pub struct NewPrismStageEvent<'a> { pub detail: Option, } -/// Column list shared by all row reads. +/// Column list shared by single-row reads (`get` / `claim` / `update`). const COLS: &str = "id, miner_hotkey, miner_coldkey, epoch, netuid, status, label, \ architecture_py, training_py, tree_blob, pod_id, pod_provider, receipt_json, metrics_json, \ bpb, review_json, similarity_json, kind, score, absence_reason, retry_count, error_detail"; +/// Listing projection: same `PrismSubmissionRow` shape without source trees, +/// telemetry series, or review blobs. Site/reconcile list of hundreds of rows +/// loading `tree_blob` (up to 17 MiB) + scripts was out-of-memory killing the +/// 16 GiB master host and bouncing `prism-challenge` (`control_plane_restart` +/// storms). +/// +/// Presence of `receipt_json` / `metrics_json` is preserved as slim stubs so +/// `resume_measurement` / `mid_pod_resume` still classify post-measure vs +/// mid-pod. `n_params` is kept for the submissions list view. +const LIST_COLS: &str = "id, miner_hotkey, miner_coldkey, epoch, netuid, status, label, \ + ''::text AS architecture_py, ''::text AS training_py, NULL::bytea AS tree_blob, \ + pod_id, pod_provider, \ + CASE WHEN receipt_json IS NULL THEN NULL ELSE jsonb_build_object(\ + 'provider', COALESCE(receipt_json->>'provider', ''), \ + 'pod_id', COALESCE(receipt_json->>'pod_id', ''), \ + 'image_digest', '', 'submission_hash', '', 'metrics_hash', '', \ + 'termination_verified', COALESCE((receipt_json->>'termination_verified')::boolean, false)\ + ) END AS receipt_json, \ + CASE WHEN metrics_json IS NULL THEN NULL ELSE jsonb_strip_nulls(jsonb_build_object(\ + 'bpb', COALESCE(metrics_json->'bpb', '0'::jsonb), \ + 'tokens_seen', COALESCE(metrics_json->'tokens_seen', '0'::jsonb), \ + 'wall_clock_seconds', COALESCE(metrics_json->'wall_clock_seconds', '0'::jsonb), \ + 'notes', '', \ + 'n_params', metrics_json->'n_params'\ + )) END AS metrics_json, \ + bpb, NULL::jsonb AS review_json, NULL::jsonb AS similarity_json, \ + kind, score, absence_reason, retry_count, error_detail"; + +/// Champion corpus for copy/similarity: keep `architecture_py`, drop trees +/// and training scripts (gates compare architecture only). +const CHAMPION_COLS: &str = "id, miner_hotkey, miner_coldkey, epoch, netuid, status, label, \ + architecture_py, ''::text AS training_py, NULL::bytea AS tree_blob, \ + pod_id, pod_provider, receipt_json, \ + CASE WHEN metrics_json IS NULL THEN NULL ELSE jsonb_strip_nulls(jsonb_build_object(\ + 'bpb', COALESCE(metrics_json->'bpb', '0'::jsonb), \ + 'tokens_seen', COALESCE(metrics_json->'tokens_seen', '0'::jsonb), \ + 'wall_clock_seconds', COALESCE(metrics_json->'wall_clock_seconds', '0'::jsonb), \ + 'notes', '', \ + 'n_params', metrics_json->'n_params'\ + )) END AS metrics_json, \ + bpb, review_json, similarity_json, kind, score, absence_reason, retry_count, error_detail"; + /// Insert the queued row. /// /// # Errors @@ -141,8 +183,11 @@ pub async fn prism_submission( Ok(row) } -/// Atomically claim the oldest queued row (`SKIP LOCKED`), moving it to -/// `provisioning`. Returns None when empty. +/// Atomically claim the next queued row (`SKIP LOCKED`), moving it to +/// `provisioning`. Prefer rows that already have a `pod_id` (control-plane +/// resume) so they are not pushed behind a long FIFO of new submits — those +/// would take the concurrent slots and trigger a second `lium rent`. +/// Among the same class, oldest `created_at` first. /// /// # Errors /// SQL error. @@ -151,7 +196,8 @@ pub async fn claim_prism_submission(pool: &PgPool) -> Result(&q) @@ -291,7 +337,7 @@ pub async fn list_prism_submissions( .filter(|s| !s.is_empty()) .map(str::to_ascii_lowercase); let q = format!( - "SELECT {COLS} FROM prism_submission \ + "SELECT {LIST_COLS} FROM prism_submission \ WHERE ($1::TEXT IS NULL OR status = $1) \ AND ($2::TEXT IS NULL OR lower(miner_hotkey) = $2) \ ORDER BY created_at DESC LIMIT $3" @@ -314,7 +360,7 @@ pub async fn list_prism_champions( limit: i64, ) -> Result, DbError> { let q = format!( - "SELECT {COLS} FROM prism_submission \ + "SELECT {CHAMPION_COLS} FROM prism_submission \ WHERE kind = 'score' AND score > 0 \ ORDER BY created_at DESC LIMIT $1" ); diff --git a/crates/prism-challenge/src/orchestrator.rs b/crates/prism-challenge/src/orchestrator.rs index ba45359c9..9b8f5e3fd 100644 --- a/crates/prism-challenge/src/orchestrator.rs +++ b/crates/prism-challenge/src/orchestrator.rs @@ -696,6 +696,8 @@ impl Orchestrator { } } let backend = self.backend_for(id)?; + // `pod_id` present ⇒ reattach (never rent a second offer). Fresh + // submits have no pod and take the provision path below. let resume = mid_pod_resume(row); let (pod_id, provider) = if let Some(pid) = row.pod_id.clone() { ( diff --git a/crates/prism-orphan/src/lib.rs b/crates/prism-orphan/src/lib.rs index 3edfae7ee..66a8b1781 100644 --- a/crates/prism-orphan/src/lib.rs +++ b/crates/prism-orphan/src/lib.rs @@ -3,8 +3,10 @@ //! Workers register in [`ActiveJobs`] for the whole `run_row` hold. On boot //! (and periodically) [`reconcile_once`] **resume-first**: mid-pod rows whose //! Lium instance is still alive are requeued with `pod_id` kept (no terminate). -//! Only unreattachable rows fail-closed (`control_plane_restart` / -//! `harness_detached`). Post-measure review stages requeue as before. +//! [`PrismStore::claim_next`] then prefers those rows so they reattach before +//! new submits take the concurrent slots. Only unreattachable rows fail-closed +//! (`control_plane_restart` / `harness_detached`). Post-measure review stages +//! requeue as before. #![forbid(unsafe_code)] #![allow(clippy::missing_errors_doc)] @@ -271,6 +273,13 @@ mod tests { let got = store.get(&r.id).await.unwrap().unwrap(); assert_eq!(got.status, Stage::Queued); assert_eq!(got.pod_id.as_deref(), Some("pod-live")); + store + .insert_queued(&row("aaaaaaaaaaaaaaaa", Stage::Queued, None)) + .await + .unwrap(); + let claimed = store.claim_next().await.unwrap().unwrap(); + assert_eq!(claimed.id, r.id, "resume row must skip the new-submit FIFO"); + assert_eq!(claimed.pod_id.as_deref(), Some("pod-live")); } #[tokio::test] diff --git a/crates/prism-store/src/store.rs b/crates/prism-store/src/store.rs index 055803798..b26dff58b 100644 --- a/crates/prism-store/src/store.rs +++ b/crates/prism-store/src/store.rs @@ -23,7 +23,8 @@ pub trait PrismStore: Send + Sync + std::fmt::Debug { async fn get(&self, id: &str) -> Result, StoreError>; /// Atomically claim the next queued row for work (`SKIP LOCKED`, moves - /// to provisioning). + /// to provisioning). Prefer queued rows that already have a `pod_id` + /// (reattach) over brand-new submits. async fn claim_next(&self) -> Result, StoreError>; /// Apply a partial update + journal the event (one logical step). @@ -216,7 +217,10 @@ impl PrismStore for MemoryPrismStore { async fn claim_next(&self) -> Result, StoreError> { let mut rows = lock(&self.rows)?; - let pos = rows.iter().position(|r| r.status == Stage::Queued); + let pos = rows + .iter() + .position(|r| r.status == Stage::Queued && r.pod_id.is_some()) + .or_else(|| rows.iter().position(|r| r.status == Stage::Queued)); let Some(i) = pos else { return Ok(None) }; let mut row = rows.remove(i).ok_or(StoreError::Backend("pop".into()))?; row.status = Stage::Provisioning; @@ -626,6 +630,21 @@ mod tests { assert_eq!(row.status, Stage::Provisioning); } + #[tokio::test] + async fn claim_prefers_queued_row_with_pod() { + let s = MemoryPrismStore::new(); + s.insert_queued(&row("new", "11")).await.unwrap(); + let mut resume = row("resume", "22"); + resume.pod_id = Some("pod-live".into()); + s.insert_queued(&resume).await.unwrap(); + let first = s.claim_next().await.unwrap().unwrap(); + assert_eq!(first.id, "resume"); + assert_eq!(first.pod_id.as_deref(), Some("pod-live")); + let second = s.claim_next().await.unwrap().unwrap(); + assert_eq!(second.id, "new"); + assert!(second.pod_id.is_none()); + } + #[tokio::test] async fn insert_queued_rejects_duplicate_id() { let s = MemoryPrismStore::new(); diff --git a/docs/PRISM.md b/docs/PRISM.md index 48ab7cf5f..1cba5789d 100644 --- a/docs/PRISM.md +++ b/docs/PRISM.md @@ -81,10 +81,14 @@ not SIGHUP GPU work. On boot (and every ~30s) orphan reconcile is still alive and whose BYOK key can be restored from the sealed vault (`PRISM_PAYER_VAULT_DIR`, default TTL ≥**36h** / train+eval+skew; heartbeats re-seal; measure start refreshes the seal and **measure Err keeps the vault -entry** so auto-/miner-retry can re-rent) are requeued with `pod_id` kept — the +entry** so auto-/miner-retry can re-rent) are requeued with `pod_id` kept. `claim_next` prefers those resume rows over +new submits so they do not wait behind the FIFO (and so workers call +`resume_eval` instead of a second `lium rent`). The orchestrator reattaches (log/event poll → wait terminal → harvest → score) without terminating the -pod. Only unreattachable rows fail-closed (`control_plane_restart` / +pod. `GET /v1/submissions` lists omit source trees / telemetry series so the +control plane cannot OOM the master host by hydrating hundreds of `tree_blob` +columns. Only unreattachable rows fail-closed (`control_plane_restart` / `harness_detached`) with best-effort terminate. Post-measure review stages still requeue. Residual gap: expired seal + no operator fallback ⇒ cannot call Lium API ⇒ fail-orphan (miner must stop the pod and resubmit). The stuck