fix: let a failed operator or watchtower proof task restart, and tell the node when it never will

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<ProofData>,
     pub error: Option<String>,
 }
@@ -343,6 +359,7 @@ pub struct WatchtowerProofRequest {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct WatchtowerProofResponse {
+    pub state: ProofTaskState,
     pub proof_data: Option<ProofData>,
     pub error: Option<String>,
 }

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);

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<u64> {
+        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<i64>,
+    ) -> anyhow::Result<u64> {
+        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<u64> {
+        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<u64> {
+        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,

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<String>,
     pub extra: Option<String>,
     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<String>,
+    /// Hash of the request inputs, to tell a resubmission with new inputs from a retry.
+    pub input_hash: Option<String>,
     pub extra: Option<String>,
     pub created_at: i64,
     pub updated_at: i64,

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<Step>
     ) {
         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) => {