graph: Move subgraph.block_ptr() check (#3058)

Author: mangas

chain/ethereum/src/chain.rs

@@ -154,6 +154,8 @@ impl Chain {
             true => 0,
         };
 
+        let start_block = writable.block_ptr()?;
+
         Ok(Box::new(PollingBlockStream::new(
             writable,
             chain_store,
@@ -169,6 +171,7 @@ impl Chain {
             *MAX_BLOCK_RANGE_SIZE,
             *TARGET_TRIGGERS_PER_BLOCK_RANGE,
             unified_api_version,
+            start_block,
         )))
     }
 

graph/src/blockchain/polling_block_stream.rs

@@ -62,10 +62,10 @@ enum ReconciliationStep<C>
 where
     C: Blockchain,
 {
-    /// Revert the current block pointed at by the subgraph pointer. The pointer is to the current
+    /// Revert(from, to) the current block pointed at by the subgraph pointer. The pointer is to the current
     /// subgraph head, and a single block will be reverted so the new head will be the parent of the
-    /// current one.
-    Revert(BlockPtr),
+    /// current one. The second BlockPtr is the parent.
+    Revert(BlockPtr, BlockPtr),
 
     /// Move forwards, processing one or more blocks. Second element is the block range size.
     ProcessDescendantBlocks(Vec<BlockWithTriggers<C>>, BlockNumber),
@@ -101,6 +101,7 @@ where
     max_block_range_size: BlockNumber,
     target_triggers_per_block_range: u64,
     unified_api_version: UnifiedMappingApiVersion,
+    current_block: Option<BlockPtr>,
 }
 
 impl<C: Blockchain> Clone for PollingBlockStreamContext<C> {
@@ -121,6 +122,7 @@ impl<C: Blockchain> Clone for PollingBlockStreamContext<C> {
             max_block_range_size: self.max_block_range_size,
             target_triggers_per_block_range: self.target_triggers_per_block_range,
             unified_api_version: self.unified_api_version.clone(),
+            current_block: self.current_block.clone(),
         }
     }
 }
@@ -140,9 +142,9 @@ where
     /// Blocks and range size
     Blocks(VecDeque<BlockWithTriggers<C>>, BlockNumber),
 
-    // The payload is the current subgraph head pointer, which should be reverted, such that the
+    // The payload is the current subgraph head pointer, which should be reverted and it's parent, such that the
     // parent of the current subgraph head becomes the new subgraph head.
-    Revert(BlockPtr),
+    Revert(BlockPtr, BlockPtr),
     Done,
 }
 
@@ -165,12 +167,14 @@ where
         max_block_range_size: BlockNumber,
         target_triggers_per_block_range: u64,
         unified_api_version: UnifiedMappingApiVersion,
+        start_block: Option<BlockPtr>,
     ) -> Self {
         Self {
             state: BlockStreamState::BeginReconciliation,
             consecutive_err_count: 0,
             chain_head_update_stream,
             ctx: PollingBlockStreamContext {
+                current_block: start_block,
                 subgraph_store,
                 chain_store,
                 adapter,
@@ -216,7 +220,7 @@ where
 
                     return Ok(NextBlocks::Done);
                 }
-                ReconciliationStep::Revert(block) => return Ok(NextBlocks::Revert(block)),
+                ReconciliationStep::Revert(from, to) => return Ok(NextBlocks::Revert(from, to)),
             }
         }
     }
@@ -229,7 +233,7 @@ where
 
         // Get pointers from database for comparison
         let head_ptr_opt = ctx.chain_store.chain_head_ptr()?;
-        let subgraph_ptr = ctx.subgraph_store.block_ptr()?;
+        let subgraph_ptr = self.current_block.clone();
 
         // If chain head ptr is not set yet
         let head_ptr = match head_ptr_opt {
@@ -249,7 +253,7 @@ where
         trace!(
             ctx.logger, "Subgraph pointer";
             "hash" => format!("{:?}", subgraph_ptr.as_ref().map(|block| &block.hash)),
-            "number" => subgraph_ptr.as_ref().map(|block| block.number),
+            "number" => subgraph_ptr.as_ref().map(|block| &block.number),
         );
 
         // Make sure not to include genesis in the reorg threshold.
@@ -317,7 +321,10 @@ where
                 //
                 // Note: We can safely unwrap the subgraph ptr here, because
                 // if it was `None`, `is_on_main_chain` would be true.
-                return Ok(ReconciliationStep::Revert(subgraph_ptr.unwrap()));
+                let from = subgraph_ptr.unwrap();
+                let parent = self.parent_ptr(&from).await?;
+
+                return Ok(ReconciliationStep::Revert(from, parent));
             }
 
             // The subgraph ptr points to a block on the main chain.
@@ -422,7 +429,8 @@ where
 
             #[cfg(debug_assertions)]
             if test_reorg(subgraph_ptr.clone()) {
-                return Ok(ReconciliationStep::Revert(subgraph_ptr));
+                let parent = self.parent_ptr(&subgraph_ptr).await?;
+                return Ok(ReconciliationStep::Revert(subgraph_ptr.clone(), parent));
             }
 
             // Precondition: subgraph_ptr.number < head_ptr.number
@@ -457,16 +465,28 @@ where
                             .await?;
                         Ok(ReconciliationStep::ProcessDescendantBlocks(vec![block], 1))
                     } else {
+                        let parent = self.parent_ptr(&subgraph_ptr).await?;
+
                         // The subgraph ptr is not on the main chain.
                         // We will need to step back (possibly repeatedly) one block at a time
                         // until we are back on the main chain.
-                        Ok(ReconciliationStep::Revert(subgraph_ptr))
+                        Ok(ReconciliationStep::Revert(subgraph_ptr, parent))
                     }
                 }
             }
         }
     }
 
