tests: Add test for reverts using fail points

Author: leoyvens

Cargo.lock

@@ -15,6 +15,7 @@ serde = "1.0"
 config = { version = "0.10", features = ["toml"], default-features = false }
 dirs = "3.0"
 anyhow = "1.0"
+fail = "0.4"
 
 [dev-dependencies]
 diesel = { version = "1.4.5", features = ["postgres", "serde_json", "numeric", "r2d2"] }

chain/ethereum/Cargo.toml

@@ -11,6 +11,8 @@ use graph::prelude::{
     BlockStream as BlockStreamTrait, BlockStreamBuilder as BlockStreamBuilderTrait, *,
 };
 
+use fail::fail_point;
+
 lazy_static! {
     /// Maximum number of blocks to request in each chunk.
     static ref MAX_BLOCK_RANGE_SIZE: u64 = std::env::var("GRAPH_ETHEREUM_MAX_BLOCK_RANGE_SIZE")
@@ -444,6 +446,11 @@ where
             let subgraph_ptr =
                 subgraph_ptr.expect("subgraph block pointer should not be `None` here");
 
+            #[cfg(debug_assertions)]
+            if test_reorg(subgraph_ptr) {
+                return Box::new(future::ok(ReconciliationStep::Revert(subgraph_ptr)));
+            }
+
             // Precondition: subgraph_ptr.number < head_ptr.number
             // Walk back to one block short of subgraph_ptr.number
             let offset = head_ptr.number - subgraph_ptr.number - 1;
@@ -822,3 +829,25 @@ where
         )
     }
 }
+
+// This always returns `false` in a normal build. A test may configure reorg by enabling
+// "test_reorg" fail point with the number of the block that should be reorged.
+fn test_reorg(ptr: EthereumBlockPointer) -> bool {
+    fail_point!("test_reorg", |reorg_at| {
+        use std::str::FromStr;
+
+        static REORGED: std::sync::Once = std::sync::Once::new();
+
+        if REORGED.is_completed() {
+            return false;
+        }
+        let reorg_at = u64::from_str(&reorg_at.unwrap()).unwrap();
+        let should_reorg = ptr.number == reorg_at;
+        if should_reorg {
+            REORGED.call_once(|| {})
+        }
+        should_reorg
+    });
+
+    false
+}

chain/ethereum/src/block_stream.rs

@@ -18,6 +18,7 @@ semver = "0.10.0"
 serde = "1.0"
 serde_json = "1.0"
 serde_yaml = "0.8"
+fail = "0.4"
 
 [dev-dependencies]
 graph-mock = { path = "../mock" }

core/Cargo.toml

@@ -1,4 +1,5 @@
 use atomic_refcell::AtomicRefCell;
+use fail::fail_point;
 use futures01::sync::mpsc::{channel, Receiver, Sender};
 use lazy_static::lazy_static;
 use std::collections::{BTreeSet, HashMap, HashSet};
@@ -1055,16 +1056,19 @@ where
                 data_sources.push(data_source);
                 runtime_hosts.push(host);
             }
-            None => warn!(
-                logger,
-                "no runtime hosted created, there is already a runtime host instantiated for \
-                 this data source";
-                "name" => &data_source.name,
-                "address" => &data_source.source.address
-                    .map(|address| address.to_string())
-                    .unwrap_or("none".to_string()),
-                "abi" => &data_source.source.abi
-            ),
+            None => {
+                fail_point!("error_on_duplicate_ds", |_| Err(anyhow!("duplicate ds")));
+                warn!(
+                    logger,
+                    "no runtime hosted created, there is already a runtime host instantiated for \
+                     this data source";
+                    "name" => &data_source.name,
+                    "address" => &data_source.source.address
+                        .map(|address| address.to_string())
+                        .unwrap_or("none".to_string()),
+                    "abi" => &data_source.source.abi
+                )
+            }
         }
     }
 

core/src/subgraph/instance_manager.rs

@@ -42,6 +42,7 @@ structopt = "0.3.20"
 toml = "0.5.7"
 shellexpand = "2.0.0"
 diesel = "1.4.5"
+fail = "0.4"
 
 [dev-dependencies]
 assert_cli = "0.6"

node/Cargo.toml

@@ -101,6 +101,10 @@ fn read_expensive_queries() -> Result<Vec<Arc<q::Document>>, std::io::Error> {
 async fn main() {
     env_logger::init();
 
+    // Allow configuring fail points on debug builds. Used for integration tests.
+    #[cfg(debug_assertions)]
+    std::mem::forget(fail::FailScenario::setup());
+
     let opt = opt::Opt::from_args();
 
     // Set up logger

node/src/main.rs

@@ -269,7 +269,6 @@ fn graft() {
             TEST_SUBGRAPH_ID.as_str(),
             BLOCKS[1],
         );
-        dbg!(&res);
         assert!(res.is_ok());
 
         let query = EntityQuery::new(

store/postgres/tests/graft.rs

@@ -6,3 +6,5 @@ edition = "2018"
 [dev-dependencies]
 duct = "0.13"
 lazy_static = "1.4.0"
+fail = { version = "0.4", features = ["failpoints"] }
+defer = "0.1"

tests/Cargo.toml

@@ -0,0 +1 @@
+[{"inputs":[],"stateMutability":"nonpayable","type":"constructor"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"uint16","name":"x","type":"uint16"}],"name":"Trigger","type":"event"},{"inputs":[{"internalType":"uint16","name":"x","type":"uint16"}],"name":"emitTrigger","outputs":[],"stateMutability":"nonpayable","type":"function"}]