diff --git a/circuits/proof-builder/src/lib.rs b/circuits/proof-builder/src/lib.rs index 94656ee5e..e11421456 100644 --- a/circuits/proof-builder/src/lib.rs +++ b/circuits/proof-builder/src/lib.rs @@ -324,8 +324,24 @@ impl ProofData { } } +/// Where an Operator or Watchtower proof task stands, as a submission reports it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProofTaskState { + /// The task is queued or being proven; submit again later. + Pending, + /// The proof is in `proof_data`. + Proven, + /// The task failed for good: its inputs can never be proven, or it failed again + /// after the restarts a resubmission allows. `error` says why. + Failed, + /// The task is no longer wanted (its request timed out). + Cancelled, +} + #[derive(Debug, Serialize, Deserialize)] pub struct OperatorProofResponse { + pub state: ProofTaskState, pub proof_data: Option, pub error: Option, } @@ -343,6 +359,7 @@ pub struct WatchtowerProofRequest { #[derive(Debug, Serialize, Deserialize)] pub struct WatchtowerProofResponse { + pub state: ProofTaskState, pub proof_data: Option, pub error: Option, } diff --git a/crates/store/migrations/20261111000000_on_demand_proof_reset.sql b/crates/store/migrations/20261111000000_on_demand_proof_reset.sql new file mode 100644 index 000000000..043710c5d --- /dev/null +++ b/crates/store/migrations/20261111000000_on_demand_proof_reset.sql @@ -0,0 +1,98 @@ +-- Operator and Watchtower tasks gain: +-- * state 4 (Cancelled): the task is no longer wanted (its request timed out, or +-- the Watchtower never asked for a proof before the Operator task fixed the list); +-- * reset_count: how often the task owner restarted the failed task; +-- * invalid_reason: set when the task inputs can never be proven, which a restart +-- with the same inputs cannot repair; +-- * input_hash (Operator only): the request inputs, to tell a resubmission with new +-- inputs from a plain retry. +-- SQLite cannot change a CHECK constraint in place, so both tables are rebuilt. + +CREATE TABLE operator_proof_new +( + `id` INTEGER PRIMARY KEY, + `instance_id` TEXT NOT NULL, + `graph_id` TEXT NOT NULL, + `execution_layer_block_number` BIGINT NOT NULL DEFAULT 0, + `path_to_proof` TEXT, + `proof_size` BIGINT NOT NULL DEFAULT 0, + `public_value_hex` TEXT, + `cycles` BIGINT NOT NULL DEFAULT 0, + `proof_state` BIGINT NOT NULL DEFAULT 0 CHECK (proof_state IN (0, 1, 2, 3, 4)), + `total_time_to_proof` BIGINT NOT NULL DEFAULT 0, + `proving_time` BIGINT NOT NULL DEFAULT 0, + `zkm_version` TEXT NOT NULL DEFAULT '', + `operator_committed_blockhash` TEXT NOT NULL, + `reset_count` BIGINT NOT NULL DEFAULT 0, + `invalid_reason` TEXT, + `input_hash` TEXT, + `extra` TEXT, + `created_at` BIGINT NOT NULL DEFAULT 0, + `updated_at` BIGINT NOT NULL DEFAULT 0 +); + +-- One task per graph: a proven task wins, then an unfinished one, then the oldest. +INSERT INTO operator_proof_new + (id, instance_id, graph_id, execution_layer_block_number, path_to_proof, proof_size, + public_value_hex, cycles, proof_state, total_time_to_proof, proving_time, zkm_version, + operator_committed_blockhash, extra, created_at, updated_at) +SELECT id, instance_id, graph_id, execution_layer_block_number, path_to_proof, proof_size, + public_value_hex, cycles, proof_state, total_time_to_proof, proving_time, zkm_version, + operator_committed_blockhash, extra, created_at, updated_at +FROM ( + SELECT *, row_number() OVER ( + PARTITION BY instance_id, graph_id + ORDER BY CASE proof_state WHEN 2 THEN 0 WHEN 3 THEN 2 ELSE 1 END, id + ) AS graph_rank + FROM operator_proof +) +WHERE graph_rank = 1; + +DROP TABLE operator_proof; +ALTER TABLE operator_proof_new RENAME TO operator_proof; + +CREATE UNIQUE INDEX idx_operator_proof_graph + ON operator_proof (instance_id, graph_id); + +CREATE TABLE watchtower_proof_new +( + `id` INTEGER PRIMARY KEY AUTOINCREMENT, + `instance_id` TEXT NOT NULL, + `graph_id` TEXT NOT NULL, + `public_key` TEXT NOT NULL, + `challenge_txid` TEXT NOT NULL, + `challenge_init_txid` TEXT NOT NULL, + `execution_layer_block_number` BIGINT NOT NULL DEFAULT 0, + `path_to_proof` TEXT, + `proof_size` BIGINT NOT NULL DEFAULT 0, + `public_value_hex` TEXT, + `cycles` BIGINT NOT NULL DEFAULT 0, + `proof_state` BIGINT NOT NULL DEFAULT 0 CHECK (proof_state IN (0, 1, 2, 3, 4)), + `total_time_to_proof` BIGINT NOT NULL DEFAULT 0, + `proving_time` BIGINT NOT NULL DEFAULT 0, + `zkm_version` TEXT NOT NULL DEFAULT '', + `node_index` INTEGER NOT NULL DEFAULT 0, + `included` BOOLEAN NOT NULL DEFAULT 0, + `reset_count` BIGINT NOT NULL DEFAULT 0, + `invalid_reason` TEXT, + `extra` TEXT, + `created_at` BIGINT NOT NULL DEFAULT 0, + `updated_at` BIGINT NOT NULL DEFAULT 0 +); + +INSERT INTO watchtower_proof_new + (id, instance_id, graph_id, public_key, challenge_txid, challenge_init_txid, + execution_layer_block_number, path_to_proof, proof_size, public_value_hex, cycles, + proof_state, total_time_to_proof, proving_time, zkm_version, node_index, included, + extra, created_at, updated_at) +SELECT id, instance_id, graph_id, public_key, challenge_txid, challenge_init_txid, + execution_layer_block_number, path_to_proof, proof_size, public_value_hex, cycles, + proof_state, total_time_to_proof, proving_time, zkm_version, node_index, included, + extra, created_at, updated_at +FROM watchtower_proof; + +DROP TABLE watchtower_proof; +ALTER TABLE watchtower_proof_new RENAME TO watchtower_proof; + +CREATE UNIQUE INDEX idx_watchtower_proof_key + ON watchtower_proof (instance_id, graph_id, public_key); diff --git a/crates/store/src/localdb/proof.rs b/crates/store/src/localdb/proof.rs index 18e9f81c7..d5555313b 100644 --- a/crates/store/src/localdb/proof.rs +++ b/crates/store/src/localdb/proof.rs @@ -663,6 +663,111 @@ impl StorageProcessor<'_> { Ok(res.rows_affected()) } + /// Fails an on-demand task whose inputs can never be proven, only while it is still + /// `New`, and records why. + pub async fn fail_invalid_on_demand_proof( + &mut self, + table: OnDemandProofTable, + id: i64, + reason: &str, + ) -> anyhow::Result { + let sql = format!( + "UPDATE {} SET proof_state = ?, invalid_reason = ?, updated_at = ? + WHERE id = ? AND proof_state = ?", + table.table_name() + ); + let res = sqlx::query(&sql) + .bind(crate::ProofState::Failed.to_i64()) + .bind(reason) + .bind(get_current_timestamp_secs()) + .bind(id) + .bind(crate::ProofState::New.to_i64()) + .execute(self.conn()) + .await?; + Ok(res.rows_affected()) + } + + /// Puts a failed on-demand task back to `New` and clears its result. + /// + /// `max_resets` is a restart with the same inputs: it counts, and it is refused once + /// `max_resets` restarts were made or when the inputs are invalid. `None` follows new + /// inputs, which start the count over. + pub async fn reset_failed_on_demand_proof( + &mut self, + table: OnDemandProofTable, + id: i64, + max_resets: Option, + ) -> anyhow::Result { + let (reset_count, guard) = match max_resets { + Some(_) => ("reset_count + 1", "AND invalid_reason IS NULL AND reset_count < ?"), + None => ("0", ""), + }; + let sql = format!( + "UPDATE {} + SET proof_state = ?, reset_count = {reset_count}, invalid_reason = NULL, + path_to_proof = NULL, public_value_hex = NULL, proof_size = 0, cycles = 0, + total_time_to_proof = 0, proving_time = 0, zkm_version = '', updated_at = ? + WHERE id = ? AND proof_state = ? {guard}", + table.table_name() + ); + let mut query = sqlx::query(&sql) + .bind(crate::ProofState::New.to_i64()) + .bind(get_current_timestamp_secs()) + .bind(id) + .bind(crate::ProofState::Failed.to_i64()); + if let Some(max_resets) = max_resets { + query = query.bind(max_resets); + } + Ok(query.execute(self.conn()).await?.rows_affected()) + } + + /// Sets the inputs of a graph's Operator task. + pub async fn update_operator_proof_inputs( + &mut self, + instance_id: &Uuid, + graph_id: &Uuid, + operator_committed_blockhash: &str, + execution_layer_block_number: i64, + input_hash: &str, + ) -> anyhow::Result { + let res = sqlx::query( + "UPDATE operator_proof + SET operator_committed_blockhash = ?, execution_layer_block_number = ?, + input_hash = ?, updated_at = ? + WHERE instance_id = ? AND graph_id = ?", + ) + .bind(operator_committed_blockhash) + .bind(execution_layer_block_number) + .bind(input_hash) + .bind(get_current_timestamp_secs()) + .bind(instance_id) + .bind(graph_id) + .execute(self.conn()) + .await?; + Ok(res.rows_affected()) + } + + /// Sets the inputs of a Watchtower task. + pub async fn update_watchtower_proof_inputs( + &mut self, + id: i64, + challenge_init_txid: &crate::SerializableTxid, + execution_layer_block_number: i64, + ) -> anyhow::Result { + let res = sqlx::query( + "UPDATE watchtower_proof + SET challenge_init_txid = ?, execution_layer_block_number = ?, updated_at = ? + WHERE id = ?", + ) + .bind(challenge_init_txid) + .bind(execution_layer_block_number) + .bind(get_current_timestamp_secs()) + .bind(id) + .execute(self.conn()) + .await?; + Ok(res.rows_affected()) + } + /// Returns the current proof state of an on-demand proof task. pub async fn get_on_demand_proof_state( &mut self, diff --git a/crates/store/src/schema.rs b/crates/store/src/schema.rs index b5a9f87d5..b12ca0c23 100644 --- a/crates/store/src/schema.rs +++ b/crates/store/src/schema.rs @@ -753,6 +753,9 @@ pub enum ProofState { Proving, Proven, Failed, + /// No longer wanted: the request timed out, or (for a Watchtower task) the + /// Operator task fixed the graph's list without a proof from this Watchtower. + Cancelled, } impl ProofState { pub fn to_i64(&self) -> i64 { @@ -761,6 +764,7 @@ impl ProofState { Self::Proving => 1, Self::Proven => 2, Self::Failed => 3, + Self::Cancelled => 4, } } @@ -770,6 +774,7 @@ impl ProofState { 1 => Some(Self::Proving), 2 => Some(Self::Proven), 3 => Some(Self::Failed), + 4 => Some(Self::Cancelled), _ => None, } } @@ -823,6 +828,10 @@ pub struct WatchtowerProof { pub zkm_version: String, pub node_index: i32, pub included: bool, + /// How often the task owner restarted the failed task. + pub reset_count: i64, + /// Why the task inputs can never be proven; `None` when the proof itself failed. + pub invalid_reason: Option, pub extra: Option, pub created_at: i64, pub updated_at: i64, @@ -843,6 +852,12 @@ pub struct OperatorProof { pub total_time_to_proof: i64, pub proving_time: i64, pub zkm_version: String, + /// How often the task owner restarted the failed task. + pub reset_count: i64, + /// Why the task inputs can never be proven; `None` when the proof itself failed. + pub invalid_reason: Option, + /// Hash of the request inputs, to tell a resubmission with new inputs from a retry. + pub input_hash: Option, pub extra: Option, pub created_at: i64, pub updated_at: i64, diff --git a/node/src/aggregates/graph_pegout_jobs.rs b/node/src/aggregates/graph_pegout_jobs.rs index 5bfef22de..3777e94c1 100644 --- a/node/src/aggregates/graph_pegout_jobs.rs +++ b/node/src/aggregates/graph_pegout_jobs.rs @@ -50,7 +50,7 @@ use goat::wots::{Wots, Wots96}; use std::str::FromStr; use store::localdb::{GraphQuery, LocalDB}; use store::{Graph, GraphStatus, InstanceBridgeInStatus, SerializableTxid}; -use tracing::{info, warn}; +use tracing::{error, info, warn}; use uuid::Uuid; pub(super) async fn run_pegout_step( @@ -1285,12 +1285,31 @@ async fn prover_assert(deps: &JobDeps, mut graph: BitvmGcGraph) -> Result ) { return step; } - let (proof, wait_secs) = - get_operator_proof(&deps.local_db, &deps.http_client, &graph, btc, instance_id, graph_id) - .await?; - let Some(proof) = proof else { - // Not ready yet, never "never": the assert must follow once the proof exists. - return Step::retry("operator proof is not ready", (wait_secs as i64).max(1)); + let proof = match get_operator_proof( + &deps.local_db, + &deps.http_client, + &graph, + btc, + instance_id, + graph_id, + ) + .await? + { + ProofPoll::Ready(proof) => proof, + ProofPoll::NotReady(wait_secs) => { + // Not ready yet, never "never": the assert must follow once the proof exists. + return Step::retry("operator proof is not ready", (wait_secs as i64).max(1)); + } + ProofPoll::Failed(reason) => { + error!(event = "graph_challenge", graph_id = %graph_id, %reason, "the proof builder cannot produce the operator proof; no assert"); + alert("operator_proof_failed"); + return Step::ignore(format!("operator proof failed: {reason}")); + } + ProofPoll::Cancelled => { + error!(event = "graph_challenge", graph_id = %graph_id, "the operator proof task was cancelled; no assert"); + alert("operator_proof_failed"); + return Step::ignore("operator proof task cancelled"); + } }; let (master, _) = local_operator()?; let assert_secret = master.assert_wots_keypair_for_graph(graph_id).0; @@ -1647,13 +1666,20 @@ async fn watchtower_challenge(deps: &JobDeps, row: &Graph, graph: &BitvmGcGraph) ) .await? { - (Some(proof), _) => proof, - (None, wait_secs) => { + ProofPoll::Ready(proof) => proof, + ProofPoll::NotReady(wait_secs) => { return Step::retry( "watchtower proof commitment is not ready", (wait_secs as i64).max(1), ); } + ProofPoll::Failed(reason) => { + error!(event = "graph_challenge", graph_id = %row.graph_id, watchtower_index = index, %reason, "the proof builder cannot produce the watchtower proof; no challenge"); + alert("watchtower_proof_failed"); + return Step::ignore(format!("watchtower proof failed: {reason}")); + } + // This node's own timeout: the operator asserted, so the challenge is moot. + ProofPoll::Cancelled => return Step::ignore("watchtower proof task cancelled"), }; match send_watchtower_challenge_tx(btc, graph, index, proof).await { Ok(txid) => { diff --git a/node/src/bin/mock_proof_builder.rs b/node/src/bin/mock_proof_builder.rs index d85709081..c73aeae06 100644 --- a/node/src/bin/mock_proof_builder.rs +++ b/node/src/bin/mock_proof_builder.rs @@ -36,7 +36,7 @@ use axum::response::Json; use axum::routing::{get, post}; use clap::Parser; use proof_builder::{ - OperatorProofRequest, OperatorProofResponse, ProofData, ProofDescResponse, + OperatorProofRequest, OperatorProofResponse, ProofData, ProofDescResponse, ProofTaskState, WatchtowerProofRequest, WatchtowerProofResponse, WatchtowerProofTimeoutUpdateRequest, WatchtowerProofTimeoutUpdateResponse, }; @@ -82,8 +82,8 @@ struct Args { struct AppState { /// First-seen instant per proof task key -> used to gate readiness. seen: Mutex>, - /// Watchtower task keys marked failed by the timeout endpoint (stay "not ready" forever). - failed_watchtower: Mutex>, + /// Watchtower task keys cancelled by the timeout endpoint. + cancelled_watchtower: Mutex>, /// Pre-parsed operator proof template. operator_template: ZKMProofWithPublicValues, operator_vk: String, @@ -254,6 +254,7 @@ async fn handle_operator_proof( let key = format!("operator:{}:{}", req.instance_id, req.graph_id); if !task_ready(&state, &key, operator_delay()) { return Json(OperatorProofResponse { + state: ProofTaskState::Pending, proof_data: None, error: Some("operator proof is not ready".to_string()), }); @@ -261,11 +262,19 @@ async fn handle_operator_proof( match operator_proof_data(&state, &req) { Ok(proof_data) => { tracing::info!(graph_id = %req.graph_id, "serving mock operator proof"); - Json(OperatorProofResponse { proof_data: Some(proof_data), error: None }) + Json(OperatorProofResponse { + state: ProofTaskState::Proven, + proof_data: Some(proof_data), + error: None, + }) } Err(error) => { tracing::warn!(graph_id = %req.graph_id, %error, "cannot build mock operator proof"); - Json(OperatorProofResponse { proof_data: None, error: Some(error) }) + Json(OperatorProofResponse { + state: ProofTaskState::Pending, + proof_data: None, + error: Some(error), + }) } } } @@ -278,22 +287,32 @@ async fn handle_watchtower_proof( log_auth_headers("watchtower", &headers); response_latency().await; let key = format!("watchtower:{}:{}:{}", req.instance_id, req.graph_id, req.public_key); - if state.failed_watchtower.lock().expect("failed set poisoned").contains(&key) { + if state.cancelled_watchtower.lock().expect("cancelled set poisoned").contains(&key) { return Json(WatchtowerProofResponse { + state: ProofTaskState::Cancelled, proof_data: None, - error: Some("watchtower proof task was marked failed".to_string()), + error: Some("watchtower proof task was cancelled".to_string()), }); } if !task_ready(&state, &key, watchtower_delay()) { return Json(WatchtowerProofResponse { + state: ProofTaskState::Pending, proof_data: None, error: Some("watchtower proof is not ready".to_string()), }); } tracing::info!(graph_id = %req.graph_id, "serving mock watchtower proof"); match watchtower_proof_data(&state, &key) { - Ok(data) => Json(WatchtowerProofResponse { proof_data: Some(data), error: None }), - Err(error) => Json(WatchtowerProofResponse { proof_data: None, error: Some(error) }), + Ok(data) => Json(WatchtowerProofResponse { + state: ProofTaskState::Proven, + proof_data: Some(data), + error: None, + }), + Err(error) => Json(WatchtowerProofResponse { + state: ProofTaskState::Pending, + proof_data: None, + error: Some(error), + }), } } @@ -304,7 +323,7 @@ async fn handle_watchtower_timeout( ) -> Json { log_auth_headers("watchtower-timeout", &headers); let key = format!("watchtower:{}:{}:{}", req.instance_id, req.graph_id, req.public_key); - state.failed_watchtower.lock().expect("failed set poisoned").insert(key); + state.cancelled_watchtower.lock().expect("cancelled set poisoned").insert(key); Json(WatchtowerProofTimeoutUpdateResponse { instance_id: req.instance_id, graph_id: req.graph_id, @@ -353,7 +372,7 @@ fn build_state() -> AppState { AppState { seen: Mutex::new(HashMap::new()), - failed_watchtower: Mutex::new(HashSet::new()), + cancelled_watchtower: Mutex::new(HashSet::new()), operator_template, operator_vk, operator_zkm_version, @@ -401,7 +420,7 @@ mod tests { let operator_template = mock_operator_template(); AppState { seen: Mutex::new(HashMap::new()), - failed_watchtower: Mutex::new(HashSet::new()), + cancelled_watchtower: Mutex::new(HashSet::new()), operator_template, operator_vk: DEFAULT_OPERATOR_VK.to_string(), operator_zkm_version: zkm_sdk::ZKM_CIRCUIT_VERSION.to_string(), diff --git a/node/src/utils/proof_client.rs b/node/src/utils/proof_client.rs index 618ffaf39..ba1d09211 100644 --- a/node/src/utils/proof_client.rs +++ b/node/src/utils/proof_client.rs @@ -32,7 +32,7 @@ use bitcoin_light_client_circuit::hash_operator_constant; use client::http_client::async_client::HttpAsyncClient; use proof_builder::api_auth::{ProofBuilderAuthRole, sign_proof_builder_request}; use proof_builder::{ - OperatorProofRequest, OperatorProofResponse, ProofData, WatchtowerProofRequest, + OperatorProofRequest, OperatorProofResponse, ProofData, ProofTaskState, WatchtowerProofRequest, WatchtowerProofResponse, WatchtowerProofTimeoutUpdateRequest, WatchtowerProofTimeoutUpdateResponse, }; @@ -73,6 +73,40 @@ pub struct OperatorStatement { pub constant: [u8; 32], } +/// What the proof builder said about a requested proof. +pub enum ProofPoll { + Ready(T), + /// Not ready yet; ask again after this many seconds. + NotReady(usize), + /// The proof builder will not produce the proof: the task failed for good or the + /// request was refused as invalid. Submitting again does not help. + Failed(String), + /// The task is no longer wanted on the proof builder. + Cancelled, +} + +/// The reason of a request the proof builder refused as invalid (HTTP 400); the same +/// request is refused again, so it is not retried. +fn refused_request(error: &anyhow::Error) -> Option { + match error.downcast_ref::() { + Some(esplora_client::Error::HttpResponse { status: 400, message }) => { + Some(format!("the proof builder refused the request: {message}")) + } + _ => None, + } +} + +/// The poll result of a task that has no proof yet. +fn unfinished(state: ProofTaskState, error: Option, wait_secs: usize) -> ProofPoll { + match state { + ProofTaskState::Failed => { + ProofPoll::Failed(error.unwrap_or_else(|| "the proof task failed".to_string())) + } + ProofTaskState::Cancelled => ProofPoll::Cancelled, + ProofTaskState::Pending | ProofTaskState::Proven => ProofPoll::NotReady(wait_secs), + } +} + fn gen_watchtower_commitment(graph_id: Uuid, proof_data: ProofData) -> Result> { let graph_id = graph_id.as_bytes(); let proof = @@ -111,16 +145,15 @@ fn proof_builder_auth_headers( Ok(sign_proof_builder_request(keypair, role, "POST", path, body)?.to_header_pairs()) } -/// Returns: -/// - `Ok(Some(WatchtowerCommitment), _)` if watchtower proof is available -/// - `Ok(None, wait_secs)` if watchtower proof is not yet available, with suggested wait time +/// Requests this watchtower's proof for the graph and returns its commitment once the +/// proof is available. pub async fn get_watchtower_commitment( local_db: &LocalDB, btc_client: &BTCClient, http_client: &HttpAsyncClient, instance_id: Uuid, graph_id: Uuid, -) -> Result<(Option>, usize)> { +) -> Result>> { let mut storage_processor = local_db.acquire().await?; if let Some(graph) = storage_processor.find_graph(&graph_id).await? && let Some(challenge_init_txid) = graph.watchtower_challenge_init_txid @@ -129,7 +162,7 @@ pub async fn get_watchtower_commitment( let tx_status = btc_client.get_tx_status(&challenge_init_txid.0).await?; if !tx_status.confirmed { warn!("graph {graph_id} challenge tx is not confirmed"); - return Ok((None, get_watchtower_proof_wait_secs())); + return Ok(ProofPoll::NotReady(get_watchtower_proof_wait_secs())); } let base_url = Url::parse( @@ -147,7 +180,7 @@ pub async fn get_watchtower_commitment( execution_layer_block_number: graph.proceed_withdraw_height, // NOTE: this number may be zero }; let auth_keypair = get_bitvm_key()?; - let response = http_client + let response = match http_client .post_response_json_with_dynamic_headers::( url.as_str(), &payload, @@ -164,13 +197,17 @@ pub async fn get_watchtower_commitment( .inspect(|_| crate::metrics_service::record_backend_ok("proof")) .inspect_err(|error| { crate::metrics_service::record_backend_failure("proof", "watchtower_proofs", error) - })?; + }) { + Ok(response) => response, + Err(error) => return refused_request(&error).map(ProofPoll::Failed).ok_or(error), + }; match response.proof_data { Some(proof_data) => { - let watchtower_commitment = gen_watchtower_commitment(graph_id, proof_data)?; - Ok((Some(watchtower_commitment), 0)) + Ok(ProofPoll::Ready(gen_watchtower_commitment(graph_id, proof_data)?)) + } + None => { + Ok(unfinished(response.state, response.error, get_watchtower_proof_wait_secs())) } - None => Ok((None, get_watchtower_proof_wait_secs())), } } else { warn!("graph:{graph_id} not found or related txn is none",); @@ -404,9 +441,8 @@ pub fn derive_operator_statement( Ok(OperatorStatement { static_input, vk_hash, zkm_version, constant }) } -/// Returns: -/// - `Ok(Some(OperatorProof), _)` if operator proof is available and valid -/// - `Ok(None, wait_secs)` if operator proof is not yet available +/// Requests the operator proof for the graph and returns it once it is available and +/// checked against the graph. pub async fn get_operator_proof( local_db: &LocalDB, http_client: &HttpAsyncClient, @@ -414,7 +450,7 @@ pub async fn get_operator_proof( btc_client: &BTCClient, instance_id: Uuid, graph_id: Uuid, -) -> Result<(Option, usize)> { +) -> Result> { let mut storage_processor = local_db.acquire().await?; let Some(graph) = storage_processor.find_graph(&graph_id).await? else { warn!("graph:{graph_id} not found"); @@ -424,7 +460,7 @@ pub async fn get_operator_proof( if graph.proceed_withdraw_height <= 0 { warn!("graph {graph_id} proceed_withdraw_height <= 0, waiting to been updated"); - return Ok((None, get_operator_proof_wait_secs())); + return Ok(ProofPoll::NotReady(get_operator_proof_wait_secs())); } let watchtower_challenge_init_txid = graph @@ -448,7 +484,7 @@ pub async fn get_operator_proof( Ok(info) => info, Err(e) => { warn!("Failed to get watchtower challenge info: {e}"); - return Ok((None, get_operator_proof_wait_secs())); + return Ok(ProofPoll::NotReady(get_operator_proof_wait_secs())); } }; let Some(operator_committed_blockhash) = get_operator_committed_blockhash( @@ -459,7 +495,7 @@ pub async fn get_operator_proof( ) .await? else { - return Ok((None, get_operator_proof_wait_secs())); + return Ok(ProofPoll::NotReady(get_operator_proof_wait_secs())); }; let base_url = Url::parse( @@ -484,7 +520,7 @@ pub async fn get_operator_proof( .collect(), }; let auth_keypair = get_bitvm_key()?; - let operator_response = http_client + let operator_response = match http_client .post_response_json_with_dynamic_headers::( operator_url.as_str(), &payload, @@ -501,13 +537,20 @@ pub async fn get_operator_proof( .inspect(|_| crate::metrics_service::record_backend_ok("proof")) .inspect_err(|error| { crate::metrics_service::record_backend_failure("proof", "operator_proofs", error) - })?; + }) { + Ok(response) => response, + Err(error) => return refused_request(&error).map(ProofPoll::Failed).ok_or(error), + }; let Some(proof_data) = operator_response.proof_data else { - if let Some(error) = operator_response.error { + if let Some(error) = &operator_response.error { info!("operator proof is not ready for graph_id:{graph_id}: {error}"); } - return Ok((None, get_operator_proof_wait_secs())); + return Ok(unfinished( + operator_response.state, + operator_response.error, + get_operator_proof_wait_secs(), + )); }; let statement = derive_operator_statement( @@ -543,17 +586,14 @@ pub async fn get_operator_proof( bail!("operator proof static public inputs do not match graph setup statement"); } - Ok(( - Some(ValidatedOperatorProof { - proof: ark_proof.proof, - public_inputs: ark_proof.public_inputs.into(), - verifying_key: ark_proof.groth16_vk.into(), - public_values: proof.public_values.to_vec(), - vk_hash: proof_data.vk, - zkm_version: proof.zkm_version, - }), - 0, - )) + Ok(ProofPoll::Ready(ValidatedOperatorProof { + proof: ark_proof.proof, + public_inputs: ark_proof.public_inputs.into(), + verifying_key: ark_proof.groth16_vk.into(), + public_values: proof.public_values.to_vec(), + vk_hash: proof_data.vk, + zkm_version: proof.zkm_version, + })) } /// Tell the proof builder that this watchtower's proof for the graph is no @@ -824,3 +864,29 @@ mod commit_pubin_tests { assert_eq!(&bitmap[1..], &[0u8; 31]); } } + +#[cfg(test)] +mod poll_tests { + use super::*; + + #[test] + fn only_a_bad_request_and_a_failed_task_are_final() { + let refused = |status| { + refused_request( + &esplora_client::Error::HttpResponse { status, message: "bad".to_string() }.into(), + ) + }; + assert!(refused(400).is_some()); + for status in [401, 403, 429, 500, 503] { + assert!(refused(status).is_none(), "{status} is retried"); + } + assert!(refused_request(&anyhow!("connection reset")).is_none()); + + let poll = |state| unfinished::<()>(state, Some("why".to_string()), 7); + assert!(matches!(poll(ProofTaskState::Pending), ProofPoll::NotReady(7))); + assert!( + matches!(poll(ProofTaskState::Failed), ProofPoll::Failed(reason) if reason == "why") + ); + assert!(matches!(poll(ProofTaskState::Cancelled), ProofPoll::Cancelled)); + } +} diff --git a/proof-builder-rpc/src/api/metrics_service.rs b/proof-builder-rpc/src/api/metrics_service.rs index 960876cce..43194bdeb 100644 --- a/proof-builder-rpc/src/api/metrics_service.rs +++ b/proof-builder-rpc/src/api/metrics_service.rs @@ -319,6 +319,7 @@ fn proof_state_label(state: &str) -> &'static str { "1" => "proving", "2" => "proven", "3" => "failed", + "4" => "cancelled", _ => "unknown", } } @@ -390,7 +391,7 @@ mod tests { }, store::MetricsStateCount { category: "operator".to_owned(), - state: "4".to_owned(), + state: "5".to_owned(), count: 2, oldest_created_at: Some(70), last_success_at: None, diff --git a/proof-builder-rpc/src/api/proof_handler.rs b/proof-builder-rpc/src/api/proof_handler.rs index 08061b528..1305cc375 100644 --- a/proof-builder-rpc/src/api/proof_handler.rs +++ b/proof-builder-rpc/src/api/proof_handler.rs @@ -5,7 +5,8 @@ use crate::api::response::{ApiErrorExt, ApiResult, ok_response}; use crate::api::routes; use crate::api::validation::InputValidator; use crate::task::{ - add_operator_task, add_watchtower_task, find_operator_task, find_watchtower_task, + OperatorTaskInput, add_operator_task, add_watchtower_task, find_operator_task, + find_watchtower_task, restart_failed_operator_task, restart_failed_watchtower_task, update_operator_task_state, update_watchtower_task_state, watchtower_task_key, }; use axum::Json; @@ -15,13 +16,40 @@ use proof_builder::api_auth::ProofBuilderAuthRole; use proof_builder::{ ChainProofDescRequest, OperatorProofDescRequest, OperatorProofRequest, OperatorProofResponse, OperatorProofTimeoutUpdateRequest, OperatorProofTimeoutUpdateResponse, ProofData, ProofDesc, - ProofDescResponse, ProofType, WatchtowerProofRequest, WatchtowerProofResponse, + ProofDescResponse, ProofTaskState, ProofType, WatchtowerProofRequest, WatchtowerProofResponse, WatchtowerProofTimeoutUpdateRequest, WatchtowerProofTimeoutUpdateResponse, }; use std::sync::Arc; use store::ProofState; use tracing::info; +/// The API result label of a task in `proof_state`. +fn state_result(proof_state: i64) -> &'static str { + match ProofState::from_i64(proof_state) { + Some(ProofState::Proven) => "ready", + Some(ProofState::Failed) => "failed", + Some(ProofState::Cancelled) => "cancelled", + _ => "pending", + } +} + +/// The submission state of an unfinished task in `proof_state`. +fn task_state(proof_state: i64) -> ProofTaskState { + match ProofState::from_i64(proof_state) { + Some(ProofState::Failed) => ProofTaskState::Failed, + Some(ProofState::Cancelled) => ProofTaskState::Cancelled, + _ => ProofTaskState::Pending, + } +} + +/// Why a failed task is not restarted. +fn failure_reason(invalid_reason: Option, reset_count: i64) -> String { + match invalid_reason { + Some(reason) => format!("the task inputs can never be proven: {reason}"), + None => format!("the proof failed again after {reset_count} restarts"), + } +} + /// Marks validation failures as invalid API results while preserving the original result. fn record_invalid(api_result: &mut ApiResultGuard, result: Result) -> Result { result.inspect_err(|_| api_result.set("invalid")) @@ -83,13 +111,7 @@ pub(super) async fn get_chain_proof_task_desc( .api_error("GET_CHAIN_PROOF_ERROR")? .map(|v| v.block_start); - api_result.set(if proof.proof_state == ProofState::Proven.to_i64() { - "ready" - } else if proof.proof_state == ProofState::Failed.to_i64() { - "failed" - } else { - "pending" - }); + api_result.set(state_result(proof.proof_state)); ok_response(ProofDescResponse { proof_desc: Some(ProofDesc { block_start: proof.block_start, @@ -142,13 +164,7 @@ pub(super) async fn get_operator_proof_task_desc( .api_error("POST_OPERATOR_PROOF_TASK_ERROR")?; match operator_proof { Some(operator_proof) => { - api_result.set(if operator_proof.proof_state == ProofState::Proven.to_i64() { - "ready" - } else if operator_proof.proof_state == ProofState::Failed.to_i64() { - "failed" - } else { - "pending" - }); + api_result.set(state_result(operator_proof.proof_state)); info!("Get Operator Proof:{operator_proof:?}"); ok_response(ProofDescResponse { proof_desc: Some(ProofDesc { @@ -230,6 +246,7 @@ pub(super) async fn post_operator_proof_task( { api_result.set("ready"); ok_response(OperatorProofResponse { + state: ProofTaskState::Proven, proof_data: Some(ProofData::load_proof_data( &operator_proof.path_to_proof.unwrap(), ProofType::Operator, @@ -237,13 +254,41 @@ pub(super) async fn post_operator_proof_task( error: None, }) } + Some(operator_proof) if operator_proof.proof_state == ProofState::Failed.to_i64() => { + record_invalid( + &mut api_result, + InputValidator::validate_operator_task(&payload, &graph_data), + )?; + let restarted = restart_failed_operator_task( + &api_state.local_db, + &operator_proof, + &OperatorTaskInput::from_request(&payload), + ) + .await + .api_error("POST_OPERATOR_PROOF_TASK_ERROR")?; + if restarted { + info!(%instance_id, %graph_id, "restarting the failed operator task"); + api_result.set("restarted"); + return ok_response(OperatorProofResponse { + state: ProofTaskState::Pending, + proof_data: None, + error: Some("The failed task was restarted".to_string()), + }); + } + api_result.set("failed"); + ok_response(OperatorProofResponse { + state: ProofTaskState::Failed, + proof_data: None, + error: Some(failure_reason( + operator_proof.invalid_reason, + operator_proof.reset_count, + )), + }) + } Some(operator_proof) => { - api_result.set(if operator_proof.proof_state == ProofState::Failed.to_i64() { - "failed" - } else { - "pending" - }); + api_result.set(state_result(operator_proof.proof_state)); ok_response(OperatorProofResponse { + state: task_state(operator_proof.proof_state), proof_data: None, error: Some(format!( "The proof is not ready, state {}, path:{:?}", @@ -261,17 +306,13 @@ pub(super) async fn post_operator_proof_task( &api_state.local_db, instance_id, graph_id, - payload.operator_committed_blockhash, - payload.execution_layer_block_number, - payload.watchtower_challenge_txids.clone(), - payload.included_watchtowers.clone(), - payload.watchtower_challenge_init_txid.clone(), - payload.watchtower_challenge_pubkeys.clone(), + &OperatorTaskInput::from_request(&payload), ) .await .api_error("POST_OPERATOR_PROOF_TASK_ERROR")?; api_result.set("created"); ok_response(OperatorProofResponse { + state: ProofTaskState::Pending, proof_data: None, error: Some("No proof found".to_string()), }) @@ -322,7 +363,7 @@ pub(super) async fn update_operator_proof_task_timeout( instance_id, graph_id, ProofState::New, - ProofState::Failed, + ProofState::Cancelled, ) .await { @@ -404,6 +445,7 @@ pub(super) async fn post_watchtower_proof_task( { api_result.set("ready"); ok_response(WatchtowerProofResponse { + state: ProofTaskState::Proven, proof_data: Some(ProofData::load_proof_data( &watchtower_proof.path_to_proof.unwrap(), ProofType::Watchtower, @@ -411,16 +453,41 @@ pub(super) async fn post_watchtower_proof_task( error: None, }) } + Some(watchtower_proof) if watchtower_proof.proof_state == ProofState::Failed.to_i64() => { + let restarted = restart_failed_watchtower_task( + &api_state.local_db, + &watchtower_proof, + &challenge_init_txid, + payload.execution_layer_block_number, + ) + .await + .api_error("POST_WATCHTOWER_PROOF_TASK_ERROR")?; + if restarted { + info!(%instance_id, %graph_id, "restarting the failed watchtower task"); + api_result.set("restarted"); + return ok_response(WatchtowerProofResponse { + state: ProofTaskState::Pending, + proof_data: None, + error: Some("The failed task was restarted".to_string()), + }); + } + api_result.set("failed"); + ok_response(WatchtowerProofResponse { + state: ProofTaskState::Failed, + proof_data: None, + error: Some(failure_reason( + watchtower_proof.invalid_reason, + watchtower_proof.reset_count, + )), + }) + } Some(watchtower_proof) => { - api_result.set(if watchtower_proof.proof_state == ProofState::Failed.to_i64() { - "failed" - } else { - "pending" - }); + api_result.set(state_result(watchtower_proof.proof_state)); ok_response(WatchtowerProofResponse { + state: task_state(watchtower_proof.proof_state), proof_data: None, error: Some(format!( - "No proof is not ready, state {}, path:{:?}", + "The proof is not ready, state {}, path:{:?}", watchtower_proof.proof_state, watchtower_proof.path_to_proof )), }) @@ -443,6 +510,7 @@ pub(super) async fn post_watchtower_proof_task( } api_result.set("created"); ok_response(WatchtowerProofResponse { + state: ProofTaskState::Pending, proof_data: None, error: Some("No proof found".to_string()), }) @@ -494,7 +562,7 @@ pub(super) async fn update_watchtower_proof_task_timeout( graph_id, &watchtower_task_key(&request.signer), ProofState::New, - ProofState::Failed, + ProofState::Cancelled, ) .await { diff --git a/proof-builder-rpc/src/api/validation.rs b/proof-builder-rpc/src/api/validation.rs index 4ea91c320..8ce65f64c 100644 --- a/proof-builder-rpc/src/api/validation.rs +++ b/proof-builder-rpc/src/api/validation.rs @@ -1,4 +1,5 @@ use crate::api::response::ErrorResponse; +use crate::task::validate_indexed_watchtower_challenges; use axum::Json; use axum::http::StatusCode; use bitcoin::hashes::Hash; @@ -111,6 +112,13 @@ impl InputValidator { ), ); } + if let Err(error) = validate_indexed_watchtower_challenges( + &request.watchtower_challenge_txids, + &request.included_watchtowers, + &request.watchtower_challenge_pubkeys, + ) { + return invalid("INVALID_WATCHTOWER_CHALLENGES", format!("{error:#}")); + } Ok(()) } @@ -173,3 +181,47 @@ impl InputValidator { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_bad_watchtower_list_is_a_bad_request() { + let graph_data = GraphData { + operator_pubkey_prefix: 0, + operator_pubkey: [0; 32], + pegin_txid: [0; 32], + kickoff_txid: [0; 32], + take1_txid: [0; 32], + take2_txid: [0; 32], + watchtower_challenge_init_txid: [0x22; 32], + prover_assert_txid: [0; 32], + disprove_txids: vec![], + watchtower_challenge_timeout_txids: vec![[0; 32]], + operator_challenge_nack_txids: vec![], + operator_commit_timeout_txid: [0; 32], + }; + let mut request = OperatorProofRequest { + instance_id: Uuid::nil().to_string(), + graph_id: Uuid::nil().to_string(), + gateway_address: None, + operator_committed_blockhash: "11".repeat(32), + execution_layer_block_number: 1, + watchtower_challenge_txids: vec![None], + included_watchtowers: vec![false], + watchtower_challenge_init_txid: "22".repeat(32), + watchtower_challenge_pubkeys: vec![ + "0272efe7ccae21d2541ad85d4f2961f2e5593c29dc8bc37bf87035fc2d5527a651".to_string(), + ], + }; + assert!(InputValidator::validate_operator_task(&request, &graph_data).is_ok()); + + // Included, but no challenge txid. + request.included_watchtowers = vec![true]; + let (status, Json(error)) = + InputValidator::validate_operator_task(&request, &graph_data).unwrap_err(); + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(error.error, "INVALID_WATCHTOWER_CHALLENGES"); + } +} diff --git a/proof-builder-rpc/src/task/mod.rs b/proof-builder-rpc/src/task/mod.rs index c048c74f8..ebf25561a 100644 --- a/proof-builder-rpc/src/task/mod.rs +++ b/proof-builder-rpc/src/task/mod.rs @@ -25,12 +25,14 @@ use std::sync::Arc; use std::time::{Instant, UNIX_EPOCH}; use uuid::Uuid; +use bitcoin::hashes::{Hash, sha256}; use futures::future::Either; -use proof_builder::{OnDemandTask, ProofBuilder, ProofRequest}; +use proof_builder::{OnDemandTask, OperatorProofRequest, ProofBuilder, ProofRequest}; +use serde::Serialize; use store::localdb::LocalDB; use store::{ LongRunningTaskProof, OnDemandProofCompletion, OnDemandProofTable, OperatorProof, ProofState, - WatchtowerProof, + SerializableTxid, WatchtowerProof, }; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; @@ -360,13 +362,7 @@ pub(crate) async fn fail_invalid_on_demand_task( reason: &str, ) -> anyhow::Result<()> { warn!(task_index, "{} task cannot be proven; marking it failed: {reason}", table.table_name()); - let completion = - OnDemandProofCompletion { proof_state: ProofState::Failed.to_i64(), ..Default::default() }; - local_db - .acquire() - .await? - .complete_on_demand_proof_if_new(table, task_index, &completion) - .await?; + local_db.acquire().await?.fail_invalid_on_demand_proof(table, task_index, reason).await?; Ok(()) } @@ -1020,7 +1016,7 @@ pub(crate) fn watchtower_task_key(public_key: &XOnlyPublicKey) -> String { /// table schema: (index, instance_id, graph_id, public_key, challenge_txid, challenge_init_txid, path_to_proof, cycles, state, update_time) /// * index: incremental id -/// * state: 0-new, 1-doing, 2-done, 3-failed +/// * state: 0-new, 1-doing, 2-done, 3-failed, 4-cancelled /// /// Invoked by API. /// @@ -1063,6 +1059,43 @@ pub(crate) async fn add_watchtower_task( Ok(true) } +/// Invoked by API. +/// +/// Restarts a failed Watchtower task that its Watchtower submitted again, like +/// [`restart_failed_operator_task`]. Returns whether the task was restarted. +pub(crate) async fn restart_failed_watchtower_task( + local_db: &LocalDB, + task: &WatchtowerProof, + challenge_init_txid: &str, + execution_layer_block_number: i64, +) -> anyhow::Result { + let challenge_init_txid: SerializableTxid = Txid::from_str(challenge_init_txid)?.into(); + let same_inputs = task.challenge_init_txid.0 == challenge_init_txid.0 + && task.execution_layer_block_number == execution_layer_block_number; + let mut storage_processor = local_db.start_transaction().await?; + let restarted = storage_processor + .reset_failed_on_demand_proof( + OnDemandProofTable::Watchtower, + task.id, + same_inputs.then_some(MAX_TASK_RESETS), + ) + .await?; + if restarted == 0 { + return Ok(false); + } + if !same_inputs { + storage_processor + .update_watchtower_proof_inputs( + task.id, + &challenge_init_txid, + execution_layer_block_number, + ) + .await?; + } + storage_processor.commit().await?; + Ok(true) +} + pub(crate) async fn find_watchtower_task( local_db: &LocalDB, instance_id: Uuid, @@ -1095,30 +1128,151 @@ pub(crate) async fn update_watchtower_task_state( .await } -/// table schema: (index, instance_id, graph_id, execution_layer_block_number, path_to_proof, cycles, state, update_time) -/// * state: 0-new, 1-doing, 2-done, 3-failed -/// * execution_layer_block_number: proceedWithdraw's block number -/// * index: incremental id -/// Invocated by API +/// How often a task owner may restart a failed Operator or Watchtower task by submitting +/// the same inputs again. New inputs start the count over. +pub(crate) const MAX_TASK_RESETS: i64 = 2; + +/// The inputs of an Operator task, as the Operator submits them. +#[derive(Debug, Clone, Serialize)] +pub(crate) struct OperatorTaskInput { + pub operator_committed_blockhash: String, + /// proceedWithdraw's block number. + pub execution_layer_block_number: i64, + pub watchtower_challenge_txids: Vec>, + pub included_watchtowers: Vec, + pub watchtower_challenge_init_txid: String, + pub watchtower_challenge_pubkeys: Vec, +} + +impl OperatorTaskInput { + pub(crate) fn from_request(request: &OperatorProofRequest) -> Self { + Self { + operator_committed_blockhash: request.operator_committed_blockhash.clone(), + execution_layer_block_number: request.execution_layer_block_number, + watchtower_challenge_txids: request.watchtower_challenge_txids.clone(), + included_watchtowers: request.included_watchtowers.clone(), + watchtower_challenge_init_txid: request.watchtower_challenge_init_txid.clone(), + watchtower_challenge_pubkeys: request.watchtower_challenge_pubkeys.clone(), + } + } + + fn hash(&self) -> anyhow::Result { + Ok(sha256::Hash::hash(&serde_json::to_vec(self)?).to_string()) + } +} + +/// Invoked by API. +/// +/// Adds the Operator task of a graph and fixes the graph's Watchtower slots. #[tracing::instrument(level = "info", skip(local_db))] pub(crate) async fn add_operator_task( local_db: &LocalDB, instance_id: Uuid, graph_id: Uuid, - operator_committed_blockhash: String, - execution_layer_block_number: i64, - watchtower_challenge_txids: Vec>, - included_watchtowers: Vec, - watchtower_challenge_init_txid: String, - watchtower_challenge_pubkeys: Vec, + input: &OperatorTaskInput, ) -> anyhow::Result { let graph_watchtower_xonly_keys = validate_indexed_watchtower_challenges( - &watchtower_challenge_txids, - &included_watchtowers, - &watchtower_challenge_pubkeys, + &input.watchtower_challenge_txids, + &input.included_watchtowers, + &input.watchtower_challenge_pubkeys, + )?; + let input_hash = input.hash()?; + let mut storage_processor = local_db.start_transaction().await?; + sync_watchtower_slots( + &mut storage_processor, + instance_id, + graph_id, + &graph_watchtower_xonly_keys, + input, + ) + .await?; + let affected_rows = storage_processor + .create_operator_proof(&OperatorProof { + id: 1, + instance_id, + graph_id, + execution_layer_block_number: input.execution_layer_block_number, + proof_state: ProofState::New.to_i64(), + created_at: current_time_secs(), + updated_at: current_time_secs(), + cycles: 0, + operator_committed_blockhash: input.operator_committed_blockhash.clone(), + ..Default::default() + }) + .await?; + storage_processor + .update_operator_proof_inputs( + &instance_id, + &graph_id, + &input.operator_committed_blockhash, + input.execution_layer_block_number, + &input_hash, + ) + .await?; + storage_processor.commit().await?; + Ok(affected_rows) +} + +/// Invoked by API. +/// +/// Restarts a failed Operator task that the Operator submitted again. New inputs replace +/// the stored ones. The same inputs restart it at most [`MAX_TASK_RESETS`] times, and +/// never when they can never be proven. Returns whether the task was restarted. +pub(crate) async fn restart_failed_operator_task( + local_db: &LocalDB, + task: &OperatorProof, + input: &OperatorTaskInput, +) -> anyhow::Result { + let graph_watchtower_xonly_keys = validate_indexed_watchtower_challenges( + &input.watchtower_challenge_txids, + &input.included_watchtowers, + &input.watchtower_challenge_pubkeys, )?; + let input_hash = input.hash()?; + let same_inputs = task.input_hash.as_deref() == Some(input_hash.as_str()); let mut storage_processor = local_db.start_transaction().await?; + let restarted = storage_processor + .reset_failed_on_demand_proof( + OnDemandProofTable::Operator, + task.id, + same_inputs.then_some(MAX_TASK_RESETS), + ) + .await?; + if restarted == 0 { + return Ok(false); + } + if !same_inputs { + sync_watchtower_slots( + &mut storage_processor, + task.instance_id, + task.graph_id, + &graph_watchtower_xonly_keys, + input, + ) + .await?; + storage_processor + .update_operator_proof_inputs( + &task.instance_id, + &task.graph_id, + &input.operator_committed_blockhash, + input.execution_layer_block_number, + &input_hash, + ) + .await?; + } + storage_processor.commit().await?; + Ok(true) +} +/// Gives each of the graph's Watchtowers one task row at its slot. A Watchtower that has +/// not asked for a proof gets a `Cancelled` row: its challenge window is over. +async fn sync_watchtower_slots( + storage_processor: &mut store::localdb::StorageProcessor<'_>, + instance_id: Uuid, + graph_id: Uuid, + graph_watchtower_xonly_keys: &[[u8; 32]], + input: &OperatorTaskInput, +) -> anyhow::Result<()> { let existing_watchtower_proof_task = storage_processor .find_watchtower_proof_by_instance_and_graph(&instance_id, &graph_id) .await?; @@ -1154,69 +1308,52 @@ pub(crate) async fn add_operator_task( .filter(|node_index| !existing_watchtower_indices.contains(node_index)) .collect(); tracing::info!("missing_watchtower_indices: {:?}", missing_watchtower_indices); - // insert timeout watchtower proof task with failed state. + let init_txid = &input.watchtower_challenge_init_txid; for node_index in missing_watchtower_indices { - let stored_txid = watchtower_challenge_txids[node_index] - .as_deref() - .unwrap_or(&watchtower_challenge_init_txid); + let stored_txid = + input.watchtower_challenge_txids[node_index].as_deref().unwrap_or(init_txid); let task = WatchtowerProof { id: 1, instance_id, graph_id, - challenge_init_txid: Txid::from_str(&watchtower_challenge_init_txid)?.into(), + challenge_init_txid: Txid::from_str(init_txid)?.into(), challenge_txid: Txid::from_str(stored_txid)?.into(), - proof_state: ProofState::Failed.to_i64(), + proof_state: ProofState::Cancelled.to_i64(), created_at: current_time_secs(), updated_at: current_time_secs(), - execution_layer_block_number, + execution_layer_block_number: input.execution_layer_block_number, // The Graph key identifies the missing Watchtower slot. public_key: watchtower_task_key(&XOnlyPublicKey::from_slice( &graph_watchtower_xonly_keys[node_index], )?), node_index: node_index as i32, - included: included_watchtowers[node_index], + included: input.included_watchtowers[node_index], ..Default::default() }; let affected = storage_processor.create_watchtower_proof(&task).await?; tracing::info!( - "mark timeout watchtower proof as failed, node_index: {node_index}, affected rows: {affected}", + "cancelled the watchtower slot that asked for no proof, node_index: {node_index}, affected rows: {affected}", ); } // update watchtower's challenge txid. - for (i, txid) in watchtower_challenge_txids.iter().enumerate() { - let stored_txid = txid.as_deref().unwrap_or(&watchtower_challenge_init_txid); + for (i, txid) in input.watchtower_challenge_txids.iter().enumerate() { + let stored_txid = txid.as_deref().unwrap_or(init_txid); let affected = storage_processor .update_watchtower_proof_challenge_txid( &instance_id, &graph_id, i as i32, stored_txid, - included_watchtowers[i], + input.included_watchtowers[i], ) .await?; tracing::info!("update watchtower proof, node_index: {i} affected rows: {affected}",); } - - let affected_rows = storage_processor - .create_operator_proof(&OperatorProof { - id: 1, - instance_id, - graph_id, - execution_layer_block_number, - proof_state: ProofState::New.to_i64(), - created_at: current_time_secs(), - updated_at: current_time_secs(), - cycles: 0, - operator_committed_blockhash, - ..Default::default() - }) - .await?; - storage_processor.commit().await?; - Ok(affected_rows) + Ok(()) } -fn validate_indexed_watchtower_challenges( +pub(crate) fn validate_indexed_watchtower_challenges( challenge_txids: &[Option], included_watchtowers: &[bool], challenge_pubkeys: &[String], @@ -1424,19 +1561,15 @@ mod tests { ]; let included_watchtowers = vec![true, false]; - add_operator_task( - &local_db, - instance_id, - graph_id, + let input = OperatorTaskInput { operator_committed_blockhash, - number, + execution_layer_block_number: number, watchtower_challenge_txids, included_watchtowers, watchtower_challenge_init_txid, watchtower_challenge_pubkeys, - ) - .await - .unwrap(); + }; + add_operator_task(&local_db, instance_id, graph_id, &input).await.unwrap(); //use crate::task::read_watchtower_challenge_details; //let is_watchtower = false; @@ -1565,19 +1698,15 @@ mod watchtower_slot_tests { // The Operator lists the member as compressed with odd parity. let member_odd = format!("03{member}"); - add_operator_task( - &local_db, - Uuid::nil(), - graph_id, - "11".repeat(32), - 1, - vec![None], - vec![false], - "22".repeat(32), - vec![member_odd], - ) - .await - .unwrap(); + let input = OperatorTaskInput { + operator_committed_blockhash: "11".repeat(32), + execution_layer_block_number: 1, + watchtower_challenge_txids: vec![None], + included_watchtowers: vec![false], + watchtower_challenge_init_txid: "22".repeat(32), + watchtower_challenge_pubkeys: vec![member_odd], + }; + add_operator_task(&local_db, Uuid::nil(), graph_id, &input).await.unwrap(); let rows = local_db .acquire() @@ -1593,3 +1722,164 @@ mod watchtower_slot_tests { assert!(!add(&local_db, graph_id, &outsider).await.unwrap()); } } + +#[cfg(test)] +mod restart_tests { + use super::*; + use secp256k1::{Keypair, SECP256K1}; + use store::create_local_db; + + fn input(committed_blockhash: &str) -> OperatorTaskInput { + let key = Keypair::from_seckey_slice(SECP256K1, &[1; 32]).unwrap().x_only_public_key().0; + OperatorTaskInput { + operator_committed_blockhash: committed_blockhash.to_string(), + execution_layer_block_number: 1, + watchtower_challenge_txids: vec![None], + included_watchtowers: vec![false], + watchtower_challenge_init_txid: "22".repeat(32), + watchtower_challenge_pubkeys: vec![watchtower_task_key(&key)], + } + } + + async fn task(local_db: &LocalDB, graph_id: Uuid) -> OperatorProof { + find_operator_task(local_db, Uuid::nil(), graph_id).await.unwrap().unwrap() + } + + async fn fail(local_db: &LocalDB, id: i64) { + let completion = OnDemandProofCompletion { + proof_state: ProofState::Failed.to_i64(), + ..Default::default() + }; + let mut storage = local_db.acquire().await.unwrap(); + assert_eq!( + storage + .complete_on_demand_proof_if_new(OnDemandProofTable::Operator, id, &completion) + .await + .unwrap(), + 1 + ); + } + + #[tokio::test] + async fn a_failed_operator_task_restarts_twice_on_the_same_inputs_and_again_on_new_ones() { + let local_db = create_local_db("sqlite::memory:").await; + let graph_id = Uuid::new_v4(); + let first = input(&"11".repeat(32)); + add_operator_task(&local_db, Uuid::nil(), graph_id, &first).await.unwrap(); + // The Watchtower never asked for a proof: its slot is cancelled, not failed. + let slots = local_db + .acquire() + .await + .unwrap() + .find_watchtower_proof_by_instance_and_graph(&Uuid::nil(), &graph_id) + .await + .unwrap(); + assert_eq!(slots.len(), 1); + assert_eq!(slots[0].proof_state, ProofState::Cancelled.to_i64()); + // One task per graph. + assert!(add_operator_task(&local_db, Uuid::nil(), graph_id, &first).await.is_err()); + + // A task that is not failed is not restarted. + let id = task(&local_db, graph_id).await.id; + assert!( + !restart_failed_operator_task(&local_db, &task(&local_db, graph_id).await, &first) + .await + .unwrap() + ); + + for restarts in 1..=MAX_TASK_RESETS { + fail(&local_db, id).await; + let failed = task(&local_db, graph_id).await; + assert!(restart_failed_operator_task(&local_db, &failed, &first).await.unwrap()); + let restarted = task(&local_db, graph_id).await; + assert_eq!(restarted.proof_state, ProofState::New.to_i64()); + assert_eq!(restarted.reset_count, restarts); + } + fail(&local_db, id).await; + let failed = task(&local_db, graph_id).await; + assert!(!restart_failed_operator_task(&local_db, &failed, &first).await.unwrap()); + assert_eq!(task(&local_db, graph_id).await.proof_state, ProofState::Failed.to_i64()); + + // New inputs replace the old ones and start the count over. + let second = input(&"33".repeat(32)); + assert!(restart_failed_operator_task(&local_db, &failed, &second).await.unwrap()); + let replaced = task(&local_db, graph_id).await; + assert_eq!(replaced.id, id); + assert_eq!(replaced.proof_state, ProofState::New.to_i64()); + assert_eq!(replaced.reset_count, 0); + assert_eq!(replaced.operator_committed_blockhash, "33".repeat(32)); + } + + #[tokio::test] + async fn a_task_with_inputs_that_can_never_be_proven_is_not_restarted() { + let local_db = create_local_db("sqlite::memory:").await; + let graph_id = Uuid::new_v4(); + let first = input(&"11".repeat(32)); + add_operator_task(&local_db, Uuid::nil(), graph_id, &first).await.unwrap(); + let id = task(&local_db, graph_id).await.id; + fail_invalid_on_demand_task(&local_db, OnDemandProofTable::Operator, id, "bad block") + .await + .unwrap(); + let failed = task(&local_db, graph_id).await; + assert_eq!(failed.invalid_reason.as_deref(), Some("bad block")); + assert!(!restart_failed_operator_task(&local_db, &failed, &first).await.unwrap()); + // New inputs may be provable. + assert!( + restart_failed_operator_task(&local_db, &failed, &input(&"33".repeat(32))) + .await + .unwrap() + ); + assert_eq!(task(&local_db, graph_id).await.invalid_reason, None); + } + + #[tokio::test] + async fn a_failed_watchtower_task_restarts_like_an_operator_task() { + let local_db = create_local_db("sqlite::memory:").await; + let graph_id = Uuid::new_v4(); + let key = watchtower_task_key( + &Keypair::from_seckey_slice(SECP256K1, &[1; 32]).unwrap().x_only_public_key().0, + ); + let init_txid = "22".repeat(32); + add_watchtower_task(&local_db, Uuid::nil(), graph_id, key.clone(), init_txid.clone(), 5) + .await + .unwrap(); + let load = || async { + find_watchtower_task(&local_db, Uuid::nil(), graph_id, &key).await.unwrap().unwrap() + }; + let id = load().await.id; + let completion = OnDemandProofCompletion { + proof_state: ProofState::Failed.to_i64(), + ..Default::default() + }; + for _ in 0..MAX_TASK_RESETS { + local_db + .acquire() + .await + .unwrap() + .complete_on_demand_proof_if_new(OnDemandProofTable::Watchtower, id, &completion) + .await + .unwrap(); + assert!( + restart_failed_watchtower_task(&local_db, &load().await, &init_txid, 5) + .await + .unwrap() + ); + } + local_db + .acquire() + .await + .unwrap() + .complete_on_demand_proof_if_new(OnDemandProofTable::Watchtower, id, &completion) + .await + .unwrap(); + assert!( + !restart_failed_watchtower_task(&local_db, &load().await, &init_txid, 5).await.unwrap() + ); + // A new execution-layer height is a new input. + assert!( + restart_failed_watchtower_task(&local_db, &load().await, &init_txid, 6).await.unwrap() + ); + let restarted = load().await; + assert_eq!((restarted.reset_count, restarted.execution_layer_block_number), (0, 6)); + } +}