make reorg_threshold private so that it can be modified for tests

Author: Zoran Cvetkov

chain/ethereum/src/chain.rs

@@ -610,7 +610,7 @@ impl Blockchain for Chain {
                 // present in the DB.
                 Box::new(PollingBlockIngestor::new(
                     logger,
-                    graph::env::ENV_VARS.reorg_threshold,
+                    graph::env::ENV_VARS.reorg_threshold(),
                     self.chain_client(),
                     self.chain_store().cheap_clone(),
                     self.polling_ingestor_interval,

graph/src/data/subgraph/mod.rs

@@ -504,9 +504,9 @@ impl Graft {
             // The graft point must be at least `reorg_threshold` blocks
             // behind the subgraph head so that a reorg can not affect the
             // data that we copy for grafting
-            (Some(ptr), true) if self.block + ENV_VARS.reorg_threshold > ptr.number => Err(GraftBaseInvalid(format!(
+            (Some(ptr), true) if self.block + ENV_VARS.reorg_threshold() > ptr.number => Err(GraftBaseInvalid(format!(
                 "failed to graft onto `{}` at block {} since it's only at block {} which is within the reorg threshold of {} blocks",
-                self.base, self.block, ptr.number, ENV_VARS.reorg_threshold
+                self.base, self.block, ptr.number, ENV_VARS.reorg_threshold()
             ))),
             // If the base deployment is failed *and* the `graft.block` is not
             // less than the `base.block`, the graft shouldn't be permitted.
@@ -696,36 +696,43 @@ impl<C: Blockchain> UnvalidatedSubgraphManifest<C> {
         validate_graft_base: bool,
     ) -> Result<SubgraphManifest<C>, Vec<SubgraphManifestValidationError>> {
         let mut errors: Vec<SubgraphManifestValidationError> = vec![];
+        println!("validate 1");
 
         // Validate that the manifest has at least one data source
         if self.0.data_sources.is_empty() {
             errors.push(SubgraphManifestValidationError::NoDataSources);
         }
+        println!("validate 2");
 
         for ds in &self.0.data_sources {
             errors.extend(ds.validate(&self.0.spec_version).into_iter().map(|e| {
                 SubgraphManifestValidationError::DataSourceValidation(ds.name().to_owned(), e)
             }));
         }
+        println!("validate 3");
 
         // For API versions newer than 0.0.5, validate that all mappings uses the same api_version
         if let Err(different_api_versions) = self.0.unified_mapping_api_version() {
             errors.push(different_api_versions.into());
         };
+        println!("validate 4 errors: {:?}", errors);
 
         let mut networks = self
             .0
             .data_sources
             .iter()
             .filter_map(|d| Some(d.network()?.to_string()))
             .collect::<Vec<String>>();
+        println!("validate 5");
         networks.sort();
         networks.dedup();
+        println!("validate 6");
         match networks.len() {
             0 => errors.push(SubgraphManifestValidationError::EthereumNetworkRequired),
             1 => (),
             _ => errors.push(SubgraphManifestValidationError::MultipleEthereumNetworks),
         }
+        println!("validate 7 errors: {:?}", errors);
 
         if let Some(graft) = &self.0.graft {
             if validate_graft_base {
@@ -734,13 +741,15 @@ impl<C: Blockchain> UnvalidatedSubgraphManifest<C> {
                 }
             }
         }
+        println!("validate 8 errors: {:?}", errors);
 
         // Validate subgraph feature usage and declaration.
         if self.0.spec_version >= SPEC_VERSION_0_0_4 {
             if let Err(feature_validation_error) = validate_subgraph_features(&self.0) {
                 errors.push(feature_validation_error.into())
             }
         }
+        println!("validate 9");
 
         match errors.is_empty() {
             true => Ok(self.0),

graph/src/env/mod.rs

@@ -5,6 +5,7 @@ mod store;
 use envconfig::Envconfig;
 use lazy_static::lazy_static;
 use semver::Version;
+use std::sync::Mutex;
 use std::{collections::HashSet, env::VarError, fmt, str::FromStr, time::Duration};
 
 use self::graphql::*;
@@ -17,6 +18,7 @@ use crate::{
 
 lazy_static! {
     pub static ref ENV_VARS: EnvVars = EnvVars::from_env().unwrap();
+    pub static ref TEST_WITH_NO_REORG: Mutex<bool> = Mutex::new(false);
 }
 
 /// Panics if:
@@ -181,7 +183,7 @@ pub struct EnvVars {
     pub static_filters_threshold: usize,
     /// Set by the environment variable `ETHEREUM_REORG_THRESHOLD`. The default
     /// value is 250 blocks.
-    pub reorg_threshold: BlockNumber,
+    reorg_threshold: BlockNumber,
     /// The time to wait between polls when using polling block ingestor.
     /// The value is set by `ETHERUM_POLLING_INTERVAL` in millis and the
     /// default is 1000.
@@ -362,6 +364,13 @@ impl EnvVars {
             .filter(|x| !x.is_empty())
             .collect()
     }
+    pub fn reorg_threshold(&self) -> i32 {
+        if *TEST_WITH_NO_REORG.lock().unwrap() {
+            0
+        } else {
+            self.reorg_threshold
+        }
+    }
 }
 
 impl Default for EnvVars {

node/src/chain.rs

@@ -457,7 +457,7 @@ pub async fn networks_as_chains(
                     Arc::new(adapter_selector),
                     Arc::new(EthereumRuntimeAdapterBuilder {}),
                     eth_adapters,
-                    ENV_VARS.reorg_threshold,
+                    ENV_VARS.reorg_threshold(),
                     polling_interval,
                     true,
                 );

node/src/manager/commands/prune.rs

@@ -188,13 +188,13 @@ pub async fn run(
 
     println!("prune {deployment}");
     println!("    latest: {latest}");
-    println!("     final: {}", latest - ENV_VARS.reorg_threshold);
+    println!("     final: {}", latest - ENV_VARS.reorg_threshold());
     println!("  earliest: {}\n", latest - history);
 
     let mut req = PruneRequest::new(
         &deployment,
         history,
-        ENV_VARS.reorg_threshold,
+        ENV_VARS.reorg_threshold(),
         status.earliest_block_number,
         latest,
     )?;
@@ -217,7 +217,7 @@ pub async fn run(
         store.subgraph_store().set_history_blocks(
             &deployment,
             history,
-            ENV_VARS.reorg_threshold,
+            ENV_VARS.reorg_threshold(),
         )?;
     }
 

node/src/manager/commands/rewind.rs

@@ -133,13 +133,13 @@ pub async fn run(
         let deployment_details = deployment_store.deployment_details_for_id(locator)?;
         let block_number_to = block_ptr_to.as_ref().map(|b| b.number).unwrap_or(0);
 
-        if block_number_to < deployment_details.earliest_block_number + ENV_VARS.reorg_threshold {
+        if block_number_to < deployment_details.earliest_block_number + ENV_VARS.reorg_threshold() {
             bail!(
                 "The block number {} is not safe to rewind to for deployment {}. The earliest block number of this deployment is {}. You can only safely rewind to block number {}",
                 block_ptr_to.as_ref().map(|b| b.number).unwrap_or(0),
                 locator,
                 deployment_details.earliest_block_number,
-                deployment_details.earliest_block_number + ENV_VARS.reorg_threshold
+                deployment_details.earliest_block_number + ENV_VARS.reorg_threshold()
             );
         }
     }

store/postgres/src/block_store.rs

@@ -507,7 +507,7 @@ impl BlockStore {
             };
 
             if let Some(head_block) = store.remove_cursor(&&store.chain)? {
-                let lower_bound = head_block.saturating_sub(ENV_VARS.reorg_threshold * 2);
+                let lower_bound = head_block.saturating_sub(ENV_VARS.reorg_threshold() * 2);
                 info!(&self.logger, "Removed cursor for non-firehose chain, now cleaning shallow blocks"; "network" => &store.chain, "lower_bound" => lower_bound);
                 store.cleanup_shallow_blocks(lower_bound)?;
             }

store/postgres/src/deployment.rs

@@ -547,7 +547,7 @@ pub fn revert_block_ptr(
     let number = format!("{}::numeric", ptr.number);
 
     // Block numbers can't be negative, so make it >= 0
-    let earliest_block = i32::max(ptr.number - ENV_VARS.reorg_threshold, 0);
+    let earliest_block = i32::max(ptr.number - ENV_VARS.reorg_threshold(), 0);
     let affected_rows = update(
         d::table
             .filter(d::deployment.eq(id.as_str()))

store/postgres/src/deployment_store.rs

@@ -1272,7 +1272,7 @@ impl DeploymentStore {
             let req = PruneRequest::new(
                 &site.as_ref().into(),
                 history_blocks,
-                ENV_VARS.reorg_threshold,
+                ENV_VARS.reorg_threshold(),
                 earliest_block,
                 latest_block,
             )?;

tests/src/fixture/ethereum.rs

@@ -64,7 +64,7 @@ pub async fn chain(
         triggers_adapter,
         Arc::new(NoopRuntimeAdapterBuilder {}),
         eth_adapters,
-        ENV_VARS.reorg_threshold,
+        ENV_VARS.reorg_threshold(),
         ENV_VARS.ingestor_polling_interval,
         // We assume the tested chain is always ingestible for now
         true,