diff --git a/cli/src/utils/processProverServer.ts b/cli/src/utils/processProverServer.ts index 15d47fdf0f..49792ff16f 100644 --- a/cli/src/utils/processProverServer.ts +++ b/cli/src/utils/processProverServer.ts @@ -17,6 +17,7 @@ import { const LIGHT_CONFIG_DIR = path.join(os.homedir(), ".config", "light"); const PROVER_BIN_DIR = path.join(LIGHT_CONFIG_DIR, "bin"); const KEYS_DIR = path.join(LIGHT_CONFIG_DIR, "proving-keys"); +const PROVING_KEYS_BASE_URL = "https://d1wbn9ra8wjh7t.cloudfront.net"; export async function killProver() { await killProcess(getProverNameByArch()); @@ -94,7 +95,8 @@ export async function startProver(proverPort: number, redisUrl?: string) { args.push("--keys-dir", KEYS_DIR + "/"); args.push("--prover-address", `0.0.0.0:${proverPort}`); - args.push("--auto-download", "true"); + args.push("--auto-download"); + args.push("--download-url", PROVING_KEYS_BASE_URL); if (redisUrl) { args.push("--redis-url", redisUrl); diff --git a/forester-utils/src/utils.rs b/forester-utils/src/utils.rs index b78cb88df6..0fa07e1546 100644 --- a/forester-utils/src/utils.rs +++ b/forester-utils/src/utils.rs @@ -5,8 +5,7 @@ use light_client::{ rpc::{Rpc, RpcError}, }; use solana_sdk::{signature::Signer, transaction::Transaction}; -use tokio::time::sleep; -use tracing::{error, warn}; +use tracing::error; use crate::error::ForesterUtilsError; @@ -30,10 +29,7 @@ pub async fn airdrop_lamports( pub async fn wait_for_indexer(rpc: &R) -> Result<(), ForesterUtilsError> { let rpc_slot = rpc.get_slot().await?; - - let indexer_slot = rpc.indexer()?.get_indexer_slot(None).await; - - let mut indexer_slot = match indexer_slot { + let indexer_slot = match rpc.indexer()?.get_indexer_slot(None).await { Ok(slot) => slot, Err(e) => { error!("failed to get indexer slot from indexer: {:?}", e); @@ -43,32 +39,15 @@ pub async fn wait_for_indexer(rpc: &R) -> Result<(), ForesterUtilsError> } }; - let max_attempts = 100; - let mut attempts = 0; - - while rpc_slot > indexer_slot { - if attempts >= max_attempts { - return Err(ForesterUtilsError::Indexer( - "Maximum attempts reached waiting for indexer to catch up".into(), - )); - } - - if rpc_slot - indexer_slot > 50 { - warn!( - "indexer is behind {} slots (rpc_slot: {}, indexer_slot: {})", - rpc_slot - indexer_slot, - rpc_slot, - indexer_slot - ); - } - - sleep(std::time::Duration::from_millis(1000)).await; - indexer_slot = rpc.indexer()?.get_indexer_slot(None).await.map_err(|e| { - error!("failed to get indexer slot from indexer: {:?}", e); - ForesterUtilsError::Indexer("Failed to get indexer slot".into()) - })?; - - attempts += 1; + let max_lag_slots = std::env::var("INDEXER_MAX_LAG_SLOTS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(100); + let lag = rpc_slot.saturating_sub(indexer_slot); + if lag > max_lag_slots { + return Err(ForesterUtilsError::Indexer(format!( + "Indexer is behind {lag} slots (maximum allowed: {max_lag_slots})" + ))); } Ok(()) } diff --git a/forester/src/api_server.rs b/forester/src/api_server.rs index d80c7757b9..fbd9109db2 100644 --- a/forester/src/api_server.rs +++ b/forester/src/api_server.rs @@ -1118,15 +1118,23 @@ pub fn spawn_api_server(config: ApiServerConfig) -> anyhow::Result, + #[arg( + long, + env = "PROVER_MAX_CONCURRENT_JOBS", + default_value = "4", + help = "Maximum queued or running prover jobs per forester (default: 4)" + )] + pub prover_max_concurrent_jobs: usize, + #[arg(long, env = "PAYER")] pub payer: Option, @@ -168,7 +176,7 @@ pub struct StartArgs { )] pub priority_fee_microlamports: Option, - #[arg(long, env = "RPC_POOL_SIZE", default_value = "100")] + #[arg(long, env = "RPC_POOL_SIZE", default_value = "32")] pub rpc_pool_size: u32, #[arg(long, env = "RPC_POOL_CONNECTION_TIMEOUT_SECS", default_value = "15")] @@ -303,6 +311,14 @@ pub struct StartArgs { )] pub enable_v1_multi_nullify: bool, + #[arg( + long, + env = "ENABLE_V1_PRESORT", + help = "Fetch queue leaf indices from the indexer (get_queue_leaf_indices) to pre-sort V1 work items for better dedup grouping. Requires an indexer that implements the endpoint. Best-effort; disabled by default.", + default_value = "false" + )] + pub enable_v1_presort: bool, + #[arg( long, env = "WORK_ITEM_BATCH_SIZE", diff --git a/forester/src/config.rs b/forester/src/config.rs index 8f88cecf22..1a59460eeb 100644 --- a/forester/src/config.rs +++ b/forester/src/config.rs @@ -36,6 +36,10 @@ pub struct ForesterConfig { /// Enable nullify_state_v1_multi instruction for batching 2-4 V1 state nullifications. /// Requires lookup_table_address to be set. pub enable_v1_multi_nullify: bool, + /// Enable the get_queue_leaf_indices pre-sort path for V1 work items. + /// Best-effort optimization; disabled by default. Only enable against an indexer + /// that implements the endpoint. + pub enable_v1_presort: bool, /// Number of queue items to process per batch cycle. Default: 50. pub work_item_batch_size: usize, } @@ -52,6 +56,7 @@ pub struct ExternalServicesConfig { pub prover_api_key: Option, pub prover_polling_interval: Option, pub prover_max_wait_time: Option, + pub prover_max_concurrent_jobs: usize, pub photon_grpc_url: Option, pub pushgateway_url: Option, pub pagerduty_routing_key: Option, @@ -284,6 +289,7 @@ impl ForesterConfig { prover_api_key: args.prover_api_key.clone(), prover_polling_interval: args.prover_polling_interval_ms.map(Duration::from_millis), prover_max_wait_time: args.prover_max_wait_time_secs.map(Duration::from_secs), + prover_max_concurrent_jobs: args.prover_max_concurrent_jobs, photon_grpc_url: args.photon_grpc_url.clone(), pushgateway_url: args.push_gateway_url.clone(), pagerduty_routing_key: args.pagerduty_routing_key.clone(), @@ -431,6 +437,7 @@ impl ForesterConfig { .transpose()?, min_queue_items: args.min_queue_items, enable_v1_multi_nullify: args.enable_v1_multi_nullify, + enable_v1_presort: args.enable_v1_presort, work_item_batch_size: args.work_item_batch_size.unwrap_or(50) as usize, }) } @@ -450,6 +457,7 @@ impl ForesterConfig { prover_api_key: None, prover_polling_interval: None, prover_max_wait_time: None, + prover_max_concurrent_jobs: 4, photon_grpc_url: None, pushgateway_url: args.push_gateway_url.clone(), pagerduty_routing_key: args.pagerduty_routing_key.clone(), @@ -488,6 +496,7 @@ impl ForesterConfig { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, work_item_batch_size: 50, }) } @@ -511,6 +520,7 @@ impl Clone for ForesterConfig { lookup_table_address: self.lookup_table_address, min_queue_items: self.min_queue_items, enable_v1_multi_nullify: self.enable_v1_multi_nullify, + enable_v1_presort: self.enable_v1_presort, work_item_batch_size: self.work_item_batch_size, } } diff --git a/forester/src/epoch_manager.rs b/forester/src/epoch_manager.rs index a253b4bad3..9f7a15be50 100644 --- a/forester/src/epoch_manager.rs +++ b/forester/src/epoch_manager.rs @@ -13,6 +13,7 @@ use dashmap::DashMap; use forester_utils::{ forester_epoch::{get_epoch_phases, Epoch, ForesterSlot, TreeAccounts, TreeForesterSchedule}, rpc_pool::SolanaRpcPool, + utils::wait_for_indexer, }; use futures::future::join_all; use light_client::{ @@ -39,7 +40,7 @@ use solana_sdk::{ transaction::TransactionError, }; use tokio::{ - sync::{mpsc, oneshot, Mutex}, + sync::{mpsc, oneshot, Mutex, Semaphore}, task::JoinHandle, time::{sleep, Instant, MissedTickBehavior}, }; @@ -280,6 +281,8 @@ pub struct EpochManager { run_id: Arc, /// Per-epoch registration trackers to coordinate re-finalization when new foresters register mid-epoch registration_trackers: Arc>>, + /// Process-wide limiter for V1 transaction sends across all trees. + v1_send_permits: Arc, } impl Clone for EpochManager { @@ -310,6 +313,7 @@ impl Clone for EpochManager { heartbeat: self.heartbeat.clone(), run_id: self.run_id.clone(), registration_trackers: self.registration_trackers.clone(), + v1_send_permits: self.v1_send_permits.clone(), } } } @@ -333,6 +337,7 @@ impl EpochManager { run_id: String, ) -> Result { let authority = Arc::new(config.payer_keypair.insecure_clone()); + let v1_send_permit_count = config.transaction_config.max_concurrent_sends.max(1); Ok(Self { config, protocol_config, @@ -359,6 +364,7 @@ impl EpochManager { heartbeat, run_id: Arc::::from(run_id), registration_trackers: Arc::new(DashMap::new()), + v1_send_permits: Arc::new(Semaphore::new(v1_send_permit_count)), }) } @@ -2242,6 +2248,19 @@ impl EpochManager { ) .await?; + if let Err(error) = wait_for_indexer(&*rpc).await { + if should_emit_rate_limited_warning("v2_wait_for_indexer", Duration::from_secs(30)) { + warn!( + event = "v2_wait_for_indexer_error", + run_id = %self.run_id, + tree = %tree_pubkey, + error = %error, + "Skipping V2 proof work because the indexer is not fresh" + ); + } + return Ok(()); + } + // Try to send any cached proofs first let cached_send_start = Instant::now(); if let Some(items_sent) = self @@ -2265,8 +2284,14 @@ impl EpochManager { let mut estimated_slot = self.slot_tracker.estimated_current_slot(); - // Polling interval for checking queue - const POLL_INTERVAL: Duration = Duration::from_millis(200); + // Adaptive queue polling: start responsive, then back off (capped) while the + // queue has nothing ready to process, and reset to the minimum as soon as work + // is found. A fixed 200ms poll made idle V2 trees re-fetch the queue ~5x/sec for + // the whole eligible window, which is the dominant source of wasted RPC/indexer + // load (and can exhaust a shared RPC credit budget). + const POLL_INTERVAL_MIN: Duration = Duration::from_millis(200); + const POLL_INTERVAL_MAX: Duration = Duration::from_secs(10); + let mut poll_interval = POLL_INTERVAL_MIN; 'inner_processing_loop: loop { if estimated_slot >= forester_slot_details.end_solana_slot { @@ -2334,9 +2359,12 @@ impl EpochManager { processing_start_time.elapsed(), ) .await; + // Work found: stay responsive while the queue drains. + poll_interval = POLL_INTERVAL_MIN; } else { - // No items to process, wait before polling again - tokio::time::sleep(POLL_INTERVAL).await; + // Nothing ready: wait, then back off (capped) to avoid hammering RPC. + tokio::time::sleep(poll_interval).await; + poll_interval = (poll_interval * 2).min(POLL_INTERVAL_MAX); } } Err(e) => { @@ -2350,7 +2378,8 @@ impl EpochManager { error = ?e, "V2 processing failed for tree" ); - tokio::time::sleep(POLL_INTERVAL).await; + tokio::time::sleep(poll_interval).await; + poll_interval = (poll_interval * 2).min(POLL_INTERVAL_MAX); } } @@ -3062,8 +3091,7 @@ impl EpochManager { } else { None }, - enable_presort: self.config.enable_v1_multi_nullify - && !self.address_lookup_tables.is_empty(), + enable_presort: self.config.enable_v1_presort, work_item_batch_size: self.config.work_item_batch_size, }; @@ -3083,6 +3111,7 @@ impl EpochManager { &batched_tx_config, *tree_accounts, transaction_builder, + self.v1_send_permits.clone(), ) .await?; @@ -3161,6 +3190,9 @@ impl EpochManager { .external_services .prover_max_wait_time .unwrap_or(Duration::from_secs(600)), + max_concurrent_jobs: self.config.external_services.prover_max_concurrent_jobs, + network: std::env::var("FORESTER_NETWORK") + .unwrap_or_else(|_| "default".to_string()), }), ops_cache: self.ops_cache.clone(), epoch_phases: epoch_info.phases.clone(), @@ -4703,6 +4735,7 @@ mod tests { send_tx_rate_limit: None, prover_polling_interval: None, prover_max_wait_time: None, + prover_max_concurrent_jobs: 4, fallback_rpc_url: None, fallback_indexer_url: None, }, @@ -4730,6 +4763,7 @@ mod tests { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, work_item_batch_size: 50, } } diff --git a/forester/src/processor/v1/helpers.rs b/forester/src/processor/v1/helpers.rs index 8af0e9c4ba..295d69cf30 100644 --- a/forester/src/processor/v1/helpers.rs +++ b/forester/src/processor/v1/helpers.rs @@ -96,15 +96,16 @@ pub async fn fetch_proofs_and_create_instructions( }; let rpc = pool.get_connection().await?; - if let Err(e) = wait_for_indexer(&*rpc).await { + wait_for_indexer(&*rpc).await.map_err(|e| { if should_emit_rate_limited_warning("v1_wait_for_indexer", Duration::from_secs(30)) { warn!( event = "v1_wait_for_indexer_error", error = %e, - "Indexer not fully caught up, but proceeding anyway" + "Skipping V1 proof work because the indexer is not fresh" ); } - } + e + })?; let address_proofs = if let Some((merkle_tree, addresses)) = address_data { let total_addresses = addresses.len(); diff --git a/forester/src/processor/v1/send_transaction.rs b/forester/src/processor/v1/send_transaction.rs index 86e7bfb650..28ae8886ac 100644 --- a/forester/src/processor/v1/send_transaction.rs +++ b/forester/src/processor/v1/send_transaction.rs @@ -16,7 +16,10 @@ use solana_sdk::{ pubkey::Pubkey, signature::{Keypair, Signature, Signer}, }; -use tokio::time::Instant; +use tokio::{ + sync::{OwnedSemaphorePermit, Semaphore}, + time::{timeout, Instant}, +}; use tracing::{error, info, trace, warn}; use crate::{ @@ -36,10 +39,21 @@ struct PreparedBatchData { timeout_deadline: Instant, } +// Stay within the merkle tree changelog capacity even when the configured +// per-cycle batch size is larger than expected. +const MAX_ITEMS_PER_CYCLE: usize = 1400; + +// Keep one tree from reserving the entire process-wide send semaphore while it +// waits on proof building and transaction confirmation. +const MAX_WORK_ITEMS_PER_SEND_CHUNK: usize = 4; +const MAX_CONCURRENT_CHUNKS_PER_TREE: usize = 1; + #[derive(Clone)] struct ChunkSendContext { pool: Arc>, max_concurrent_sends: usize, + send_permits: Arc, + permits_pre_acquired: bool, timeout_deadline: Instant, cancel_signal: Arc, num_sent_transactions: Arc, @@ -67,6 +81,7 @@ pub async fn send_batched_transactions, + send_permits: Arc, ) -> std::result::Result { let function_start_time = Instant::now(); @@ -144,37 +159,53 @@ pub async fn send_batched_transactions MAX_ITEMS_PER_CYCLE { - &data.work_items[..MAX_ITEMS_PER_CYCLE] - } else { - &data.work_items - }; + let items_to_process = &data.work_items[..cycle_item_count]; - // Process all chunks concurrently: each chunk fetches proofs, builds, and sends in parallel. + // Process chunks concurrently: each chunk fetches proofs and builds independently, + // while transaction sends share the process-wide V1 concurrency limit. let chunks: Vec> = items_to_process .chunks(work_item_batch_size) .map(|c| c.to_vec()) .collect(); let num_chunks = chunks.len(); + let max_concurrent_chunks = MAX_CONCURRENT_CHUNKS_PER_TREE.min(num_chunks.max(1)); info!( tree = %tree_accounts.merkle_tree, - "Processing {} concurrent chunks of up to {} items each", - num_chunks, work_item_batch_size + "Processing {} chunks of up to {} items each with chunk_concurrency={} and send_concurrency={}", + num_chunks, + work_item_batch_size, + max_concurrent_chunks, + effective_max_concurrent_sends ); let chunk_futures: Vec<_> = chunks @@ -190,13 +221,31 @@ pub async fn send_batched_transactions= timeout_deadline { return Ok(()); } - // Each chunk gets a fresh blockhash + let chunk_admission_permit = + match acquire_send_permits(&send_permits, 1, timeout_deadline).await { + Ok(permit) => permit, + Err(e) => { + warn!( + tree = %tree_id, + error = ?e, + "Skipping chunk because send concurrency permits were not available before deadline" + ); + return Ok(()); + } + }; + + if cancel_signal.load(Ordering::SeqCst) || Instant::now() >= timeout_deadline { + return Ok(()); + } + + // Each chunk gets a fresh blockhash after it owns send capacity. let (recent_blockhash, last_valid_block_height) = { let mut rpc = pool.get_connection().await.map_err(ForesterError::from)?; rpc.get_latest_blockhash().await.map_err(|e| { @@ -229,9 +278,13 @@ pub async fn send_batched_transactions>() + .await; for result in results { if let Err(ForesterError::NotEligible) = result { return Err(ForesterError::NotEligible); @@ -392,6 +448,49 @@ fn compute_effective_max_concurrent_sends( effective.max(1) } +fn cycle_item_limit(configured_work_item_batch_size: usize, work_item_count: usize) -> usize { + let limit = configured_work_item_batch_size.clamp(1, MAX_ITEMS_PER_CYCLE); + work_item_count.min(limit) +} + +fn chunk_item_limit( + configured_work_item_batch_size: usize, + effective_max_concurrent_sends: usize, +) -> usize { + configured_work_item_batch_size + .max(1) + .min(effective_max_concurrent_sends.max(1)) + .min(MAX_WORK_ITEMS_PER_SEND_CHUNK) +} + +async fn acquire_send_permits( + send_permits: &Arc, + permit_count: usize, + timeout_deadline: Instant, +) -> std::result::Result { + if Instant::now() >= timeout_deadline { + return Err(ForesterError::General { + error: "Timed out waiting for send concurrency permits".to_string(), + }); + } + + let permit_count = u32::try_from(permit_count).map_err(|_| ForesterError::General { + error: "send concurrency permit count exceeds supported limit".to_string(), + })?; + let permit_wait = timeout_deadline.saturating_duration_since(Instant::now()); + timeout( + permit_wait, + send_permits.clone().acquire_many_owned(permit_count), + ) + .await + .map_err(|_| ForesterError::General { + error: "Timed out waiting for send concurrency permits".to_string(), + })? + .map_err(|_| ForesterError::General { + error: "send concurrency limiter closed".to_string(), + }) +} + async fn execute_transaction_chunk_sending( transactions: Vec, context: &ChunkSendContext, @@ -406,11 +505,14 @@ async fn execute_transaction_chunk_sending( let num_sent_transactions = Arc::clone(&context.num_sent_transactions); let timeout_deadline = context.timeout_deadline; let max_concurrent_sends = context.max_concurrent_sends; + let send_permits = Arc::clone(&context.send_permits); + let permits_pre_acquired = context.permits_pre_acquired; let confirmation = context.confirmation; let transaction_send_futures = transactions.into_iter().map(|prepared_transaction| { let pool_clone = Arc::clone(&pool); let cancel_signal_clone = Arc::clone(&cancel_signal); let num_sent_transactions_clone = Arc::clone(&num_sent_transactions); + let send_permits_clone = Arc::clone(&send_permits); let tx_label = prepared_transaction.label().to_string(); async move { @@ -423,6 +525,32 @@ async fn execute_transaction_chunk_sending( .unwrap_or_default(); let tx_signature_str = tx_signature.to_string(); + let _send_permit = if permits_pre_acquired { + None + } else { + let permit_wait = timeout_deadline.saturating_duration_since(Instant::now()); + match timeout(permit_wait, send_permits_clone.acquire_owned()).await { + Ok(Ok(permit)) => Some(permit), + Ok(Err(_)) => { + error!( + tx.signature_attempt = %tx_signature_str, + "Send concurrency limiter was closed" + ); + return TransactionSendResult::SendFailure( + ForesterError::General { + error: "send concurrency limiter closed".to_string(), + }, + Some(tx_signature), + ); + } + Err(_) => return TransactionSendResult::Timeout, + } + }; + + if cancel_signal_clone.load(Ordering::SeqCst) || Instant::now() >= timeout_deadline { + return TransactionSendResult::Cancelled; + } + match pool_clone.get_connection().await { Ok(mut rpc) => { if Instant::now() >= timeout_deadline { @@ -562,3 +690,83 @@ async fn execute_transaction_chunk_sending( } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn acquire_send_permits_acquires_requested_permits() { + let send_permits = Arc::new(Semaphore::new(3)); + let permit = acquire_send_permits( + &send_permits, + 2, + Instant::now() + Duration::from_millis(100), + ) + .await + .expect("permit reservation should succeed"); + + assert_eq!(send_permits.available_permits(), 1); + + drop(permit); + assert_eq!(send_permits.available_permits(), 3); + } + + #[tokio::test] + async fn acquire_send_permits_waits_for_full_reservation() { + let send_permits = Arc::new(Semaphore::new(2)); + let first_permit = acquire_send_permits( + &send_permits, + 2, + Instant::now() + Duration::from_millis(100), + ) + .await + .expect("first reservation should succeed"); + let waiter_permits = Arc::clone(&send_permits); + + let waiter = tokio::spawn(async move { + acquire_send_permits( + &waiter_permits, + 2, + Instant::now() + Duration::from_millis(500), + ) + .await + }); + + tokio::time::sleep(Duration::from_millis(25)).await; + + assert!(!waiter.is_finished()); + + drop(first_permit); + let second_permit = timeout(Duration::from_millis(100), waiter) + .await + .expect("second reservation should complete after permits are released") + .expect("waiter task should not panic") + .expect("second reservation should succeed"); + + assert_eq!(send_permits.available_permits(), 0); + + drop(second_permit); + assert_eq!(send_permits.available_permits(), 2); + } + + #[test] + fn cycle_item_limit_uses_configured_batch_size_as_total_cycle_cap() { + assert_eq!(cycle_item_limit(50, 1252), 50); + assert_eq!(cycle_item_limit(50, 12), 12); + } + + #[test] + fn cycle_item_limit_stays_within_changelog_cap() { + assert_eq!(cycle_item_limit(10_000, 2_000), MAX_ITEMS_PER_CYCLE); + assert_eq!(cycle_item_limit(0, 10), 1); + } + + #[test] + fn chunk_item_limit_caps_single_tree_send_reservation() { + assert_eq!(chunk_item_limit(50, 10), MAX_WORK_ITEMS_PER_SEND_CHUNK); + assert_eq!(chunk_item_limit(3, 10), 3); + assert_eq!(chunk_item_limit(50, 2), 2); + assert_eq!(chunk_item_limit(0, 10), 1); + } +} diff --git a/forester/src/processor/v1/tx_builder.rs b/forester/src/processor/v1/tx_builder.rs index 36fee1eafc..f1c46d1dfd 100644 --- a/forester/src/processor/v1/tx_builder.rs +++ b/forester/src/processor/v1/tx_builder.rs @@ -1,9 +1,10 @@ -use std::{sync::Arc, time::Duration}; +use std::{collections::HashSet, sync::Arc, time::Duration}; use account_compression::processor::initialize_address_merkle_tree::Pubkey; use async_trait::async_trait; use forester_utils::rpc_pool::SolanaRpcPool; use light_client::rpc::Rpc; +use light_registry::account_compression_cpi::sdk::nullify_state_v1_multi_lookup_table_accounts; use solana_program::hash::Hash; use solana_sdk::{ address_lookup_table::AddressLookupTableAccount, @@ -11,7 +12,7 @@ use solana_sdk::{ signature::{Keypair, Signer}, }; use tokio::sync::Mutex; -use tracing::{trace, warn}; +use tracing::{debug, trace, warn}; use crate::{ epoch_manager::WorkItem, @@ -69,6 +70,31 @@ impl EpochManagerTransactions { } } +fn lookup_tables_cover_v1_multi_nullify( + lookup_tables: &[AddressLookupTableAccount], + work_items: &[WorkItem], +) -> bool { + let Some(first_item) = work_items.first() else { + return false; + }; + if lookup_tables.is_empty() { + return false; + } + + let required_accounts = nullify_state_v1_multi_lookup_table_accounts( + first_item.tree_account.merkle_tree, + first_item.tree_account.queue, + ); + let lookup_accounts = lookup_tables + .iter() + .flat_map(|table| table.addresses.iter().copied()) + .collect::>(); + + required_accounts + .iter() + .all(|account| lookup_accounts.contains(account)) +} + #[async_trait] impl TransactionBuilder for EpochManagerTransactions { fn epoch(&self) -> u64 { @@ -124,8 +150,19 @@ impl TransactionBuilder for EpochManagerTransactions { .map(|&item| item.clone()) .collect::>(); - let use_multi_nullify = - self.enable_v1_multi_nullify && !self.address_lookup_tables.is_empty(); + let use_multi_nullify = self.enable_v1_multi_nullify + && lookup_tables_cover_v1_multi_nullify(&self.address_lookup_tables, &work_items); + if self.enable_v1_multi_nullify && !use_multi_nullify { + let tree = work_items + .first() + .map(|item| item.tree_account.merkle_tree.to_string()) + .unwrap_or_default(); + debug!( + tree, + lookup_tables = self.address_lookup_tables.len(), + "V1 multi-nullify disabled for batch because lookup tables do not cover required accounts" + ); + } let mut transactions = vec![]; let all_instructions = match fetch_proofs_and_create_instructions( payer.pubkey(), diff --git a/forester/src/processor/v2/common.rs b/forester/src/processor/v2/common.rs index 191160bea4..80e2abc29f 100644 --- a/forester/src/processor/v2/common.rs +++ b/forester/src/processor/v2/common.rs @@ -79,6 +79,8 @@ pub struct ProverConfig { pub api_key: Option, pub polling_interval: Duration, pub max_wait_time: Duration, + pub max_concurrent_jobs: usize, + pub network: String, } #[derive(Debug)] diff --git a/forester/src/processor/v2/processor.rs b/forester/src/processor/v2/processor.rs index 3de6dea860..4441cab3c5 100644 --- a/forester/src/processor/v2/processor.rs +++ b/forester/src/processor/v2/processor.rs @@ -4,7 +4,7 @@ use std::{ }; use anyhow::anyhow; -use forester_utils::{forester_epoch::EpochPhases, utils::wait_for_indexer}; +use forester_utils::forester_epoch::EpochPhases; use light_client::rpc::Rpc; use light_compressed_account::QueueType; use solana_sdk::pubkey::Pubkey; @@ -14,7 +14,6 @@ use tracing::{debug, info, warn}; use crate::{ epoch_manager::{CircuitMetrics, ProcessingMetrics}, errors::ForesterError, - logging::should_emit_rate_limited_warning, processor::v2::{ batch_job_builder::BatchJobBuilder, common::WorkerPool, @@ -136,6 +135,20 @@ where self.worker_pool = Some(WorkerPool { job_tx }); } + if self.cached_state.is_some() { + let onchain_root = self.strategy.fetch_onchain_root(&self.context).await?; + if onchain_root != self.current_root { + warn!( + tree = %self.context.merkle_tree, + expected_root_prefix = ?&self.current_root[..4], + onchain_root_prefix = ?&onchain_root[..4], + "Discarding cached proof state because the on-chain root advanced" + ); + self.current_root = onchain_root; + self.clear_cache().await; + } + } + if let Some(cached) = self.cached_state.take() { let actual_available = self .strategy @@ -188,26 +201,6 @@ where ); } - { - let rpc = self.context.rpc_pool.get_connection().await?; - if let Err(e) = wait_for_indexer(&*rpc).await { - if should_emit_rate_limited_warning("v2_wait_for_indexer", Duration::from_secs(30)) - { - warn!( - event = "wait_for_indexer_error", - error = %e, - "wait_for_indexer error (proceeding anyway)" - ); - } else { - debug!( - event = "wait_for_indexer_error_suppressed", - error = %e, - "Suppressing repeated wait_for_indexer warning" - ); - } - } - } - let queue_data = match self .strategy .fetch_queue_data(&self.context, fetch_batches, self.zkp_batch_size) @@ -217,14 +210,6 @@ where None => return Ok(ProcessingResult::default()), }; - if self.current_root == [0u8; 32] || queue_data.initial_root == self.current_root { - let total_batches = queue_data.num_batches; - let process_now = total_batches.min(self.context.max_batches_per_tree); - return self - .process_batches(queue_data, 0, process_now, total_batches) - .await; - } - let onchain_root = self.strategy.fetch_onchain_root(&self.context).await?; match reconcile_roots(self.current_root, queue_data.initial_root, onchain_root) { RootReconcileDecision::Proceed => { diff --git a/forester/src/processor/v2/proof_worker.rs b/forester/src/processor/v2/proof_worker.rs index b7afeacf0b..98a81f24fb 100644 --- a/forester/src/processor/v2/proof_worker.rs +++ b/forester/src/processor/v2/proof_worker.rs @@ -1,6 +1,12 @@ -use std::{sync::Arc, time::Duration}; +use std::{ + sync::{Arc, OnceLock}, + time::Duration, +}; -const MAX_CONCURRENT_PROOFS: usize = 64; +#[cfg(not(test))] +const QUEUE_FULL_RETRY_DELAY: Duration = Duration::from_secs(5); +#[cfg(test)] +const QUEUE_FULL_RETRY_DELAY: Duration = Duration::from_millis(10); use async_channel::Receiver; use light_batched_merkle_tree::merkle_tree::{ @@ -139,19 +145,22 @@ impl ProofClients { config.polling_interval, config.max_wait_time, config.api_key.clone(), - ), + ) + .with_network(config.network.clone()), nullify_client: ProofClient::with_config( config.update_url.clone(), config.polling_interval, config.max_wait_time, config.api_key.clone(), - ), + ) + .with_network(config.network.clone()), address_append_client: ProofClient::with_config( config.address_append_url.clone(), config.polling_interval, config.max_wait_time, config.api_key.clone(), - ), + ) + .with_network(config.network.clone()), } } @@ -167,16 +176,23 @@ impl ProofClients { pub fn spawn_proof_workers(config: &ProverConfig) -> async_channel::Sender { let (job_tx, job_rx) = async_channel::bounded::(256); let clients = Arc::new(ProofClients::new(config)); - tokio::spawn(async move { run_proof_pipeline(job_rx, clients).await }); + let semaphore = process_proof_semaphore(config.max_concurrent_jobs); + tokio::spawn(async move { run_proof_pipeline(job_rx, clients, semaphore).await }); job_tx } +fn process_proof_semaphore(max_concurrent_jobs: usize) -> Arc { + static SEMAPHORE: OnceLock> = OnceLock::new(); + SEMAPHORE + .get_or_init(|| Arc::new(tokio::sync::Semaphore::new(max_concurrent_jobs.max(1)))) + .clone() +} + async fn run_proof_pipeline( job_rx: Receiver, clients: Arc, + semaphore: Arc, ) -> crate::Result<()> { - let semaphore = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_PROOFS)); - while let Ok(job) = job_rx.recv().await { let clients = clients.clone(); let permit = semaphore.clone().acquire_owned().await; @@ -199,7 +215,13 @@ async fn submit_and_poll_proof(clients: Arc, job: ProofJob) { let round_trip_start = std::time::Instant::now(); - match client.submit_proof_async(inputs_json, circuit_type).await { + let Some(submit_result) = + submit_with_backpressure(client, &inputs_json, circuit_type, job.seq, &job.result_tx).await + else { + return; + }; + + match submit_result { Ok(SubmitProofResult::Queued(job_id)) => { debug!( "Submitted proof job seq={} type={} job_id={}", @@ -248,6 +270,58 @@ async fn submit_and_poll_proof(clients: Arc, job: ProofJob) { } } +async fn submit_with_backpressure( + client: &ProofClient, + inputs_json: &str, + circuit_type: &str, + seq: u64, + result_tx: &mpsc::Sender, +) -> Option> { + let mut attempts = 0u64; + loop { + if result_tx.is_closed() { + debug!( + "Stopping proof submission retry for seq={}: result channel closed", + seq + ); + return None; + } + + match client + .submit_proof_async(inputs_json.to_string(), circuit_type) + .await + { + Err(error) if is_queue_full(&error) => { + attempts += 1; + let jitter = Duration::from_millis(seq.wrapping_add(attempts) % 1000); + let delay = QUEUE_FULL_RETRY_DELAY + jitter; + warn!( + "Prover queue full for seq={} type={}; retrying in {:?} (attempt {})", + seq, circuit_type, delay, attempts + ); + tokio::select! { + _ = tokio::time::sleep(delay) => {} + _ = result_tx.closed() => { + debug!( + "Stopping proof submission retry for seq={}: result channel closed", + seq + ); + return None; + } + } + } + result => return Some(result), + } + } +} + +fn is_queue_full(err: &ProverClientError) -> bool { + matches!( + err, + ProverClientError::ProverServerError(message) if message.contains("queue_full") + ) +} + async fn poll_and_send_result( clients: Arc, job_id: String, @@ -278,7 +352,13 @@ async fn poll_and_send_result( let inputs_json = inputs.to_json(&tree_id, seq); let circuit_type = inputs.circuit_type(); - match client.submit_proof_async(inputs_json, circuit_type).await { + let Some(submit_result) = + submit_with_backpressure(client, &inputs_json, circuit_type, seq, &result_tx).await + else { + return; + }; + + match submit_result { Ok(SubmitProofResult::Queued(new_job_id)) => { debug!( "Resubmitted proof job seq={} type={} new_job_id={}", @@ -421,3 +501,87 @@ fn build_proof_result( submitted_at, } } + +#[cfg(test)] +mod tests { + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + }; + + use super::*; + + #[test] + fn queue_full_detection_is_specific() { + let queue_full = ProverClientError::ProverServerError( + "Prover server error: queue_full - network proof queue is full".to_string(), + ); + let other = ProverClientError::ProverServerError("invalid_input".to_string()); + + assert!(is_queue_full(&queue_full)); + assert!(!is_queue_full(&other)); + } + + #[tokio::test] + async fn retries_queue_full_until_job_is_accepted() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let request_count = Arc::new(AtomicUsize::new(0)); + let server_request_count = request_count.clone(); + + let server = tokio::spawn(async move { + for request_index in 0..2 { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = vec![0u8; 16 * 1024]; + let _ = stream.read(&mut request).await.unwrap(); + server_request_count.fetch_add(1, Ordering::SeqCst); + + let (status, body) = if request_index == 0 { + ( + "429 Too Many Requests", + r#"{"code":"queue_full","message":"network proof queue is full"}"#, + ) + } else { + ( + "202 Accepted", + r#"{"job_id":"job-1","estimated_time":null}"#, + ) + }; + let response = format!( + "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + } + }); + + let client = ProofClient::with_config( + format!("http://{address}"), + Duration::from_millis(10), + Duration::from_secs(1), + None, + ); + let (result_tx, _result_rx) = mpsc::channel(1); + + let submit_result = tokio::time::timeout( + Duration::from_secs(2), + submit_with_backpressure(&client, "{}", "append", 7, &result_tx), + ) + .await + .expect("submission retry timed out") + .expect("result channel unexpectedly closed") + .expect("submission failed after queue became available"); + + match submit_result { + SubmitProofResult::Queued(job_id) => assert_eq!(job_id, "job-1"), + SubmitProofResult::Immediate(_) => panic!("expected queued proof job"), + } + server.await.unwrap(); + assert_eq!(request_count.load(Ordering::SeqCst), 2); + } +} diff --git a/forester/src/processor/v2/root_guard.rs b/forester/src/processor/v2/root_guard.rs index 8416204454..472d3030df 100644 --- a/forester/src/processor/v2/root_guard.rs +++ b/forester/src/processor/v2/root_guard.rs @@ -29,29 +29,15 @@ pub fn reconcile_roots( indexer_root: [u8; 32], onchain_root: [u8; 32], ) -> RootReconcileDecision { - if expected_root == [0u8; 32] { - // Uninitialized expected root — proceed but adopt the indexer root. - // Validate that indexer and on-chain agree when possible. - if indexer_root != onchain_root { - tracing::warn!( - "Proceeding with uninitialized expected root, but indexer root ({:?}) != onchain root ({:?}). Indexer may be stale.", - &indexer_root[..4], - &onchain_root[..4], - ); - } + if indexer_root == onchain_root && expected_root == onchain_root { return RootReconcileDecision::Proceed; } - if indexer_root == expected_root { - return RootReconcileDecision::Proceed; - } - - if onchain_root == expected_root { - return RootReconcileDecision::WaitForIndexer; - } - if indexer_root == onchain_root { return RootReconcileDecision::ResetToOnchainAndProceed(onchain_root); } + if expected_root == onchain_root { + return RootReconcileDecision::WaitForIndexer; + } RootReconcileDecision::ResetToOnchainAndStop(onchain_root) } @@ -93,18 +79,26 @@ mod tests { } #[test] - fn proceeds_when_expected_is_zero() { + fn cold_start_stops_when_indexer_and_chain_disagree() { assert_eq!( reconcile_roots(root(0), root(1), root(2)), - RootReconcileDecision::Proceed + RootReconcileDecision::ResetToOnchainAndStop(root(2)) + ); + } + + #[test] + fn cold_start_proceeds_only_when_indexer_matches_chain() { + assert_eq!( + reconcile_roots(root(0), root(2), root(2)), + RootReconcileDecision::ResetToOnchainAndProceed(root(2)) ); } #[test] - fn proceeds_when_expected_matches_indexer() { + fn stops_when_expected_matches_stale_indexer() { assert_eq!( reconcile_roots(root(9), root(9), root(8)), - RootReconcileDecision::Proceed + RootReconcileDecision::ResetToOnchainAndStop(root(8)) ); } diff --git a/forester/src/smart_transaction.rs b/forester/src/smart_transaction.rs index 7f093b0f26..66b5762c1b 100644 --- a/forester/src/smart_transaction.rs +++ b/forester/src/smart_transaction.rs @@ -18,6 +18,7 @@ use solana_sdk::{ use solana_transaction_status::TransactionConfirmationStatus; use thiserror::Error; use tokio::time::{sleep, Instant}; +use tracing::{debug, warn}; use crate::{ errors::{is_blockhash_not_found, rpc_is_already_processed}, @@ -45,6 +46,10 @@ impl Default for ConfirmationConfig { } } +// Let RPC-side retries fan out before doing another app-level send request. +const MIN_TRANSACTION_RESEND_INTERVAL: Duration = Duration::from_secs(5); +const RPC_SEND_TRANSACTION_MAX_RETRIES: usize = 3; + #[derive(Debug, Clone, Copy)] pub struct TransactionPolicy { pub priority_fee_config: PriorityFeeConfig, @@ -439,7 +444,14 @@ async fn send_prepared_transaction( .ok_or_else(|| RpcError::CustomError("Prepared transaction missing signature".into()))?; let last_valid_block_height = transaction.last_valid_block_height(); let rpc = &*rpc; + let started_at = Instant::now(); + let mut send_attempts = 0; let mut last_send_error = None; + let resend_interval = confirmation + .poll_interval + .saturating_mul(4) + .max(MIN_TRANSACTION_RESEND_INTERVAL); + let mut next_send_at = Instant::now(); for attempt in 0..confirmation.max_attempts { if let Some(signature) = confirmed_signature_or_error(rpc, signature).await? { @@ -450,23 +462,53 @@ async fn send_prepared_transaction( return Err(SmartTransactionError::ConfirmationDeadlineExceeded { signature }); } - if rpc.get_block_height().await? > last_valid_block_height { + let current_block_height = rpc.get_block_height().await?; + if current_block_height > last_valid_block_height { + log_blockhash_expired( + transaction, + signature, + last_valid_block_height, + current_block_height, + send_attempts, + started_at.elapsed(), + last_send_error.as_ref(), + ); return Err(SmartTransactionError::BlockhashExpired { signature, last_valid_block_height, }); } - match transaction.send_with_confirmation_config(rpc).await { - Ok(_) => last_send_error = None, - Err(error) if rpc_is_already_processed(&error) => last_send_error = None, - // BlockhashNotFound is transient: the sending RPC may not have - // propagated the blockhash from the fetcher yet. Retry until the - // blockhash either propagates or the outer `get_block_height > - // last_valid_block_height` check exits with BlockhashExpired. - Err(error) if is_blockhash_not_found(&error) => last_send_error = Some(error), - Err(error) if rpc.should_retry(&error) => last_send_error = Some(error), - Err(error) => return Err(error.into()), + if Instant::now() >= next_send_at { + send_attempts += 1; + match transaction.send_with_confirmation_config(rpc).await { + Ok(_) => last_send_error = None, + Err(error) if rpc_is_already_processed(&error) => last_send_error = None, + // BlockhashNotFound is transient: the sending RPC may not have + // propagated the blockhash from the fetcher yet. Retry until the + // blockhash either propagates or the outer `get_block_height > + // last_valid_block_height` check exits with BlockhashExpired. + Err(error) if is_blockhash_not_found(&error) => { + debug_retryable_send_error(transaction, signature, send_attempts, &error); + last_send_error = Some(error); + } + Err(error) if is_permanent_send_error(&error) => { + warn!( + signature = %signature, + label = %transaction.label(), + send_attempts, + error = ?error, + "Permanent transaction send error" + ); + return Err(error.into()); + } + Err(error) if rpc.should_retry(&error) => { + debug_retryable_send_error(transaction, signature, send_attempts, &error); + last_send_error = Some(error); + } + Err(error) => return Err(error.into()), + } + next_send_at = Instant::now() + resend_interval; } if let Some(signature) = confirmed_signature_or_error(rpc, signature).await? { @@ -491,7 +533,17 @@ async fn send_prepared_transaction( return Ok(signature); } - if rpc.get_block_height().await? > last_valid_block_height { + let current_block_height = rpc.get_block_height().await?; + if current_block_height > last_valid_block_height { + log_blockhash_expired( + transaction, + signature, + last_valid_block_height, + current_block_height, + send_attempts, + started_at.elapsed(), + last_send_error.as_ref(), + ); return Err(SmartTransactionError::BlockhashExpired { signature, last_valid_block_height, @@ -499,6 +551,14 @@ async fn send_prepared_transaction( } if let Some(error) = last_send_error { + warn!( + signature = %signature, + label = %transaction.label(), + send_attempts, + elapsed_ms = started_at.elapsed().as_millis(), + error = ?error, + "Transaction confirmation ended with retryable send error" + ); return Err(error.into()); } @@ -506,16 +566,70 @@ async fn send_prepared_transaction( return Err(SmartTransactionError::ConfirmationDeadlineExceeded { signature }); } + warn!( + signature = %signature, + label = %transaction.label(), + send_attempts, + elapsed_ms = started_at.elapsed().as_millis(), + max_attempts = confirmation.max_attempts, + last_send_error = ?last_send_error.as_ref(), + "Transaction confirmation timed out" + ); Err(SmartTransactionError::ConfirmationTimedOut { signature, max_attempts: confirmation.max_attempts, }) } +fn is_permanent_send_error(error: &RpcError) -> bool { + let message = error.to_string().to_ascii_lowercase(); + message.contains("base64 encoded too large") + || message.contains("encoded too large") + || message.contains("transaction too large") + || message.contains("invalidparams") + || message.contains("invalid params") +} + +fn debug_retryable_send_error( + transaction: &PreparedTransaction, + signature: Signature, + send_attempts: u32, + error: &RpcError, +) { + debug!( + signature = %signature, + label = %transaction.label(), + send_attempts, + error = ?error, + "Retryable transaction send error" + ); +} + +fn log_blockhash_expired( + transaction: &PreparedTransaction, + signature: Signature, + last_valid_block_height: u64, + current_block_height: u64, + send_attempts: u32, + elapsed: Duration, + last_send_error: Option<&RpcError>, +) { + warn!( + signature = %signature, + label = %transaction.label(), + last_valid_block_height, + current_block_height, + send_attempts, + elapsed_ms = elapsed.as_millis(), + last_send_error = ?last_send_error, + "Transaction blockhash expired before confirmation" + ); +} + fn confirmation_send_transaction_config() -> RpcSendTransactionConfig { RpcSendTransactionConfig { skip_preflight: true, - max_retries: Some(0), + max_retries: Some(RPC_SEND_TRANSACTION_MAX_RETRIES), ..Default::default() } } diff --git a/forester/tests/e2e_test.rs b/forester/tests/e2e_test.rs index c11500f29a..76680db3cb 100644 --- a/forester/tests/e2e_test.rs +++ b/forester/tests/e2e_test.rs @@ -249,6 +249,7 @@ async fn e2e_test() { prover_api_key: get_prover_api_key(), prover_polling_interval: None, prover_max_wait_time: None, + prover_max_concurrent_jobs: 4, photon_grpc_url: get_photon_grpc_url(), pushgateway_url: None, pagerduty_routing_key: None, @@ -303,6 +304,7 @@ async fn e2e_test() { }), min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, work_item_batch_size: 50, }; let test_mode = TestMode::from_env(); diff --git a/forester/tests/legacy/priority_fee_test.rs b/forester/tests/legacy/priority_fee_test.rs index 37f2ed010a..96e22e4134 100644 --- a/forester/tests/legacy/priority_fee_test.rs +++ b/forester/tests/legacy/priority_fee_test.rs @@ -89,6 +89,7 @@ async fn test_priority_fee_request() { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, api_server_port: 8080, api_server_public_bind: false, group_authority: None, @@ -101,6 +102,7 @@ async fn test_priority_fee_request() { prover_api_key: None, prover_polling_interval_ms: None, prover_max_wait_time_secs: None, + prover_max_concurrent_jobs: 4, photon_grpc_url: None, max_concurrent_sends: 50, max_batches_per_tree: 4, diff --git a/forester/tests/legacy/test_utils.rs b/forester/tests/legacy/test_utils.rs index fdac97ba34..faafb5ee53 100644 --- a/forester/tests/legacy/test_utils.rs +++ b/forester/tests/legacy/test_utils.rs @@ -81,6 +81,7 @@ pub fn forester_config() -> ForesterConfig { prover_api_key: None, prover_polling_interval: None, prover_max_wait_time: None, + prover_max_concurrent_jobs: 4, photon_grpc_url: None, pushgateway_url: None, pagerduty_routing_key: None, @@ -125,6 +126,7 @@ pub fn forester_config() -> ForesterConfig { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, work_item_batch_size: 50, } } diff --git a/forester/tests/priority_fee_test.rs b/forester/tests/priority_fee_test.rs index 5067342d2f..1c7b695606 100644 --- a/forester/tests/priority_fee_test.rs +++ b/forester/tests/priority_fee_test.rs @@ -49,6 +49,7 @@ async fn test_priority_fee_request() { prover_api_key: None, prover_polling_interval_ms: None, prover_max_wait_time_secs: None, + prover_max_concurrent_jobs: 4, payer: Some(std::env::var("PAYER").expect("PAYER must be set in environment")), derivation: Some( std::env::var("DERIVATION_PUBKEY") @@ -94,6 +95,7 @@ async fn test_priority_fee_request() { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, api_server_port: 8080, group_authority: None, light_pda_programs: vec![], diff --git a/forester/tests/test_utils.rs b/forester/tests/test_utils.rs index 4ae9352482..b87c05b23f 100644 --- a/forester/tests/test_utils.rs +++ b/forester/tests/test_utils.rs @@ -91,6 +91,7 @@ pub fn forester_config() -> ForesterConfig { prover_api_key: None, prover_polling_interval: None, prover_max_wait_time: None, + prover_max_concurrent_jobs: 4, photon_grpc_url: None, pushgateway_url: None, pagerduty_routing_key: None, @@ -138,6 +139,7 @@ pub fn forester_config() -> ForesterConfig { lookup_table_address: None, min_queue_items: None, enable_v1_multi_nullify: false, + enable_v1_presort: false, work_item_batch_size: 50, } } diff --git a/program-tests/account-compression-test/tests/batched_merkle_tree_test.rs b/program-tests/account-compression-test/tests/batched_merkle_tree_test.rs index ba9446c594..b8a949b928 100644 --- a/program-tests/account-compression-test/tests/batched_merkle_tree_test.rs +++ b/program-tests/account-compression-test/tests/batched_merkle_tree_test.rs @@ -1017,8 +1017,8 @@ async fn test_init_batch_state_merkle_trees() { let params = InitStateTreeAccountsInstructionData::test_default(); let e2e_test_params = InitStateTreeAccountsInstructionData::e2e_test_default(); let default_params = InitStateTreeAccountsInstructionData::default(); - let param_vec = vec![params, e2e_test_params, default_params]; - for params in param_vec.iter() { + let param_vec = [params, e2e_test_params, default_params]; + for params in ¶m_vec { println!("Init new mt with params {:?}", params); let merkle_tree_keypair = Keypair::new(); let nullifier_queue_keypair = Keypair::new(); @@ -1595,8 +1595,8 @@ async fn test_init_batch_address_merkle_trees() { let params = InitAddressTreeAccountsInstructionData::test_default(); let e2e_test_params = InitAddressTreeAccountsInstructionData::e2e_test_default(); let default_params = InitAddressTreeAccountsInstructionData::default(); - let param_vec = vec![params, e2e_test_params, default_params]; - for params in param_vec.iter() { + let param_vec = [params, e2e_test_params, default_params]; + for params in ¶m_vec { println!("Init new mt with params {:?}", params); let merkle_tree_keypair = Keypair::new(); let merkle_tree_pubkey = merkle_tree_keypair.pubkey(); diff --git a/program-tests/compressed-token-test/tests/v1.rs b/program-tests/compressed-token-test/tests/v1.rs index 81c01c82fd..5ecf57e960 100644 --- a/program-tests/compressed-token-test/tests/v1.rs +++ b/program-tests/compressed-token-test/tests/v1.rs @@ -4816,7 +4816,7 @@ async fn test_transfer_with_transaction_hash() { let payer = recipient_keypair.insecure_clone(); let input_compressed_account_token_data = test_indexer.token_compressed_accounts[0].token_data.clone(); - let input_compressed_accounts = vec![test_indexer.token_compressed_accounts[0].clone()]; + let input_compressed_accounts = [test_indexer.token_compressed_accounts[0].clone()]; let change_out_compressed_account_0 = TokenTransferOutputData { amount: input_compressed_account_token_data.amount, diff --git a/program-tests/system-test/tests/test.rs b/program-tests/system-test/tests/test.rs index 4262a24232..e577f0389b 100644 --- a/program-tests/system-test/tests/test.rs +++ b/program-tests/system-test/tests/test.rs @@ -1636,7 +1636,7 @@ async fn test_with_compression() { &mut rpc, &mut test_indexer, &payer, - &vec![compressed_account_with_context], + &[compressed_account_with_context], &recipient_pubkey, compress_amount, &env.v1_state_trees[0].merkle_tree, diff --git a/programs/registry/src/account_compression_cpi/sdk.rs b/programs/registry/src/account_compression_cpi/sdk.rs index bc29789c0b..6aee5f904d 100644 --- a/programs/registry/src/account_compression_cpi/sdk.rs +++ b/programs/registry/src/account_compression_cpi/sdk.rs @@ -176,8 +176,8 @@ pub fn compress_proofs(proofs: &[&[[u8; 32]; 16]]) -> Option { let mut proof_bitvecs = [0u32; 4]; for (proof_idx, _) in proofs.iter().enumerate() { let bv = proof_bitvecs[proof_idx].view_bits_mut::(); - for level in 0..16 { - bv.set(pool_indices[proof_idx][level], true); + for &pool_index in &pool_indices[proof_idx] { + bv.set(pool_index, true); } } diff --git a/prover/client/src/proof_client.rs b/prover/client/src/proof_client.rs index 859ea4917f..da90946c48 100644 --- a/prover/client/src/proof_client.rs +++ b/prover/client/src/proof_client.rs @@ -63,6 +63,7 @@ pub struct ProofClient { polling_interval: Duration, max_wait_time: Duration, api_key: Option, + network: Option, initial_poll_delay: Duration, } @@ -74,6 +75,7 @@ impl ProofClient { polling_interval: Duration::from_millis(DEFAULT_POLLING_INTERVAL_MS), max_wait_time: Duration::from_secs(DEFAULT_MAX_WAIT_TIME_SECS), api_key: None, + network: None, initial_poll_delay: Duration::from_millis(INITIAL_POLL_DELAY_SMALL_CIRCUIT_MS), } } @@ -97,6 +99,7 @@ impl ProofClient { polling_interval, max_wait_time, api_key, + network: None, initial_poll_delay, } } @@ -115,10 +118,16 @@ impl ProofClient { polling_interval, max_wait_time, api_key, + network: None, initial_poll_delay, } } + pub fn with_network(mut self, network: String) -> Self { + self.network = Some(network); + self + } + pub async fn submit_proof_async( &self, inputs_json: String, @@ -242,6 +251,9 @@ impl ProofClient { if let Some(api_key) = &self.api_key { request = request.header("X-API-Key", api_key); } + if let Some(network) = &self.network { + request = request.header("X-Light-Network", network); + } request .body(inputs_json.to_string()) diff --git a/prover/server/prover/common/key_downloader.go b/prover/server/prover/common/key_downloader.go index d3e93e29ae..6a03cbe0f1 100644 --- a/prover/server/prover/common/key_downloader.go +++ b/prover/server/prover/common/key_downloader.go @@ -15,7 +15,7 @@ import ( ) const ( - DefaultBaseURL = "https://storage.googleapis.com/light-protocol-proving-keys/light-protocol-keys" + DefaultBaseURL = "https://d1wbn9ra8wjh7t.cloudfront.net" DefaultMaxRetries = 10 DefaultRetryDelay = 5 * time.Second DefaultMaxRetryDelay = 5 * time.Minute @@ -68,7 +68,7 @@ func downloadChecksum(config *DownloadConfig) error { return nil } - checksumURL := config.BaseURL + "/CHECKSUM" + checksumURL := strings.TrimRight(config.BaseURL, "/") + "/CHECKSUM" logging.Logger().Info(). Str("url", checksumURL). Msg("Downloading CHECKSUM file") @@ -284,6 +284,23 @@ func downloadFileWithResume(url, outputPath string, config *DownloadConfig) erro func DownloadKey(keyPath string, config *DownloadConfig) error { filename := filepath.Base(keyPath) + if !config.AutoDownload { + fileInfo, err := os.Stat(keyPath) + if err != nil { + if os.IsNotExist(err) { + return fmt.Errorf("required key file not found: %s (auto-download disabled)", filename) + } + return fmt.Errorf("failed to check key file %s: %w", filename, err) + } + if !fileInfo.Mode().IsRegular() || fileInfo.Size() == 0 { + return fmt.Errorf("required key file is not a non-empty regular file: %s", filename) + } + logging.Logger().Info(). + Str("file", filename). + Int64("size", fileInfo.Size()). + Msg("Using existing key file with auto-download disabled") + return nil + } if err := downloadChecksum(config); err != nil { return fmt.Errorf("failed to load checksums: %w", err) @@ -303,6 +320,10 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { } if fileInfo, err := os.Stat(keyPath); err == nil { + if !fileInfo.Mode().IsRegular() || fileInfo.Size() == 0 { + return fmt.Errorf("existing key file is not a non-empty regular file: %s", filename) + } + logging.Logger().Info(). Str("file", filename). Int64("size", fileInfo.Size()). @@ -343,7 +364,7 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { return fmt.Errorf("failed to create directory: %w", err) } - url := fmt.Sprintf("%s/%s", config.BaseURL, filename) + url := fmt.Sprintf("%s/%s", strings.TrimRight(config.BaseURL, "/"), filename) logging.Logger().Info(). Str("file", filename). Str("url", url). @@ -364,7 +385,7 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { logging.Logger().Info(). Str("file", filename). - Msg("Key file downloaded and verified successfully") + Msg("Key file download completed successfully") return nil } diff --git a/prover/server/prover/common/key_downloader_offline_test.go b/prover/server/prover/common/key_downloader_offline_test.go new file mode 100644 index 0000000000..afe19b61bb --- /dev/null +++ b/prover/server/prover/common/key_downloader_offline_test.go @@ -0,0 +1,106 @@ +package common + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" +) + +func TestDownloadKeyOfflineUsesExistingNonEmptyFile(t *testing.T) { + keyPath := filepath.Join(t.TempDir(), "batch_update_32_500.key") + if err := os.WriteFile(keyPath, []byte("existing-key"), 0o600); err != nil { + t.Fatal(err) + } + + config := DefaultDownloadConfig() + config.AutoDownload = false + config.BaseURL = "http://127.0.0.1:1" + if err := DownloadKey(keyPath, config); err != nil { + t.Fatalf("existing offline key rejected: %v", err) + } +} + +func TestDownloadKeyOfflineRejectsMissingOrEmptyFile(t *testing.T) { + config := DefaultDownloadConfig() + config.AutoDownload = false + config.BaseURL = "http://127.0.0.1:1" + + missing := filepath.Join(t.TempDir(), "missing.key") + if err := DownloadKey(missing, config); err == nil { + t.Fatal("missing offline key unexpectedly accepted") + } + + empty := filepath.Join(t.TempDir(), "empty.key") + if err := os.WriteFile(empty, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := DownloadKey(empty, config); err == nil { + t.Fatal("empty offline key unexpectedly accepted") + } +} + +func TestDownloadKeyUsesBaseURLRootWithTrailingSlash(t *testing.T) { + const filename = "test.key" + const contents = "downloaded-key" + const checksum = "1195e8d870f621c94ac378c38846612f075d0bd8da7fb727a873220ba6434a63" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/CHECKSUM": + _, _ = w.Write([]byte(checksum + " " + filename + "\n")) + case "/" + filename: + _, _ = w.Write([]byte(contents)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := DefaultDownloadConfig() + config.BaseURL = server.URL + "/" + config.MaxRetries = 1 + config.RetryDelay = time.Millisecond + config.MaxRetryDelay = time.Millisecond + + keyPath := filepath.Join(t.TempDir(), filename) + if err := DownloadKey(keyPath, config); err != nil { + t.Fatalf("download failed: %v", err) + } + + got, err := os.ReadFile(keyPath) + if err != nil { + t.Fatal(err) + } + if string(got) != contents { + t.Fatalf("downloaded contents = %q, want %q", got, contents) + } +} + +func TestDownloadKeyRejectsMissingChecksumManifest(t *testing.T) { + const filename = "test.key" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/CHECKSUM" { + w.WriteHeader(http.StatusForbidden) + return + } + t.Fatal("key object requested without required checksum manifest") + })) + defer server.Close() + + config := DefaultDownloadConfig() + config.BaseURL = server.URL + config.MaxRetries = 1 + + keyPath := filepath.Join(t.TempDir(), filename) + err := DownloadKey(keyPath, config) + if err == nil { + t.Fatal("download without checksum manifest unexpectedly succeeded") + } + if got, want := err.Error(), "failed to load checksums: failed to download CHECKSUM file: HTTP 403"; got != want { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/prover/server/server/queue.go b/prover/server/server/queue.go index 314849b2a8..c94ec98318 100644 --- a/prover/server/server/queue.go +++ b/prover/server/server/queue.go @@ -5,15 +5,71 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "light/light-prover/logging" "light/light-prover/prover/common" + "os" + "strconv" + "strings" "time" "github.com/google/uuid" "github.com/redis/go-redis/v9" ) +var ErrQueueFull = errors.New("network proof queue is full") + +var enqueueWithLimitScript = redis.NewScript(` +local total = redis.call("LLEN", KEYS[1]) +local trees = redis.call("SMEMBERS", KEYS[3]) +for _, tree in ipairs(trees) do + total = total + redis.call("LLEN", KEYS[1] .. ":" .. tree) +end +if total >= tonumber(ARGV[3]) then + return 0 +end +if ARGV[1] ~= "" then + redis.call("SADD", KEYS[3], ARGV[1]) +end +redis.call("RPUSH", KEYS[2], ARGV[2]) +return 1 +`) + +const ( + MainnetNetwork = "mainnet" + DevnetNetwork = "devnet" +) + +func NetworkQueueName(baseQueue, network string) string { + return strings.TrimSuffix(baseQueue, "_queue") + "_" + network + "_queue" +} + +func ProcessingQueueName(queueName string) string { + return strings.TrimSuffix(queueName, "_queue") + "_processing_queue" +} + +func normalizeNetwork(network string) (string, error) { + network = strings.ToLower(strings.TrimSpace(network)) + if network != MainnetNetwork && network != DevnetNetwork { + return "", fmt.Errorf("unsupported prover network %q", network) + } + return network, nil +} + +func queueLimit(queueName string) int64 { + envName, fallback := "PROVER_MAX_PENDING_DEFAULT", int64(16) + if strings.Contains(queueName, "_mainnet_queue") { + envName, fallback = "PROVER_MAX_PENDING_MAINNET", 64 + } else if strings.Contains(queueName, "_devnet_queue") { + envName, fallback = "PROVER_MAX_PENDING_DEVNET", 8 + } + if value, err := strconv.ParseInt(os.Getenv(envName), 10, 64); err == nil && value > 0 { + return value + } + return fallback +} + const ( // ResultsIndexKey is the Redis hash that maps inputHash → jobID ResultsIndexKey = "zk_results_index" @@ -70,13 +126,35 @@ func (rq *RedisQueue) EnqueueProof(queueName string, job *ProofJob) error { return fmt.Errorf("failed to marshal job: %w", err) } - // Use tree-specific sub-queue for fair queuing if TreeID is set + // Use tree-specific sub-queue for fair queuing if TreeID is set. actualQueueName := queueName if job.TreeID != "" && isFairQueueEnabled(queueName) { actualQueueName = fmt.Sprintf("%s:%s", queueName, job.TreeID) - // Track this tree in the trees set for round-robin - treesSetKey := fmt.Sprintf("%s:trees", queueName) - rq.Client.SAdd(rq.Ctx, treesSetKey, job.TreeID) + } + + if isFairQueueEnabled(queueName) && + (strings.Contains(queueName, "_mainnet_queue") || strings.Contains(queueName, "_devnet_queue")) { + limit := queueLimit(queueName) + accepted, err := enqueueWithLimitScript.Run( + rq.Ctx, + rq.Client, + []string{queueName, actualQueueName, queueName + ":trees"}, + job.TreeID, + data, + limit, + ).Int() + if err != nil { + return fmt.Errorf("failed to atomically enqueue job: %w", err) + } + if accepted == 0 { + pending, _ := rq.pendingQueueLen(queueName) + return fmt.Errorf("%w: queue=%s pending=%d limit=%d", ErrQueueFull, queueName, pending, limit) + } + return nil + } + + if job.TreeID != "" && isFairQueueEnabled(queueName) { + rq.Client.SAdd(rq.Ctx, queueName+":trees", job.TreeID) } err = rq.Client.RPush(rq.Ctx, actualQueueName, data).Err() @@ -95,9 +173,85 @@ func (rq *RedisQueue) EnqueueProof(queueName string, job *ProofJob) error { // isFairQueueEnabled returns true for queues that support fair queuing per tree func isFairQueueEnabled(queueName string) bool { - return queueName == "zk_update_queue" || - queueName == "zk_append_queue" || - queueName == "zk_address_append_queue" + return !strings.Contains(queueName, "_processing_") && + strings.HasSuffix(queueName, "_queue") && + (strings.HasPrefix(queueName, "zk_update") || + strings.HasPrefix(queueName, "zk_append") || + strings.HasPrefix(queueName, "zk_address_append")) +} + +func (rq *RedisQueue) pendingQueueLen(queueName string) (int64, error) { + total, err := rq.Client.LLen(rq.Ctx, queueName).Result() + if err != nil { + return 0, err + } + trees, err := rq.Client.SMembers(rq.Ctx, queueName+":trees").Result() + if err != nil { + return 0, err + } + for _, tree := range trees { + length, err := rq.Client.LLen(rq.Ctx, queueName+":"+tree).Result() + if err != nil { + return 0, err + } + total += length + } + return total, nil +} + +func (rq *RedisQueue) DequeuePrioritizedProof(baseQueue string) (*ProofJob, error) { + cycle, err := rq.Client.Incr(rq.Ctx, baseQueue+":network_cycle").Result() + if err != nil { + return nil, err + } + networks := []string{MainnetNetwork, DevnetNetwork} + if cycle%5 == 0 { + networks = []string{DevnetNetwork, MainnetNetwork} + } + for _, network := range networks { + queueName := NetworkQueueName(baseQueue, network) + pending, err := rq.pendingQueueLen(queueName) + if err != nil { + return nil, err + } + if pending == 0 { + continue + } + job, err := rq.DequeueProof(queueName, time.Second) + if err != nil { + return nil, err + } + if job != nil { + return job, nil + } + } + // Drain pre-upgrade jobs after checking both network queues so a rolling + // deployment does not strand work in the legacy queue. + pending, err := rq.pendingQueueLen(baseQueue) + if err != nil { + return nil, err + } + if pending > 0 { + return rq.DequeueProof(baseQueue, time.Second) + } + return nil, nil +} + +func requestQueueNames() []string { + bases := []string{"zk_update_queue", "zk_append_queue", "zk_address_append_queue"} + queues := append([]string{}, bases...) + for _, base := range bases { + queues = append(queues, NetworkQueueName(base, MainnetNetwork), NetworkQueueName(base, DevnetNetwork)) + } + return queues +} + +func processingQueueNames() []string { + queues := []string{"zk_update_processing_queue", "zk_append_processing_queue", "zk_address_append_processing_queue"} + for _, requestQueue := range requestQueueNames()[3:] { + queues = append(queues, ProcessingQueueName(requestQueue)) + } + return queues } // StoreJobMeta stores job metadata when a job is submitted to enable reliable status lookups. @@ -389,7 +543,8 @@ func (rq *RedisQueue) dequeueLowestBatchIndex(queueName string) (*ProofJob, erro func (rq *RedisQueue) GetQueueStats() (map[string]int64, error) { stats := make(map[string]int64) - queues := []string{"zk_update_queue", "zk_append_queue", "zk_address_append_queue", "zk_update_processing_queue", "zk_append_processing_queue", "zk_address_append_processing_queue", "zk_failed_queue", "zk_results_queue"} + queues := append(requestQueueNames(), processingQueueNames()...) + queues = append(queues, "zk_failed_queue", "zk_results_queue") for _, queue := range queues { length, err := rq.Client.LLen(rq.Ctx, queue).Result() @@ -429,8 +584,16 @@ func (rq *RedisQueue) GetQueueHealth() (map[string]interface{}, error) { health["queue_lengths"] = stats health["timestamp"] = time.Now().Unix() - health["total_pending"] = stats["zk_update_queue"] + stats["zk_append_queue"] + stats["zk_address_append_queue"] - health["total_processing"] = stats["zk_update_processing_queue"] + stats["zk_append_processing_queue"] + stats["zk_address_append_processing_queue"] + var totalPending int64 + for _, queue := range requestQueueNames() { + totalPending += stats[queue] + stats[queue+"_tree_subqueues"] + } + var totalProcessing int64 + for _, queue := range processingQueueNames() { + totalProcessing += stats[queue] + } + health["total_pending"] = totalPending + health["total_processing"] = totalProcessing health["total_failed"] = stats["zk_failed_queue"] health["total_results"] = stats["zk_results_queue"] @@ -451,11 +614,7 @@ func (rq *RedisQueue) GetQueueHealth() (map[string]interface{}, error) { func (rq *RedisQueue) countStuckJobs() int64 { stuckTimeout := time.Now().Add(-2 * time.Minute) - processingQueues := []string{ - "zk_update_processing_queue", - "zk_append_processing_queue", - "zk_address_append_processing_queue", - } + processingQueues := processingQueueNames() var totalStuck int64 @@ -616,11 +775,7 @@ func (rq *RedisQueue) CleanupOldRequests() error { cutoffTime := time.Now().Add(-30 * time.Minute) // Queues to clean up old requests from - queuesToClean := []string{ - "zk_update_queue", - "zk_append_queue", - "zk_address_append_queue", - } + queuesToClean := requestQueueNames() totalRemoved := int64(0) @@ -720,11 +875,7 @@ func (rq *RedisQueue) CleanupStuckProcessingJobs() error { // (proof generation can take 3-4 minutes under load) processingTimeout := time.Now().Add(-10 * time.Minute) - processingQueues := []string{ - "zk_update_processing_queue", - "zk_append_processing_queue", - "zk_address_append_processing_queue", - } + processingQueues := processingQueueNames() totalRecovered := int64(0) totalFailed := int64(0) @@ -854,10 +1005,13 @@ func (rq *RedisQueue) recoverStuckJobsFromQueue(queueName string, timeoutCutoff originalQueue := getOriginalQueueFromProcessing(queueName) if originalQueue != "" { originalJob := &ProofJob{ - ID: originalJobID, - Type: "zk_proof", - Payload: job.Payload, - CreatedAt: job.CreatedAt, + ID: originalJobID, + Type: "zk_proof", + Payload: job.Payload, + CreatedAt: job.CreatedAt, + TreeID: job.TreeID, + BatchIndex: job.BatchIndex, + Network: job.Network, } err = rq.EnqueueProof(originalQueue, originalJob) @@ -887,16 +1041,10 @@ func (rq *RedisQueue) recoverStuckJobsFromQueue(queueName string, timeoutCutoff } func getOriginalQueueFromProcessing(processingQueueName string) string { - switch processingQueueName { - case "zk_update_processing_queue": - return "zk_update_queue" - case "zk_append_processing_queue": - return "zk_append_queue" - case "zk_address_append_processing_queue": - return "zk_address_append_queue" - default: - return "" + if strings.HasSuffix(processingQueueName, "_processing_queue") { + return strings.TrimSuffix(processingQueueName, "_processing_queue") + "_queue" } + return "" } func (rq *RedisQueue) cleanupOldRequestsFromQueue(queueName string, cutoffTime time.Time) (int64, error) { @@ -971,6 +1119,21 @@ func ComputeInputHash(payload json.RawMessage) string { return hex.EncodeToString(hash[:]) } +func ComputeNetworkInputHash(network string, payload json.RawMessage) string { + if network == "" { + return ComputeInputHash(payload) + } + prefixed := make([]byte, 0, len(network)+1+len(payload)) + prefixed = append(prefixed, network...) + prefixed = append(prefixed, 0) + prefixed = append(prefixed, payload...) + return ComputeInputHash(prefixed) +} + +func ComputeJobInputHash(job *ProofJob) string { + return ComputeNetworkInputHash(job.Network, job.Payload) +} + // FindCachedResult searches for a cached result by input hash. // Returns the proof result (as ProofWithTiming) and job ID if found, otherwise returns nil. func (rq *RedisQueue) FindCachedResult(inputHash string) (*common.ProofWithTiming, string, error) { diff --git a/prover/server/server/queue_job.go b/prover/server/server/queue_job.go index 9a225290fe..ee8e2369f4 100644 --- a/prover/server/server/queue_job.go +++ b/prover/server/server/queue_job.go @@ -107,7 +107,8 @@ type ProofJob struct { // BatchIndex is the batch sequence number within a tree - used to process batches in order // Lower batch indices should be processed first to enable sequential transaction submission // -1 means no batch index (legacy requests, FIFO) - BatchIndex int64 `json:"batch_index"` + BatchIndex int64 `json:"batch_index"` + Network string `json:"network,omitempty"` } type QueueWorker interface { @@ -204,7 +205,7 @@ func (w *BaseQueueWorker) Stop() { } func (w *BaseQueueWorker) processJobs() { - job, err := w.queue.DequeueProof(w.queueName, 5*time.Second) + job, err := w.queue.DequeuePrioritizedProof(w.queueName) if err != nil { logging.Logger().Error().Err(err).Str("queue", w.queueName).Msg("Error dequeuing from queue") time.Sleep(2 * time.Second) @@ -234,7 +235,7 @@ func (w *BaseQueueWorker) processJobs() { // Add to failed queue with expiration reason expirationErr := fmt.Errorf("job expired after %v (max: %v)", jobAge, JobExpirationTimeout) - expiredInputHash := ComputeInputHash(job.Payload) + expiredInputHash := ComputeJobInputHash(job) w.addToFailedQueue(job, expiredInputHash, expirationErr) return } @@ -259,7 +260,7 @@ func (w *BaseQueueWorker) processJobs() { Msg("Dequeued proof job") // Check for duplicate inputs before processing - inputHash := ComputeInputHash(job.Payload) + inputHash := ComputeJobInputHash(job) // Check if we already have a successful result for this input cachedProof, cachedJobID, err := w.queue.FindCachedResult(inputHash) @@ -366,24 +367,31 @@ func (w *BaseQueueWorker) processJobs() { Str("queue", w.queueName). Msg("Starting proof generation") + processingQueueName := w.processingQueueName + if job.Network != "" { + processingQueueName = ProcessingQueueName(NetworkQueueName(w.queueName, job.Network)) + } processingJob := &ProofJob{ - ID: job.ID + "_processing", - Type: "processing", - Payload: job.Payload, - CreatedAt: time.Now(), + ID: job.ID + "_processing", + Type: "processing", + Payload: job.Payload, + CreatedAt: time.Now(), + TreeID: job.TreeID, + BatchIndex: job.BatchIndex, + Network: job.Network, } - err := w.queue.EnqueueProof(w.processingQueueName, processingJob) + err := w.queue.EnqueueProof(processingQueueName, processingJob) if err != nil { logging.Logger().Error(). Err(err). Str("job_id", job.ID). - Str("processing_queue", w.processingQueueName). + Str("processing_queue", processingQueueName). Msg("Failed to add job to processing queue") return } proof, err := w.generateProof(job) - w.removeFromProcessingQueue(job.ID) + w.removeFromProcessingQueue(job.ID, processingQueueName) proofDuration := time.Since(proofStartTime) @@ -685,18 +693,18 @@ func (w *BaseQueueWorker) processBatchAddressAppendProof(payload json.RawMessage return v2.ProveBatchAddressAppend(ps, ¶ms) } -func (w *BaseQueueWorker) removeFromProcessingQueue(jobID string) { - processingQueueLength, _ := w.queue.Client.LLen(w.queue.Ctx, w.processingQueueName).Result() +func (w *BaseQueueWorker) removeFromProcessingQueue(jobID, processingQueueName string) { + processingQueueLength, _ := w.queue.Client.LLen(w.queue.Ctx, processingQueueName).Result() for i := range processingQueueLength { - item, err := w.queue.Client.LIndex(w.queue.Ctx, w.processingQueueName, i).Result() + item, err := w.queue.Client.LIndex(w.queue.Ctx, processingQueueName, i).Result() if err != nil { continue } var job ProofJob if json.Unmarshal([]byte(item), &job) == nil && job.ID == jobID+"_processing" { - w.queue.Client.LRem(w.queue.Ctx, w.processingQueueName, 1, item) + w.queue.Client.LRem(w.queue.Ctx, processingQueueName, 1, item) break } } diff --git a/prover/server/server/queue_network_test.go b/prover/server/server/queue_network_test.go new file mode 100644 index 0000000000..75d6b7a7f4 --- /dev/null +++ b/prover/server/server/queue_network_test.go @@ -0,0 +1,68 @@ +package server + +import ( + "encoding/json" + "testing" +) + +func TestNormalizeNetwork(t *testing.T) { + for input, expected := range map[string]string{ + "mainnet": MainnetNetwork, + " MAINNET ": MainnetNetwork, + "devnet": DevnetNetwork, + } { + actual, err := normalizeNetwork(input) + if err != nil { + t.Fatalf("normalizeNetwork(%q): %v", input, err) + } + if actual != expected { + t.Fatalf("normalizeNetwork(%q) = %q, want %q", input, actual, expected) + } + } + + for _, input := range []string{"", "default", "testnet", "mainnet-beta"} { + if _, err := normalizeNetwork(input); err == nil { + t.Fatalf("normalizeNetwork(%q) unexpectedly succeeded", input) + } + } +} + +func TestNetworkQueueNames(t *testing.T) { + request := NetworkQueueName("zk_address_append_queue", MainnetNetwork) + if request != "zk_address_append_mainnet_queue" { + t.Fatalf("unexpected request queue: %s", request) + } + processing := ProcessingQueueName(request) + if processing != "zk_address_append_mainnet_processing_queue" { + t.Fatalf("unexpected processing queue: %s", processing) + } + if original := getOriginalQueueFromProcessing(processing); original != request { + t.Fatalf("processing queue maps to %s, want %s", original, request) + } +} + +func TestNetworkInputHashIsolation(t *testing.T) { + payload := json.RawMessage(`{"circuitType":"batchAppend"}`) + mainnet := ComputeNetworkInputHash(MainnetNetwork, payload) + devnet := ComputeNetworkInputHash(DevnetNetwork, payload) + legacy := ComputeNetworkInputHash("", payload) + + if mainnet == devnet || mainnet == legacy || devnet == legacy { + t.Fatal("identical payloads must have distinct cache keys across networks") + } + if legacy != ComputeInputHash(payload) { + t.Fatal("legacy input hash changed") + } +} + +func TestQueueLimits(t *testing.T) { + t.Setenv("PROVER_MAX_PENDING_MAINNET", "21") + t.Setenv("PROVER_MAX_PENDING_DEVNET", "3") + + if got := queueLimit("zk_append_mainnet_queue"); got != 21 { + t.Fatalf("mainnet queue limit = %d, want 21", got) + } + if got := queueLimit("zk_append_devnet_queue"); got != 3 { + t.Fatalf("devnet queue limit = %d, want 3", got) + } +} diff --git a/prover/server/server/server.go b/prover/server/server/server.go index a332e30c13..a14f0d241e 100644 --- a/prover/server/server/server.go +++ b/prover/server/server/server.go @@ -443,6 +443,14 @@ func (handler proveHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { forceSync := r.Header.Get("X-Sync") == "true" || r.URL.Query().Get("sync") == "true" shouldUseQueue := handler.shouldUseQueueForCircuit(proofRequestMeta.CircuitType, forceAsync, forceSync) + network := "" + if shouldUseQueue { + network, err = normalizeNetwork(r.Header.Get("X-Light-Network")) + if err != nil { + malformedBodyError(err).send(w) + return + } + } logging.Logger().Info(). Str("circuit_type", string(proofRequestMeta.CircuitType)). @@ -453,7 +461,7 @@ func (handler proveHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { Msg("Processing prove request") if shouldUseQueue && handler.enableQueue && handler.redisQueue != nil { - handler.handleAsyncProof(w, r, buf, proofRequestMeta) + handler.handleAsyncProof(w, r, buf, proofRequestMeta, network) } else { handler.handleSyncProof(w, r, buf, proofRequestMeta) } @@ -659,11 +667,16 @@ func RunEnhanced(config *EnhancedConfig, redisQueue *RedisQueue, keyManager *com malformedBodyError(err).send(w) return } + network, err := normalizeNetwork(r.Header.Get("X-Light-Network")) + if err != nil { + malformedBodyError(err).send(w) + return + } - queueName := GetQueueNameForCircuit(proofRequestMeta.CircuitType) + queueName := NetworkQueueName(GetQueueNameForCircuit(proofRequestMeta.CircuitType), network) // Compute input hash for deduplication - inputHash := ComputeInputHash(json.RawMessage(buf)) + inputHash := ComputeNetworkInputHash(network, json.RawMessage(buf)) // Check for existing in-flight job with same input dedupResult, err := redisQueue.DeduplicateJob(inputHash) @@ -715,6 +728,7 @@ func RunEnhanced(config *EnhancedConfig, redisQueue *RedisQueue, keyManager *com CreatedAt: time.Now(), TreeID: proofRequestMeta.TreeID, BatchIndex: proofRequestMeta.BatchIndex, + Network: network, } // Store job metadata BEFORE enqueueing to prevent race condition where worker @@ -739,6 +753,11 @@ func RunEnhanced(config *EnhancedConfig, redisQueue *RedisQueue, keyManager *com if delErr := redisQueue.DeleteJobMeta(jobID); delErr != nil { logging.Logger().Error().Err(delErr).Str("job_id", jobID).Msg("Failed to cleanup job metadata after enqueue failure") } + if errors.Is(err, ErrQueueFull) { + w.Header().Set("Retry-After", "5") + (&Error{StatusCode: http.StatusTooManyRequests, Code: "queue_full", Message: err.Error()}).send(w) + return + } unexpectedError(err).send(w) return } @@ -865,14 +884,14 @@ func spawnServerJob(server *http.Server, label string) RunningJob { type healthHandler struct { } -func (handler proveHandler) handleAsyncProof(w http.ResponseWriter, r *http.Request, buf []byte, meta common.ProofRequestMeta) { +func (handler proveHandler) handleAsyncProof(w http.ResponseWriter, r *http.Request, buf []byte, meta common.ProofRequestMeta, network string) { ProofRequestsTotal.WithLabelValues(string(meta.CircuitType)).Inc() RecordCircuitInputSize(string(meta.CircuitType), len(buf)) - queueName := GetQueueNameForCircuit(meta.CircuitType) + queueName := NetworkQueueName(GetQueueNameForCircuit(meta.CircuitType), network) // Compute input hash for deduplication - inputHash := ComputeInputHash(json.RawMessage(buf)) + inputHash := ComputeNetworkInputHash(network, json.RawMessage(buf)) // Check for existing in-flight job with same input dedupResult, err := handler.redisQueue.DeduplicateJob(inputHash) @@ -924,6 +943,7 @@ func (handler proveHandler) handleAsyncProof(w http.ResponseWriter, r *http.Requ CreatedAt: time.Now(), TreeID: meta.TreeID, BatchIndex: meta.BatchIndex, + Network: network, } // Store job metadata BEFORE enqueueing to prevent race condition where worker @@ -951,6 +971,15 @@ func (handler proveHandler) handleAsyncProof(w http.ResponseWriter, r *http.Requ logging.Logger().Error().Err(delErr).Str("job_id", jobID).Msg("Failed to cleanup job metadata after enqueue failure") } + if errors.Is(err, ErrQueueFull) { + w.Header().Set("Retry-After", "5") + (&Error{ + StatusCode: http.StatusTooManyRequests, + Code: "queue_full", + Message: err.Error(), + }).send(w) + return + } if handler.isBatchOperation(meta.CircuitType) { serviceUnavailableError := &Error{ StatusCode: http.StatusServiceUnavailable, diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 43e5784a18..ff79a41f96 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,3 @@ [toolchain] -channel = "1.90.0" +channel = "1.91" components = ["rustfmt", "clippy"] diff --git a/scripts/lint.sh b/scripts/lint.sh index abce5885e9..0e4f39794f 100755 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -19,8 +19,10 @@ cargo clippy --workspace --all-features --all-targets -- -D warnings # Check that READMEs are up-to-date with cargo-rdme echo "Checking READMEs are up-to-date..." -if ! command -v cargo-rdme &> /dev/null; then - cargo install --locked cargo-rdme +CARGO_RDME_VERSION="1.5.0" +if ! command -v cargo-rdme &> /dev/null || + [[ "$(cargo rdme --version)" != "cargo-rdme ${CARGO_RDME_VERSION}" ]]; then + cargo install --locked --version "$CARGO_RDME_VERSION" cargo-rdme fi for toml in $(find program-libs sdk-libs -name '.cargo-rdme.toml' -type f); do crate_dir=$(dirname "$toml") diff --git a/sdk-libs/client/src/indexer/photon_indexer.rs b/sdk-libs/client/src/indexer/photon_indexer.rs index afc7bd3c21..7e35e9995c 100644 --- a/sdk-libs/client/src/indexer/photon_indexer.rs +++ b/sdk-libs/client/src/indexer/photon_indexer.rs @@ -65,9 +65,18 @@ impl PhotonIndexer { } Err(e) => { let is_retryable = match &e { - IndexerError::ApiError(_) => { - warn!("API Error: {}", e); - true + IndexerError::ApiError(msg) => { + // A 404 / "method not found" means the indexer does not implement + // this endpoint; retrying cannot succeed and only burns the backoff + // budget (up to ~44s with the default config). Treat it as non-retryable. + let lower = msg.to_lowercase(); + if lower.contains("method not found") || lower.contains("status 404") { + warn!("Non-retryable API error (endpoint not available): {}", e); + false + } else { + warn!("API Error: {}", e); + true + } } IndexerError::PhotonError { context: _,