Merge pull request #11 from buffrr/prover-limits-ui-fixes

Prover limits UI fixes

Author: buffrr

core/src/app.rs

@@ -1303,7 +1303,13 @@ impl Operator {
             };
 
         // has_pending is true until fully done (published)
-        let has_pending = !is_done;
+        // Publishing does not block the next commitment. broadcast/confirmed/
+        // finalized are on-chain and strictly sequential — two commitments
+        // cannot race — but publishing is downstream distribution of an
+        // already-final commitment. The background loop publishes on its own
+        // timer, and publish_certs selects across commitments, so a later
+        // commitment's certificates are simply picked up by the same sweep.
+        let has_pending = !is_done && current_step.as_deref() != Some("published");
 
         // Reset stale temp certs when the on-chain tip has changed,
         // so handles published against an old tip get republished

prover/src/main.rs

@@ -40,6 +40,15 @@ struct Cli {
     #[arg(long, default_value = "8888")]
     server_port: u16,
 
+    /// Skip startup calibration (for --server mode).
+    ///
+    /// Calibration proves a small batch to measure throughput, and the server
+    /// does not accept connections until it finishes — including /health. On a
+    /// short-lived GPU pod that delay is billed on every cold start, so skip it
+    /// when estimates aren't needed. Also settable via PROVER_NO_CALIBRATE.
+    #[arg(long)]
+    no_calibrate: bool,
+
     #[command(subcommand)]
     cmd: Option<Commands>,
 }
