Offchain usdc (#112)

core/services/ocr2/delegate.go

@@ -293,7 +293,7 @@ func (d *Delegate) cleanupEVM(jb job.Job, q pg.Queryer, relayID relay.ID) error
 		}
 		return nil
 	case types.CCIPExecution:
-		err = ccip.UnregisterExecPluginLpFilters(context.Background(), spec, d.legacyChains, pg.WithQueryer(q))
+		err = ccip.UnregisterExecPluginLpFilters(context.Background(), d.lggr, spec, d.legacyChains, pg.WithQueryer(q))
 		if err != nil {
 			d.lggr.Errorw("failed to unregister ccip exec plugin filters", "err", err, "spec", spec)
 		}

core/services/ocr2/plugins/ccip/abihelpers/abi_helpers.go

@@ -45,6 +45,8 @@ var EventSignatures struct {
 	FeeTokenAdded        common.Hash
 	FeeTokenRemoved      common.Hash
 
+	USDCMessageSent common.Hash
+
 	// offset || sourceChainID || seqNum || ...
 	SendRequestedSequenceNumberWord int
 	// offset || priceUpdatesOffset || minSeqNum || maxSeqNum || merkleRoot
@@ -128,6 +130,8 @@ func init() {
 		panic("missing event 'manuallyExecute'")
 	}
 	ExecutionReportArgs = manuallyExecuteMethod.Inputs[:1]
+
+	EventSignatures.USDCMessageSent = utils.Keccak256Fixed([]byte("MessageSent(bytes)"))
 }
 
 func MessagesFromExecutionReport(report types.Report) ([]evm_2_evm_offramp.InternalEVM2EVMMessage, error) {

core/services/ocr2/plugins/ccip/config/config.go

@@ -1,5 +1,12 @@
 package config
 
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/pkg/errors"
+
+	"github.com/smartcontractkit/chainlink/v2/core/utils"
+)
+
 // CommitPluginJobSpecConfig contains the plugin specific variables for the ccip.CCIPCommit plugin.
 // We use ID here to keep it as general as possible, e.g. abstracting for chains which don't have an address concept.
 type CommitPluginJobSpecConfig struct {
@@ -14,4 +21,29 @@ type CommitPluginJobSpecConfig struct {
 // ExecutionPluginJobSpecConfig contains the plugin specific variables for the ccip.CCIPExecution plugin.
 type ExecutionPluginJobSpecConfig struct {
 	SourceStartBlock, DestStartBlock int64 // Only for first time job add.
+	USDCConfig                       USDCConfig
+}
+
+type USDCConfig struct {
+	SourceTokenAddress              common.Address
+	SourceMessageTransmitterAddress common.Address
+	AttestationAPI                  string
+}
+
+func (uc *USDCConfig) ValidateUSDCConfig() error {
+	if uc.AttestationAPI == "" && uc.SourceTokenAddress == utils.ZeroAddress && uc.SourceMessageTransmitterAddress == utils.ZeroAddress {
+		return nil
+	}
+
+	if uc.AttestationAPI == "" {
+		return errors.New("AttestationAPI is required")
+	}
+	if uc.SourceTokenAddress == utils.ZeroAddress {
+		return errors.New("SourceTokenAddress is required")
+	}
+	if uc.SourceMessageTransmitterAddress == utils.ZeroAddress {
+		return errors.New("SourceMessageTransmitterAddress is required")
+	}
+
+	return nil
 }

core/services/ocr2/plugins/ccip/execution_plugin.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/url"
 	"strconv"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -15,6 +16,7 @@ import (
 	libocr2 "github.com/smartcontractkit/libocr/offchainreporting2plus"
 
 	relaylogger "github.com/smartcontractkit/chainlink-relay/pkg/logger"
+
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/ccipdata"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/contractutil"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/internal/hashlib"
@@ -34,6 +36,8 @@ import (
 	"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/observability"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/tokendata"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/tokendata/usdc"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/promwrapper"
 	"github.com/smartcontractkit/chainlink/v2/core/services/pg"
 )
@@ -110,12 +114,19 @@ func NewExecutionServices(lggr logger.Logger, jb job.Job, chainSet evm.LegacyCha
 		"sourceChain", ChainName(int64(chainId)),
 		"destChain", ChainName(destChainID))
 
+	sourceChainEventClient := ccipdata.NewLogPollerReader(sourceChain.LogPoller(), execLggr, sourceChain.Client())
+
+	tokenDataProviders, err := getTokenDataProviders(lggr, pluginConfig, offRampConfig.OnRamp, sourceChainEventClient)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get token data providers")
+	}
+
 	wrappedPluginFactory := NewExecutionReportingPluginFactory(
 		ExecutionPluginConfig{
 			lggr:                     execLggr,
 			sourceLP:                 sourceChain.LogPoller(),
 			destLP:                   destChain.LogPoller(),
-			sourceReader:             ccipdata.NewLogPollerReader(sourceChain.LogPoller(), execLggr, sourceChain.Client()),
+			sourceReader:             sourceChainEventClient,
 			destReader:               ccipdata.NewLogPollerReader(destChain.LogPoller(), execLggr, destChain.Client()),
 			onRamp:                   onRamp,
 			offRamp:                  offRamp,
@@ -126,6 +137,7 @@ func NewExecutionServices(lggr logger.Logger, jb job.Job, chainSet evm.LegacyCha
 			sourceClient:             sourceChain.Client(),
 			destGasEstimator:         destChain.GasEstimator(),
 			leafHasher:               hashlib.NewLeafHasher(offRampConfig.SourceChainSelector, offRampConfig.ChainSelector, onRamp.Address(), hashlib.NewKeccakCtx()),
+			tokenDataProviders:       tokenDataProviders,
 		})
 
 	err = wrappedPluginFactory.UpdateLogPollerFilters(utils.ZeroAddress, qopts...)
@@ -161,8 +173,42 @@ func NewExecutionServices(lggr logger.Logger, jb job.Job, chainSet evm.LegacyCha
 	return []job.ServiceCtx{job.NewServiceAdapter(oracle)}, nil
 }
 
-func getExecutionPluginSourceLpChainFilters(onRamp, priceRegistry common.Address) []logpoller.Filter {
-	return []logpoller.Filter{
+func getTokenDataProviders(lggr logger.Logger, pluginConfig ccipconfig.ExecutionPluginJobSpecConfig, onRampAddress common.Address, sourceChainEventClient *ccipdata.LogPollerReader) (map[common.Address]tokendata.Reader, error) {
+	tokenDataProviders := make(map[common.Address]tokendata.Reader)
+
+	if pluginConfig.USDCConfig.AttestationAPI != "" {
+		lggr.Infof("USDC token data provider enabled")
+		err := pluginConfig.USDCConfig.ValidateUSDCConfig()
+		if err != nil {
+			return nil, err
+		}
+
+		attestationURI, err2 := url.ParseRequestURI(pluginConfig.USDCConfig.AttestationAPI)
+		if err2 != nil {
+			return nil, errors.Wrap(err2, "failed to parse USDC attestation API")
+		}
+
+		tokenDataProviders[pluginConfig.USDCConfig.SourceTokenAddress] = tokendata.NewCachedReader(
+			usdc.NewUSDCTokenDataReader(
+				sourceChainEventClient,
+				pluginConfig.USDCConfig.SourceTokenAddress,
+				pluginConfig.USDCConfig.SourceMessageTransmitterAddress,
+				onRampAddress,
+				attestationURI,
+			),
+		)
+	}
+
+	return tokenDataProviders, nil
+}
+
+func getExecutionPluginSourceLpChainFilters(onRamp, priceRegistry common.Address, tokenDataProviders map[common.Address]tokendata.Reader) []logpoller.Filter {
+	var filters []logpoller.Filter
+	for _, provider := range tokenDataProviders {
+		filters = append(filters, provider.GetSourceLogPollerFilters()...)
+	}
+
+	return append(filters, []logpoller.Filter{
 		{
 			Name:      logpoller.FilterName(EXEC_CCIP_SENDS, onRamp.String()),
 			EventSigs: []common.Hash{abihelpers.EventSignatures.SendRequested},
@@ -178,7 +224,7 @@ func getExecutionPluginSourceLpChainFilters(onRamp, priceRegistry common.Address
 			EventSigs: []common.Hash{abihelpers.EventSignatures.FeeTokenRemoved},
 			Addresses: []common.Address{priceRegistry},
 		},
-	}
+	}...)
 }
 
 func getExecutionPluginDestLpChainFilters(commitStore, offRamp, priceRegistry common.Address) []logpoller.Filter {
@@ -217,7 +263,7 @@ func getExecutionPluginDestLpChainFilters(commitStore, offRamp, priceRegistry co
 }
 
 // UnregisterExecPluginLpFilters unregisters all the registered filters for both source and dest chains.
-func UnregisterExecPluginLpFilters(ctx context.Context, spec *job.OCR2OracleSpec, chainSet evm.LegacyChainContainer, qopts ...pg.QOpt) error {
+func UnregisterExecPluginLpFilters(ctx context.Context, lggr logger.Logger, spec *job.OCR2OracleSpec, chainSet evm.LegacyChainContainer, qopts ...pg.QOpt) error {
 	if spec == nil {
 		return errors.New("spec is nil")
 	}
@@ -264,17 +310,19 @@ func UnregisterExecPluginLpFilters(ctx context.Context, spec *job.OCR2OracleSpec
 		return errors.Wrap(err, "failed loading onRamp")
 	}
 
-	return unregisterExecutionPluginLpFilters(ctx, sourceChain.LogPoller(), destChain.LogPoller(), offRamp, offRampConfig, sourceOnRamp, sourceChain.Client(), qopts...)
+	return unregisterExecutionPluginLpFilters(ctx, lggr, sourceChain.LogPoller(), destChain.LogPoller(), offRamp, offRampConfig, sourceOnRamp, sourceChain.Client(), pluginConfig, qopts...)
 }
 
 func unregisterExecutionPluginLpFilters(
 	ctx context.Context,
+	lggr logger.Logger,
 	sourceLP logpoller.LogPoller,
 	destLP logpoller.LogPoller,
 	destOffRamp evm_2_evm_offramp.EVM2EVMOffRampInterface,
 	destOffRampConfig evm_2_evm_offramp.EVM2EVMOffRampStaticConfig,
 	sourceOnRamp evm_2_evm_onramp.EVM2EVMOnRampInterface,
 	sourceChainClient client.Client,
+	pluginConfig ccipconfig.ExecutionPluginJobSpecConfig,
 	qopts ...pg.QOpt) error {
 	destOffRampDynCfg, err := destOffRamp.GetDynamicConfig(&bind.CallOpts{Context: ctx})
 	if err != nil {
@@ -286,9 +334,15 @@ func unregisterExecutionPluginLpFilters(
 		return err
 	}
 
-	if err := logpollerutil.UnregisterLpFilters(
+	// SourceChainEventClient can be nil because it is not used in unregisterExecutionPluginLpFilters
+	tokenDataProviders, err := getTokenDataProviders(lggr, pluginConfig, destOffRampConfig.OnRamp, nil)
+	if err != nil {
+		return err
+	}
+
+	if err = logpollerutil.UnregisterLpFilters(
 		sourceLP,
-		getExecutionPluginSourceLpChainFilters(destOffRampConfig.OnRamp, onRampDynCfg.PriceRegistry),
+		getExecutionPluginSourceLpChainFilters(destOffRampConfig.OnRamp, onRampDynCfg.PriceRegistry, tokenDataProviders),
 		qopts...,
 	); err != nil {
 		return err

core/services/ocr2/plugins/ccip/execution_plugin_test.go

@@ -12,8 +12,12 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/mocks"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_offramp"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_onramp"
+	"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/config"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/testhelpers"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ccip/tokendata/usdc"
+	"github.com/smartcontractkit/chainlink/v2/core/utils"
 )
 
 func TestGetExecutionPluginFilterNamesFromSpec(t *testing.T) {
@@ -53,7 +57,7 @@ func TestGetExecutionPluginFilterNamesFromSpec(t *testing.T) {
 	for _, tc := range testCases {
 		chainSet := &mocks.LegacyChainContainer{}
 		t.Run(tc.description, func(t *testing.T) {
-			err := UnregisterExecPluginLpFilters(context.Background(), tc.spec, chainSet)
+			err := UnregisterExecPluginLpFilters(context.Background(), logger.TestLogger(t), tc.spec, chainSet)
 			if tc.expectingErr {
 				assert.Error(t, err)
 			} else {
@@ -74,11 +78,20 @@ func TestGetExecutionPluginFilterNames(t *testing.T) {
 	mockOnRamp, onRampAddr := testhelpers.NewFakeOnRamp(t)
 	mockOnRamp.SetDynamicCfg(evm_2_evm_onramp.EVM2EVMOnRampDynamicConfig{PriceRegistry: srcPriceRegAddr})
 
+	pluginConfig := config.ExecutionPluginJobSpecConfig{
+		USDCConfig: config.USDCConfig{
+			SourceTokenAddress:              utils.RandomAddress(),
+			SourceMessageTransmitterAddress: utils.RandomAddress(),
+			AttestationAPI:                  "http://localhost:8080",
+		},
+	}
+
 	srcLP := mocklp.NewLogPoller(t)
 	srcFilters := []string{
 		"Exec ccip sends - " + onRampAddr.String(),
 		"Fee token added - 0xdAFea492D9c6733aE3d56B7ed1ADb60692c98bC9",
 		"Fee token removed - 0xdAFea492D9c6733aE3d56B7ed1ADb60692c98bC9",
+		usdc.MESSAGE_SENT_FILTER_NAME + " - " + pluginConfig.USDCConfig.SourceMessageTransmitterAddress.Hex(),
 	}
 	for _, f := range srcFilters {
 		srcLP.On("UnregisterFilter", f, mock.Anything).Return(nil)
@@ -99,15 +112,18 @@ func TestGetExecutionPluginFilterNames(t *testing.T) {
 
 	err := unregisterExecutionPluginLpFilters(
 		context.Background(),
+		logger.TestLogger(t),
 		srcLP,
 		dstLP,
 		mockOffRamp,
 		evm_2_evm_offramp.EVM2EVMOffRampStaticConfig{
-			CommitStore: commitStoreAddr,
-			OnRamp:      onRampAddr,
+			CommitStore:         commitStoreAddr,
+			OnRamp:              onRampAddr,
+			SourceChainSelector: 5009297550715157269,
 		},
 		mockOnRamp,
 		nil,
+		pluginConfig,
 	)
 	assert.NoError(t, err)