Don't require finality when fetching CCIPSendRequested in the execution phase

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

@@ -239,7 +239,7 @@ func (r *CommitReportingPlugin) calculateMinMaxSequenceNumbers(ctx context.Conte
 		return 0, 0, err
 	}
 
-	msgRequests, err := r.onRampReader.GetSendRequestsBetweenSeqNums(ctx, nextInflightMin, nextInflightMin+OnRampMessagesScanLimit)
+	msgRequests, err := r.onRampReader.GetSendRequestsBetweenSeqNums(ctx, nextInflightMin, nextInflightMin+OnRampMessagesScanLimit, true)
 	if err != nil {
 		return 0, 0, err
 	}
@@ -696,7 +696,7 @@ func (r *CommitReportingPlugin) buildReport(ctx context.Context, lggr logger.Log
 
 	// Logs are guaranteed to be in order of seq num, since these are finalized logs only
 	// and the contract's seq num is auto-incrementing.
-	sendRequests, err := r.onRampReader.GetSendRequestsBetweenSeqNums(ctx, interval.Min, interval.Max)
+	sendRequests, err := r.onRampReader.GetSendRequestsBetweenSeqNums(ctx, interval.Min, interval.Max, true)
 	if err != nil {
 		return ccipdata.CommitStoreReport{}, err
 	}

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

@@ -103,7 +103,7 @@ func TestCommitReportingPlugin_Observation(t *testing.T) {
 
 			onRampReader := ccipdatamocks.NewOnRampReader(t)
 			if len(tc.sendReqs) > 0 {
-				onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.commitStoreSeqNum, tc.commitStoreSeqNum+OnRampMessagesScanLimit).
+				onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.commitStoreSeqNum, tc.commitStoreSeqNum+OnRampMessagesScanLimit, true).
 					Return(tc.sendReqs, nil)
 			}
 
@@ -269,7 +269,7 @@ func TestCommitReportingPlugin_Report(t *testing.T) {
 
 			onRampReader := ccipdatamocks.NewOnRampReader(t)
 			if len(tc.sendRequests) > 0 {
-				onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expSeqNumRange.Min, tc.expSeqNumRange.Max).Return(tc.sendRequests, nil)
+				onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expSeqNumRange.Min, tc.expSeqNumRange.Max, true).Return(tc.sendRequests, nil)
 			}
 
 			gasPriceEstimator := prices.NewMockGasPriceEstimatorCommit(t)
@@ -1318,7 +1318,7 @@ func TestCommitReportingPlugin_calculateMinMaxSequenceNumbers(t *testing.T) {
 					},
 				})
 			}
-			onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMin+OnRampMessagesScanLimit).Return(sendReqs, nil)
+			onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMin+OnRampMessagesScanLimit, true).Return(sendReqs, nil)
 			p.onRampReader = onRampReader
 
 			minSeqNum, maxSeqNum, err := p.calculateMinMaxSequenceNumbers(ctx, lggr)

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

