core: stop scanning reaching max endBlock

Author: shuaibbapputty

core/src/subgraph/inputs.rs

@@ -17,6 +17,7 @@ pub struct IndexingInputs<C: Blockchain> {
     pub start_blocks: Vec<BlockNumber>,
     pub end_blocks: BTreeSet<BlockNumber>,
     pub stop_block: Option<BlockNumber>,
+    pub max_end_block: Option<BlockNumber>,
     pub store: Arc<dyn WritableStore>,
     pub debug_fork: Option<Arc<dyn SubgraphFork>>,
     pub triggers_adapter: Arc<dyn TriggersAdapter<C>>,
@@ -40,6 +41,7 @@ impl<C: Blockchain> IndexingInputs<C> {
             start_blocks,
             end_blocks,
             stop_block,
+            max_end_block,
             store: _,
             debug_fork,
             triggers_adapter,
@@ -57,6 +59,7 @@ impl<C: Blockchain> IndexingInputs<C> {
             start_blocks: start_blocks.clone(),
             end_blocks: end_blocks.clone(),
             stop_block: stop_block.clone(),
+            max_end_block: max_end_block.clone(),
             store,
             debug_fork: debug_fork.clone(),
             triggers_adapter: triggers_adapter.clone(),

core/src/subgraph/instance_manager.rs

@@ -331,6 +331,13 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             })
             .collect();
 
+        let max_end_block: Option<BlockNumber> = if manifest.data_sources.len() == end_blocks.len()
+        {
+            end_blocks.iter().max().cloned()
+        } else {
+            None
+        };
+
         let templates = Arc::new(manifest.templates.clone());
 
         // Obtain the debug fork from the subgraph store
@@ -419,6 +426,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             start_blocks,
             end_blocks,
             stop_block,
+            max_end_block,
             store,
             debug_fork,
             triggers_adapter,

core/src/subgraph/runner.rs

@@ -197,6 +197,17 @@ where
                     .unfail_deterministic_error(&current_ptr, &parent_ptr)
                     .await?;
             }
+
+            // Stop subgraph when we reach maximum endblock.
+            if let Some(max_end_block) = self.inputs.max_end_block {
+                if max_end_block < current_ptr.block_number() {
+                    info!(self.logger, "Stopping subgraph as we reached maximum endBlock";
+                                "max_end_block" => max_end_block,
+                                "current_block" => current_ptr.block_number());
+                    self.inputs.store.flush().await?;
+                    return Ok(self);
+                }
+            }
         }
 
         loop {
@@ -987,44 +998,57 @@ where
         cancel_handle: &CancelHandle,
     ) -> Result<Action, Error> {
         let action = match event {
-            Some(Ok(BlockStreamEvent::ProcessWasmBlock(
-                block_ptr,
-                block_time,
-                data,
-                handler,
-                cursor,
-            ))) => {
-                let _section = self
-                    .metrics
-                    .stream
-                    .stopwatch
-                    .start_section(PROCESS_WASM_BLOCK_SECTION_NAME);
-                self.handle_process_wasm_block(
-                    block_ptr,
-                    block_time,
-                    data,
-                    handler,
-                    cursor,
-                    cancel_handle,
-                )
-                .await?
-            }
-            Some(Ok(BlockStreamEvent::ProcessBlock(block, cursor))) => {
-                let _section = self
-                    .metrics
-                    .stream
-                    .stopwatch
-                    .start_section(PROCESS_BLOCK_SECTION_NAME);
-                self.handle_process_block(block, cursor, cancel_handle)
-                    .await?
-            }
-            Some(Ok(BlockStreamEvent::Revert(revert_to_ptr, cursor))) => {
-                let _section = self
-                    .metrics
-                    .stream
-                    .stopwatch
-                    .start_section(HANDLE_REVERT_SECTION_NAME);
-                self.handle_revert(revert_to_ptr, cursor).await?
+            Some(Ok(event)) => {
+                if self.inputs.max_end_block.map_or(false, |max_end_block| {
+                    event.block_ptr().block_number() > max_end_block
+                }) {
+                    info!(self.logger, "Stopping subgraph as we reached maximum endBlock";
+                                "max_end_block" => self.inputs.max_end_block,
+                                "current_block" => event.block_ptr().block_number());
+                    return Ok(Action::Stop);
+                }
+
+                match event {
+                    BlockStreamEvent::ProcessWasmBlock(
+                        block_ptr,
+                        block_time,
+                        data,
+                        handler,
+                        cursor,
+                    ) => {
+                        let _section = self
+                            .metrics
+                            .stream
+                            .stopwatch
+                            .start_section(PROCESS_WASM_BLOCK_SECTION_NAME);
+                        self.handle_process_wasm_block(
+                            block_ptr,
+                            block_time,
+                            data,
+                            handler,
+                            cursor,
+                            cancel_handle,
+                        )
+                        .await?
+                    }
+                    BlockStreamEvent::ProcessBlock(block, cursor) => {
+                        let _section = self
+                            .metrics
+                            .stream
+                            .stopwatch
+                            .start_section(PROCESS_BLOCK_SECTION_NAME);
+                        self.handle_process_block(block, cursor, cancel_handle)
+                            .await?
+                    }
+                    BlockStreamEvent::Revert(revert_to_ptr, cursor) => {
+                        let _section = self
+                            .metrics
+                            .stream
+                            .stopwatch
+                            .start_section(HANDLE_REVERT_SECTION_NAME);
+                        self.handle_revert(revert_to_ptr, cursor).await?
+                    }
+                }
             }
             // Log and drop the errors from the block_stream
             // The block stream will continue attempting to produce blocks

graph/src/blockchain/block_stream.rs

@@ -541,6 +541,16 @@ pub enum BlockStreamEvent<C: Blockchain> {
     ProcessWasmBlock(BlockPtr, BlockTime, Box<[u8]>, String, FirehoseCursor),
 }
 
+impl<C: Blockchain> BlockStreamEvent<C> {
+    pub fn block_ptr(&self) -> BlockPtr {
+        match self {
+            BlockStreamEvent::Revert(ptr, _) => ptr.clone(),
+            BlockStreamEvent::ProcessBlock(block, _) => block.ptr(),
+            BlockStreamEvent::ProcessWasmBlock(ptr, _, _, _, _) => ptr.clone(),
+        }
+    }
+}
+
 impl<C: Blockchain> Clone for BlockStreamEvent<C>
 where
     C::TriggerData: Clone,