diff --git a/Cargo.lock b/Cargo.lock index 81ba246..1041533 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2299,7 +2299,7 @@ dependencies = [ [[package]] name = "fabric-resolver" version = "0.2.7" -source = "git+https://github.com/spacesprotocol/certrelay.git#15234646f27ff9fcee96d6d1aa79e1fce60d4f88" +source = "git+https://github.com/spacesprotocol/certrelay.git#e680aff03be1b87778f22b8c4f284da52b443333" dependencies = [ "borsh", "dashmap", @@ -4918,7 +4918,7 @@ dependencies = [ [[package]] name = "relay" version = "0.2.7" -source = "git+https://github.com/spacesprotocol/certrelay.git#15234646f27ff9fcee96d6d1aa79e1fce60d4f88" +source = "git+https://github.com/spacesprotocol/certrelay.git#e680aff03be1b87778f22b8c4f284da52b443333" dependencies = [ "anyhow", "axum 0.8.3", diff --git a/Cargo.toml b/Cargo.toml index 5fdbfb7..05cb984 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,9 @@ subs-core = { path = "core" } libveritas = { version = "=0.3.3", features = ["elf"] } libveritas_zk = "=0.1.1" libveritas_testutil = "=0.3.3" -spacedb = "0.1.3" +# hash-idx builds a sidecar index that speeds up .prove(); it is not a default +# feature and pulls in rusqlite, which we already depend on. +spacedb = { version = "0.1.3", features = ["hash-idx"] } relay = { git = "https://github.com/spacesprotocol/certrelay.git" } fabric = { package = "fabric-resolver", version = "=0.2.7" } diff --git a/REGISTRY.md b/REGISTRY.md index fd25471..573c7ba 100644 --- a/REGISTRY.md +++ b/REGISTRY.md @@ -11,65 +11,50 @@ work. (yours) (private) ``` -**subs always initiates.** The registry never calls subs, never needs to reach -it, and never needs a public address for it. Your registry is an HTTP server -that subs polls. +**subs always initiates.** The registry never calls subs and never needs a +public address for it. Your registry is an HTTP server that subs polls. ## Scope -This document covers **only** the four endpoints subs calls. Everything else -about your registry is yours to decide and subs neither sees nor cares about: +This document covers **only** the four endpoints subs calls. Everything else is +yours to decide and subs neither sees nor cares about: how requests get in, how +you authenticate or bill the people making them, how you store or display them. -- how handle requests get in — a public form, a paid checkout, an admin panel -- how you authenticate, bill, or rate-limit the people making requests -- how you store registrations, notify users, or expose status to them - -Implement the four endpoints below and subs will work with it. [`examples/registry-server`](examples/registry-server) is a working -implementation you can run and read; the intake it happens to ship is -illustrative, not part of this contract. +implementation you can run and read. --- ## The cycle -Each pass, subs: +Each pass, per space, subs: 1. `GET /pending` — collect handles awaiting registration -2. stages them locally (validating and de-duplicating) -3. `POST /ack` — report which ones it took +2. stages them locally, validating and de-duplicating +3. `POST /ack` — report what it took 4. publishes certificates to the relay network -Steps 1–3 are what your registry participates in. Step 4 is internal to subs. - -Later, when a batch is committed on-chain, subs can call -[`POST /committed`](#post-committed) — that one is **not** automatic. +Steps 1–3 involve your registry; step 4 is internal to subs. `POST /committed` +fires separately and is **not** automatic. --- ## Authentication -All four endpoints are authenticated with a single shared bearer token: +All four endpoints take a single shared bearer token: ``` Authorization: Bearer ``` Configure it in subs under **Settings → Registry Server → Auth Token**. Reject -anything without a valid token with **`401`**. subs surfaces `401` and `403` -distinctly from other failures, so an operator sees "registry rejected the auth -token" rather than a generic upstream error. - -This token grants access to your work queue — reading pending handles and -marking them staged or committed. If other systems of yours write to the -registry, give them their own credentials; there's no reason for them to share -the one subs holds. +anything without a valid token with **`401`**; subs surfaces `401`/`403` +distinctly, so an operator sees "registry rejected the auth token" rather than a +generic upstream error. -**`/health` is authenticated too**, deliberately. subs' **Test** button probes -it, so a successful test proves both reachability *and* that the token is -accepted — rather than showing green for a registry that will reject every -request after it. If you need an open liveness probe for a load balancer, -expose it on a path subs doesn't use. +**`/health` is authenticated too.** subs' **Test** button probes it, so a +successful test proves both reachability and that the token works. If you need +an open liveness probe, expose it on a path subs doesn't use. --- @@ -81,7 +66,24 @@ Return `200` with any body. ### `GET /pending` -Return handles waiting to be staged. +Return handles waiting to be staged. subs asks about **one space per request**: + +``` +GET /pending?space=@example +GET /pending?space=%233438-1-0 +``` + +**Return only handles in that space.** A handle for a space this operator cannot +act on can never be staged, so it is never acked — serve it and it comes back +every cycle, forever. + +The value is a canonical space label, percent-encoded; numeric spaces look like +`#3438-1-0`, so `#` arrives as `%23`. + +Scope includes spaces **delegated to the operator's wallet but not yet +started** — staging the first handle adopts such a space automatically. Scope is +recomputed every cycle, so a new delegation takes effect without restarting +subs. ```json { @@ -99,31 +101,59 @@ Return handles waiting to be staged. Return `{"handles": []}` when there's nothing pending — not `404`. -**Any non-2xx aborts the entire sync**, including the ack step, so nothing is -staged that pass. subs retries on the next cycle. The request times out after -10 seconds. +**Any non-2xx aborts that space's sync**, including its ack, so nothing is +staged for it that pass. subs retries next cycle. The request times out after 10 +seconds. Returning already-staged handles is harmless — subs de-duplicates — but filtering them keeps payloads small. +**Pagination** is optional: return the full set for the space, or cap it and let +the next cycle collect the rest. subs stages an entire response in one pass. +Capping per space starves nothing, since each request covers one space. + ### `POST /ack` -Called after subs stages the handles it pulled. +Called once subs has decided each handle's fate. ```json -{ "handles": ["alice@example", "bob@example"] } +{ + "handles": [ + { "handle": "alice@example", "outcome": "staged" }, + { "handle": "bob@example", "outcome": "already_committed_different_spk" } + ] +} ``` Move these out of your pending set. Return `2xx`; the body is ignored. -**This must be idempotent.** subs acks every handle it pulled, including ones -already staged locally, and re-acks after a failure. Re-acking an -already-acked handle must succeed, not error. +**Every outcome is terminal** — a handle appears here only once settled, and +will not be offered again. + +| Outcome | Meaning | What to tell the user | +|---|---|---| +| `staged` | Accepted; awaiting commitment | In progress | +| `already_staged_same_spk` | Already pending under the same owner | In progress — a duplicate request | +| `already_committed_same_spk` | Already registered to this owner | Already theirs | +| `already_staged_different_spk` | Another owner has it pending | **Cannot be fulfilled** | +| `already_committed_different_spk` | Another owner already holds it | **Cannot be fulfilled** | +| `invalid` | Not a parseable `name@space` handle | **Cannot be fulfilled** | + +The last three can never succeed, however many times they are retried. If you +take payment, they are your refund signal. + +**This must be idempotent.** If the ack fails, subs logs it and continues — +staging already happened on its side — so the next cycle re-pulls, re-stages (a +no-op) and re-acks. The flow self-heals, but only if re-acking succeeds rather +than errors. -If the ack fails, subs logs it and continues — staging already succeeded on its -side. Your registry still has those handles pending, so the next cycle -re-pulls, re-stages (a no-op), and re-acks. **The flow self-heals, but only if -`/ack` is idempotent.** +#### Retryable failures + +Some handles are neither staged nor settled: subs could not reach a decision, +typically because the space could not be loaded or the wallet could not operate +it. These are **deliberately absent** from the ack body. They stay in +`/pending`, and a later cycle picks them up once the operator's configuration is +fixed. Acking them would mark them done and lose them. ### `POST /committed` @@ -134,47 +164,30 @@ re-pulls, re-stages (a no-op), and re-acks. **The flow self-heals, but only if Mark these committed and record the root. Return `2xx`. **This is not automatic.** It fires only when someone calls -`POST /registry/notify` on subs with a `space` and `root`; nothing triggers it -on a timer. If you need committed state and aren't calling `/registry/notify`, -track it by watching the chain or by querying the relay network for the -handle's certificate. +`POST /registry/notify` on subs with a `space` and `root`. If you need committed +state and aren't calling that, track it by watching the chain or by querying the +relay network for the handle's certificate. --- ## Delivery semantics **At-least-once, never exactly-once.** A handle may be delivered more than once -— an ack that fails after subs staged the handle is the ordinary case. Design -so a repeat delivery is a no-op. +— an ack that fails after subs staged the handle is the ordinary case. Design so +a repeat delivery is a no-op. -**subs is the source of truth for what is registered**, not your registry. -`/ack` means subs accepted a handle into staging; it does not mean the handle -is committed on-chain. Only the commit notification means that. +**subs is the source of truth for what is registered**, not your registry. An +ack of `staged` means subs accepted the handle into staging; it does not mean +the handle is committed on-chain. Only the commit notification means that. --- ## Handle validation -`handle` must parse as a spaces name in `name@space` form. Handles that don't -parse are **skipped and never acked** — so they stay pending on your side and -are re-pulled on every cycle, forever. - -Validate at intake. A malformed handle accepted into your pending queue becomes -a permanent poison entry. - -subs may also decline to stage a well-formed handle: - -| Reason | Meaning | -|---|---| -| `already staged` | Same handle and script pubkey already pending locally | -| `already committed` | Already committed on-chain | -| `already staged with different spk` | Conflicts with a pending entry under a different owner | -| `already committed with different spk` | The handle is taken | - -These are logged by subs and **still acked** — they're settled outcomes, not -retryable failures. The last two mean the request cannot be fulfilled; your -registry has no way to learn this today, so surfacing it to users requires -polling on-chain state yourself. +`handle` must parse as a spaces name in `name@space` form. One that doesn't is +acked `invalid` and settled immediately, so it won't linger in your queue — but +validate at intake anyway. A malformed handle that reaches `/pending` has +already cost a round trip and, if you charged for it, a refund. --- @@ -191,9 +204,10 @@ In subs' **Settings → Registry Server**: With automatic sync **off**, the cycle runs only on **Sync Now** (or `POST /registry/sync`). -With it **on**, subs runs continuously: every 30 seconds when idle, every 5 -seconds while certificates remain to publish. It defaults to off because -publishing broadcasts to the relay network. +With it **on**, subs sleeps 2 seconds *after* each cycle finishes rather than +running on a fixed schedule, so cycles never overlap and a slow one simply +delays the next. It defaults to off because publishing broadcasts to the relay +network. --- @@ -202,7 +216,11 @@ publishing broadcasts to the relay network. - [ ] `/health`, `/pending`, `/ack`, `/committed` all require the bearer token - [ ] Missing or wrong token returns `401` - [ ] `GET /pending` returns the documented shape, `{"handles": []}` when empty -- [ ] `POST /ack` is idempotent and returns `2xx` for already-acked handles +- [ ] `GET /pending` filters on `?space=` +- [ ] `GET /pending` includes spaces delegated to the operator but not yet started +- [ ] `POST /ack` reads a per-handle `outcome` and is idempotent +- [ ] The `*_different_spk` and `invalid` outcomes are surfaced to the user, and refunded if paid +- [ ] Handles absent from an ack stay pending — they are retryable, not settled - [ ] Handles are validated as `name@space` **at intake** - [ ] Repeat delivery of the same handle is a no-op -- [ ] `POST /committed` implemented, if you need committed state +- [ ] `POST /committed` implemented, if you need committed state \ No newline at end of file diff --git a/core/src/app.rs b/core/src/app.rs index c82f114..a804ba8 100644 --- a/core/src/app.rs +++ b/core/src/app.rs @@ -185,9 +185,24 @@ impl LiveSpaceInfo { } let sub = name.subspace().unwrap(); + + // Split the two tree operations. Both resolve a snapshot by root and + // then walk the tree, so when a cert is slow this says which half — + // the membership check or the inclusion proof — is responsible. + let lookup_started = std::time::Instant::now(); let is_final = self.local.lookup_handle_in_tree(&sub, tip).await?.is_some(); + let lookup_ms = lookup_started.elapsed().as_millis(); + if is_final { - return self.local.issue_cert(&name, tip.unwrap()).await; + let proof_started = std::time::Instant::now(); + let cert = self.local.issue_cert(&name, tip.unwrap()).await; + log::debug!( + " {} final: lookup {}ms, proof {}ms", + name, + lookup_ms, + proof_started.elapsed().as_millis() + ); + return cert; } // temp cert @@ -255,8 +270,52 @@ pub struct Operator { fabric: Option, fabric_seeds: Vec, spaces: Arc>>>, + /// Last measured (certificates, message bytes) per space. + /// + /// Sizing is driven by the serialized message rather than a model of its + /// parts. A message is one root certificate carrying a ~250 KB recursive + /// receipt plus ~1 KB handle certificates, wrapped in framing and a chain + /// proof — so any per-certificate average describes none of it, and a + /// model built from parts drifts from what the relay actually measures. + /// Scaling from the whole is correct whichever component grows. + /// + /// Not persisted: a restart falls back to a pessimistic default and the + /// first batch replaces it with a real measurement. + last_message: Arc>>, +} + +/// Outcome of submitting a batch. +/// +/// Carries the measurement even when nothing was sent, which is the point: a +/// batch refused for being oversized must still teach the next one its size, +/// or the same batch is rebuilt and refused forever. +#[derive(Debug, Clone, Copy)] +pub struct SubmitOutcome { + pub cert_count: usize, + pub message_bytes: usize, + /// False when the message was built and measured but deliberately not + /// broadcast. Distinct from an `Err`, which means something failed — + /// a network blip must not be mistaken for a sizing problem, or repeated + /// blips would ratchet the batch down with nothing to grow it back. + pub sent: bool, } +/// Largest message a relay accepts, matching certrelay's default +/// `max_message_size`. A relay may be configured lower, which we cannot see. +const RELAY_MAX_MESSAGE_BYTES: usize = 512 * 1024; + +/// Fraction of that budget to aim for, leaving room for batch-to-batch +/// variation between the measurement and the next send. +const RELAY_BUDGET_PERCENT: usize = 85; + +/// Batch size assumed before anything has been measured. +/// +/// Pessimistic on purpose: measured messages are ~50% of the limit at 50 +/// certificates, so this is safe by a wide margin. Too small merely costs one +/// undersized batch after a restart; too large costs a refused one. Being +/// wrong in the cheap direction is what makes persistence unnecessary. +const ASSUMED_BATCH: usize = 50; + impl Operator { /// Create a new Operator with RPC client. /// @@ -270,6 +329,7 @@ impl Operator { wallet: wallet.into(), rpc: Some(rpc), spaces: Arc::new(Mutex::new(HashMap::new())), + last_message: Arc::new(Mutex::new(HashMap::new())), fabric: None, fabric_seeds: Vec::new(), } @@ -286,6 +346,7 @@ impl Operator { fabric: None, fabric_seeds: Vec::new(), spaces: Arc::new(Mutex::new(HashMap::new())), + last_message: Arc::new(Mutex::new(HashMap::new())), } } @@ -978,6 +1039,16 @@ impl Operator { local_space.save_estimate(commitment_id, estimate_json).await } + /// Drop a commitment's estimate once the proof it described has finished. + /// + /// The next proof of the same commitment is a different shape — a fold is + /// not a step — so leaving the old figures visible presents them as a + /// forecast for work they say nothing about. + pub async fn clear_estimate(&self, space: &SLabel, commitment_id: i64) -> anyhow::Result<()> { + let local_space = self.get_local_space(space)?; + local_space.clear_estimate(commitment_id).await + } + /// Get input for SNARK compression. pub async fn get_compress_input( &self, @@ -1147,7 +1218,7 @@ impl Operator { // No commitments yet - show message based on staged count let Some(commitment) = commitment else { - let unpublished = storage.select_handles(crate::storage::HandleSelector::Unpublished(None)).await?.len(); + let unpublished = storage.count_unpublished(None).await?; let message = if staged_count > 0 { Some(format!( "{} handle(s) staged. Ready to commit.", @@ -1316,7 +1387,7 @@ impl Operator { if let Some(_) = confirmed_idx { storage.reset_stale_temp_certs(Some(&commitment.root)).await?; } - let unpublished = storage.select_handles(crate::storage::HandleSelector::Unpublished(confirmed_idx)).await?.len(); + let unpublished = storage.count_unpublished(confirmed_idx).await?; let commitments = storage.list_commitments().await?; let pending_proofs = crate::core::count_pending_proofs(&commitments); @@ -1382,23 +1453,98 @@ impl Operator { Ok(new_txid) } - pub async fn submit_certs(&self, certs: Vec) -> anyhow::Result<()> { - log::info!("submit_certs: building message for {} certs", certs.len()); + pub async fn submit_certs(&self, certs: Vec) -> anyhow::Result { + let count = certs.len(); + log::info!("submit_certs: building message for {} certs", count); + let msg = self.build_message(certs).await?; - log::info!("submit_certs: message built, broadcasting via fabric"); let fabric = self.require_fabric()?; let relays = fabric .bootstrap() .await .map_err(|e| anyhow!("fabric bootstrap error: {}", e))?; - log::info!("relays available: {:?}", relays); + let bytes = msg.to_bytes(); + + // Relays reject a message over max_message_size outright, so batch + // size is bounded by bytes rather than by count. Logging the measured + // size is what makes that budget legible: certificate size grows with + // tree depth, so a batch that fits today can stop fitting as the space + // grows. + log::info!( + "submit_certs: {} certs = {} bytes, {:.1}% of the {} KB relay limit", + count, + bytes.len(), + bytes.len() as f64 / RELAY_MAX_MESSAGE_BYTES as f64 * 100.0, + RELAY_MAX_MESSAGE_BYTES / 1024, + ); + + // Refuse rather than let the relay reject it. Returned as Ok, not Err, + // so the caller can tell "measured and declined" from "something + // failed" — the measurement is what shrinks the next batch, and a + // network error must not be read as a size problem. + if bytes.len() > RELAY_MAX_MESSAGE_BYTES { + log::warn!( + "submit_certs: {} bytes exceeds the {} KB relay limit; not sending", + bytes.len(), + RELAY_MAX_MESSAGE_BYTES / 1024, + ); + return Ok(SubmitOutcome { + cert_count: count, + message_bytes: bytes.len(), + sent: false, + }); + } + + log::debug!("relays available: {:?}", relays); fabric - .broadcast(&msg.to_bytes()) + .broadcast(&bytes) .await .map_err(|e| anyhow!("Could not broadcast message: {}", e))?; log::info!("submit_certs: broadcast OK"); - Ok(()) + Ok(SubmitOutcome { + cert_count: count, + message_bytes: bytes.len(), + sent: true, + }) + } + + /// How many certificates to put in the next batch for this space. + /// + /// Scaled from the last message's measured size, so it converges in one + /// cycle whichever component grew — root certificate, handle certificates, + /// framing or chain proof. Falls back to a conservative assumption when + /// nothing has been measured yet. + /// + /// `ceiling` is a sanity cap, not a target: bytes decide the batch, and a + /// ceiling set near the expected size would silently prevent it growing. + fn next_batch_size(&self, space: &SLabel, ceiling: usize) -> usize { + let target = RELAY_MAX_MESSAGE_BYTES * RELAY_BUDGET_PERCENT / 100; + + let scaled = match self.last_message.lock().unwrap().get(space) { + Some(&(certs, bytes)) if certs > 0 && bytes > 0 => { + // Certificates per byte, applied to the budget. Integer maths + // deliberately rounds down. + (certs * target / bytes).max(1) + } + _ => ASSUMED_BATCH, + }; + + scaled.min(ceiling).max(1) + } + + /// Record what a batch of this size actually serialized to. + /// + /// Called before the message is sent — and regardless of whether it is — + /// so a batch refused for being oversized still teaches the next one. + fn record_message_size(&self, space: &SLabel, outcome: &SubmitOutcome) { + if outcome.cert_count == 0 || outcome.message_bytes == 0 { + return; + } + self.last_message + .lock() + .unwrap() + .insert(space.clone(), (outcome.cert_count, outcome.message_bytes)); } pub async fn build_message(&self, certs: Vec) -> anyhow::Result { @@ -1493,18 +1639,50 @@ impl Operator { for space_data in space_datas { let space = SName::from_space(&space_data.info.space); + + let root_started = std::time::Instant::now(); let root_cert = space_data .info .issue_cert(rpc, &self.wallet, &space) .await?; + log::debug!( + "[{}] root cert issued in {}ms", + space_data.info.space, + root_started.elapsed().as_millis() + ); certs.push(root_cert); + + // Per-handle timing. Certificate issuance dominates publishing, and + // the cost per handle is what tells you whether a change actually + // helped — a batch total hides which handles were slow and folds in + // fixed per-batch work. + let batch_started = std::time::Instant::now(); + let mut slowest_ms = 0u128; + let count = space_data.handles.len(); + for handle in space_data.handles { + let started = std::time::Instant::now(); let cert = space_data .info .issue_cert(rpc, &self.wallet, &handle) .await?; + let ms = started.elapsed().as_millis(); + slowest_ms = slowest_ms.max(ms); + log::debug!("[{}] cert for {} in {}ms", space_data.info.space, handle, ms); certs.push(cert); } + + if count > 0 { + let total = batch_started.elapsed(); + log::info!( + "[{}] issued {} certs in {}ms ({:.1}ms/cert, slowest {}ms)", + space_data.info.space, + count, + total.as_millis(), + total.as_millis() as f64 / count as f64, + slowest_ms, + ); + } } Ok(certs) @@ -1537,19 +1715,42 @@ impl Operator { log::info!("[{}] Reset {} stale temp cert(s) for republishing", space, reset); } - let all_handles = storage.select_handles( - if only.is_empty() { - crate::storage::HandleSelector::Unpublished(confirmed_idx) + // A relay rejects an oversized message outright, so the caller's limit + // is an upper bound rather than the batch size: what actually fits is + // derived from the last message this space produced. + let limit = if only.is_empty() { + let sized = self.next_batch_size(space, limit); + if sized != limit { + log::debug!("[{}] batch sized to {} (ceiling {})", space, sized, limit); + } + sized + } else { + // An explicit handle list is the caller's choice; honour it and + // let the size guard catch it if it is too large. + limit + }; + + // Only the batch is fetched. `total` comes from a COUNT so the caller + // still learns how much is left without paying to materialise it. + let batch: Vec<_> = storage + .select_handles(if only.is_empty() { + crate::storage::HandleSelector::Unpublished(confirmed_idx, Some(limit)) } else { crate::storage::HandleSelector::ByName(only.to_vec()) - } - ).await?; - if all_handles.is_empty() { + }) + .await? + .into_iter() + .take(limit) + .collect(); + if batch.is_empty() { return Ok((0, 0)); } - let total = all_handles.len(); - let batch: Vec<_> = all_handles.into_iter().take(limit).collect(); + let total = if only.is_empty() { + storage.count_unpublished(confirmed_idx).await? + } else { + batch.len() + }; let handle_names: Vec = batch .iter() @@ -1559,7 +1760,25 @@ impl Operator { let count = handle_names.len(); let certs = self.issue_certs(handle_names).await?; - self.submit_certs(certs).await?; + + let outcome = self.submit_certs(certs).await?; + // Recorded whether or not it was sent: a refused batch is exactly the + // case that must inform the next one, or the same oversized batch is + // rebuilt and refused forever. + self.record_message_size(space, &outcome); + + if !outcome.sent { + // Nothing is marked published, so these handles are picked up + // again next cycle — by then sized from the measurement above. + log::warn!( + "[{}] batch of {} produced a {} byte message and was not sent; \ + retrying smaller", + space, + outcome.cert_count, + outcome.message_bytes, + ); + return Ok((0, total)); + } // Determine temp vs final per handle based on confirmed idx let mut temp_names = Vec::new(); @@ -1581,8 +1800,17 @@ impl Operator { } if !final_names.is_empty() { storage.mark_handles_published(&final_names, "final", None).await?; - // If no more committed handles need publishing, mark commitment as published - if storage.select_handles(crate::storage::HandleSelector::Unpublished(confirmed_idx)).await?.iter().all(|h| h.commitment_root.is_none()) { + // If no more committed handles need publishing, mark commitment as + // published. Only the first page is examined: the predicate is + // "does any committed handle remain", so one counter-example is + // enough and scanning the whole backlog to learn it was waste. + let remaining = storage + .select_handles(crate::storage::HandleSelector::Unpublished( + confirmed_idx, + Some(limit.max(1)), + )) + .await?; + if remaining.iter().all(|h| h.commitment_root.is_none()) { if let Some(commitment) = storage.get_last_commitment().await? { if commitment.published_at.is_none() { storage.mark_commitment_published(commitment.id).await?; diff --git a/core/src/core.rs b/core/src/core.rs index 809d6e7..ced149b 100644 --- a/core/src/core.rs +++ b/core/src/core.rs @@ -21,7 +21,7 @@ use serde::Serialize; use spacedb::db::Database; use spacedb::subtree::SubTree; use spacedb::tx::{ProofType, ReadTransaction}; -use spacedb::{Hash, NodeHasher, Sha256Hasher}; +use spacedb::{Configuration, Hash, NodeHasher, Sha256Hasher}; use spaces_protocol::slabel::SLabel; pub use subs_types::{CompressInput, ProvingRequest}; use tokio::task::spawn_blocking; @@ -68,6 +68,20 @@ impl SkipReason { SkipReason::AlreadyStaged => "already staged", } } + + /// Stable token reported to a registry in the ack body. + /// + /// Separate from `as_str`, which is prose for logs and can be reworded + /// freely. These are a protocol contract: a registry matches on them to + /// decide what to tell a user, so they must not change once published. + pub fn as_outcome(&self) -> &'static str { + match self { + SkipReason::AlreadyCommittedDifferentSpk => "already_committed_different_spk", + SkipReason::AlreadyStagedDifferentSpk => "already_staged_different_spk", + SkipReason::AlreadyCommitted => "already_committed_same_spk", + SkipReason::AlreadyStaged => "already_staged_same_spk", + } + } } /// Result of committing staged entries @@ -202,7 +216,28 @@ impl LocalSpace { } let db_path = dir.join(format!("{}.sdb", name)); - let db = spawn_blocking(move || Database::open(db_path.to_str().unwrap())).await??; + let config = Configuration::standard() + .with_auto_hash_index(true); + + // Index building is idempotent — it fingerprints the root and skips a + // valid existing index — so this is only expensive the first time a + // space is opened after the feature is enabled. It runs on the + // blocking pool with the open because indexing every snapshot of a + // large tree is real CPU and I/O, not something to do on the runtime. + // + // A failure is logged rather than propagated: the index is an + // optimisation, and a space that cannot be indexed must still open. + let db = spawn_blocking(move || -> anyhow::Result<_> { + let db = Database::open_with_config(db_path.to_str().unwrap(), config)?; + for snap in db.iter() { + match snap.and_then(|mut s| s.build_hash_index()) { + Ok(()) => {} + Err(e) => log::warn!("could not build hash index: {} (proofs will be slower)", e), + } + } + Ok(db) + }) + .await??; Ok(Self { name, @@ -996,7 +1031,16 @@ impl LocalSpace { commitment_id: i64, estimate_json: &str, ) -> anyhow::Result<()> { - self.storage.update_commitment_estimate(commitment_id, estimate_json).await + self.storage + .update_commitment_estimate(commitment_id, Some(estimate_json)) + .await + } + + /// Drop the stored estimate once the proof it described has finished. + pub async fn clear_estimate(&self, commitment_id: i64) -> anyhow::Result<()> { + self.storage + .update_commitment_estimate(commitment_id, None) + .await } } @@ -1403,6 +1447,7 @@ fn get_snapshot_for_tip(db: &Database, tip: [u8;32]) -> anyhow::Re Err(anyhow!("no snapshot for {}", hex::encode(&tip))) } + fn rollback_local_commitment(db: &Database, root: [u8; 32]) -> anyhow::Result<()> { let mut found = false; diff --git a/core/src/storage.rs b/core/src/storage.rs index d5fe0db..d3092b7 100644 --- a/core/src/storage.rs +++ b/core/src/storage.rs @@ -91,8 +91,11 @@ pub struct Handle { /// Selector for querying handles for publishing. pub enum HandleSelector { - /// Handles needing certs: unpublished or temp-published ready for finalization. - Unpublished(Option), + /// Handles needing certs: unpublished or temp-published ready for + /// finalization. The second field caps how many are returned; `None` means + /// unbounded, which is only appropriate when the caller genuinely needs + /// every row. + Unpublished(Option, Option), /// Specific committed handles by name (regardless of publish status). ByName(Vec), } @@ -409,13 +412,18 @@ impl Storage { .await? } + /// Store or clear a commitment's proving estimate. + /// + /// `None` clears it. An estimate describes one specific proof, and the + /// commitment row holds only one, so a finished proof's estimate has to be + /// cleared rather than left to be read as a prediction for the next. pub async fn update_commitment_estimate( &self, commitment_id: i64, - estimate_json: &str, + estimate_json: Option<&str>, ) -> anyhow::Result<()> { let conn = self.conn.clone(); - let json = estimate_json.to_string(); + let json = estimate_json.map(|s| s.to_string()); spawn_blocking(move || { let conn = conn.lock().unwrap(); conn.execute( @@ -921,6 +929,26 @@ impl Storage { // Publishing /// Select handles for publishing based on the given selector. + /// How many handles still need certificates. + /// + /// Counted in SQL. The pipeline reports this on every poll, and loading + /// every row just to call `.len()` made that cost grow with the backlog. + pub async fn count_unpublished(&self, confirmed_idx: Option) -> anyhow::Result { + let conn = self.conn.clone(); + spawn_blocking(move || { + let conn = conn.lock().unwrap(); + let idx_param = confirmed_idx.map(|v| v as i64).unwrap_or(-1); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM handles WHERE publish_status IS NULL \ + OR (publish_status = 'temp' AND commitment_idx IS NOT NULL AND commitment_idx <= ?)", + params![idx_param], + |row| row.get(0), + )?; + Ok(count as usize) + }) + .await? + } + pub async fn select_handles(&self, selector: HandleSelector) -> anyhow::Result> { let conn = self.conn.clone(); spawn_blocking(move || { @@ -940,15 +968,23 @@ impl Storage { }) }; match selector { - HandleSelector::Unpublished(confirmed_idx) => { + HandleSelector::Unpublished(confirmed_idx, limit) => { + // Bounded in SQL. Callers publish a batch at a time, so + // materialising the whole backlog to take the first N made + // each cycle proportional to work already queued — slower + // the further behind it fell. let sql = format!( "SELECT {} FROM handles WHERE publish_status IS NULL \ OR (publish_status = 'temp' AND commitment_idx IS NOT NULL AND commitment_idx <= ?) \ - ORDER BY name ASC", cols + ORDER BY name ASC LIMIT ?", cols ); let idx_param = confirmed_idx.map(|v| v as i64).unwrap_or(-1); + // SQLite treats a negative LIMIT as unbounded. + let limit_param = limit.map(|v| v as i64).unwrap_or(-1); let mut stmt = conn.prepare(&sql)?; - let result: Vec = stmt.query_map(params![idx_param], map_row)?.collect::, _>>()?; + let result: Vec = stmt + .query_map(params![idx_param, limit_param], map_row)? + .collect::, _>>()?; Ok(result) } HandleSelector::ByName(names) => { diff --git a/examples/registry-server/README.md b/examples/registry-server/README.md index db6f2ed..b71b2e2 100644 --- a/examples/registry-server/README.md +++ b/examples/registry-server/README.md @@ -69,8 +69,8 @@ All authenticated endpoints expect `Authorization: Bearer ` and return | Method | Endpoint | Description | |--------|----------|-------------| | GET | `/health` | Liveness, and confirms the token is accepted | -| GET | `/pending` | Get pending handles to stage | -| POST | `/ack` | Acknowledge handles were staged | +| GET | `/pending` | Get pending handles to stage; filters on `?space=` | +| POST | `/ack` | Record the per-handle outcome subsd reached | | POST | `/committed` | Notify when handles are committed | The keys are separate because they have different blast radii: the subsd key @@ -101,7 +101,9 @@ curl http://localhost:8080/status/alice@example ### Get pending handles (subsd) ```bash +# subsd asks one space at a time; unscoped returns everything. curl -H "Authorization: Bearer $SUBSD_API_KEY" \ + --get --data-urlencode "space=@example" \ http://localhost:8080/pending ``` @@ -111,7 +113,10 @@ curl -H "Authorization: Bearer $SUBSD_API_KEY" \ curl -X POST http://localhost:8080/ack \ -H "Authorization: Bearer $SUBSD_API_KEY" \ -H "Content-Type: application/json" \ - -d '{"handles": ["alice@example"]}' + -d '{"handles": [ + {"handle": "alice@example", "outcome": "staged"}, + {"handle": "bob@example", "outcome": "already_committed_different_spk"} + ]}' ``` ## Production Considerations diff --git a/examples/registry-server/src/main.rs b/examples/registry-server/src/main.rs index cc4b14c..ce8439e 100644 --- a/examples/registry-server/src/main.rs +++ b/examples/registry-server/src/main.rs @@ -31,7 +31,7 @@ use std::net::SocketAddr; use std::sync::Arc; use axum::{ - extract::{Path, Request, State}, + extract::{Path, Query, Request, State}, http::{HeaderMap, StatusCode}, middleware::{self, Next}, response::{IntoResponse, Response}, @@ -80,6 +80,7 @@ enum RegistrationStatus { Pending, // Waiting to be pulled by subsd Staged, // Pulled by subsd, waiting for commit Committed, // On-chain + Rejected, // subsd reported a terminal failure; see the ack outcome } #[tokio::main] @@ -301,6 +302,7 @@ async fn get_status( RegistrationStatus::Pending => "pending", RegistrationStatus::Staged => "staged", RegistrationStatus::Committed => "committed", + RegistrationStatus::Rejected => "rejected", }; Json(StatusResponse { handle: reg.handle.clone(), @@ -328,25 +330,61 @@ struct PendingResponse { } /// GET /pending - Get pending handles for subsd to stage -async fn get_pending_handles(State(state): State>) -> impl IntoResponse { +#[derive(Deserialize)] +struct PendingQuery { + /// The single space the caller is asking about, e.g. "@example". + /// Absent means unscoped: return everything, which is what a registry + /// written before scoping existed does by default. + space: Option, +} + +/// The space a handle belongs to: everything after the last '@'. +fn handle_space(handle: &str) -> Option { + handle.rsplit_once('@').map(|(_, space)| format!("@{}", space)) +} + +async fn get_pending_handles( + State(state): State>, + Query(q): Query, +) -> impl IntoResponse { let registrations = state.registrations.read().await; let pending: Vec = registrations .iter() .filter(|r| r.status == RegistrationStatus::Pending) + // Only hand over work for the space that was asked about. Without + // this, handles for a space subsd cannot act on come back on every + // cycle and are never acked, because they can never stage. + .filter(|r| match &q.space { + None => true, + Some(want) => handle_space(&r.handle).as_deref() == Some(want.as_str()), + }) + // A real registry would paginate here; subsd stages a whole response + // in one pass, so capping the page and letting the next cycle collect + // the rest is the natural place to bound it. .map(|r| PendingHandle { handle: r.handle.clone(), script_pubkey: r.script_pubkey.clone(), }) .collect(); - tracing::info!("Returning {} pending handles", pending.len()); + match &q.space { + Some(space) => tracing::info!("Returning {} pending handles for {}", pending.len(), space), + None => tracing::info!("Returning {} pending handles (unscoped)", pending.len()), + } Json(PendingResponse { handles: pending }) } +#[derive(Deserialize)] +struct AckEntry { + handle: String, + /// Why the handle is settled. See REGISTRY.md; every value is terminal. + outcome: String, +} + #[derive(Deserialize)] struct AckRequest { - handles: Vec, + handles: Vec, } #[derive(Serialize)] @@ -354,7 +392,12 @@ struct AckResponse { acknowledged: usize, } -/// POST /ack - Acknowledge handles were staged by subsd +/// POST /ack - Record the outcome subsd reached for each handle. +/// +/// Every outcome is terminal, so all of them leave /pending. A real registry +/// would branch here: `staged` is on its way, while the `*_different_spk` and +/// `invalid` outcomes mean the request can never be fulfilled and the user +/// should be told — and refunded, if they paid. async fn ack_handles( State(state): State>, Json(req): Json, @@ -362,12 +405,36 @@ async fn ack_handles( let mut registrations = state.registrations.write().await; let mut count = 0; - for handle in &req.handles { - if let Some(reg) = registrations.iter_mut().find(|r| r.handle == *handle) { + for entry in &req.handles { + if let Some(reg) = registrations.iter_mut().find(|r| r.handle == entry.handle) { if reg.status == RegistrationStatus::Pending { - reg.status = RegistrationStatus::Staged; + reg.status = match entry.outcome.as_str() { + "staged" | "already_staged_same_spk" => RegistrationStatus::Staged, + // Already registered to the requested owner: the request is + // fulfilled, not refused. Rejecting it would tell a paying + // user their own handle was denied. + "already_committed_same_spk" => RegistrationStatus::Committed, + // Unfulfillable: taken by another script pubkey, or never + // a valid handle. Parked as Rejected rather than Staged so + // it isn't reported as in-flight forever. + "already_committed_different_spk" + | "already_staged_different_spk" + | "invalid" => RegistrationStatus::Rejected, + // An outcome this example predates. Treated as unfulfillable + // so nothing is stuck pending, but listed separately from + // the known-terminal arm above: a new outcome is a prompt to + // read the table in REGISTRY.md, not to assume refusal. + other => { + tracing::warn!( + "Handle {}: unrecognised outcome {:?}; treating as rejected", + entry.handle, + other + ); + RegistrationStatus::Rejected + } + }; count += 1; - tracing::info!("Handle {} acknowledged as staged", handle); + tracing::info!("Handle {} acked: {}", entry.handle, entry.outcome); } } } diff --git a/prover/src/lib.rs b/prover/src/lib.rs index 66ae237..4ee5d8f 100644 --- a/prover/src/lib.rs +++ b/prover/src/lib.rs @@ -6,12 +6,20 @@ pub mod server; use std::time::Instant; +pub mod progress; + use anyhow::{anyhow, Result}; +use std::sync::Arc; use libveritas::constants::{FOLD_ELF, FOLD_ID, STEP_ELF, STEP_ID}; use libveritas_zk::guest::Commitment; -use risc0_zkvm::{default_executor, default_prover, ExecutorEnv, ProverOpts, Receipt}; +use risc0_zkvm::{ + default_executor, default_prover, get_prover_server, ExecutorEnv, ExecutorImpl, ProverOpts, + Receipt, VerifierContext, +}; use spacedb::{NodeHasher, Sha256Hasher}; use spacedb::subtree::{ProofType, SubTree, ValueOrHash}; +pub use crate::progress::{JobProgress, ProgressSink}; +use crate::progress::SegmentProgress; use subs_types::{CalibrationInfo, CompressInput, EstimateResult, ProvingRequest, SegmentEstimate}; /// Build a synthetic ProvingRequest::Step for benchmarking/calibration. @@ -82,13 +90,22 @@ impl Prover { /// Prove a ProvingRequest and return the serialized receipt. pub fn prove(&self, request: &ProvingRequest) -> Result> { + self.prove_with_progress(request, None) + } + + /// Prove, reporting progress into `sink` as segments complete. + pub fn prove_with_progress( + &self, + request: &ProvingRequest, + sink: Option>, + ) -> Result> { match request { ProvingRequest::Step { idx, exclusion_proof, zk_batch, .. - } => self.prove_step(*idx, exclusion_proof, zk_batch), + } => self.prove_step_inner(*idx, exclusion_proof, zk_batch, sink), ProvingRequest::Fold { idx, acc_receipt, @@ -96,17 +113,70 @@ impl Prover { step_receipt, step_commitment, .. - } => self.prove_fold( + } => self.prove_fold_inner( *idx, acc_receipt, acc_commitment, step_receipt, step_commitment, + sink, ), } } - fn prove_step(&self, idx: usize, exclusion_proof: &[u8], zk_batch: &[u8]) -> Result> { + /// Execute, publish the session facts, then prove — reporting each segment. + /// + /// This is the same split risc0 performs internally: `prove_with_opts` + /// resolves to `ExecutorImpl::from_elf(env, elf).run()` followed by + /// `prove_session`, with the same opts on the same prover server. Doing it + /// explicitly changes nothing about the receipt; it only makes the session + /// visible so cycle counts and segment progress can be reported while the + /// job is still running, instead of arriving with the receipt. + /// + /// Note this proves in-process via `get_prover_server`, so unlike + /// `default_prover` it ignores RISC0_PROVER and BONSAI_API_*. That is the + /// intent — this binary *is* the prover a client dials — but it means + /// setting those env vars will not redirect proving elsewhere. + fn prove_session_with_progress( + &self, + idx: usize, + env: ExecutorEnv<'_>, + elf: &[u8], + sink: Option>, + ) -> Result { + let opts = ProverOpts::succinct(); + + let mut session = ExecutorImpl::from_elf(env, elf) + .map_err(|e| anyhow!("[#{}] executor init: {}", idx, e))? + .run() + .map_err(|e| anyhow!("[#{}] execute failed: {}", idx, e))?; + + if let Some(sink) = &sink { + // Published before proving starts: the segment count and cycle + // total are the whole point of reporting early. + sink.on_session_ready(session.user_cycles, session.segments.len()); + session.add_hook(SegmentProgress::new(sink.clone())); + } + + let ctx = VerifierContext::default().with_dev_mode(opts.dev_mode()); + let prover = get_prover_server(&opts) + .map_err(|e| anyhow!("[#{}] prover server: {}", idx, e))?; + + let info = prover + .prove_session(&ctx, &session) + .map_err(|e| anyhow!("[#{}] prove failed: {}", idx, e))?; + + Ok(info.receipt) + } + + + fn prove_step_inner( + &self, + idx: usize, + exclusion_proof: &[u8], + zk_batch: &[u8], + sink: Option>, + ) -> Result> { let env = ExecutorEnv::builder() .write(&( exclusion_proof.to_vec(), @@ -118,23 +188,22 @@ impl Prover { .build() .map_err(|e| anyhow!("[#{}] env build: {}", idx, e))?; - let prove_info = default_prover() - .prove_with_opts(env, STEP_ELF, &ProverOpts::succinct()) - .map_err(|e| anyhow!("[#{}] prove step failed: {}", idx, e))?; + let receipt = self.prove_session_with_progress(idx, env, STEP_ELF, sink)?; - let receipt_bytes = borsh::to_vec(&prove_info.receipt) + let receipt_bytes = borsh::to_vec(&receipt) .map_err(|e| anyhow!("[#{}] serialize receipt: {}", idx, e))?; Ok(receipt_bytes) } - fn prove_fold( + fn prove_fold_inner( &self, idx: usize, acc_receipt: &[u8], acc_commitment: &Commitment, step_receipt: &[u8], step_commitment: &Commitment, + sink: Option>, ) -> Result> { let acc: Receipt = borsh::from_slice(acc_receipt) .map_err(|e| anyhow!("deserialize acc receipt: {}", e))?; @@ -149,11 +218,11 @@ impl Prover { .build() .map_err(|e| anyhow!("[#{}] env build: {}", idx, e))?; - let prove_info = default_prover() - .prove_with_opts(env, FOLD_ELF, &ProverOpts::succinct()) - .map_err(|e| anyhow!("[#{}] fold prove failed: {}", idx, e))?; + // Same instrumented path as the step half. Assumption resolution lives + // inside prove_session, which is all prove_with_opts would have added. + let receipt = self.prove_session_with_progress(idx, env, FOLD_ELF, sink)?; - let receipt_bytes = borsh::to_vec(&prove_info.receipt) + let receipt_bytes = borsh::to_vec(&receipt) .map_err(|e| anyhow!("[#{}] serialize receipt: {}", idx, e))?; Ok(receipt_bytes) diff --git a/prover/src/main.rs b/prover/src/main.rs index b7b35e1..5c320e1 100644 --- a/prover/src/main.rs +++ b/prover/src/main.rs @@ -40,14 +40,15 @@ struct Cli { #[arg(long, default_value = "8888")] server_port: u16, - /// Skip startup calibration (for --server mode). + /// Run 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. + /// short-lived GPU pod that delay is billed on every cold start, so it is + /// off by default and only worth paying for when /estimate is wanted. + /// Also settable via PROVER_CALIBRATE. #[arg(long)] - no_calibrate: bool, + calibrate: bool, #[command(subcommand)] cmd: Option, @@ -89,9 +90,9 @@ async fn main() -> Result<()> { let cli = Cli::parse(); if cli.server { - 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?; + let calibrate = cli.calibrate + || std::env::var("PROVER_CALIBRATE").is_ok_and(|v| !v.is_empty() && v != "0"); + subs_prover::server::run_server(cli.server_port, calibrate).await?; return Ok(()); } diff --git a/prover/src/progress.rs b/prover/src/progress.rs new file mode 100644 index 0000000..672a7f3 --- /dev/null +++ b/prover/src/progress.rs @@ -0,0 +1,234 @@ +//! Live progress for an in-flight proving job. +//! +//! The prover is the only place that knows how long a job will take: the +//! session gives the segment count before proving starts, and risc0 fires a +//! hook around each segment as it is proven. Together those turn "processing" +//! into an ETA measured on this GPU, for this job — rather than extrapolated +//! from a synthetic calibration run on some other pod. + +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +use risc0_zkvm::{Segment, SessionEvents}; +use serde::Serialize; + +/// A snapshot of a job's progress, safe to serialize into a status response. +#[derive(Debug, Clone, Serialize)] +pub struct JobProgress { + /// User cycles executed by the guest. + pub total_cycles: u64, + /// Padded proving cycles across the segments proven so far. Equals the + /// job's true total once `segments_done == segments`. + pub proving_cycles_done: u64, + /// Segments this job will prove. Known before proving starts. + pub segments: usize, + /// Segments proven so far. + pub segments_done: usize, + /// Wall-clock since proving began. + pub elapsed_seconds: f64, + /// Projected total wall-clock. `None` until a segment completes — there is + /// nothing to extrapolate from before that. + pub estimated_total_seconds: Option, + /// Which phase the job is in, 1-based. + /// + /// 1. Proving segments. Determinate: `segments_done` of `segments`. + /// 2. Lift/join/resolve, turning the composite segment receipts into the + /// succinct receipt. risc0 exposes no `SessionEvents` hook for this, so + /// nothing can be observed while it runs — it is genuinely + /// indeterminate, not merely unmeasured. + /// + /// Reporting it matters because phase 2 is not a tail: on a measured + /// single-segment step proof, segments finished at 10.7s of 38.8s, so 72% + /// of the wall-clock happened in phase 2 with the bar already full. + pub phase: u8, + /// Total phases, so the UI need not hardcode it. + pub phase_total: u8, + /// Fraction of phase 1 complete (0.0–1.0), interpolated within the segment + /// currently being proven so the bar advances between completions. + /// + /// `None` while the first segment is proving — there is no measured segment + /// duration to interpolate against yet — and throughout phase 2. + pub phase_one_fraction: Option, + /// Wall-clock of the first segment. + /// + /// Worth reporting separately: only sm_80 gets native SASS in the shipped + /// image, so every other GPU JIT-compiles PTX on its first kernel launch + /// and pays for it here. Comparing this against the mean is what tells an + /// operator whether baking in their card's SASS is worth the build time. + pub first_segment_seconds: Option, +} + +/// Shared progress counters, written by the proving hook and read by the +/// status endpoint while the job runs. +pub struct ProgressSink { + started: Instant, + total_cycles: AtomicU64, + segments: AtomicUsize, + segments_done: AtomicUsize, + proving_cycles_done: AtomicU64, + /// Millis, so it fits an atomic without a lock. + first_segment_millis: AtomicU64, + /// When the most recent segment landed. + /// + /// Estimates must extrapolate from segment *completions*, not from current + /// elapsed time: between two completions no new information arrives, so an + /// elapsed-based projection inflates on every poll and "remaining" tracks + /// "elapsed" exactly. + last_segment_millis: AtomicU64, + /// Elapsed at the moment proving stopped, or 0 while it runs. + /// + /// Without this, `snapshot()` keeps deriving elapsed from `started`, so a + /// finished job's /jobs/:id keeps ageing — reporting an ever-growing + /// duration for work that ended. subs stops polling once the receipt is + /// pulled, but anything else reading the prover API sees it. + finished_millis: AtomicU64, +} + +impl ProgressSink { + pub fn new() -> Arc { + Arc::new(Self { + started: Instant::now(), + total_cycles: AtomicU64::new(0), + segments: AtomicUsize::new(0), + segments_done: AtomicUsize::new(0), + proving_cycles_done: AtomicU64::new(0), + first_segment_millis: AtomicU64::new(0), + last_segment_millis: AtomicU64::new(0), + finished_millis: AtomicU64::new(0), + }) + } + + /// Stop the clock. Idempotent — the first call wins, so a job that is + /// finished twice (completed then cancelled, say) keeps its real duration. + pub fn finish(&self) { + let ms = self.started.elapsed().as_millis() as u64; + let _ = self + .finished_millis + .compare_exchange(0, ms.max(1), Ordering::Relaxed, Ordering::Relaxed); + } + + /// Record what the session tells us before any segment is proven. + pub fn on_session_ready(&self, total_cycles: u64, segments: usize) { + self.total_cycles.store(total_cycles, Ordering::Relaxed); + self.segments.store(segments, Ordering::Relaxed); + } + + fn on_segment_proven(&self, po2: usize) { + self.proving_cycles_done + .fetch_add(1u64 << po2, Ordering::Relaxed); + let done = self.segments_done.fetch_add(1, Ordering::Relaxed) + 1; + let ms = self.started.elapsed().as_millis() as u64; + self.last_segment_millis.store(ms, Ordering::Relaxed); + if done == 1 { + self.first_segment_millis.store(ms, Ordering::Relaxed); + } + } + + pub fn snapshot(&self) -> JobProgress { + let segments = self.segments.load(Ordering::Relaxed); + let done = self.segments_done.load(Ordering::Relaxed); + // Frozen once the job ends, so a finished job reports the duration it + // actually took rather than time since it started. + let finished_ms = self.finished_millis.load(Ordering::Relaxed); + let elapsed = if finished_ms > 0 { + finished_ms as f64 / 1000.0 + } else { + self.started.elapsed().as_secs_f64() + }; + let first_ms = self.first_segment_millis.load(Ordering::Relaxed); + + // Extrapolate from observed segments. Remaining segments' po2 is not + // known without resolving them, so the best available projection is the + // mean rate so far — which is exactly what a po2-weighted estimate + // reduces to when the remaining work is unknown. + // + // Crucially this extrapolates from segment *completion* timestamps, not + // from current elapsed time. Between two completions no new information + // arrives, so an elapsed-based projection grows on every poll: with one + // of two segments done it computed `elapsed / 1 * 2`, making "remaining" + // exactly equal "elapsed" and climb forever. The first segment's + // duration is already measured — the estimate must hold still until the + // next one actually lands. + // + // The first segment carries GPU warm-up and any PTX JIT, so once a + // second segment lands it is excluded from the rate and only the steady + // -state segments are extrapolated. + // + // Phase 2 (lift/join/resolve) is unobservable, so once every segment is + // proven there is nothing left to extrapolate from. The old code fell + // through to the `done == segments` case and reported + // `total == elapsed` on every poll, which claims "finishing now" for + // the entire phase -- 28.1s of a measured 38.8s proof. No estimate is + // the honest answer; the UI shows phase 2 as indeterminate. + let in_phase_two = segments > 0 && done >= segments; + let first = first_ms as f64 / 1000.0; + let last = self.last_segment_millis.load(Ordering::Relaxed) as f64 / 1000.0; + + let estimated_total_seconds = if in_phase_two || done == 0 || segments == 0 { + None + } else if done == 1 { + // Only the first segment has been timed, and it includes warm-up, + // so this runs high. It is at least stable. + Some(first * segments as f64) + } else { + let steady_rate = (last - first) / (done - 1) as f64; + Some(first + steady_rate * (segments - 1) as f64) + }; + + // If elapsed has already overtaken the projection, the current segment + // is slower than the ones it was extrapolated from and the estimate is + // simply wrong. Dropping it shows no ETA; clamping it to `elapsed` + // would report "~0 seconds remaining" while work continues, which is + // the same falsehood this pass set out to remove. + let estimated_total_seconds = estimated_total_seconds.filter(|est| *est > elapsed); + + // Fraction of phase 1 complete, interpolated inside the segment being + // proven so the bar moves continuously instead of stepping once per + // completion. Segments here run over a minute, so a step-only bar looks + // stalled for most of the job. + // + // The current segment is capped at 90% of its share: it must not look + // finished before it is. Interpolation needs a segment duration to + // scale against, so it only starts once one has been measured — the + // very first segment has no basis and reports None, which the UI shows + // as an indeterminate bar rather than a fabricated position. + let phase_one_fraction = if in_phase_two || segments == 0 || done == 0 || last <= 0.0 { + None + } else { + let mean_segment = last / done as f64; + let within = ((elapsed - last).max(0.0) / mean_segment).min(0.9); + Some(((done as f64 + within) / segments as f64).min(1.0)) + }; + + JobProgress { + total_cycles: self.total_cycles.load(Ordering::Relaxed), + proving_cycles_done: self.proving_cycles_done.load(Ordering::Relaxed), + segments, + segments_done: done, + elapsed_seconds: elapsed, + estimated_total_seconds, + phase: if in_phase_two { 2 } else { 1 }, + phase_total: 2, + phase_one_fraction, + first_segment_seconds: (first_ms > 0).then_some(first_ms as f64 / 1000.0), + } + } +} + +/// Bridges risc0's per-segment proving hook into a [`ProgressSink`]. +pub struct SegmentProgress { + sink: Arc, +} + +impl SegmentProgress { + pub fn new(sink: Arc) -> Self { + Self { sink } + } +} + +impl SessionEvents for SegmentProgress { + fn on_post_prove_segment(&self, segment: &Segment) { + self.sink.on_segment_proven(segment.po2()); + } +} diff --git a/prover/src/server.rs b/prover/src/server.rs index 3a1b2ca..39177a6 100644 --- a/prover/src/server.rs +++ b/prover/src/server.rs @@ -20,7 +20,7 @@ use tokio::sync::{mpsc, RwLock}; use tower_http::cors::{Any, CorsLayer}; use tower_http::trace::TraceLayer; -use crate::Prover; +use crate::{JobProgress, ProgressSink, Prover}; use subs_types::{CompressInput, ProvingRequest}; /// Job status @@ -31,6 +31,7 @@ pub enum JobStatus { Processing, Complete, Failed, + Cancelled, } /// Job type @@ -51,6 +52,13 @@ pub struct Job { pub request: JobRequest, pub receipt: Option>, pub error: Option, + /// Live counters, shared with the proving thread. Attached before proving + /// starts so /jobs/:id can report progress while the job is still running. + pub progress: Option>, + /// Set by /jobs/:id/cancel. A queued job never starts; a running one is + /// abandoned when it finishes — see cancel_job for why it cannot be + /// interrupted mid-proof. + pub cancel_requested: bool, } #[derive(Clone)] @@ -92,6 +100,10 @@ pub struct JobStatusResponse { pub job_type: JobType, pub status: JobStatus, pub error: Option, + /// Absent until execution finishes and proving begins, and for compress + /// jobs, which have no segments. Consumers that predate this ignore it. + #[serde(skip_serializing_if = "Option::is_none")] + pub progress: Option, } /// Error response @@ -114,7 +126,7 @@ pub struct ErrorResponse { /// 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<()> { +pub async fn run_server(port: u16, calibrate: bool) -> anyhow::Result<()> { // Initialize tracing tracing_subscriber::fmt() .with_env_filter( @@ -131,9 +143,10 @@ pub async fn run_server(port: u16, no_calibrate: bool) -> anyhow::Result<()> { // 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"); + // estimate is useless before it, but billable on a short-lived pod. Hence + // opt-in: most runs want the server answering immediately. + if !calibrate { + tracing::info!("Calibration off (pass --calibrate to enable); /estimate will be unavailable"); } else { tracing::info!("Calibrating proving throughput..."); let calibrate_state = state.clone(); @@ -183,7 +196,8 @@ pub async fn run_server(port: u16, no_calibrate: bool) -> anyhow::Result<()> { .route("/compress", post(submit_compress)) .route("/jobs/:job_id", get(get_job_status)) .route("/jobs/:job_id/receipt", get(get_job_receipt)) - .route("/calibration", get(get_calibration)); + .route("/calibration", get(get_calibration)) + .route("/jobs/:job_id/cancel", post(cancel_job)); if let Some(token) = auth_token { app = app.layer(middleware::from_fn(move |req: Request, next: Next| { let token = token.clone(); @@ -215,6 +229,60 @@ pub async fn run_server(port: u16, no_calibrate: bool) -> anyhow::Result<()> { } /// Health check endpoint +/// POST /jobs/:job_id/cancel - Stop a job. +/// +/// A queued job is dropped before it starts, which is the case that matters: +/// it frees the worker for everything behind it. +/// +/// A running job cannot be interrupted. risc0's `prove_session` proves the +/// whole session in one call, and its per-segment hook returns `()`, so there +/// is no cancellation point that does not involve unwinding through the +/// prover — not worth the risk of leaving a CUDA context in a bad state to +/// reclaim part of one proof. Such a job is marked cancelled, keeps running to +/// completion, and has its receipt discarded. The GPU time is already spent; +/// the honest way to reclaim it is to terminate the pod. +async fn cancel_job( + State(state): State>, + Path(job_id): Path, +) -> impl IntoResponse { + let mut jobs = state.jobs.write().await; + let Some(job) = jobs.get_mut(&job_id) else { + return ( + StatusCode::NOT_FOUND, + Json(ErrorResponse { + error: "Job not found".to_string(), + }), + ) + .into_response(); + }; + + match job.status { + JobStatus::Complete | JobStatus::Failed | JobStatus::Cancelled => ( + StatusCode::CONFLICT, + Json(ErrorResponse { + error: format!("job already {:?}", job.status), + }), + ) + .into_response(), + JobStatus::Pending => { + job.cancel_requested = true; + job.status = JobStatus::Cancelled; + tracing::info!("Job {} cancelled before starting", job_id); + Json(serde_json::json!({ "cancelled": true, "was_running": false })).into_response() + } + JobStatus::Processing => { + job.cancel_requested = true; + tracing::info!("Job {} cancel requested while running", job_id); + Json(serde_json::json!({ + "cancelled": true, + "was_running": true, + "note": "proof runs to completion; receipt discarded" + })) + .into_response() + } + } +} + /// GET /calibration - Measured proving throughput of this machine. /// /// The number that characterises a GPU for cost purposes: cost per proof is @@ -289,6 +357,8 @@ async fn submit_prove( request: JobRequest::Prove(request), receipt: None, error: None, + progress: None, + cancel_requested: false, }; // Add to queue @@ -388,6 +458,8 @@ async fn submit_compress( request: JobRequest::Compress(input), receipt: None, error: None, + progress: None, + cancel_requested: false, }; // Add to queue @@ -428,6 +500,7 @@ async fn get_job_status( job_type: job.job_type.clone(), status: job.status.clone(), error: job.error.clone(), + progress: job.progress.as_ref().map(|p| p.snapshot()), }), ) .into_response(), @@ -515,6 +588,11 @@ async fn run_worker(state: Arc, mut rx: mpsc::Receiver) { let job_request = { let mut jobs = state.jobs.write().await; match jobs.get_mut(&job_id) { + // Cancelled while queued: never start it. + Some(job) if job.cancel_requested => { + tracing::info!("Skipping cancelled job {}", job_id); + None + } Some(job) => { job.status = JobStatus::Processing; Some(job.request.clone()) @@ -537,7 +615,18 @@ async fn run_worker(state: Arc, mut rx: mpsc::Receiver) { JobRequest::Prove(req) => { let idx = req.idx(); tracing::info!("[#{}] Starting proof...", idx); - prover.prove(req) + + // Publish the sink before proving so the status endpoint can + // read counters as segments land, rather than only once the + // receipt exists. + let sink = ProgressSink::new(); + { + let mut jobs = state.jobs.write().await; + if let Some(job) = jobs.get_mut(&job_id) { + job.progress = Some(sink.clone()); + } + } + prover.prove_with_progress(req, Some(sink)) } JobRequest::Compress(input) => { tracing::info!("Starting SNARK compression..."); @@ -549,12 +638,36 @@ async fn run_worker(state: Arc, mut rx: mpsc::Receiver) { { let mut jobs = state.jobs.write().await; if let Some(job) = jobs.get_mut(&job_id) { + // Stop the progress clock before recording the outcome, so a + // finished job reports how long it took rather than how long + // ago it started. + if let Some(sink) = &job.progress { + sink.finish(); + } match result { + // Cancelled mid-proof: the work finished, but the caller no + // longer wants it, so the receipt is dropped rather than + // stored. + Ok(receipt) if job.cancel_requested => { + tracing::info!( + "Job {} finished but was cancelled; discarding {} byte receipt", + job_id, + receipt.len() + ); + job.status = JobStatus::Cancelled; + } Ok(receipt) => { tracing::info!("Job {} complete ({} bytes)", job_id, receipt.len()); job.status = JobStatus::Complete; job.receipt = Some(receipt); } + // A cancelled job that then errored is still cancelled: the + // caller asked for it to stop and does not want the failure + // of work they abandoned reported back as a fault. + Err(e) if job.cancel_requested => { + tracing::info!("Job {} cancelled; it ended with: {}", job_id, e); + job.status = JobStatus::Cancelled; + } Err(e) => { tracing::error!("Job {} failed: {}", job_id, e); job.status = JobStatus::Failed; diff --git a/subs/src/background.rs b/subs/src/background.rs index b2051c7..2c01fe7 100644 --- a/subs/src/background.rs +++ b/subs/src/background.rs @@ -108,6 +108,19 @@ async fn registry_loop(state: AppState) { .await { Ok((0, 0)) => {} + // Nothing published but work remains: the batch was built and + // measured, then declined for exceeding the relay's message + // limit. submit_certs has already logged the size and recorded + // it, so the next pass rebuilds a smaller one — saying + // "Published 0" here would read as a stall rather than a resize. + Ok((0, remaining)) => { + sent_any = true; + tracing::debug!( + "[{}] Batch declined as oversized, {} cert(s) still pending", + space, + remaining + ); + } Ok((published, remaining)) => { sent_any = true; tracing::info!( @@ -179,6 +192,14 @@ async fn proving_loop(state: AppState) { match poll_job(&state, &prover_endpoint, prover_auth_token.as_deref(), space, &job_key, &job_id, commitment_id, is_fold).await { Ok(true) => { tracing::info!("[{}] Proof complete for commitment {}", space, commitment_id); + // The estimate described the proof that just finished. + // A commitment holds only one, so leaving it would show + // the completed step's figures beside a Prove button + // offering the fold — a different shape of work. + // Cleared here; the loop refetches for whatever is next. + if let Err(e) = state.operator.clear_estimate(space, commitment_id).await { + tracing::debug!("[{}] Could not clear estimate: {}", space, e); + } } Ok(false) => {} Err(e) => { @@ -186,10 +207,13 @@ async fn proving_loop(state: AppState) { } } } else { - // Only fetch and store the estimate; proving is user-initiated via the UI - if let Err(e) = fetch_and_store_estimate(&state, &prover_endpoint, prover_auth_token.as_deref(), space, commitment_id, &request).await { - tracing::debug!("[{}] Could not fetch estimate: {}", space, e); - } + // Estimate fetching is disabled along with its UI. It ran on + // every pass — unguarded — and /estimate executes the guest to + // count cycles, so an unproven commitment re-executed it every + // 10s indefinitely, for a figure nothing renders and that read + // ~43% low anyway (calibration measures composite(), real jobs + // run succinct()). Re-enable together with the display. + let _ = &request; } } @@ -199,6 +223,11 @@ async fn proving_loop(state: AppState) { } /// Fetch a proving estimate from the prover and store it on the commitment. +/// +/// Currently unused: see the call site in the proving loop for why estimates +/// are off. Kept so re-enabling is a one-line change once calibration measures +/// the phase real jobs actually run. +#[allow(dead_code)] async fn fetch_and_store_estimate( state: &AppState, prover_endpoint: &str, diff --git a/subs/src/routes/commits.rs b/subs/src/routes/commits.rs index 32393e9..b57ae8e 100644 --- a/subs/src/routes/commits.rs +++ b/subs/src/routes/commits.rs @@ -192,12 +192,46 @@ pub async fn get_commit_status( Ok(Json(response)) } -/// Maximum handles to publish per request to avoid oversized relay messages. -/// Certificates per publish batch. All of them go into a single message, and -/// the relay rejects anything over its 512 KB max_message_size outright, so -/// this stays well under the point where proof-carrying certs could add up -/// to a flat 413. -pub(crate) const PUBLISH_BATCH_SIZE: usize = 50; +/// Read live progress for one job, best-effort. +async fn fetch_job_progress( + state: &AppState, + prover_endpoint: &str, + job_id: &str, +) -> Option { + #[derive(serde::Deserialize)] + struct Resp { + #[serde(default)] + progress: Option, + } + + let client = reqwest::Client::builder() + // Short: this runs inside a UI poll, so a slow prover should degrade + // to "no progress bar", not hold up the whole pipeline response. + .timeout(std::time::Duration::from_secs(3)) + .build() + .ok()?; + + let url = format!("{}/jobs/{}", prover_endpoint.trim_end_matches('/'), job_id); + let mut req = client.get(&url); + if let Some(t) = state.config.prover_auth_token().ok().flatten() { + req = req.bearer_auth(t); + } + + req.send().await.ok()?.json::().await.ok()?.progress +} + +/// Sanity cap on certificates per publish batch. +/// +/// Not the target: publish_certs sizes each batch from the last message's +/// measured bytes, because a relay rejects anything over its 512 KB +/// max_message_size outright and the real constraint is bytes rather than +/// count. This only bounds the pathological case where a message is somehow +/// tiny — it must stay well *above* the byte-derived size, or it silently +/// caps growth and the measurement does nothing. +/// +/// Measured against a 20k-handle backlog: batches converge to ~500 +/// certificates for ~445 KB, so bytes bind well below this. +pub(crate) const PUBLISH_BATCH_SIZE: usize = 1000; #[derive(Serialize)] pub struct PublishResponse { @@ -343,6 +377,15 @@ pub struct PipelineResponse { /// 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, + /// Prover-side id of the in-flight job, so it can be correlated with the + /// prover's own logs and with the runpod proxy. + #[serde(skip_serializing_if = "Option::is_none")] + pub proving_job_id: Option, + /// Live progress of the in-flight proof, fetched from the prover. + /// Best-effort: absent if the prover is unreachable or predates progress + /// reporting, which must not fail the pipeline view. + #[serde(skip_serializing_if = "Option::is_none")] + pub proving_progress: Option, } pub async fn get_pipeline_status( @@ -376,6 +419,7 @@ pub async fn get_pipeline_status( // is a single step proof and later ones are step + fold. let mut proof_index = None; let mut proof_total = None; + let mut active_job_id: Option = 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 { @@ -387,7 +431,8 @@ pub async fn get_pipeline_status( 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() + active_job_id = state.config.get(&job_key).unwrap_or(None); + active_job_id.is_some() } else { false } @@ -395,11 +440,22 @@ pub async fn get_pipeline_status( false }; + // Ask the prover how far along it is. Only the prover knows, and the value + // is stale the moment it is cached, so it is fetched per request rather + // than stored. Failures are swallowed: a missing progress bar is a much + // better outcome than a broken pipeline view. + let proving_progress = match (active_job_id.as_deref(), state.config.prover_endpoint().ok().flatten()) { + (Some(job_id), Some(endpoint)) => fetch_job_progress(&state, &endpoint, job_id).await, + _ => None, + }; + Ok(Json(PipelineResponse { status, prover_configured, proving_job_active, proof_index, proof_total, + proving_job_id: active_job_id, + proving_progress, })) } diff --git a/subs/src/routes/mod.rs b/subs/src/routes/mod.rs index efe548c..e9e3cfb 100644 --- a/subs/src/routes/mod.rs +++ b/subs/src/routes/mod.rs @@ -60,6 +60,7 @@ pub fn router() -> Router { .route("/spaces/:space/proving/fulfill", post(proving::fulfill)) .route("/spaces/:space/proving/push", post(proving::push_to_prover)) .route("/spaces/:space/proving/poll", post(proving::poll_prover)) + .route("/spaces/:space/proving/cancel", post(proving::cancel_proving)) .route("/spaces/:space/proving/estimate", get(proving::get_estimate)) .route("/spaces/:space/compress", get(proving::get_compress_input)) .route("/spaces/:space/snark", post(proving::save_snark)) diff --git a/subs/src/routes/proving.rs b/subs/src/routes/proving.rs index 03e1fc7..4cdfc65 100644 --- a/subs/src/routes/proving.rs +++ b/subs/src/routes/proving.rs @@ -15,6 +15,44 @@ use serde::{Deserialize, Serialize}; use subs_core::CompressInput; use crate::state::AppState; + +fn one() -> u8 { + 1 +} + +/// Live proving progress, forwarded verbatim from the prover. +/// +/// The prover is the only place that knows how far along a proof is; subs just +/// relays it so the UI can show a bar instead of a spinner. Fields are optional +/// so a prover that predates progress reporting still deserializes. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobProgress { + pub total_cycles: u64, + pub proving_cycles_done: u64, + pub segments: usize, + pub segments_done: usize, + pub elapsed_seconds: f64, + pub estimated_total_seconds: Option, + pub first_segment_seconds: Option, + /// Which phase the prover is in, and how many there are. Defaulted rather + /// than optional so a prover that predates phase reporting still + /// deserializes and simply reads as "phase 1 of 1" — one determinate bar, + /// which is exactly how it used to behave. + #[serde(default = "one")] + pub phase: u8, + #[serde(default = "one")] + pub phase_total: u8, + #[serde(default)] + pub phase_one_fraction: Option, + /// Anything else the prover reported. + /// + /// A custom prover knows things this one cannot — which GPU it rented, what + /// the pod costs, where it is queued. Without this those fields would be + /// dropped on deserialization; flattening keeps them so the UI can display + /// them generically, without subs needing to know what they mean. + #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")] + pub extra: serde_json::Map, +} use super::json_error; /// Request type for fulfill payload @@ -288,6 +326,90 @@ pub struct PollResponse { pub message: Option, } +/// POST /spaces/:space/proving/cancel - Stop the in-flight proving job. +/// +/// Clears the local job key regardless of what the prover says, so the UI stops +/// waiting on a job it has abandoned. A queued job never runs; a running one +/// finishes on the prover and has its receipt discarded — see the prover's +/// cancel_job for why it cannot be interrupted mid-proof. +pub async fn cancel_proving( + State(state): State, + Path(space): Path, +) -> Result, Response> { + let space_label: spaces_protocol::slabel::SLabel = space + .parse() + .map_err(|e| json_error(StatusCode::BAD_REQUEST, format!("invalid space: {}", e)))?; + + let prover_endpoint = state + .config + .prover_endpoint() + .map_err(|e| json_error(StatusCode::INTERNAL_SERVER_ERROR, e))? + .ok_or_else(|| json_error(StatusCode::BAD_REQUEST, "prover_endpoint not configured"))?; + + let request = state + .operator + .get_next_proving_request(&space_label) + .await + .map_err(|e| json_error(StatusCode::INTERNAL_SERVER_ERROR, e))? + .ok_or_else(|| json_error(StatusCode::BAD_REQUEST, "no proving request in flight"))?; + + let commitment_id = request.commitment_id(); + let is_fold = matches!(&request, subs_core::ProvingRequest::Fold { .. }); + let job_key = format!( + "job:{}:{}:{}", + space, + commitment_id, + if is_fold { "fold" } else { "step" } + ); + + let job_id = state + .config + .get(&job_key) + .map_err(|e| json_error(StatusCode::INTERNAL_SERVER_ERROR, e))? + .ok_or_else(|| json_error(StatusCode::BAD_REQUEST, "no job in flight for this commitment"))?; + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .unwrap(); + + let url = format!( + "{}/jobs/{}/cancel", + prover_endpoint.trim_end_matches('/'), + job_id + ); + let mut req = client.post(&url); + if let Some(t) = state.config.prover_auth_token().ok().flatten() { + req = req.bearer_auth(t); + } + + let prover_said = match req.send().await { + Ok(r) => { + let status = r.status(); + let body = r.text().await.unwrap_or_default(); + if status.is_success() { + serde_json::from_str::(&body).unwrap_or(serde_json::json!({})) + } else { + // A job the prover has already lost or finished is still worth + // clearing locally, so this is reported rather than fatal. + serde_json::json!({ "prover_status": status.as_u16(), "prover_body": body }) + } + } + Err(e) => serde_json::json!({ "prover_error": e.to_string() }), + }; + + // Drop the key either way: whatever the prover does with the work, subs is + // no longer waiting on it, and leaving the key would keep the UI showing a + // job that will never be collected. + let _ = state.config.delete(&job_key); + + Ok(Json(serde_json::json!({ + "success": true, + "job_id": job_id, + "prover": prover_said, + }))) +} + /// POST /spaces/:space/proving/poll - Poll prover for job completion and save receipt pub async fn poll_prover( State(state): State, @@ -367,6 +489,11 @@ pub async fn poll_prover( )); } + /// Only what this poll acts on. Progress is deliberately absent: the + /// pipeline view fetches it separately, and parsing it here would make a + /// custom prover's differently-shaped `progress` fail the whole status + /// response — turning a healthy poll into a gateway error over a field + /// nothing reads. #[derive(Deserialize)] struct JobStatusResponse { status: String, @@ -413,6 +540,11 @@ pub async fn poll_prover( // Clean up job key let _ = state.config.delete(&job_key); + // The stored estimate described the proof that just finished; the + // next one is a different shape. Cleared so it isn't read as a + // forecast for work it says nothing about. + let _ = state.operator.clear_estimate(&space_label, commitment_id).await; + Ok(Json(PollResponse { success: true, status: Some("complete".to_string()), @@ -500,3 +632,63 @@ pub async fn get_estimate( Ok((StatusCode::OK, Json(estimate)).into_response()) } + +#[cfg(test)] +mod tests { + use super::JobProgress; + + /// A custom prover's extra fields must survive deserialization. + /// + /// Without `#[serde(flatten)]` these are dropped silently — the struct + /// still parses, the UI just shows nothing — so this is the kind of + /// regression that would not surface until someone asked why their proxy's + /// fields never appeared. + #[test] + fn unknown_fields_are_preserved() { + let json = r#"{ + "total_cycles": 4081240, + "proving_cycles_done": 3145728, + "segments": 5, + "segments_done": 3, + "elapsed_seconds": 333.4, + "estimated_total_seconds": 572.1, + "first_segment_seconds": 94.0, + "gpu": "NVIDIA A100 80GB PCIe", + "hourly_rate": 1.19 + }"#; + + let p: JobProgress = serde_json::from_str(json).expect("deserialize"); + + assert_eq!(p.segments_done, 3); + assert_eq!(p.segments, 5); + assert_eq!( + p.extra.get("gpu").and_then(|v| v.as_str()), + Some("NVIDIA A100 80GB PCIe") + ); + assert_eq!(p.extra.get("hourly_rate").and_then(|v| v.as_f64()), Some(1.19)); + // Known fields must not leak into extra, or the UI renders them twice. + assert!(!p.extra.contains_key("segments")); + assert!(!p.extra.contains_key("elapsed_seconds")); + } + + /// A prover that reports nothing extra round-trips with an empty map, and + /// re-serializes without an `extra` key. + #[test] + fn plain_progress_round_trips() { + let json = r#"{ + "total_cycles": 100, + "proving_cycles_done": 50, + "segments": 2, + "segments_done": 1, + "elapsed_seconds": 1.5, + "estimated_total_seconds": null, + "first_segment_seconds": null + }"#; + + let p: JobProgress = serde_json::from_str(json).expect("deserialize"); + assert!(p.extra.is_empty()); + + let out = serde_json::to_string(&p).expect("serialize"); + assert!(!out.contains("extra"), "flattened map must not emit a key: {out}"); + } +} diff --git a/subs/src/routes/registry.rs b/subs/src/routes/registry.rs index 563a57a..9372716 100644 --- a/subs/src/routes/registry.rs +++ b/subs/src/routes/registry.rs @@ -33,6 +33,27 @@ struct PendingHandle { script_pubkey: String, } +/// One handle's fate, reported to the registry on ack. +/// +/// Outcomes are a protocol contract — a registry matches on them to decide +/// what to show a user. All of them are terminal: the handle is settled and +/// will not be offered again. Anything retryable is simply left unacked and +/// stays in /pending. +#[derive(Serialize)] +pub struct AckEntry { + pub handle: String, + pub outcome: &'static str, +} + +impl AckEntry { + fn new(handle: &str, outcome: &'static str) -> Self { + Self { + handle: handle.to_string(), + outcome, + } + } +} + /// Outcome of a single pull -> stage -> ack cycle. pub struct SyncOutcome { pub pulled: usize, @@ -56,7 +77,80 @@ pub async fn sync_once( .timeout(std::time::Duration::from_secs(10)) .build()?; - let mut req = client.get(format!("{}/pending", base)); + // Scope is what this wallet can act on, which is wider than what it is + // currently running: a space delegated on-chain but never started still + // counts, because staging one of its handles auto-adopts it via + // load_or_create_space. Leaving those out would be a closed loop — the + // handles would never arrive, so the space would never be created, so it + // would never enter scope, and the work would sit in the registry + // invisible to both sides. + let mut spaces: Vec = state + .operator + .list_spaces() + .iter() + .map(|s| s.to_string()) + .collect(); + + match state.operator.list_delegated_spaces().await { + Ok(delegated) => spaces.extend(delegated.iter().map(|s| s.to_string())), + Err(e) => { + // One RPC call; if the node is unreachable we still sync the + // spaces already running rather than stalling the whole cycle. + tracing::debug!("Could not list delegated spaces: {}", e); + } + } + spaces.sort(); + spaces.dedup(); + + if spaces.is_empty() { + return Ok(SyncOutcome { + pulled: 0, + staged: 0, + errors: vec![], + }); + } + + let mut pulled = 0usize; + let mut staged = 0usize; + let mut errors: Vec = Vec::new(); + + // One space per request, pulled and settled before moving on. Keeps the + // URL bounded however many spaces an operator runs, gives the registry a + // natural place to paginate, and means a space that cannot be staged + // cannot affect any other. + for space in &spaces { + match sync_space(state, &client, base, auth_token, space).await { + Ok((p, s, mut errs)) => { + pulled += p; + staged += s; + errors.append(&mut errs); + } + Err(e) => errors.push(format!("{}: {}", space, e)), + } + } + + Ok(SyncOutcome { + pulled, + staged, + errors, + }) +} + +/// Pull, stage and acknowledge one space's pending handles. +/// +/// Returns (pulled, staged, non-fatal errors). An Err is a failure of the +/// space as a whole — an unreachable registry, a rejected token — and leaves +/// everything for that space pending. +async fn sync_space( + state: &AppState, + client: &reqwest::Client, + base: &str, + auth_token: Option<&str>, + space: &str, +) -> anyhow::Result<(usize, usize, Vec)> { + let mut req = client + .get(format!("{}/pending", base)) + .query(&[("space", space)]); if let Some(t) = auth_token { req = req.bearer_auth(t); } @@ -78,108 +172,111 @@ pub async fn sync_once( .map_err(|e| anyhow::anyhow!("invalid response from registry: {}", e))?; if pending.handles.is_empty() { - return Ok(SyncOutcome { - pulled: 0, - staged: 0, - errors: vec![], - }); + return Ok((0, 0, vec![])); } - tracing::info!("Pulled {} pending handles from registry", pending.handles.len()); - - let mut errors = Vec::new(); - - // Group by space before staging. add_requests aborts the whole call if any - // one space can't be loaded — an unknown space, a wallet that can't operate - // it — so batching every space together lets one bad handle sink handles - // that would otherwise stage fine. - let mut by_space: std::collections::HashMap> = + let mut errors: Vec = Vec::new(); + let mut outcomes: Vec = Vec::new(); + let mut requests: Vec = Vec::new(); + let mut submitted: std::collections::HashMap = std::collections::HashMap::new(); for handle in &pending.handles { - let handle_name: spaces_protocol::sname::SName = match handle.handle.parse() { + let parsed: spaces_protocol::sname::SName = match handle.handle.parse() { Ok(h) => h, Err(e) => { errors.push(format!("{}: invalid handle: {}", handle.handle, e)); + // Terminal: it will never parse, so settle it rather than + // leave it to be re-pulled forever as a poison entry. + outcomes.push(AckEntry::new(&handle.handle, "invalid")); continue; } }; - let space = match handle_name.space() { - Some(s) => s.to_string(), - None => { - errors.push(format!("{}: handle has no space", handle.handle)); + // We asked for one space; anything else is a registry bug. Skipping + // rather than forwarding keeps add_requests to a single space, which + // is what stops one bad space aborting the whole call. + match parsed.space().map(|s| s.to_string()) { + Some(s) if s == space => {} + other => { + errors.push(format!( + "{}: registry returned a handle for {:?} when {} was requested", + handle.handle, other, space + )); continue; } - }; + } - by_space.entry(space).or_default().push(( - handle.handle.clone(), - subs_core::HandleRequest { - handle: handle_name, - script_pubkey: handle.script_pubkey.clone(), - dev_private_key: None, - }, - )); + submitted.insert(parsed.to_string(), handle.handle.clone()); + requests.push(subs_core::HandleRequest { + handle: parsed, + script_pubkey: handle.script_pubkey.clone(), + dev_private_key: None, + }); } - let mut staged = 0; - let mut to_ack: Vec = Vec::new(); - - for (space, entries) in by_space { - let (names, requests): (Vec, Vec) = - entries.into_iter().unzip(); + let original = |parsed: &spaces_protocol::sname::SName| -> String { + submitted + .get(&parsed.to_string()) + .cloned() + .unwrap_or_else(|| parsed.to_string()) + }; + let mut staged = 0usize; + if !requests.is_empty() { match state.operator.add_requests(requests).await { Ok(result) => { tracing::info!("[{}] Staged {} handles", space, result.total_added); + staged = result.total_added; + for space_result in &result.by_space { + for added in &space_result.added { + outcomes.push(AckEntry::new(&original(added), "staged")); + } + // Skips are settled: already staged, already committed, or + // taken under another script pubkey. The outcome is what + // tells the registry which it was. for skip in &space_result.skipped { - tracing::info!("Skipped: {} ({:?})", skip.handle, skip.reason); + tracing::info!("Skipped: {} ({})", skip.handle, skip.reason.as_str()); + outcomes.push(AckEntry::new( + &original(&skip.handle), + skip.reason.as_outcome(), + )); } } - staged += result.total_added; - // Skips are settled outcomes (already staged, already - // committed, taken by another spk), so they're acked too — - // leaving them pending would re-pull them forever. - to_ack.extend(names); } Err(e) => { - // Deliberately not acked. Staging never happened, so acking - // would mark them done at the registry and lose them; leaving - // them pending means a fixed operator config picks them up. - errors.push(format!("{}: failed to stage: {}", space, e)); + // Retryable: staging never happened, so these get no outcome + // and stay pending for a later cycle to collect once the + // operator's configuration is fixed. + errors.push(format!("failed to stage: {}", e)); } } } - // A failed ack leaves handles Pending at the registry, and the next cycle - // re-pulls and re-acks them (add_requests dedupes), so this self-heals. - if !to_ack.is_empty() { - if let Err(e) = ack(&client, base, auth_token, &to_ack).await { - // Not fatal: staging already succeeded locally. + // A failed ack leaves handles pending, and the next cycle re-pulls and + // re-acks them (add_requests dedupes), so this self-heals. + if !outcomes.is_empty() { + if let Err(e) = ack(client, base, auth_token, &outcomes).await { tracing::warn!("Failed to acknowledge handles to registry: {}", e); errors.push(format!("ack failed: {}", e)); } } - Ok(SyncOutcome { - pulled: pending.handles.len(), - staged, - errors, - }) + Ok((pending.handles.len(), staged, errors)) } + /// POST /ack, checking the response rather than firing and forgetting. async fn ack( client: &reqwest::Client, base: &str, auth_token: Option<&str>, - handles: &[String], + handles: &[AckEntry], ) -> anyhow::Result<()> { #[derive(Serialize)] struct AckRequest<'a> { - handles: &'a [String], + handles: &'a [AckEntry], } let mut req = client diff --git a/subs/src/routes/requests.rs b/subs/src/routes/requests.rs index 0789189..6f747ee 100644 --- a/subs/src/routes/requests.rs +++ b/subs/src/routes/requests.rs @@ -133,8 +133,14 @@ pub async fn bulk_generate( let secp = Secp256k1::new(); let mut requests = Vec::with_capacity(body.count); - for i in 0..body.count { - let handle_str = format!("{}{}{}", body.prefix, i, body.space); + for _ in 0..body.count { + // Random rather than sequential from 0. Sequential names collide with + // everything generated before them, and duplicates are skipped + // silently, so asking for 100 against a space holding 100k handles + // used to stage nothing at all -- you had to ask for 100_100 to reach + // fresh names. 64 bits of suffix makes a collision irrelevant here. + let suffix: u64 = rand::random(); + let handle_str = format!("{}{:016x}{}", body.prefix, suffix, body.space); let handle: spaces_protocol::sname::SName = handle_str .parse() .map_err(|e| json_error(StatusCode::BAD_REQUEST, format!("invalid handle {}: {}", handle_str, e)))?; @@ -156,12 +162,16 @@ pub async fn bulk_generate( }); } - let count = requests.len(); - state + // Report what was actually added, not what was generated. These differ + // whenever a name collides, and reporting the generated count made a + // no-op bulk generate look like a success. + let result = state .operator .add_requests(requests) .await .map_err(|e| json_error(StatusCode::INTERNAL_SERVER_ERROR, e))?; - Ok(Json(BulkGenerateResponse { staged: count })) + Ok(Json(BulkGenerateResponse { + staged: result.total_added, + })) } diff --git a/subs/templates/base.html b/subs/templates/base.html index 6836455..6c0a53d 100644 --- a/subs/templates/base.html +++ b/subs/templates/base.html @@ -756,6 +756,73 @@ .animate-pulse { animation: pulse 2s ease-in-out infinite; } @keyframes shimmer { 0%, 100% { opacity: 0.7; } 50% { opacity: 1; } } .shimmer { animation: shimmer 2s ease-in-out infinite; } +/* === PROVING PROGRESS === */ +.prove-panel { + margin-top: 12px; padding: 14px; + background: var(--bg-base); border: 1px solid var(--border-subtle); + border-radius: var(--radius); +} +.prove-head { + display: flex; align-items: baseline; justify-content: space-between; + gap: 12px; margin-bottom: 10px; +} +.prove-phase { font-size: 13px; color: var(--text-primary); font-weight: 500; } +.prove-phase-num { + font-family: var(--mono); font-size: 10px; color: var(--text-muted); + text-transform: uppercase; letter-spacing: 0.04em; +} +/* Taller than a hairline: this is the primary indicator on the panel, and at + 6px it read as a divider rather than a progress bar. */ +.prove-track { + height: 10px; background: var(--bg-raised); + border-radius: 5px; overflow: hidden; position: relative; +} +.prove-fill { + height: 100%; background: var(--accent); border-radius: 5px; + transition: width 1s linear; +} +/* Indeterminate: work is happening but its duration is unknowable. Used for + proving phase 2, where risc0 exposes no hook to measure against. Travels the + full track rather than parking at one end, so it reads as "still working". */ +.prove-fill-indeterminate { + position: absolute; top: 0; left: 0; height: 100%; width: 42%; + border-radius: 5px; + background: linear-gradient(90deg, + transparent, var(--accent) 18%, var(--accent) 82%, transparent); + animation: indeterminate 1.8s ease-in-out infinite; +} +/* Travel is kept just short of fully clearing the track at each end, so the + stripe is visible at essentially every frame rather than spending part of + each cycle off-screen looking like a stalled bar. */ +@keyframes indeterminate { + 0% { transform: translateX(-75%); } + 100% { transform: translateX(240%); } +} +/* Auto-fit so a custom prover's extra fields flow into the same grid without + the layout needing to know how many there will be. */ +.prove-stats { + margin-top: 12px; display: grid; gap: 10px; + grid-template-columns: repeat(auto-fit, minmax(118px, 1fr)); +} +.prove-stat { + background: var(--bg-surface); border: 1px solid var(--border-subtle); + border-radius: 7px; padding: 8px 10px; + display: flex; flex-direction: column; gap: 3px; min-width: 0; +} +.prove-stat-label { + font-family: var(--mono); font-size: 9px; color: var(--text-muted); + text-transform: uppercase; letter-spacing: 0.05em; + white-space: nowrap; overflow: hidden; text-overflow: ellipsis; +} +.prove-stat-value { + font-family: var(--mono); font-size: 14px; color: var(--text-primary); + line-height: 1.25; overflow-wrap: anywhere; +} +/* The one number people are actually waiting on. */ +.prove-stat-accent .prove-stat-value { color: var(--accent); } +.prove-stat-note { + font-size: 10px; color: var(--text-muted); line-height: 1.3; +} /* === SCROLLBAR === */ .scrollbar-thin::-webkit-scrollbar { width: 4px; } diff --git a/subs/templates/space.html b/subs/templates/space.html index d5c2a8a..c275a66 100644 --- a/subs/templates/space.html +++ b/subs/templates/space.html @@ -229,6 +229,10 @@

Handles

// 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; +// Last rendered step, so a transition always gets one more render. Without it +// the poll disarms on entering a state it does not watch — e.g. proving -> +// broadcast — leaving the finished proof's UI frozen on screen. +let lastPipelineStep; let currentFilter = null; let searchTimeout = null; let selectedHandles = new Set(); @@ -240,10 +244,16 @@

Handles

// --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- +// Escapes for text *and* attribute contexts. The previous version round-tripped +// through div.textContent/innerHTML, which escapes only & < > — leaving quotes +// intact, so anything interpolated into title="…" or onclick="…('…')" could +// break out of the attribute. Values reaching here include field names chosen +// by whichever prover is configured. Entities still render as the literal +// characters in text contexts, so this is safe everywhere. function esc(t) { - const d = document.createElement('div'); - d.textContent = t; - return d.innerHTML; + return String(t == null ? '' : t).replace(/[&<>"']/g, c => ({ + '&': '&', '<': '<', '>': '>', '"': '"', "'": ''', + })[c]); } function $(id) { return document.getElementById(id); } @@ -309,11 +319,18 @@

Handles

clearTimeout(pipelinePollTimer); pipelinePollTimer = null; } + const stepChanged = lastPipelineStep !== undefined && lastPipelineStep !== j.current_step; + lastPipelineStep = j.current_step; + if ((j.steps && j.steps.proving === 'in_progress' && j.prover_configured) + || j.proving_job_active || j.current_step === 'confirmed' || j.current_step === 'finalized' || j.current_step === 'published' - || (j.unpublished || 0) > 0) { + || (j.unpublished || 0) > 0 + // One more pass after any transition, so the state we just left is + // never what stays on screen. + || stepChanged) { pipelinePollTimer = setTimeout(refreshPipeline, 5000); } @@ -408,6 +425,163 @@

Handles

$('pipelineStepper').innerHTML = h; } +// Cycle counts run to millions, where exact digits are noise. +function fmtCycles(n) { + if (n == null) return null; + if (n >= 1e9) return `${(n / 1e9).toFixed(1)}B`; + if (n >= 1e6) return `${(n / 1e6).toFixed(1)}M`; + if (n >= 1e3) return `${(n / 1e3).toFixed(1)}K`; + return `${n}`; +} + +// Compact form for stat tiles, where "1 minute 32 seconds" wraps to two lines +// and buries the number. Prose elsewhere still uses fmtDuration. +function fmtDurationShort(sec) { + if (sec == null) return null; + const total = Math.max(0, Math.round(sec)); + if (total < 60) return `${total}s`; + const mins = Math.floor(total / 60); + if (mins < 60) { + const secs = total % 60; + return secs === 0 ? `${mins}m` : `${mins}m ${secs}s`; + } + const hours = Math.floor(mins / 60); + const remMins = mins % 60; + return remMins === 0 ? `${hours}h` : `${hours}h ${remMins}m`; +} + +function fmtDuration(sec) { + if (sec == null) return null; + const total = Math.max(0, Math.round(sec)); + const unit = (n, word) => `${n} ${word}${n === 1 ? '' : 's'}`; + + if (total < 60) return unit(total, 'second'); + + const mins = Math.floor(total / 60); + const secs = total % 60; + if (mins < 60) { + // Seconds stop mattering once it is a long wait. + return mins >= 10 || secs === 0 + ? unit(mins, 'minute') + : `${unit(mins, 'minute')} ${unit(secs, 'second')}`; + } + + const hours = Math.floor(mins / 60); + const remMins = mins % 60; + return remMins === 0 + ? unit(hours, 'hour') + : `${unit(hours, 'hour')} ${unit(remMins, 'minute')}`; +} + +function renderProvingProgress(p, jobId) { + // Shown even with no progress yet: the id is what correlates this proof + // with the prover's logs and with the runpod proxy. + const idRow = jobId + ? `
+ job + ${esc(jobId)} + +
` + : ''; + + // Absent for a prover that predates progress reporting, or while the + // executor is still running before the first segment is proven. + if (!p || !p.segments) return idRow; + + // `title` carries any explanation, so the tiles stay uniform instead of + // growing a line of prose underneath. + const stat = (label, value, opts = {}) => ` +
+ ${esc(label)} + ${esc(value)} +
`; + + const elapsed = fmtDurationShort(p.elapsed_seconds); + const phaseTotal = p.phase_total || 1; + // Phase 2 is lift/join/resolve. risc0 fires no hook during it, so there is + // nothing to measure and nothing to extrapolate — it gets a moving bar + // with no percentage rather than a full one that sits there. On a measured + // single-segment proof this phase was 28.1s of 38.8s, so a bar pinned at + // 100% for its duration was the most misleading thing on the page. + const inPhaseTwo = phaseTotal > 1 && p.phase >= 2; + // An indeterminate bar is also right for the first segment: nothing has + // been timed yet, so there is no honest position to draw. + const indeterminate = inPhaseTwo || p.phase_one_fraction == null; + // Just the counter — the phase's description is the panel heading. + const phaseLabel = phaseTotal > 1 ? `Phase ${p.phase} of ${phaseTotal}` : null; + + // No ETA until a segment lands — there is nothing to extrapolate from. + // Absent for all of phase 2, by design. Its absence is not annotated: it is + // the normal case here, and saying so reads as a fault. + const remaining = p.estimated_total_seconds != null + ? fmtDurationShort(Math.max(0, p.estimated_total_seconds - p.elapsed_seconds)) + : null; + + // Driven by the prover's interpolated fraction, which advances within the + // segment being proven. A bar keyed on segments_done alone would sit still + // for the minute-plus each segment takes, then jump. + const bar = indeterminate + ? `
` + : `
`; + + let h = `
+
+ ${esc(inPhaseTwo ? 'Producing succinct receipt' : 'Proving segments')} + ${phaseLabel ? `${esc(phaseLabel)}` : ''} +
+
${bar}
+
`; + + // Ordered by what someone watching a proof actually wants: how much longer, + // then how long so far, then the work being done. + if (remaining) h += stat('remaining', `~${remaining}`, { accent: true }); + h += stat('elapsed', elapsed); + if (!inPhaseTwo) h += stat('segments', `${p.segments_done}/${p.segments}`); + // Collected by the prover and forwarded all along, but previously listed in + // KNOWN (so skipped by the extras block) without being rendered anywhere — + // so cycle counts never reached the page at all. + if (p.total_cycles) h += stat('cycles', fmtCycles(p.total_cycles)); + // Gated on `segments_done > 1` before, which a single-segment job never + // reaches — so on the proofs this actually produces it never rendered. It + // is the only number separating warm-up (PTX JIT) from steady-state rate. + if (p.first_segment_seconds != null && p.segments_done >= 1) { + h += stat('first segment', fmtDurationShort(p.first_segment_seconds), { + title: 'The first segment includes GPU warm-up and any PTX JIT, so it runs slower than the ones after it.', + }); + } + + // Whatever else the prover chose to report. A custom prover — the runpod + // proxy, say — knows things this UI cannot anticipate: the GPU it rented, + // the pod's hourly rate, queue position. They flow into the same grid as + // the built-in stats, so a new field looks native without a subs change. + const KNOWN = new Set([ + 'total_cycles', 'proving_cycles_done', 'segments', 'segments_done', + 'elapsed_seconds', 'estimated_total_seconds', 'first_segment_seconds', + 'phase', 'phase_total', 'phase_one_fraction', + ]); + for (const [k, v] of Object.entries(p)) { + if (KNOWN.has(k) || v == null || typeof v === 'object') continue; + const label = k.replace(/_/g, ' '); + // Labels are ellipsized to keep tiles uniform, so the full name has to + // stay reachable — a custom prover can name a field anything. + h += stat(label, typeof v === 'number' ? v.toLocaleString() : String(v), { title: label }); + } + + h += '
'; + return h + '
' + idRow; +} + +async function cancelProving() { + if (!confirm('Cancel this proof?\n\nA queued job stops immediately. One already running finishes on the prover and its result is discarded — the GPU time is not reclaimed.')) return; + const { ok, data } = await api(`${spaceUrl}/proving/cancel`, { method: 'POST' }); + if (ok) { + logAction(`Proving cancelled: ${data.job_id}`); + refreshPipeline(); + } else { + logAction(`Cancel failed: ${data.error || 'unknown'}`); + } +} + function renderActions(r) { const el = $('pipelineActions'); let h = ''; @@ -427,24 +601,23 @@

Handles

Generating proof${proofLabel}... (${total} pending)`; + h += renderProvingProgress(r.proving_progress, r.proving_job_id); } - if (r.estimate) { - h += '
'; - for (const [k, v] of Object.entries(r.estimate)) { - if (v == null || typeof v === 'object') continue; - const label = k.replace(/_/g, ' ').replace(/\b\w/g, c => c.toUpperCase()); - const val = typeof v === 'number' ? v.toLocaleString() : String(v); - h += `
- ${esc(label)}: ${esc(val)}
`; - } - h += '
'; - } + // The pre-prove estimate is not rendered. It is persisted on the + // commitment when fetched, so it outlives the prover that produced + // it and kept showing after calibration was turned off. Its timings + // are also wrong: calibration measures ProverOpts::composite(), + // which skips lift/join, while real jobs run succinct() — so it + // read ~43% low against a measured proof. The live progress in + // renderProvingProgress supersedes it. if (!r.proving_job_active) { h += `
`; } else { - h += ''; + h += `
+ +
`; } } else { h += `