Attempt fallback source when query is not retriable (#799)

Author: DZakh

codegenerator/cli/npm/envio/src/sources/HyperFuelSource.res

@@ -258,14 +258,10 @@ let make = ({chain, endpointUrl}: options): t => {
                 backoffMillis,
               })
             | UnexpectedMissingParams({missingParams}) =>
-              WithBackoff({
-                message: `Received page response with invalid data. Attempt a retry. Missing params: ${missingParams->Js.Array2.joinWith(
-                    ",",
+              ImpossibleForTheQuery({
+                message: `Source returned invalid data with missing required fields: ${missingParams->Js.Array2.joinWith(
+                    ", ",
                   )}`,
-                backoffMillis: switch retry {
-                | 0 => 1000
-                | _ => 4000 * retry
-                },
               })
             },
           }),

codegenerator/cli/npm/envio/src/sources/HyperSyncSource.res

@@ -288,14 +288,10 @@ let make = (
                 backoffMillis,
               })
             | UnexpectedMissingParams({missingParams}) =>
-              WithBackoff({
-                message: `Received page response with invalid data. Attempt a retry. Missing params: ${missingParams->Js.Array2.joinWith(
-                    ",",
+              ImpossibleForTheQuery({
+                message: `Source returned invalid data with missing required fields: ${missingParams->Js.Array2.joinWith(
+                    ", ",
                   )}`,
-                backoffMillis: switch retry {
-                | 0 => 1000
-                | _ => 4000 * retry
-                },
               })
             },
           }),

codegenerator/cli/npm/envio/src/sources/RpcSource.res

@@ -678,11 +678,12 @@ let make = (
       | exn =>
         raise(
           Source.GetItemsError(
-            FailedParsingItems({
-              message: "Failed to parse events using hypersync client decoder. Please double-check your ABI.",
+            FailedGettingItems({
               exn,
-              blockNumber: fromBlock,
-              logIndex: 0,
+              attemptedToBlock: toBlock,
+              retry: ImpossibleForTheQuery({
+                message: "Failed to parse events using hypersync client decoder. Please double-check your ABI.",
+              }),
             }),
           ),
         )
@@ -812,11 +813,12 @@ let make = (
                 | exn =>
                   raise(
                     Source.GetItemsError(
-                      FailedParsingItems({
-                        message: "Failed to parse event with viem, please double-check your ABI.",
+                      FailedGettingItems({
                         exn,
-                        blockNumber,
-                        logIndex,
+                        attemptedToBlock: toBlock,
+                        retry: ImpossibleForTheQuery({
+                          message: `Failed to parse event with viem, please double-check your ABI. Block number: ${blockNumber->Int.toString}, log index: ${logIndex->Int.toString}`,
+                        }),
                       }),
                     ),
                   )

codegenerator/cli/npm/envio/src/sources/Source.res

@@ -23,11 +23,11 @@ type blockRangeFetchResponse = {
 type getItemsRetry =
   | WithSuggestedToBlock({toBlock: int})
   | WithBackoff({message: string, backoffMillis: int})
+  | ImpossibleForTheQuery({message: string})
 
 type getItemsError =
   | UnsupportedSelection({message: string})
   | FailedGettingFieldSelection({exn: exn, blockNumber: int, logIndex: int, message: string})
-  | FailedParsingItems({exn: exn, blockNumber: int, logIndex: int, message: string})
   | FailedGettingItems({exn: exn, attemptedToBlock: int, retry: getItemsRetry})
 
 exception GetItemsError(getItemsError)

codegenerator/cli/npm/envio/src/sources/SourceManager.res

@@ -304,6 +304,7 @@ let getNextSyncSource = (
   sourceManager,
   // This is needed to include the Fallback source to rotation
   ~initialSource,
+  ~currentSource,
   // After multiple failures start returning fallback sources as well
   // But don't try it when main sync sources fail because of invalid configuration
   // note: The logic might be changed in the future
@@ -315,7 +316,7 @@ let getNextSyncSource = (
   let hasActive = ref(false)
 
   sourceManager.sources->Utils.Set.forEach(source => {
-    if source === sourceManager.activeSource {
+    if source === currentSource {
       hasActive := true
     } else if (
       switch source.sourceFor {
@@ -332,7 +333,7 @@ let getNextSyncSource = (
   | None =>
     switch before->Array.get(0) {
     | Some(s) => s
-    | None => sourceManager.activeSource
+    | None => currentSource
     }
   }
 }
@@ -349,9 +350,11 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
   let responseRef = ref(None)
   let retryRef = ref(0)
   let initialSource = sourceManager.activeSource
+  let sourceRef = ref(initialSource)
+  let shouldUpdateActiveSource = ref(false)
 
   while responseRef.contents->Option.isNone {
-    let source = sourceManager.activeSource
+    let source = sourceRef.contents
     let toBlock = toBlockRef.contents
     let retry = retryRef.contents
 
@@ -391,9 +394,8 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
     | Source.GetItemsError(error) =>
       switch error {
       | UnsupportedSelection(_)
-      | FailedGettingFieldSelection(_)
-      | FailedParsingItems(_) => {
-          let nextSource = sourceManager->getNextSyncSource(~initialSource)
+      | FailedGettingFieldSelection(_) => {
+          let nextSource = sourceManager->getNextSyncSource(~initialSource, ~currentSource=source)
 
           // These errors are impossible to recover, so we delete the source
           // from sourceManager so it's not attempted anymore
@@ -404,8 +406,7 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
           if notAlreadyDeleted {
             switch error {
             | UnsupportedSelection({message}) => logger->Logging.childError(message)
-            | FailedGettingFieldSelection({exn, message, blockNumber, logIndex})
-            | FailedParsingItems({exn, message, blockNumber, logIndex}) =>
+            | FailedGettingFieldSelection({exn, message, blockNumber, logIndex}) =>
               logger->Logging.childError({
                 "msg": message,
                 "err": exn->Utils.prettifyExn,
@@ -426,7 +427,8 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
               "msg": "Switching to another data-source",
               "source": nextSource.name,
             })
-            sourceManager.activeSource = nextSource
+            sourceRef := nextSource
+            shouldUpdateActiveSource := true
             retryRef := 0
           }
         }
@@ -438,6 +440,33 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
         })
         toBlockRef := Some(toBlock)
         retryRef := 0
+      | FailedGettingItems({exn, attemptedToBlock, retry: ImpossibleForTheQuery({message})}) =>
+        let nextSource =
+          sourceManager->getNextSyncSource(
+            ~initialSource,
+            ~currentSource=source,
+            ~attemptFallbacks=true,
+          )
+
+        let hasAnotherSource = nextSource !== initialSource
+
+        logger->Logging.childWarn({
+          "msg": message ++ (hasAnotherSource ? " - Attempting to another source" : ""),
+          "toBlock": attemptedToBlock,
+          "err": exn->Utils.prettifyExn,
+        })
+
+        if !hasAnotherSource {
+          %raw(`null`)->ErrorHandling.mkLogAndRaise(
+            ~logger,
+            ~msg="The indexer doesn't have data-sources which can continue fetching. Please, check the error logs or reach out to the Envio team.",
+          )
+        } else {
+          sourceRef := nextSource
+          shouldUpdateActiveSource := false
+          retryRef := 0
+        }
+
       | FailedGettingItems({exn, attemptedToBlock, retry: WithBackoff({message, backoffMillis})}) =>
         // Starting from the 11th failure (retry=10)
         // include fallback sources for switch
@@ -454,7 +483,11 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
         | _ =>
           // Then try to switch every second failure
           if retry->mod(2) === 0 {
-            sourceManager->getNextSyncSource(~initialSource, ~attemptFallbacks)
+            sourceManager->getNextSyncSource(
+              ~initialSource,
+              ~attemptFallbacks,
+              ~currentSource=source,
+            )
           } else {
             source
           }
@@ -476,16 +509,22 @@ let executeQuery = async (sourceManager: t, ~query: FetchState.query, ~currentBl
             "msg": "Switching to another data-source",
             "source": nextSource.name,
           })
-          sourceManager.activeSource = nextSource
+          sourceRef := nextSource
+          shouldUpdateActiveSource := true
         } else {
           await Utils.delay(Pervasives.min(backoffMillis, 60_000))
         }
         retryRef := retryRef.contents + 1
       }
+
     // TODO: Handle more error cases and hang/retry instead of throwing
     | exn => exn->ErrorHandling.mkLogAndRaise(~logger, ~msg="Failed to fetch block Range")
     }
   }
 
+  if shouldUpdateActiveSource.contents {
+    sourceManager.activeSource = sourceRef.contents
+  }
+
   responseRef.contents->Option.getUnsafe
 }

scenarios/test_codegen/test/E2E_test.res

@@ -558,4 +558,79 @@ describe("E2E tests", () => {
       ~message="Shouldn't increment on invalidation",
     )
   })
+
+  Async.it(
+    "Should attempt fallback source when primary source fails with missing params",
+    async () => {
+      let sourceMockPrimary = Mock.Source.make(
+        [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes],
+        ~chain=#1337,
+      )
+      let sourceMockFallback = Mock.Source.make(
+        [#getHeightOrThrow, #getItemsOrThrow, #getBlockHashes],
+        ~chain=#1337,
+      )
+      let indexerMock = await Mock.Indexer.make(
+        ~chains=[
+          {
+            chain: #1337,
+            sources: [sourceMockPrimary.source, sourceMockFallback.source],
+          },
+        ],
+      )
+      await Utils.delay(0)
+
+      // Resolve initial height request from primary source
+      Assert.deepEqual(
+        sourceMockPrimary.getHeightOrThrowCalls->Array.length,
+        1,
+        ~message="should have called getHeightOrThrow on primary source",
+      )
+      sourceMockPrimary.resolveGetHeightOrThrow(300)
+      await Utils.delay(0)
+      await Utils.delay(0)
+
+      // Primary source should now attempt to fetch items
+      Assert.deepEqual(
+        sourceMockPrimary.getItemsOrThrowCalls->Array.length,
+        1,
+        ~message="should have called getItemsOrThrow on primary source",
+      )
+
+      // Simulate missing params error from HyperSync (converted to InvalidData by the source)
+      sourceMockPrimary.rejectGetItemsOrThrow(
+        Source.GetItemsError(
+          FailedGettingItems({
+            exn: %raw(`null`),
+            attemptedToBlock: 100,
+            retry: ImpossibleForTheQuery({
+              message: "Source returned invalid data with missing required fields: log.address",
+            }),
+          }),
+        ),
+      )
+      await Utils.delay(0)
+      await Utils.delay(0)
+
+      // The fallback source should now be called immediately
+      Assert.deepEqual(
+        sourceMockFallback.getItemsOrThrowCalls->Array.length,
+        1,
+        ~message="fallback source should be called after primary fails with invalid data",
+      )
+
+      // Resolve the fallback source successfully
+      sourceMockFallback.resolveGetItemsOrThrow([], ~latestFetchedBlockNumber=100)
+      await indexerMock.getBatchWritePromise()
+
+      Assert.deepEqual(
+        (
+          sourceMockPrimary.getItemsOrThrowCalls->Array.length,
+          sourceMockFallback.getItemsOrThrowCalls->Array.length,
+        ),
+        (2, 1),
+        ~message="Shouldn't switch to fallback source for the next query",
+      )
+    },
+  )
 })