diff --git a/GNUmakefile b/GNUmakefile index 55d2de618e..dfe6d80ccb 100644 --- a/GNUmakefile +++ b/GNUmakefile @@ -68,6 +68,10 @@ install-medianpoc: ## Build & install the chainlink-medianpoc binary. install-ocr3-capability: ## Build & install the chainlink-ocr3-capability binary. go install $(GOFLAGS) ./plugins/cmd/chainlink-ocr3-capability +.PHONY: install-ccip-exec +install-ccip-exec: + go install $(GOFLAGS) ./plugins/cmd/chainlink-ccip-exec + .PHONY: docker ## Build the chainlink docker image docker: docker buildx build \ diff --git a/core/config/env/env.go b/core/config/env/env.go index 0ebfc357bf..3f846da258 100644 --- a/core/config/env/env.go +++ b/core/config/env/env.go @@ -28,6 +28,7 @@ var ( MercuryPlugin = NewPlugin("mercury") SolanaPlugin = NewPlugin("solana") StarknetPlugin = NewPlugin("starknet") + CCIPExecPlugin = NewPlugin("ccip-exec") // PrometheusDiscoveryHostName is the externally accessible hostname // published by the node in the `/discovery` endpoint. Generally, it is expected to match // the public hostname of node. diff --git a/core/scripts/go.mod b/core/scripts/go.mod index b8cddfbb5a..4c46a1c0c7 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -24,7 +24,7 @@ require ( github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chain-selectors v1.0.23 github.com/smartcontractkit/chainlink-automation v1.0.4 - github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 + github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000 github.com/smartcontractkit/libocr v0.0.0-20240717100443-f6226e09bee7 github.com/spf13/cobra v1.8.0 diff --git a/core/scripts/go.sum b/core/scripts/go.sum index db4c3cd316..2a415c9b2a 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1074,8 +1074,8 @@ github.com/smartcontractkit/chain-selectors v1.0.23 h1:D2Eaex4Cw/O7Lg3tX6WklOqnj github.com/smartcontractkit/chain-selectors v1.0.23/go.mod h1:d4Hi+E1zqjy9HqMkjBE5q1vcG9VGgxf5VxiRHfzi2kE= github.com/smartcontractkit/chainlink-automation v1.0.4 h1:iyW181JjKHLNMnDleI8umfIfVVlwC7+n5izbLSFgjw8= github.com/smartcontractkit/chainlink-automation v1.0.4/go.mod h1:u4NbPZKJ5XiayfKHD/v3z3iflQWqvtdhj13jVZXj/cM= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 h1:pdEpjgbZ5w/Sd5lzg/XiuC5gVyrmSovOo+3nUD46SP8= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b h1:Dak5bYlqcDh3kUbmNZaV5ib+MykYuzAgPkWmtqkPj7U= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45 h1:NBQLtqk8zsyY4qTJs+NElI3aDFTcAo83JHvqD04EvB0= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45/go.mod h1:LV0h7QBQUpoC2UUi6TcUvcIFm1xjP/DtEcqV8+qeLUs= github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240718160222-2dc0c8136bfa h1:g75H8oh2ws52s8BekwvGQ9XvBVu3E7WM1rfiA0PN0zk= diff --git a/core/services/ocr2/delegate.go b/core/services/ocr2/delegate.go index 7c99083a97..6a06ed6fab 100644 --- a/core/services/ocr2/delegate.go +++ b/core/services/ocr2/delegate.go @@ -11,19 +11,19 @@ import ( "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/config" - "github.com/ethereum/go-ethereum/common" "github.com/pkg/errors" "github.com/prometheus/client_golang/prometheus" + "google.golang.org/grpc" + chainselectors "github.com/smartcontractkit/chain-selectors" + 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/config" "github.com/smartcontractkit/libocr/commontypes" libocr2 "github.com/smartcontractkit/libocr/offchainreporting2plus" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" - "google.golang.org/grpc" ocr2keepers20 "github.com/smartcontractkit/chainlink-automation/pkg/v2" ocr2keepers20config "github.com/smartcontractkit/chainlink-automation/pkg/v2/config" @@ -41,7 +41,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" @@ -52,8 +51,10 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/keystore/chaintype" "github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/ocr2key" "github.com/smartcontractkit/chainlink/v2/core/services/llo" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/ccipcommit" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/ccipexec" + ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/functions" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/generic" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/liquiditymanager" @@ -63,7 +64,6 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/median" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/mercury" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/autotelemetry21" ocr2keeper21core "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/core" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/validate" @@ -79,8 +79,6 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/synchronization" "github.com/smartcontractkit/chainlink/v2/core/services/telemetry" "github.com/smartcontractkit/chainlink/v2/plugins" - - ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" ) type ErrJobSpecNoRelayer struct { @@ -1730,12 +1728,29 @@ func (d *Delegate) ccipCommitGetSrcProvider(ctx context.Context, jb job.Job, plu } func (d *Delegate) newServicesCCIPExecution(ctx context.Context, lggr logger.SugaredLogger, jb job.Job, bootstrapPeers []commontypes.BootstrapperLocator, kb ocr2key.KeyBundle, ocrDB *db, lc ocrtypes.LocalConfig, transmitterID string) ([]job.ServiceCtx, error) { + if jb.OCR2OracleSpec == nil { + return nil, fmt.Errorf("spec is nil") + } spec := jb.OCR2OracleSpec if spec.Relay != relay.NetworkEVM { return nil, fmt.Errorf("non evm chains are not supported for CCIP execution") } - dstRid, err := spec.RelayID() + var pluginJobSpecConfig ccipconfig.ExecPluginJobSpecConfig + err := json.Unmarshal(spec.PluginConfig.Bytes(), &pluginJobSpecConfig) + if err != nil { + return nil, err + } + if pluginJobSpecConfig.USDCConfig.AttestationAPI != "" { + err = pluginJobSpecConfig.USDCConfig.ValidateUSDCConfig() + } + if err != nil { + return nil, err + } + + sourceTokenAddress := ccip.EvmAddrToGeneric(pluginJobSpecConfig.USDCConfig.SourceTokenAddress) + + dstRid, err := spec.RelayID() if err != nil { return nil, ErrJobSpecNoRelayer{Err: err, PluginName: string(spec.PluginType)} } @@ -1746,12 +1761,6 @@ func (d *Delegate) newServicesCCIPExecution(ctx context.Context, lggr logger.Sug // PROVIDER BASED ARG CONSTRUCTION // Write PluginConfig bytes to send source/dest relayer provider + info outside of top level rargs/pargs over the wire - var pluginJobSpecConfig ccipconfig.ExecPluginJobSpecConfig - err = json.Unmarshal(spec.PluginConfig.Bytes(), &pluginJobSpecConfig) - if err != nil { - return nil, err - } - dstChainID, err := strconv.ParseInt(dstRid.ChainID, 10, 64) if err != nil { return nil, err @@ -1786,7 +1795,7 @@ func (d *Delegate) newServicesCCIPExecution(ctx context.Context, lggr logger.Sug MetricsRegisterer: prometheus.WrapRegistererWith(map[string]string{"job_name": jb.Name.ValueOrZero()}, prometheus.DefaultRegisterer), } - return ccipexec.NewExecServices(ctx, lggr, jb, srcProvider, dstProvider, int64(srcChainID), dstChainID, d.isNewlyCreatedJob, oracleArgsNoPlugin2, logError) + return ccipexec.NewExecServices(ctx, lggr, d.cfg, jb, string(sourceTokenAddress), srcProvider, dstProvider, int64(srcChainID), dstChainID, d.isNewlyCreatedJob, oracleArgsNoPlugin2, logError) } func (d *Delegate) ccipExecGetDstProvider(ctx context.Context, jb job.Job, pluginJobSpecConfig ccipconfig.ExecPluginJobSpecConfig, transmitterID string) (types.CCIPExecProvider, error) { diff --git a/core/services/ocr2/plugins/ccip/ccipcommit/initializers.go b/core/services/ocr2/plugins/ccip/ccipcommit/initializers.go index c5dc8ffb45..e8a4302e18 100644 --- a/core/services/ocr2/plugins/ccip/ccipcommit/initializers.go +++ b/core/services/ocr2/plugins/ccip/ccipcommit/initializers.go @@ -8,17 +8,20 @@ import ( "strings" "time" + "github.com/Masterminds/semver/v3" + + commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/pricegetter" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/rpclib" - "github.com/Masterminds/semver/v3" "github.com/ethereum/go-ethereum/common" libocr2 "github.com/smartcontractkit/libocr/offchainreporting2plus" "go.uber.org/multierr" "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" - commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" cciporm "github.com/smartcontractkit/chainlink/v2/core/services/ccip" @@ -28,7 +31,6 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" "github.com/smartcontractkit/chainlink/v2/core/chains/legacyevm" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/job" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" diff --git a/core/services/ocr2/plugins/ccip/ccipexec/batching.go b/core/services/ocr2/plugins/ccip/ccipexec/batching.go index f096953f5d..548a7b9783 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/batching.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/batching.go @@ -15,7 +15,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/types" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/abihelpers" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata" @@ -34,7 +34,7 @@ type BatchContext struct { report commitReportWithSendRequests inflight []InflightInternalExecutionReport inflightAggregateValue *big.Int - lggr logger.Logger + lggr logger.SugaredLogger availableDataLen int availableGas uint64 expectedNonces map[cciptypes.Address]uint64 diff --git a/core/services/ocr2/plugins/ccip/ccipexec/batching_test.go b/core/services/ocr2/plugins/ccip/ccipexec/batching_test.go index 4393fc5f58..43bcb84d50 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/batching_test.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/batching_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/binary" + "math" "math/big" "reflect" @@ -14,10 +15,10 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/types" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/prices" @@ -802,7 +803,7 @@ func runBatchingStrategyTests(t *testing.T, strategy BatchingStrategy, available for _, tc := range testCases { tc := tc t.Run(tc.name, func(t *testing.T) { - lggr := logger.TestLogger(t) + lggr := commonlogger.TestSugared(t) gasPriceEstimator := prices.NewMockGasPriceEstimatorExec(t) if !tc.skipGasPriceEstimator { diff --git a/core/services/ocr2/plugins/ccip/ccipexec/factory.go b/core/services/ocr2/plugins/ccip/ccipexec/factory.go index 7e5cadc7ea..866b08a42c 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/factory.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/factory.go @@ -5,8 +5,17 @@ import ( "fmt" "sync" + "go.uber.org/multierr" + + "github.com/smartcontractkit/chainlink-common/pkg/services" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/observability" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/tokendata" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcommon" @@ -23,6 +32,161 @@ type ExecutionReportingPluginFactory struct { destPriceRegReader ccipdata.PriceRegistryReader destPriceRegAddr cciptypes.Address readersMu *sync.Mutex + lggr logger.Logger + services []services.Service +} + +func (rf *ExecutionReportingPluginFactory) Name() string { + return rf.lggr.Name() +} + +// Start is used to run chainHealthcheck and tokenDataWorker, which were previously passed +// back to the delegate as top level job.ServiceCtx to be managed in core alongside the reporting +// plugin factory +func (rf *ExecutionReportingPluginFactory) Start(ctx context.Context) (err error) { + rf.readersMu.Lock() + defer rf.readersMu.Unlock() + for _, service := range rf.services { + serviceErr := service.Start(ctx) + err = multierr.Append(err, serviceErr) + } + return +} + +func (rf *ExecutionReportingPluginFactory) Close() (err error) { + rf.readersMu.Lock() + defer rf.readersMu.Unlock() + for _, service := range rf.services { + closeErr := service.Close() + err = multierr.Append(err, closeErr) + } + + return +} + +func (rf *ExecutionReportingPluginFactory) Ready() error { + return nil +} + +func (rf *ExecutionReportingPluginFactory) HealthReport() map[string]error { + return make(map[string]error) +} + +func NewExecutionReportingPluginFactoryV2(ctx context.Context, lggr logger.Logger, sourceTokenAddress string, srcChainID int64, dstChainID int64, srcProvider commontypes.CCIPExecProvider, dstProvider commontypes.CCIPExecProvider) (*ExecutionReportingPluginFactory, error) { + // NewOffRampReader doesn't need addr param when provided in job spec + offRampReader, err := dstProvider.NewOffRampReader(ctx, "") + if err != nil { + return nil, fmt.Errorf("create offRampReader: %w", err) + } + + offRampConfig, err := offRampReader.GetStaticConfig(ctx) + if err != nil { + return nil, fmt.Errorf("get offRamp static config: %w", err) + } + + srcChainSelector := offRampConfig.SourceChainSelector + dstChainSelector := offRampConfig.ChainSelector + onRampReader, err := srcProvider.NewOnRampReader(ctx, offRampConfig.OnRamp, srcChainSelector, dstChainSelector) + if err != nil { + return nil, fmt.Errorf("create onRampReader: %w", err) + } + + dynamicOnRampConfig, err := onRampReader.GetDynamicConfig(ctx) + if err != nil { + return nil, fmt.Errorf("get onramp dynamic config: %w", err) + } + + sourceWrappedNative, err := srcProvider.SourceNativeToken(ctx, dynamicOnRampConfig.Router) + if err != nil { + return nil, fmt.Errorf("get source wrapped native token: %w", err) + } + + srcCommitStore, err := srcProvider.NewCommitStoreReader(ctx, offRampConfig.CommitStore) + if err != nil { + return nil, fmt.Errorf("could not create src commitStoreReader reader: %w", err) + } + + dstCommitStore, err := dstProvider.NewCommitStoreReader(ctx, offRampConfig.CommitStore) + if err != nil { + return nil, fmt.Errorf("could not create dst commitStoreReader reader: %w", err) + } + + var commitStoreReader ccipdata.CommitStoreReader + commitStoreReader = ccip.NewProviderProxyCommitStoreReader(srcCommitStore, dstCommitStore) + + tokenDataProviders := make(map[cciptypes.Address]tokendata.Reader) + // init usdc token data provider + usdcReader, err2 := srcProvider.NewTokenDataReader(ctx, "") + tokenDataProviders[cciptypes.Address(sourceTokenAddress)] = usdcReader + if err2 != nil { + // in order to not wire the attestation API through this factory, we wire it through the provider + // when the provider is created. In some cases the attestation API can be nil, which means we + // don't want any token data providers. This should not cause creating the job to fail, so we + // give an empty map and move on. + if err2.Error() != "empty USDC attestation API" { + return nil, fmt.Errorf("new usdc reader: %w", err2) + } + tokenDataProviders = make(map[cciptypes.Address]tokendata.Reader) + } + + // Prom wrappers + onRampReader = observability.NewObservedOnRampReader(onRampReader, srcChainID, ccip.ExecPluginLabel) + commitStoreReader = observability.NewObservedCommitStoreReader(commitStoreReader, dstChainID, ccip.ExecPluginLabel) + offRampReader = observability.NewObservedOffRampReader(offRampReader, dstChainID, ccip.ExecPluginLabel) + metricsCollector := ccip.NewPluginMetricsCollector(ccip.ExecPluginLabel, srcChainID, dstChainID) + + tokenPoolBatchedReader, err := dstProvider.NewTokenPoolBatchedReader(ctx, "", srcChainSelector) + if err != nil { + return nil, fmt.Errorf("new token pool batched reader: %w", err) + } + + chainHealthcheck := cache.NewObservedChainHealthCheck( + cache.NewChainHealthcheck( + // Adding more details to Logger to make healthcheck logs more informative + // It's safe because healthcheck logs only in case of unhealthy state + logger.With(lggr, "onramp", offRampConfig.OnRamp, + "commitStore", offRampConfig.CommitStore, + ), + onRampReader, + commitStoreReader, + ), + ccip.ExecPluginLabel, + srcChainID, + dstChainID, + offRampConfig.OnRamp, + ) + + tokenBackgroundWorker := tokendata.NewBackgroundWorker( + tokenDataProviders, + tokenDataWorkerNumWorkers, + tokenDataWorkerTimeout, + 2*tokenDataWorkerTimeout, + ) + + return &ExecutionReportingPluginFactory{ + config: ExecutionPluginStaticConfig{ + lggr: logger.Sugared(lggr), + onRampReader: onRampReader, + commitStoreReader: commitStoreReader, + offRampReader: offRampReader, + sourcePriceRegistryProvider: ccip.NewChainAgnosticPriceRegistry(srcProvider), + sourceWrappedNativeToken: sourceWrappedNative, + destChainSelector: dstChainSelector, + priceRegistryProvider: ccip.NewChainAgnosticPriceRegistry(dstProvider), + tokenPoolBatchedReader: tokenPoolBatchedReader, + tokenDataWorker: tokenBackgroundWorker, + metricsCollector: metricsCollector, + chainHealthcheck: chainHealthcheck, + newReportingPluginRetryConfig: defaultNewReportingPluginRetryConfig, + }, + services: []services.Service{chainHealthcheck, tokenBackgroundWorker}, + readersMu: &sync.Mutex{}, + lggr: logger.Sugared(lggr), + + // the fields below are initially empty and populated on demand + destPriceRegReader: nil, + destPriceRegAddr: "", + }, nil } func NewExecutionReportingPluginFactory(config ExecutionPluginStaticConfig) *ExecutionReportingPluginFactory { @@ -36,7 +200,7 @@ func NewExecutionReportingPluginFactory(config ExecutionPluginStaticConfig) *Exe } } -func (rf *ExecutionReportingPluginFactory) UpdateDynamicReaders(ctx context.Context, newPriceRegAddr cciptypes.Address) error { +func (rf *ExecutionReportingPluginFactory) UpdateDynamicReaders(_ context.Context, newPriceRegAddr cciptypes.Address) error { rf.readersMu.Lock() defer rf.readersMu.Unlock() // TODO: Investigate use of Close() to cleanup. @@ -89,6 +253,14 @@ func (rf *ExecutionReportingPluginFactory) NewReportingPluginFn(config types.Rep newReportingPluginFn := func() (reportingPluginAndInfo, error) { ctx := context.Background() // todo: consider setting a timeout + // Start the chainHealthcheck and tokenDataWorker services + // Using Start, while a bit more obtuse, allows us to manage these services + // in the same process as the plugin factory in LOOP mode + err := rf.Start(ctx) + if err != nil { + return reportingPluginAndInfo{}, err + } + destPriceRegistry, destWrappedNative, err := rf.config.offRampReader.ChangeConfig(ctx, config.OnchainConfig, config.OffchainConfig) if err != nil { return reportingPluginAndInfo{}, err diff --git a/core/services/ocr2/plugins/ccip/ccipexec/factory_test.go b/core/services/ocr2/plugins/ccip/ccipexec/factory_test.go index 869aa7e332..8aedf16ed3 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/factory_test.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/factory_test.go @@ -9,8 +9,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "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" @@ -24,7 +24,7 @@ import ( // retries a sufficient number of times to get through the transient errors and eventually succeed. func TestNewReportingPluginRetriesUntilSuccess(t *testing.T) { execConfig := ExecutionPluginStaticConfig{} - execConfig.lggr = logger.TestLogger(t) + execConfig.lggr = logger.TestSugared(t) execConfig.metricsCollector = ccip2.NoopMetricsCollector // For this unit test, ensure that there is no delay between retries @@ -57,7 +57,7 @@ func TestNewReportingPluginRetriesUntilSuccess(t *testing.T) { priceRegistryProvider.On("NewPriceRegistryReader", mock.Anything, mock.Anything).Return(nil, nil).Once() execConfig.priceRegistryProvider = priceRegistryProvider - execConfig.lggr, _ = logger.NewLogger() + execConfig.lggr = logger.TestSugared(t) factory := NewExecutionReportingPluginFactory(execConfig) reportingConfig := types.ReportingPluginConfig{} diff --git a/core/services/ocr2/plugins/ccip/ccipexec/inflight.go b/core/services/ocr2/plugins/ccip/ccipexec/inflight.go index c76bfdf778..fd6fb86cff 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/inflight.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/inflight.go @@ -6,8 +6,8 @@ import ( "github.com/pkg/errors" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/logger" ) // InflightInternalExecutionReport serves the same purpose as InflightCommitReport diff --git a/core/services/ocr2/plugins/ccip/ccipexec/initializers.go b/core/services/ocr2/plugins/ccip/ccipexec/initializers.go index aa42ff2828..9b5f74cfb1 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/initializers.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/initializers.go @@ -2,36 +2,33 @@ package ccipexec import ( "context" - "encoding/json" "fmt" "math/big" "time" + "github.com/Masterminds/semver/v3" + + "github.com/smartcontractkit/chainlink/v2/plugins" + + "github.com/smartcontractkit/chainlink-common/pkg/loop" + + "github.com/smartcontractkit/chainlink/v2/core/config/env" + "github.com/smartcontractkit/chainlink-common/pkg/types" - "github.com/Masterminds/semver/v3" "go.uber.org/multierr" libocr2 "github.com/smartcontractkit/libocr/offchainreporting2plus" commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" - cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - - "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/statuschecker" - "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/job" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/cache" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/factory" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/observability" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/oraclelib" - "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/tokendata" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/promwrapper" ) @@ -54,128 +51,39 @@ var defaultNewReportingPluginRetryConfig = ccipdata.RetryConfig{ MaxRetries: (6 * 4) + 10, } -func NewExecServices(ctx context.Context, lggr logger.Logger, jb job.Job, srcProvider types.CCIPExecProvider, dstProvider types.CCIPExecProvider, srcChainID int64, dstChainID int64, new bool, argsNoPlugin libocr2.OCR2OracleArgs, logError func(string)) ([]job.ServiceCtx, error) { - if jb.OCR2OracleSpec == nil { - return nil, fmt.Errorf("spec is nil") - } - spec := jb.OCR2OracleSpec - var pluginConfig ccipconfig.ExecPluginJobSpecConfig - err := json.Unmarshal(spec.PluginConfig.Bytes(), &pluginConfig) - if err != nil { - return nil, err - } - - offRampAddress := ccipcalc.HexToAddress(spec.ContractID) - offRampReader, err := dstProvider.NewOffRampReader(ctx, offRampAddress) - if err != nil { - return nil, fmt.Errorf("create offRampReader: %w", err) - } +func NewExecServices(ctx context.Context, lggr logger.Logger, cfg plugins.RegistrarConfig, jb job.Job, sourceTokenAddress string, srcProvider types.CCIPExecProvider, dstProvider types.CCIPExecProvider, srcChainID int64, dstChainID int64, new bool, argsNoPlugin libocr2.OCR2OracleArgs, logError func(string)) ([]job.ServiceCtx, error) { + lggr = lggr.Named("ccip-exec").Named(string(jb.ID)) - offRampConfig, err := offRampReader.GetStaticConfig(ctx) - if err != nil { - return nil, fmt.Errorf("get offRamp static config: %w", err) - } + loopCmd := env.CCIPExecPlugin.Cmd.Get() + loopEnabled := loopCmd != "" - srcChainSelector := offRampConfig.SourceChainSelector - dstChainSelector := offRampConfig.ChainSelector - onRampReader, err := srcProvider.NewOnRampReader(ctx, offRampConfig.OnRamp, srcChainSelector, dstChainSelector) - if err != nil { - return nil, fmt.Errorf("create onRampReader: %w", err) - } - - dynamicOnRampConfig, err := onRampReader.GetDynamicConfig(ctx) - if err != nil { - return nil, fmt.Errorf("get onramp dynamic config: %w", err) - } - - sourceWrappedNative, err := srcProvider.SourceNativeToken(ctx, dynamicOnRampConfig.Router) - if err != nil { - return nil, fmt.Errorf("get source wrapped native token: %w", err) - } - - srcCommitStore, err := srcProvider.NewCommitStoreReader(ctx, offRampConfig.CommitStore) - if err != nil { - return nil, fmt.Errorf("could not create src commitStoreReader reader: %w", err) - } - - dstCommitStore, err := dstProvider.NewCommitStoreReader(ctx, offRampConfig.CommitStore) - if err != nil { - return nil, fmt.Errorf("could not create dst commitStoreReader reader: %w", err) - } - - var commitStoreReader ccipdata.CommitStoreReader - commitStoreReader = ccip.NewProviderProxyCommitStoreReader(srcCommitStore, dstCommitStore) - - tokenDataProviders := make(map[cciptypes.Address]tokendata.Reader) - // init usdc token data provider - if pluginConfig.USDCConfig.AttestationAPI != "" { - lggr.Infof("USDC token data provider enabled") - err2 := pluginConfig.USDCConfig.ValidateUSDCConfig() - if err2 != nil { - return nil, err2 + var pluginFactory types.ReportingPluginFactory + if loopEnabled { + // find loop command + envVars, err := plugins.ParseEnvFile(env.CCIPExecPlugin.Env.Get()) + if err != nil { + return nil, fmt.Errorf("failed to parse ccip exec env file: %w", err) } - - usdcReader, err2 := srcProvider.NewTokenDataReader(ctx, ccip.EvmAddrToGeneric(pluginConfig.USDCConfig.SourceTokenAddress)) + cmdFn, grpcOpts, err := cfg.RegisterLOOP(plugins.CmdConfig{ + ID: lggr.Name(), + Cmd: loopCmd, + Env: envVars, + }) + if err != nil { + return nil, fmt.Errorf("failed to register ccip exec plugin: %w", err) + } + // get reporting plugin factory from loop + factoryServer := loop.NewExecutionService(lggr, grpcOpts, cmdFn, srcProvider, dstProvider, uint32(srcChainID), uint32(dstChainID), sourceTokenAddress) + pluginFactory = factoryServer + } else { + var err2 error + pluginFactory, err2 = NewExecutionReportingPluginFactoryV2(ctx, lggr, sourceTokenAddress, srcChainID, dstChainID, srcProvider, dstProvider) if err2 != nil { - return nil, fmt.Errorf("new usdc reader: %w", err2) + return nil, err2 } - tokenDataProviders[cciptypes.Address(pluginConfig.USDCConfig.SourceTokenAddress.String())] = usdcReader } - // Prom wrappers - onRampReader = observability.NewObservedOnRampReader(onRampReader, srcChainID, ccip.ExecPluginLabel) - commitStoreReader = observability.NewObservedCommitStoreReader(commitStoreReader, dstChainID, ccip.ExecPluginLabel) - offRampReader = observability.NewObservedOffRampReader(offRampReader, dstChainID, ccip.ExecPluginLabel) - metricsCollector := ccip.NewPluginMetricsCollector(ccip.ExecPluginLabel, srcChainID, dstChainID) - - tokenPoolBatchedReader, err := dstProvider.NewTokenPoolBatchedReader(ctx, offRampAddress, srcChainSelector) - if err != nil { - return nil, fmt.Errorf("new token pool batched reader: %w", err) - } - - chainHealthcheck := cache.NewObservedChainHealthCheck( - cache.NewChainHealthcheck( - // Adding more details to Logger to make healthcheck logs more informative - // It's safe because healthcheck logs only in case of unhealthy state - lggr.With( - "onramp", offRampConfig.OnRamp, - "commitStore", offRampConfig.CommitStore, - "offramp", offRampAddress, - ), - onRampReader, - commitStoreReader, - ), - ccip.ExecPluginLabel, - srcChainID, - dstChainID, - offRampConfig.OnRamp, - ) - - tokenBackgroundWorker := tokendata.NewBackgroundWorker( - tokenDataProviders, - tokenDataWorkerNumWorkers, - tokenDataWorkerTimeout, - 2*tokenDataWorkerTimeout, - ) - - wrappedPluginFactory := NewExecutionReportingPluginFactory(ExecutionPluginStaticConfig{ - lggr: lggr, - onRampReader: onRampReader, - commitStoreReader: commitStoreReader, - offRampReader: offRampReader, - sourcePriceRegistryProvider: ccip.NewChainAgnosticPriceRegistry(srcProvider), - sourceWrappedNativeToken: sourceWrappedNative, - destChainSelector: dstChainSelector, - priceRegistryProvider: ccip.NewChainAgnosticPriceRegistry(dstProvider), - tokenPoolBatchedReader: tokenPoolBatchedReader, - tokenDataWorker: tokenBackgroundWorker, - metricsCollector: metricsCollector, - chainHealthcheck: chainHealthcheck, - newReportingPluginRetryConfig: defaultNewReportingPluginRetryConfig, - txmStatusChecker: statuschecker.NewTxmStatusChecker(dstProvider.GetTransactionStatus), - }) - - argsNoPlugin.ReportingPluginFactory = promwrapper.NewPromFactory(wrappedPluginFactory, "CCIPExecution", jb.OCR2OracleSpec.Relay, big.NewInt(0).SetInt64(dstChainID)) + argsNoPlugin.ReportingPluginFactory = promwrapper.NewPromFactory(pluginFactory, "CCIPExecution", jb.OCR2OracleSpec.Relay, big.NewInt(0).SetInt64(dstChainID)) argsNoPlugin.Logger = commonlogger.NewOCRWrapper(lggr, true, logError) oracle, err := libocr2.NewOracle(argsNoPlugin) if err != nil { @@ -190,14 +98,10 @@ func NewExecServices(ctx context.Context, lggr logger.Logger, jb job.Job, srcPro dstProvider, job.NewServiceAdapter(oracle), ), - chainHealthcheck, - tokenBackgroundWorker, }, nil } return []job.ServiceCtx{ job.NewServiceAdapter(oracle), - chainHealthcheck, - tokenBackgroundWorker, }, nil } diff --git a/core/services/ocr2/plugins/ccip/ccipexec/ocr2.go b/core/services/ocr2/plugins/ccip/ccipexec/ocr2.go index 2c70cac497..70a380d199 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/ocr2.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/ocr2.go @@ -16,8 +16,8 @@ import ( "github.com/smartcontractkit/libocr/offchainreporting2plus/types" "github.com/smartcontractkit/chainlink-common/pkg/hashutil" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/cache" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcommon" @@ -50,7 +50,7 @@ var ( ) type ExecutionPluginStaticConfig struct { - lggr logger.Logger + lggr logger.SugaredLogger onRampReader ccipdata.OnRampReader offRampReader ccipdata.OffRampReader commitStoreReader ccipdata.CommitStoreReader @@ -69,7 +69,7 @@ type ExecutionPluginStaticConfig struct { type ExecutionReportingPlugin struct { // Misc F int - lggr logger.Logger + lggr logger.SugaredLogger offchainConfig cciptypes.ExecOffchainConfig tokenDataWorker tokendata.Worker metricsCollector ccip.PluginMetricsCollector @@ -101,7 +101,7 @@ func (r *ExecutionReportingPlugin) Query(context.Context, types.ReportTimestamp) return types.Query{}, nil } -func (r *ExecutionReportingPlugin) Observation(ctx context.Context, timestamp types.ReportTimestamp, query types.Query) (types.Observation, error) { +func (r *ExecutionReportingPlugin) Observation(ctx context.Context, _ types.ReportTimestamp, _ types.Query) (types.Observation, error) { lggr := r.lggr.Named("ExecutionObservation") if healthy, err := r.chainHealthcheck.IsHealthy(ctx); err != nil { return nil, err @@ -142,7 +142,7 @@ func (r *ExecutionReportingPlugin) Observation(ctx context.Context, timestamp ty return ccip.NewExecutionObservation(executableObservations).Marshal() } -func (r *ExecutionReportingPlugin) getExecutableObservations(ctx context.Context, lggr logger.Logger, inflight []InflightInternalExecutionReport) ([]ccip.ObservedMessage, error) { +func (r *ExecutionReportingPlugin) getExecutableObservations(ctx context.Context, lggr logger.SugaredLogger, inflight []InflightInternalExecutionReport) ([]ccip.ObservedMessage, error) { unexpiredReports, err := r.commitRootsCache.RootsEligibleForExecution(ctx) if err != nil { return nil, err @@ -256,7 +256,7 @@ func (r *ExecutionReportingPlugin) getExecutedSeqNrsInRange(ctx context.Context, func (r *ExecutionReportingPlugin) buildBatch( ctx context.Context, inflight []InflightInternalExecutionReport, - lggr logger.Logger, + lggr logger.SugaredLogger, report commitReportWithSendRequests, aggregateTokenLimit *big.Int, sourceTokenPricesUSD map[cciptypes.Address]*big.Int, @@ -573,7 +573,7 @@ func calculateObservedMessagesConsensus(observations []ccip.ExecutionObservation return finalSequenceNumbers, nil } -func (r *ExecutionReportingPlugin) ShouldAcceptFinalizedReport(ctx context.Context, timestamp types.ReportTimestamp, report types.Report) (bool, error) { +func (r *ExecutionReportingPlugin) ShouldAcceptFinalizedReport(ctx context.Context, _ types.ReportTimestamp, report types.Report) (bool, error) { lggr := r.lggr.Named("ShouldAcceptFinalizedReport") execReport, err := r.offRampReader.DecodeExecutionReport(ctx, report) if err != nil { @@ -607,7 +607,7 @@ func (r *ExecutionReportingPlugin) ShouldAcceptFinalizedReport(ctx context.Conte return true, nil } -func (r *ExecutionReportingPlugin) ShouldTransmitAcceptedReport(ctx context.Context, timestamp types.ReportTimestamp, report types.Report) (bool, error) { +func (r *ExecutionReportingPlugin) ShouldTransmitAcceptedReport(ctx context.Context, _ types.ReportTimestamp, report types.Report) (bool, error) { lggr := r.lggr.Named("ShouldTransmitAcceptedReport") execReport, err := r.offRampReader.DecodeExecutionReport(ctx, report) if err != nil { diff --git a/core/services/ocr2/plugins/ccip/ccipexec/ocr2_test.go b/core/services/ocr2/plugins/ccip/ccipexec/ocr2_test.go index fea05ab61c..862204bb50 100644 --- a/core/services/ocr2/plugins/ccip/ccipexec/ocr2_test.go +++ b/core/services/ocr2/plugins/ccip/ccipexec/ocr2_test.go @@ -20,11 +20,12 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" + cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/libocr/commontypes" "github.com/smartcontractkit/libocr/offchainreporting2/types" ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" - cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" lpMocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller/mocks" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" @@ -137,7 +138,7 @@ func TestExecutionReportingPlugin_Observation(t *testing.T) { p := &ExecutionReportingPlugin{} p.inflightReports = newInflightExecReportsContainer(time.Minute) p.inflightReports.reports = tc.inflightReports - p.lggr = logger.TestLogger(t) + p.lggr = commonlogger.TestSugared(t) p.tokenDataWorker = tokendata.NewBackgroundWorker( make(map[cciptypes.Address]tokendata.Reader), 10, 5*time.Second, time.Hour) p.metricsCollector = ccip.NoopMetricsCollector @@ -284,7 +285,7 @@ func TestExecutionReportingPlugin_Report(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { p := ExecutionReportingPlugin{} - p.lggr = logger.TestLogger(t) + p.lggr = commonlogger.TestSugared(t) p.F = tc.f bs, err := NewBatchingStrategy(tc.batchingStrategyId, &statuschecker.TxmStatusChecker{}) assert.NoError(t, err) @@ -342,7 +343,7 @@ func TestExecutionReportingPlugin_ShouldAcceptFinalizedReport(t *testing.T) { plugin := ExecutionReportingPlugin{ offRampReader: mockOffRampReader, - lggr: logger.TestLogger(t), + lggr: commonlogger.TestSugared(t), inflightReports: newInflightExecReportsContainer(1 * time.Hour), chainHealthcheck: chainHealthcheck, metricsCollector: ccip.NoopMetricsCollector, @@ -394,7 +395,7 @@ func TestExecutionReportingPlugin_ShouldTransmitAcceptedReport(t *testing.T) { plugin := ExecutionReportingPlugin{ commitStoreReader: mockCommitStoreReader, offRampReader: mockOffRampReader, - lggr: logger.TestLogger(t), + lggr: commonlogger.TestSugared(t), inflightReports: newInflightExecReportsContainer(1 * time.Hour), chainHealthcheck: chainHealthcheck, } @@ -428,7 +429,7 @@ func TestExecutionReportingPlugin_buildReport(t *testing.T) { // ensure that buildReport should cap the built report to fit in MaxExecutionReportLength p := &ExecutionReportingPlugin{} - p.lggr = logger.TestLogger(t) + p.lggr = commonlogger.TestSugared(t) commitStore := ccipdatamocks.NewCommitStoreReader(t) commitStore.On("VerifyExecutionReport", mock.Anything, mock.Anything, mock.Anything).Return(true, nil) @@ -559,7 +560,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) { } ctx := testutils.Context(t) - lggr := logger.TestLogger(t) + lggr := commonlogger.TestSugared(t) for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { p := &ExecutionReportingPlugin{} @@ -1411,7 +1412,7 @@ func encodeExecutionReport(t *testing.T, report cciptypes.ExecReport) []byte { // Verify the price registry update mechanism in case of configuration change on the source onRamp. func TestExecutionReportingPlugin_ensurePriceRegistrySynchronization(t *testing.T) { p := &ExecutionReportingPlugin{} - p.lggr = logger.TestLogger(t) + p.lggr = commonlogger.TestSugared(t) p.sourcePriceRegistryLock = sync.RWMutex{} sourcePriceRegistryAddress1 := cciptypes.Address(utils.RandomAddress().String()) diff --git a/core/services/ocr2/plugins/ccip/internal/cache/chain_health.go b/core/services/ocr2/plugins/ccip/internal/cache/chain_health.go index 00f90615eb..bc85c994d4 100644 --- a/core/services/ocr2/plugins/ccip/internal/cache/chain_health.go +++ b/core/services/ocr2/plugins/ccip/internal/cache/chain_health.go @@ -9,10 +9,9 @@ import ( "github.com/pkg/errors" "golang.org/x/sync/errgroup" + "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/services" - "github.com/smartcontractkit/chainlink/v2/core/logger" - "github.com/smartcontractkit/chainlink/v2/core/services/job" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata" ) @@ -31,7 +30,7 @@ import ( // // Additionally, to reduce the number of calls to the RPC, we refresh RMN state in the background based on defaultRMNStateRefreshInterval type ChainHealthcheck interface { - job.ServiceCtx + services.Service IsHealthy(ctx context.Context) (bool, error) } @@ -52,7 +51,7 @@ type chainHealthcheck struct { globalStatusExpiration time.Duration rmnStatusRefreshInterval time.Duration - lggr logger.Logger + lggr logger.SugaredLogger onRamp ccipdata.OnRampReader commitStore ccipdata.CommitStoreReader @@ -62,6 +61,14 @@ type chainHealthcheck struct { backgroundCancel context.CancelFunc } +func (c *chainHealthcheck) HealthReport() map[string]error { + return make(map[string]error) +} + +func (c *chainHealthcheck) Name() string { + return c.lggr.Name() +} + func NewChainHealthcheck(lggr logger.Logger, onRamp ccipdata.OnRampReader, commitStore ccipdata.CommitStoreReader) *chainHealthcheck { ctx, cancel := context.WithCancel(context.Background()) @@ -73,7 +80,7 @@ func NewChainHealthcheck(lggr logger.Logger, onRamp ccipdata.OnRampReader, commi globalStatusExpiration: defaultGlobalStatusExpirationDuration, rmnStatusRefreshInterval: defaultRMNStateRefreshInterval, - lggr: lggr, + lggr: logger.Sugared(lggr), onRamp: onRamp, commitStore: commitStore, @@ -95,7 +102,7 @@ func newChainHealthcheckWithCustomEviction(lggr logger.Logger, onRamp ccipdata.O globalStatusExpiration: globalStatusDuration, rmnStatusRefreshInterval: rmnStatusRefreshInterval, - lggr: lggr, + lggr: logger.Sugared(lggr), onRamp: onRamp, commitStore: commitStore, diff --git a/core/services/ocr2/plugins/ccip/internal/cache/commit_roots.go b/core/services/ocr2/plugins/ccip/internal/cache/commit_roots.go index 5f8bd5edc5..0fa5d4c2f7 100644 --- a/core/services/ocr2/plugins/ccip/internal/cache/commit_roots.go +++ b/core/services/ocr2/plugins/ccip/internal/cache/commit_roots.go @@ -7,11 +7,9 @@ import ( "time" "github.com/patrickmn/go-cache" - orderedmap "github.com/wk8/go-ordered-map/v2" - + "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - - "github.com/smartcontractkit/chainlink/v2/core/logger" + orderedmap "github.com/wk8/go-ordered-map/v2" ) const ( diff --git a/core/services/ocr2/plugins/ccip/internal/cache/mocks/chain_health_mock.go b/core/services/ocr2/plugins/ccip/internal/cache/mocks/chain_health_mock.go index 595b15774a..9c5df256e8 100644 --- a/core/services/ocr2/plugins/ccip/internal/cache/mocks/chain_health_mock.go +++ b/core/services/ocr2/plugins/ccip/internal/cache/mocks/chain_health_mock.go @@ -66,6 +66,53 @@ func (_c *ChainHealthcheck_Close_Call) RunAndReturn(run func() error) *ChainHeal return _c } +// HealthReport provides a mock function with given fields: +func (_m *ChainHealthcheck) HealthReport() map[string]error { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for HealthReport") + } + + var r0 map[string]error + if rf, ok := ret.Get(0).(func() map[string]error); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(map[string]error) + } + } + + return r0 +} + +// ChainHealthcheck_HealthReport_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'HealthReport' +type ChainHealthcheck_HealthReport_Call struct { + *mock.Call +} + +// HealthReport is a helper method to define mock.On call +func (_e *ChainHealthcheck_Expecter) HealthReport() *ChainHealthcheck_HealthReport_Call { + return &ChainHealthcheck_HealthReport_Call{Call: _e.mock.On("HealthReport")} +} + +func (_c *ChainHealthcheck_HealthReport_Call) Run(run func()) *ChainHealthcheck_HealthReport_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *ChainHealthcheck_HealthReport_Call) Return(_a0 map[string]error) *ChainHealthcheck_HealthReport_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *ChainHealthcheck_HealthReport_Call) RunAndReturn(run func() map[string]error) *ChainHealthcheck_HealthReport_Call { + _c.Call.Return(run) + return _c +} + // IsHealthy provides a mock function with given fields: ctx func (_m *ChainHealthcheck) IsHealthy(ctx context.Context) (bool, error) { ret := _m.Called(ctx) @@ -122,6 +169,96 @@ func (_c *ChainHealthcheck_IsHealthy_Call) RunAndReturn(run func(context.Context return _c } +// Name provides a mock function with given fields: +func (_m *ChainHealthcheck) Name() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Name") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + +// ChainHealthcheck_Name_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Name' +type ChainHealthcheck_Name_Call struct { + *mock.Call +} + +// Name is a helper method to define mock.On call +func (_e *ChainHealthcheck_Expecter) Name() *ChainHealthcheck_Name_Call { + return &ChainHealthcheck_Name_Call{Call: _e.mock.On("Name")} +} + +func (_c *ChainHealthcheck_Name_Call) Run(run func()) *ChainHealthcheck_Name_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *ChainHealthcheck_Name_Call) Return(_a0 string) *ChainHealthcheck_Name_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *ChainHealthcheck_Name_Call) RunAndReturn(run func() string) *ChainHealthcheck_Name_Call { + _c.Call.Return(run) + return _c +} + +// Ready provides a mock function with given fields: +func (_m *ChainHealthcheck) Ready() error { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Ready") + } + + var r0 error + if rf, ok := ret.Get(0).(func() error); ok { + r0 = rf() + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// ChainHealthcheck_Ready_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Ready' +type ChainHealthcheck_Ready_Call struct { + *mock.Call +} + +// Ready is a helper method to define mock.On call +func (_e *ChainHealthcheck_Expecter) Ready() *ChainHealthcheck_Ready_Call { + return &ChainHealthcheck_Ready_Call{Call: _e.mock.On("Ready")} +} + +func (_c *ChainHealthcheck_Ready_Call) Run(run func()) *ChainHealthcheck_Ready_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *ChainHealthcheck_Ready_Call) Return(_a0 error) *ChainHealthcheck_Ready_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *ChainHealthcheck_Ready_Call) RunAndReturn(run func() error) *ChainHealthcheck_Ready_Call { + _c.Call.Return(run) + return _c +} + // Start provides a mock function with given fields: _a0 func (_m *ChainHealthcheck) Start(_a0 context.Context) error { ret := _m.Called(_a0) diff --git a/core/services/ocr2/plugins/ccip/internal/cache/observed_chain_health.go b/core/services/ocr2/plugins/ccip/internal/cache/observed_chain_health.go index 941162448a..b88aacb06d 100644 --- a/core/services/ocr2/plugins/ccip/internal/cache/observed_chain_health.go +++ b/core/services/ocr2/plugins/ccip/internal/cache/observed_chain_health.go @@ -2,6 +2,7 @@ package cache import ( "context" + "fmt" "strconv" "github.com/prometheus/client_golang/prometheus" @@ -29,6 +30,18 @@ type ObservedChainHealthcheck struct { laneHealthStatus *prometheus.GaugeVec } +func (o *ObservedChainHealthcheck) Ready() error { + return nil +} + +func (o *ObservedChainHealthcheck) HealthReport() map[string]error { + return make(map[string]error) +} + +func (o *ObservedChainHealthcheck) Name() string { + return fmt.Sprintf("ObservedChainHealthcheck.%v.%v.%v", o.plugin, o.sourceChain, o.destChain) +} + func NewObservedChainHealthCheck( chainHealthcheck ChainHealthcheck, plugin string, diff --git a/core/services/ocr2/plugins/ccip/internal/ccipcalc/calc.go b/core/services/ocr2/plugins/ccip/internal/ccipcalc/calc.go index 8ba57e77ed..869e73b752 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipcalc/calc.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipcalc/calc.go @@ -4,7 +4,7 @@ import ( "math/big" "sort" - "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink-common/pkg/logger" ) // ContiguousReqs checks if seqNrs contains all numbers from min to max. diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/batchreader/token_pool_batch_reader.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/batchreader/token_pool_batch_reader.go index 57e8df1bde..47999fa47e 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/batchreader/token_pool_batch_reader.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/batchreader/token_pool_batch_reader.go @@ -10,10 +10,11 @@ import ( "github.com/ethereum/go-ethereum/common" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" type_and_version "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/type_and_version_interface_wrapper" - "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/abihelpers" ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/ccipdataprovider/provider.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/ccipdataprovider/provider.go index d1666d548a..ad7216d53c 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/ccipdataprovider/provider.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/ccipdataprovider/provider.go @@ -3,10 +3,12 @@ package ccipdataprovider import ( "context" + "github.com/smartcontractkit/chainlink/v2/core/logger" + cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" - "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/factory" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/observability" ) diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/onramp.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/onramp.go index e82584ac7c..4efee119cf 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/onramp.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/onramp.go @@ -4,7 +4,6 @@ import ( "github.com/pkg/errors" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" "github.com/smartcontractkit/chainlink/v2/core/logger" diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/price_registry.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/price_registry.go index f1fa7c4e81..f232e1ccf8 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/price_registry.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/factory/price_registry.go @@ -28,7 +28,7 @@ func ClosePriceRegistryReader(ctx context.Context, lggr logger.Logger, versionFi return err } -func initOrClosePriceRegistryReader(ctx context.Context, lggr logger.Logger, versionFinder VersionFinder, priceRegistryAddress cciptypes.Address, lp logpoller.LogPoller, cl client.Client, closeReader bool) (ccipdata.PriceRegistryReader, error) { +func initOrClosePriceRegistryReader(_ context.Context, lggr logger.Logger, versionFinder VersionFinder, priceRegistryAddress cciptypes.Address, lp logpoller.LogPoller, cl client.Client, closeReader bool) (ccipdata.PriceRegistryReader, error) { registerFilters := !closeReader priceRegistryEvmAddr, err := ccipcalc.GenericAddrToEvm(priceRegistryAddress) diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go index 3f57d419e1..8655bc98ac 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/reader.go @@ -6,10 +6,10 @@ import ( "github.com/ethereum/go-ethereum/core/types" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types" - "github.com/smartcontractkit/chainlink/v2/core/logger" ) const ( diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/price_registry.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/price_registry.go index d2104f985b..6080657a1e 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/price_registry.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/price_registry.go @@ -13,6 +13,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" @@ -20,7 +21,6 @@ import ( evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/price_registry_1_0_0" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/shared/generated/erc20" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/abihelpers" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/cache" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/commit_store.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/commit_store.go index ecc8acb576..b4ed11f267 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/commit_store.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/commit_store.go @@ -14,6 +14,7 @@ import ( "github.com/pkg/errors" "github.com/smartcontractkit/chainlink-common/pkg/config" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink-common/pkg/types/query" "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" @@ -23,7 +24,6 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/commit_store_1_2_0" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/abihelpers" ccipconfig "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/config" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/price_registry.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/price_registry.go index 9aac30e612..91d94b278d 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/price_registry.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/price_registry.go @@ -7,11 +7,11 @@ import ( "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/price_registry_1_2_0" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0" diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/commit_store.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/commit_store.go index d5545174cb..78c99091b6 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/commit_store.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/commit_store.go @@ -8,12 +8,12 @@ import ( "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/commit_store" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0" ) diff --git a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go index 5a9377858d..d8575eb440 100644 --- a/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go +++ b/core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go @@ -11,13 +11,12 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/smartcontractkit/chainlink-common/pkg/hashutil" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_onramp" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/rmn_contract" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/abihelpers" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/cache" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" diff --git a/core/services/ocr2/plugins/ccip/internal/oraclelib/backfilled_oracle.go b/core/services/ocr2/plugins/ccip/internal/oraclelib/backfilled_oracle.go index d2851e3a07..5e6133e12a 100644 --- a/core/services/ocr2/plugins/ccip/internal/oraclelib/backfilled_oracle.go +++ b/core/services/ocr2/plugins/ccip/internal/oraclelib/backfilled_oracle.go @@ -2,16 +2,16 @@ package oraclelib import ( "context" + "sync" "sync/atomic" "time" - "github.com/smartcontractkit/chainlink/v2/core/services" - "go.uber.org/multierr" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services" "github.com/smartcontractkit/chainlink/v2/core/services/job" ) diff --git a/core/services/ocr2/plugins/ccip/internal/rpclib/evm.go b/core/services/ocr2/plugins/ccip/internal/rpclib/evm.go index 71357029dd..6c4aabb435 100644 --- a/core/services/ocr2/plugins/ccip/internal/rpclib/evm.go +++ b/core/services/ocr2/plugins/ccip/internal/rpclib/evm.go @@ -13,7 +13,7 @@ import ( "github.com/pkg/errors" "golang.org/x/sync/errgroup" - "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink-common/pkg/logger" ) var ErrEmptyOutput = errors.New("rpc call output is empty (make sure that the contract method exists and rpc is healthy)") diff --git a/core/services/ocr2/plugins/ccip/observations.go b/core/services/ocr2/plugins/ccip/observations.go index 29fa85021f..de2f8ffd16 100644 --- a/core/services/ocr2/plugins/ccip/observations.go +++ b/core/services/ocr2/plugins/ccip/observations.go @@ -10,12 +10,13 @@ import ( "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + "github.com/smartcontractkit/chainlink-common/pkg/logger" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipcalc" ) +// CommitObservation are stored to the commit store. // Note if a breaking change is introduced to this struct nodes running different versions // will not be able to unmarshal each other's observations. Do not modify unless you // know what you are doing. diff --git a/core/services/ocr2/plugins/ccip/tokendata/bgworker.go b/core/services/ocr2/plugins/ccip/tokendata/bgworker.go index 1a74ab2305..19e4bed103 100644 --- a/core/services/ocr2/plugins/ccip/tokendata/bgworker.go +++ b/core/services/ocr2/plugins/ccip/tokendata/bgworker.go @@ -12,7 +12,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services" cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip" - "github.com/smartcontractkit/chainlink/v2/core/services/job" ) type msgResult struct { @@ -22,7 +21,7 @@ type msgResult struct { } type Worker interface { - job.ServiceCtx + services.Service // AddJobsFromMsgs will include the provided msgs for background processing. AddJobsFromMsgs(ctx context.Context, msgs []cciptypes.EVM2EVMOnRampCCIPSendRequestedWithMeta) @@ -46,6 +45,14 @@ type BackgroundWorker struct { backgroundCancel context.CancelFunc } +func (w *BackgroundWorker) HealthReport() map[string]error { + return make(map[string]error) +} + +func (w *BackgroundWorker) Name() string { + return "TokenDataBackgroundWorker" +} + func NewBackgroundWorker( tokenDataReaders map[cciptypes.Address]Reader, numWorkers int, diff --git a/core/services/relay/evm/evm.go b/core/services/relay/evm/evm.go index ae5244f5f0..41a2b752a2 100644 --- a/core/services/relay/evm/evm.go +++ b/core/services/relay/evm/evm.go @@ -515,6 +515,7 @@ func (r *Relayer) NewCCIPExecProvider(rargs commontypes.RelayArgs, pargs commont execPluginConfig.SourceStartBlock, execPluginConfig.JobID, usdcConfig.AttestationAPI, + string(ccip.EvmAddrToGeneric(usdcConfig.SourceTokenAddress)), int(usdcConfig.AttestationAPITimeoutSeconds), usdcConfig.AttestationAPIIntervalMilliseconds, usdcConfig.SourceMessageTransmitterAddress, diff --git a/core/services/relay/evm/exec_provider.go b/core/services/relay/evm/exec_provider.go index e50ae41351..69be653f9a 100644 --- a/core/services/relay/evm/exec_provider.go +++ b/core/services/relay/evm/exec_provider.go @@ -37,6 +37,7 @@ type SrcExecProvider struct { maxGasPrice *big.Int usdcReader *ccip.USDCReaderImpl usdcAttestationAPI string + usdcSourceTokenAddress string usdcAttestationAPITimeoutSeconds int usdcAttestationAPIIntervalMilliseconds int usdcSrcMsgTransmitterAddr common.Address @@ -59,6 +60,7 @@ func NewSrcExecProvider( startBlock uint64, jobID string, usdcAttestationAPI string, + usdcSourceTokenAddress string, usdcAttestationAPITimeoutSeconds int, usdcAttestationAPIIntervalMilliseconds int, usdcSrcMsgTransmitterAddr common.Address, @@ -83,6 +85,7 @@ func NewSrcExecProvider( startBlock: startBlock, usdcReader: usdcReader, usdcAttestationAPI: usdcAttestationAPI, + usdcSourceTokenAddress: usdcSourceTokenAddress, usdcAttestationAPITimeoutSeconds: usdcAttestationAPITimeoutSeconds, usdcAttestationAPIIntervalMilliseconds: usdcAttestationAPIIntervalMilliseconds, usdcSrcMsgTransmitterAddr: usdcSrcMsgTransmitterAddr, @@ -194,12 +197,15 @@ func (s *SrcExecProvider) NewPriceRegistryReader(ctx context.Context, addr ccipt return } -func (s *SrcExecProvider) NewTokenDataReader(ctx context.Context, tokenAddress cciptypes.Address) (tokenDataReader cciptypes.TokenDataReader, err error) { +func (s SrcExecProvider) NewTokenDataReader(ctx context.Context, _ cciptypes.Address) (tokenDataReader cciptypes.TokenDataReader, err error) { + if s.usdcAttestationAPI == "" { + return nil, fmt.Errorf("empty USDC attestation API") + } attestationURI, err2 := url.ParseRequestURI(s.usdcAttestationAPI) if err2 != nil { return nil, fmt.Errorf("failed to parse USDC attestation API: %w", err2) } - tokenAddr, err2 := ccip.GenericAddrToEvm(tokenAddress) + tokenAddr, err2 := ccip.GenericAddrToEvm(cciptypes.Address(s.usdcSourceTokenAddress)) if err2 != nil { return nil, fmt.Errorf("failed to parse token address: %w", err2) } @@ -362,8 +368,11 @@ func (d *DstExecProvider) NewCommitStoreReader(ctx context.Context, addr cciptyp return } -func (d *DstExecProvider) NewOffRampReader(ctx context.Context, offRampAddress cciptypes.Address) (offRampReader cciptypes.OffRampReader, err error) { - offRampReader, err = ccip.NewOffRampReader(d.lggr, d.versionFinder, offRampAddress, d.client, d.lp, d.gasEstimator, &d.maxGasPrice, true, d.feeEstimatorConfig) +// NewOffRampReader constructs a reader for the offramp contract on the dest chain. +// The offramp address is known when the provider is constructed - by consuming it from the provider instead of at runtime +// we save ourselves wiring it through the execution reporting plugin factory grpc server + client +func (d DstExecProvider) NewOffRampReader(ctx context.Context, _ cciptypes.Address) (offRampReader cciptypes.OffRampReader, err error) { + offRampReader, err = ccip.NewOffRampReader(d.lggr, d.versionFinder, d.offRampAddress, d.client, d.lp, d.gasEstimator, &d.maxGasPrice, true, d.feeEstimatorConfig) return } @@ -381,6 +390,9 @@ func (d *DstExecProvider) NewTokenDataReader(ctx context.Context, tokenAddress c return nil, fmt.Errorf("invalid: NewTokenDataReader called on DstExecProvider. It should only be called on SrcExecProvider") } +// NewTokenPoolBatchedReader constructs a batched caller to read token prices from the destination pool. +// The offramp address is known when the provider is constructed - by consuming it from the provider instead of at runtime +// we save ourselves wiring it through the execution reporting plugin factory grpc server + client func (d *DstExecProvider) NewTokenPoolBatchedReader(ctx context.Context, offRampAddress cciptypes.Address, sourceChainSelector uint64) (tokenPoolBatchedReader cciptypes.TokenPoolBatchedReader, err error) { batchCaller := ccip.NewDynamicLimitedBatchCaller( d.lggr, @@ -390,7 +402,7 @@ func (d *DstExecProvider) NewTokenPoolBatchedReader(ctx context.Context, offRamp uint(ccip.DefaultMaxParallelRpcCalls), ) - tokenPoolBatchedReader, err = ccip.NewEVMTokenPoolBatchedReader(d.lggr, sourceChainSelector, offRampAddress, batchCaller) + tokenPoolBatchedReader, err = ccip.NewEVMTokenPoolBatchedReader(d.lggr, sourceChainSelector, d.offRampAddress, batchCaller) if err != nil { return nil, fmt.Errorf("new token pool batched reader: %w", err) } diff --git a/dashboard-lib/go.mod b/dashboard-lib/go.mod index f115486231..0a4064b20e 100644 --- a/dashboard-lib/go.mod +++ b/dashboard-lib/go.mod @@ -21,7 +21,7 @@ require ( github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/common v0.45.0 // indirect github.com/stretchr/testify v1.9.0 // indirect - golang.org/x/sys v0.16.0 // indirect + golang.org/x/sys v0.21.0 // indirect ) exclude github.com/sourcegraph/sourcegraph/lib v0.0.0-20221216004406-749998a2ac74 diff --git a/dashboard-lib/go.sum b/dashboard-lib/go.sum index 7eb74088f1..18b2a90648 100644 --- a/dashboard-lib/go.sum +++ b/dashboard-lib/go.sum @@ -31,7 +31,7 @@ github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU= -golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/go.mod b/go.mod index 0c6c29d0e6..017397a527 100644 --- a/go.mod +++ b/go.mod @@ -75,7 +75,7 @@ require ( github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chain-selectors v1.0.23 github.com/smartcontractkit/chainlink-automation v1.0.4 - github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 + github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45 github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240718160222-2dc0c8136bfa github.com/smartcontractkit/chainlink-feeds v0.0.0-20240710170203-5b41615da827 diff --git a/go.sum b/go.sum index d65d3fa705..c364267110 100644 --- a/go.sum +++ b/go.sum @@ -1030,8 +1030,8 @@ github.com/smartcontractkit/chain-selectors v1.0.23 h1:D2Eaex4Cw/O7Lg3tX6WklOqnj github.com/smartcontractkit/chain-selectors v1.0.23/go.mod h1:d4Hi+E1zqjy9HqMkjBE5q1vcG9VGgxf5VxiRHfzi2kE= github.com/smartcontractkit/chainlink-automation v1.0.4 h1:iyW181JjKHLNMnDleI8umfIfVVlwC7+n5izbLSFgjw8= github.com/smartcontractkit/chainlink-automation v1.0.4/go.mod h1:u4NbPZKJ5XiayfKHD/v3z3iflQWqvtdhj13jVZXj/cM= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 h1:pdEpjgbZ5w/Sd5lzg/XiuC5gVyrmSovOo+3nUD46SP8= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b h1:Dak5bYlqcDh3kUbmNZaV5ib+MykYuzAgPkWmtqkPj7U= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45 h1:NBQLtqk8zsyY4qTJs+NElI3aDFTcAo83JHvqD04EvB0= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45/go.mod h1:LV0h7QBQUpoC2UUi6TcUvcIFm1xjP/DtEcqV8+qeLUs= github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240718160222-2dc0c8136bfa h1:g75H8oh2ws52s8BekwvGQ9XvBVu3E7WM1rfiA0PN0zk= diff --git a/integration-tests/ccip-tests/Makefile b/integration-tests/ccip-tests/Makefile index 8fdd635fd6..c008a35a38 100644 --- a/integration-tests/ccip-tests/Makefile +++ b/integration-tests/ccip-tests/Makefile @@ -67,3 +67,7 @@ test_smoke_ccip_default: set_config .PHONY: build_ccip_image build_ccip_image: docker build -f ../../core/chainlink.Dockerfile --build-arg COMMIT_SHA=$(git rev-parse HEAD) --build-arg CHAINLINK_USER=chainlink -t $(image):$(tag) ../../ + +.PHONY: build_ccip_plugin_image +build_ccip_plugin_image: + docker build -f ../../plugins/chainlink.Dockerfile --build-arg COMMIT_SHA=$(git rev-parse HEAD) --build-arg CHAINLINK_USER=chainlink -t $(image):$(tag) ../../ \ No newline at end of file diff --git a/integration-tests/go.mod b/integration-tests/go.mod index 635332c58e..19632687d6 100644 --- a/integration-tests/go.mod +++ b/integration-tests/go.mod @@ -34,7 +34,7 @@ require ( github.com/slack-go/slack v0.12.2 github.com/smartcontractkit/chain-selectors v1.0.23 github.com/smartcontractkit/chainlink-automation v1.0.4 - github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 + github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b github.com/smartcontractkit/chainlink-testing-framework/havoc v1.50.0 github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.5 github.com/smartcontractkit/chainlink-testing-framework/lib/grafana v1.50.0 diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 727c8bf04a..e180f9a4d7 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -1417,8 +1417,8 @@ github.com/smartcontractkit/chain-selectors v1.0.23 h1:D2Eaex4Cw/O7Lg3tX6WklOqnj github.com/smartcontractkit/chain-selectors v1.0.23/go.mod h1:d4Hi+E1zqjy9HqMkjBE5q1vcG9VGgxf5VxiRHfzi2kE= github.com/smartcontractkit/chainlink-automation v1.0.4 h1:iyW181JjKHLNMnDleI8umfIfVVlwC7+n5izbLSFgjw8= github.com/smartcontractkit/chainlink-automation v1.0.4/go.mod h1:u4NbPZKJ5XiayfKHD/v3z3iflQWqvtdhj13jVZXj/cM= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 h1:pdEpjgbZ5w/Sd5lzg/XiuC5gVyrmSovOo+3nUD46SP8= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b h1:Dak5bYlqcDh3kUbmNZaV5ib+MykYuzAgPkWmtqkPj7U= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45 h1:NBQLtqk8zsyY4qTJs+NElI3aDFTcAo83JHvqD04EvB0= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45/go.mod h1:LV0h7QBQUpoC2UUi6TcUvcIFm1xjP/DtEcqV8+qeLUs= github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240718160222-2dc0c8136bfa h1:g75H8oh2ws52s8BekwvGQ9XvBVu3E7WM1rfiA0PN0zk= diff --git a/integration-tests/load/go.mod b/integration-tests/load/go.mod index 2956080624..37e5a4e85a 100644 --- a/integration-tests/load/go.mod +++ b/integration-tests/load/go.mod @@ -16,7 +16,7 @@ require ( github.com/rs/zerolog v1.33.0 github.com/slack-go/slack v0.12.2 github.com/smartcontractkit/chainlink-automation v1.0.4 - github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 + github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.5 github.com/smartcontractkit/chainlink-testing-framework/seth v1.50.2 github.com/smartcontractkit/chainlink-testing-framework/wasp v1.50.0 diff --git a/integration-tests/load/go.sum b/integration-tests/load/go.sum index 8162c16965..93462a9af3 100644 --- a/integration-tests/load/go.sum +++ b/integration-tests/load/go.sum @@ -1389,8 +1389,8 @@ github.com/smartcontractkit/chain-selectors v1.0.23 h1:D2Eaex4Cw/O7Lg3tX6WklOqnj github.com/smartcontractkit/chain-selectors v1.0.23/go.mod h1:d4Hi+E1zqjy9HqMkjBE5q1vcG9VGgxf5VxiRHfzi2kE= github.com/smartcontractkit/chainlink-automation v1.0.4 h1:iyW181JjKHLNMnDleI8umfIfVVlwC7+n5izbLSFgjw8= github.com/smartcontractkit/chainlink-automation v1.0.4/go.mod h1:u4NbPZKJ5XiayfKHD/v3z3iflQWqvtdhj13jVZXj/cM= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1 h1:pdEpjgbZ5w/Sd5lzg/XiuC5gVyrmSovOo+3nUD46SP8= -github.com/smartcontractkit/chainlink-common v0.2.2-0.20240723123524-e407ecd120b1/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b h1:Dak5bYlqcDh3kUbmNZaV5ib+MykYuzAgPkWmtqkPj7U= +github.com/smartcontractkit/chainlink-common v0.2.2-0.20240925011859-eac77b0c392b/go.mod h1:Jg1sCTsbxg76YByI8ifpFby3FvVqISStHT8ypy9ocmY= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45 h1:NBQLtqk8zsyY4qTJs+NElI3aDFTcAo83JHvqD04EvB0= github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240710121324-3ed288aa9b45/go.mod h1:LV0h7QBQUpoC2UUi6TcUvcIFm1xjP/DtEcqV8+qeLUs= github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240718160222-2dc0c8136bfa h1:g75H8oh2ws52s8BekwvGQ9XvBVu3E7WM1rfiA0PN0zk= diff --git a/plugins/chainlink.Dockerfile b/plugins/chainlink.Dockerfile index 6d42567c74..ea6a3def89 100644 --- a/plugins/chainlink.Dockerfile +++ b/plugins/chainlink.Dockerfile @@ -25,6 +25,9 @@ RUN make install-medianpoc # Install ocr3-capability binary RUN make install-ocr3-capability +# Install ccip-exec binary +RUN make install-ccip-exec + # Link LOOP Plugin source dirs with simple names RUN go list -m -f "{{.Dir}}" github.com/smartcontractkit/chainlink-feeds | xargs -I % ln -s % /chainlink-feeds RUN go list -m -f "{{.Dir}}" github.com/smartcontractkit/chainlink-data-streams | xargs -I % ln -s % /chainlink-data-streams @@ -68,6 +71,7 @@ RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add - \ COPY --from=buildgo /go/bin/chainlink /usr/local/bin/ COPY --from=buildgo /go/bin/chainlink-medianpoc /usr/local/bin/ COPY --from=buildgo /go/bin/chainlink-ocr3-capability /usr/local/bin/ +COPY --from=buildgo /go/bin/chainlink-ccip-exec /usr/local/bin/ COPY --from=buildplugins /go/bin/chainlink-feeds /usr/local/bin/ ENV CL_MEDIAN_CMD chainlink-feeds @@ -77,6 +81,7 @@ COPY --from=buildplugins /go/bin/chainlink-solana /usr/local/bin/ ENV CL_SOLANA_CMD chainlink-solana COPY --from=buildplugins /go/bin/chainlink-starknet /usr/local/bin/ ENV CL_STARKNET_CMD chainlink-starknet +ENV CL_CCIP-EXEC_CMD chainlink-ccip-exec # Dependency of CosmWasm/wasmd COPY --from=buildgo /go/pkg/mod/github.com/\!cosm\!wasm/wasmvm@v*/internal/api/libwasmvm.*.so /usr/lib/ diff --git a/plugins/cmd/chainlink-ccip-exec/main.go b/plugins/cmd/chainlink-ccip-exec/main.go new file mode 100644 index 0000000000..d184dbe131 --- /dev/null +++ b/plugins/cmd/chainlink-ccip-exec/main.go @@ -0,0 +1,39 @@ +package main + +import ( + "github.com/hashicorp/go-plugin" + + "github.com/smartcontractkit/chainlink-common/pkg/loop" +) + +const ( + loggerName = "PluginCCIPExec" +) + +func main() { + s := loop.MustNewStartedServer(loggerName) + defer s.Stop() + + p := NewPlugin(s.Logger) + defer s.Logger.ErrorIfFn(p.Close, "Failed to close") + + s.MustRegister(p) + + stop := make(chan struct{}) + defer close(stop) + + plugin.Serve(&plugin.ServeConfig{ + HandshakeConfig: loop.PluginCCIPExecutionHandshakeConfig(), + Plugins: map[string]plugin.Plugin{ + loop.CCIPExecutionLOOPName: &loop.ExecutionLoop{ + PluginServer: p, + BrokerConfig: loop.BrokerConfig{ + StopCh: stop, + Logger: s.Logger, + GRPCOpts: s.GRPCOpts, + }, + }, + }, + GRPCServer: s.GRPCOpts.NewServer, + }) +} diff --git a/plugins/cmd/chainlink-ccip-exec/plugin.go b/plugins/cmd/chainlink-ccip-exec/plugin.go new file mode 100644 index 0000000000..5ee9fc06a9 --- /dev/null +++ b/plugins/cmd/chainlink-ccip-exec/plugin.go @@ -0,0 +1,25 @@ +package main + +import ( + "context" + + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/ccipexec" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-common/pkg/types" +) + +type Plugin struct { + loop.Plugin + stop services.StopChan +} + +func NewPlugin(lggr logger.Logger) *Plugin { + return &Plugin{Plugin: loop.Plugin{Logger: lggr}, stop: make(services.StopChan)} +} + +func (p *Plugin) NewExecutionFactory(ctx context.Context, srcProvider types.CCIPExecProvider, dstProvider types.CCIPExecProvider, srcChainID int64, dstChainID int64, sourceTokenAddress string) (types.ReportingPluginFactory, error) { + return ccipexec.NewExecutionReportingPluginFactoryV2(ctx, nil, sourceTokenAddress, srcChainID, dstChainID, srcProvider, dstProvider) +}