@@ -80,7 +89,9 @@ async fn main() -> Result<()> {
     let cli = Cli::parse();
 
     if cli.server {
-        subs_prover::server::run_server(cli.server_port).await?;
+        let no_calibrate = cli.no_calibrate
+            || std::env::var("PROVER_NO_CALIBRATE").is_ok_and(|v| !v.is_empty() && v != "0");
+        subs_prover::server::run_server(cli.server_port, no_calibrate).await?;
         return Ok(());
     }
 
@@ -160,7 +171,10 @@ fn compress(input: &CompressInput) -> Result<Vec<u8>> {
 }
 
 fn run_bench(existing: usize, insert: usize) -> Result<()> {
-    eprintln!("Building tree with {} existing handles, {} inserts...", existing, insert);
+    eprintln!(
+        "Building tree with {} existing handles, {} inserts...",
+        existing, insert
+    );
     let start = std::time::Instant::now();
     let request = subs_prover::build_bench_request(existing, insert)?;
     eprintln!("Request built in {:.2}s", start.elapsed().as_secs_f64());
@@ -181,15 +195,22 @@ fn run_bench(existing: usize, insert: usize) -> Result<()> {
         }
     };
 
-    eprintln!("Estimating proof for {} handles inserted into tree of {}...", insert, existing);
+    eprintln!(
+        "Estimating proof for {} handles inserted into tree of {}...",
+        insert, existing
+    );
     let estimate = prover.estimate(&request, calibration.as_ref())?;
 
     eprintln!("\n=== Estimate ===");
     eprintln!("Total user cycles:    {}", estimate.total_cycles);
-    eprintln!("Total proving cycles: {} (padded)", estimate.total_proving_cycles);
+    eprintln!(
+        "Total proving cycles: {} (padded)",
+        estimate.total_proving_cycles
+    );
     eprintln!("Segments:             {}", estimate.segments);
     for (i, seg) in estimate.segment_details.iter().enumerate() {
-        let time_str = seg.estimated_seconds
+        let time_str = seg
+            .estimated_seconds
             .map(|s| format!("{:.2}s", s))
             .unwrap_or_else(|| "n/a".into());
         eprintln!(

prover/src/server.rs

@@ -8,7 +8,7 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use axum::{
-    extract::{Path, Request, State},
+    extract::{DefaultBodyLimit, Path, Request, State},
     http::StatusCode,
     middleware::{self, Next},
     response::{IntoResponse, Response},
@@ -101,7 +101,20 @@ pub struct ErrorResponse {
 }
 
 /// Start the prover server
-pub async fn run_server(port: u16) -> anyhow::Result<()> {
+/// Maximum accepted request body.
+///
+/// Proving requests scale with the batch: the zk input alone is 64 bytes per
+/// handle, so a 50k-handle commitment is ~3 MB before the exclusion proof, and
+/// axum's 2 MB default rejects it with a 413 that reads like a prover fault.
+/// Bodies are buffered in memory, so this is a memory bound as much as a
+/// policy one — generous rather than tight, since the failure mode of setting
+/// it too low is a rejected commitment, and PROVER_AUTH_TOKEN gates the port.
+///
+/// Kept in step with MAX_BODY_BYTES in the runpod proxy: whichever hop has the
+/// lower ceiling is the one that 413s.
+const MAX_BODY_BYTES: usize = 512 * 1024 * 1024;
+
+pub async fn run_server(port: u16, no_calibrate: bool) -> anyhow::Result<()> {
     // Initialize tracing
     tracing_subscriber::fmt()
         .with_env_filter(
@@ -116,28 +129,34 @@ pub async fn run_server(port: u16) -> anyhow::Result<()> {
     // Create shared state
     let state = Arc::new(ServerState::new(tx));
 
-    // Calibrate proving throughput on startup
-    tracing::info!("Calibrating proving throughput...");
-    let calibrate_state = state.clone();
-    let calibrate_handle = tokio::task::spawn_blocking(move || {
-        let prover = Prover::new();
-        prover.calibrate()
-    });
-    match calibrate_handle.await {
-        Ok(Ok(info)) => {
-            tracing::info!(
-                "Calibration complete: {:.2}s per segment at po2={}, {:.0} cycles/sec",
-                info.seconds_per_segment,
-                info.calibration_po2,
-                info.cycles_per_sec,
-            );
-            *calibrate_state.calibration.write().await = Some(info);
-        }
-        Ok(Err(e)) => {
-            tracing::warn!("Calibration failed (estimates will be unavailable): {}", e);
-        }
-        Err(e) => {
-            tracing::warn!("Calibration task panicked: {}", e);
+    // Calibrate proving throughput on startup. This blocks the listener, so
+    // /health stays unanswered until it completes — deliberate, since an
+    // estimate is useless before it, but billable on a short-lived pod.
+    if no_calibrate {
+        tracing::info!("Calibration skipped (--no-calibrate); /estimate will be unavailable");
+    } else {
+        tracing::info!("Calibrating proving throughput...");
+        let calibrate_state = state.clone();
+        let calibrate_handle = tokio::task::spawn_blocking(move || {
+            let prover = Prover::new();
+            prover.calibrate()
+        });
+        match calibrate_handle.await {
+            Ok(Ok(info)) => {
+                tracing::info!(
+                    "Calibration complete: {:.2}s per segment at po2={}, {:.0} cycles/sec",
+                    info.seconds_per_segment,
+                    info.calibration_po2,
+                    info.cycles_per_sec,
+                );
+                *calibrate_state.calibration.write().await = Some(info);
+            }
+            Ok(Err(e)) => {
+                tracing::warn!("Calibration failed (estimates will be unavailable): {}", e);
+            }
+            Err(e) => {
+                tracing::warn!("Calibration task panicked: {}", e);
+            }
         }
     }
 
@@ -150,7 +169,9 @@ pub async fn run_server(port: u16) -> anyhow::Result<()> {
     // Optional bearer-token auth. If PROVER_AUTH_TOKEN is set, every route
     // (including /health) requires `Authorization: Bearer <token>` — that
     // way a successful /health probe also confirms auth is wired correctly.
-    let auth_token = std::env::var("PROVER_AUTH_TOKEN").ok().filter(|s| !s.is_empty());
+    let auth_token = std::env::var("PROVER_AUTH_TOKEN")
+        .ok()
+        .filter(|s| !s.is_empty());
     if auth_token.is_some() {
         tracing::info!("PROVER_AUTH_TOKEN set, requiring bearer auth on all routes");
     }
@@ -161,7 +182,8 @@ pub async fn run_server(port: u16) -> anyhow::Result<()> {
         .route("/estimate", post(submit_estimate))
         .route("/compress", post(submit_compress))
         .route("/jobs/:job_id", get(get_job_status))
-        .route("/jobs/:job_id/receipt", get(get_job_receipt));
+        .route("/jobs/:job_id/receipt", get(get_job_receipt))
+        .route("/calibration", get(get_calibration));
     if let Some(token) = auth_token {
         app = app.layer(middleware::from_fn(move |req: Request, next: Next| {
             let token = token.clone();
@@ -170,6 +192,7 @@ pub async fn run_server(port: u16) -> anyhow::Result<()> {
     }
 
     let app = app
+        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
         .layer(TraceLayer::new_for_http())
         .layer(
             CorsLayer::new()
@@ -192,6 +215,29 @@ pub async fn run_server(port: u16) -> anyhow::Result<()> {
 }
 
 /// Health check endpoint
+/// GET /calibration - Measured proving throughput of this machine.
+///
+/// The number that characterises a GPU for cost purposes: cost per proof is
+/// total_proving_cycles / cycles_per_sec. Previously only reachable by reading
+/// the startup log line or running `subs-prover bench` on the box.
+///
+/// 503 when calibration was skipped or failed.
+async fn get_calibration(State(state): State<Arc<ServerState>>) -> impl IntoResponse {
+    match state.calibration.read().await.clone() {
+        Some(info) => Json(serde_json::json!({
+            "seconds_per_segment": info.seconds_per_segment,
+            "calibration_po2": info.calibration_po2,
+            "cycles_per_sec": info.cycles_per_sec,
+        }))
+        .into_response(),
+        None => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "calibration unavailable (skipped or failed)",
+        )
+            .into_response(),
+    }
+}
+
 async fn health() -> &'static str {
     "ok"
 }
@@ -435,10 +481,7 @@ async fn get_job_receipt(
                 tracing::info!("Job {} receipt pulled, removing job", job_id);
                 (
                     StatusCode::OK,
-                    [(
-                        axum::http::header::CONTENT_TYPE,
-                        "application/octet-stream",
-                    )],
+                    [(axum::http::header::CONTENT_TYPE, "application/octet-stream")],
                     receipt,
                 )
                     .into_response()

subs/src/routes/commits.rs

@@ -338,6 +338,11 @@ pub struct PipelineResponse {
     pub prover_configured: bool,
     /// Whether a proving job is currently in flight on the prover
     pub proving_job_active: bool,
+    /// Which proof of the commitment is next (1-based), when proving.
+    pub proof_index: Option<usize>,
+    /// How many proofs this commitment needs in total: the first commitment
+    /// after genesis needs only a step, later ones need step + fold.
+    pub proof_total: Option<usize>,
 }
 
 pub async fn get_pipeline_status(
@@ -366,11 +371,21 @@ pub async fn get_pipeline_status(
     // Check if there's an active proving job by looking for a job key in config.
     // The job key uses the commitment's SQLite row id from the proving request,
     // matching the format used by push_to_prover and the background loop.
-    let proving_job_active = if status.commitment_idx.is_some() {
+    // Also report which proof of this commitment is next. count_pending_proofs
+    // charges a fold only from idx >= 2, so the first commitment after genesis
+    // is a single step proof and later ones are step + fold.
+    let mut proof_index = None;
+    let mut proof_total = None;
+
+    let proving_job_active = if let Some(idx) = status.commitment_idx {
         if let Ok(Some(req)) = state.operator.get_next_proving_request(&space_label).await {
             let cid = req.commitment_id();
             let is_fold = matches!(&req, subs_types::ProvingRequest::Fold { .. });
             let kind = if is_fold { "fold" } else { "step" };
+
+            proof_total = Some(if idx >= 2 { 2 } else { 1 });
+            proof_index = Some(if is_fold { 2 } else { 1 });
+
             let job_key = format!("job:{}:{}:{}", space, cid, kind);
             state.config.get(&job_key).unwrap_or(None).is_some()
         } else {
@@ -384,5 +399,7 @@ pub async fn get_pipeline_status(
         status,
         prover_configured,
         proving_job_active,
+        proof_index,
+        proof_total,
     }))
 }

subs/templates/space.html

@@ -225,7 +225,10 @@ <h3>Handles</h3>
 let currentPage = 1;
 let perPage = 20;
 let currentPipeline = null;
-let provingPollActive = false;
+// Handle for the pipeline auto-refresh timer. Kept as a handle rather than a
+// boolean so it can be cleared and re-armed on every pass: a flag that is only
+// ever set true stops the chain after a single tick.
+let pipelinePollTimer = null;
 let currentFilter = null;
 let searchTimeout = null;
 let selectedHandles = new Set();
@@ -297,6 +300,23 @@ <h3>Handles</h3>
 
         updatePublishBar(j.unpublished || 0);
 
+        // Re-arm before the has_pending branch below: publishing no longer
+        // sets has_pending, so that branch returns early and would otherwise
+        // never reach this. Poll while the pipeline advances on its own — the
+        // prover working, the chain confirming, the background loop
+        // publishing. 'broadcast' is excluded; it waits on the operator.
+        if (pipelinePollTimer) {
+            clearTimeout(pipelinePollTimer);
+            pipelinePollTimer = null;
+        }
+        if ((j.steps && j.steps.proving === 'in_progress' && j.prover_configured)
+            || j.current_step === 'confirmed'
+            || j.current_step === 'finalized'
+            || j.current_step === 'published'
+            || (j.unpublished || 0) > 0) {
+            pipelinePollTimer = setTimeout(refreshPipeline, 5000);
+        }
+
         if (!j.has_pending) {
             const sc = j.staged_count || 0;
             if (sc > 0) {
@@ -323,14 +343,6 @@ <h3>Handles</h3>
         renderActions(j);
         $('provingEndpoint').innerHTML = '';
 
-        if (j.steps.proving === 'in_progress' && j.prover_configured) {
-            if (!provingPollActive) {
-                provingPollActive = true;
-                setTimeout(refreshPipeline, 5000);
-            }
-        } else {
-            provingPollActive = false;
-        }
         loadHandles(currentPage);
     } catch (e) {
         $('pipelineStepper').innerHTML = `<span style="color:var(--red);font-size:13px">${e.message}</span>`;
@@ -401,14 +413,19 @@ <h3>Handles</h3>
     let h = '';
 
     if (r.current_step === 'proving') {
+        // "1 of 2" / "2 of 2" so an interrupted run and a normal handoff
+        // between the step and fold proofs don't look identical.
+        const proofLabel = (r.proof_index && r.proof_total && r.proof_total > 1)
+            ? ` ${r.proof_index} of ${r.proof_total}`
+            : '';
         if (r.prover_configured) {
             if (r.proving_job_active) {
                 const total = r.pending_proofs || 0;
                 h += `<div class="flex items-center gap-2" style="color:var(--accent)">
                     <svg width="14" height="14" viewBox="0 0 24 24" fill="none" class="animate-spin">
                         <circle opacity="0.25" cx="12" cy="12" r="10" stroke="currentColor" stroke-width="4"/>
                         <path opacity="0.75" fill="currentColor" d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4z"/></svg>
-                    <span style="font-size:13px">Generating proof...
+                    <span style="font-size:13px">Generating proof${proofLabel}...
                         <span style="color:var(--text-muted)">(${total} pending)</span></span></div>`;
             }
             if (r.estimate) {
@@ -424,7 +441,7 @@ <h3>Handles</h3>
             }
             if (!r.proving_job_active) {
                 h += `<div class="flex gap-3" style="margin-top:10px">
-                    <button onclick="startProving()" class="btn-primary">Prove</button>
+                    <button onclick="startProving()" class="btn-primary">Prove${proofLabel ? ' ' + proofLabel.trim() : ''}</button>
                     <button onclick="rollbackLocal()" class="btn-secondary">Rollback</button></div>`;
             } else {
                 h += '<button onclick="rollbackLocal()" class="btn-sm" style="margin-top:10px">Rollback Local</button>';