From ba67b11546ed8fce8e26aa47a1e581ea7b4a75af Mon Sep 17 00:00:00 2001 From: Ethan Date: Wed, 11 Nov 2026 07:40:26 +0000 Subject: [PATCH 1/3] fix: keep proven headers in one archive and refetch headers that break the chain --- Cargo.lock | 2 + circuits/header-chain-proof/host/src/lib.rs | 83 +++++- circuits/operator-proof/host/src/lib.rs | 12 +- circuits/proof-builder/Cargo.toml | 2 + circuits/proof-builder/src/headers.rs | 309 ++++++++++++++++++++ circuits/proof-builder/src/lib.rs | 1 + circuits/watchtower-proof/host/src/lib.rs | 13 +- 7 files changed, 395 insertions(+), 27 deletions(-) create mode 100644 circuits/proof-builder/src/headers.rs diff --git a/Cargo.lock b/Cargo.lock index b62934fb8..0e90986c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9144,7 +9144,9 @@ name = "proof-builder" version = "0.4.0" dependencies = [ "anyhow", + "bincode", "bitcoin", + "borsh", "commit-chain", "header-chain", "hex", diff --git a/circuits/header-chain-proof/host/src/lib.rs b/circuits/header-chain-proof/host/src/lib.rs index e1dccea64..b150b65c5 100644 --- a/circuits/header-chain-proof/host/src/lib.rs +++ b/circuits/header-chain-proof/host/src/lib.rs @@ -226,6 +226,7 @@ async fn fetch_header_chain_with_client( let mut i = block_headers.len(); let mut retries = 9; + let mut reorgs = 0; while i < end { tracing::info!("get block by height: {i}"); let tip_height = btc_client.get_height().await?; @@ -238,6 +239,25 @@ async fn fetch_header_chain_with_client( Ok(block) => { tracing::info!("block_id {i}: {}", block.block_hash().to_string()); let header: header_chain::CircuitBlockHeader = block.header.into(); + if let Some(last) = block_headers.last() + && header.prev_block_hash != last.compute_block_hash() + { + // A reorg replaced headers fetched for this batch: drop them and fetch + // again. Below `start` they are proven, and only a human can sort that out. + block_headers.truncate(start); + writer.set_len((start * 80) as u64)?; + anyhow::ensure!( + i > start && reorgs < MAX_BATCH_REORGS, + "header {i} does not extend header {}; the proven chain or the \ + Bitcoin source changed", + i - 1 + ); + tracing::warn!(height = i, start, "Bitcoin reorg inside the batch; refetching"); + reorgs += 1; + writer.seek(std::io::SeekFrom::Start((start * 80) as u64))?; + i = start; + continue; + } block_headers.push(header.clone()); header.serialize(&mut writer)?; retries = 3; @@ -255,12 +275,15 @@ async fn fetch_header_chain_with_client( }; } writer.set_len((block_headers.len() * 80) as u64)?; - // The proof's `.blocks` sidecar is published with the proof (see `save_proof`). + // The proven headers are archived with the proof (see `save_proof`). Ok(block_headers) } -/// The `.blocks` sidecar of a header-chain proof: every header up to the proof's last -/// block, 80 bytes each. Operator and watchtower proofs read it to build SPV proofs. +/// Reorgs one batch fetch rides out before it gives up. +const MAX_BATCH_REORGS: usize = 3; + +/// Headers as the header cache, the header archive and the Genesis seed sidecar hold +/// them: 80 bytes each. pub fn proof_block_headers(headers: &[CircuitBlockHeader]) -> anyhow::Result> { let mut bytes = Vec::with_capacity(headers.len() * 80); for header in headers { @@ -437,10 +460,12 @@ impl ProofBuilder for HeaderChainProofBuilder { else { anyhow::bail!("invalid context"); }; - let proven_headers = total_block_headers - .get(..start + batch_size) - .context("block headers do not cover the proven batch")?; - let blocks = proof_block_headers(proven_headers)?; + // Headers go to the shared archive, durably, before the proof that needs them. + proof_builder::headers::extend_header_archive( + proof_builder::headers::header_archive_path(output_proof), + total_block_headers, + start + batch_size, + )?; //fs::write(&output_proof, bincode::serialize(&proof)?)?; //let public_value_hex = hex::encode(proof.public_values.as_slice()); //let proof_size = proof.bytes().len(); @@ -459,9 +484,15 @@ impl ProofBuilder for HeaderChainProofBuilder { (".public_inputs.bin", public_values.as_slice()), (".vk_hash.bin", vk_hash.as_bytes()), (".zkm_version.bin", proof.zkm_version.as_bytes()), - (".blocks", blocks.as_slice()), ], )?; + if let Some(dir) = std::path::Path::new(output_proof).parent() { + match proof_builder::headers::remove_superseded_sidecars(dir) { + Ok(0) => {} + Ok(removed) => tracing::info!(removed, "removed per-proof header sidecars"), + Err(error) => tracing::warn!("remove per-proof header sidecars: {error:#}"), + } + } tracing::info!("Generate proof successfully, proof: {:?}", proof); Ok((public_value_hex, proof_size)) @@ -666,6 +697,40 @@ mod tests { } } + #[tokio::test] + async fn a_header_that_does_not_extend_the_chain_is_refetched_or_refused() { + let blocks = test_blocks(); + let mut fork = blocks[2].clone(); + fork.header.prev_blockhash = blocks[0].block_hash(); + for start in [1, 2] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("headers.bin"); + let cached: Vec = + blocks[..start].iter().map(|b| b.header.into()).collect(); + fs::write(&path, proof_block_headers(&cached).unwrap()).unwrap(); + let (client, mock) = BTCClient::new_mock_client(); + mock.set_height(100); + for (height, block) in [(1, &blocks[1]), (2, &fork)] { + mock.set_block_hash(height, block.block_hash()); + mock.set_block(block.block_hash(), block.clone()); + } + // From 1, height 2 never extends the batch's own height 1: refetched, then + // given up. From 2, it does not extend the proven chain: refused at once. + let error = fetch_header_chain_with_client( + &client, + start, + 3 - start, + path.to_str().unwrap(), + false, + false, + ) + .await + .unwrap_err(); + assert!(format!("{error:#}").contains("does not extend header 1"), "{error:#}"); + assert_eq!(fs::metadata(&path).unwrap().len(), (start * 80) as u64); + } + } + #[test] fn genesis_batch_rotates_to_single_block_batches() { let mut args = Args::try_parse_from([ @@ -709,7 +774,7 @@ mod tests { let headers = vec![genesis.clone(), regtest, genesis]; let bytes = proof_block_headers(&headers).unwrap(); assert_eq!(bytes.len(), 3 * 80); - // Operator and watchtower proofs parse the sidecar 80 bytes at a time. + // Operator and watchtower proofs parse the headers 80 bytes at a time. let parsed: Vec<_> = bytes .chunks(80) .map(|header| CircuitBlockHeader::try_from_slice(header).unwrap()) diff --git a/circuits/operator-proof/host/src/lib.rs b/circuits/operator-proof/host/src/lib.rs index f7df0ddc0..46c632db4 100644 --- a/circuits/operator-proof/host/src/lib.rs +++ b/circuits/operator-proof/host/src/lib.rs @@ -3,7 +3,6 @@ use alloy_primitives::U256; use anyhow::Context; use bitcoin::{Block, BlockHash, Network, Transaction, Txid, hashes::Hash, secp256k1::PublicKey}; use bitcoin_light_client_circuit::{IndexedWatchtowerChallenge, build_spv}; -use borsh::BorshDeserialize; use clap::Parser; use client::btc_chain::BTCClient; use commit_chain::{CommitChainCircuitInput, CommitChainPrevProofType}; @@ -502,14 +501,9 @@ impl ProofBuilder for OperatorProofBuilder { let operator_genesis_sequencer_commit_txid = Txid::from_str(genesis_sequencer_commit_txid)?; - let bitcoin_block_headers = { - let headers: Vec = std::fs::read(format!("{header_chain_input_proof}.blocks")) - .context("read header chain blocks error")?; - headers - .chunks(80) - .map(|header| CircuitBlockHeader::try_from_slice(header).unwrap()) - .collect::>() - }; + let bitcoin_block_headers = + proof_builder::headers::proven_headers(&header_chain_input_proof) + .context("read the headers of the header-chain proof")?; let found = bitcoin_block_headers.iter().position(|h| { h.compute_block_hash() == *target_block_ss_commit.block_hash().as_byte_array() diff --git a/circuits/proof-builder/Cargo.toml b/circuits/proof-builder/Cargo.toml index 824718e98..74df8838e 100644 --- a/circuits/proof-builder/Cargo.toml +++ b/circuits/proof-builder/Cargo.toml @@ -11,6 +11,8 @@ serde.workspace = true serde_json.workspace = true secp256k1.workspace = true hex.workspace = true +bincode.workspace = true +borsh.workspace = true rand.workspace = true # Ziren diff --git a/circuits/proof-builder/src/headers.rs b/circuits/proof-builder/src/headers.rs new file mode 100644 index 000000000..2a2ff7fd8 --- /dev/null +++ b/circuits/proof-builder/src/headers.rs @@ -0,0 +1,309 @@ +//! The archive of proven Bitcoin headers. +//! +//! A header-chain proof covers every header from Genesis to its last block. The headers +//! are kept once, in [`HEADER_ARCHIVE`] next to the proofs, 80 bytes each and only ever +//! appended: a proof covering `n` headers reads the archive's first `n`, so readers need +//! no lock against the prover. + +use anyhow::{Context, Result, bail, ensure}; +use bincode::Options; +use borsh::{BorshDeserialize, BorshSerialize}; +use header_chain::{BlockHeaderCircuitOutput, CircuitBlockHeader}; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; + +pub const HEADER_ARCHIVE: &str = "headers.blocks"; +const HEADER_SIZE: u64 = 80; + +/// The archive next to `proof_path`. +pub fn header_archive_path(proof_path: impl AsRef) -> PathBuf { + match proof_path.as_ref().parent() { + Some(dir) if !dir.as_os_str().is_empty() => dir.join(HEADER_ARCHIVE), + _ => PathBuf::from(HEADER_ARCHIVE), + } +} + +fn header_bytes(header: &CircuitBlockHeader) -> Result> { + let mut bytes = Vec::with_capacity(HEADER_SIZE as usize); + header.serialize(&mut bytes)?; + ensure!(bytes.len() as u64 == HEADER_SIZE, "block headers must be 80 bytes each"); + Ok(bytes) +} + +fn read_header(file: &mut File, index: u64) -> Result { + let mut bytes = [0u8; HEADER_SIZE as usize]; + file.seek(SeekFrom::Start(index * HEADER_SIZE))?; + file.read_exact(&mut bytes)?; + Ok(CircuitBlockHeader::try_from_slice(&bytes)?) +} + +/// Extend the archive to `headers[..end]`, the headers the proof being saved covers. +/// Headers already archived must be the same ones, and every appended header must +/// extend the one before it. A torn tail from an interrupted append is dropped first. +pub fn extend_header_archive( + archive: impl AsRef, + headers: &[CircuitBlockHeader], + end: usize, +) -> Result<()> { + let archive = archive.as_ref(); + let headers = headers.get(..end).context("block headers do not cover the proven batch")?; + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(archive) + .with_context(|| format!("open {}", archive.display()))?; + let len = file.metadata()?.len(); + let archived = len / HEADER_SIZE; + if len % HEADER_SIZE != 0 { + file.set_len(archived * HEADER_SIZE)?; + } + // The archive is a hash chain, so one shared header equal on both sides means the + // whole shared prefix is. + let shared = archived.min(end as u64); + if shared > 0 && read_header(&mut file, shared - 1)? != headers[shared as usize - 1] { + bail!( + "{} holds another header at height {}; the proven chain changed", + archive.display(), + shared - 1 + ); + } + if archived >= end as u64 { + return Ok(()); + } + let mut previous = match archived { + 0 => None, + index => Some(read_header(&mut file, index - 1)?.compute_block_hash()), + }; + file.seek(SeekFrom::Start(archived * HEADER_SIZE))?; + let mut bytes = Vec::with_capacity((end - archived as usize) * HEADER_SIZE as usize); + for header in &headers[archived as usize..] { + if let Some(previous) = previous { + ensure!( + header.prev_block_hash == previous, + "header at height {} does not extend the archived chain", + archived as usize + bytes.len() / HEADER_SIZE as usize + ); + } + bytes.extend(header_bytes(header)?); + previous = Some(header.compute_block_hash()); + } + file.write_all(&bytes)?; + file.sync_all().with_context(|| format!("sync {}", archive.display())) +} + +fn decode_header_chain_output(public_values: &[u8]) -> Result { + // The encoding the header-chain guest commits its output with. + Ok(bincode::DefaultOptions::new() + .with_fixint_encoding() + .reject_trailing_bytes() + .deserialize(public_values)?) +} + +/// The headers a header-chain proof covers, from Genesis to its last block. They come +/// from the archive next to the proof, or from the proof's own `.blocks` sidecar (the +/// Genesis seed), and the last one must be the proof's best block. +pub fn proven_headers(proof_path: &str) -> Result> { + let public_values = fs::read(format!("{proof_path}.public_inputs.bin")) + .with_context(|| format!("read {proof_path}.public_inputs.bin"))?; + let chain_state = decode_header_chain_output(&public_values) + .with_context(|| format!("decode the header-chain output of {proof_path}"))? + .chain_state; + let count = u64::from(chain_state.block_height.wrapping_add(1)); + ensure!(count > 0, "{proof_path} covers no header"); + let bytes = count * HEADER_SIZE; + + let archive = header_archive_path(proof_path); + let source = match fs::metadata(&archive) { + Ok(metadata) if metadata.len() >= bytes => archive, + _ => PathBuf::from(format!("{proof_path}.blocks")), + }; + let mut raw = Vec::with_capacity(bytes as usize); + File::open(&source) + .with_context(|| format!("open {}", source.display()))? + .take(bytes) + .read_to_end(&mut raw)?; + ensure!( + raw.len() as u64 == bytes, + "{} holds {} headers, {proof_path} covers {count}", + source.display(), + raw.len() as u64 / HEADER_SIZE + ); + let headers = raw + .chunks_exact(HEADER_SIZE as usize) + .map(CircuitBlockHeader::try_from_slice) + .collect::, _>>()?; + let last = headers.last().map(CircuitBlockHeader::compute_block_hash); + ensure!( + last == Some(chain_state.best_block_hash), + "{} does not end at the best block of {proof_path}", + source.display() + ); + Ok(headers) +} + +/// Delete the per-proof `.blocks` sidecars older proofs were published with, once per +/// process: only those the archive in `dir` covers (their last header is the archive's +/// at the same height), so nothing else ending in `.blocks` is touched. Sidecars of +/// Genesis batches (`0-*`) stay: they seed a fresh archive. +pub fn remove_superseded_sidecars(dir: impl AsRef) -> Result { + static DONE: AtomicBool = AtomicBool::new(false); + if DONE.swap(true, Ordering::SeqCst) { + return Ok(0); + } + let dir = dir.as_ref(); + let mut archive = File::open(dir.join(HEADER_ARCHIVE))?; + let archived = archive.metadata()?.len() / HEADER_SIZE; + let mut removed = 0; + for entry in fs::read_dir(dir)? { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + if !name.ends_with(".bin.blocks") || name.starts_with("0-") || !entry.file_type()?.is_file() + { + continue; + } + let mut sidecar = File::open(entry.path())?; + let len = sidecar.metadata()?.len(); + let count = len / HEADER_SIZE; + if len == 0 || len % HEADER_SIZE != 0 || count > archived { + continue; + } + let covered = read_header(&mut sidecar, count - 1) + .and_then(|last| Ok(last == read_header(&mut archive, count - 1)?)) + .unwrap_or(false); + if covered { + fs::remove_file(entry.path())?; + removed += 1; + } + } + Ok(removed) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn chain(len: usize) -> Vec { + let genesis = bitcoin::constants::genesis_block(bitcoin::Network::Regtest).header; + let mut headers: Vec = vec![genesis.into()]; + while headers.len() < len { + let mut next = headers.last().unwrap().clone(); + next.prev_block_hash = next.compute_block_hash(); + next.nonce = next.nonce.wrapping_add(1); + headers.push(next); + } + headers + } + + #[test] + fn the_archive_grows_by_batch_and_refuses_a_changed_or_broken_chain() { + let dir = tempfile::tempdir().unwrap(); + let archive = dir.path().join(HEADER_ARCHIVE); + let headers = chain(6); + extend_header_archive(&archive, &headers, 4).unwrap(); + // A torn append and a repeated save are both harmless. + let mut file = OpenOptions::new().append(true).open(&archive).unwrap(); + file.write_all(&[1, 2, 3]).unwrap(); + extend_header_archive(&archive, &headers, 4).unwrap(); + extend_header_archive(&archive, &headers, 5).unwrap(); + assert_eq!(fs::metadata(&archive).unwrap().len(), 5 * HEADER_SIZE); + + let mut changed = headers.clone(); + changed[4].nonce += 7; + assert!(extend_header_archive(&archive, &changed, 5).is_err()); + let mut broken = headers.clone(); + broken[5].prev_block_hash = [9; 32]; + assert!(extend_header_archive(&archive, &broken, 6).is_err()); + assert_eq!(fs::metadata(&archive).unwrap().len(), 5 * HEADER_SIZE); + } + + fn write_proof(dir: &Path, name: &str, headers: &[CircuitBlockHeader]) -> String { + let mut chain_state = header_chain::ChainState::new(); + chain_state.block_height = headers.len() as u32 - 1; + chain_state.best_block_hash = headers.last().unwrap().compute_block_hash(); + let output = BlockHeaderCircuitOutput { + chain_state, + self_program_id: [0; 32], + program_history_hash: [0; 32], + upgrade_checkpoint_hash: [0; 32], + }; + let public_values = + bincode::DefaultOptions::new().with_fixint_encoding().serialize(&output).unwrap(); + let proof = dir.join(name).to_str().unwrap().to_string(); + fs::write(format!("{proof}.public_inputs.bin"), public_values).unwrap(); + proof + } + + #[test] + fn a_proof_reads_its_own_prefix_of_the_archive_or_its_seed_sidecar() { + let dir = tempfile::tempdir().unwrap(); + let headers = chain(5); + let seed = write_proof(dir.path(), "0-2.bin", &headers[..2]); + let mut sidecar = Vec::new(); + for header in &headers[..2] { + sidecar.extend(header_bytes(header).unwrap()); + } + fs::write(format!("{seed}.blocks"), sidecar).unwrap(); + // No archive yet: the seed's own sidecar. + assert_eq!(proven_headers(&seed).unwrap(), headers[..2]); + + let later = write_proof(dir.path(), "3-1.bin", &headers[..4]); + assert!(proven_headers(&later).is_err(), "neither archive nor sidecar covers it"); + extend_header_archive(header_archive_path(&later), &headers, 5).unwrap(); + assert_eq!(proven_headers(&later).unwrap(), headers[..4]); + assert_eq!(proven_headers(&seed).unwrap(), headers[..2]); + + // An archive of another chain is refused, not served. + let other = chain(4).into_iter().map(|mut h| { + h.time += 1; + h + }); + let stranger = write_proof(dir.path(), "9-1.bin", &other.collect::>()); + assert!(proven_headers(&stranger).is_err()); + } + + #[test] + fn only_sidecars_the_archive_covers_are_removed() { + let dir = tempfile::tempdir().unwrap(); + let headers = chain(4); + extend_header_archive(dir.path().join(HEADER_ARCHIVE), &headers, 3).unwrap(); + let bytes = |headers: &[CircuitBlockHeader]| -> Vec { + headers.iter().flat_map(|header| header_bytes(header).unwrap()).collect() + }; + let other = chain(2).into_iter().map(|mut h| { + h.time += 1; + h + }); + for (name, content) in [ + ("0-2.bin.blocks", bytes(&headers[..2])), + ("2-1.bin.blocks", bytes(&headers[..3])), + ("3-1.bin.blocks", bytes(&headers[..4])), + ("5-1.bin.blocks", bytes(&other.collect::>())), + ("7-40.bin.blocks", b"[{\"state\":1}]".to_vec()), + ("2-1.bin", b"proof".to_vec()), + ] { + fs::write(dir.path().join(name), content).unwrap(); + } + assert_eq!(remove_superseded_sidecars(dir.path()).unwrap(), 1); + let mut left: Vec<_> = fs::read_dir(dir.path()) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect(); + left.sort(); + assert_eq!( + left, + [ + "0-2.bin.blocks", + "2-1.bin", + "3-1.bin.blocks", + "5-1.bin.blocks", + "7-40.bin.blocks", + HEADER_ARCHIVE + ] + ); + } +} diff --git a/circuits/proof-builder/src/lib.rs b/circuits/proof-builder/src/lib.rs index 94656ee5e..dba7fafa9 100644 --- a/circuits/proof-builder/src/lib.rs +++ b/circuits/proof-builder/src/lib.rs @@ -13,6 +13,7 @@ use zkm_sdk::{Prover, ZKMProofKind, ZKMProvingKey, ZKMStdin, ZKMVerifyingKey}; pub mod api_auth; pub mod artifacts; +pub mod headers; #[derive(Debug, Clone, Serialize, Deserialize)] pub enum ProofRequest { diff --git a/circuits/watchtower-proof/host/src/lib.rs b/circuits/watchtower-proof/host/src/lib.rs index beb63bb96..1e02f7c03 100644 --- a/circuits/watchtower-proof/host/src/lib.rs +++ b/circuits/watchtower-proof/host/src/lib.rs @@ -1,9 +1,8 @@ #![feature(trim_prefix_suffix)] //! Generate watchtower proof use anyhow::Context; -use borsh::BorshDeserialize; use commit_chain::{CommitChainCircuitInput, CommitChainPrevProofType}; -use header_chain::{CircuitBlockHeader, HeaderChainCircuitInput, HeaderChainPrevProofType}; +use header_chain::{HeaderChainCircuitInput, HeaderChainPrevProofType}; use zkm_sdk::{ HashableKey, ProverClient, ZKMProofKind, ZKMProofWithPublicValues, ZKMStdin, include_elf, }; @@ -284,13 +283,9 @@ impl ProofBuilder for WatchtowerProofBuilder { }; // --- spv --- // let genesis_sequencer_commit_txid = Txid::from_str(genesis_sequencer_commit_txid)?; - let bitcoin_block_headers = { - let headers: Vec = std::fs::read(format!("{header_chain_input_proof}.blocks"))?; - headers - .chunks(80) - .map(|header| CircuitBlockHeader::try_from_slice(header).unwrap()) - .collect::>() - }; + let bitcoin_block_headers = + proof_builder::headers::proven_headers(&header_chain_input_proof) + .context("read the headers of the header-chain proof")?; tracing::info!("block headers: {:?}", bitcoin_block_headers.len()); let found = bitcoin_block_headers .iter() From aa37d65da66f438e0a29b5b86c858f974dc9722b Mon Sep 17 00:00:00 2001 From: Ethan Date: Wed, 11 Nov 2026 07:40:26 +0000 Subject: [PATCH 2/3] fix: resume the commit chain from stored proofs on restart --- .../src/task/commit_chain_proof.rs | 52 +++++++++---------- 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/proof-builder-rpc/src/task/commit_chain_proof.rs b/proof-builder-rpc/src/task/commit_chain_proof.rs index 358296710..ed7368486 100644 --- a/proof-builder-rpc/src/task/commit_chain_proof.rs +++ b/proof-builder-rpc/src/task/commit_chain_proof.rs @@ -38,33 +38,33 @@ pub(crate) fn spawn_commit_chain_proof_task( loop { tokio::select! { _ = tokio::time::sleep(Duration::from_secs(interval)) => { - let init_input = args.init_input; - if !init_input { - let next_task = fetch_latest_long_running_task(&local_db, CommitChainProofBuilder::name()).await?; - if let Some(next_task) = next_task { - info!("Commit chain's next task: {next_task:?}"); - args.start = fetch_next_commit_task_index(&local_db).await?; - args.input_proof = next_task.path_to_proof.unwrap(); - args.commit_info = format!( - "{}/commit_info.json.{}", - std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), - args.start, - ); - args.output_proof = format!( - "{}/{}-{}.bin", - std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), - args.start, - args.batch_size - ); - args.commits = format!( - "{}/{}-{}.bin.commits", - std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), - args.start, - args.batch_size - ); - args.init_input = false; - } + // Stored progress wins over `init_input`, as for the header chain: a + // restart must not prove from Genesis again and replace the chain. + let next_task = fetch_latest_long_running_task(&local_db, CommitChainProofBuilder::name()).await?; + if let Some(next_task) = next_task { + info!("Commit chain's next task: {next_task:?}"); + args.start = fetch_next_commit_task_index(&local_db).await?; + args.input_proof = next_task.path_to_proof.unwrap(); + args.commit_info = format!( + "{}/commit_info.json.{}", + std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), + args.start, + ); + args.output_proof = format!( + "{}/{}-{}.bin", + std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), + args.start, + args.batch_size + ); + args.commits = format!( + "{}/{}-{}.bin.commits", + std::path::Path::new(&args.output_proof).parent().unwrap().to_str().unwrap(), + args.start, + args.batch_size + ); + args.init_input = false; } + let init_input = args.init_input; info!("Commit chain proof generate task: generate proof, args: {args:?}"); let commits = match fetch_commit_chain(&args.esplora_url, &args.commit_info, &args.commits, args.bitcoin_network).await { From e6306dbc65435d79c0fa04d490c2a3b1a0cebd3c Mon Sep 17 00:00:00 2001 From: Ethan Date: Wed, 11 Nov 2026 07:40:26 +0000 Subject: [PATCH 3/3] fix: reset a state-chain batch that keeps failing and alert once resets run out --- crates/store/src/localdb/proof.rs | 19 +++ crates/store/src/localdb/tests.rs | 9 ++ .../src/task/state_chain_proof.rs | 119 +++++++++++++++++- 3 files changed, 146 insertions(+), 1 deletion(-) diff --git a/crates/store/src/localdb/proof.rs b/crates/store/src/localdb/proof.rs index 18e9f81c7..03952fabe 100644 --- a/crates/store/src/localdb/proof.rs +++ b/crates/store/src/localdb/proof.rs @@ -674,6 +674,25 @@ impl StorageProcessor<'_> { Ok(state) } + /// Deletes the unproven tasks of one proof chain starting at or after `from_block` and + /// returns their starts. + pub async fn delete_unproven_long_running_task_proofs( + &mut self, + chain_name: &str, + from_block: i64, + ) -> anyhow::Result> { + Ok(sqlx::query_scalar::<_, i64>( + "DELETE FROM long_running_task_proof + WHERE chain_name = ? AND block_start >= ? AND proof_state != ? + RETURNING block_start", + ) + .bind(chain_name) + .bind(from_block) + .bind(crate::ProofState::Proven.to_i64()) + .fetch_all(self.conn()) + .await?) + } + /// Counts unfinished (`New` or `Proving`) tasks of one proof chain starting at or after /// `from_block`. pub async fn count_pending_long_running_task_proofs( diff --git a/crates/store/src/localdb/tests.rs b/crates/store/src/localdb/tests.rs index 07950a40c..e98011010 100644 --- a/crates/store/src/localdb/tests.rs +++ b/crates/store/src/localdb/tests.rs @@ -555,6 +555,15 @@ async fn test_count_pending_long_running_task_proofs() { assert_eq!(s.count_pending_long_running_task_proofs(&chain, 20).await.unwrap(), 2); assert_eq!(s.count_pending_long_running_task_proofs(&chain, 0).await.unwrap(), 3); assert_eq!(s.count_pending_long_running_task_proofs("header-chain", 0).await.unwrap(), 0); + + // A batch reset drops every unproven batch from its start, of its own chain only. + let mut deleted = s.delete_unproven_long_running_task_proofs(&chain, 10).await.unwrap(); + deleted.sort(); + assert_eq!(deleted, [20, 30, 40]); + assert_eq!(s.count_pending_long_running_task_proofs(&chain, 0).await.unwrap(), 1); + assert!( + s.delete_unproven_long_running_task_proofs("header-chain", 0).await.unwrap().is_empty() + ); } #[tokio::test] diff --git a/proof-builder-rpc/src/task/state_chain_proof.rs b/proof-builder-rpc/src/task/state_chain_proof.rs index b9a13092f..251600f8f 100644 --- a/proof-builder-rpc/src/task/state_chain_proof.rs +++ b/proof-builder-rpc/src/task/state_chain_proof.rs @@ -1,4 +1,5 @@ use crate::prover::{Cancelled, ProverRuntime}; +use crate::prover::{Failure, ProveError}; use crate::task::ProofState::{Failed, New, Proven, Proving}; use crate::task::fetch_latest_long_running_task_by_state; use crate::task::{ @@ -8,13 +9,51 @@ use crate::task::{ use crate::{ProofBuilderConfig, task::fetch_latest_long_running_task}; use proof_builder::{ProofBuilder, ProofRequest}; use state_chain_proof::{StateChainProofBuilder, fetch_state_chain}; +use std::sync::Arc; use std::time::Duration; use store::localdb::LocalDB; +use tokio::sync::Mutex; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::api::metrics_service::{ApiMetricsState, STATE_CHAIN_PROOF}; +/// Consecutive fatal failures of one batch before its snapshot is dropped and fetched +/// again: the configured RPC may have served wrong block data. +const FATAL_RUNS_PER_BATCH: u32 = 3; +/// Resets of one batch before the prover gives up on it and only alerts. +const RESETS_PER_BATCH: u32 = 3; + +/// Held by the ctx builder for each batch it prepares and by the prover for a reset, +/// so a reset never races a batch row the builder is about to create. +type BatchLock = Arc>; + +/// Drop every unproven batch from `start` with its snapshot; the ctx builder then +/// prepares them again from the last proven batch. +async fn reset_state_chain_batches( + local_db: &LocalDB, + snap_path: &str, + start: u64, +) -> anyhow::Result { + let starts = local_db + .acquire() + .await? + .delete_unproven_long_running_task_proofs(&StateChainProofBuilder::name(), start as i64) + .await?; + for batch in &starts { + for suffix in ["args", "ctx"] { + let path = format!("{snap_path}/{batch}.{suffix}"); + match std::fs::remove_file(&path) { + Err(error) if error.kind() != std::io::ErrorKind::NotFound => { + return Err(error.into()); + } + _ => {} + } + } + } + Ok(starts.len()) +} + /// Whether the prover already has `max_pending` unproved batches waiting. async fn state_chain_backlog_full(local_db: &LocalDB, max_pending: i64) -> anyhow::Result { if max_pending <= 0 { @@ -42,13 +81,16 @@ async fn spawn_state_chain_ctx_builder( local_db: LocalDB, init_interval: u64, max_pending: i64, + batch_lock: BatchLock, cancellation_token: CancellationToken, ) -> anyhow::Result<()> { + let configured = args.clone(); let mut args = args; let mut interval = init_interval; loop { tokio::select! { _ = tokio::time::sleep(Duration::from_secs(interval)) => { + let _batch = batch_lock.lock().await; if state_chain_backlog_full(&local_db, max_pending).await? { tracing::debug!("ctx builder: {max_pending} batches await the prover"); continue; @@ -75,6 +117,9 @@ async fn spawn_state_chain_ctx_builder( ); args.init_input = false; } + } else { + // Nothing stored, or a reset dropped every batch: start over. + args = configured.clone(); } tracing::info!("ctx builder: fetching blocks, args={args:?}"); @@ -163,17 +208,37 @@ async fn spawn_state_chain_prover( local_db: LocalDB, metrics_state: ApiMetricsState, interval: u64, + batch_lock: BatchLock, cancellation_token: CancellationToken, ) -> anyhow::Result<()> { let builder = runtime.builder(StateChainProofBuilder::with_client).await?; let mut start_index = start_index; + // Of the batch at `start_index`. + let (mut fatal_runs, mut resets) = (0, 0); loop { tokio::select! { _ = tokio::time::sleep(Duration::from_secs(interval)) => { + let snap_path = std::path::Path::new(&input_proof).parent().unwrap().to_str().unwrap(); + if fatal_runs >= FATAL_RUNS_PER_BATCH && resets >= RESETS_PER_BATCH { + if fatal_runs == FATAL_RUNS_PER_BATCH { + fatal_runs += 1; + tracing::error!(block_start = start_index, resets, "state chain batch keeps failing after fresh fetches; proving stopped until restart"); + } + metrics_state.record_attempt(STATE_CHAIN_PROOF, "stalled", Duration::ZERO); + continue; + } + if fatal_runs >= FATAL_RUNS_PER_BATCH { + let _batch = batch_lock.lock().await; + let dropped = reset_state_chain_batches(&local_db, snap_path, start_index).await?; + resets += 1; + fatal_runs = 0; + tracing::error!(block_start = start_index, dropped, resets, "state chain batch failed {FATAL_RUNS_PER_BATCH} times; fetching it again"); + metrics_state.record_attempt(STATE_CHAIN_PROOF, "batch_reset", Duration::ZERO); + continue; + } tracing::info!("prover: start proving, block_start: {start_index}, batch_size: {batch_size}"); - let snap_path = std::path::Path::new(&input_proof).parent().unwrap().to_str().unwrap(); let args: state_chain_proof::Args = match std::fs::read(format!("{}/{}.args", snap_path, start_index)) { Ok(x) => match serde_json::from_slice(&x) { Ok(args) => args, @@ -259,6 +324,10 @@ async fn spawn_state_chain_prover( } Ok(Ok(Err(error))) => { metrics_state.record_attempt(STATE_CHAIN_PROOF, failure_outcome(&error), attempt_start.elapsed()); + // Only fatal failures count against the snapshot. + if matches!(error, ProveError::Prove(_, Failure::Fatal)) { + fatal_runs += 1; + } settle_chain_failure(&runtime, error, &StateChainProofBuilder::name())?; continue; } @@ -268,6 +337,7 @@ async fn spawn_state_chain_prover( metrics_state.record_attempt(STATE_CHAIN_PROOF, outcome, attempt_start.elapsed()); tracing::info!("Updated long running task to proven state, affected rows: {affected}"); start_index += batch_size; + (fatal_runs, resets) = (0, 0); } _ = cancellation_token.cancelled() => { anyhow::bail!("prover cancelled"); @@ -289,11 +359,13 @@ pub(crate) fn spawn_state_chain_proof_task( tokio::spawn(async move { tokio::time::sleep(Duration::from_secs(initial_delay)).await; + let batch_lock = BatchLock::default(); let ctx_builder = tokio::spawn(spawn_state_chain_ctx_builder( args.clone(), local_db.clone(), interval + 1, options.state_chain_max_pending_batches, + batch_lock.clone(), cancellation_token.clone(), )); @@ -346,6 +418,7 @@ pub(crate) fn spawn_state_chain_proof_task( local_db.clone(), metrics_state, interval, + batch_lock, cancellation_token.clone(), )); @@ -364,3 +437,47 @@ pub(crate) fn spawn_state_chain_proof_task( Ok(()) }) } + +#[cfg(test)] +mod reset_tests { + use super::*; + use store::{LongRunningTaskProof, create_local_db}; + + #[tokio::test] + async fn a_reset_drops_the_unproven_batches_and_their_snapshots() { + let local_db = create_local_db("sqlite::memory:").await; + let dir = tempfile::tempdir().unwrap(); + let snap = dir.path().to_str().unwrap(); + let mut storage = local_db.acquire().await.unwrap(); + for (start, state) in [(0, Proven), (10, Proving), (20, New)] { + storage + .create_long_running_task_proof(&LongRunningTaskProof { + block_start: start, + block_end: start + 10, + chain_name: StateChainProofBuilder::name(), + proof_state: state.to_i64(), + ..Default::default() + }) + .await + .unwrap(); + for suffix in ["args", "ctx"] { + std::fs::write(format!("{snap}/{start}.{suffix}"), b"{}").unwrap(); + } + } + drop(storage); + + assert_eq!(reset_state_chain_batches(&local_db, snap, 10).await.unwrap(), 2); + // The builder restarts after the proven batch. + let latest = fetch_latest_long_running_task(&local_db, StateChainProofBuilder::name()) + .await + .unwrap() + .unwrap(); + assert_eq!((latest.block_start, latest.block_end), (0, 10)); + let mut left: Vec<_> = std::fs::read_dir(snap) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect(); + left.sort(); + assert_eq!(left, ["0.args", "0.ctx"]); + } +}