CCIP-4420 Merging back CCIP codebase

.mockery.yaml

@@ -490,12 +490,20 @@ packages:
       PriceRegistryReader:
         config:
           filename: price_registry_reader_mock.go
+      FeeEstimatorConfigReader:
+        config:
+          filename: fee_estimator_config_mock.go
       TokenPoolReader:
         config:
           filename: token_pool_reader_mock.go
       USDCReader:
         config:
           filename: usdc_reader_mock.go
+  github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/estimatorconfig:
+    interfaces:
+      GasPriceInterceptor:
+        config:
+          filename: gas_price_interceptor_mock.go
   github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/batchreader:
     config:
       filename: token_pool_batched_reader_mock.go

ccip/config/evm/Sei_Mainnet.toml

@@ -0,0 +1,18 @@
+ChainID = '1329'
+ChainType = 'sei'
+# finality_depth: instant
+FinalityDepth = 10
+# block_time: ~0.4s, adding 1 second buffer
+LogPollInterval = '2s'
+# finality_depth * block_time / 60 secs = ~0.8 min (finality time)
+NoNewFinalizedHeadsThreshold = '5m'
+# "RPC node returned multiple missing blocks on query for block numbers [31592085 31592084] even though the WS subscription already sent us these blocks. It might help to increase EVM.RPCBlockQueryDelay (currently 1)"
+RPCBlockQueryDelay = 5 
+
+[GasEstimator]
+EIP1559DynamicFees = false
+Mode = 'BlockHistory'
+PriceMax = '3000 gwei' # recommended by ds&a
+
+[GasEstimator.BlockHistory]
+BlockHistorySize = 200

core/chains/evm/client/errors.go

@@ -265,9 +265,11 @@ var aStar = ClientErrors{
 }
 
 var mantle = ClientErrors{
-	InsufficientEth: regexp.MustCompile(`(: |^)'*insufficient funds for gas \* price \+ value`),
-	Fatal:           regexp.MustCompile(`(: |^)'*invalid sender`),
-	NonceTooLow:     regexp.MustCompile(`(: |^)'*nonce too low`),
+	InsufficientEth:                   regexp.MustCompile(`(: |^)'*insufficient funds for gas \* price \+ value`),
+	Fatal:                             regexp.MustCompile(`(: |^)'*invalid sender`),
+	NonceTooLow:                       regexp.MustCompile(`(: |^)'*nonce too low`),
+	ReplacementTransactionUnderpriced: regexp.MustCompile(`(: |^)'*replacement transaction underpriced`),
+	TransactionAlreadyInMempool:       regexp.MustCompile(`(: |^)'*already known`),
 }
 
 var hederaFatal = regexp.MustCompile(`(: |^)(execution reverted)(:|$) | ^Transaction gas limit '(\d+)' exceeds block gas limit '(\d+)' | ^Transaction gas limit provided '(\d+)' is insufficient of intrinsic gas required '(\d+)' | ^Oversized data:|status INVALID_SIGNATURE`)

core/chains/evm/client/errors_test.go

