Parallelise State loading & FeeQuoter validation for faster resolution (#22065)

* Parallelise State loading & FeeQuoter validation for faster resolution

* fixes

Author: simsonraj

deployment/ccip/shared/stateview/evm/validate_feequoter.go

@@ -9,6 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
+	"golang.org/x/sync/errgroup"
 
 	chain_selectors "github.com/smartcontractkit/chain-selectors"
 
@@ -312,57 +313,72 @@ func (c CCIPChainState) validateAllDestChainConfigs(
 	v16FeeTokens []common.Address,
 	fqV2 *fqv2ops.FeeQuoterContract,
 ) error {
+	var mu sync.Mutex
 	var errs []error
+	grp := errgroup.Group{}
+	grp.SetLimit(20)
 
 	for _, destChainSel := range connectedChains {
-		var v16Cfg *fee_quoter.FeeQuoterDestChainConfig
-		var v20Cfg *fqv2ops.DestChainConfig
-		var legacyCfg *evm_2_evm_onramp.EVM2EVMOnRampDynamicConfig
-
-		if c.FeeQuoter != nil && v16FeeTokens != nil {
-			cfg, err := c.FeeQuoter.GetDestChainConfig(callOpts, destChainSel)
-			if err != nil {
-				errs = append(errs, fmt.Errorf("failed to get FeeQuoter v1.6 dest chain config for chain %d: %w", destChainSel, err))
-			} else {
-				v16Cfg = &cfg
+		dest := destChainSel
+		grp.Go(func() error {
+			var destErrs []error
+			var v16Cfg *fee_quoter.FeeQuoterDestChainConfig
+			var v20Cfg *fqv2ops.DestChainConfig
+			var legacyCfg *evm_2_evm_onramp.EVM2EVMOnRampDynamicConfig
+
+			if c.FeeQuoter != nil && v16FeeTokens != nil {
+				cfg, err := c.FeeQuoter.GetDestChainConfig(callOpts, dest)
+				if err != nil {
+					destErrs = append(destErrs, fmt.Errorf("failed to get FeeQuoter v1.6 dest chain config for chain %d: %w", dest, err))
+				} else {
+					v16Cfg = &cfg
+				}
 			}
-		}
-		if fqV2 != nil {
-			cfg, err := fqV2.GetDestChainConfig(callOpts, destChainSel)
-			if err != nil {
-				errs = append(errs, fmt.Errorf("failed to get FeeQuoter v2.0 dest chain config for chain %d: %w", destChainSel, err))
-			} else {
-				v20Cfg = &cfg
+			if fqV2 != nil {
+				cfg, err := fqV2.GetDestChainConfig(callOpts, dest)
+				if err != nil {
+					destErrs = append(destErrs, fmt.Errorf("failed to get FeeQuoter v2.0 dest chain config for chain %d: %w", dest, err))
+				} else {
+					v20Cfg = &cfg
+				}
 			}
-		}
-		if legacyOnRamp := c.EVM2EVMOnRamp[destChainSel]; legacyOnRamp != nil {
-			cfg, err := legacyOnRamp.GetDynamicConfig(callOpts)
-			if err != nil {
-				errs = append(errs, fmt.Errorf("failed to get v1.5 OnRamp dynamic config for dest chain %d: %w", destChainSel, err))
-			} else {
-				legacyCfg = &cfg
+			if legacyOnRamp := c.EVM2EVMOnRamp[dest]; legacyOnRamp != nil {
+				cfg, err := legacyOnRamp.GetDynamicConfig(callOpts)
+				if err != nil {
+					destErrs = append(destErrs, fmt.Errorf("failed to get v1.5 OnRamp dynamic config for dest chain %d: %w", dest, err))
+				} else {
+					legacyCfg = &cfg
+				}
 			}
-		}
 
-		v16Enabled := v16Cfg != nil && v16Cfg.IsEnabled
-		v20Enabled := v20Cfg != nil && v20Cfg.IsEnabled
+			v16Enabled := v16Cfg != nil && v16Cfg.IsEnabled
+			v20Enabled := v20Cfg != nil && v20Cfg.IsEnabled
 
-		// Skip v1.6 checks when lane is enabled only in v2.0.
-		if v16Cfg != nil && (v16Enabled || !v20Enabled) {
-			if err := c.validateV16DestChainConfig(callOpts, sourceChainSel, destChainSel, *v16Cfg, legacyCfg); err != nil {
-				errs = append(errs, err)
+			// Skip v1.6 checks when lane is enabled only in v2.0.
+			if v16Cfg != nil && (v16Enabled || !v20Enabled) {
+				if err := c.validateV16DestChainConfig(callOpts, sourceChainSel, dest, *v16Cfg, legacyCfg); err != nil {
+					destErrs = append(destErrs, err)
+				}
 			}
-		}
-		if v20Cfg != nil {
-			v16ForV20 := v16Cfg
-			if !v16Enabled && v20Enabled {
-				v16ForV20 = nil
+			if v20Cfg != nil {
+				v16ForV20 := v16Cfg
+				if !v16Enabled && v20Enabled {
+					v16ForV20 = nil
+				}
+				if err := c.validateV20DestChainConfig(callOpts, sourceChainSel, dest, *v20Cfg, v16ForV20, legacyCfg, fqV2); err != nil {
+					destErrs = append(destErrs, err)
+				}
 			}
-			if err := c.validateV20DestChainConfig(callOpts, sourceChainSel, destChainSel, *v20Cfg, v16ForV20, legacyCfg, fqV2); err != nil {
-				errs = append(errs, err)
+
+			if len(destErrs) > 0 {
+				mu.Lock()
+				errs = append(errs, destErrs...)
+				mu.Unlock()
 			}
-		}
+			return nil
+		})
 	}
+	_ = grp.Wait()
 
 	return errors.Join(errs...)
 }
@@ -533,33 +549,40 @@ func (c CCIPChainState) validateAllTokenTransferFeeConfigs(
 	e.Logger.Debugw("Validating TokenTransferFeeConfigs", "tokens", len(allTokens), "connectedChains", len(connectedChains))
 	var mu sync.Mutex
 	var errs []error
-	var wg sync.WaitGroup
-	sem := make(chan struct{}, 20)
+	outerGrp := errgroup.Group{}
+	outerGrp.SetLimit(20)
 	for _, tokenAddr := range allTokens {
 		token := tokenAddr
 		tokenLabel := token.Hex()
 		if sym, ok := addrToSymbol[token]; ok {
 			tokenLabel = fmt.Sprintf("%s (%s)", sym, token.Hex())
 		}
-		wg.Add(1)
-		sem <- struct{}{}
-		go func() {
-			defer wg.Done()
-			defer func() { <-sem }()
+		outerGrp.Go(func() error {
 			var tokenErrs []error
+			var tokenMu sync.Mutex
+			innerGrp := errgroup.Group{}
+			innerGrp.SetLimit(10)
 			for _, destChainSel := range connectedChains {
-				if err := c.validateTokenTransferFee(callOpts, destChainSel, token, tokenLabel, fqV2); err != nil {
-					tokenErrs = append(tokenErrs, err)
-				}
+				dest := destChainSel
+				innerGrp.Go(func() error {
+					if err := c.validateTokenTransferFee(callOpts, dest, token, tokenLabel, fqV2); err != nil {
+						tokenMu.Lock()
+						tokenErrs = append(tokenErrs, err)
+						tokenMu.Unlock()
+					}
+					return nil
+				})
 			}
+			_ = innerGrp.Wait()
 			if len(tokenErrs) > 0 {
 				mu.Lock()
 				errs = append(errs, tokenErrs...)
 				mu.Unlock()
 			}
-		}()
+			return nil
+		})
 	}
-	wg.Wait()
+	_ = outerGrp.Wait()
 
 	return errors.Join(errs...)
 }

deployment/ccip/shared/stateview/state.go

@@ -1048,20 +1048,30 @@ func LoadOnchainState(e cldf.Environment, opts ...LoadOption) (CCIPOnChainState,
 		TonChains:   tonChains,
 		evmMu:       &sync.RWMutex{},
 	}
+	grp, ctx := errgroup.WithContext(e.GetContext())
+	grp.SetLimit(10) // parallel EVM chain loading with bounded concurrency
 	for chainSelector, chain := range e.BlockChains.EVMChains() {
-		// get all addresses for chain from addressbook
-		// here we do not load addresses from datastore as there can be multiple
-		// contracts of the same type and version in datastore which can lead to
-		// ambiguity while loading the state
-		addresses, err := e.ExistingAddresses.AddressesForChain(chainSelector)
-		if err != nil && !errors.Is(err, cldf.ErrChainNotFound) {
-			return state, fmt.Errorf("failed to get addresses for chain %d: %w", chainSelector, err)
-		}
-		chainState, err := LoadChainState(e.GetContext(), chain, addresses, opts...)
-		if err != nil {
-			return state, err
-		}
-		state.WriteEVMChainState(chainSelector, chainState)
+		sel := chainSelector
+		ch := chain
+		grp.Go(func() error {
+			// get all addresses for chain from addressbook
+			// here we do not load addresses from datastore as there can be multiple
+			// contracts of the same type and version in datastore which can lead to
+			// ambiguity while loading the state
+			addresses, err := e.ExistingAddresses.AddressesForChain(sel)
+			if err != nil && !errors.Is(err, cldf.ErrChainNotFound) {
+				return fmt.Errorf("failed to get addresses for chain %d: %w", sel, err)
+			}
+			chainState, err := LoadChainState(ctx, ch, addresses, opts...)
+			if err != nil {
+				return err
+			}
+			state.WriteEVMChainState(sel, chainState)
+			return nil
+		})
+	}
+	if err := grp.Wait(); err != nil {
+		return state, err
 	}
 	return state, state.Validate()
 }