improve error log

graphprotocol · Feb 20, 2024 · b9a219e · b9a219e
1 parent b1bf41c
commit b9a219e
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 7 deletions.
diff --git a/chain/substreams/src/mapper.rs b/chain/substreams/src/mapper.rs
@@ -168,7 +168,7 @@ impl BlockStreamMapper<Chain> for Mapper {
 fn parse_changes(
     changes: &EntityChanges,
     schema: &InputSchema,
-) -> anyhow::Result<Vec<ParsedChanges>> {
+) -> Result<Vec<ParsedChanges>, SubstreamsError> {
     let mut parsed_changes = vec![];
     for entity_change in changes.entity_changes.iter() {
         let mut parsed_data: HashMap<Word, Value> = HashMap::default();

diff --git a/graph/src/blockchain/block_stream.rs b/graph/src/blockchain/block_stream.rs
@@ -465,12 +465,15 @@ pub enum SubstreamsError {
     #[error("invalid undo message")]
     InvalidUndoError,
 
+    #[error("entity validation failed {0}")]
+    EntityValidationError(#[from] crate::data::store::EntityValidationError),
+
     /// We were unable to decode the received block payload into the chain specific Block struct (e.g. chain_ethereum::pb::Block)
     #[error("received gRPC block payload cannot be decoded: {0}")]
     DecodingError(#[from] prost::DecodeError),
 
     /// Some unknown error occurred
-    #[error("unknown error")]
+    #[error("unknown error {0}")]
     UnknownError(#[from] anyhow::Error),
 
     #[error("multiple module output error")]
@@ -483,9 +486,28 @@ pub enum SubstreamsError {
     UnexpectedStoreDeltaOutput,
 }
 
+impl SubstreamsError {
+    pub fn is_deterministic(&self) -> bool {
+        use SubstreamsError::*;
+
+        match self {
+            EntityValidationError(_) => true,
+            MissingClockError
+            | InvalidUndoError
+            | DecodingError(_)
+            | UnknownError(_)
+            | MultipleModuleOutputError
+            | ModuleOutputNotPresentOrUnexpected
+            | UnexpectedStoreDeltaOutput => false,
+        }
+    }
+}
+
 #[derive(Debug, Error)]
 pub enum BlockStreamError {
-    #[error("block stream error")]
+    #[error("substreams error: {0}")]
+    SubstreamsError(#[from] SubstreamsError),
+    #[error("block stream error {0}")]
     Unknown(#[from] anyhow::Error),
     #[error("block stream fatal error")]
     Fatal(String),

diff --git a/graph/src/blockchain/substreams_block_stream.rs b/graph/src/blockchain/substreams_block_stream.rs
@@ -237,9 +237,12 @@ fn stream_blocks<C: Blockchain, F: BlockStreamMapper<C>>(
                                     }
                                 }
                             },
-                            Err(BlockStreamError::Fatal(msg)) => {
-                                Err(BlockStreamError::Fatal(msg))?
-                            }
+                            Err(BlockStreamError::SubstreamsError(e)) if e.is_deterministic()  =>
+                                Err(BlockStreamError::Fatal(e.to_string()))?,
+
+                            Err(BlockStreamError::Fatal(msg)) =>
+                                Err(BlockStreamError::Fatal(msg))?,
+
                             Err(err) => {
 
                                 info!(&logger, "received err");
@@ -308,7 +311,7 @@ async fn process_substreams_response<C: Blockchain, F: BlockStreamMapper<C>>(
     match mapper
         .to_block_stream_event(logger, response.message, log_data)
         .await
-        .context("Mapping message to BlockStreamEvent failed")?
+        .map_err(BlockStreamError::from)?
     {
         Some(event) => {
             let cursor = match &event {