Fee boosting

commit/tokenprice/observation.go

@@ -44,22 +44,12 @@ func (p *processor) ObserveFeedTokenPrices(ctx context.Context) []cciptypes.Toke
 	//sort tokens to query to ensure deterministic order
 	sort.Slice(tokensToQuery, func(i, j int) bool { return tokensToQuery[i] < tokensToQuery[j] })
 	p.lggr.Infow("observing feed token prices", "tokens", tokensToQuery)
-	tokenPrices, err := p.tokenPriceReader.GetTokenFeedPricesUSD(ctx, tokensToQuery)
+	tokenPricesUSD, err := p.tokenPriceReader.GetTokenFeedPricesUSD(ctx, tokensToQuery)
 	if err != nil {
 		p.lggr.Errorw("call to GetTokenFeedPricesUSD failed", "err", err)
 		return []cciptypes.TokenPrice{}
 	}
 
-	// If we couldn't fetch all prices log and return only the ones we could fetch
-	if len(tokenPrices) != len(tokensToQuery) {
-		p.lggr.Errorw("token prices length mismatch", "got", tokenPrices, "want", tokensToQuery)
-	}
-
-	tokenPricesUSD := make([]cciptypes.TokenPrice, 0, len(tokenPrices))
-	for i, token := range tokensToQuery {
-		tokenPricesUSD = append(tokenPricesUSD, cciptypes.NewTokenPrice(token, tokenPrices[i]))
-	}
-
 	return tokenPricesUSD
 }
 

commit/tokenprice/observation_test.go

