Update error handling for strategy execution

Author: Ferossgp

packages/transaction-decoder/src/abi-loader.ts

@@ -61,7 +61,7 @@ const getMany = (requests: Array<AbiStore.AbiParams>) =>
       return yield* Effect.all(
         requests.map(({ chainID, address, event, signature }) => get({ chainID, address, event, signature })),
         {
-          concurrency: 'inherit',
+          concurrency: 'unbounded',
           batching: 'inherit',
         },
       )
@@ -170,6 +170,7 @@ export const AbiLoaderRequestResolver = RequestResolver.makeBatched((requests: A
       },
       {
         discard: true,
+        concurrency: 'unbounded',
       },
     )
 
@@ -182,6 +183,7 @@ export const AbiLoaderRequestResolver = RequestResolver.makeBatched((requests: A
       }
     }
 
+    const concurrency = Math.min(...[...concurrencyMap.values(), 50]) // Use minimum concurrency across all chains, capped at 25
     // NOTE: Firstly we batch strategies by address because in a transaction most of events and traces are from the same abi
     const response = yield* Effect.forEach(
       remaining,
@@ -196,10 +198,14 @@ export const AbiLoaderRequestResolver = RequestResolver.makeBatched((requests: A
             chainId: req.chainID,
             strategyId: 'address-batch',
           })
-          .pipe(Effect.map((result) => (result ? Either.left(result) : Either.right(req))))
+          .pipe(
+            Effect.tapError(Effect.logWarning),
+            Effect.orElseSucceed(() => null),
+            Effect.map((result) => (result ? Either.left(result) : Either.right(req))),
+          )
       },
       {
-        concurrency: Math.min(...[...concurrencyMap.values()], 50), // Use minimum to be conservative
+        concurrency,
       },
     )
 
@@ -213,16 +219,21 @@ export const AbiLoaderRequestResolver = RequestResolver.makeBatched((requests: A
           (strategy) => strategy.type === 'fragment',
         )
 
-        return strategyExecutor.executeStrategiesSequentially(allAvailableStrategies, {
-          address,
-          chainId: chainID,
-          event,
-          signature,
-          strategyId: 'fragment-batch',
-        })
+        return strategyExecutor
+          .executeStrategiesSequentially(allAvailableStrategies, {
+            address,
+            chainId: chainID,
+            event,
+            signature,
+            strategyId: 'fragment-batch',
+          })
+          .pipe(
+            Effect.tapError(Effect.logWarning),
+            Effect.orElseSucceed(() => null),
+          ) // If no strategies found, return null
       },
       {
-        concurrency: Math.min(...[...concurrencyMap.values()], 25), // More conservative for fragments
+        concurrency,
         batching: true,
       },
     )

packages/transaction-decoder/src/abi-strategy/strategy-executor.ts

@@ -1,9 +1,18 @@
-import { Effect, Schedule, Duration, pipe } from 'effect'
+import { Effect, Schedule, Duration, pipe, Data } from 'effect'
 import { GetContractABIStrategyParams, ContractAbiResolverStrategy } from '../abi-strategy/request-model.js'
 import type { CircuitBreaker } from '../circuit-breaker/circuit-breaker.js'
 import { RequestPool } from '../circuit-breaker/request-pool.js'
 import * as Constants from '../circuit-breaker/constants.js'
 
+export class MissingHealthyStrategy extends Data.TaggedError('MissingHealthyStrategy')<{
+  chainId: number
+  strategies: string[]
+}> {
+  constructor(params: { chainId: number; strategies: string[] }) {
+    super(params)
+  }
+}
+
 export const make = (circuitBreaker: CircuitBreaker, requestPool: RequestPool) => {
   const executeStrategy = (strategy: ContractAbiResolverStrategy, params: GetContractABIStrategyParams) => {
     return pipe(
@@ -44,14 +53,16 @@ export const make = (circuitBreaker: CircuitBreaker, requestPool: RequestPool) =
       }
 
       if (healthyStrategies.length === 0) {
-        yield* Effect.logWarning(`No healthy strategies available for chain ${params.chainId}`)
-        return null
+        return yield* Effect.fail(
+          new MissingHealthyStrategy({
+            chainId: params.chainId,
+            strategies: strategies.map((s) => s.id),
+          }),
+        )
       }
 
       // Try strategies one by one until one succeeds
-      return yield* Effect.validateFirst(healthyStrategies, (strategy) => executeStrategy(strategy, params)).pipe(
-        Effect.orElseSucceed(() => null),
-      )
+      return yield* Effect.validateFirst(healthyStrategies, (strategy) => executeStrategy(strategy, params))
     })
 
   return {

packages/transaction-decoder/src/contract-meta-loader.ts

@@ -138,11 +138,16 @@ const ContractMetaLoaderRequestResolver = RequestResolver.makeBatched((requests:
       ({ chainID, address }) => {
         const allAvailableStrategies = Array.prependAll(strategies.default, strategies[chainID] ?? [])
 
-        return metaStrategyExecutor.executeStrategiesSequentially(allAvailableStrategies, {
-          address,
-          chainId: chainID,
-          strategyId: 'meta-batch',
-        })
+        return metaStrategyExecutor
+          .executeStrategiesSequentially(allAvailableStrategies, {
+            address,
+            chainId: chainID,
+            strategyId: 'meta-batch',
+          })
+          .pipe(
+            Effect.tapError(Effect.logWarning),
+            Effect.orElseSucceed(() => null),
+          )
       },
       {
         concurrency: Math.min(...[...concurrencyMap.values()], 25), // Conservative concurrency

packages/transaction-decoder/src/meta-strategy/strategy-executor.ts

@@ -1,9 +1,18 @@
-import { Effect, Schedule, Duration, pipe } from 'effect'
+import { Effect, Schedule, Duration, pipe, Data } from 'effect'
 import { FetchMetaParams, ContractMetaResolverStrategy, GetContractMetaStrategy } from './request-model.js'
 import type { CircuitBreaker } from '../circuit-breaker/circuit-breaker.js'
 import { RequestPool } from '../circuit-breaker/request-pool.js'
 import * as Constants from '../circuit-breaker/constants.js'
 
+export class MissingHealthyStrategy extends Data.TaggedError('MissingHealthyStrategy')<{
+  chainId: number
+  strategies: string[]
+}> {
+  constructor(params: { chainId: number; strategies: string[] }) {
+    super(params)
+  }
+}
+
 export const make = (circuitBreaker: CircuitBreaker, requestPool: RequestPool) => {
   const executeStrategy = (strategy: ContractMetaResolverStrategy, params: FetchMetaParams) => {
     return pipe(
@@ -49,14 +58,16 @@ export const make = (circuitBreaker: CircuitBreaker, requestPool: RequestPool) =
       }
 
       if (healthyStrategies.length === 0) {
-        yield* Effect.logWarning(`No healthy meta strategies available for chain ${params.chainId}`)
-        return null
+        return yield* Effect.fail(
+          new MissingHealthyStrategy({
+            chainId: params.chainId,
+            strategies: strategies.map((s) => s.id),
+          }),
+        )
       }
 
       // Try strategies one by one until one succeeds
-      return yield* Effect.validateFirst(healthyStrategies, (strategy) => executeStrategy(strategy, params)).pipe(
-        Effect.orElseSucceed(() => null),
-      )
+      return yield* Effect.validateFirst(healthyStrategies, (strategy) => executeStrategy(strategy, params))
     })
 
   return {