@@ -17,7 +17,8 @@ func getProofData(
 	sourceReader ccipdata.OnRampReader,
 	interval ccipdata.CommitStoreInterval,
 ) (sendReqsInRoot []ccipdata.Event[internal.EVM2EVMMessage], leaves [][32]byte, tree *merklemulti.Tree[[32]byte], err error) {
-	sendReqs, err := sourceReader.GetSendRequestsBetweenSeqNums(ctx, interval.Min, interval.Max)
+	// We don't need to double-check if logs are finalized because we already checked that in the Commit phase.
+	sendReqs, err := sourceReader.GetSendRequestsBetweenSeqNums(ctx, interval.Min, interval.Max, false)
 	if err != nil {
 		return nil, nil, nil, err
 	}

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

@@ -768,7 +768,8 @@ func (r *ExecutionReportingPlugin) getReportsWithSendRequests(
 
 	var sendRequests []ccipdata.Event[internal.EVM2EVMMessage]
 	eg.Go(func() error {
-		sendReqs, err := r.config.onRampReader.GetSendRequestsBetweenSeqNums(ctx, intervalMin, intervalMax)
+		// We don't need to double-check if logs are finalized because we already checked that in the Commit phase.
+		sendReqs, err := r.config.onRampReader.GetSendRequestsBetweenSeqNums(ctx, intervalMin, intervalMax, false)
 		if err != nil {
 			return err
 		}

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

@@ -134,7 +134,7 @@ func TestExecutionReportingPlugin_Observation(t *testing.T) {
 			p.config.offRampReader = mockOffRampReader
 
 			mockOnRampReader := ccipdatamocks.NewOnRampReader(t)
-			mockOnRampReader.On("GetSendRequestsBetweenSeqNums", ctx, mock.Anything, mock.Anything).
+			mockOnRampReader.On("GetSendRequestsBetweenSeqNums", ctx, mock.Anything, mock.Anything, false).
 				Return(tc.sendRequests, nil).Maybe()
 			p.config.onRampReader = mockOnRampReader
 
@@ -396,7 +396,7 @@ func TestExecutionReportingPlugin_buildReport(t *testing.T) {
 		sendReqs[i] = ccipdata.Event[internal.EVM2EVMMessage]{Data: msg}
 	}
 	sourceReader.On("GetSendRequestsBetweenSeqNums",
-		ctx, observations[0].SeqNr, observations[len(observations)-1].SeqNr).Return(sendReqs, nil)
+		ctx, observations[0].SeqNr, observations[len(observations)-1].SeqNr, false).Return(sendReqs, nil)
 	p.config.onRampReader = sourceReader
 
 	execReport, err := p.buildReport(ctx, p.lggr, observations)
@@ -1010,7 +1010,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) {
 			p.config.offRampReader = offRampReader
 
 			sourceReader := ccipdatamocks.NewOnRampReader(t)
-			sourceReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMax).
+			sourceReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMax, false).
 				Return(tc.onchainEvents, nil).Maybe()
 			p.config.onRampReader = sourceReader
 

core/services/ocr2/plugins/ccip/internal/ccipdata/onramp_reader.go

@@ -41,7 +41,7 @@ type OnRampDynamicConfig struct {
 type OnRampReader interface {
 	Closer
 	// GetSendRequestsBetweenSeqNums returns all the finalized message send requests in the provided sequence numbers range (inclusive).
-	GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64) ([]Event[internal.EVM2EVMMessage], error)
+	GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]Event[internal.EVM2EVMMessage], error)
 	// Get router configured in the onRamp
 	RouterAddress() (common.Address, error)
 	Address() (common.Address, error)
@@ -66,3 +66,10 @@ func NewOnRampReader(lggr logger.Logger, sourceSelector, destSelector uint64, on
 		return nil, errors.Errorf("got unexpected version %v", version.String())
 	}
 }
+
+func logsConfirmations(finalized bool) logpoller.Confirmations {
+	if finalized {
+		return logpoller.Finalized
+	}
+	return logpoller.Confirmations(0)
+}

core/services/ocr2/plugins/ccip/internal/ccipdata/onramp_reader_test.go

@@ -324,7 +324,7 @@ func testOnRampReader(t *testing.T, th onRampReaderTH, expectedRouterAddress com
 	require.NoError(t, err)
 	require.Equal(t, expectedRouterAddress, res)
 
-	msg, err := th.reader.GetSendRequestsBetweenSeqNums(ctx, 0, 10)
+	msg, err := th.reader.GetSendRequestsBetweenSeqNums(ctx, 0, 10, true)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
 	require.Equal(t, []ccipdata.Event[internal.EVM2EVMMessage]{}, msg)

core/services/ocr2/plugins/ccip/internal/ccipdata/onramp_v1_0_0.go

@@ -211,14 +211,14 @@ func (o *OnRampV1_0_0) logToMessage(log types.Log) (*internal.EVM2EVMMessage, er
 	}, nil
 }
 
