fix: isolate prover queues by network

Author: sergeytimoshin

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<R: Rpc>(
 
 pub async fn wait_for_indexer<R: Rpc>(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<R: Rpc>(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::<u64>().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(())
 }

forester/src/epoch_manager.rs

@@ -3176,6 +3176,8 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                     .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(),

forester/src/processor/v1/helpers.rs

@@ -96,15 +96,16 @@ pub async fn fetch_proofs_and_create_instructions<R: Rpc>(
     };
 
     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();

forester/src/processor/v2/common.rs

@@ -79,6 +79,7 @@ pub struct ProverConfig {
     pub api_key: Option<String>,
     pub polling_interval: Duration,
     pub max_wait_time: Duration,
+    pub network: String,
 }
 
 #[derive(Debug)]

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 => {

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()),
         }
     }
 

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))
         );
     }
 

prover/client/src/proof_client.rs

@@ -63,6 +63,7 @@ pub struct ProofClient {
     polling_interval: Duration,
     max_wait_time: Duration,
     api_key: Option<String>,
+    network: Option<String>,
     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())