Renamed of the Firehose service gRPC ids and packages (#3054)

This will require a coordinate update for NEAR when Firehose service ids will change there, `graph-node` with this version will need to be deployed.

Author: Matthieu Vachon

chain/ethereum/examples/firehose.rs

@@ -1,11 +1,8 @@
 use anyhow::Error;
 use graph::{
-    firehose::{
-        bstream::BlockResponseV2, bstream::BlocksRequestV2, bstream::ForkStep,
-        endpoints::FirehoseEndpoint,
-    },
     log::logger,
     prelude::{prost, tokio, tonic},
+    {firehose, firehose::FirehoseEndpoint, firehose::ForkStep},
 };
 use graph_chain_ethereum::codec;
 use prost::Message;
@@ -23,9 +20,9 @@ async fn main() -> Result<(), Error> {
 
     loop {
         println!("connecting to the stream!");
-        let mut stream: Streaming<BlockResponseV2> = match firehose
+        let mut stream: Streaming<firehose::Response> = match firehose
             .clone()
-            .stream_blocks(BlocksRequestV2 {
+            .stream_blocks(firehose::Request {
                 start_block_num: 7000000,
                 start_cursor: match &cursor {
                     Some(c) => c.clone(),

chain/ethereum/proto/codec.proto

@@ -1,6 +1,8 @@
 syntax = "proto3";
 
-package dfuse.ethereum.codec.v1;
+package sf.ethereum.codec.v1;
+
+option go_package = "github.com/streamingfast/sf-ethereum/pb/sf/ethereum/codec/v1;pbcodec";
 
 import "google/protobuf/timestamp.proto";
 

chain/ethereum/src/chain.rs

@@ -1,7 +1,7 @@
 use anyhow::{Context, Error};
 use graph::blockchain::BlockchainKind;
 use graph::data::subgraph::UnifiedMappingApiVersion;
-use graph::firehose::endpoints::FirehoseNetworkEndpoints;
+use graph::firehose::FirehoseNetworkEndpoints;
 use graph::prelude::{
     EthereumBlock, EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, StopwatchMetrics,
 };
@@ -19,7 +19,7 @@ use graph::{
     },
     cheap_clone::CheapClone,
     components::store::DeploymentLocator,
-    firehose::bstream,
+    firehose,
     log::factory::{ComponentLoggerConfig, ElasticComponentLoggerConfig},
     prelude::{
         async_trait, error, lazy_static, o, serde_json as json, web3::types::H256, BlockNumber,
@@ -562,11 +562,13 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
     async fn to_block_stream_event(
         &self,
         logger: &Logger,
-        response: &bstream::BlockResponseV2,
+        response: &firehose::Response,
         adapter: &TriggersAdapter,
         filter: &TriggerFilter,
     ) -> Result<BlockStreamEvent<Chain>, FirehoseError> {
-        let step = bstream::ForkStep::from_i32(response.step).unwrap_or_else(|| {
+        use firehose::ForkStep;
+
+        let step = ForkStep::from_i32(response.step).unwrap_or_else(|| {
             panic!(
                 "unknown step i32 value {}, maybe you forgot update & re-regenerate the protobuf definitions?",
                 response.step
@@ -579,15 +581,16 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 
         // Right now, this is done in all cases but in reality, with how the BlockStreamEvent::Revert
         // is defined right now, only block hash and block number is necessary. However, this information
-        // is not part of the actual bstream::BlockResponseV2 payload. As such, we need to decode the full
+        // is not part of the actual firehose::Response payload. As such, we need to decode the full
         // block which is useless.
         //
-        // Check about adding basic information about the block in the bstream::BlockResponseV2 or maybe
+        // Check about adding basic information about the block in the firehose::Response or maybe
         // define a slimmed down stuct that would decode only a few fields and ignore all the rest.
         let block = codec::Block::decode(any_block.value.as_ref())?;
 
+        use firehose::ForkStep::*;
         match step {
-            bstream::ForkStep::StepNew => {
+            StepNew => {
                 let ethereum_block: EthereumBlockWithCalls = (&block).into();
                 let block_with_triggers = adapter
                     .triggers_in_block(logger, BlockFinality::NonFinal(ethereum_block), filter)
@@ -599,7 +602,7 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
                 ))
             }
 
-            bstream::ForkStep::StepUndo => Ok(BlockStreamEvent::Revert(
+            StepUndo => Ok(BlockStreamEvent::Revert(
                 BlockPtr {
                     hash: BlockHash::from(block.hash),
                     number: block.number as i32,
@@ -611,11 +614,11 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
                 }),
             )),
 
-            bstream::ForkStep::StepIrreversible => {
+            StepIrreversible => {
                 unreachable!("irreversible step is not handled and should not be requested in the Firehose request")
             }
 
-            bstream::ForkStep::StepUnknown => {
+            StepUnknown => {
                 unreachable!("unknown step should not happen in the Firehose response")
             }
         }

chain/ethereum/src/codec.rs

@@ -1,4 +1,4 @@
-#[path = "protobuf/dfuse.ethereum.codec.v1.rs"]
+#[path = "protobuf/sf.ethereum.codec.v1.rs"]
 mod pbcodec;
 
 use graph::{

chain/ethereum/src/protobuf/dfuse.ethereum.codec.v1.rs renamed to chain/ethereum/src/protobuf/sf.ethereum.codec.v1.rs

@@ -1,7 +1,7 @@
 use graph::blockchain::BlockchainKind;
 use graph::cheap_clone::CheapClone;
 use graph::data::subgraph::UnifiedMappingApiVersion;
-use graph::firehose::endpoints::FirehoseNetworkEndpoints;
+use graph::firehose::FirehoseNetworkEndpoints;
 use graph::prelude::StopwatchMetrics;
 use graph::{
     anyhow,
@@ -14,7 +14,7 @@ use graph::{
         BlockHash, BlockPtr, Blockchain, IngestorAdapter as IngestorAdapterTrait, IngestorError,
     },
     components::store::DeploymentLocator,
-    firehose::bstream,
+    firehose::{self as firehose, ForkStep},
     log::factory::{ComponentLoggerConfig, ElasticComponentLoggerConfig},
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
@@ -288,11 +288,11 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
     async fn to_block_stream_event(
         &self,
         logger: &Logger,
-        response: &bstream::BlockResponseV2,
+        response: &firehose::Response,
         adapter: &TriggersAdapter,
         filter: &TriggerFilter,
     ) -> Result<BlockStreamEvent<Chain>, FirehoseError> {
-        let step = bstream::ForkStep::from_i32(response.step).unwrap_or_else(|| {
+        let step = ForkStep::from_i32(response.step).unwrap_or_else(|| {
             panic!(
                 "unknown step i32 value {}, maybe you forgot update & re-regenerate the protobuf definitions?",
                 response.step
@@ -312,13 +312,14 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
         // define a slimmed down stuct that would decode only a few fields and ignore all the rest.
         let block = codec::Block::decode(any_block.value.as_ref())?;
 
+        use ForkStep::*;
         match step {
-            bstream::ForkStep::StepNew => Ok(BlockStreamEvent::ProcessBlock(
+            StepNew => Ok(BlockStreamEvent::ProcessBlock(
                 adapter.triggers_in_block(logger, block, filter).await?,
                 Some(response.cursor.clone()),
             )),
 
-            bstream::ForkStep::StepUndo => {
+            StepUndo => {
                 let header = block.header();
 
                 Ok(BlockStreamEvent::Revert(
@@ -331,11 +332,11 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
                 ))
             }
 
-            bstream::ForkStep::StepIrreversible => {
+            StepIrreversible => {
                 panic!("irreversible step is not handled and should not be requested in the Firehose request")
             }
 
-            bstream::ForkStep::StepUnknown => {
+            StepUnknown => {
                 panic!("unknown step should not happen in the Firehose response")
             }
         }

chain/near/src/chain.rs

@@ -3,6 +3,6 @@ fn main() {
     tonic_build::configure()
         .out_dir("src/firehose")
         .format(true)
-        .compile(&["proto/bstream.proto"], &["proto"])
+        .compile(&["proto/firehose.proto"], &["proto"])
         .expect("Failed to compile Firehose proto(s)");
 }

graph/build.rs

@@ -1,51 +1,17 @@
 syntax = "proto3";
 
-package dfuse.bstream.v1;
+package sf.firehose.v1;
 
-import "google/protobuf/timestamp.proto";
 import "google/protobuf/any.proto";
 
-option go_package = "github.com/streamingfast/pbgo/dfuse/bstream/v1;pbbstream";
+option go_package = "github.com/streamingfast/pbgo/sf/firehose/v1;pbfirehose";
 
-service BlockStream {
-  rpc Blocks(BlockRequest) returns (stream Block);
-}
-
-service BlockStreamV2 {
-  rpc Blocks(BlocksRequestV2) returns (stream BlockResponseV2);
-}
-
-// Version 1 request
-
-message BlockRequest {
-  // Number of blocks we want to get in burst upon connection, on a best effort basis.
-  int64 burst = 1;
-
-  // Type of blocks we're after here, is it 'ethereum' data, 'eos', etc.. The server can fail early
-  // if he doesn't match the data he serves (services mismatch, etc..)
-  string content_type = 2;
-
-  // Whether we can assume the data will come ordered, unless there is a chain reorganization.
-  // mindreaders output ordered data, whereas relayers can output unordered data.
-  // The server can fail early if the assumption of the caller cannot be fulfilled.
-  enum Order {
-    UNSPECIFIED = 0;
-    ORDERED = 1;
-    UNORDERED = 2;
-  }
-  Order order = 3;
-
-  string requester = 4;
-}
-
-// Version 2 request
-
-message IrreversibleBlocksRequestV2 {
-  int64 start_block_num = 1;
+service Stream {
+  rpc Blocks(Request) returns (stream Response);
 }
 
 // For historical segments, forks are not passed
-message BlocksRequestV2 {
+message Request {
   // Controls where the stream of blocks will start.
   //
   // The stream will start **inclusively** at the requested block num.
@@ -66,7 +32,7 @@ message BlocksRequestV2 {
   // Controls where the stream of blocks will start which will be immediately after
   // the Block pointed by this opaque cursor.
   //
-  // Obtain this value from a previously received BlockResponseV2.cursor.
+  // Obtain this value from a previously received from `Response.cursor`.
   //
   // This value takes precedence over `start_block_num`.
   string start_cursor = 13;
@@ -92,17 +58,30 @@ message BlocksRequestV2 {
 
   // **Warning** Experimental API, controls how blocks are trimmed for extraneous information before
   // being sent back. The actual trimming is chain dependent.
-  BlockDetails details = 15;
+  //BlockDetails details = 15;
+  reserved 15;
 
   // controls how many confirmations will consider a given block as final (STEP_IRREVERSIBLE). Warning, if any reorg goes beyond that number of confirmations, the request will stall forever
-  uint64 confirmations = 16;
+  //uint64 confirmations = 16;
+  reserved 16;
+
 
+  //- EOS "handoffs:3"
+  //- EOS "lib"
+  //- EOS "confirms:3"
+  //- ETH "confirms:200"
+  //- ETH "confirms:7"
+  //- SOL "commmitement:finalized"
+  //- SOL "confirms:200"
+  string irreversibility_condition = 17;
+
+  repeated google.protobuf.Any transforms = 18;
 }
 
-message BlockResponseV2 {
+message Response {
   // Chain specific block payload, one of:
-  // - dfuse.eosio.codec.v1.Block
-  // - dfuse.ethereum.codec.v1.Block
+  // - sf.eosio.codec.v1.Block
+  // - sf.ethereum.codec.v1.Block
   // - sf.near.codec.v1.Block
   // - sf.solana.codec.v1.Block
   google.protobuf.Any block = 1;
@@ -124,41 +103,8 @@ enum ForkStep {
   reserved 5 ;
 }
 
+// TODO: move to ethereum specific transforms
 enum BlockDetails {
   BLOCK_DETAILS_FULL = 0;
   BLOCK_DETAILS_LIGHT = 1;
-}
-
-message Cursor {
-  BlockRef block = 1;
-  BlockRef head_block = 2;
-  BlockRef lib = 3;
-  ForkStep step = 4;
-}
-
-// General response and structs
-
-message Block {
-  uint64 number = 1;
-  string id = 2;
-  string previous_id = 3;
-  google.protobuf.Timestamp timestamp = 4;
-  uint64 lib_num = 5;
-
-  Protocol payload_kind = 6;
-  int32 payload_version = 7;
-  bytes payload_buffer = 8;
-}
-
-message BlockRef {
-  uint64 num = 1;
-  string id = 2;
-}
-
-enum Protocol {
-  UNKNOWN = 0;
-  EOS = 1;
-  ETH = 2;
-  SOLANA = 3;
-  NEAR = 4;
-}
+}

graph/proto/bstream.proto renamed to graph/proto/firehose.proto

@@ -5,7 +5,7 @@ use thiserror::Error;
 
 use super::{Block, BlockPtr, Blockchain};
 use crate::components::store::BlockNumber;
-use crate::firehose::bstream;
+use crate::firehose;
 use crate::{prelude::*, prometheus::labels};
 
 pub trait BlockStream<C: Blockchain>:
@@ -82,7 +82,7 @@ pub trait FirehoseMapper<C: Blockchain>: Send + Sync {
     async fn to_block_stream_event(
         &self,
         logger: &Logger,
-        response: &bstream::BlockResponseV2,
+        response: &firehose::Response,
         adapter: &C::TriggersAdapter,
         filter: &C::TriggerFilter,
     ) -> Result<BlockStreamEvent<C>, FirehoseError>;