diff --git a/crates/core/generate-pie/src/lib.rs b/crates/core/generate-pie/src/lib.rs index 3e8581c7..a1cb86e7 100644 --- a/crates/core/generate-pie/src/lib.rs +++ b/crates/core/generate-pie/src/lib.rs @@ -63,11 +63,13 @@ // Standard library imports use std::path::Path; use std::sync::Arc; +use std::time::{Duration, Instant}; // External crate imports use anyhow::bail; use cairo_vm::types::layout_name::LayoutName; use futures::future::join_all; use log::{info, warn}; +use rpc_client::utils::{reset_rpc_timing, rpc_timing_snapshot}; use rpc_client::RpcClient; use starknet_api::core::OsChainInfo; use starknet_os::{ @@ -79,7 +81,7 @@ use tokio::sync::Semaphore; use crate::constants::{DEFAULT_MAX_PARALLEL_BLOCKS, MAX_EXECUTION_STEPS_WARNING_THRESHOLD}; use block_processor::collect_single_block_info; use error::PieGenerationError; -use types::{PieGenerationInput, PieGenerationResult}; +use types::{PieGenerationInput, PieGenerationResult, PieGenerationTiming}; use utils::sort_abi_entries_for_deprecated_class; const MAX_PARALLEL_BLOCKS_ENV: &str = "SNOS_MAX_PARALLEL_BLOCKS"; @@ -154,6 +156,8 @@ pub mod types; /// } /// ``` pub async fn generate_pie(input: PieGenerationInput) -> Result { + reset_rpc_timing(); + let snos_started_at = Instant::now(); info!("Starting PIE generation for {} blocks: {:?}", input.blocks.len(), input.blocks); // Validate input configuration @@ -299,9 +303,43 @@ pub async fn generate_pie(input: PieGenerationInput) -> Result String { + format!("{:.3}s", duration.as_secs_f64()) +} + +fn duration_millis(duration: Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 } pub fn parse_layout(layout: &str) -> anyhow::Result { diff --git a/crates/core/generate-pie/src/types/pie.rs b/crates/core/generate-pie/src/types/pie.rs index f3cbf7df..ca51cbd1 100644 --- a/crates/core/generate-pie/src/types/pie.rs +++ b/crates/core/generate-pie/src/types/pie.rs @@ -2,6 +2,7 @@ use blockifier::blockifier_versioned_constants::VersionedConstants; use cairo_vm::types::layout_name::LayoutName; use starknet_os::io::os_output::StarknetOsRunnerOutput; use starknet_types_core::felt::Felt; +use std::collections::HashMap; use crate::error::PieGenerationError; use crate::types::{ChainConfig, OsHintsConfiguration}; @@ -91,4 +92,19 @@ pub struct PieGenerationResult { pub blocks_processed: Vec, /// The output file path where the PIE was saved (if specified). pub output_path: Option, + /// Timing details for the SNOS run. + pub timing: PieGenerationTiming, +} + +/// Wall-clock timing details for a SNOS run. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct PieGenerationTiming { + /// Total wall-clock time spent in SNOS processing. + pub total_processing_time_ms: u64, + /// Wall-clock time spent with at least one RPC call in flight. + pub rpc_wait_time_ms: u64, + /// Wall-clock time spent on local execution/processing outside RPC waits. + pub execution_time_ms: u64, + /// RPC calls grouped by method name, including a `total` entry. + pub rpc_calls_by_method: HashMap, } diff --git a/crates/core/generate-pie/src/types/proof.rs b/crates/core/generate-pie/src/types/proof.rs index 6cd0d972..1aa32cb3 100644 --- a/crates/core/generate-pie/src/types/proof.rs +++ b/crates/core/generate-pie/src/types/proof.rs @@ -112,6 +112,7 @@ impl ProofCollectionResult { .chain(current_contract_commitment_facts) .map(|(key, value)| (HashOutput(key), value)) .collect(); + let merged_contract_fact_count = global_contract_commitment_facts.len(); let contract_state_commitment_info = CommitmentInfo { previous_root: HashOutput(previous_contract_storage_root), @@ -120,6 +121,15 @@ impl ProofCollectionResult { commitment_facts: global_contract_commitment_facts, }; + info!( + "Prepared storage commitment info for block {:?}, contract {:#x}: previous_root={:#x} updated_root={:#x} merged_facts={}", + block_id, + contract_address, + previous_contract_storage_root, + current_contract_data.root, + merged_contract_fact_count + ); + address_to_storage_commitment_info.insert( ContractAddress::try_from(contract_address) .map_err(|e| BlockProcessingError::new_custom(format!("Invalid contract address: {:?}", e)))?, @@ -197,6 +207,7 @@ impl ProofCollectionResult { .chain(current_state_commitment_facts) .map(|(k, v)| (HashOutput(k), v)) .collect(); + let merged_global_state_fact_count = global_state_commitment_facts.len(); let contract_state_commitment_info = CommitmentInfo { previous_root: HashOutput(previous_contract_trie_root), @@ -205,6 +216,18 @@ impl ProofCollectionResult { commitment_facts: global_state_commitment_facts, }; + info!( + "Prepared global contract-state commitment for block {:?}: previous_root={:#x} updated_root={:#x} merged_facts={}", + block_id, + previous_contract_trie_root, + current_contract_trie_root, + merged_global_state_fact_count + ); + info!( + "Prepared class commitment roots for block {:?}: previous_root={:#x} updated_root={:#x}", + block_id, previous_root, updated_root + ); + // Compute class commitment let contract_class_commitment_info = compute_class_commitment(&self.previous_class_proofs, &self.class_proofs, previous_root, updated_root); diff --git a/crates/core/generate-pie/src/utils/rpc.rs b/crates/core/generate-pie/src/utils/rpc.rs index c65a8e66..2fec6fbb 100644 --- a/crates/core/generate-pie/src/utils/rpc.rs +++ b/crates/core/generate-pie/src/utils/rpc.rs @@ -4,7 +4,7 @@ use blockifier::execution::call_info::CallInfo; use blockifier::state::cached_state::StateMaps; use blockifier::transaction::objects::TransactionExecutionInfo; use cairo_vm::Felt252; -use log::info; +use log::{info, warn}; use rpc_client::client::ProofClient; use rpc_client::error::ClientError; use rpc_client::types::{ClassProof, ContractData, ContractProof}; @@ -16,6 +16,14 @@ use starknet_api::state::StorageKey; use starknet_types_core::felt::Felt; use std::collections::{HashMap, HashSet}; +fn summarize_felts(values: &[Felt], limit: usize) -> String { + let mut summary: Vec = values.iter().take(limit).map(|felt| format!("{:#x}", felt)).collect(); + if values.len() > limit { + summary.push(format!("... +{}", values.len() - limit)); + } + summary.join(", ") +} + /// Comprehensive structure that captures all access information from transaction execution #[derive(Debug, Clone)] pub struct BlockAccessInfo { @@ -148,7 +156,12 @@ pub(crate) async fn get_class_proofs( let proof = execute_with_retry(&operation_name, || rpc_client.starknet_rpc().get_class_proof(block_number, class_hash)) .await - .map_err(ClientError::ProviderError)?; + .map_err(|e| { + let message = + format!("class proof request failed for block {block_number} class_hash {class_hash:#x}: {e}"); + warn!("{message}"); + ClientError::CustomError(message) + })?; // TODO: need to combine these, similar to merge_chunked_storage_proofs above? proofs.insert(**class_hash, proof); } @@ -188,18 +201,37 @@ async fn get_storage_proof_for_contract>( let contract_data = match &contract_proof.contract_data { None => { + warn!( + "Storage proof for contract {} at block {} returned no contract_data", + contract_address, block_number + ); return Ok(contract_proof); } Some(contract_data) => contract_data, }; + info!( + "Fetched initial storage proof for contract {} at block {}: root={:#x} storage_proof_sets={} contract_nodes={} requested_keys=[{}]", + contract_address, + block_number, + contract_data.root, + contract_data.storage_proofs.len(), + contract_proof.contract_proof.len(), + summarize_felts(&keys, 8) + ); + let additional_keys = if contract_data.root != Felt::ZERO { contract_data.get_additional_keys(&keys).map_err(|e| ClientError::CustomError(format!("{}", e)))? } else { vec![] }; - info!("Got {} additional keys for contract {}", additional_keys.len(), contract_address); + info!( + "Got {} additional keys for contract {} [{}]", + additional_keys.len(), + contract_address, + summarize_felts(&additional_keys, 8) + ); // Fetch additional proofs required to fill gaps in the storage trie that could make // the OS crash otherwise. @@ -210,7 +242,12 @@ async fn get_storage_proof_for_contract>( // Combine all storage proofs into a single vector match &additional_proof.contract_data { None => { - panic!("Failed to fetch additional proof for contract {}", contract_address) + let message = format!( + "Additional storage proof for contract {} at block {} returned no contract_data", + contract_address, block_number + ); + warn!("{message}"); + return Err(ClientError::CustomError(message)); } Some(contract_data) => { additional_proof.contract_data = Some(ContractData { @@ -223,6 +260,17 @@ async fn get_storage_proof_for_contract>( contract_proof = merge_storage_proofs(vec![contract_proof.clone(), additional_proof]); } + if let Some(contract_data) = &contract_proof.contract_data { + info!( + "Final merged storage proof for contract {} at block {}: root={:#x} storage_proof_sets={} contract_nodes={}", + contract_address, + block_number, + contract_data.root, + contract_data.storage_proofs.len(), + contract_proof.contract_proof.len() + ); + } + Ok(contract_proof) } @@ -235,8 +283,32 @@ async fn fetch_storage_proof_for_contract( keys: &[Felt], block_number: u64, ) -> Result { - info!("Fetching storage proof for contract {} with {} keys", contract_address, keys.len()); - rpc_client.starknet_rpc().get_proof(block_number, contract_address, keys).await.map_err(ClientError::ProviderError) + info!( + "Fetching storage proof for contract {} with {} keys [{}]", + contract_address, + keys.len(), + summarize_felts(keys, 8) + ); + + let operation_name = format!( + "get_proof(block_number: {block_number}, contract_address: {contract_address:#x}, keys: {})", + keys.len() + ); + + execute_with_retry(&operation_name, || rpc_client.starknet_rpc().get_proof(block_number, contract_address, keys)) + .await + .map_err(|e| { + let message = format!( + "storage proof request failed for block {} contract {:#x} keys={} [{}]: {}", + block_number, + contract_address, + keys.len(), + summarize_felts(keys, 8), + e + ); + warn!("{message}"); + ClientError::CustomError(message) + }) } /// Merges the storage proofs of the SAME contract. diff --git a/crates/rpc-client/src/state_reader/mod.rs b/crates/rpc-client/src/state_reader/mod.rs index 32fd9d86..3d152174 100644 --- a/crates/rpc-client/src/state_reader/mod.rs +++ b/crates/rpc-client/src/state_reader/mod.rs @@ -2,7 +2,7 @@ use blockifier::execution::contract_class::{CompiledClassV0, CompiledClassV1, Ru use blockifier::state::errors::StateError; use blockifier::state::state_api::{StateReader, StateResult}; use cairo_lang_starknet_classes::contract_class::version_id_from_serialized_sierra_program; -use log::{debug, warn}; +use log::{debug, info, warn}; use starknet::core::types::{BlockId, Felt, StarknetError}; use starknet::providers::{Provider, ProviderError}; use starknet_api::contract_class::compiled_class_hash::{HashVersion, HashableCompiledClass}; @@ -50,6 +50,26 @@ impl AsyncRpcStateReader { } } +fn log_cached_state_zero_fallback( + field_name: &str, + block_id: BlockId, + contract_address: ContractAddress, + key: Option, + error: &ProviderError, +) { + let key_suffix = key.map(|key| format!(", key={:#x}", Felt::from(*key.0.key()))).unwrap_or_default(); + let message = format!( + "Cached state {field_name} fallback to zero for block {block_id:?}, contract={:#x}{key_suffix}: {error}", + Felt::from(*contract_address.key()) + ); + + match error { + ProviderError::StarknetError(StarknetError::ContractNotFound) => info!("{message}"), + ProviderError::StarknetError(StarknetError::ClassHashNotFound) => info!("{message}"), + _ => warn!("{message}"), + } +} + // Helper function to convert provider error to state error fn provider_error_to_state_error(provider_error: ProviderError) -> StateError { StateError::StateReadError(provider_error.to_string()) @@ -80,7 +100,10 @@ impl AsyncRpcStateReader { .await { Ok(value) => Ok(value.value()), - Err(ProviderError::StarknetError(StarknetError::ContractNotFound)) => Ok(Felt::ZERO), + Err(err @ ProviderError::StarknetError(StarknetError::ContractNotFound)) => { + log_cached_state_zero_fallback("storage", block_id, contract_address, Some(key), &err); + Ok(Felt::ZERO) + } Err(e) => Err(provider_error_to_state_error(e)), }?; @@ -103,7 +126,10 @@ impl AsyncRpcStateReader { .await { Ok(value) => Ok(value), - Err(ProviderError::StarknetError(StarknetError::ContractNotFound)) => Ok(Felt::ZERO), + Err(err @ ProviderError::StarknetError(StarknetError::ContractNotFound)) => { + log_cached_state_zero_fallback("nonce", block_id, contract_address, None, &err); + Ok(Felt::ZERO) + } Err(e) => Err(provider_error_to_state_error(e)), }?; Ok(Nonce(nonce)) @@ -125,7 +151,10 @@ impl AsyncRpcStateReader { .await { Ok(class_hash) => Ok(class_hash), - Err(ProviderError::StarknetError(StarknetError::ContractNotFound)) => Ok(ClassHash::default().0), + Err(err @ ProviderError::StarknetError(StarknetError::ContractNotFound)) => { + log_cached_state_zero_fallback("class_hash", block_id, contract_address, None, &err); + Ok(ClassHash::default().0) + } Err(e) => Err(provider_error_to_state_error(e)), }?; diff --git a/crates/rpc-client/src/types/proofs/contract.rs b/crates/rpc-client/src/types/proofs/contract.rs index e17ba781..a062d109 100644 --- a/crates/rpc-client/src/types/proofs/contract.rs +++ b/crates/rpc-client/src/types/proofs/contract.rs @@ -1,5 +1,5 @@ use anyhow::bail; -use log::info; +use log::{info, warn}; use rayon::prelude::*; use serde::{Deserialize, Serialize}; use starknet::providers::ProviderError; @@ -50,12 +50,31 @@ impl ContractData { info!("Fetching additional keys for a contract which already have {} keys", keys.len()); let mut additional_keys = vec![]; if let Err(errors) = self.verify(keys) { + warn!( + "Contract proof verification produced {} errors for root {:#x} across {} requested keys", + errors.len(), + self.root, + keys.len() + ); + for (index, error) in errors.iter().take(10).enumerate() { + warn!("Proof verification error {}/{}: {:?}", index + 1, errors.len(), error); + } + if errors.len() > 10 { + warn!("... {} additional proof verification errors omitted", errors.len() - 10); + } for error in errors { match error { ProofVerificationError::NonExistenceProof { key, height, node } => { if let TrieNode::Edge { child: _, path, .. } = &node { if height.0 < DEFAULT_STORAGE_TREE_HEIGHT { let modified_key = path.get_key_following_edge(key, height); + info!( + "Derived additional key {:#x} from non-existence proof on root {:#x} (original key {:#x}, height={})", + modified_key, + self.root, + key, + height.0 + ); additional_keys.push(modified_key); } } diff --git a/crates/rpc-client/src/utils.rs b/crates/rpc-client/src/utils.rs index e4aee9dd..0158f113 100644 --- a/crates/rpc-client/src/utils.rs +++ b/crates/rpc-client/src/utils.rs @@ -3,14 +3,16 @@ use log::warn; use starknet::core::types::StarknetError; use starknet::providers::ProviderError; +use std::collections::HashMap; use std::future::Future; -use std::sync::OnceLock; -use std::time::Duration; +use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, Instant}; use tokio::time::sleep; /// Global Tokio runtime for executing async operations in non-async contexts. /// This is used when there's no current runtime available (e.g., in worker threads). static GLOBAL_RUNTIME: OnceLock = OnceLock::new(); +static RPC_TIMING_STATE: OnceLock> = OnceLock::new(); /// Maximum number of retry attempts for RPC calls. const MAX_RETRY_ATTEMPTS: u32 = 5; @@ -21,6 +23,24 @@ const INITIAL_BACKOFF_MS: u64 = 100; /// Maximum delay for exponential backoff (in milliseconds). const MAX_BACKOFF_MS: u64 = 5000; +#[derive(Debug, Clone, Default)] +pub struct RpcTimingSnapshot { + pub wait_elapsed: Duration, + pub cumulative_call_elapsed: Duration, + pub calls: u64, + pub calls_by_method: HashMap, +} + +#[derive(Debug, Default)] +struct RpcTimingState { + active_calls: u64, + wait_started_at: Option, + wait_elapsed: Duration, + cumulative_call_elapsed: Duration, + calls: u64, + calls_by_method: HashMap, +} + /// Gets or creates the global Tokio runtime. fn get_global_runtime() -> &'static tokio::runtime::Runtime { GLOBAL_RUNTIME.get_or_init(|| { @@ -28,6 +48,71 @@ fn get_global_runtime() -> &'static tokio::runtime::Runtime { }) } +fn rpc_timing_state() -> &'static Mutex { + RPC_TIMING_STATE.get_or_init(|| Mutex::new(RpcTimingState::default())) +} + +pub fn reset_rpc_timing() { + *rpc_timing_state().lock().expect("RPC timing mutex poisoned") = RpcTimingState::default(); +} + +pub fn rpc_timing_snapshot() -> RpcTimingSnapshot { + let state = rpc_timing_state().lock().expect("RPC timing mutex poisoned"); + let mut wait_elapsed = state.wait_elapsed; + + if let Some(wait_started_at) = state.wait_started_at { + wait_elapsed += wait_started_at.elapsed(); + } + + RpcTimingSnapshot { + wait_elapsed, + cumulative_call_elapsed: state.cumulative_call_elapsed, + calls: state.calls, + calls_by_method: state.calls_by_method.clone(), + } +} + +fn record_rpc_call_started(operation_name: &str) -> (Instant, String) { + let now = Instant::now(); + let mut state = rpc_timing_state().lock().expect("RPC timing mutex poisoned"); + + if state.active_calls == 0 { + state.wait_started_at = Some(now); + } + state.active_calls += 1; + + (now, rpc_method_name(operation_name).to_string()) +} + +fn record_rpc_call_finished(call_started_at: Instant, method_name: &str) { + let now = Instant::now(); + let mut state = rpc_timing_state().lock().expect("RPC timing mutex poisoned"); + + state.calls += 1; + *state.calls_by_method.entry(method_name.to_string()).or_default() += 1; + state.cumulative_call_elapsed += now.duration_since(call_started_at); + state.active_calls = state.active_calls.saturating_sub(1); + + if state.active_calls == 0 { + if let Some(wait_started_at) = state.wait_started_at.take() { + state.wait_elapsed += now.duration_since(wait_started_at); + } + } +} + +fn rpc_method_name(operation_name: &str) -> &str { + let base_name = operation_name.split(['(', ' ']).next().unwrap_or(operation_name); + + match base_name { + "get_nonce_at" => "get_nonce", + "get_compiled_class" + | "get_pre_snip34_compiled_class_hash" + | "get_compiled_class_hash_v1" + | "get_compiled_class_hash_v2" => "get_class", + method_name => method_name, + } +} + /// Executes a coroutine (future) in a Tokio runtime context. /// /// This function is useful for executing async code in contexts where you need to get a @@ -81,7 +166,11 @@ where loop { attempts += 1; - match f().await { + let (call_started_at, method_name) = record_rpc_call_started(operation_name); + let result = f().await; + record_rpc_call_finished(call_started_at, &method_name); + + match result { Ok(result) => { if attempts > 1 { warn!("{operation_name}: succeeded after {attempts} attempts"); @@ -113,3 +202,67 @@ where } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn rpc_timing_tracks_successful_rpc_wait() { + let before = rpc_timing_snapshot(); + + execute_with_retry("timed_sleep", || async { + sleep(Duration::from_millis(5)).await; + Ok::<_, ProviderError>(()) + }) + .await + .unwrap(); + + let after = rpc_timing_snapshot(); + assert!(after.calls >= before.calls + 1); + assert!( + after.calls_by_method.get("timed_sleep").copied().unwrap_or_default() + >= before.calls_by_method.get("timed_sleep").copied().unwrap_or_default() + 1 + ); + assert!(after.wait_elapsed >= before.wait_elapsed); + assert!(after.cumulative_call_elapsed >= before.cumulative_call_elapsed + Duration::from_millis(5)); + } + + #[tokio::test] + async fn rpc_timing_counts_overlapped_wait_once_for_wall_clock_summary() { + let before = rpc_timing_snapshot(); + + let first = execute_with_retry("timed_sleep_1", || async { + sleep(Duration::from_millis(20)).await; + Ok::<_, ProviderError>(()) + }); + let second = execute_with_retry("timed_sleep_2", || async { + sleep(Duration::from_millis(20)).await; + Ok::<_, ProviderError>(()) + }); + + let (first_result, second_result) = tokio::join!(first, second); + first_result.unwrap(); + second_result.unwrap(); + + let after = rpc_timing_snapshot(); + assert!(after.calls >= before.calls + 2); + assert!( + after.calls_by_method.get("timed_sleep_1").copied().unwrap_or_default() + >= before.calls_by_method.get("timed_sleep_1").copied().unwrap_or_default() + 1 + ); + assert!( + after.calls_by_method.get("timed_sleep_2").copied().unwrap_or_default() + >= before.calls_by_method.get("timed_sleep_2").copied().unwrap_or_default() + 1 + ); + assert!(after.wait_elapsed >= before.wait_elapsed); + assert!(after.cumulative_call_elapsed >= before.cumulative_call_elapsed + Duration::from_millis(40)); + } + + #[test] + fn rpc_method_name_normalizes_state_reader_helpers_to_rpc_methods() { + assert_eq!(rpc_method_name("get_nonce_at(contract: 0x123)"), "get_nonce"); + assert_eq!(rpc_method_name("get_compiled_class_hash_v2(class_hash: 0x456)"), "get_class"); + assert_eq!(rpc_method_name("get_proof(block_number: 1, keys: 3)"), "get_proof"); + } +}