Reset substreams/firehose block ingestor backoff (#5047)

* Reset substreams/firehose block ingestor backoff fixes #4976 * fix cursor comparison
graphprotocol · Mar 15, 2024 · 1b94a73 · 1b94a73
1 parent 3c03b3f
commit 1b94a73
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/chain/substreams/src/block_ingestor.rs b/chain/substreams/src/block_ingestor.rs
@@ -169,9 +169,14 @@ impl BlockIngestor for SubstreamsBlockIngestor {
             // If the error is retryable it will print the error and return the cursor
             // therefore if we get an error here it has to be a fatal error.
             // This is a bit brittle and should probably be improved at some point.
-            let res = self.process_blocks(latest_cursor, stream).await;
+            let res = self.process_blocks(latest_cursor.clone(), stream).await;
             match res {
-                Ok(cursor) => latest_cursor = cursor,
+                Ok(cursor) => {
+                    if cursor.as_ref() != latest_cursor.as_ref() {
+                        backoff.reset();
+                        latest_cursor = cursor;
+                    }
+                }
                 Err(BlockStreamError::Fatal(e)) => {
                     error!(
                         self.logger,

diff --git a/graph/src/blockchain/firehose_block_ingestor.rs b/graph/src/blockchain/firehose_block_ingestor.rs
@@ -207,7 +207,11 @@ where
                     info!(logger, "Blockstream connected, consuming blocks");
 
                     // Consume the stream of blocks until an error is hit
-                    latest_cursor = self.process_blocks(latest_cursor, stream).await
+                    let cursor = self.process_blocks(latest_cursor.clone(), stream).await;
+                    if cursor != latest_cursor {
+                        backoff.reset();
+                        latest_cursor = cursor;
+                    }
                 }
                 Err(e) => {
                     error!(logger, "Unable to connect to endpoint: {:#}", e);