fix firehose UNDO case on old blocks

When firehose sends an undo event from an older block, the
instance-manager would try to fetch the block from RPC to
retrieve its parent block info.

Now, the firehose will include the parent's hash and num in
the Revert event to remove that extraneous RPC call

Author: sduchesneau

chain/ethereum/src/chain.rs

@@ -597,6 +597,13 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
                     number: block.number as i32,
                 },
                 FirehoseCursor::Some(response.cursor.clone()),
+                match block.header {
+                    Some(header) => Some(BlockPtr {
+                        hash: BlockHash::from(header.parent_hash),
+                        number: (block.number as i32) - 1,
+                    }),
+                    None => None,
+                },
             )),
 
             bstream::ForkStep::StepIrreversible => {

chain/near/src/chain.rs

@@ -285,6 +285,7 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
                         number: header.height as i32,
                     },
                     Some(response.cursor.clone()),
+                    None, // FIXME: we should get the parent block pointer when we have access to parent block height
                 ))
             }
 

core/src/subgraph/instance_manager.rs

@@ -493,7 +493,7 @@ where
         loop {
             let (block, cursor) = match block_stream.next().await {
                 Some(Ok(BlockStreamEvent::ProcessBlock(block, cursor))) => (block, cursor),
-                Some(Ok(BlockStreamEvent::Revert(subgraph_ptr, _))) => {
+                Some(Ok(BlockStreamEvent::Revert(subgraph_ptr, _, optional_parent_ptr))) => {
                     info!(
                         logger,
                         "Reverting block to get back to main chain";
@@ -502,31 +502,49 @@ where
                     );
 
                     // We would like to revert the DB state to the parent of the current block.
-                    // First, load the block in order to get the parent hash.
-                    if let Err(e) = ctx
-                        .inputs
-                        .triggers_adapter
-                        .parent_ptr(&subgraph_ptr)
-                        .await
-                        .map(|parent_ptr| parent_ptr.expect("genesis block cannot be reverted"))
-                        .and_then(|parent_ptr| {
-                            // Revert entity changes from this block, and update subgraph ptr.
-                            ctx.inputs
-                                .store
-                                .revert_block_operations(parent_ptr)
-                                .map_err(Into::into)
-                        })
-                    {
-                        error!(
-                            &logger,
-                            "Could not revert block. \
-                            The likely cause is the block not being found due to a deep reorg. \
-                            Retrying";
-                            "block_number" => format!("{}", subgraph_ptr.number),
-                            "block_hash" => format!("{}", subgraph_ptr.hash),
-                            "error" => e.to_string(),
-                        );
-                        continue;
+                    match optional_parent_ptr {
+                        Some(parent_ptr) => {
+                            if let Err(e) = ctx.inputs.store.revert_block_operations(parent_ptr) {
+                                error!(
+                                    &logger,
+                                    "Could not revert block. Retrying";
+                                    "block_number" => format!("{}", subgraph_ptr.number),
+                                    "block_hash" => format!("{}", subgraph_ptr.hash),
+                                    "error" => e.to_string(),
+                                );
+                                continue;
+                            }
+                        }
+                        None => {
+                            // First, load the block in order to get the parent hash.
+                            if let Err(e) = ctx
+                                .inputs
+                                .triggers_adapter
+                                .parent_ptr(&subgraph_ptr)
+                                .await
+                                .map(|parent_ptr| {
+                                    parent_ptr.expect("genesis block cannot be reverted")
+                                })
+                                .and_then(|parent_ptr| {
+                                    // Revert entity changes from this block, and update subgraph ptr.
+                                    ctx.inputs
+                                        .store
+                                        .revert_block_operations(parent_ptr)
+                                        .map_err(Into::into)
+                                })
+                            {
+                                error!(
+                                    &logger,
+                                    "Could not revert block. \
+                                    The likely cause is the block not being found due to a deep reorg. \
+                                    Retrying";
+                                    "block_number" => format!("{}", subgraph_ptr.number),
+                                    "block_hash" => format!("{}", subgraph_ptr.hash),
+                                    "error" => e.to_string(),
+                                );
+                                continue;
+                            }
+                        }
                     }
 
                     ctx.block_stream_metrics

graph/src/blockchain/block_stream.rs

@@ -102,7 +102,8 @@ pub enum FirehoseError {
 pub enum BlockStreamEvent<C: Blockchain> {
     // The payload is the current subgraph head pointer, which should be reverted, such that the
     // parent of the current subgraph head becomes the new subgraph head.
-    Revert(BlockPtr, FirehoseCursor),
+    // An optional pointer to the parent block will save a round trip operation when reverting.
+    Revert(BlockPtr, FirehoseCursor, Option<BlockPtr>),
 
     ProcessBlock(BlockWithTriggers<C>, FirehoseCursor),
 }

graph/src/blockchain/polling_block_stream.rs

@@ -546,6 +546,7 @@ impl<C: Blockchain> Stream for PollingBlockStream<C> {
                             break Poll::Ready(Some(Ok(BlockStreamEvent::Revert(
                                 block,
                                 FirehoseCursor::None,
+                                None,
                             ))));
                         }
                         Poll::Pending => {