From 412b3356ded4b718ac4817b01fd1d4622756ede7 Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Mon, 8 Jun 2026 11:30:12 +0100 Subject: [PATCH 01/16] forester: fix V1 nullify presort retry storm and reduce RPC pressure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The V1 send path's optional get_queue_leaf_indices "presort" was effectively always on (gated only by enable_v1_multi_nullify). On an indexer that does not implement the endpoint it returns 404, which the Photon client treated as a retryable ApiError and backed off ~44s (10 retries) per send. That consumed the entire per-slot send budget, so the chunk aborted at the deadline guard before building/sending any transaction — V1 queues never drained. Changes: - photon_indexer: treat an ApiError whose message contains "method not found" or "status 404" as non-retryable (fail fast instead of ~44s backoff). - forester: add `--enable-v1-presort` (env ENABLE_V1_PRESORT), default off, and gate the get_queue_leaf_indices presort path on it. - forester: replace the fixed 200ms V2 queue poll with adaptive backoff (200ms -> 10s cap, reset to 200ms when work is found) to stop idle V2 trees from re-fetching the queue ~5x/sec for the whole eligible window. - forester: lower default --rpc-pool-size 100 -> 32. Co-Authored-By: Claude Opus 4.8 (1M context) --- forester/src/cli.rs | 10 +++++++- forester/src/config.rs | 7 ++++++ forester/src/epoch_manager.rs | 24 +++++++++++++------ sdk-libs/client/src/indexer/photon_indexer.rs | 15 +++++++++--- 4 files changed, 45 insertions(+), 11 deletions(-) diff --git a/forester/src/cli.rs b/forester/src/cli.rs index f67ce54b06..0fbe713e22 100644 --- a/forester/src/cli.rs +++ b/forester/src/cli.rs @@ -168,7 +168,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 +303,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..c944a36634 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, } @@ -431,6 +435,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, }) } @@ -488,6 +493,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 +517,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..2390f50f8c 100644 --- a/forester/src/epoch_manager.rs +++ b/forester/src/epoch_manager.rs @@ -2265,8 +2265,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 +2340,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 +2359,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 +3072,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, }; @@ -4730,6 +4739,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/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: _, From d630d514207ea2ff7389a2e02b7976ef2c3712f0 Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Wed, 10 Jun 2026 11:00:57 +0100 Subject: [PATCH 02/16] forester: skip compressible-count RPC scan when compression is disabled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The api_server unconditionally spawned run_compressible_provider, which with no in-memory trackers falls back to a full paginated getProgramAccounts scan every 30s. Even with compressible tracking disabled (the default), that heavy scan ties up RPC pool connections and triggers "Failed to get RPC connection" pool failures on mainnet. Only spawn the provider when there is a cheap source — in-memory trackers (compression enabled) or upstream forester APIs to aggregate. Otherwise skip it; the dashboard simply reports no compressible counts. Co-Authored-By: Claude Opus 4.8 (1M context) --- forester/src/api_server.rs | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) 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 Date: Wed, 10 Jun 2026 12:49:26 +0000 Subject: [PATCH 03/16] refetch blockhash for v1 chunk --- forester/src/epoch_manager.rs | 8 +- forester/src/processor/v1/send_transaction.rs | 123 ++++++++++++++++-- forester/src/smart_transaction.rs | 30 +++-- rust-toolchain.toml | 2 +- 4 files changed, 141 insertions(+), 22 deletions(-) diff --git a/forester/src/epoch_manager.rs b/forester/src/epoch_manager.rs index 2390f50f8c..6cf29d6c4e 100644 --- a/forester/src/epoch_manager.rs +++ b/forester/src/epoch_manager.rs @@ -39,7 +39,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 +280,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 +312,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 +336,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 +363,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)), }) } @@ -3092,6 +3097,7 @@ impl EpochManager { &batched_tx_config, *tree_accounts, transaction_builder, + self.v1_send_permits.clone(), ) .await?; diff --git a/forester/src/processor/v1/send_transaction.rs b/forester/src/processor/v1/send_transaction.rs index 86e7bfb650..8a83dc06ac 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::{ @@ -40,6 +43,8 @@ struct PreparedBatchData { 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 +72,7 @@ pub async fn send_batched_transactions, + send_permits: Arc, ) -> std::result::Result { let function_start_time = Instant::now(); @@ -144,18 +150,20 @@ pub async fn send_batched_transactions MAX_ITEMS_PER_CYCLE { @@ -164,17 +172,25 @@ pub async fn send_batched_transactions> = items_to_process .chunks(work_item_batch_size) .map(|c| c.to_vec()) .collect(); let num_chunks = chunks.len(); + let max_concurrent_chunks = effective_max_concurrent_sends + .div_ceil(work_item_batch_size) + .max(1) + .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 +206,37 @@ pub async fn send_batched_transactions= timeout_deadline { return Ok(()); } - // Each chunk gets a fresh blockhash + let permits_to_reserve = work_chunk.len().min(effective_max_concurrent_sends).max(1); + let _chunk_send_permits = match acquire_send_permits( + &send_permits, + permits_to_reserve, + timeout_deadline, + ) + .await + { + Ok(permits) => permits, + 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| { @@ -232,6 +272,8 @@ pub async fn send_batched_transactions>() + .await; for result in results { if let Err(ForesterError::NotEligible) = result { return Err(ForesterError::NotEligible); @@ -392,6 +437,35 @@ fn compute_effective_max_concurrent_sends( effective.max(1) } +async fn acquire_send_permits( + send_permits: &Arc, + permit_count: usize, + timeout_deadline: Instant, +) -> std::result::Result, ForesterError> { + let mut permits = Vec::with_capacity(permit_count); + + for _ in 0..permit_count { + if Instant::now() >= timeout_deadline { + return Err(ForesterError::General { + error: "Timed out waiting for send concurrency permits".to_string(), + }); + } + + let permit_wait = timeout_deadline.saturating_duration_since(Instant::now()); + let permit = timeout(permit_wait, send_permits.clone().acquire_owned()) + .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(), + })?; + permits.push(permit); + } + + Ok(permits) +} + async fn execute_transaction_chunk_sending( transactions: Vec, context: &ChunkSendContext, @@ -406,11 +480,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 +500,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 { diff --git a/forester/src/smart_transaction.rs b/forester/src/smart_transaction.rs index 7f093b0f26..6a3a4b0302 100644 --- a/forester/src/smart_transaction.rs +++ b/forester/src/smart_transaction.rs @@ -45,6 +45,8 @@ impl Default for ConfirmationConfig { } } +const MIN_TRANSACTION_RESEND_INTERVAL: Duration = Duration::from_secs(2); + #[derive(Debug, Clone, Copy)] pub struct TransactionPolicy { pub priority_fee_config: PriorityFeeConfig, @@ -440,6 +442,11 @@ async fn send_prepared_transaction( let last_valid_block_height = transaction.last_valid_block_height(); let rpc = &*rpc; 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? { @@ -457,16 +464,19 @@ async fn send_prepared_transaction( }); } - 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 { + 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()), + } + next_send_at = Instant::now() + resend_interval; } if let Some(signature) = confirmed_signature_or_error(rpc, signature).await? { 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"] From 2627db3d3a848e153df27f4cc09851dc124fb297 Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Wed, 10 Jun 2026 14:22:58 +0100 Subject: [PATCH 04/16] Fix atomic V1 send permit reservation --- forester/src/processor/v1/send_transaction.rs | 101 ++++++++++++++---- 1 file changed, 80 insertions(+), 21 deletions(-) diff --git a/forester/src/processor/v1/send_transaction.rs b/forester/src/processor/v1/send_transaction.rs index 8a83dc06ac..7a67b632ba 100644 --- a/forester/src/processor/v1/send_transaction.rs +++ b/forester/src/processor/v1/send_transaction.rs @@ -441,29 +441,28 @@ async fn acquire_send_permits( send_permits: &Arc, permit_count: usize, timeout_deadline: Instant, -) -> std::result::Result, ForesterError> { - let mut permits = Vec::with_capacity(permit_count); - - for _ in 0..permit_count { - if Instant::now() >= timeout_deadline { - return Err(ForesterError::General { - error: "Timed out waiting for send concurrency permits".to_string(), - }); - } - - let permit_wait = timeout_deadline.saturating_duration_since(Instant::now()); - let permit = timeout(permit_wait, send_permits.clone().acquire_owned()) - .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(), - })?; - permits.push(permit); +) -> std::result::Result { + if Instant::now() >= timeout_deadline { + return Err(ForesterError::General { + error: "Timed out waiting for send concurrency permits".to_string(), + }); } - Ok(permits) + 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( @@ -665,3 +664,63 @@ 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); + } +} From aa948ea99f216d44ebe60338575713f246d37b2b Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Wed, 10 Jun 2026 13:36:11 +0000 Subject: [PATCH 05/16] fix tests --- forester/tests/e2e_test.rs | 1 + forester/tests/legacy/priority_fee_test.rs | 1 + forester/tests/legacy/test_utils.rs | 1 + forester/tests/priority_fee_test.rs | 1 + forester/tests/test_utils.rs | 1 + 5 files changed, 5 insertions(+) diff --git a/forester/tests/e2e_test.rs b/forester/tests/e2e_test.rs index c11500f29a..21460bb276 100644 --- a/forester/tests/e2e_test.rs +++ b/forester/tests/e2e_test.rs @@ -303,6 +303,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..b475e6acd6 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, diff --git a/forester/tests/legacy/test_utils.rs b/forester/tests/legacy/test_utils.rs index fdac97ba34..960d73873c 100644 --- a/forester/tests/legacy/test_utils.rs +++ b/forester/tests/legacy/test_utils.rs @@ -125,6 +125,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..a02df66b56 100644 --- a/forester/tests/priority_fee_test.rs +++ b/forester/tests/priority_fee_test.rs @@ -94,6 +94,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..9061a526f4 100644 --- a/forester/tests/test_utils.rs +++ b/forester/tests/test_utils.rs @@ -138,6 +138,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, } } From 8de3b8fd66086f6b6a9efac5b906d0d9f40199cc Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 10:38:42 +0000 Subject: [PATCH 06/16] fix(forester): bound v1 RPC fanout --- forester/src/processor/v1/send_transaction.rs | 112 ++++++++++++----- forester/src/processor/v1/tx_builder.rs | 45 ++++++- forester/src/smart_transaction.rs | 116 +++++++++++++++++- .../tests/integration_tests.rs | 2 +- 4 files changed, 232 insertions(+), 43 deletions(-) diff --git a/forester/src/processor/v1/send_transaction.rs b/forester/src/processor/v1/send_transaction.rs index 7a67b632ba..baa7109992 100644 --- a/forester/src/processor/v1/send_transaction.rs +++ b/forester/src/processor/v1/send_transaction.rs @@ -39,6 +39,15 @@ 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>, @@ -151,12 +160,17 @@ pub async fn send_batched_transactions MAX_ITEMS_PER_CYCLE { - &data.work_items[..MAX_ITEMS_PER_CYCLE] - } else { - &data.work_items - }; + if cycle_item_count < data.work_items.len() { + info!( + tree = %tree_accounts.merkle_tree, + total_work_items = data.work_items.len(), + items_this_cycle = cycle_item_count, + configured_work_item_batch_size, + "Capping V1 work items for this cycle" + ); + } + + let items_to_process = &data.work_items[..cycle_item_count]; // Process chunks concurrently: each chunk fetches proofs and builds independently, // while transaction sends share the process-wide V1 concurrency limit. @@ -180,10 +198,7 @@ pub async fn send_batched_transactions permits, - Err(e) => { - warn!( - tree = %tree_id, - error = ?e, - "Skipping chunk because send concurrency permits were not available before deadline" - ); - return Ok(()); - } - }; + 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(()); @@ -269,11 +278,13 @@ pub async fn send_batched_transactions usize { + let limit = configured_work_item_batch_size + .max(1) + .min(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, @@ -723,4 +751,24 @@ mod tests { 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/smart_transaction.rs b/forester/src/smart_transaction.rs index 6a3a4b0302..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,7 +46,9 @@ impl Default for ConfirmationConfig { } } -const MIN_TRANSACTION_RESEND_INTERVAL: Duration = Duration::from_secs(2); +// 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 { @@ -441,6 +444,8 @@ 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 @@ -457,7 +462,17 @@ 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, @@ -465,6 +480,7 @@ async fn send_prepared_transaction( } 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, @@ -472,8 +488,24 @@ async fn send_prepared_transaction( // 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) 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; @@ -501,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, @@ -509,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()); } @@ -516,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/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs b/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs index 2c3e82972a..9b40b900e5 100644 --- a/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs +++ b/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs @@ -3863,7 +3863,7 @@ async fn test_d9_edge_many_literals() { #[tokio::test] async fn test_d9_edge_mixed() { use csdk_anchor_full_derived_test::d9_seeds::{ - edge_cases::{_UNDERSCORE_CONST, AB, SEED_123}, + edge_cases::{AB, SEED_123, _UNDERSCORE_CONST}, D9EdgeMixedParams, }; From 5c08319c0dbe660ff21cc6b62886f2691f1feb3e Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 10:38:30 +0000 Subject: [PATCH 07/16] fix: isolate prover queues by network --- forester-utils/src/utils.rs | 45 ++-- forester/src/epoch_manager.rs | 2 + forester/src/processor/v1/helpers.rs | 7 +- forester/src/processor/v2/common.rs | 1 + forester/src/processor/v2/processor.rs | 45 ++-- forester/src/processor/v2/proof_worker.rs | 9 +- forester/src/processor/v2/root_guard.rs | 38 ++-- prover/client/src/proof_client.rs | 12 ++ prover/server/server/queue.go | 239 +++++++++++++++++---- prover/server/server/queue_job.go | 38 ++-- prover/server/server/queue_network_test.go | 68 ++++++ prover/server/server/server.go | 41 +++- 12 files changed, 397 insertions(+), 148 deletions(-) create mode 100644 prover/server/server/queue_network_test.go diff --git a/forester-utils/src/utils.rs b/forester-utils/src/utils.rs index b78cb88df6..e3739245cb 100644 --- a/forester-utils/src/utils.rs +++ b/forester-utils/src/utils.rs @@ -5,7 +5,6 @@ use light_client::{ rpc::{Rpc, RpcError}, }; use solana_sdk::{signature::Signer, transaction::Transaction}; -use tokio::time::sleep; use tracing::{error, warn}; 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,19 @@ 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 { + warn!( + lag, + max_lag_slots, rpc_slot, indexer_slot, "indexer freshness gate rejected proof work" + ); + return Err(ForesterUtilsError::Indexer(format!( + "Indexer is behind {lag} slots (maximum allowed: {max_lag_slots})" + ))); } Ok(()) } diff --git a/forester/src/epoch_manager.rs b/forester/src/epoch_manager.rs index 6cf29d6c4e..b4b1bb5a4d 100644 --- a/forester/src/epoch_manager.rs +++ b/forester/src/epoch_manager.rs @@ -3176,6 +3176,8 @@ impl EpochManager { .external_services .prover_max_wait_time .unwrap_or(Duration::from_secs(600)), + network: std::env::var("FORESTER_NETWORK") + .unwrap_or_else(|_| "default".to_string()), }), ops_cache: self.ops_cache.clone(), epoch_phases: epoch_info.phases.clone(), 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/v2/common.rs b/forester/src/processor/v2/common.rs index 191160bea4..c72868b842 100644 --- a/forester/src/processor/v2/common.rs +++ b/forester/src/processor/v2/common.rs @@ -79,6 +79,7 @@ pub struct ProverConfig { pub api_key: Option, pub polling_interval: Duration, pub max_wait_time: Duration, + 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..540ab349f2 100644 --- a/forester/src/processor/v2/proof_worker.rs +++ b/forester/src/processor/v2/proof_worker.rs @@ -139,19 +139,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()), } } 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/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/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, From 52c62c6074cad7e1eae3b29841f0df2513fe3994 Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 11:03:18 +0000 Subject: [PATCH 08/16] fix(prover): support offline mounted keys --- prover/server/prover/common/key_downloader.go | 17 ++++++++ .../common/key_downloader_offline_test.go | 40 +++++++++++++++++++ 2 files changed, 57 insertions(+) create mode 100644 prover/server/prover/common/key_downloader_offline_test.go diff --git a/prover/server/prover/common/key_downloader.go b/prover/server/prover/common/key_downloader.go index d3e93e29ae..2ac59320e1 100644 --- a/prover/server/prover/common/key_downloader.go +++ b/prover/server/prover/common/key_downloader.go @@ -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) 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..f05064b83c --- /dev/null +++ b/prover/server/prover/common/key_downloader_offline_test.go @@ -0,0 +1,40 @@ +package common + +import ( + "os" + "path/filepath" + "testing" +) + +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") + } +} From 63a819783b02284381a57b4f9405f8cde1b48d11 Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 11:38:51 +0000 Subject: [PATCH 09/16] fix(forester): gate v2 work on indexer freshness --- forester-utils/src/utils.rs | 6 +----- forester/src/epoch_manager.rs | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/forester-utils/src/utils.rs b/forester-utils/src/utils.rs index e3739245cb..0fa07e1546 100644 --- a/forester-utils/src/utils.rs +++ b/forester-utils/src/utils.rs @@ -5,7 +5,7 @@ use light_client::{ rpc::{Rpc, RpcError}, }; use solana_sdk::{signature::Signer, transaction::Transaction}; -use tracing::{error, warn}; +use tracing::error; use crate::error::ForesterUtilsError; @@ -45,10 +45,6 @@ pub async fn wait_for_indexer(rpc: &R) -> Result<(), ForesterUtilsError> .unwrap_or(100); let lag = rpc_slot.saturating_sub(indexer_slot); if lag > max_lag_slots { - warn!( - lag, - max_lag_slots, rpc_slot, indexer_slot, "indexer freshness gate rejected proof work" - ); return Err(ForesterUtilsError::Indexer(format!( "Indexer is behind {lag} slots (maximum allowed: {max_lag_slots})" ))); diff --git a/forester/src/epoch_manager.rs b/forester/src/epoch_manager.rs index b4b1bb5a4d..7875a00509 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::{ @@ -2247,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 From 4b243d43c3f3f5f9bf44c57e38652aef01710387 Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 17:53:50 +0000 Subject: [PATCH 10/16] fix(prover): download keys from public S3 bucket --- prover/server/prover/common/key_downloader.go | 73 +++++++++++++++---- .../common/key_downloader_offline_test.go | 67 +++++++++++++++++ 2 files changed, 125 insertions(+), 15 deletions(-) diff --git a/prover/server/prover/common/key_downloader.go b/prover/server/prover/common/key_downloader.go index 2ac59320e1..284993f493 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://light-protocol-proving-keys.s3.eu-north-1.amazonaws.com" DefaultMaxRetries = 10 DefaultRetryDelay = 5 * time.Second DefaultMaxRetryDelay = 5 * time.Minute @@ -27,15 +27,19 @@ type DownloadConfig struct { RetryDelay time.Duration MaxRetryDelay time.Duration AutoDownload bool + // RequireChecksum rejects downloads when BaseURL does not publish a CHECKSUM + // manifest. The default S3 bucket currently publishes keys without a manifest. + RequireChecksum bool } func DefaultDownloadConfig() *DownloadConfig { return &DownloadConfig{ - BaseURL: DefaultBaseURL, - MaxRetries: DefaultMaxRetries, - RetryDelay: DefaultRetryDelay, - MaxRetryDelay: DefaultMaxRetryDelay, - AutoDownload: true, + BaseURL: DefaultBaseURL, + MaxRetries: DefaultMaxRetries, + RetryDelay: DefaultRetryDelay, + MaxRetryDelay: DefaultMaxRetryDelay, + AutoDownload: true, + RequireChecksum: false, } } @@ -80,6 +84,18 @@ func downloadChecksum(config *DownloadConfig) error { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { + if !config.RequireChecksum && + (resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusNotFound) { + globalChecksumCaches.caches[config.BaseURL] = &checksumCacheEntry{ + checksums: make(map[string]string), + loaded: true, + } + logging.Logger().Warn(). + Int("status_code", resp.StatusCode). + Str("base_url", config.BaseURL). + Msg("CHECKSUM manifest unavailable; downloaded keys will not be checksum-verified") + return nil + } return fmt.Errorf("failed to download CHECKSUM file: HTTP %d", resp.StatusCode) } @@ -315,11 +331,23 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { expectedChecksum, checksumExists := entry.checksums[filename] globalChecksumCaches.mu.RUnlock() - if !checksumExists { + if !checksumExists && config.RequireChecksum { return fmt.Errorf("no checksum found for %s", filename) } 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) + } + + if !checksumExists { + logging.Logger().Warn(). + Str("file", filename). + Int64("size", fileInfo.Size()). + Msg("Using existing key file without checksum verification") + return nil + } + logging.Logger().Info(). Str("file", filename). Int64("size", fileInfo.Size()). @@ -370,18 +398,33 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { return err } - valid, err := verifyChecksum(keyPath, expectedChecksum) - if err != nil { - return fmt.Errorf("failed to verify downloaded file: %w", err) - } - if !valid { - os.Remove(keyPath) - return fmt.Errorf("downloaded file checksum mismatch") + if checksumExists { + valid, err := verifyChecksum(keyPath, expectedChecksum) + if err != nil { + return fmt.Errorf("failed to verify downloaded file: %w", err) + } + if !valid { + os.Remove(keyPath) + return fmt.Errorf("downloaded file checksum mismatch") + } + } else { + fileInfo, err := os.Stat(keyPath) + if err != nil { + return fmt.Errorf("failed to stat downloaded file: %w", err) + } + if !fileInfo.Mode().IsRegular() || fileInfo.Size() == 0 { + os.Remove(keyPath) + return fmt.Errorf("downloaded key is not a non-empty regular file") + } + logging.Logger().Warn(). + Str("file", filename). + Int64("size", fileInfo.Size()). + Msg("Key file downloaded without checksum verification") } 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 index f05064b83c..e517444202 100644 --- a/prover/server/prover/common/key_downloader_offline_test.go +++ b/prover/server/prover/common/key_downloader_offline_test.go @@ -1,9 +1,13 @@ package common import ( + "net/http" + "net/http/httptest" "os" "path/filepath" + "strings" "testing" + "time" ) func TestDownloadKeyOfflineUsesExistingNonEmptyFile(t *testing.T) { @@ -38,3 +42,66 @@ func TestDownloadKeyOfflineRejectsMissingOrEmptyFile(t *testing.T) { t.Fatal("empty offline key unexpectedly accepted") } } + +func TestDownloadKeyAllowsMissingChecksumManifest(t *testing.T) { + const filename = "test.key" + const contents = "downloaded-key" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/CHECKSUM": + w.WriteHeader(http.StatusForbidden) + 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("manifest-less 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 TestDownloadKeyCanRequireChecksumManifest(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 + config.RequireChecksum = true + + keyPath := filepath.Join(t.TempDir(), filename) + err := DownloadKey(keyPath, config) + if err == nil { + t.Fatal("download without required checksum manifest unexpectedly succeeded") + } + if !strings.Contains(err.Error(), "CHECKSUM file: HTTP 403") { + t.Fatalf("unexpected error: %v", err) + } +} From bf6a45da48a8d398907d158275de83a4bbfc863c Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 17:54:10 +0000 Subject: [PATCH 11/16] fix(forester): backpressure saturated prover queues --- forester/src/cli.rs | 8 + forester/src/config.rs | 3 + forester/src/epoch_manager.rs | 2 + forester/src/processor/v2/common.rs | 1 + forester/src/processor/v2/proof_worker.rs | 175 ++++++++++++++++++++- forester/tests/e2e_test.rs | 1 + forester/tests/legacy/priority_fee_test.rs | 1 + forester/tests/legacy/test_utils.rs | 1 + forester/tests/priority_fee_test.rs | 1 + forester/tests/test_utils.rs | 1 + 10 files changed, 187 insertions(+), 7 deletions(-) diff --git a/forester/src/cli.rs b/forester/src/cli.rs index 0fbe713e22..e8f066a3bd 100644 --- a/forester/src/cli.rs +++ b/forester/src/cli.rs @@ -79,6 +79,14 @@ pub struct StartArgs { )] pub prover_max_wait_time_secs: Option, + #[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, diff --git a/forester/src/config.rs b/forester/src/config.rs index c944a36634..1a59460eeb 100644 --- a/forester/src/config.rs +++ b/forester/src/config.rs @@ -56,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, @@ -288,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(), @@ -455,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(), diff --git a/forester/src/epoch_manager.rs b/forester/src/epoch_manager.rs index 7875a00509..9f7a15be50 100644 --- a/forester/src/epoch_manager.rs +++ b/forester/src/epoch_manager.rs @@ -3190,6 +3190,7 @@ 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()), }), @@ -4734,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, }, diff --git a/forester/src/processor/v2/common.rs b/forester/src/processor/v2/common.rs index c72868b842..80e2abc29f 100644 --- a/forester/src/processor/v2/common.rs +++ b/forester/src/processor/v2/common.rs @@ -79,6 +79,7 @@ pub struct ProverConfig { pub api_key: Option, pub polling_interval: Duration, pub max_wait_time: Duration, + pub max_concurrent_jobs: usize, pub network: String, } diff --git a/forester/src/processor/v2/proof_worker.rs b/forester/src/processor/v2/proof_worker.rs index 540ab349f2..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::{ @@ -170,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; @@ -202,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={}", @@ -251,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, @@ -281,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={}", @@ -424,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/tests/e2e_test.rs b/forester/tests/e2e_test.rs index 21460bb276..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, diff --git a/forester/tests/legacy/priority_fee_test.rs b/forester/tests/legacy/priority_fee_test.rs index b475e6acd6..96e22e4134 100644 --- a/forester/tests/legacy/priority_fee_test.rs +++ b/forester/tests/legacy/priority_fee_test.rs @@ -102,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 960d73873c..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, diff --git a/forester/tests/priority_fee_test.rs b/forester/tests/priority_fee_test.rs index a02df66b56..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") diff --git a/forester/tests/test_utils.rs b/forester/tests/test_utils.rs index 9061a526f4..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, From d60515c4bfdaaa6a2a4445ffb1ea4735b5c859a3 Mon Sep 17 00:00:00 2001 From: Sergey Timoshin Date: Mon, 20 Jul 2026 18:01:29 +0000 Subject: [PATCH 12/16] chore: satisfy nightly rustfmt --- .../csdk-anchor-full-derived-test/tests/integration_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs b/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs index 9b40b900e5..2c3e82972a 100644 --- a/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs +++ b/sdk-tests/csdk-anchor-full-derived-test/tests/integration_tests.rs @@ -3863,7 +3863,7 @@ async fn test_d9_edge_many_literals() { #[tokio::test] async fn test_d9_edge_mixed() { use csdk_anchor_full_derived_test::d9_seeds::{ - edge_cases::{AB, SEED_123, _UNDERSCORE_CONST}, + edge_cases::{_UNDERSCORE_CONST, AB, SEED_123}, D9EdgeMixedParams, }; From 262568deb216fb9237ac2b3b8eb1c73c7268533e Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Tue, 21 Jul 2026 14:15:53 +0100 Subject: [PATCH 13/16] fix prover key downloads and rust 1.91 lint --- cli/src/utils/processProverServer.ts | 2 + forester/src/processor/v1/send_transaction.rs | 4 +- .../tests/batched_merkle_tree_test.rs | 8 +-- .../compressed-token-test/tests/v1.rs | 2 +- program-tests/system-test/tests/test.rs | 2 +- .../src/account_compression_cpi/sdk.rs | 4 +- prover/server/prover/common/key_downloader.go | 71 +++++-------------- .../common/key_downloader_offline_test.go | 17 +++-- 8 files changed, 35 insertions(+), 75 deletions(-) diff --git a/cli/src/utils/processProverServer.ts b/cli/src/utils/processProverServer.ts index 15d47fdf0f..2b09fcee1e 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()); @@ -95,6 +96,7 @@ 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("--download-url", PROVING_KEYS_BASE_URL); if (redisUrl) { args.push("--redis-url", redisUrl); diff --git a/forester/src/processor/v1/send_transaction.rs b/forester/src/processor/v1/send_transaction.rs index baa7109992..28ae8886ac 100644 --- a/forester/src/processor/v1/send_transaction.rs +++ b/forester/src/processor/v1/send_transaction.rs @@ -449,9 +449,7 @@ fn compute_effective_max_concurrent_sends( } fn cycle_item_limit(configured_work_item_batch_size: usize, work_item_count: usize) -> usize { - let limit = configured_work_item_batch_size - .max(1) - .min(MAX_ITEMS_PER_CYCLE); + let limit = configured_work_item_batch_size.clamp(1, MAX_ITEMS_PER_CYCLE); work_item_count.min(limit) } 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/server/prover/common/key_downloader.go b/prover/server/prover/common/key_downloader.go index 284993f493..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://light-protocol-proving-keys.s3.eu-north-1.amazonaws.com" + DefaultBaseURL = "https://d1wbn9ra8wjh7t.cloudfront.net" DefaultMaxRetries = 10 DefaultRetryDelay = 5 * time.Second DefaultMaxRetryDelay = 5 * time.Minute @@ -27,19 +27,15 @@ type DownloadConfig struct { RetryDelay time.Duration MaxRetryDelay time.Duration AutoDownload bool - // RequireChecksum rejects downloads when BaseURL does not publish a CHECKSUM - // manifest. The default S3 bucket currently publishes keys without a manifest. - RequireChecksum bool } func DefaultDownloadConfig() *DownloadConfig { return &DownloadConfig{ - BaseURL: DefaultBaseURL, - MaxRetries: DefaultMaxRetries, - RetryDelay: DefaultRetryDelay, - MaxRetryDelay: DefaultMaxRetryDelay, - AutoDownload: true, - RequireChecksum: false, + BaseURL: DefaultBaseURL, + MaxRetries: DefaultMaxRetries, + RetryDelay: DefaultRetryDelay, + MaxRetryDelay: DefaultMaxRetryDelay, + AutoDownload: true, } } @@ -72,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") @@ -84,18 +80,6 @@ func downloadChecksum(config *DownloadConfig) error { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - if !config.RequireChecksum && - (resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusNotFound) { - globalChecksumCaches.caches[config.BaseURL] = &checksumCacheEntry{ - checksums: make(map[string]string), - loaded: true, - } - logging.Logger().Warn(). - Int("status_code", resp.StatusCode). - Str("base_url", config.BaseURL). - Msg("CHECKSUM manifest unavailable; downloaded keys will not be checksum-verified") - return nil - } return fmt.Errorf("failed to download CHECKSUM file: HTTP %d", resp.StatusCode) } @@ -331,7 +315,7 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { expectedChecksum, checksumExists := entry.checksums[filename] globalChecksumCaches.mu.RUnlock() - if !checksumExists && config.RequireChecksum { + if !checksumExists { return fmt.Errorf("no checksum found for %s", filename) } @@ -340,14 +324,6 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { return fmt.Errorf("existing key file is not a non-empty regular file: %s", filename) } - if !checksumExists { - logging.Logger().Warn(). - Str("file", filename). - Int64("size", fileInfo.Size()). - Msg("Using existing key file without checksum verification") - return nil - } - logging.Logger().Info(). Str("file", filename). Int64("size", fileInfo.Size()). @@ -388,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). @@ -398,28 +374,13 @@ func DownloadKey(keyPath string, config *DownloadConfig) error { return err } - if checksumExists { - valid, err := verifyChecksum(keyPath, expectedChecksum) - if err != nil { - return fmt.Errorf("failed to verify downloaded file: %w", err) - } - if !valid { - os.Remove(keyPath) - return fmt.Errorf("downloaded file checksum mismatch") - } - } else { - fileInfo, err := os.Stat(keyPath) - if err != nil { - return fmt.Errorf("failed to stat downloaded file: %w", err) - } - if !fileInfo.Mode().IsRegular() || fileInfo.Size() == 0 { - os.Remove(keyPath) - return fmt.Errorf("downloaded key is not a non-empty regular file") - } - logging.Logger().Warn(). - Str("file", filename). - Int64("size", fileInfo.Size()). - Msg("Key file downloaded without checksum verification") + valid, err := verifyChecksum(keyPath, expectedChecksum) + if err != nil { + return fmt.Errorf("failed to verify downloaded file: %w", err) + } + if !valid { + os.Remove(keyPath) + return fmt.Errorf("downloaded file checksum mismatch") } logging.Logger().Info(). diff --git a/prover/server/prover/common/key_downloader_offline_test.go b/prover/server/prover/common/key_downloader_offline_test.go index e517444202..afe19b61bb 100644 --- a/prover/server/prover/common/key_downloader_offline_test.go +++ b/prover/server/prover/common/key_downloader_offline_test.go @@ -5,7 +5,6 @@ import ( "net/http/httptest" "os" "path/filepath" - "strings" "testing" "time" ) @@ -43,14 +42,15 @@ func TestDownloadKeyOfflineRejectsMissingOrEmptyFile(t *testing.T) { } } -func TestDownloadKeyAllowsMissingChecksumManifest(t *testing.T) { +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.WriteHeader(http.StatusForbidden) + _, _ = w.Write([]byte(checksum + " " + filename + "\n")) case "/" + filename: _, _ = w.Write([]byte(contents)) default: @@ -60,14 +60,14 @@ func TestDownloadKeyAllowsMissingChecksumManifest(t *testing.T) { defer server.Close() config := DefaultDownloadConfig() - config.BaseURL = server.URL + 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("manifest-less download failed: %v", err) + t.Fatalf("download failed: %v", err) } got, err := os.ReadFile(keyPath) @@ -79,7 +79,7 @@ func TestDownloadKeyAllowsMissingChecksumManifest(t *testing.T) { } } -func TestDownloadKeyCanRequireChecksumManifest(t *testing.T) { +func TestDownloadKeyRejectsMissingChecksumManifest(t *testing.T) { const filename = "test.key" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -94,14 +94,13 @@ func TestDownloadKeyCanRequireChecksumManifest(t *testing.T) { config := DefaultDownloadConfig() config.BaseURL = server.URL config.MaxRetries = 1 - config.RequireChecksum = true keyPath := filepath.Join(t.TempDir(), filename) err := DownloadKey(keyPath, config) if err == nil { - t.Fatal("download without required checksum manifest unexpectedly succeeded") + t.Fatal("download without checksum manifest unexpectedly succeeded") } - if !strings.Contains(err.Error(), "CHECKSUM file: HTTP 403") { + if got, want := err.Error(), "failed to load checksums: failed to download CHECKSUM file: HTTP 403"; got != want { t.Fatalf("unexpected error: %v", err) } } From aeeeab9fbf54371d84a5cf2b9b8d7b46a5695772 Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Tue, 21 Jul 2026 14:29:46 +0100 Subject: [PATCH 14/16] fix prover boolean flag parsing --- cli/src/utils/processProverServer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cli/src/utils/processProverServer.ts b/cli/src/utils/processProverServer.ts index 2b09fcee1e..49792ff16f 100644 --- a/cli/src/utils/processProverServer.ts +++ b/cli/src/utils/processProverServer.ts @@ -95,7 +95,7 @@ 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) { From fa29aff10a9dd6fe3637517cb17980a001aa9bf2 Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Tue, 21 Jul 2026 14:48:29 +0100 Subject: [PATCH 15/16] install cargo-rdme intralink toolchain in lint --- scripts/lint.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/lint.sh b/scripts/lint.sh index abce5885e9..658e0a3dae 100755 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -22,6 +22,7 @@ echo "Checking READMEs are up-to-date..." if ! command -v cargo-rdme &> /dev/null; then cargo install --locked cargo-rdme fi +cargo rdme install-rust-toolchain-for-intralinks for toml in $(find program-libs sdk-libs -name '.cargo-rdme.toml' -type f); do crate_dir=$(dirname "$toml") echo "Checking README in $crate_dir..." From bf6c58bc7621cb3bde5c729a1748092b2d7af4ee Mon Sep 17 00:00:00 2001 From: Sergei Timoshin Date: Tue, 21 Jul 2026 14:53:36 +0100 Subject: [PATCH 16/16] pin cargo-rdme for deterministic lint --- scripts/lint.sh | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/scripts/lint.sh b/scripts/lint.sh index 658e0a3dae..0e4f39794f 100755 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -19,10 +19,11 @@ 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 -cargo rdme install-rust-toolchain-for-intralinks for toml in $(find program-libs sdk-libs -name '.cargo-rdme.toml' -type f); do crate_dir=$(dirname "$toml") echo "Checking README in $crate_dir..."