@@ -57,7 +57,7 @@ func Test_Observation(t *testing.T) {
 
 				tokenPriceReader := readermock.NewMockPriceReader(t)
 				tokenPriceReader.EXPECT().GetTokenFeedPricesUSD(mock.Anything, []types.Account{tokenA, tokenB}).
-					Return([]*big.Int{bi100, bi200}, nil)
+					Return(feedTokenPrices, nil)
 
 				tokenPriceReader.EXPECT().GetFeeQuoterTokenUpdates(mock.Anything, mock.Anything).Return(
 					map[types.Account]shared.TimestampedBig{

execute/exectypes/observation.go

@@ -3,6 +3,8 @@ package exectypes
 import (
 	"encoding/json"
 
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
+
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 )
 
@@ -13,6 +15,8 @@ type CommitObservations map[cciptypes.ChainSelector][]CommitData
 // and sequence number.
 type MessageObservations map[cciptypes.ChainSelector]map[cciptypes.SeqNum]cciptypes.Message
 
+type FeeTokenPriceObservations map[types.Account]cciptypes.TokenPrice
+
 // NonceObservations contain the latest nonce for senders in the previously observed messages.
 // Nonces are organized by source chain selector and the string encoded sender address. The address
 // must be encoding according to the destination chain requirements with typeconv.AddressBytesToString.
@@ -28,22 +32,33 @@ type Observation struct {
 
 	// Messages are determined during the second phase of execute.
 	// Ideally, it contains all the messages identified by the previous outcome's
-	// NextCommits. With the previous outcome, and these messsages, we can build the
+	// NextCommits. With the previous outcome, and these messages, we can build the
 	// execute report.
 	Messages MessageObservations `json:"messages"`
 
+	// FeeTokenPrices are determined during the second phase of execute.
+	// We need to observe the token prices for fee tokens so that we can calculate the fees in USD for each
+	// message and compare them to the execution cost for each message in Outcome. If the fees are greater than
+	// the execution cost, we will not execute the message.
+	FeeTokenPrices FeeTokenPriceObservations `json:"feeTokenPrices"`
+
 	// Nonces are determined during the third phase of execute.
 	// It contains the nonces of senders who are being considered for the final report.
 	Nonces NonceObservations `json:"nonces"`
 }
 
-// NewObservation constructs a Observation object.
+// NewObservation constructs an Observation object.
 func NewObservation(
-	commitReports CommitObservations, messages MessageObservations, nonces NonceObservations) Observation {
+	commitReports CommitObservations,
+	messages MessageObservations,
+	nonces NonceObservations,
+	feeTokenPrices FeeTokenPriceObservations,
+) Observation {
 	return Observation{
-		CommitReports: commitReports,
-		Messages:      messages,
-		Nonces:        nonces,
+		CommitReports:  commitReports,
+		Messages:       messages,
+		Nonces:         nonces,
+		FeeTokenPrices: feeTokenPrices,
 	}
 }
 

execute/factory.go

@@ -4,22 +4,26 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 
 	"google.golang.org/grpc"
 
+	"github.com/smartcontractkit/libocr/commontypes"
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
-	"github.com/smartcontractkit/libocr/commontypes"
-	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
-	ragep2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/execute/internal/gas"
 	"github.com/smartcontractkit/chainlink-ccip/internal/reader"
 	readerpkg "github.com/smartcontractkit/chainlink-ccip/pkg/reader"
 	"github.com/smartcontractkit/chainlink-ccip/pluginconfig"
+
+	ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 )
 
 // PluginFactoryConstructor implements common OCR3ReportingPluginClient and is used for initializing a plugin factory
@@ -109,6 +113,12 @@ func (p PluginFactory) NewReportingPlugin(
 		p.ocrConfig.Config.OfframpAddress,
 	)
 
+	// TODO: Replace this with a price reader that reads from a configured price feed
+	defaultTokenPrice := big.NewInt(1)
+	onChainTokenPricesReader := reader.NewStaticPriceReader(map[ocr2types.Account]*big.Int{}, defaultTokenPrice)
+
+	messageExecutionCostEstimator := gas.NewStaticMessageExecutionCostEstimator(cciptypes.NewBigIntFromInt64(0))
+
 	return NewPlugin(
 			config,
 			pluginconfig.ExecutePluginConfig{
@@ -123,6 +133,8 @@ func (p PluginFactory) NewReportingPlugin(
 			p.tokenDataReader,
 			p.estimateProvider,
 			p.lggr,
+			onChainTokenPricesReader,
+			messageExecutionCostEstimator,
 		), ocr3types.ReportingPluginInfo{
 			Name: "CCIPRoleExecute",
 			Limits: ocr3types.ReportingPluginLimits{

execute/internal/gas/cost_estimator.go

@@ -0,0 +1,26 @@
+package gas
+
+import (
+	"github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+)
+
+type MessageExecutionCostEstimator interface {
+	EstimateMsgCostUSD(msg ccipocr3.Message) (ccipocr3.BigInt, error)
+}
+
+type StaticMessageExecutionCostEstimator struct {
+	StaticCostUSD ccipocr3.BigInt
+}
+
+func NewStaticMessageExecutionCostEstimator(staticCostUSD ccipocr3.BigInt) *StaticMessageExecutionCostEstimator {
+	return &StaticMessageExecutionCostEstimator{
+		StaticCostUSD: staticCostUSD,
+	}
+}
+
+func (s StaticMessageExecutionCostEstimator) EstimateMsgCostUSD(msg ccipocr3.Message) (ccipocr3.BigInt, error) {
+	return s.StaticCostUSD, nil
+}
+
+// Ensure StaticMessageExecutionCostEstimator implements MessageExecutionCostEstimator
+var _ MessageExecutionCostEstimator = (*StaticMessageExecutionCostEstimator)(nil)

execute/plugin.go

@@ -9,13 +9,14 @@ import (
 	mapset "github.com/deckarep/golang-set/v2"
 	"golang.org/x/exp/maps"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/logger"
-	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 	libocrtypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/execute/internal/gas"
 	"github.com/smartcontractkit/chainlink-ccip/execute/report"
@@ -41,10 +42,12 @@ type Plugin struct {
 	msgHasher    cciptypes.MessageHasher
 	homeChain    reader.HomeChain
 
-	oracleIDToP2pID  map[commontypes.OracleID]libocrtypes.PeerID
-	tokenDataReader  exectypes.TokenDataReader
-	estimateProvider gas.EstimateProvider
-	lggr             logger.Logger
+	oracleIDToP2pID               map[commontypes.OracleID]libocrtypes.PeerID
+	tokenDataReader               exectypes.TokenDataReader
+	estimateProvider              gas.EstimateProvider
+	lggr                          logger.Logger
+	onChainTokenPricesReader      reader.PriceReader
+	messageExecutionCostEstimator gas.MessageExecutionCostEstimator
 }
 
 func NewPlugin(
@@ -58,6 +61,8 @@ func NewPlugin(
 	tokenDataReader exectypes.TokenDataReader,
 	estimateProvider gas.EstimateProvider,
 	lggr logger.Logger,
+	onChainTokenPricesReader reader.PriceReader,
+	messageExecutionCostEstimator gas.MessageExecutionCostEstimator,
 ) *Plugin {
 	readerSyncer := plugincommon.NewBackgroundReaderSyncer(
 		lggr,
@@ -70,17 +75,19 @@ func NewPlugin(
 	}
 
 	return &Plugin{
-		reportingCfg:     reportingCfg,
-		cfg:              cfg,
-		oracleIDToP2pID:  oracleIDToP2pID,
-		ccipReader:       ccipReader,
-		readerSyncer:     readerSyncer,
-		reportCodec:      reportCodec,
-		msgHasher:        msgHasher,
-		homeChain:        homeChain,
-		tokenDataReader:  tokenDataReader,
-		estimateProvider: estimateProvider,
-		lggr:             lggr,
+		reportingCfg:                  reportingCfg,
+		cfg:                           cfg,
+		oracleIDToP2pID:               oracleIDToP2pID,
+		ccipReader:                    ccipReader,
+		readerSyncer:                  readerSyncer,
+		reportCodec:                   reportCodec,
+		msgHasher:                     msgHasher,
+		homeChain:                     homeChain,
+		tokenDataReader:               tokenDataReader,
+		estimateProvider:              estimateProvider,
+		lggr:                          lggr,
+		onChainTokenPricesReader:      onChainTokenPricesReader,
+		messageExecutionCostEstimator: messageExecutionCostEstimator,
 	}
 }
 
@@ -188,14 +195,15 @@ func (p *Plugin) Observation(
 			}
 
 			// TODO: truncate grouped to a maximum observation size?
-			return exectypes.NewObservation(groupedCommits, nil, nil).Encode()
+			return exectypes.NewObservation(groupedCommits, nil, nil, nil).Encode()
 		}
 
 		// No observation for non-dest readers.
 		return types.Observation{}, nil
 	case exectypes.GetMessages:
 		// Phase 2: Gather messages from the source chains and build the execution report.
 		messages := make(exectypes.MessageObservations)
+
 		if len(previousOutcome.PendingCommitReports) == 0 {
 			p.lggr.Debug("TODO: No reports to execute. This is expected after a cold start.")
 			// No reports to execute.
@@ -233,6 +241,14 @@ func (p *Plugin) Observation(
 			}
 		}
 
+		// We need to observe the token prices for fee tokens so that we can calculate the fees in USD for each
+		// message and compare them to the execution cost for each message in Outcome. If the paid fees are less than
+		// the execution cost, we will not execute the message.
+		feeTokenPriceObservation, err := observeFeeTokenPrices(ctx, p.onChainTokenPricesReader, messages)
+		if err != nil {
+			return types.Observation{}, fmt.Errorf("unable to get token prices: %w", err)
+		}
+
 		// Regroup the commit reports back into the observation format.
 		// TODO: use same format for Observation and Outcome.
 		groupedCommits := make(exectypes.CommitObservations)
@@ -244,7 +260,7 @@ func (p *Plugin) Observation(
 		}
 
 		// TODO: Fire off messages for an attestation check service.
-		return exectypes.NewObservation(groupedCommits, messages, nil).Encode()
+		return exectypes.NewObservation(groupedCommits, messages, nil, feeTokenPriceObservation).Encode()
 
 	case exectypes.Filter:
 		// Phase 3: observe nonce for each unique source/sender pair.
@@ -274,7 +290,7 @@ func (p *Plugin) Observation(
 			nonceObservations[srcChain] = nonces
 		}
 
-		return exectypes.NewObservation(nil, nil, nonceObservations).Encode()
+		return exectypes.NewObservation(nil, nil, nonceObservations, nil).Encode()
 	default:
 		return types.Observation{}, fmt.Errorf("unknown state")
 	}
@@ -419,7 +435,9 @@ func (p *Plugin) Outcome(
 			observation.Nonces,
 			p.cfg.DestChain,
 			uint64(maxReportSizeBytes),
-			p.cfg.OffchainConfig.BatchGasLimit)
+			p.cfg.OffchainConfig.BatchGasLimit,
+			p.cfg.OffchainConfig.RelativeBoostPerWaitHour)
+
 		outcomeReports, commitReports, err := selectReport(
 			p.lggr,
 			commitReports,

execute/plugin_e2e_test.go

@@ -17,6 +17,10 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 
+	"github.com/smartcontractkit/chainlink-ccip/execute/internal/gas"
+
+	reader_mock "github.com/smartcontractkit/chainlink-ccip/mocks/internal_/reader"
+
 	"github.com/smartcontractkit/chainlink-ccip/chainconfig"
 	"github.com/smartcontractkit/chainlink-ccip/execute/exectypes"
 	"github.com/smartcontractkit/chainlink-ccip/execute/internal/gas/evm"
@@ -143,6 +147,7 @@ func makeMsg(seqNum cciptypes.SeqNum, src, dest cciptypes.ChainSelector, execute
 				SourceChainSelector: src,
 				SequenceNumber:      seqNum,
 			},
+			FeeTokenAmount: cciptypes.NewBigIntFromInt64(100),
 		},
 		Destination: dest,
 		Executed:    executed,
@@ -239,11 +244,19 @@ func setupSimpleTest(
 	tokenDataReader := mock_types.NewMockTokenDataReader(t)
 	tokenDataReader.On("ReadTokenData", mock.Anything, mock.Anything, mock.Anything).Return([][]byte{}, nil)
 
+	tokenPricesReader := reader_mock.NewMockPriceReader(t)
+	tokenPricesReader.On("GetTokenFeedPricesUSD", mock.Anything, mock.Anything).Return([]cciptypes.TokenPrice{}, nil)
+
+	messageExecutionCostEstimator := gas.NewStaticMessageExecutionCostEstimator(cciptypes.NewBigIntFromInt64(0))
+
 	oracleIDToP2pID := GetP2pIDs(1, 2, 3)
 	nodes := []nodeSetup{
-		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, oracleIDToP2pID, 1, 1),
-		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, oracleIDToP2pID, 2, 1),
-		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, oracleIDToP2pID, 3, 1),
+		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, tokenPricesReader,
+			messageExecutionCostEstimator, oracleIDToP2pID, 1, 1),
+		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, tokenPricesReader,
+			messageExecutionCostEstimator, oracleIDToP2pID, 2, 1),
+		newNode(ctx, t, lggr, cfg, msgHasher, ccipReader, homeChain, tokenDataReader, tokenPricesReader,
+			messageExecutionCostEstimator, oracleIDToP2pID, 3, 1),
 	}
 
 	err = homeChain.Close()
@@ -262,6 +275,8 @@ func newNode(
 	ccipReader readerpkg.CCIPReader,
 	homeChain reader.HomeChain,
 	tokenDataReader exectypes.TokenDataReader,
+	onChainTokenPricesReader reader.PriceReader,
+	messageExecutionCostEstimator gas.MessageExecutionCostEstimator,
 	oracleIDToP2pID map[commontypes.OracleID]libocrtypes.PeerID,
 	id int,
 	N int,
@@ -283,7 +298,9 @@ func newNode(
 		homeChain,
 		tokenDataReader,
 		evm.EstimateProvider{},
-		lggr)
+		lggr,
+		onChainTokenPricesReader,
+		messageExecutionCostEstimator)
 
 	return nodeSetup{
 		node:        node1,