diff --git a/core/src/preflight/mod.rs b/core/src/preflight/mod.rs index 27bd6e041..303a66689 100644 --- a/core/src/preflight/mod.rs +++ b/core/src/preflight/mod.rs @@ -228,6 +228,11 @@ pub async fn batch_preflight( }: BatchPreflightData, ) -> RaikoResult { let measurement = Measurement::start("Fetching block data...", false); + info!( + "batch preflight fetching block and parent data for batch {} across {} blocks", + batch_id, + block_numbers.len() + ); let all_block_parent_pairs = get_batch_blocks_and_parent_data(&provider, &block_numbers).await?; diff --git a/core/src/provider/rpc.rs b/core/src/provider/rpc.rs index 96403b222..110225675 100644 --- a/core/src/provider/rpc.rs +++ b/core/src/provider/rpc.rs @@ -6,8 +6,9 @@ use alloy_transport_http::Http; use raiko_lib::clear_line; use reqwest_alloy::Client; use reth_primitives::revm_primitives::{AccountInfo, Bytecode}; -use std::collections::HashMap; -use tracing::debug; +use std::{collections::HashMap, env, future::Future, time::Duration}; +use tokio::time::sleep; +use tracing::{debug, warn}; use crate::{ interfaces::{RaikoError, RaikoResult}, @@ -22,6 +23,94 @@ pub struct RpcBlockDataProvider { block_numbers: Vec, } +#[derive(Clone, Copy, Debug)] +struct RpcClientConfig { + connect_timeout_secs: u64, + request_timeout_secs: u64, + max_retries: usize, + initial_backoff_ms: u64, + max_backoff_ms: u64, +} + +impl RpcClientConfig { + fn from_env() -> Self { + Self { + connect_timeout_secs: env_u64("RAIKO_RPC_CONNECT_TIMEOUT_SECS", 10), + request_timeout_secs: env_u64("RAIKO_RPC_TIMEOUT_SECS", 120), + max_retries: env_usize("RAIKO_RPC_MAX_RETRIES", 4), + initial_backoff_ms: env_u64("RAIKO_RPC_RETRY_INITIAL_BACKOFF_MS", 1_000), + max_backoff_ms: env_u64("RAIKO_RPC_RETRY_MAX_BACKOFF_MS", 8_000), + } + } +} + +fn env_u64(name: &str, default: u64) -> u64 { + env::var(name) + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(default) +} + +fn env_usize(name: &str, default: usize) -> usize { + env::var(name) + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(default) +} + +fn next_backoff_ms(current_ms: u64, max_ms: u64) -> u64 { + current_ms.saturating_mul(2).min(max_ms.max(current_ms)) +} + +fn is_retryable_rpc_error(error: &RaikoError) -> bool { + match error { + RaikoError::RPC(message) => { + if message.starts_with("Error sending batch request") { + return true; + } + + let message = message.to_ascii_lowercase(); + [ + "timeout", + "timed out", + "deadline", + "connection", + "connect", + "refused", + "reset", + "broken pipe", + "temporarily unavailable", + "429", + "502", + "503", + "504", + "dns", + "eof", + "closed", + ] + .iter() + .any(|needle| message.contains(needle)) + } + _ => false, + } +} + +fn build_http_client() -> RaikoResult { + let config = RpcClientConfig::from_env(); + Client::builder() + .connect_timeout(Duration::from_secs(config.connect_timeout_secs)) + .timeout(Duration::from_secs(config.request_timeout_secs)) + .build() + .map_err(|err| RaikoError::RPC(format!("Failed to build RPC HTTP client: {err}"))) +} + +fn build_rpc_client(url: reqwest::Url) -> RaikoResult>> { + let http_client = build_http_client()?; + let transport = Http::with_client(http_client, url); + let is_local = transport.guess_local(); + Ok(ClientBuilder::default().transport(transport, is_local)) +} + impl RpcBlockDataProvider { /// async will be used for future preflight optimization pub async fn new(url: &str, block_number: u64) -> RaikoResult { @@ -31,9 +120,10 @@ impl RpcBlockDataProvider { "provider rpc url: {:?} for block_number {}", url, block_number ); + let client = build_rpc_client(url.clone())?; Ok(Self { - provider: ProviderBuilder::new().on_provider(RootProvider::new_http(url.clone())), - client: ClientBuilder::default().http(url), + provider: ProviderBuilder::new().on_provider(RootProvider::new(client.clone())), + client, block_numbers: vec![block_number, block_number + 1], }) } @@ -49,9 +139,10 @@ impl RpcBlockDataProvider { "Batch provider rpc: {:?} for block_number {}", url, block_numbers[0] ); + let client = build_rpc_client(url.clone())?; Ok(Self { - provider: ProviderBuilder::new().on_provider(RootProvider::new_http(url.clone())), - client: ClientBuilder::default().http(url), + provider: ProviderBuilder::new().on_provider(RootProvider::new(client.clone())), + client, block_numbers, }) } @@ -59,6 +150,51 @@ impl RpcBlockDataProvider { pub fn provider(&self) -> &ReqwestProvider { &self.provider } + + async fn with_rpc_retry( + &self, + operation: &'static str, + mut make_request: F, + ) -> RaikoResult + where + F: FnMut() -> Fut, + Fut: Future>, + { + let config = RpcClientConfig::from_env(); + let total_attempts = config.max_retries.saturating_add(1).max(1); + let mut backoff_ms = config.initial_backoff_ms; + + for attempt in 1..=total_attempts { + match make_request().await { + Ok(value) => return Ok(value), + Err(err) if attempt < total_attempts && is_retryable_rpc_error(&err) => { + if backoff_ms > 0 { + warn!( + operation = operation, + attempt, + total_attempts, + delay_ms = backoff_ms, + error = %err, + "Retrying transient RPC failure" + ); + sleep(Duration::from_millis(backoff_ms)).await; + backoff_ms = next_backoff_ms(backoff_ms, config.max_backoff_ms); + } else { + warn!( + operation = operation, + attempt, + total_attempts, + error = %err, + "Retrying transient RPC failure without backoff" + ); + } + } + Err(err) => return Err(err), + } + } + + unreachable!("rpc retry loop returns on success or final error") + } } impl BlockDataProvider for RpcBlockDataProvider { @@ -67,40 +203,43 @@ impl BlockDataProvider for RpcBlockDataProvider { let max_batch_size = 32; for blocks_to_fetch in blocks_to_fetch.chunks(max_batch_size) { - let mut batch = self.client.new_batch(); - let mut requests = Vec::with_capacity(max_batch_size); - - for (block_number, full) in blocks_to_fetch { - requests.push(Box::pin( - batch - .add_call( - "eth_getBlockByNumber", - &(BlockNumberOrTag::from(*block_number), full), - ) - .map_err(|_| { - RaikoError::RPC( - "Failed adding eth_getBlockByNumber call to batch".to_owned(), - ) - })?, - )); - } + let mut blocks = self + .with_rpc_retry("eth_getBlockByNumber", || async { + let mut batch = self.client.new_batch(); + let mut requests = Vec::with_capacity(blocks_to_fetch.len()); + + for (block_number, full) in blocks_to_fetch { + requests.push(Box::pin( + batch + .add_call( + "eth_getBlockByNumber", + &(BlockNumberOrTag::from(*block_number), full), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getBlockByNumber call to batch" + .to_owned(), + ) + })?, + )); + } - batch.send().await.map_err(|e| { - RaikoError::RPC(format!( - "Error sending batch request for block {blocks_to_fetch:?}: {e}" - )) - })?; - - let mut blocks = Vec::with_capacity(max_batch_size); - // Collect the data from the batch - for request in requests { - blocks.push( - request.await.map_err(|e| { - RaikoError::RPC(format!("Error collecting request data: {e}")) - })?, - ); - } + batch.send().await.map_err(|e| { + RaikoError::RPC(format!( + "Error sending batch request for block {blocks_to_fetch:?}: {e}" + )) + })?; + + let mut blocks = Vec::with_capacity(blocks_to_fetch.len()); + for request in requests { + blocks.push(request.await.map_err(|e| { + RaikoError::RPC(format!("Error collecting request data: {e}")) + })?); + } + Ok(blocks) + }) + .await?; all_blocks.append(&mut blocks); } @@ -122,83 +261,92 @@ impl BlockDataProvider for RpcBlockDataProvider { let max_batch_size = 250; for accounts in accounts.chunks(max_batch_size) { - let mut batch = self.client.new_batch(); - - let mut nonce_requests = Vec::with_capacity(max_batch_size); - let mut balance_requests = Vec::with_capacity(max_batch_size); - let mut code_requests = Vec::with_capacity(max_batch_size); + let mut fetched_accounts = self + .with_rpc_retry("eth_getAccountState", || async { + let mut batch = self.client.new_batch(); + + let mut nonce_requests = Vec::with_capacity(accounts.len()); + let mut balance_requests = Vec::with_capacity(accounts.len()); + let mut code_requests = Vec::with_capacity(accounts.len()); + + for address in accounts { + nonce_requests.push(Box::pin( + batch + .add_call::<_, Uint<64, 1>>( + "eth_getTransactionCount", + &(address, Some(BlockId::from(block_number))), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getTransactionCount call to batch" + .to_owned(), + ) + })?, + )); + balance_requests.push(Box::pin( + batch + .add_call::<_, Uint<256, 4>>( + "eth_getBalance", + &(address, Some(BlockId::from(block_number))), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getBalance call to batch".to_owned(), + ) + })?, + )); + code_requests.push(Box::pin( + batch + .add_call::<_, Bytes>( + "eth_getCode", + &(address, Some(BlockId::from(block_number))), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getCode call to batch".to_owned(), + ) + })?, + )); + } - for address in accounts { - nonce_requests.push(Box::pin( - batch - .add_call::<_, Uint<64, 1>>( - "eth_getTransactionCount", - &(address, Some(BlockId::from(block_number))), - ) - .map_err(|_| { - RaikoError::RPC( - "Failed adding eth_getTransactionCount call to batch".to_owned(), - ) - })?, - )); - balance_requests.push(Box::pin( - batch - .add_call::<_, Uint<256, 4>>( - "eth_getBalance", - &(address, Some(BlockId::from(block_number))), - ) - .map_err(|_| { - RaikoError::RPC("Failed adding eth_getBalance call to batch".to_owned()) - })?, - )); - code_requests.push(Box::pin( batch - .add_call::<_, Bytes>( - "eth_getCode", - &(address, Some(BlockId::from(block_number))), - ) - .map_err(|_| { - RaikoError::RPC("Failed adding eth_getCode call to batch".to_owned()) - })?, - )); - } + .send() + .await + .map_err(|e| RaikoError::RPC(format!("Error sending batch request {e}")))?; + + let mut accounts = Vec::with_capacity(accounts.len()); + for ((nonce_request, balance_request), code_request) in nonce_requests + .into_iter() + .zip(balance_requests.into_iter()) + .zip(code_requests.into_iter()) + { + let (nonce, balance, code) = ( + nonce_request.await.map_err(|e| { + RaikoError::RPC(format!("Failed to collect nonce request: {e}")) + })?, + balance_request.await.map_err(|e| { + RaikoError::RPC(format!("Failed to collect balance request: {e}")) + })?, + code_request.await.map_err(|e| { + RaikoError::RPC(format!("Failed to collect code request: {e}")) + })?, + ); - batch - .send() - .await - .map_err(|e| RaikoError::RPC(format!("Error sending batch request {e}")))?; - - let mut accounts = vec![]; - // Collect the data from the batch - for ((nonce_request, balance_request), code_request) in nonce_requests - .into_iter() - .zip(balance_requests.into_iter()) - .zip(code_requests.into_iter()) - { - let (nonce, balance, code) = ( - nonce_request.await.map_err(|e| { - RaikoError::RPC(format!("Failed to collect nonce request: {e}")) - })?, - balance_request.await.map_err(|e| { - RaikoError::RPC(format!("Failed to collect balance request: {e}")) - })?, - code_request.await.map_err(|e| { - RaikoError::RPC(format!("Failed to collect code request: {e}")) - })?, - ); - - let nonce = nonce.try_into().map_err(|_| { - RaikoError::Conversion("Failed to convert nonce to u64".to_owned()) - })?; - - let bytecode = Bytecode::new_raw(code); - - let account_info = AccountInfo::new(balance, nonce, bytecode.hash_slow(), bytecode); - - accounts.push(account_info); - } + let nonce = nonce.try_into().map_err(|_| { + RaikoError::Conversion("Failed to convert nonce to u64".to_owned()) + })?; - all_accounts.append(&mut accounts); + let bytecode = Bytecode::new_raw(code); + let account_info = + AccountInfo::new(balance, nonce, bytecode.hash_slow(), bytecode); + accounts.push(account_info); + } + + Ok(accounts) + }) + .await?; + + all_accounts.append(&mut fetched_accounts); } Ok(all_accounts) @@ -219,39 +367,42 @@ impl BlockDataProvider for RpcBlockDataProvider { let max_batch_size = 1000; for accounts in accounts.chunks(max_batch_size) { - let mut batch = self.client.new_batch(); - - let mut requests = Vec::with_capacity(max_batch_size); + let mut values = self + .with_rpc_retry("eth_getStorageAt", || async { + let mut batch = self.client.new_batch(); + + let mut requests = Vec::with_capacity(accounts.len()); + + for (address, key) in accounts { + requests.push(Box::pin( + batch + .add_call::<_, U256>( + "eth_getStorageAt", + &(address, key, Some(BlockId::from(block_number))), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getStorageAt call to batch".to_owned(), + ) + })?, + )); + } - for (address, key) in accounts { - requests.push(Box::pin( batch - .add_call::<_, U256>( - "eth_getStorageAt", - &(address, key, Some(BlockId::from(block_number))), - ) - .map_err(|_| { - RaikoError::RPC( - "Failed adding eth_getStorageAt call to batch".to_owned(), - ) - })?, - )); - } + .send() + .await + .map_err(|e| RaikoError::RPC(format!("Error sending batch request {e}")))?; + + let mut values = Vec::with_capacity(accounts.len()); + for request in requests { + values.push(request.await.map_err(|e| { + RaikoError::RPC(format!("Error collecting request data: {e}")) + })?); + } - batch - .send() - .await - .map_err(|e| RaikoError::RPC(format!("Error sending batch request {e}")))?; - - let mut values = Vec::with_capacity(max_batch_size); - // Collect the data from the batch - for request in requests { - values.push( - request.await.map_err(|e| { - RaikoError::RPC(format!("Error collecting request data: {e}")) - })?, - ); - } + Ok(values) + }) + .await?; all_values.append(&mut values); } @@ -286,75 +437,78 @@ impl BlockDataProvider for RpcBlockDataProvider { #[cfg(not(debug_assertions))] tracing::trace!("Fetching storage proof {idx}/{num_storage_proofs}..."); - // Create a batch for all storage proofs - let mut batch = self.client.new_batch(); - - // Collect all requests - let mut requests = Vec::new(); - - let mut batch_size = 0; - while !accounts.is_empty() && batch_size < batch_limit { - let mut address_to_remove = None; - - if let Some((address, keys)) = accounts.iter_mut().next() { - // Calculate how many keys we can still process - let num_keys_to_process = if batch_size + keys.len() < batch_limit { - keys.len() - } else { - batch_limit - batch_size - }; - - // If we can process all keys, remove the address from the map after the loop - if num_keys_to_process == keys.len() { - address_to_remove = Some(*address); + let (proofs, remaining_accounts) = self + .with_rpc_retry("eth_getProof", || { + let mut pending_accounts = accounts.clone(); + + async move { + let mut batch = self.client.new_batch(); + let mut requests = Vec::new(); + + let mut batch_size = 0; + while !pending_accounts.is_empty() && batch_size < batch_limit { + let mut address_to_remove = None; + + if let Some((address, keys)) = pending_accounts.iter_mut().next() { + let num_keys_to_process = if batch_size + keys.len() < batch_limit { + keys.len() + } else { + batch_limit - batch_size + }; + + if num_keys_to_process == keys.len() { + address_to_remove = Some(*address); + } + + let keys_to_process = keys + .drain(0..num_keys_to_process) + .map(StorageKey::from) + .collect::>(); + + requests.push(Box::pin( + batch + .add_call::<_, EIP1186AccountProofResponse>( + "eth_getProof", + &( + *address, + keys_to_process.clone(), + BlockId::from(block_number), + ), + ) + .map_err(|_| { + RaikoError::RPC( + "Failed adding eth_getProof call to batch" + .to_owned(), + ) + })?, + )); + + batch_size += 1 + keys_to_process.len(); + } + + if let Some(address) = address_to_remove { + pending_accounts.remove(&address); + } + } + + batch.send().await.map_err(|e| { + RaikoError::RPC(format!("Error sending batch request {e}")) + })?; + + let mut proofs = Vec::with_capacity(requests.len()); + for request in requests { + proofs.push(request.await.map_err(|e| { + RaikoError::RPC(format!("Error collecting request data: {e}")) + })?); + } + + Ok((proofs, pending_accounts)) } + }) + .await?; + accounts = remaining_accounts; - // Extract the keys to process - let keys_to_process = keys - .drain(0..num_keys_to_process) - .map(StorageKey::from) - .collect::>(); - - // Add the request - requests.push(Box::pin( - batch - .add_call::<_, EIP1186AccountProofResponse>( - "eth_getProof", - &( - *address, - keys_to_process.clone(), - BlockId::from(block_number), - ), - ) - .map_err(|_| { - RaikoError::RPC( - "Failed adding eth_getProof call to batch".to_owned(), - ) - })?, - )); - - // Keep track of how many keys were processed - // Add an additional 1 for the account proof itself - batch_size += 1 + keys_to_process.len(); - } - - // Remove the address if all keys were processed for this account - if let Some(address) = address_to_remove { - accounts.remove(&address); - } - } - - // Send the batch - batch - .send() - .await - .map_err(|e| RaikoError::RPC(format!("Error sending batch request {e}")))?; - - // Collect the data from the batch - for request in requests { - let mut proof = request - .await - .map_err(|e| RaikoError::RPC(format!("Error collecting request data: {e}")))?; + for mut proof in proofs { idx += proof.storage_proof.len(); if let Some(map_proof) = storage_proofs.get_mut(&proof.address) { map_proof.storage_proof.append(&mut proof.storage_proof); diff --git a/gaiko b/gaiko index 375d2c38a..f478cb45a 160000 --- a/gaiko +++ b/gaiko @@ -1 +1 @@ -Subproject commit 375d2c38a2f1abdb3d32a05da5268969cd304341 +Subproject commit f478cb45a8c76144800f89c8a6cab0ad8b992e03 diff --git a/provers/sp1/driver/src/lib.rs b/provers/sp1/driver/src/lib.rs index 0a6f5e003..6235940e6 100644 --- a/provers/sp1/driver/src/lib.rs +++ b/provers/sp1/driver/src/lib.rs @@ -250,7 +250,7 @@ impl Prover for Sp1Prover { })?; info!("Sp1: network proof id: {proof_id:?} for aggregation"); network_client - .wait_proof(proof_id.clone(), Some(Duration::from_secs(3600))) + .wait_proof(proof_id.clone(), Some(sp1_network_wait_timeout())) .await .map_err(|e| ProverError::GuestError(format!("Sp1: network proof failed {e:?}")))? }; @@ -402,7 +402,7 @@ impl Prover for Sp1Prover { input.taiko.batch_id ); network_client - .wait_proof(proof_id.clone(), Some(Duration::from_secs(3600))) + .wait_proof(proof_id.clone(), Some(sp1_network_wait_timeout())) .await .map_err(|e| ProverError::GuestError(format!("Sp1: network proof failed {e:?}")))? }; @@ -571,7 +571,7 @@ impl Prover for Sp1Prover { })?; info!("Sp1: network proof id: {proof_id:?} for shasta aggregation"); network_client - .wait_proof(proof_id.clone(), Some(Duration::from_secs(3600))) + .wait_proof(proof_id.clone(), Some(sp1_network_wait_timeout())) .await .map_err(|e| ProverError::GuestError(format!("Sp1: network proof failed {e:?}")))? }; @@ -657,6 +657,15 @@ fn get_env_mock() -> ProverMode { } } +fn sp1_network_wait_timeout() -> Duration { + Duration::from_secs( + env::var("RAIKO_SP1_NETWORK_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(3_600), + ) +} + /// A fixture that can be used to test the verification of SP1 zkVM proofs inside Solidity. #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] diff --git a/reqactor/src/backend.rs b/reqactor/src/backend.rs index 084e4f9df..ace9fc130 100644 --- a/reqactor/src/backend.rs +++ b/reqactor/src/backend.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{env, sync::Arc, time::Duration}; use raiko_core::{ interfaces::{aggregate_proofs, aggregate_shasta_proposals, ProofRequest}, @@ -26,7 +26,10 @@ use raiko_reqpool::{ ShastaProofRequestEntity, SingleProofRequestEntity, Status, StatusWithContext, }; use reth_primitives::B256; -use tokio::sync::{mpsc, Mutex, Notify, Semaphore}; +use tokio::{ + sync::{mpsc, Mutex, Notify, Semaphore}, + time::timeout, +}; use tracing::{debug, trace}; use crate::queue::Queue; @@ -42,6 +45,60 @@ pub(crate) struct Backend { semaphore: Arc, } +#[derive(Clone, Copy, Debug)] +struct BackendTimeoutConfig { + guest_input_timeout_secs: u64, + batch_guest_input_timeout_secs: u64, + proof_timeout_secs: u64, + batch_proof_timeout_secs: u64, + aggregation_timeout_secs: u64, +} + +impl BackendTimeoutConfig { + fn from_env() -> Self { + Self { + guest_input_timeout_secs: env_u64("RAIKO_GUEST_INPUT_TIMEOUT_SECS", 900), + batch_guest_input_timeout_secs: env_u64("RAIKO_BATCH_GUEST_INPUT_TIMEOUT_SECS", 1_800), + proof_timeout_secs: env_u64("RAIKO_PROOF_TIMEOUT_SECS", 7_200), + batch_proof_timeout_secs: env_u64("RAIKO_BATCH_PROOF_TIMEOUT_SECS", 7_200), + aggregation_timeout_secs: env_u64("RAIKO_AGGREGATION_TIMEOUT_SECS", 7_200), + } + } +} + +fn env_u64(name: &str, default: u64) -> u64 { + env::var(name) + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(default) +} + +fn request_timeout_for(request_entity: &RequestEntity) -> (&'static str, Duration) { + let config = BackendTimeoutConfig::from_env(); + match request_entity { + RequestEntity::SingleProof(_) => ( + "single proof request", + Duration::from_secs(config.proof_timeout_secs), + ), + RequestEntity::Aggregation(_) | RequestEntity::ShastaAggregation(_) => ( + "aggregation request", + Duration::from_secs(config.aggregation_timeout_secs), + ), + RequestEntity::BatchProof(_) | RequestEntity::ShastaProof(_) => ( + "batch proof request", + Duration::from_secs(config.batch_proof_timeout_secs), + ), + RequestEntity::GuestInput(_) => ( + "guest input request", + Duration::from_secs(config.guest_input_timeout_secs), + ), + RequestEntity::BatchGuestInput(_) | RequestEntity::ShastaGuestInput(_) => ( + "batch guest input request", + Duration::from_secs(config.batch_guest_input_timeout_secs), + ), + } +} + impl Backend { pub fn new( pool: Pool, @@ -96,58 +153,76 @@ impl Backend { let request_key_ = request_key.clone(); let mut pool_ = self.pool.clone(); let chain_specs = self.chain_specs.clone(); + let (request_kind, request_timeout) = request_timeout_for(&request_entity); let handle = tokio::spawn(async move { let _permit = permit; - - let result = match request_entity { - RequestEntity::SingleProof(entity) => { - do_prove_single(&mut pool_, &chain_specs, request_key_.clone(), entity) + let timeout_secs = request_timeout.as_secs(); + let result = match timeout(request_timeout, async { + match request_entity { + RequestEntity::SingleProof(entity) => { + do_prove_single(&mut pool_, &chain_specs, request_key_.clone(), entity) + .await + } + RequestEntity::Aggregation(entity) => { + do_prove_aggregation(&mut pool_, request_key_.clone(), entity).await + } + RequestEntity::BatchProof(entity) => { + do_prove_batch(&mut pool_, &chain_specs, request_key_.clone(), entity) + .await + } + RequestEntity::GuestInput(entity) => { + do_generate_guest_input( + &mut pool_, + &chain_specs, + request_key_.clone(), + entity, + ) .await + } + RequestEntity::BatchGuestInput(entity) => { + do_generate_batch_guest_input( + &mut pool_, + &chain_specs, + request_key_.clone(), + entity, + ) + .await + } + RequestEntity::ShastaGuestInput(entity) => { + do_generate_shasta_proposal_guest_input( + &mut pool_, + &chain_specs, + request_key_.clone(), + entity, + ) + .await + } + RequestEntity::ShastaProof(entity) => { + do_prove_shasta_proposal( + &mut pool_, + &chain_specs, + request_key_.clone(), + entity, + ) + .await + } + RequestEntity::ShastaAggregation(entity) => { + do_shasta_aggregation(&mut pool_, request_key_.clone(), entity).await + } } - RequestEntity::Aggregation(entity) => { - do_prove_aggregation(&mut pool_, request_key_.clone(), entity).await - } - RequestEntity::BatchProof(entity) => { - do_prove_batch(&mut pool_, &chain_specs, request_key_.clone(), entity).await - } - RequestEntity::GuestInput(entity) => { - do_generate_guest_input( - &mut pool_, - &chain_specs, - request_key_.clone(), - entity, - ) - .await - } - RequestEntity::BatchGuestInput(entity) => { - do_generate_batch_guest_input( - &mut pool_, - &chain_specs, - request_key_.clone(), - entity, - ) - .await - } - RequestEntity::ShastaGuestInput(entity) => { - do_generate_shasta_proposal_guest_input( - &mut pool_, - &chain_specs, - request_key_.clone(), - entity, - ) - .await - } - RequestEntity::ShastaProof(entity) => { - do_prove_shasta_proposal( - &mut pool_, - &chain_specs, - request_key_.clone(), - entity, - ) - .await - } - RequestEntity::ShastaAggregation(entity) => { - do_shasta_aggregation(&mut pool_, request_key_.clone(), entity).await + }) + .await + { + Ok(result) => result, + Err(_) => { + let message = format!("{request_kind} timed out after {timeout_secs}s"); + tracing::error!( + request_key = %request_key_, + request_kind, + timeout_secs, + "Actor backend timed out" + ); + Err(message) } }; let status = match result {