From 6927087042443d354147909554c5cf70c4411c5c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sun, 11 Oct 2026 02:02:58 +0000 Subject: [PATCH] feat: integrate the Ziren 2.0 EthProofs client into a durable Fleet worker --- Cargo.lock | 58 ++-- Cargo.toml | 1 + bin/eth-proofs/Cargo.toml | 12 +- bin/eth-proofs/build.rs | 11 +- bin/eth-proofs/src/cli.rs | 48 ++- bin/eth-proofs/src/eth_proofs.rs | 189 ++++++++++- bin/eth-proofs/src/main.rs | 383 ++++++++++++++++++++-- bin/eth-proofs/src/state.rs | 330 +++++++++++++++++++ crates/executor/host/src/full_executor.rs | 32 +- 9 files changed, 980 insertions(+), 84 deletions(-) create mode 100644 bin/eth-proofs/src/state.rs diff --git a/Cargo.lock b/Cargo.lock index b2c0abb..93c284e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2481,12 +2481,16 @@ dependencies = [ "alloy-transport", "alloy-transport-ws", "base64 0.22.1", + "bincode", "clap", "dotenv", "eyre", + "fs2", "futures", "guest-executor", + "hex", "host-executor", + "libc", "primitives", "provider", "reqwest 0.12.25", @@ -2494,12 +2498,16 @@ dependencies = [ "reqwest-retry", "serde", "serde_json", + "sha2", + "tempfile", "time", "tokio", "tracing", "tracing-subscriber", "url", "zkm-build", + "zkm-cuda", + "zkm-prover", "zkm-sdk", ] @@ -2678,6 +2686,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "funty" version = "2.0.0" @@ -3390,7 +3408,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2 0.6.1", - "system-configuration 0.6.1", + "system-configuration", "tokio", "tower-service", "tracing", @@ -3830,11 +3848,10 @@ checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de" [[package]] name = "libredox" -version = "0.1.10" +version = "0.1.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb" +checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d" dependencies = [ - "bitflags 2.10.0", "libc", ] @@ -4125,9 +4142,9 @@ dependencies = [ [[package]] name = "ntapi" -version = "0.4.1" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8a3895c6391c39d7fe7ebc444a87eb2991b2a0bc718fdabd071eec617fc68e4" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" dependencies = [ "winapi", ] @@ -5119,9 +5136,9 @@ checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483" [[package]] name = "portable-atomic-util" -version = "0.2.4" +version = "0.2.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8a2f0d8d040d7848a709caf78912debcc3f33ee4b3cac47d73d1e1069e83507" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" dependencies = [ "portable-atomic", ] @@ -5710,7 +5727,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "sync_wrapper 0.1.2", - "system-configuration 0.5.1", + "system-configuration", "tokio", "tokio-rustls 0.24.1", "tokio-util", @@ -7436,18 +7453,7 @@ checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7" dependencies = [ "bitflags 1.3.2", "core-foundation", - "system-configuration-sys 0.5.0", -] - -[[package]] -name = "system-configuration" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" -dependencies = [ - "bitflags 2.10.0", - "core-foundation", - "system-configuration-sys 0.6.0", + "system-configuration-sys", ] [[package]] @@ -7460,16 +7466,6 @@ dependencies = [ "libc", ] -[[package]] -name = "system-configuration-sys" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "tap" version = "1.0.1" diff --git a/Cargo.toml b/Cargo.toml index 1f7fb48..8c9332b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,6 +50,7 @@ provider = { path = "./crates/provider" } zkm-build = { git = "https://github.com/ProjectZKM/Ziren" } zkm-core-executor = { git = "https://github.com/ProjectZKM/Ziren" } zkm-prover = { git = "https://github.com/ProjectZKM/Ziren" } +zkm-cuda = { git = "https://github.com/ProjectZKM/Ziren" } zkm-sdk = { git = "https://github.com/ProjectZKM/Ziren" } # reth diff --git a/bin/eth-proofs/Cargo.toml b/bin/eth-proofs/Cargo.toml index bd8a843..99b5c92 100644 --- a/bin/eth-proofs/Cargo.toml +++ b/bin/eth-proofs/Cargo.toml @@ -4,7 +4,7 @@ name = "eth-proofs" edition = "2021" [dependencies] -tokio.workspace = true +tokio = { workspace = true, features = ["signal"] } url.workspace = true tracing.workspace = true serde_json.workspace = true @@ -13,6 +13,11 @@ tracing-subscriber = "0.3.18" dotenv = "0.15.0" clap = { version = "4.5.7", features = ["derive", "env"] } eyre = "0.6.12" +bincode = "1.3.3" +sha2 = "0.10.8" +hex = "0.4.3" +fs2 = "0.4.3" +libc = "0.2" reqwest = { workspace = true, features = ["json"] } reqwest-middleware = { version = "0.3.3", features = ["json"] } reqwest-retry = "0.6.1" @@ -28,6 +33,8 @@ provider.workspace = true # Ziren zkm-sdk.workspace = true +zkm-cuda.workspace = true +zkm-prover.workspace = true # alloy alloy-chains.workspace = true @@ -41,3 +48,6 @@ zkm-build.workspace = true [features] test = [] + +[dev-dependencies] +tempfile = "3" diff --git a/bin/eth-proofs/build.rs b/bin/eth-proofs/build.rs index dcfa0c5..5029151 100644 --- a/bin/eth-proofs/build.rs +++ b/bin/eth-proofs/build.rs @@ -1,5 +1,12 @@ -use zkm_build::build_program; +use zkm_build::{build_program_with_args, BuildArgs}; fn main() { - build_program("../guest"); + println!("cargo:rerun-if-env-changed=ETH_PROOFS_BUILD_ELF"); + if let Ok(path) = std::env::var("ETH_PROOFS_BUILD_ELF") { + let path = std::fs::canonicalize(path).expect("ETH_PROOFS_BUILD_ELF must exist"); + println!("cargo:rerun-if-changed={}", path.display()); + println!("cargo:rustc-env=ZKM_ELF_reth={}", path.display()); + } else { + build_program_with_args("../guest", BuildArgs { locked: true, ..Default::default() }); + } } diff --git a/bin/eth-proofs/src/cli.rs b/bin/eth-proofs/src/cli.rs index 03ef128..ba5dbd1 100644 --- a/bin/eth-proofs/src/cli.rs +++ b/bin/eth-proofs/src/cli.rs @@ -14,7 +14,7 @@ pub struct Args { /// The WS rpc url used to fetch data about the block. #[clap(long, env)] - pub ws_rpc_url: Url, + pub ws_rpc_url: Option, /// The debug HTTP rpc url used to fetch data about the block trace. #[clap(long, env)] @@ -28,6 +28,24 @@ pub struct Args { #[clap(long, default_value_t = 100)] pub block_interval: u64, + /// Which residue class of `block_interval` this instance proves: blocks + /// with `number % block_interval == block_residue`. Two instances with + /// `--block-interval 2` and residues 0 and 1 split the chain between two + /// provers and prove two blocks at a time. + #[clap(long, default_value_t = 0)] + pub block_residue: u64, + + /// How many prepared blocks (fetched + natively executed) may wait for the + /// prover. The fetch of block N+1 always overlaps the proof of block N; + /// this bounds how far ahead the fetcher runs. + #[clap(long, default_value_t = 1)] + pub prefetch_depth: usize, + + /// Skip a block when it is more than this many blocks behind the newest + /// header seen (0 = never skip: prove every block the stream delivers). + #[clap(long, default_value_t = 0)] + pub max_lag: u64, + /// ETH proofs endpoint. #[clap(long, env)] pub eth_proofs_endpoint: String, @@ -44,6 +62,34 @@ pub struct Args { #[clap(long, env)] pub pager_duty_integration_key: Option, + /// Direct local GPU server, set by Fleet for an integrated app worker. + #[clap(long, env = "FLEET_GPU_ENDPOINT")] + pub local_gpu_endpoint: Option, + + /// Persistent local worker state. Required with --local-gpu-endpoint. + #[clap(long, env = "FLEET_APP_STORAGE_ROOT", requires = "local_gpu_endpoint")] + pub state_dir: Option, + + /// Use an existing production guest instead of the embedded ELF. + #[clap(long, env = "ETH_PROOFS_ELF_FILE")] + pub elf_file: Option, + + /// Reject a guest whose digest differs from this app's configured program. + #[clap(long, env = "FLEET_APP_ELF_ID")] + pub expected_elf_id: Option, + + /// First block for a new durable stream; restarts resume persisted progress. + #[clap(long)] + pub start_block: Option, + + /// Witness preparation concurrency (currently 1; GPU proving remains serial). + #[clap(long, default_value_t = 1)] + pub prepare_concurrency: usize, + + /// Pause proving when this many completed reports await acknowledgment. + #[clap(long, default_value_t = 128)] + pub max_pending_reports: usize, + /// Moongate server endpoint. #[clap(long, env)] pub moongate_endpoint: Option, diff --git a/bin/eth-proofs/src/eth_proofs.rs b/bin/eth-proofs/src/eth_proofs.rs index 327437d..ab1aa05 100644 --- a/bin/eth-proofs/src/eth_proofs.rs +++ b/bin/eth-proofs/src/eth_proofs.rs @@ -1,29 +1,80 @@ -use std::time::Duration; +use crate::state::State; +use std::{sync::Arc, time::Duration}; use base64::{engine::general_purpose::STANDARD, Engine}; use eyre::eyre; use host_executor::ExecutionHooks; use reqwest_middleware::{ClientBuilder, ClientWithMiddleware}; use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware}; -use tracing::error; +use tracing::{error, info}; use zkm_sdk::{HashableKey, ZKMVerifyingKey}; -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct EthProofsClient { cluster_id: u64, endpoint: String, api_token: String, client: ClientWithMiddleware, + state: Option>, } impl EthProofsClient { pub fn new(cluster_id: u64, endpoint: String, api_token: String) -> Self { let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3); - let client = ClientBuilder::new(reqwest::Client::new()) - .with(RetryTransientMiddleware::new_with_policy(retry_policy)) - .build(); + let client = ClientBuilder::new( + reqwest::Client::builder().timeout(Duration::from_secs(60)).build().unwrap(), + ) + .with(RetryTransientMiddleware::new_with_policy(retry_policy)) + .build(); - Self { cluster_id, endpoint, api_token, client } + Self { cluster_id, endpoint, api_token, client, state: None } + } + + pub fn with_state(mut self, state: Arc) -> Self { + self.state = Some(state); + self + } + + pub fn start_reporter(&self) -> tokio::task::JoinHandle<()> { + let this = self.clone(); + tokio::spawn(async move { + let Some(state) = &this.state else { + return; + }; + loop { + match state.pending() { + Ok(pending) => { + for (block, payload) in pending { + let result = this + .client + .post(format!("{}/proofs/proved", this.endpoint)) + .header("Authorization", format!("Bearer {}", this.api_token)) + .json(&payload) + .send() + .await; + match result { + Ok(response) if response.status().is_success() => { + if let Err(err) = state.acknowledge(block) { + error!(block, %err, "report accepted but acknowledgment persistence failed"); + } else { + info!(block, "durable EthProofs report acknowledged"); + } + } + Ok(response) => { + error!(block, status = %response.status(), "EthProofs report remains pending") + } + Err(err) => error!(block, %err, "EthProofs report remains pending"), + } + } + } + Err(err) => error!(%err, "reading persistent reporting outbox failed"), + } + tokio::select! { + _ = state.report_ready() => {}, + _ = tokio::time::sleep(Duration::from_secs(3)) => {}, + } + } + }) } pub async fn queued(&self, block_number: u64) { @@ -113,8 +164,12 @@ impl EthProofsClient { .map_err(|e| eyre!(e)) .and_then(|r| r.error_for_status().map_err(|e| eyre!(e))); - if let Err(err) = response { - error!("Failed to report proof proving: {}", err) + match response { + Ok(_) => info!( + "Reported proved block {} to eth-proofs: {} cycles, {:.1} s", + block_number, cycles, elapsed + ), + Err(err) => error!("Failed to report proof proved: {}", err), } }); } @@ -145,9 +200,121 @@ impl ExecutionHooks for EthProofsClient { cycles: u64, proving_duration: Duration, ) -> eyre::Result<()> { - #[cfg(not(feature = "test"))] - self.proved(proof_bytes, block_number, cycles, proving_duration.as_secs_f32(), vk).await; + if let Some(state) = &self.state { + let payload = serde_json::json!({ + "proof": STANDARD.encode(proof_bytes), "block_number": block_number, + "proving_cycles": cycles, "proving_time": proving_duration.as_millis() as u64, + "verifier_id": vk.bytes32(), "cluster_id": self.cluster_id, + }); + // Commit the output and its retryable report before the cursor advances. + state.commit(block_number, proof_bytes, _public_values_bytes, payload)?; + } else { + #[cfg(not(feature = "test"))] + self.proved(proof_bytes, block_number, cycles, proving_duration.as_secs_f32(), vk) + .await; + } Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::state::Identity; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[tokio::test] + async fn report_is_retained_until_http_success() { + let root = tempfile::tempdir().unwrap(); + let state = Arc::new( + State::open( + root.path(), + Identity { + namespace: "test".into(), + tenant: "ethproofs".into(), + chain_id: 1, + cluster_id: 8, + elf_id: "elf".into(), + interval: 1, + residue: 0, + }, + Some(10), + ) + .unwrap(), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = EthProofsClient::new( + 8, + format!("http://{}", listener.local_addr().unwrap()), + "test".into(), + ) + .with_state(state.clone()); + let reporter = client.start_reporter(); + // Let the reporter enter its idle wait, then commit a new output. + tokio::time::sleep(Duration::from_millis(50)).await; + state.commit(10, b"proof", b"values", serde_json::json!({"block_number":10})).unwrap(); + for status in ["401 Unauthorized", "200 OK"] { + let timeout = if status.starts_with("401") { 1 } else { 10 }; + let (mut socket, _) = + tokio::time::timeout(Duration::from_secs(timeout), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut request = Vec::new(); + loop { + let mut bytes = [0; 2048]; + let n = socket.read(&mut bytes).await.unwrap(); + request.extend_from_slice(&bytes[..n]); + if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&request[..end]).to_ascii_lowercase(); + let length = headers + .lines() + .find_map(|line| line.strip_prefix("content-length: ")) + .unwrap() + .parse::() + .unwrap(); + if request.len() >= end + 4 + length { + break; + } + } + assert!(n > 0); + } + assert!(String::from_utf8_lossy(&request).starts_with("POST /proofs/proved")); + assert_eq!(state.pending_count().unwrap(), 1); + socket + .write_all( + format!("HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .as_bytes(), + ) + .await + .unwrap(); + } + tokio::time::timeout(Duration::from_secs(5), async { + while state.pending_count().unwrap() != 0 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + reporter.abort(); + let _ = reporter.await; + drop(client); + drop(state); + let restored = State::open( + root.path(), + Identity { + namespace: "test".into(), + tenant: "ethproofs".into(), + chain_id: 1, + cluster_id: 8, + elf_id: "elf".into(), + interval: 1, + residue: 0, + }, + None, + ) + .unwrap(); + assert_eq!(restored.pending_count().unwrap(), 0); + } +} diff --git a/bin/eth-proofs/src/main.rs b/bin/eth-proofs/src/main.rs index 688a05b..ceefc59 100644 --- a/bin/eth-proofs/src/main.rs +++ b/bin/eth-proofs/src/main.rs @@ -1,25 +1,32 @@ -use std::{env, sync::Arc}; +use sha2::{Digest, Sha256}; +use state::{Identity, State}; +use std::{env, sync::Arc, time::Duration}; +use zkm_cuda::ZKMGpuServer; +use zkm_prover::ZKMProver; +use zkm_sdk::provers::CudaProver; use alloy_provider::{Provider, ProviderBuilder, WsConnect}; use clap::Parser; use cli::Args; use eth_proofs::EthProofsClient; -use futures::{future::ready, StreamExt}; +use futures::{channel::mpsc, future::ready, FutureExt, SinkExt, StreamExt}; use host_executor::{ - alerting::AlertingClient, create_eth_block_execution_strategy_factory, BlockExecutor, - EthExecutorComponents, FullExecutor, + alerting::AlertingClient, create_eth_block_execution_strategy_factory, EthExecutorComponents, + FullExecutor, }; use provider::create_provider; -use tracing::{error, info}; +use tracing::{error, info, warn}; use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; use zkm_sdk::{include_elf, ProverClient}; mod cli; +mod state; mod eth_proofs; #[tokio::main] async fn main() -> eyre::Result<()> { + exit_with_parent()?; // Initialize the environment variables. dotenv::dotenv().ok(); @@ -43,7 +50,56 @@ async fn main() -> eyre::Result<()> { let args = Args::parse(); let config = args.as_config().await?; - let elf = include_elf!("reth").to_vec(); + eyre::ensure!(args.block_interval > 0, "block interval must be positive"); + eyre::ensure!(args.max_pending_reports > 0, "pending report limit must be positive"); + eyre::ensure!( + args.prepare_concurrency == 1 && args.prefetch_depth > 0, + "this migration starts with serial preparation and a positive queue depth" + ); + let elf = match &args.elf_file { + Some(path) => std::fs::read(path)?, + None => include_elf!("reth").to_vec(), + }; + let elf_id = hex::encode(Sha256::digest(&elf)); + if let Some(expected) = &args.expected_elf_id { + eyre::ensure!(*expected == elf_id, "guest ELF differs from the app's configured digest"); + } + let state = if let Some(endpoint) = &args.local_gpu_endpoint { + eyre::ensure!( + endpoint.scheme() == "http" && + matches!(endpoint.host_str(), Some("127.0.0.1" | "localhost" | "[::1]")), + "local GPU endpoint must be loopback HTTP" + ); + eyre::ensure!( + !args.execute_only && args.max_lag == 0, + "durable local mode requires proving and never skips lagging blocks" + ); + let root = args + .state_dir + .as_ref() + .ok_or_else(|| eyre::eyre!("local proving requires --state-dir"))?; + eyre::ensure!(args.expected_elf_id.is_some(), "local proving requires --expected-elf-id"); + let state = Arc::new(State::open( + root, + Identity { + namespace: std::env::var("FLEET_APP_NAMESPACE")?, + tenant: std::env::var("FLEET_APP_TENANT")?, + chain_id: 1, + cluster_id: args.eth_proofs_cluster_id, + elf_id: elf_id.clone(), + interval: args.block_interval, + residue: args.block_residue % args.block_interval, + }, + args.start_block, + )?); + // FullExecutor's verification-key file belongs to this replica's storage. + std::env::set_current_dir(std::fs::canonicalize(root)?)?; + Some(state) + } else { + eyre::ensure!(args.state_dir.is_none(), "durable worker state requires local proving"); + None + }; + info!(%elf_id, local = state.is_some(), "starting EthProofs app pipeline"); let block_execution_strategy_factory = create_eth_block_execution_strategy_factory(&config.genesis, None); @@ -52,25 +108,102 @@ async fn main() -> eyre::Result<()> { args.eth_proofs_endpoint, args.eth_proofs_api_token, ); + let eth_proofs_client = match &state { + Some(state) => eth_proofs_client.with_state(state.clone()), + None => eth_proofs_client, + }; + let reporter = eth_proofs_client.start_reporter(); + let (shutdown_tx, mut shutdown) = tokio::sync::watch::channel(false); + if state.is_some() { + tokio::spawn(async move { + #[cfg(unix)] + { + let mut terminate = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .unwrap(); + tokio::select! { _ = terminate.recv() => {}, _ = tokio::signal::ctrl_c() => {} } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } + info!("draining EthProofs worker; finish the current proof and persist its cursor"); + let _ = shutdown_tx.send(true); + }); + } + let alerting_client = args.pager_duty_integration_key.map(AlertingClient::new); - let ws = WsConnect::new(args.ws_rpc_url); - #[allow(deprecated)] - let ws_provider = ProviderBuilder::new().on_ws(ws).await?; let http_provider = create_provider(args.http_rpc_url); let debug_http_provider = create_provider(args.debug_http_rpc_url); - - // Subscribe to block headers. - let subscription = ws_provider.subscribe_blocks().await?; - let mut stream = - subscription.into_stream().filter(|h| ready(h.number % args.block_interval == 0)); - - // let mut builder = ProverClient::builder().cuda(); - if let Some(_endpoint) = &args.moongate_endpoint { - // builder = builder.with_moongate_endpoint(endpoint) - } - - let client = Arc::new(ProverClient::new()); + eyre::ensure!( + http_provider.get_chain_id().await? == 1, + "EthProofs worker requires Ethereum mainnet RPC" + ); + let block_interval = args.block_interval; + let block_residue = args.block_residue % block_interval; + // Local mode uses HTTP head polling: a WS disconnect cannot discard the cursor. + let ws_provider = if state.is_none() { + #[allow(deprecated)] + let provider = ProviderBuilder::new() + .on_ws(WsConnect::new( + args.ws_rpc_url + .ok_or_else(|| eyre::eyre!("network proving requires --ws-rpc-url"))?, + )) + .await?; + Some(provider) + } else { + None + }; + let head_shutdown = shutdown.clone(); + let mut stream = if state.is_some() { + futures::stream::unfold( + (http_provider.clone(), None, head_shutdown), + |(provider, mut previous, shutdown)| async move { + loop { + if *shutdown.borrow() { + return None; + } + match provider.get_block_number().await { + Ok(head) if previous != Some(head) => { + previous = Some(head); + return Some((head, (provider, previous, shutdown))); + } + Ok(_) => {} + Err(err) => { + warn!(%err, "head RPC failed; retrying without advancing cursor") + } + } + tokio::time::sleep(Duration::from_secs(2)).await; + } + }, + ) + .boxed() + } else { + ws_provider + .as_ref() + .unwrap() + .subscribe_blocks() + .await? + .into_stream() + .filter(move |h| ready(h.number % block_interval == block_residue)) + .map(|h| h.number) + .boxed() + }; + let client = if let Some(endpoint) = args.local_gpu_endpoint { + let endpoint = endpoint.to_string(); + Arc::new( + tokio::task::spawn_blocking(move || ProverClient { + prover: Box::new(CudaProver::new( + ZKMProver::new(), + ZKMGpuServer::External { endpoint }, + )), + }) + .await?, + ) + } else { + Arc::new(ProverClient::new()) + }; let executor = FullExecutor::, _>::try_new( http_provider.clone(), @@ -85,19 +218,211 @@ async fn main() -> eyre::Result<()> { info!("Latest block number: {}", http_provider.get_block_number().await?); - while let Some(header) = stream.next().await { - // Wait for the block to be avaliable in the HTTP provider - executor.wait_for_block(header.number).await?; + // Two-stage pipeline. The fetcher pulls headers off the subscription, + // fetches the block + witness and runs the native execution; the prover + // loop below consumes the prepared inputs one at a time. Block N+1's + // fetch therefore overlaps block N's proof instead of leaving the cards + // idle for the RPC round trips. + let executor = Arc::new(executor); + let alerting_client = Arc::new(alerting_client); + let (mut tx, mut rx) = mpsc::channel(args.prefetch_depth.max(1)); + let fetcher = { + let executor = executor.clone(); + let alerting_client = alerting_client.clone(); + let max_lag = args.max_lag; + let state = state.clone(); + let preparation_shutdown = shutdown.clone(); + tokio::spawn(async move { + // The subscription is only a wake-up: it says how far the chain + // is. Which block to prepare next is a cursor that advances by + // `block_interval`, so falling behind never skips a block — the + // backlog is worked through in order (unless `max_lag` says to + // drop it). A header stream that lags drops headers; the cursor + // does not. + let mut head = 0u64; + let mut next = state.as_ref().and_then(|s| s.next_block()).unwrap_or(0); + loop { + if *preparation_shutdown.borrow() { + break; + } + if next == 0 || next > head { + match stream.next().await { + Some(header) => { + head = head.max(header); + if next == 0 { + // The stream only delivers this instance's + // residue class, so the cursor starts aligned. + next = header - + ((header + block_interval - block_residue) % block_interval); + if let Some(state) = &state { + state.initialize(next)?; + } + } + continue; + } + None => break, + } + } + // Absorb whatever headers are already queued so `head` is fresh. + while let Some(Some(header)) = stream.next().now_or_never() { + head = head.max(header); + } + + let number = next; + next += block_interval; + if max_lag > 0 && head - number > max_lag { + warn!("skipping block {number}: {} behind the newest header", head - number); + continue; + } - if let Err(err) = executor.execute(header.number).await { - let error_message = format!("Error handling block {}: {err}", header.number); - error!(error_message); + if let Some(state) = &state { + if state.is_committed(number)? { + if tx.send((number, None)).await.is_err() { + break; + } + continue; + } + } + // Retry the same block; a transient preparation failure must not skip it. + loop { + let prepared = match executor.wait_for_block(number).await { + Ok(()) => { + if let Some(state) = &state { + match state.input(number)? { + Some(bytes) => Ok(bincode::deserialize(&bytes)?), + None => executor.prepare(number).await, + } + } else { + executor.prepare(number).await + } + } + Err(err) => Err(err), + }; + match prepared { + Ok(client_input) => { + if let Some(state) = &state { + state.save_input(number, &bincode::serialize(&client_input)?)?; + } + if tx.send((number, Some(client_input))).await.is_err() { + return Ok::<(), eyre::Report>(()); + } + break; + } + Err(err) => { + report_error(&alerting_client, number, err).await; + tokio::time::sleep(Duration::from_secs(5)).await; + } + } + } + } + Ok::<(), eyre::Report>(()) + }) + }; - if let Some(alerting_client) = &alerting_client { - alerting_client.send_alert(error_message).await; + // NOT pipelined, deliberately. Sending block N+1's request early is + // worth ~0.3 s of transfer per block, but the server parks a session + // guard between a block's core and its compress and a new core request + // drops it: a request that lands in that window lets two blocks + // interleave on one card, which is what the guard is there to stop. + 'worker: loop { + if *shutdown.borrow() { + break; + } + let next = tokio::select! { + next = rx.next() => next, + _ = shutdown.changed(), if state.is_some() => None, + }; + let Some((number, client_input)) = next else { + break; + }; + if let Some(state) = &state { + while state.pending_count()? >= args.max_pending_reports { + warn!(number, "report outbox full; waiting before starting another proof"); + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(3)) => {}, + _ = shutdown.changed() => break, + } } + if *shutdown.borrow() { + break; + } + } + if let Some(client_input) = client_input { + let committed = + state.as_ref().map(|s| s.is_committed(number)).transpose()?.unwrap_or(false); + if !committed { + if let Some(state) = &state { + let block = loop { + let result = tokio::select! { + block = http_provider.get_block_by_number(number.into()) => block, + _ = shutdown.changed() => break 'worker, + }; + match result { + Ok(Some(block)) => break block, + Ok(None) => warn!(number, "waiting for prepared block to be available"), + Err(err) => { + warn!(number, %err, "canonical block lookup failed; retrying") + } + } + tokio::time::sleep(Duration::from_secs(3)).await; + }; + if block.header.hash != client_input.current_block.header.hash_slow() { + state.discard_inputs_from(number)?; + eyre::bail!( + "prepared inputs invalidated by a reorg; restart from durable cursor" + ); + } + } + if let Err(err) = executor.prove_prepared(client_input).await { + report_error(&alerting_client, number, err).await; + // The replica lifecycle recovers failed GPU servers; input stays durable. + if state.is_some() { + return Err(eyre::eyre!( + "local proof failed; durable input retained for restart" + )); + } + } + } + } + if let Some(state) = &state { + state.advance(number, true)?; } } + if *shutdown.borrow() { + fetcher.abort(); + } else { + fetcher.await??; + } + reporter.abort(); + if state.is_some() && !*shutdown.borrow() { + eyre::bail!("local worker preparation stream ended unexpectedly"); + } + + Ok(()) +} + +async fn report_error(alerting_client: &Option, number: u64, err: eyre::Report) { + let error_message = format!("Error handling block {number}: {err}"); + error!(error_message); + + if let Some(alerting_client) = alerting_client { + alerting_client.send_alert(error_message).await; + } +} +/// Actor death must not leave an orphan holding the durable stream lock. +fn exit_with_parent() -> eyre::Result<()> { + if env::var("FLEET_EXIT_WITH_PARENT").as_deref() != Ok("true") { + return Ok(()); + } + #[cfg(target_os = "linux")] + unsafe { + let parent = libc::getppid(); + eyre::ensure!(parent > 1, "Fleet worker has no live parent"); + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { + return Err(std::io::Error::last_os_error().into()); + } + eyre::ensure!(libc::getppid() == parent, "Fleet parent exited during startup"); + } Ok(()) } diff --git a/bin/eth-proofs/src/state.rs b/bin/eth-proofs/src/state.rs new file mode 100644 index 0000000..80fefca --- /dev/null +++ b/bin/eth-proofs/src/state.rs @@ -0,0 +1,330 @@ +//! Restart-safe state for a single, machine-pinned EthProofs stream. +use eyre::{bail, Context}; +use fs2::FileExt; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, File, OpenOptions}, + io::Write, + path::{Path, PathBuf}, + sync::Mutex, +}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct Identity { + pub namespace: String, + pub tenant: String, + pub chain_id: u64, + pub cluster_id: u64, + pub elf_id: String, + pub interval: u64, + pub residue: u64, +} +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Checkpoint { + version: u32, + identity: Identity, + next_block: Option, + last_proved: Option, +} +#[derive(Debug, Serialize, Deserialize)] +struct Committed { + block: u64, + proof_hash: String, + public_values_hash: String, + report: Value, +} +pub struct State { + root: PathBuf, + _lock: File, + checkpoint: Mutex, + report_ready: tokio::sync::Notify, +} +impl State { + pub fn open(root: &Path, identity: Identity, start: Option) -> eyre::Result { + if identity.interval == 0 || identity.residue >= identity.interval { + bail!("invalid block interval or residue"); + } + if start.is_some_and(|n| n % identity.interval != identity.residue) { + bail!("start block does not belong to this stream's residue"); + } + fs::create_dir_all(root)?; + let lock = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(root.join("worker.lock"))?; + lock.try_lock_exclusive().wrap_err("another worker owns this storage root")?; + for name in ["inputs", "outputs", "outbox", "reported"] { + fs::create_dir_all(root.join(name))?; + } + let path = root.join("checkpoint.json"); + let checkpoint = if path.exists() { + let checkpoint: Checkpoint = serde_json::from_slice(&fs::read(path)?)?; + if checkpoint.version != 1 || checkpoint.identity != identity { + bail!("worker state identity differs from configured chain, cluster, program or stream"); + } + checkpoint + } else { + Checkpoint { version: 1, identity, next_block: start, last_proved: None } + }; + let state = Self { + root: fs::canonicalize(root)?, + _lock: lock, + checkpoint: Mutex::new(checkpoint), + report_ready: tokio::sync::Notify::new(), + }; + state.save_checkpoint()?; + state.recover_reports()?; + Ok(state) + } + fn save_checkpoint(&self) -> eyre::Result<()> { + atomic_write( + &self.root.join("checkpoint.json"), + &serde_json::to_vec(&*self.checkpoint.lock().unwrap())?, + ) + } + pub fn next_block(&self) -> Option { + self.checkpoint.lock().unwrap().next_block + } + pub fn initialize(&self, block: u64) -> eyre::Result<()> { + { + let mut c = self.checkpoint.lock().unwrap(); + if c.next_block.is_none() { + c.next_block = Some(block); + } + } + self.save_checkpoint() + } + pub fn advance(&self, block: u64, proved: bool) -> eyre::Result<()> { + { + let mut c = self.checkpoint.lock().unwrap(); + if c.next_block != Some(block) { + bail!("refusing to advance out of order"); + } + c.next_block = Some( + block + .checked_add(c.identity.interval) + .ok_or_else(|| eyre::eyre!("block cursor overflow"))?, + ); + if proved { + c.last_proved = Some(block); + } + } + self.save_checkpoint()?; + let input = self.root.join("inputs").join(format!("{block}.bin")); + if input.exists() { + fs::remove_file(input)?; + } + let hash = self.root.join("inputs").join(format!("{block}.sha256")); + if hash.exists() { + fs::remove_file(hash)?; + } + sync_dir(&self.root.join("inputs"))?; + Ok(()) + } + pub fn input(&self, block: u64) -> eyre::Result>> { + let p = self.root.join("inputs").join(format!("{block}.bin")); + if !p.exists() { + return Ok(None); + } + let bytes = fs::read(p)?; + let hash = fs::read_to_string(self.root.join("inputs").join(format!("{block}.sha256")))?; + if digest(&bytes) != hash { + bail!("prepared input checksum mismatch"); + } + Ok(Some(bytes)) + } + pub fn save_input(&self, block: u64, bytes: &[u8]) -> eyre::Result<()> { + // A partial write can always be prepared again; the binary is published last. + atomic_write( + &self.root.join("inputs").join(format!("{block}.sha256")), + digest(bytes).as_bytes(), + )?; + atomic_write(&self.root.join("inputs").join(format!("{block}.bin")), bytes) + } + pub fn commit( + &self, + block: u64, + proof: &[u8], + public_values: &[u8], + report: Value, + ) -> eyre::Result<()> { + let dir = self.root.join("outputs").join(block.to_string()); + fs::create_dir_all(&dir)?; + sync_dir(&self.root.join("outputs"))?; + atomic_write(&dir.join("proof.bin"), proof)?; + atomic_write(&dir.join("public_values.bin"), public_values)?; + let committed = Committed { + block, + proof_hash: digest(proof), + public_values_hash: digest(public_values), + report, + }; + atomic_write(&dir.join("manifest.json"), &serde_json::to_vec(&committed)?)?; + self.ensure_report(&committed)?; + self.report_ready.notify_one(); + Ok(()) + } + pub fn is_committed(&self, block: u64) -> eyre::Result { + let dir = self.root.join("outputs").join(block.to_string()); + if !dir.join("manifest.json").exists() { + return Ok(false); + } + let m: Committed = serde_json::from_slice(&fs::read(dir.join("manifest.json"))?)?; + if m.block != block || + digest(&fs::read(dir.join("proof.bin"))?) != m.proof_hash || + digest(&fs::read(dir.join("public_values.bin"))?) != m.public_values_hash + { + bail!("committed output checksum mismatch"); + } + self.ensure_report(&m)?; + Ok(true) + } + fn ensure_report(&self, m: &Committed) -> eyre::Result<()> { + if !self.root.join("reported").join(format!("{}.json", m.block)).exists() { + let path = self.root.join("outbox").join(format!("{}.json", m.block)); + if !path.exists() { + atomic_write(&path, &serde_json::to_vec(&m.report)?)?; + } + } + Ok(()) + } + fn recover_reports(&self) -> eyre::Result<()> { + // Only the current block can be committed before its outbox/cursor write. + // Earlier blocks advanced only after both were persisted. + if let Some(block) = self.next_block() { + self.is_committed(block)?; + } + Ok(()) + } + pub fn discard_inputs_from(&self, block: u64) -> eyre::Result<()> { + for entry in fs::read_dir(self.root.join("inputs"))? { + let path = entry?.path(); + if path + .file_stem() + .and_then(|s| s.to_str()) + .and_then(|s| s.parse::().ok()) + .is_some_and(|n| n >= block) + { + fs::remove_file(path)?; + } + } + sync_dir(&self.root.join("inputs")) + } + pub async fn report_ready(&self) { + self.report_ready.notified().await; + } + pub fn pending_count(&self) -> eyre::Result { + Ok(fs::read_dir(self.root.join("outbox"))? + .filter_map(Result::ok) + .filter(|e| e.path().extension().and_then(|s| s.to_str()) == Some("json")) + .count()) + } + pub fn pending(&self) -> eyre::Result> { + let mut blocks = Vec::new(); + for entry in fs::read_dir(self.root.join("outbox"))? { + let path = entry?.path(); + if path.extension().and_then(|s| s.to_str()) != Some("json") { + continue; + } + blocks.push(path.file_stem().unwrap().to_str().unwrap().parse::()?); + } + blocks.sort_unstable(); + let mut pending = Vec::new(); + // Load one proof at a time so an API outage cannot inflate reporter memory. + if let Some(block) = blocks.first() { + pending.push(( + *block, + serde_json::from_slice(&fs::read( + self.root.join("outbox").join(format!("{block}.json")), + )?)?, + )); + } + Ok(pending) + } + pub fn acknowledge(&self, block: u64) -> eyre::Result<()> { + let source = self.root.join("outbox").join(format!("{block}.json")); + if source.exists() { + fs::rename(source, self.root.join("reported").join(format!("{block}.json")))?; + sync_dir(&self.root.join("reported"))?; + sync_dir(&self.root.join("outbox"))?; + } + Ok(()) + } +} +pub fn atomic_write(path: &Path, bytes: &[u8]) -> eyre::Result<()> { + let temp = path.with_extension("tmp"); + let mut file = OpenOptions::new().create(true).truncate(true).write(true).open(&temp)?; + file.write_all(bytes)?; + file.sync_all()?; + fs::rename(&temp, path)?; + sync_dir(path.parent().unwrap()) +} +fn sync_dir(path: &Path) -> eyre::Result<()> { + File::open(path)?.sync_all()?; + Ok(()) +} +fn digest(bytes: &[u8]) -> String { + hex::encode(Sha256::digest(bytes)) +} + +#[cfg(test)] +mod tests { + use super::*; + fn identity() -> Identity { + Identity { + namespace: "test".into(), + tenant: "ethproofs".into(), + chain_id: 1, + cluster_id: 8, + elf_id: "abc".into(), + interval: 1, + residue: 0, + } + } + #[test] + fn restart_recovers_proof_and_pending_report_without_reproving() { + let root = tempfile::tempdir().unwrap(); + let state = State::open(root.path(), identity(), Some(10)).unwrap(); + state.save_input(10, b"input").unwrap(); + state.commit(10, b"proof", b"values", serde_json::json!({"block_number":10})).unwrap(); + // Simulate the window after output commit, before cursor advancement. + fs::remove_file(root.path().join("outbox/10.json")).unwrap(); + drop(state); + let state = State::open(root.path(), identity(), None).unwrap(); + assert!(state.is_committed(10).unwrap()); + assert_eq!(state.pending().unwrap().len(), 1); + state.advance(10, true).unwrap(); + state.acknowledge(10).unwrap(); + drop(state); + let state = State::open(root.path(), identity(), None).unwrap(); + assert_eq!(state.next_block(), Some(11)); + assert!(state.pending().unwrap().is_empty()); + assert!(state.input(10).unwrap().is_none()); + } + #[test] + fn duplicate_owner_and_changed_program_are_rejected() { + let root = tempfile::tempdir().unwrap(); + let state = State::open(root.path(), identity(), Some(10)).unwrap(); + assert!(State::open(root.path(), identity(), None).is_err()); + drop(state); + let mut changed = identity(); + changed.elf_id = "other".into(); + assert!(State::open(root.path(), changed, None).is_err()); + } + #[test] + fn corrupted_input_or_output_is_not_reused() { + let root = tempfile::tempdir().unwrap(); + let state = State::open(root.path(), identity(), Some(10)).unwrap(); + state.save_input(10, b"input").unwrap(); + fs::write(root.path().join("inputs/10.bin"), b"bad").unwrap(); + assert!(state.input(10).is_err()); + state.commit(10, b"proof", b"values", Value::Null).unwrap(); + fs::write(root.path().join("outputs/10/proof.bin"), b"bad").unwrap(); + assert!(state.is_committed(10).is_err()); + assert!(state.advance(11, true).is_err()); + } +} diff --git a/crates/executor/host/src/full_executor.rs b/crates/executor/host/src/full_executor.rs index 2785c03..72c810a 100644 --- a/crates/executor/host/src/full_executor.rs +++ b/crates/executor/host/src/full_executor.rs @@ -271,14 +271,12 @@ where } Ok(()) } -} -impl BlockExecutor for FullExecutor -where - C: ExecutorComponents, - P: Provider + Clone + std::fmt::Debug, -{ - async fn execute(&self, block_number: u64) -> eyre::Result<()> { + /// Fetch the block and witness and execute natively, without starting GPU proving. + pub async fn prepare( + &self, + block_number: u64, + ) -> eyre::Result> { self.hooks.on_execution_start(block_number).await?; let client_input_from_cache = self.config.cache_dir.as_ref().and_then(|cache_dir| { @@ -333,9 +331,25 @@ where }; info!("Block {} executed in {:?}", block_number, now.elapsed()); - self.process_client(client_input, &self.hooks, self.config.prove_mode).await?; + Ok(client_input) + } - Ok(()) + /// Prove a prepared input; callers serialize core and compression together. + pub async fn prove_prepared( + &self, + client_input: ClientExecutorInput, + ) -> eyre::Result<()> { + self.process_client(client_input, &self.hooks, self.config.prove_mode).await + } +} + +impl BlockExecutor for FullExecutor +where + C: ExecutorComponents, + P: Provider + Clone + std::fmt::Debug, +{ + async fn execute(&self, block_number: u64) -> eyre::Result<()> { + self.prove_prepared(self.prepare(block_number).await?).await } fn client(&self) -> Arc {