Initial addition of FirehoseBlockIngestor for NEAR (and other Firehose chains) (#3041)

* Initial addition of FirehoseBlockIngestor for NEAR (and other Firehose chains)

The actual FirehoseBlockIngestor is generic enough to be re-usable on all chains that have Firehose support. This is the initial implementation that just follow chain head and update the head blocks. Rather naive in its approach right now.

Current limitations that I would like to discuss:
- Chain net_version and genesis_hash already exist with a default value. Now the value is fetched from Firehose correctly so there is a mismatch. How we should handle that, just delete the prior record "manually" before deploying? Automatically override if current value is XYZ? I would opt for the latter.
- Still need to discuss how we want to handler backfilling. There is different options available: backfill straight inside graph-node with a different cursor, backfill with graph-man in parallel fashion, backfill on demand when requestion block number to block hash
- Currently doing 3 calls to database: upsert_block, attemp_chain_update and set_head_cursor, I would like to refactor that inside 1 efficient transaction instead (if possible, wanted to discuss that first before doing it).

Also optimized memory usage a bit by only unpacking via `HeaderOnlyBlock` structure.

Author: Matthieu Vachon

chain/ethereum/src/codec.rs

@@ -1,12 +1,16 @@
 #[path = "protobuf/dfuse.ethereum.codec.v1.rs"]
 mod pbcodec;
 
-use graph::prelude::{
-    web3,
-    web3::types::TransactionReceipt as w3TransactionReceipt,
-    web3::types::{Bytes, H160, H2048, H256, H64, U256, U64},
-    EthereumBlock, EthereumBlockWithCalls, EthereumCall, LightEthereumBlock,
+use graph::{
+    blockchain::{Block as BlockchainBlock, BlockPtr},
+    prelude::{
+        web3,
+        web3::types::TransactionReceipt as w3TransactionReceipt,
+        web3::types::{Bytes, H160, H2048, H256, H64, U256, U64},
+        BlockNumber, EthereumBlock, EthereumBlockWithCalls, EthereumCall, LightEthereumBlock,
+    },
 };
+use std::convert::TryFrom;
 use std::sync::Arc;
 
 use crate::chain::BlockFinality;
@@ -295,3 +299,37 @@ impl Into<EthereumBlockWithCalls> for &Block {
         }
     }
 }
+
+impl From<Block> for BlockPtr {
+    fn from(b: Block) -> BlockPtr {
+        (&b).into()
+    }
+}
+
+impl<'a> From<&'a Block> for BlockPtr {
+    fn from(b: &'a Block) -> BlockPtr {
+        BlockPtr::from((H256::from_slice(b.hash.as_ref()), b.number))
+    }
+}
+
+impl BlockchainBlock for Block {
+    fn number(&self) -> i32 {
+        BlockNumber::try_from(self.number).unwrap()
+    }
+
+    fn ptr(&self) -> BlockPtr {
+        self.into()
+    }
+
+    fn parent_ptr(&self) -> Option<BlockPtr> {
+        let parent_hash = &self.header.as_ref().unwrap().parent_hash;
+
+        match parent_hash.len() {
+            0 => None,
+            _ => Some(BlockPtr::from((
+                H256::from_slice(parent_hash.as_ref()),
+                self.number - 1,
+            ))),
+        }
+    }
+}

chain/near/proto/codec.proto

@@ -12,6 +12,16 @@ message Block {
   repeated StateChangeWithCause state_changes = 5;
 }
 
+// HeaderOnlyBlock is a standard [Block] structure where all other fields are
+// removed so that hydrating that object from a [Block] bytes payload will
+// drastically reduced allocated memory required to hold the full block.
+//
+// This can be used to unpack a [Block] when only the [BlockHeader] information
+// is required and greatly reduced required memory.
+message HeaderOnlyBlock {
+  BlockHeader header = 2;
+}
+
 message StateChangeWithCause {
   StateChangeValue value = 1;
   StateChangeCause cause = 2;

chain/near/src/codec.rs

@@ -2,7 +2,7 @@
 mod pbcodec;
 
 use graph::{
-    blockchain::Block as Blockchainblock,
+    blockchain::Block as BlockchainBlock,
     blockchain::BlockPtr,
     prelude::{hex, web3::types::H256, BlockNumber},
 };
@@ -23,18 +23,28 @@ impl LowerHex for &CryptoHash {
     }
 }
 