@@ -112,6 +112,7 @@ func Test_Eth_Errors(t *testing.T) {
 			{"gas price too low", false, "Arbitrum"},
 			{"client error replacement underpriced", true, "tomlConfig"},
 			{"", false, "tomlConfig"},
+			{"failed to forward tx to sequencer, please try again. Error message: 'replacement transaction underpriced'", true, "Mantle"},
 		}
 
 		for _, test := range tests {
@@ -145,6 +146,8 @@ func Test_Eth_Errors(t *testing.T) {
 			{"client error transaction already in mempool", true, "tomlConfig"},
 			{"alreadyknown", true, "Gnosis"},
 			{"tx already exists in cache", true, "Sei"},
+			{"failed to forward tx to sequencer, please try again. Error message: 'already known'", true, "Mantle"},
+			{"tx already exists in cache", true, "Sei"},
 		}
 		for _, test := range tests {
 			err = evmclient.NewSendErrorS(test.message)

core/chains/evm/gas/block_history_estimator_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 
 	commonfee "github.com/smartcontractkit/chainlink/v2/common/fee"

core/chains/evm/gas/rollups/op_l1_oracle.go

@@ -26,6 +26,8 @@ import (
 )
 
 // Reads L2-specific precompiles and caches the l1GasPrice set by the L2.
+//
+//nolint:unused // backported from CCIP
 type optimismL1Oracle struct {
 	services.StateMachine
 	client     l1OracleClient

core/chains/evm/types/models_test.go

@@ -954,7 +954,7 @@ func TestBlock_UnmarshalJSON(t *testing.T) {
 	t.Run("unmarshals geth block", func(t *testing.T) {
 		b := new(evmtypes.Block)
 		err := b.UnmarshalJSON([]byte(gethSampleBlock))
-		assert.NoError(t, err)
+		require.NoError(t, err)
 
 		assert.Equal(t, int64(15051090), b.Number)
 		assert.Equal(t, "0x45eb0a650b6b0b9fd1ee676b870e43fa7614f1034f7404070327a332faed05c0", b.Hash.Hex())
@@ -966,14 +966,25 @@ func TestBlock_UnmarshalJSON(t *testing.T) {
 	t.Run("handles empty result", func(t *testing.T) {
 		b := new(evmtypes.Block)
 		err := b.UnmarshalJSON([]byte("null"))
-		assert.Error(t, err)
+		require.Error(t, err)
 		assert.Equal(t, pkgerrors.Cause(err), evmtypes.ErrMissingBlock)
 		assert.True(t, pkgerrors.Is(err, evmtypes.ErrMissingBlock))
 	})
 	t.Run("unmarshals EIP-4844 block", func(t *testing.T) {
 		b := new(evmtypes.Block)
 		err := b.UnmarshalJSON([]byte(eip4844Block))
-		assert.NoError(t, err)
+		require.NoError(t, err)
+		assert.Equal(t, int64(5300694), b.Number)
+		assert.Equal(t, "0x3edd900025edab70dde26a52377c3d0a9474c3f540bd0131d58f508711272590", b.Hash.Hex())
+		assert.Equal(t, "0x077c1d68b52f8203cb90a71759a09b11c2a6577f97ea1fd4a8686a387fbedac8", b.ParentHash.Hex())
+		assert.Equal(t, assets.NewWeiI(96436174005), b.BaseFeePerGas)
+		assert.Equal(t, int64(1708087260), b.Timestamp.Unix())
+		assert.Len(t, b.Transactions, 6)
+	})
+	t.Run("unmarshals EIP-4844 block", func(t *testing.T) {
+		b := new(evmtypes.Block)
+		err := b.UnmarshalJSON([]byte(eip4844Block))
+		require.NoError(t, err)
 		assert.Equal(t, int64(5300694), b.Number)
 		assert.Equal(t, "0x3edd900025edab70dde26a52377c3d0a9474c3f540bd0131d58f508711272590", b.Hash.Hex())
 		assert.Equal(t, "0x077c1d68b52f8203cb90a71759a09b11c2a6577f97ea1fd4a8686a387fbedac8", b.ParentHash.Hex())

core/services/ocr2/delegate.go

@@ -7,13 +7,15 @@ import (
 	"fmt"
 	"log"
 	"strconv"
+	"strings"
 	"time"
 
 	"gopkg.in/guregu/null.v4"
 
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -41,7 +43,6 @@ import (
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox"
 	datastreamsllo "github.com/smartcontractkit/chainlink-data-streams/llo"
-
 	"github.com/smartcontractkit/chainlink/v2/core/bridges"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/legacyevm"
 	coreconfig "github.com/smartcontractkit/chainlink/v2/core/config"
@@ -264,12 +265,12 @@ func (d *Delegate) JobType() job.Type {
 	return job.OffchainReporting2
 }
 
-func (d *Delegate) BeforeJobCreated(spec job.Job) {
+func (d *Delegate) BeforeJobCreated(_ job.Job) {
 	// This is only called first time the job is created
 	d.isNewlyCreatedJob = true
 }
-func (d *Delegate) AfterJobCreated(spec job.Job)  {}
-func (d *Delegate) BeforeJobDeleted(spec job.Job) {}
+func (d *Delegate) AfterJobCreated(_ job.Job)  {}
+func (d *Delegate) BeforeJobDeleted(_ job.Job) {}
 func (d *Delegate) OnDeleteJob(ctx context.Context, jb job.Job) error {
 	// If the job spec is malformed in any way, we report the error but return nil so that
 	//  the job deletion itself isn't blocked.
@@ -1650,7 +1651,85 @@ func (d *Delegate) newServicesCCIPCommit(ctx context.Context, lggr logger.Sugare
 		MetricsRegisterer:      prometheus.WrapRegistererWith(map[string]string{"job_name": jb.Name.ValueOrZero()}, prometheus.DefaultRegisterer),
 	}
 
-	return ccipcommit.NewCommitServices(ctx, d.ds, srcProvider, dstProvider, d.legacyChains, jb, lggr, d.pipelineRunner, oracleArgsNoPlugin, d.isNewlyCreatedJob, int64(srcChainID), dstChainID, logError)
+	priceGetter, err := d.ccipCommitPriceGetter(ctx, lggr, pluginJobSpecConfig, jb)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create price getter: %w", err)
+	}
+	//nolint:gosec // safe to cast
+	return ccipcommit.NewCommitServices(ctx, d.ds, srcProvider, dstProvider, priceGetter, jb, lggr, d.pipelineRunner, oracleArgsNoPlugin, d.isNewlyCreatedJob, int64(srcChainID), dstChainID, logError)
+}
+
+func (d *Delegate) ccipCommitPriceGetter(ctx context.Context, lggr logger.SugaredLogger, pluginJobSpecConfig ccipconfig.CommitPluginJobSpecConfig, jb job.Job) (priceGetter ccip.AllTokensPriceGetter, err error) {
+	spec := jb.OCR2OracleSpec
+	withPipeline := strings.Trim(pluginJobSpecConfig.TokenPricesUSDPipeline, "\n\t ") != ""
+	if withPipeline {
+		priceGetter, err = ccip.NewPipelineGetter(pluginJobSpecConfig.TokenPricesUSDPipeline, d.pipelineRunner, jb.ID, jb.ExternalJobID, jb.Name.ValueOrZero(), lggr)
+		if err != nil {
+			return nil, fmt.Errorf("creating pipeline price getter: %w", err)
+		}
+	} else {
+		// Use dynamic price getter.
+		if pluginJobSpecConfig.PriceGetterConfig == nil {
+			return nil, errors.New("priceGetterConfig is nil")
+		}
+
+		// Configure contract readers for all chains specified in the aggregator configurations.
+		// Some lanes (e.g. Wemix/Kroma) requires other clients than source and destination, since they use feeds from other chains.
+		aggregatorChainsToContracts := make(map[uint64][]common.Address)
+		for _, aggCfg := range pluginJobSpecConfig.PriceGetterConfig.AggregatorPrices {
+			if _, ok := aggregatorChainsToContracts[aggCfg.ChainID]; !ok {
+				aggregatorChainsToContracts[aggCfg.ChainID] = make([]common.Address, 0)
+			}
+
+			aggregatorChainsToContracts[aggCfg.ChainID] = append(aggregatorChainsToContracts[aggCfg.ChainID], aggCfg.AggregatorContractAddress)
+		}
+
+		contractReaders := map[uint64]types.ContractReader{}
+
+		for chainID, aggregatorContracts := range aggregatorChainsToContracts {
+			relayID := types.RelayID{Network: spec.Relay, ChainID: strconv.FormatUint(chainID, 10)}
+			relay, rerr := d.RelayGetter.Get(relayID)
+			if rerr != nil {
+				return nil, fmt.Errorf("get relay by id=%v: %w", relayID, err)
+			}
+
+			contractsConfig := make(map[string]evmrelaytypes.ChainContractReader, len(aggregatorContracts))
+			for i := range aggregatorContracts {
+				contractsConfig[fmt.Sprintf("%v_%v", ccip.OffchainAggregator, i)] = evmrelaytypes.ChainContractReader{
+					ContractABI: ccip.OffChainAggregatorABI,
+					Configs: map[string]*evmrelaytypes.ChainReaderDefinition{
+						"decimals": { // CR consumers choose an alias
+							ChainSpecificName: "decimals",
+						},
+						"latestRoundData": {
+							ChainSpecificName: "latestRoundData",
+						},
+					},
+				}
+			}
+			contractReaderConfig := evmrelaytypes.ChainReaderConfig{
+				Contracts: contractsConfig,
+			}
+
+			contractReaderConfigJSONBytes, jerr := json.Marshal(contractReaderConfig)
+			if jerr != nil {
+				return nil, fmt.Errorf("marshal contract reader config: %w", jerr)
+			}
+
+			contractReader, cerr := relay.NewContractReader(ctx, contractReaderConfigJSONBytes)
+			if cerr != nil {
+				return nil, fmt.Errorf("new ccip commit contract reader %w", cerr)
+			}
+
+			contractReaders[chainID] = contractReader
+		}
+
+		priceGetter, err = ccip.NewDynamicPriceGetter(*pluginJobSpecConfig.PriceGetterConfig, contractReaders)
+		if err != nil {
+			return nil, fmt.Errorf("creating dynamic price getter: %w", err)
+		}
+	}
+	return priceGetter, nil
 }
 
 func newCCIPCommitPluginBytes(isSourceProvider bool, sourceStartBlock uint64, destStartBlock uint64) config.CommitPluginConfig {
@@ -1834,7 +1913,7 @@ func (d *Delegate) ccipExecGetDstProvider(ctx context.Context, jb job.Job, plugi
 
 	// PROVIDER BASED ARG CONSTRUCTION
 	// Write PluginConfig bytes to send source/dest relayer provider + info outside of top level rargs/pargs over the wire
-	dstConfigBytes, err := newExecPluginConfig(false, pluginJobSpecConfig.SourceStartBlock, pluginJobSpecConfig.DestStartBlock, pluginJobSpecConfig.USDCConfig, string(jb.ID)).Encode()
+	dstConfigBytes, err := newExecPluginConfig(false, pluginJobSpecConfig.SourceStartBlock, pluginJobSpecConfig.DestStartBlock, pluginJobSpecConfig.USDCConfig, pluginJobSpecConfig.LBTCConfig, string(jb.ID)).Encode()
 	if err != nil {
 		return nil, err
 	}
@@ -1867,7 +1946,7 @@ func (d *Delegate) ccipExecGetDstProvider(ctx context.Context, jb job.Job, plugi
 
 func (d *Delegate) ccipExecGetSrcProvider(ctx context.Context, jb job.Job, pluginJobSpecConfig ccipconfig.ExecPluginJobSpecConfig, transmitterID string, dstProvider types.CCIPExecProvider) (srcProvider types.CCIPExecProvider, srcChainID uint64, err error) {
 	spec := jb.OCR2OracleSpec
-	srcConfigBytes, err := newExecPluginConfig(true, pluginJobSpecConfig.SourceStartBlock, pluginJobSpecConfig.DestStartBlock, pluginJobSpecConfig.USDCConfig, string(jb.ID)).Encode()
+	srcConfigBytes, err := newExecPluginConfig(true, pluginJobSpecConfig.SourceStartBlock, pluginJobSpecConfig.DestStartBlock, pluginJobSpecConfig.USDCConfig, pluginJobSpecConfig.LBTCConfig, string(jb.ID)).Encode()
 	if err != nil {
 		return nil, 0, err
 	}
@@ -1916,12 +1995,13 @@ func (d *Delegate) ccipExecGetSrcProvider(ctx context.Context, jb job.Job, plugi
 	return
 }
 
-func newExecPluginConfig(isSourceProvider bool, srcStartBlock uint64, dstStartBlock uint64, usdcConfig ccipconfig.USDCConfig, jobID string) config.ExecPluginConfig {
+func newExecPluginConfig(isSourceProvider bool, srcStartBlock uint64, dstStartBlock uint64, usdcConfig ccipconfig.USDCConfig, lbtcConfig ccipconfig.LBTCConfig, jobID string) config.ExecPluginConfig {
 	return config.ExecPluginConfig{
 		IsSourceProvider: isSourceProvider,
 		SourceStartBlock: srcStartBlock,
 		DestStartBlock:   dstStartBlock,
 		USDCConfig:       usdcConfig,
+		LBTCConfig:       lbtcConfig,
 		JobID:            jobID,
 	}
 }

core/services/ocr2/plugins/ccip/ccipcommit/factory.go

@@ -74,8 +74,11 @@ type reportingPluginAndInfo struct {
 func (rf *CommitReportingPluginFactory) NewReportingPlugin(ctx context.Context, config types.ReportingPluginConfig) (types.ReportingPlugin, types.ReportingPluginInfo, error) {
 	initialRetryDelay := rf.config.newReportingPluginRetryConfig.InitialDelay
 	maxDelay := rf.config.newReportingPluginRetryConfig.MaxDelay
+	maxRetries := rf.config.newReportingPluginRetryConfig.MaxRetries
 
-	pluginAndInfo, err := ccipcommon.RetryUntilSuccess(rf.NewReportingPluginFn(ctx, config), initialRetryDelay, maxDelay)
+	pluginAndInfo, err := ccipcommon.RetryUntilSuccess(
+		rf.NewReportingPluginFn(ctx, config), initialRetryDelay, maxDelay, maxRetries,
+	)
 	if err != nil {
 		return nil, types.ReportingPluginInfo{}, err
 	}
@@ -86,33 +89,33 @@ func (rf *CommitReportingPluginFactory) NewReportingPlugin(ctx context.Context,
 // retried via RetryUntilSuccess. NewReportingPlugin must return successfully in order for the Commit plugin to
 // function, hence why we can only keep retrying it until it succeeds.
 func (rf *CommitReportingPluginFactory) NewReportingPluginFn(ctx context.Context, config types.ReportingPluginConfig) func() (reportingPluginAndInfo, error) {
-	return func() (reportingPluginAndInfo, error) {
+	newReportingPluginFn := func() (reportingPluginAndInfo, error) {
 		destPriceReg, err := rf.config.commitStore.ChangeConfig(ctx, config.OnchainConfig, config.OffchainConfig)
 		if err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("commitStore.ChangeConfig error: %w", err)
 		}
 
 		priceRegEvmAddr, err := ccipcalc.GenericAddrToEvm(destPriceReg)
 		if err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("GenericAddrToEvm error: %w", err)
 		}
 		if err = rf.UpdateDynamicReaders(ctx, priceRegEvmAddr); err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("UpdateDynamicReaders error: %w", err)
 		}
 
 		pluginOffChainConfig, err := rf.config.commitStore.OffchainConfig(ctx)
 		if err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("commitStore.OffchainConfig error: %w", err)
 		}
 
 		gasPriceEstimator, err := rf.config.commitStore.GasPriceEstimator(ctx)
 		if err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("commitStore.GasPriceEstimator error: %w", err)
 		}
 
 		err = rf.config.priceService.UpdateDynamicConfig(ctx, gasPriceEstimator, rf.destPriceRegReader)
 		if err != nil {
-			return reportingPluginAndInfo{}, err
+			return reportingPluginAndInfo{}, fmt.Errorf("priceService.UpdateDynamicConfig error: %w", err)
 		}
 
 		lggr := rf.config.lggr.Named("CommitReportingPlugin")
@@ -145,4 +148,14 @@ func (rf *CommitReportingPluginFactory) NewReportingPluginFn(ctx context.Context
 
 		return reportingPluginAndInfo{plugin, pluginInfo}, nil
 	}
+
+	return func() (reportingPluginAndInfo, error) {
+		result, err := newReportingPluginFn()
+		if err != nil {
+			rf.config.lggr.Errorw("NewReportingPlugin failed", "err", err)
+			rf.config.metricsCollector.NewReportingPluginError()
+		}
+
+		return result, err
+	}
 }

core/services/ocr2/plugins/ccip/ccipcommit/factory_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	ccip2 "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata"
 	ccipdataprovidermocks "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/ccipdataprovider/mocks"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/mocks"
@@ -28,6 +29,8 @@ import (
 func TestNewReportingPluginRetriesUntilSuccess(t *testing.T) {
 	ctx := tests.Context(t)
 	commitConfig := CommitPluginStaticConfig{}
+	commitConfig.lggr = logger.TestLogger(t)
+	commitConfig.metricsCollector = ccip2.NoopMetricsCollector
 
 	// For this unit test, ensure that there is no delay between retries
 	commitConfig.newReportingPluginRetryConfig = ccipdata.RetryConfig{