+    async fn parent_ptr(&self, block_ptr: &BlockPtr) -> Result<BlockPtr, Error> {
+        let ptr = self
+            .adapter
+            .parent_ptr(block_ptr)
+            .await?
+            .expect("genesis block can't be reverted");
+
+        Ok(ptr)
+    }
+
     /// Set subgraph deployment entity synced flag if and only if the subgraph block pointer is
     /// caught up to the head block pointer.
     fn update_subgraph_synced_status(&self) -> Result<(), StoreError> {
@@ -505,54 +525,60 @@ impl<C: Blockchain> Stream for PollingBlockStream<C> {
                 // Waiting for the reconciliation to complete or yield blocks
                 BlockStreamState::Reconciliation(next_blocks_future) => {
                     match next_blocks_future.poll_unpin(cx) {
-                        Poll::Ready(Ok(NextBlocks::Blocks(next_blocks, block_range_size))) => {
-                            // We had only one error, so we infer that reducing the range size is
-                            // what fixed it. Reduce the max range size to prevent future errors.
-                            // See: 018c6df4-132f-4acc-8697-a2d64e83a9f0
-                            if self.consecutive_err_count == 1 {
-                                // Reduce the max range size by 10%, but to no less than 10.
-                                self.ctx.max_block_range_size =
-                                    (self.ctx.max_block_range_size * 9 / 10).max(10);
+                        Poll::Ready(Ok(next_block_step)) => match next_block_step {
+                            NextBlocks::Blocks(next_blocks, block_range_size) => {
+                                // We had only one error, so we infer that reducing the range size is
+                                // what fixed it. Reduce the max range size to prevent future errors.
+                                // See: 018c6df4-132f-4acc-8697-a2d64e83a9f0
+                                if self.consecutive_err_count == 1 {
+                                    // Reduce the max range size by 10%, but to no less than 10.
+                                    self.ctx.max_block_range_size =
+                                        (self.ctx.max_block_range_size * 9 / 10).max(10);
+                                }
+                                self.consecutive_err_count = 0;
+
+                                let total_triggers =
+                                    next_blocks.iter().map(|b| b.trigger_count()).sum::<usize>();
+                                self.ctx.previous_triggers_per_block =
+                                    total_triggers as f64 / block_range_size as f64;
+                                self.ctx.previous_block_range_size = block_range_size;
+                                if total_triggers > 0 {
+                                    debug!(
+                                        self.ctx.logger,
+                                        "Processing {} triggers", total_triggers
+                                    );
+                                }
+
+                                // Switch to yielding state until next_blocks is depleted
+                                self.state =
+                                    BlockStreamState::YieldingBlocks(Box::new(next_blocks));
+
+                                // Yield the first block in next_blocks
+                                continue;
                             }
-                            self.consecutive_err_count = 0;
-
-                            let total_triggers =
-                                next_blocks.iter().map(|b| b.trigger_count()).sum::<usize>();
-                            self.ctx.previous_triggers_per_block =
-                                total_triggers as f64 / block_range_size as f64;
-                            self.ctx.previous_block_range_size = block_range_size;
-                            if total_triggers > 0 {
-                                debug!(self.ctx.logger, "Processing {} triggers", total_triggers);
-                            }
-
-                            // Switch to yielding state until next_blocks is depleted
-                            self.state = BlockStreamState::YieldingBlocks(Box::new(next_blocks));
-
-                            // Yield the first block in next_blocks
-                            continue;
-                        }
-                        // Reconciliation completed. We're caught up to chain head.
-                        Poll::Ready(Ok(NextBlocks::Done)) => {
-                            // Reset error count
-                            self.consecutive_err_count = 0;
+                            // Reconciliation completed. We're caught up to chain head.
+                            NextBlocks::Done => {
+                                // Reset error count
+                                self.consecutive_err_count = 0;
 
-                            // Switch to idle
-                            self.state = BlockStreamState::Idle;
+                                // Switch to idle
+                                self.state = BlockStreamState::Idle;
 
-                            // Poll for chain head update
-                            continue;
-                        }
-                        Poll::Ready(Ok(NextBlocks::Revert(block))) => {
-                            self.state = BlockStreamState::BeginReconciliation;
-                            break Poll::Ready(Some(Ok(BlockStreamEvent::Revert(
-                                block,
-                                FirehoseCursor::None,
-                                None,
-                            ))));
-                        }
-                        Poll::Pending => {
-                            break Poll::Pending;
-                        }
+                                // Poll for chain head update
+                                continue;
+                            }
+                            NextBlocks::Revert(from, to) => {
+                                self.ctx.current_block = to.into();
+
+                                self.state = BlockStreamState::BeginReconciliation;
+                                break Poll::Ready(Some(Ok(BlockStreamEvent::Revert(
+                                    from,
+                                    FirehoseCursor::None,
+                                    self.ctx.current_block.clone(),
+                                ))));
+                            }
+                        },
+                        Poll::Pending => break Poll::Pending,
                         Poll::Ready(Err(e)) => {
                             // Reset the block range size in an attempt to recover from the error.
                             // See also: 018c6df4-132f-4acc-8697-a2d64e83a9f0
@@ -577,6 +603,8 @@ impl<C: Blockchain> Stream for PollingBlockStream<C> {
                     match next_blocks.pop_front() {
                         // Yield one block
                         Some(next_block) => {
+                            self.ctx.current_block = Some(next_block.block.ptr());
+
                             break Poll::Ready(Some(Ok(BlockStreamEvent::ProcessBlock(
                                 next_block,
                                 FirehoseCursor::None,