-func (o *OnRampV1_0_0) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64) ([]Event[internal.EVM2EVMMessage], error) {
+func (o *OnRampV1_0_0) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]Event[internal.EVM2EVMMessage], error) {
 	logs, err := o.lp.LogsDataWordRange(
 		o.sendRequestedEventSig,
 		o.address,
 		o.sendRequestedSeqNumberWord,
 		logpoller.EvmWord(seqNumMin),
 		logpoller.EvmWord(seqNumMax),
-		logpoller.Finalized,
+		logsConfirmations(finalized),
 		pg.WithParentCtx(ctx))
 	if err != nil {
 		return nil, err

core/services/ocr2/plugins/ccip/internal/ccipdata/onramp_v1_2_0.go

@@ -270,14 +270,14 @@ func (o *OnRampV1_2_0) logToMessage(log types.Log) (*internal.EVM2EVMMessage, er
 	}, nil
 }
 
-func (o *OnRampV1_2_0) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64) ([]Event[internal.EVM2EVMMessage], error) {
+func (o *OnRampV1_2_0) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]Event[internal.EVM2EVMMessage], error) {
 	logs, err := o.lp.LogsDataWordRange(
 		o.sendRequestedEventSig,
 		o.address,
 		o.sendRequestedSeqNumberWord,
 		logpoller.EvmWord(seqNumMin),
 		logpoller.EvmWord(seqNumMax),
-		logpoller.Finalized,
+		logsConfirmations(finalized),
 		pg.WithParentCtx(ctx))
 	if err != nil {
 		return nil, err

core/services/ocr2/plugins/ccip/internal/ccipdata/onramp_v1_2_0_test.go

@@ -89,21 +89,36 @@ func TestLogPollerClient_GetSendRequestsBetweenSeqNums(t *testing.T) {
 	limit := uint64(10)
 	lggr := logger.TestLogger(t)
 
-	lp := mocks.NewLogPoller(t)
-	onRampV2, err := NewOnRampV1_2_0(lggr, 1, 1, onRampAddr, lp, nil)
-	require.NoError(t, err)
-	lp.On("LogsDataWordRange",
-		onRampV2.sendRequestedEventSig,
-		onRampAddr,
-		onRampV2.sendRequestedSeqNumberWord,
-		abihelpers.EvmWord(seqNum),
-		abihelpers.EvmWord(seqNum+limit),
-		logpoller.Finalized,
-		mock.Anything,
-	).Return([]logpoller.Log{}, nil)
-
-	events, err := onRampV2.GetSendRequestsBetweenSeqNums(context.Background(), seqNum, seqNum+limit)
-	assert.NoError(t, err)
-	assert.Empty(t, events)
-	lp.AssertExpectations(t)
+	tests := []struct {
+		name          string
+		finalized     bool
+		confirmations logpoller.Confirmations
+	}{
+		{"finalized", true, logpoller.Finalized},
+		{"unfinalized", false, logpoller.Confirmations(0)},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			lp := mocks.NewLogPoller(t)
+			onRampV2, err := NewOnRampV1_2_0(lggr, 1, 1, onRampAddr, lp, nil)
+			require.NoError(t, err)
+
+			lp.On("LogsDataWordRange",
+				onRampV2.sendRequestedEventSig,
+				onRampAddr,
+				onRampV2.sendRequestedSeqNumberWord,
+				abihelpers.EvmWord(seqNum),
+				abihelpers.EvmWord(seqNum+limit),
+				tt.confirmations,
+				mock.Anything,
+			).Once().Return([]logpoller.Log{}, nil)
+
+			events, err1 := onRampV2.GetSendRequestsBetweenSeqNums(context.Background(), seqNum, seqNum+limit, tt.finalized)
+			assert.NoError(t, err1)
+			assert.Empty(t, events)
+
+			lp.AssertExpectations(t)
+		})
+	}
 }

core/services/ocr2/plugins/ccip/internal/observability/onramp.go

@@ -27,9 +27,9 @@ func NewObservedOnRampReader(origin ccipdata.OnRampReader, chainID int64, plugin
 	}
 }
 
-func (o ObservedOnRampReader) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64) ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
+func (o ObservedOnRampReader) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
 	return withObservedInteractionAndResults(o.metric, "GetSendRequestsBetweenSeqNums", func() ([]ccipdata.Event[internal.EVM2EVMMessage], error) {
-		return o.OnRampReader.GetSendRequestsBetweenSeqNums(ctx, seqNumMin, seqNumMax)
+		return o.OnRampReader.GetSendRequestsBetweenSeqNums(ctx, seqNumMin, seqNumMax, finalized)
 	})
 }