Feature/graphman test run

core/src/subgraph/instance_manager.rs

@@ -60,6 +60,7 @@ struct IndexingInputs<C: Blockchain> {
     deployment: DeploymentLocator,
     features: BTreeSet<SubgraphFeature>,
     start_blocks: Vec<BlockNumber>,
+    stop_block: Option<BlockNumber>,
     store: Arc<dyn WritableStore>,
     triggers_adapter: Arc<C::TriggersAdapter>,
     chain: Arc<C>,
@@ -192,6 +193,7 @@ where
         self: Arc<Self>,
         loc: DeploymentLocator,
         manifest: serde_yaml::Mapping,
+        stop_block: Option<BlockNumber>,
     ) {
         let logger = self.logger_factory.subgraph_logger(&loc);
         let err_logger = logger.clone();
@@ -201,13 +203,17 @@ where
             match BlockchainKind::from_manifest(&manifest)? {
                 BlockchainKind::Ethereum => {
                     instance_manager
-                        .start_subgraph_inner::<graph_chain_ethereum::Chain>(logger, loc, manifest)
+                        .start_subgraph_inner::<graph_chain_ethereum::Chain>(
+                            logger, loc, manifest, stop_block,
+                        )
                         .await
                 }
 
                 BlockchainKind::Near => {
                     instance_manager
-                        .start_subgraph_inner::<graph_chain_near::Chain>(logger, loc, manifest)
+                        .start_subgraph_inner::<graph_chain_near::Chain>(
+                            logger, loc, manifest, stop_block,
+                        )
                         .await
                 }
             }
@@ -274,6 +280,7 @@ where
         logger: Logger,
         deployment: DeploymentLocator,
         manifest: serde_yaml::Mapping,
+        stop_block: Option<BlockNumber>,
     ) -> Result<(), Error> {
         let subgraph_store = self.subgraph_store.cheap_clone();
         let registry = self.metrics_registry.cheap_clone();
@@ -412,6 +419,7 @@ where
                 deployment: deployment.clone(),
                 features,
                 start_blocks,
+                stop_block,
                 store,
                 triggers_adapter,
                 chain,
@@ -620,6 +628,16 @@ where
 
             let block_ptr = block.ptr();
 
+            match ctx.inputs.stop_block.clone() {
+                Some(stop_block) => {
+                    if block_ptr.number > stop_block {
+                        info!(&logger, "stop block reached for subgraph");
+                        return Ok(());
+                    }
+                }
+                _ => {}
+            }
+
             if block.trigger_count() > 0 {
                 subgraph_metrics
                     .block_trigger_count

core/src/subgraph/provider.rs

@@ -40,7 +40,11 @@ where
     L: LinkResolver,
     I: SubgraphInstanceManager,
 {
-    async fn start(&self, loc: DeploymentLocator) -> Result<(), SubgraphAssignmentProviderError> {
+    async fn start(
+        &self,
+        loc: DeploymentLocator,
+        stop_block: Option<BlockNumber>,
+    ) -> Result<(), SubgraphAssignmentProviderError> {
         let logger = self.logger_factory.subgraph_logger(&loc);
 
         // If subgraph ID already in set
@@ -63,7 +67,7 @@ where
 
         self.instance_manager
             .cheap_clone()
-            .start_subgraph(loc, raw)
+            .start_subgraph(loc, raw, stop_block)
             .await;
 
         Ok(())

core/src/subgraph/registrar.rs

@@ -409,7 +409,7 @@ async fn start_subgraph(
     trace!(logger, "Start subgraph");
 
     let start_time = Instant::now();
-    let result = provider.start(deployment.clone()).await;
+    let result = provider.start(deployment.clone(), None).await;
 
     debug!(
         logger,

graph/src/components/subgraph/instance_manager.rs

@@ -1,3 +1,4 @@
+use crate::prelude::BlockNumber;
 use std::sync::Arc;
 
 use crate::components::store::DeploymentLocator;
@@ -13,6 +14,7 @@ pub trait SubgraphInstanceManager: Send + Sync + 'static {
         self: Arc<Self>,
         deployment: DeploymentLocator,
         manifest: serde_yaml::Mapping,
+        stop_block: Option<BlockNumber>,
     );
     fn stop_subgraph(&self, deployment: DeploymentLocator);
 }

graph/src/components/subgraph/provider.rs

@@ -8,6 +8,7 @@ pub trait SubgraphAssignmentProvider: Send + Sync + 'static {
     async fn start(
         &self,
         deployment: DeploymentLocator,
+        stop_block: Option<BlockNumber>,
     ) -> Result<(), SubgraphAssignmentProviderError>;
     async fn stop(
         &self,

node/src/bin/manager.rs

@@ -144,6 +144,17 @@ pub enum Command {
         /// The deployments to rewind
         names: Vec<String>,
     },
+    /// Deploy and run an arbitrary subgraph, up to a certain block (for dev and testing purposes) -- WARNING: WILL RUN MIGRATIONS ON THE DB, DO NOT USE IN PRODUCTION
+    Run {
+        /// Network name (must fit one of the chain)
+        network_name: String,
+
+        /// Subgraph in the form `<IPFS Hash>` or `<name>:<IPFS Hash>`
+        subgraph: String,
+
+        /// Highest block number to process before stopping (inclusive)
+        stop_block: i32,
+    },
     /// Check and interrogate the configuration
     ///
     /// Print information about a configuration file without
@@ -374,6 +385,18 @@ impl Context {
         }
     }
 
+    fn metrics_registry(&self) -> Arc<MetricsRegistry> {
+        self.registry.clone()
+    }
+
+    fn config(&self) -> Cfg {
+        self.config.clone()
+    }
+
+    fn node_id(&self) -> NodeId {
+        self.node_id.clone()
+    }
+
     fn primary_pool(self) -> ConnectionPool {
         let primary = self.config.primary_store();
         let pool = StoreBuilder::main_pool(
@@ -411,6 +434,10 @@ impl Context {
         pools
     }
 
+    async fn store_builder(self) -> StoreBuilder {
+        StoreBuilder::new(&self.logger, &self.node_id, &self.config, self.registry).await
+    }
+
     fn store_and_pools(self) -> (Arc<Store>, HashMap<Shard, ConnectionPool>) {
         let (subgraph_store, pools) = StoreBuilder::make_subgraph_store_and_pools(
             &self.logger,
@@ -582,6 +609,29 @@ async fn main() {
                 sleep,
             )
         }
+        Run {
+            network_name,
+            subgraph,
+            stop_block,
+        } => {
+            let logger = ctx.logger.clone();
+            let config = ctx.config();
+            let registry = ctx.metrics_registry().clone();
+            let node_id = ctx.node_id().clone();
+            let store_builder = ctx.store_builder().await;
+
+            commands::run::run(
+                logger,
+                store_builder,
+                network_name,
+                config,
+                registry,
+                node_id,
+                subgraph,
+                stop_block,
+            )
+            .await
+        }
         Listen(cmd) => {
             use ListenCommand::*;
             match cmd {

node/src/manager/commands/mod.rs

@@ -8,6 +8,7 @@ pub mod listen;
 pub mod query;
 pub mod remove;
 pub mod rewind;
+pub mod run;
 pub mod stats;
 pub mod txn_speed;
 pub mod unused_deployments;