Updated Substreams to latest version of Protobuf definition and activated ProductionMode by default (#4211)

The production mode is required to benefits from automatic backprocessing and downloading of block scoped data message as they are produced. This will drastically improve the ingestion speed of a substreams (time to gather some metrics!).

Updated also the instructions to re-generate the `substreams.proto` file with more instructions of how we do it.

Co-authored-by: Matthieu Vachon <[email protected]>

Author: maoueh

chain/substreams/src/mapper.rs

@@ -98,7 +98,7 @@ impl SubstreamsMapper<Chain> for Mapper {
                     }
                 }
             }
-            Some(Data::StoreDeltas(_)) => Err(UnexpectedStoreDeltaOutput),
+            Some(Data::DebugStoreDeltas(_)) => Err(UnexpectedStoreDeltaOutput),
             _ => Err(SubstreamsError::ModuleOutputNotPresentOrUnexpected),
         }
     }

graph/proto/substreams.proto

@@ -1,16 +1,27 @@
+// File generated using this command at the root of `graph-node` project
+// and assuming `substreams` repository is a sibling of `graph-node` (note that you
+// might need to adjust the `head -nN` and `skip N` values in the commands below to skip
+// more/less lines):
+//
+// ```
+// cat graph/proto/substreams.proto | head -n16 > /tmp/substreams.proto && mv /tmp/substreams.proto graph/proto/substreams.proto
+// cat ../substreams/proto/sf/substreams/v1/substreams.proto | grep -Ev 'import *"sf/substreams' >> graph/proto/substreams.proto
+// cat ../substreams/proto/sf/substreams/v1/modules.proto | skip 6 >> graph/proto/substreams.proto
+// cat ../substreams/proto/sf/substreams/v1/package.proto | skip 9 >> graph/proto/substreams.proto
+// cat ../substreams/proto/sf/substreams/v1/clock.proto | skip 7 >> graph/proto/substreams.proto
+// # Manually add line `import "google/protobuf/descriptor.proto";` below `import "google/protobuf/timestamp.proto";`
+// ```
+//
+// FIXME: We copy over and inline most of the substreams files, this is bad and we need a better way to
+// generate that, outside of doing this copying over.
 syntax = "proto3";
 
 package sf.substreams.v1;
-
 option go_package = "github.com/streamingfast/substreams/pb/sf/substreams/v1;pbsubstreams";
 
 import "google/protobuf/any.proto";
-import "google/protobuf/descriptor.proto";
 import "google/protobuf/timestamp.proto";
