Merge branch 'ccip-develop' into logpoller-reorg-check

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

@@ -5,10 +5,10 @@ import (
 	"encoding/hex"
 	"fmt"
 	"math/big"
-	"slices"
 	"sort"
 	"time"
 
+	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
 
@@ -435,6 +435,9 @@ func validateObservations(ctx context.Context, lggr logger.Logger, destTokens []
 			lggr.Warnw("Skipping observation due to token count mismatch", "expecting", len(destTokens), "got", len(obs.TokenPricesUSD))
 			continue
 		}
+
+		destTokensSet := mapset.NewSet[cciptypes.Address](destTokens...)
+
 		// If any of the observed token prices is reported as nil, or not supported on dest chain, the observation is faulty, skip the observation.
 		// Printing all faulty prices instead of short-circuiting to make log more informative.
 		skipObservation := false
@@ -443,8 +446,11 @@ func validateObservations(ctx context.Context, lggr logger.Logger, destTokens []
 				lggr.Warnw("Nil value in observed TokenPricesUSD", "token", token)
 				skipObservation = true
 			}
-			if !slices.Contains(destTokens, token) {
-				lggr.Warnw("Unsupported token in observed TokenPricesUSD", "token", token)
+
+			if !destTokensSet.Contains(token) {
+				lggr.Warnw("Unsupported token in observed TokenPricesUSD",
+					"token", token,
+					"destTokens", destTokensSet.String())
 				skipObservation = true
 			}
 		}

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

@@ -100,6 +100,12 @@ func (r *ExecutionReportingPlugin) Observation(ctx context.Context, timestamp ty
 	if err := ccipcommon.VerifyNotDown(ctx, r.lggr, r.commitStoreReader, r.onRampReader); err != nil {
 		return nil, err
 	}
+
+	// Ensure that the source price registry is synchronized with the onRamp.
+	if err := r.ensurePriceRegistrySynchronization(ctx); err != nil {
+		return nil, fmt.Errorf("ensuring price registry synchronization: %w", err)
+	}
+
 	// Expire any inflight reports.
 	r.inflightReports.expire(lggr)
 	inFlight := r.inflightReports.getAll()
@@ -1042,11 +1048,6 @@ func (r *ExecutionReportingPlugin) prepareTokenExecData(ctx context.Context) (ex
 		return execTokenData{}, err
 	}
 
-	// Ensure that the source price registry is synchronized with the onRamp.
-	if err = r.ensurePriceRegistrySynchronization(ctx); err != nil {
-		return execTokenData{}, fmt.Errorf("ensuring price registry synchronization: %w", err)
-	}
-
 	sourceFeeTokens, err := r.sourcePriceRegistry.GetFeeTokens(ctx)
 	if err != nil {
 		return execTokenData{}, fmt.Errorf("get source fee tokens: %w", err)

core/services/ocr2/plugins/ccip/ccipexec/ocr2_test.go

@@ -1855,7 +1855,7 @@ func Test_prepareTokenExecData(t *testing.T) {
 			reportingPlugin := ExecutionReportingPlugin{
 				onRampReader:                onrampReader,
 				offRampReader:               offrampReader,
-				sourcePriceRegistry:         nil, // will be updated at first use.
+				sourcePriceRegistry:         sourcePriceRegistry,
 				sourcePriceRegistryProvider: sourcePriceRegistryProvider,
 				destPriceRegistry:           destPriceRegistry,
 				gasPriceEstimator:           gasPriceEstimator,

core/services/ocr2/plugins/ccip/cciptypes/models.go

@@ -10,19 +10,30 @@ import (
 
 type Address string
 
-// TODO: make JSON marshal/unmarshal non-evm specific.
-// Make sure we have casing compatibility with old versions.
-func (a *Address) UnmarshalJSON(bytes []byte) error {
-	vStr := strings.Trim(string(bytes), `"`)
+func (a *Address) UnmarshalJSON(b []byte) error {
+	vStr := strings.Trim(string(b), `"`)
 	if !common.IsHexAddress(vStr) {
 		return fmt.Errorf("invalid address: %s", vStr)
 	}
 	*a = Address(common.HexToAddress(vStr).String())
 	return nil
 }
 
-func (a *Address) MarshalJSON() ([]byte, error) {
-	return []byte(`"` + strings.ToLower(string(*a)) + `"`), nil
+func (a Address) MarshalJSON() ([]byte, error) {
+	return []byte(`"` + strings.ToLower(string(a)) + `"`), nil
+}
+
+func (a Address) MarshalText() (text []byte, err error) {
+	return []byte(strings.ToLower(string(a))), nil
+}
+
+func (a *Address) UnmarshalText(text []byte) error {
+	vStr := string(text)
+	if !common.IsHexAddress(vStr) {
+		return fmt.Errorf("invalid address: %s", vStr)
+	}
+	*a = Address(common.HexToAddress(vStr).String())
+	return nil
 }
 
 type Hash [32]byte

core/services/ocr2/plugins/ccip/cciptypes/models_test.go

@@ -1,15 +1,12 @@
 package cciptypes
 
 import (
-	"encoding/json"
 	"fmt"
-	"strings"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
+	json2 "github.com/goccy/go-json"
 	"github.com/stretchr/testify/assert"
-
-	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils"
 )
 
 func TestHash_String(t *testing.T) {
@@ -44,26 +41,30 @@ func TestHash_String(t *testing.T) {
 }
 
 func TestAddress_JSON(t *testing.T) {
-	addr1 := utils.RandomAddress()
-
-	addrArr := []Address{Address(addr1.String())}
-	evmAddrArr := []common.Address{addr1}
+	addrLower := "0xe8bade28e08b469b4eeec35b9e48b2ce49fb3fc9"
+	addrEIP55 := "0xE8BAde28E08B469B4EeeC35b9E48B2Ce49FB3FC9"
 
-	b, err := json.Marshal(addrArr)
-	assert.NoError(t, err)
-	assert.Equal(t, fmt.Sprintf(`["%s"]`, strings.ToLower(addr1.String())), string(b))
+	t.Run("arrays", func(t *testing.T) {
+		addrArr := []Address{Address(addrLower), Address(addrEIP55)}
+		b, err := json2.Marshal(addrArr)
+		assert.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf(`["%s","%s"]`, addrLower, addrLower), string(b))
 
-	b2, err := json.Marshal(evmAddrArr)
-	assert.NoError(t, err)
-	assert.Equal(t, string(b), string(b2), "marshal should produce the same result for common.Address and cciptypes.Address")
+		evmAddrArr := []common.Address{common.HexToAddress(addrLower), common.HexToAddress(addrEIP55)}
+		bEvm, err := json2.Marshal(evmAddrArr)
+		assert.NoError(t, err)
+		assert.Equal(t, b, bEvm)
+	})
 
-	var unmarshalledAddr []Address
-	err = json.Unmarshal(b, &unmarshalledAddr)
-	assert.NoError(t, err)
-	assert.Equal(t, addrArr[0], unmarshalledAddr[0])
+	t.Run("maps", func(t *testing.T) {
+		m := map[Address]int{Address(addrEIP55): 14}
+		b, err := json2.Marshal(m)
+		assert.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf(`{"%s":14}`, addrLower), string(b), "should be lower when marshalled")
 
-	var unmarshalledEvmAddr []common.Address
-	err = json.Unmarshal(b, &unmarshalledEvmAddr)
-	assert.NoError(t, err)
-	assert.Equal(t, evmAddrArr[0], unmarshalledEvmAddr[0])
+		m2 := map[Address]int{}
+		err = json2.Unmarshal(b, &m2)
+		assert.NoError(t, err)
+		assert.Equal(t, m, m2, "should be eip55 when unmarshalled")
+	})
 }

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_0_0/onramp.go

@@ -25,6 +25,7 @@ import (
 
 const (
 	CCIPSendRequestedEventName = "CCIPSendRequested"
+	ConfigSetEventName         = "ConfigSet"
 )
 
 var _ ccipdata.OnRampReader = &OnRamp{}
@@ -52,6 +53,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 	}
 	onRampABI := abihelpers.MustParseABI(evm_2_evm_onramp_1_0_0.EVM2EVMOnRampABI)
 	eventSig := abihelpers.MustGetEventID(CCIPSendRequestedEventName, onRampABI)
+	configSetEventSig := abihelpers.MustGetEventID(ConfigSetEventName, onRampABI)
 	filters := []logpoller.Filter{
 		{
 			Name:      logpoller.FilterName(ccipdata.COMMIT_CCIP_SENDS, onRampAddress),
@@ -60,7 +62,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
-			EventSigs: []common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			EventSigs: []common.Hash{configSetEventSig},
 			Addresses: []common.Address{onRampAddress},
 		},
 	}
@@ -80,7 +82,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		sendRequestedEventSig:      eventSig,
 		cachedSourcePriceRegistryAddress: cache.NewLogpollerEventsBased[cciptypes.Address](
 			sourceLP,
-			[]common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			[]common.Hash{configSetEventSig},
 			onRampAddress,
 		),
 		cachedStaticConfig: cachedStaticConfig,

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_2_0/onramp.go

@@ -27,11 +27,13 @@ import (
 var (
 	// Backwards compat for integration tests
 	CCIPSendRequestEventSig common.Hash
+	ConfigSetEventSig       common.Hash
 )
 
 const (
 	CCIPSendRequestSeqNumIndex = 4
 	CCIPSendRequestedEventName = "CCIPSendRequested"
+	ConfigSetEventName         = "ConfigSet"
 )
 
 func init() {
@@ -40,6 +42,7 @@ func init() {
 		panic(err)
 	}
 	CCIPSendRequestEventSig = abihelpers.MustGetEventID(CCIPSendRequestedEventName, onRampABI)
+	ConfigSetEventSig = abihelpers.MustGetEventID(ConfigSetEventName, onRampABI)
 }
 
 var _ ccipdata.OnRampReader = &OnRamp{}
@@ -67,7 +70,6 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 	if err != nil {
 		return nil, err
 	}
-	onRampABI := abihelpers.MustParseABI(evm_2_evm_onramp_1_2_0.EVM2EVMOnRampABI)
 	// Subscribe to the relevant logs
 	// Note we can keep the same prefix across 1.0/1.1 and 1.2 because the onramp addresses will be different
 	filters := []logpoller.Filter{
@@ -78,7 +80,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
-			EventSigs: []common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			EventSigs: []common.Hash{ConfigSetEventSig},
 			Addresses: []common.Address{onRampAddress},
 		},
 	}
@@ -97,7 +99,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		sendRequestedEventSig:      CCIPSendRequestEventSig,
 		cachedSourcePriceRegistryAddress: cache.NewLogpollerEventsBased[cciptypes.Address](
 			sourceLP,
-			[]common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			[]common.Hash{ConfigSetEventSig},
 			onRampAddress,
 		),
 		cachedStaticConfig: cachedStaticConfig,

core/services/ocr2/plugins/ccip/internal/ccipdata/v1_5_0/onramp.go

@@ -28,11 +28,13 @@ import (
 var (
 	// Backwards compat for integration tests
 	CCIPSendRequestEventSig common.Hash
+	ConfigSetEventSig       common.Hash
 )
 
 const (
 	CCIPSendRequestSeqNumIndex = 4
 	CCIPSendRequestedEventName = "CCIPSendRequested"
+	ConfigSetEventName         = "ConfigSet"
 )
 
 func init() {
@@ -41,6 +43,7 @@ func init() {
 		panic(err)
 	}
 	CCIPSendRequestEventSig = abihelpers.MustGetEventID(CCIPSendRequestedEventName, onRampABI)
+	ConfigSetEventSig = abihelpers.MustGetEventID(ConfigSetEventName, onRampABI)
 }
 
 var _ ccipdata.OnRampReader = &OnRamp{}
@@ -68,7 +71,6 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 	if err != nil {
 		return nil, err
 	}
-	onRampABI := abihelpers.MustParseABI(evm_2_evm_onramp.EVM2EVMOnRampABI)
 	// Subscribe to the relevant logs
 	// Note we can keep the same prefix across 1.0/1.1 and 1.2 because the onramp addresses will be different
 	filters := []logpoller.Filter{
@@ -79,7 +81,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		},
 		{
 			Name:      logpoller.FilterName(ccipdata.CONFIG_CHANGED, onRampAddress),
-			EventSigs: []common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			EventSigs: []common.Hash{ConfigSetEventSig},
 			Addresses: []common.Address{onRampAddress},
 		},
 	}
@@ -98,7 +100,7 @@ func NewOnRamp(lggr logger.Logger, sourceSelector, destSelector uint64, onRampAd
 		sendRequestedEventSig:      CCIPSendRequestEventSig,
 		cachedSourcePriceRegistryAddress: cache.NewLogpollerEventsBased[cciptypes.Address](
 			sourceLP,
-			[]common.Hash{abihelpers.MustGetEventID("ConfigSet", onRampABI)},
+			[]common.Hash{ConfigSetEventSig},
 			onRampAddress,
 		),
 		cachedStaticConfig: cachedStaticConfig,

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

@@ -1,9 +1,9 @@
 package ccip
 
 import (
-	"encoding/json"
 	"math/big"
 
+	json2 "github.com/goccy/go-json"
 	"github.com/smartcontractkit/libocr/commontypes"
 
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
@@ -22,7 +22,7 @@ type CommitObservation struct {
 }
 
 func (o CommitObservation) Marshal() ([]byte, error) {
-	return json.Marshal(&o)
+	return json2.Marshal(&o)
 }
 
 // ExecutionObservation stores messages as a map pointing from a sequence number (uint) to the message payload (MsgData)
@@ -64,7 +64,7 @@ func NewObservedMessage(seqNr uint64, tokenData [][]byte) ObservedMessage {
 }
 
 func (o ExecutionObservation) Marshal() ([]byte, error) {
-	return json.Marshal(&o)
+	return json2.Marshal(&o)
 }
 
 // GetParsableObservations checks the given observations for formatting and value errors.
@@ -80,7 +80,7 @@ func GetParsableObservations[O CommitObservation | ExecutionObservation](l logge
 			continue
 		}
 		var ob O
-		err := json.Unmarshal(ao.Observation, &ob)
+		err := json2.Unmarshal(ao.Observation, &ob)
 		if err != nil {
 			l.Errorw("Received unmarshallable observation", "err", err, "observation", string(ao.Observation), "observer", ao.Observer)
 			continue

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/gin-gonic/gin v1.9.1
 	github.com/go-ldap/ldap/v3 v3.4.6
 	github.com/go-webauthn/webauthn v0.9.4
+	github.com/goccy/go-json v0.10.2
 	github.com/google/pprof v0.0.0-20231023181126-ff6d637d2a7b
 	github.com/google/uuid v1.4.0
 	github.com/gorilla/securecookie v1.1.2
@@ -194,7 +195,6 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.15.5 // indirect
 	github.com/go-webauthn/x v0.1.5 // indirect
-	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect
 	github.com/gofrs/flock v0.8.1 // indirect
 	github.com/gofrs/uuid v4.3.1+incompatible // indirect