+impl BlockHeader {
+    pub fn parent_ptr(&self) -> Option<BlockPtr> {
+        match (self.prev_hash.as_ref(), self.prev_height) {
+            (Some(hash), number) => Some(BlockPtr::from((hash.into(), number))),
+            _ => None,
+        }
+    }
+}
+
+impl<'a> From<&'a BlockHeader> for BlockPtr {
+    fn from(b: &'a BlockHeader) -> BlockPtr {
+        BlockPtr::from((b.hash.as_ref().unwrap().into(), b.height))
+    }
+}
+
 impl Block {
     pub fn header(&self) -> &BlockHeader {
         self.header.as_ref().unwrap()
     }
 
     pub fn parent_ptr(&self) -> Option<BlockPtr> {
-        let header = self.header();
-
-        match (header.prev_hash.as_ref(), header.prev_height) {
-            (Some(hash), number) => Some(BlockPtr::from((hash.into(), number))),
-            _ => None,
-        }
+        self.header().parent_ptr()
     }
 }
 
@@ -46,14 +56,11 @@ impl From<Block> for BlockPtr {
 
 impl<'a> From<&'a Block> for BlockPtr {
     fn from(b: &'a Block) -> BlockPtr {
-        let header = b.header();
-        let hash: H256 = header.hash.as_ref().unwrap().into();
-
-        BlockPtr::from((hash, header.height))
+        BlockPtr::from(b.header())
     }
 }
 
-impl Blockchainblock for Block {
+impl BlockchainBlock for Block {
     fn number(&self) -> i32 {
         BlockNumber::try_from(self.header().height).unwrap()
     }
@@ -67,6 +74,32 @@ impl Blockchainblock for Block {
     }
 }
 
+impl HeaderOnlyBlock {
+    pub fn header(&self) -> &BlockHeader {
+        self.header.as_ref().unwrap()
+    }
+}
+
+impl<'a> From<&'a HeaderOnlyBlock> for BlockPtr {
+    fn from(b: &'a HeaderOnlyBlock) -> BlockPtr {
+        BlockPtr::from(b.header())
+    }
+}
+
+impl BlockchainBlock for HeaderOnlyBlock {
+    fn number(&self) -> i32 {
+        BlockNumber::try_from(self.header().height).unwrap()
+    }
+
+    fn ptr(&self) -> BlockPtr {
+        self.into()
+    }
+
+    fn parent_ptr(&self) -> Option<BlockPtr> {
+        self.header().parent_ptr()
+    }
+}
+
 impl execution_outcome::Status {
     pub fn is_success(&self) -> bool {
         use execution_outcome::Status::*;

chain/near/src/lib.rs

@@ -7,3 +7,5 @@ mod runtime;
 mod trigger;
 
 pub use crate::chain::Chain;
+pub use codec::Block;
+pub use codec::HeaderOnlyBlock;

chain/near/src/protobuf/sf.near.codec.v1.rs

@@ -11,6 +11,17 @@ pub struct Block {
     #[prost(message, repeated, tag = "5")]
     pub state_changes: ::prost::alloc::vec::Vec<StateChangeWithCause>,
 }
+/// HeaderOnlyBlock is a standard [Block] structure where all other fields are
+/// removed so that hydrating that object from a [Block] bytes payload will
+/// drastically reduced allocated memory required to hold the full block.
+///
+/// This can be used to unpack a [Block] when only the [BlockHeader] information
+/// is required and greatly reduced required memory.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct HeaderOnlyBlock {
+    #[prost(message, optional, tag = "2")]
+    pub header: ::core::option::Option<BlockHeader>,
+}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct StateChangeWithCause {
     #[prost(message, optional, tag = "1")]

graph/src/blockchain/firehose_block_ingestor.rs

@@ -0,0 +1,166 @@
+use std::{marker::PhantomData, sync::Arc, time::Duration};
+
+use crate::{
+    blockchain::Block as BlockchainBlock,
+    components::store::ChainStore,
+    firehose::{bstream, decode_firehose_block, endpoints::FirehoseEndpoint},
+    prelude::{error, info, Logger},
+    util::backoff::ExponentialBackoff,
+};
+use anyhow::{Context, Error};
+use futures03::StreamExt;
+use slog::trace;
+use tonic::Streaming;
+
+pub struct FirehoseBlockIngestor<M>
+where
+    M: prost::Message + BlockchainBlock + Default + 'static,
+{
+    chain_store: Arc<dyn ChainStore>,
+    endpoint: Arc<FirehoseEndpoint>,
+    logger: Logger,
+
+    phantom: PhantomData<M>,
+}
+
+impl<M> FirehoseBlockIngestor<M>
+where
+    M: prost::Message + BlockchainBlock + Default + 'static,
+{
+    pub fn new(
+        chain_store: Arc<dyn ChainStore>,
+        endpoint: Arc<FirehoseEndpoint>,
+        logger: Logger,
+    ) -> FirehoseBlockIngestor<M> {
+        FirehoseBlockIngestor {
+            chain_store,
+            endpoint,
+            logger,
+            phantom: PhantomData {},
+        }
+    }
+
+    pub async fn run(self) {
+        let mut latest_cursor = self.fetch_head_cursor().await;
+        let mut backoff =
+            ExponentialBackoff::new(Duration::from_millis(250), Duration::from_secs(30));
+
+        loop {
+            info!(
+                self.logger,
+                "Blockstream disconnected, connecting"; "endpoint uri" => format_args!("{}", self.endpoint), "cursor" => format_args!("{}", latest_cursor),
+            );
+
+            let result = self
+                .endpoint
+                .clone()
+                .stream_blocks(bstream::BlocksRequestV2 {
+                    // Starts at current HEAD block of the chain (viewed from Firehose side)
+                    start_block_num: -1,
+                    start_cursor: latest_cursor.clone(),
+                    fork_steps: vec![
+                        bstream::ForkStep::StepNew as i32,
+                        bstream::ForkStep::StepUndo as i32,
+                    ],
+                    ..Default::default()
+                })
+                .await;
+
+            match result {
+                Ok(stream) => {
+                    // Consume the stream of blocks until an error is hit
+                    latest_cursor = self.process_blocks(latest_cursor, stream).await
+                }
+                Err(e) => {
+                    error!(self.logger, "Unable to connect to endpoint: {:?}", e);
+                }
+            }
+
+            // If we reach this point, we must wait a bit before retrying
+            backoff.sleep_async().await;
+        }
+    }
+
+    async fn fetch_head_cursor(&self) -> String {
+        let mut backoff =
+            ExponentialBackoff::new(Duration::from_millis(250), Duration::from_secs(30));
+        loop {
+            match self.chain_store.clone().chain_head_cursor() {
+                Ok(cursor) => return cursor.unwrap_or_else(|| "".to_string()),
+                Err(e) => {
+                    error!(self.logger, "Fetching chain head cursor failed: {:?}", e);
+
+                    backoff.sleep_async().await;
+                }
+            }
+        }
+    }
+
+    /// Consumes the incoming stream of blocks infinitely until it hits an error. In which case
+    /// the error is logged right away and the latest available cursor is returned
+    /// upstream for future consumption.
+    async fn process_blocks(
+        &self,
+        cursor: String,
+        mut stream: Streaming<bstream::BlockResponseV2>,
+    ) -> String {
+        use bstream::ForkStep::*;
+
+        let mut latest_cursor = cursor;
+
+        while let Some(message) = stream.next().await {
+            match message {
+                Ok(v) => {
+                    let step = bstream::ForkStep::from_i32(v.step)
+                        .expect("Fork step should always match to known value");
+
+                    let result = match step {
+                        StepNew => self.process_new_block(&v).await,
+                        StepUndo => {
+                            trace!(self.logger, "Received undo block to ingest, skipping");
+                            Ok(())
+                        }
+                        StepIrreversible | StepUnknown => panic!(
+                            "We explicitly requested StepNew|StepUndo but received something else"
+                        ),
+                    };
+
+                    if let Err(e) = result {
+                        error!(self.logger, "Process block failed: {:?}", e);
+                        break;
+                    }
+
+                    latest_cursor = v.cursor;
+                }
+                Err(e) => {
+                    info!(
+                        self.logger,
+                        "An error occurred while streaming blocks: {}", e
+                    );
+                    break;
+                }
+            }
+        }
+
+        error!(
+            self.logger,
+            "Stream blocks complete unexpectedly, expecting stream to always stream blocks"
+        );
+        latest_cursor
+    }
+
+    async fn process_new_block(&self, response: &bstream::BlockResponseV2) -> Result<(), Error> {
+        let block = decode_firehose_block::<M>(response)
+            .context("Mapping firehose block to blockchain::Block")?;
+
+        trace!(self.logger, "Received new block to ingest {}", block.ptr());
+
+        self.chain_store
+            .clone()
+            .set_chain_head(block, response.cursor.clone())
+            .await
+            .context("Updating chain head")?;
+
+        Ok(())
+    }
+}

graph/src/blockchain/mod.rs

@@ -4,6 +4,7 @@
 
 pub mod block_ingestor;
 pub mod block_stream;
+pub mod firehose_block_ingestor;
 pub mod firehose_block_stream;
 pub mod polling_block_stream;
 mod types;

graph/src/components/store.rs

@@ -1324,6 +1324,21 @@ pub trait ChainStore: Send + Sync + 'static {
     /// The head block pointer will be None on initial set up.
     fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error>;
 
+    /// Get the current head block cursor for this chain.
+    ///
+    /// The head block cursor will be None on initial set up.
+    fn chain_head_cursor(&self) -> Result<Option<String>, Error>;
+
+    /// This method does actually three operations:
+    /// - Upserts received block into blocks table
+    /// - Update chain head block into networks table
+    /// - Update chain head cursor into networks table
+    async fn set_chain_head(
+        self: Arc<Self>,
+        block: Arc<dyn Block>,
+        cursor: String,
+    ) -> Result<(), Error>;
+
     /// Returns the blocks present in the store.
     fn blocks(&self, hashes: &[H256]) -> Result<Vec<serde_json::Value>, Error>;