-
-// FIXME: I copied over and inlined most of the substreams files, this is bad and we need a better way to
-// generate that, outside of doing this copying over. We should check maybe `buf` or a pre-populated
-// package.
+import "google/protobuf/descriptor.proto";
 
 service Stream {
   rpc Blocks(Request) returns (stream Response);
@@ -23,13 +34,26 @@ message Request {
   repeated ForkStep fork_steps = 4;
   string irreversibility_condition = 5;
 
+  // By default, the engine runs in developer mode, with richer and deeper output,
+  // * support for multiple `output_modules`, of `store` and `map` kinds
+  // * support for `initial_store_snapshot_for_modules`
+  // * log outputs for output modules
+  //
+  // With `production_mode`, however, you trade off functionality for high speed, where it:
+  // * restricts the possible requested `output_modules` to a single mapper module,
+  // * turns off support for `initial_store_snapshot_for_modules`,
+  // * still streams output linearly, with a cursor, but at higher speeds
+  // * and purges log outputs from responses.
+  bool production_mode = 9;
+
   Modules modules = 6;
   repeated string output_modules = 7;
   repeated string initial_store_snapshot_for_modules = 8;
 }
 
 message Response {
   oneof message {
+    SessionInit session = 5; // Always sent first
     ModulesProgress progress = 1; // Progress of data preparation, before sending in the stream of `data` events.
     InitialSnapshotData snapshot_data = 2;
     InitialSnapshotComplete snapshot_complete = 3;
@@ -51,6 +75,10 @@ enum ForkStep {
   reserved 5;
 }
 
+message SessionInit  {
+  string trace_id = 1;
+}
+
 message InitialSnapshotComplete {
   string cursor = 1;
 }
@@ -71,17 +99,36 @@ message BlockScopedData {
 
 message ModuleOutput {
   string name = 1;
+
   oneof data {
     google.protobuf.Any map_output = 2;
-    StoreDeltas store_deltas = 3;
-  }
-  repeated string logs = 4;
 
+    // StoreDeltas are produced for store modules in development mode.
+    // It is not possible to retrieve store models in production, with parallelization
+    // enabled. If you need the deltas directly, write a pass through mapper module
+    // that will get them down to you.
+    StoreDeltas debug_store_deltas = 3;
+  }
+  repeated string debug_logs = 4;
   // LogsTruncated is a flag that tells you if you received all the logs or if they
   // were truncated because you logged too much (fixed limit currently is set to 128 KiB).
-  bool logs_truncated = 5;
+  bool debug_logs_truncated = 5;
+
+  bool cached = 6;
 }
 
+// think about:
+// message ModuleOutput { ...
+//   ModuleOutputDebug debug_info = 6;
+// ...}
+//message ModuleOutputDebug {
+//  StoreDeltas store_deltas = 3;
+//  repeated string logs = 4;
+//  // LogsTruncated is a flag that tells you if you received all the logs or if they
+//  // were truncated because you logged too much (fixed limit currently is set to 128 KiB).
+//  bool logs_truncated = 5;
+//}
+
 message ModulesProgress {
   repeated ModuleProgress modules = 1;
 }
@@ -116,8 +163,8 @@ message ModuleProgress {
 }
 
 message BlockRange {
-  uint64 start_block = 1;
-  uint64 end_block = 2;
+  uint64 start_block = 2;
+  uint64 end_block = 3;
 }
 
 message StoreDeltas {
@@ -144,7 +191,6 @@ message Output {
   google.protobuf.Timestamp timestamp = 4;
   google.protobuf.Any value = 10;
 }
-
 message Modules {
   repeated Module modules = 1;
   repeated Binary binaries = 2;
@@ -199,8 +245,9 @@ message Module {
       UPDATE_POLICY_MIN = 4;
       // Provides a store where you can `max_*()` keys, where two stores merge by leaving the maximum value.
       UPDATE_POLICY_MAX = 5;
+      // Provides a store where you can `append()` keys, where two stores merge by concatenating the bytes in order.
+      UPDATE_POLICY_APPEND = 6;
     }
-
   }
 
   message Input {
@@ -232,13 +279,6 @@ message Module {
     string type = 1;
   }
 }
-
-message Clock {
-  string id = 1;
-  uint64 number = 2;
-  google.protobuf.Timestamp timestamp = 3;
-}
-
 message Package {
   // Needs to be one so this file can be used _directly_ as a
   // buf `Image` andor a ProtoSet for grpcurl and other tools
@@ -265,3 +305,8 @@ message ModuleMetadata {
   uint64 package_index = 1;
   string doc = 2;
 }
+message Clock {
+  string id = 1;
+  uint64 number = 2;
+  google.protobuf.Timestamp timestamp = 3;
+}

graph/src/blockchain/substreams_block_stream.rs

@@ -186,6 +186,7 @@ fn stream_blocks<C: Blockchain, F: SubstreamsMapper<C>>(
         irreversibility_condition: "".to_string(),
         modules,
         output_modules: vec![module_name],
+        production_mode: true,
         ..Default::default()
     };
 

graph/src/substreams/sf.substreams.v1.rs

@@ -10,6 +10,18 @@ pub struct Request {
     pub fork_steps: ::prost::alloc::vec::Vec<i32>,
     #[prost(string, tag="5")]
     pub irreversibility_condition: ::prost::alloc::string::String,
+    /// By default, the engine runs in developer mode, with richer and deeper output,
+    /// * support for multiple `output_modules`, of `store` and `map` kinds
+    /// * support for `initial_store_snapshot_for_modules`
+    /// * log outputs for output modules
+    ///
+    /// With `production_mode`, however, you trade off functionality for high speed, where it:
+    /// * restricts the possible requested `output_modules` to a single mapper module,
+    /// * turns off support for `initial_store_snapshot_for_modules`,
+    /// * still streams output linearly, with a cursor, but at higher speeds
+    /// * and purges log outputs from responses.
+    #[prost(bool, tag="9")]
+    pub production_mode: bool,
     #[prost(message, optional, tag="6")]
     pub modules: ::core::option::Option<Modules>,
     #[prost(string, repeated, tag="7")]
@@ -19,13 +31,16 @@ pub struct Request {
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Response {
-    #[prost(oneof="response::Message", tags="1, 2, 3, 4")]
+    #[prost(oneof="response::Message", tags="5, 1, 2, 3, 4")]
     pub message: ::core::option::Option<response::Message>,
 }
 /// Nested message and enum types in `Response`.
 pub mod response {
     #[derive(Clone, PartialEq, ::prost::Oneof)]
     pub enum Message {
+        /// Always sent first
+        #[prost(message, tag="5")]
+        Session(super::SessionInit),
         /// Progress of data preparation, before sending in the stream of `data` events.
         #[prost(message, tag="1")]
         Progress(super::ModulesProgress),
@@ -38,6 +53,11 @@ pub mod response {
     }
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SessionInit {
+    #[prost(string, tag="1")]
+    pub trace_id: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct InitialSnapshotComplete {
     #[prost(string, tag="1")]
     pub cursor: ::prost::alloc::string::String,
@@ -69,11 +89,13 @@ pub struct ModuleOutput {
     #[prost(string, tag="1")]
     pub name: ::prost::alloc::string::String,
     #[prost(string, repeated, tag="4")]
-    pub logs: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    pub debug_logs: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     /// LogsTruncated is a flag that tells you if you received all the logs or if they
     /// were truncated because you logged too much (fixed limit currently is set to 128 KiB).
     #[prost(bool, tag="5")]
-    pub logs_truncated: bool,
+    pub debug_logs_truncated: bool,
+    #[prost(bool, tag="6")]
+    pub cached: bool,
     #[prost(oneof="module_output::Data", tags="2, 3")]
     pub data: ::core::option::Option<module_output::Data>,
 }
@@ -83,10 +105,26 @@ pub mod module_output {
     pub enum Data {
         #[prost(message, tag="2")]
         MapOutput(::prost_types::Any),
+        /// StoreDeltas are produced for store modules in development mode.
+        /// It is not possible to retrieve store models in production, with parallelization
+        /// enabled. If you need the deltas directly, write a pass through mapper module
+        /// that will get them down to you.
         #[prost(message, tag="3")]
-        StoreDeltas(super::StoreDeltas),
+        DebugStoreDeltas(super::StoreDeltas),
     }
 }
+// think about:
+// message ModuleOutput { ...
+//   ModuleOutputDebug debug_info = 6;
+// ...}
+//message ModuleOutputDebug {
+//  StoreDeltas store_deltas = 3;
+//  repeated string logs = 4;
+//  // LogsTruncated is a flag that tells you if you received all the logs or if they
+//  // were truncated because you logged too much (fixed limit currently is set to 128 KiB).
+//  bool logs_truncated = 5;
+//}
+
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ModulesProgress {
     #[prost(message, repeated, tag="1")]
@@ -143,9 +181,9 @@ pub mod module_progress {
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BlockRange {
-    #[prost(uint64, tag="1")]
-    pub start_block: u64,
     #[prost(uint64, tag="2")]
+    pub start_block: u64,
+    #[prost(uint64, tag="3")]
     pub end_block: u64,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -258,6 +296,8 @@ pub mod module {
             Min = 4,
             /// Provides a store where you can `max_*()` keys, where two stores merge by leaving the maximum value.
             Max = 5,
+            /// Provides a store where you can `append()` keys, where two stores merge by concatenating the bytes in order.
+            Append = 6,
         }
     }
     #[derive(Clone, PartialEq, ::prost::Message)]
@@ -320,15 +360,6 @@ pub mod module {
     }
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct Clock {
-    #[prost(string, tag="1")]
-    pub id: ::prost::alloc::string::String,
-    #[prost(uint64, tag="2")]
-    pub number: u64,
-    #[prost(message, optional, tag="3")]
-    pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
-}
-#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Package {
     /// Needs to be one so this file can be used _directly_ as a
     /// buf `Image` andor a ProtoSet for grpcurl and other tools
@@ -362,6 +393,15 @@ pub struct ModuleMetadata {
     #[prost(string, tag="2")]
     pub doc: ::prost::alloc::string::String,
 }
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Clock {
+    #[prost(string, tag="1")]
+    pub id: ::prost::alloc::string::String,
+    #[prost(uint64, tag="2")]
+    pub number: u64,
+    #[prost(message, optional, tag="3")]
+    pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
+}
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]
 pub enum ForkStep {