Fixing tests

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

@@ -178,7 +178,7 @@ func TestCommitReportingPlugin_Observation(t *testing.T) {
 			onRampReader.On("IsSourceChainHealthy", ctx).Return(true, nil)
 			onRampReader.On("IsSourceCursed", ctx).Return(tc.sourceChainCursed, nil)
 			if len(tc.sendReqs) > 0 {
-				onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.commitStoreSeqNum, tc.commitStoreSeqNum+OnRampMessagesScanLimit, true).
+				onRampReader.On("GetSendRequestsForSeqNums", ctx, []cciptypes.SequenceNumberRange{{Min: tc.commitStoreSeqNum, Max: tc.commitStoreSeqNum + OnRampMessagesScanLimit}}, true).
 					Return(tc.sendReqs, nil)
 			}
 
@@ -405,7 +405,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, true).Return(tc.sendRequests, nil)
+				onRampReader.On("GetSendRequestsForSeqNums", ctx, []cciptypes.SequenceNumberRange{{Min: tc.expSeqNumRange.Min, Max: tc.expSeqNumRange.Max}}, true).Return(tc.sendRequests, nil)
 			}
 
 			evmEstimator := mocks.NewEvmFeeEstimator(t)
@@ -1569,7 +1569,7 @@ func TestCommitReportingPlugin_calculateMinMaxSequenceNumbers(t *testing.T) {
 					},
 				})
 			}
-			onRampReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMin+OnRampMessagesScanLimit, true).Return(sendReqs, nil)
+			onRampReader.On("GetSendRequestsForSeqNums", ctx, []cciptypes.SequenceNumberRange{{Min: tc.expQueryMin, Max: tc.expQueryMin + OnRampMessagesScanLimit}}, true).Return(sendReqs, nil)
 			p.onRampReader = onRampReader
 
 			minSeqNum, maxSeqNum, _, err := p.calculateMinMaxSequenceNumbers(ctx, lggr)

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

@@ -168,7 +168,7 @@ func TestExecutionReportingPlugin_Observation(t *testing.T) {
 			p.tokenPoolBatchedReader = tokenPoolBatchedReader
 
 			mockOffRampReader := ccipdatamocks.NewOffRampReader(t)
-			mockOffRampReader.On("GetExecutionStateChangesBetweenSeqNums", ctx, mock.Anything, mock.Anything, 0).
+			mockOffRampReader.On("GetExecutionStateChangesForSeqNums", ctx, mock.Anything, 0).
 				Return(executionEvents, nil).Maybe()
 			mockOffRampReader.On("CurrentRateLimiterState", mock.Anything).Return(tc.rateLimiterState, nil).Maybe()
 			mockOffRampReader.On("Address", ctx).Return(cciptypes.Address(offRamp.Address().String()), nil).Maybe()
@@ -189,7 +189,7 @@ func TestExecutionReportingPlugin_Observation(t *testing.T) {
 			mockOnRampReader := ccipdatamocks.NewOnRampReader(t)
 			mockOnRampReader.On("IsSourceCursed", ctx).Return(tc.sourceChainCursed, nil).Maybe()
 			mockOnRampReader.On("IsSourceChainHealthy", ctx).Return(tc.sourceChainHealthy, nil).Maybe()
-			mockOnRampReader.On("GetSendRequestsBetweenSeqNums", ctx, mock.Anything, mock.Anything, false).
+			mockOnRampReader.On("GetSendRequestsForSeqNums", ctx, mock.Anything, false).
 				Return(tc.sendRequests, nil).Maybe()
 			sourcePriceRegistryAddress := cciptypes.Address(utils.RandomAddress().String())
 			mockOnRampReader.On("SourcePriceRegistryAddress", ctx).Return(sourcePriceRegistryAddress, nil).Maybe()
@@ -449,8 +449,12 @@ func TestExecutionReportingPlugin_buildReport(t *testing.T) {
 		}
 		sendReqs[i] = cciptypes.EVM2EVMMessageWithTxMeta{EVM2EVMMessage: msg}
 	}
-	sourceReader.On("GetSendRequestsBetweenSeqNums",
-		ctx, observations[0].SeqNr, observations[len(observations)-1].SeqNr, false).Return(sendReqs, nil)
+	sourceReader.On(
+		"GetSendRequestsForSeqNums",
+		ctx,
+		[]cciptypes.SequenceNumberRange{{Min: observations[0].SeqNr, Max: observations[len(observations)-1].SeqNr}},
+		false,
+	).Return(sendReqs, nil)
 	p.onRampReader = sourceReader
 
 	execReport, err := p.buildReport(ctx, p.lggr, observations)
@@ -857,8 +861,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) {
 	testCases := []struct {
 		name                string
 		reports             []cciptypes.CommitStoreReport
-		expQueryMin         uint64 // expected min/max used in the query to get ccipevents
-		expQueryMax         uint64
+		expSeqNumbers       []cciptypes.SequenceNumberRange
 		onchainEvents       []cciptypes.EVM2EVMMessageWithTxMeta
 		destExecutedSeqNums []uint64
 
@@ -883,8 +886,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) {
 					MerkleRoot: [32]byte{200},
 				},
 			},
-			expQueryMin: 1,
-			expQueryMax: 3,
+			expSeqNumbers: []cciptypes.SequenceNumberRange{{Min: 1, Max: 2}, {Min: 3, Max: 3}},
 			onchainEvents: []cciptypes.EVM2EVMMessageWithTxMeta{
 				{EVM2EVMMessage: cciptypes.EVM2EVMMessage{SequenceNumber: 1}},
 				{EVM2EVMMessage: cciptypes.EVM2EVMMessage{SequenceNumber: 2}},
@@ -939,7 +941,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) {
 			p.offRampReader = offRampReader
 
 			sourceReader := ccipdatamocks.NewOnRampReader(t)
-			sourceReader.On("GetSendRequestsBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMax, false).
+			sourceReader.On("GetSendRequestsForSeqNums", ctx, tc.expSeqNumbers, false).
 				Return(tc.onchainEvents, nil).Maybe()
 			p.onRampReader = sourceReader
 
@@ -959,7 +961,7 @@ func TestExecutionReportingPlugin_getReportsWithSendRequests(t *testing.T) {
 					},
 				})
 			}
-			offRampReader.On("GetExecutionStateChangesBetweenSeqNums", ctx, tc.expQueryMin, tc.expQueryMax, 0).Return(executedEvents, nil).Maybe()
+			offRampReader.On("GetExecutionStateChangesForSeqNums", ctx, tc.expSeqNumbers, 0).Return(executedEvents, nil).Maybe()
 
 			populatedReports, err := p.getReportsWithSendRequests(ctx, tc.reports)
 			if tc.expErr {

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

@@ -416,7 +416,7 @@ func testOnRampReader(t *testing.T, th onRampReaderTH, expectedRouterAddress com
 	require.NoError(t, err)
 	require.Equal(t, ccipcalc.EvmAddrToGeneric(expectedRouterAddress), res)
 
-	msg, err := th.reader.GetSendRequestsBetweenSeqNums(ctx, 0, 10, true)
+	msg, err := th.reader.GetSendRequestsForSeqNums(ctx, []cciptypes.SequenceNumberRange{{Min: 0, Max: 10}}, true)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
 	require.Equal(t, []cciptypes.EVM2EVMMessageWithTxMeta{}, msg)

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

@@ -51,7 +51,6 @@ func LogsConfidence(finalized bool) primitives.ConfidenceLevel {
 		return primitives.Finalized
 	}
 	return primitives.Unconfirmed
-
 }
 
 func LogsConfirmations(finalized bool) evmtypes.Confirmations {

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

@@ -21,7 +21,6 @@ import (
 	evmclimocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client/mocks"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller/mocks"
-	evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/evm_2_evm_offramp_1_0_0"
 	mock_contracts "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/mocks/v1_0_0"
@@ -269,14 +268,7 @@ func Test_GetExecutionStateChangesForSeqNums(t *testing.T) {
 }
 
 func Test_LogsAreProperlyMarkedAsFinalized(t *testing.T) {
-	minSeqNr := uint64(10)
-	maxSeqNr := uint64(14)
-	inputLogs := []logpoller.Log{
-		CreateExecutionStateChangeEventLog(t, nil, utils.RandomAddress(), 10, 2, 0, utils.RandomBytes32()),
-		CreateExecutionStateChangeEventLog(t, nil, utils.RandomAddress(), 11, 3, 0, utils.RandomBytes32()),
-		CreateExecutionStateChangeEventLog(t, nil, utils.RandomAddress(), 12, 5, 0, utils.RandomBytes32()),
-		CreateExecutionStateChangeEventLog(t, nil, utils.RandomAddress(), 14, 7, 0, utils.RandomBytes32()),
-	}
+	ctx := testutils.Context(t)
 
 	tests := []struct {
 		name                        string
@@ -302,17 +294,30 @@ func Test_LogsAreProperlyMarkedAsFinalized(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			offrampAddress := utils.RandomAddress()
+			chainID := testutils.NewRandomEVMChainID()
+			orm := logpoller.NewORM(chainID, pgtest.NewSqlxDB(t), logger.TestLogger(t))
+			lpOpts := logpoller.Opts{
+				PollPeriod:               time.Hour,
+				FinalityDepth:            2,
+				BackfillBatchSize:        20,
+				RpcBatchSize:             10,
+				KeepFinalizedBlocksDepth: 1000,
+			}
+			lp := logpoller.NewLogPoller(orm, nil, logger.TestLogger(t), lpOpts)
 
-			lp := mocks.NewLogPoller(t)
-			lp.On("LatestBlock", mock.Anything).
-				Return(logpoller.LogPollerBlock{FinalizedBlockNumber: int64(tt.lastFinalizedBlock)}, nil)
-			lp.On("IndexedLogsTopicRange", mock.Anything, ExecutionStateChangedEvent, offrampAddress, 1, logpoller.EvmWord(minSeqNr), logpoller.EvmWord(maxSeqNr), evmtypes.Confirmations(0)).
-				Return(inputLogs, nil)
+			offrampAddress := utils.RandomAddress()
+			inputLogs := []logpoller.Log{
+				CreateExecutionStateChangeEventLog(t, chainID, offrampAddress, 10, 2, 0, utils.RandomBytes32()),
+				CreateExecutionStateChangeEventLog(t, chainID, offrampAddress, 11, 3, 0, utils.RandomBytes32()),
+				CreateExecutionStateChangeEventLog(t, chainID, offrampAddress, 12, 5, 0, utils.RandomBytes32()),
+				CreateExecutionStateChangeEventLog(t, chainID, offrampAddress, 14, 7, 0, utils.RandomBytes32()),
+			}
+			err := orm.InsertLogsWithBlock(ctx, inputLogs, logpoller.NewLogPollerBlock(utils.RandomBytes32(), 100, time.Now(), int64(tt.lastFinalizedBlock)))
+			require.NoError(t, err)
 
 			offRamp, err := NewOffRamp(logger.TestLogger(t), offrampAddress, evmclimocks.NewClient(t), lp, nil, nil)
 			require.NoError(t, err)
-			logs, err := offRamp.GetExecutionStateChangesForSeqNums(testutils.Context(t), []cciptypes.SequenceNumberRange{{Min: minSeqNr, Max: maxSeqNr}}, 0)
+			logs, err := offRamp.GetExecutionStateChangesForSeqNums(testutils.Context(t), []cciptypes.SequenceNumberRange{{Min: 0, Max: 100}}, 0)
 			require.NoError(t, err)
 			assert.Len(t, logs, len(inputLogs))
 

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

@@ -186,10 +186,10 @@ func (o *OnRamp) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, s
 	return res, nil
 }
 
-func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNrs []cciptypes.SequenceNumberRange, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
-	seqNrRanges := make([]query.Expression, 0, len(seqNrs))
-	for _, seqNr := range seqNrs {
-		seqNrRanges = append(seqNrRanges, query.And(
+func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNums []cciptypes.SequenceNumberRange, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
+	seqNumRanges := make([]query.Expression, 0, len(seqNums))
+	for _, seqNr := range seqNums {
+		seqNumRanges = append(seqNumRanges, query.And(
 			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
 				{Value: logpoller.EvmWord(seqNr.Min).Hex(), Operator: primitives.Gte},
 			}),
@@ -199,11 +199,19 @@ func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNrs []cciptyp
 		))
 	}
 
+	// TODO Move to chainlink-common, querying layer should cover cases like these
+	var seqNumsFilter query.Expression
+	if len(seqNumRanges) == 1 {
+		seqNumsFilter = seqNumRanges[0]
+	} else {
+		seqNumsFilter = query.Or(seqNumRanges...)
+	}
+
 	sendRequestsQuery, err := query.Where(
 		o.address.String(),
 		logpoller.NewAddressFilter(o.address),
 		logpoller.NewEventSigFilter(o.sendRequestedEventSig),
-		query.Or(seqNrRanges...),
+		seqNumsFilter,
 		query.Confidence(ccipdata.LogsConfidence(finalized)),
 	)
 	if err != nil {

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

@@ -189,10 +189,10 @@ func (o *OnRamp) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, s
 	return res, nil
 }
 
-func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNrs []cciptypes.SequenceNumberRange, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
-	seqNrRanges := make([]query.Expression, 0, len(seqNrs))
-	for _, seqNr := range seqNrs {
-		seqNrRanges = append(seqNrRanges, query.And(
+func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNums []cciptypes.SequenceNumberRange, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
+	seqNumRanges := make([]query.Expression, 0, len(seqNums))
+	for _, seqNr := range seqNums {
+		seqNumRanges = append(seqNumRanges, query.And(
 			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
 				{Value: logpoller.EvmWord(seqNr.Min).Hex(), Operator: primitives.Gte},
 			}),
@@ -202,11 +202,19 @@ func (o *OnRamp) GetSendRequestsForSeqNums(ctx context.Context, seqNrs []cciptyp
 		))
 	}
 
+	// TODO Move to chainlink-common, querying layer should cover cases like these
+	var seqNumsFilter query.Expression
+	if len(seqNumRanges) == 1 {
+		seqNumsFilter = seqNumRanges[0]
+	} else {
+		seqNumsFilter = query.Or(seqNumRanges...)
+	}
+
 	sendRequestsQuery, err := query.Where(
 		o.address.String(),
 		logpoller.NewAddressFilter(o.address),
 		logpoller.NewEventSigFilter(o.sendRequestedEventSig),
-		query.Or(seqNrRanges...),
+		seqNumsFilter,
 		query.Confidence(ccipdata.LogsConfidence(finalized)),
 	)
 	if err != nil {

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

@@ -26,9 +26,9 @@ func NewObservedOnRampReader(origin ccipdata.OnRampReader, chainID int64, plugin
 	}
 }
 
-func (o ObservedOnRampReader) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
-	return withObservedInteractionAndResults(o.metric, "GetSendRequestsBetweenSeqNums", func() ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
-		return o.OnRampReader.GetSendRequestsBetweenSeqNums(ctx, seqNumMin, seqNumMax, finalized)
+func (o ObservedOnRampReader) GetSendRequestsForSeqNums(ctx context.Context, seqNums []cciptypes.SequenceNumberRange, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
+	return withObservedInteractionAndResults(o.metric, "GetSendRequestsForSeqNums", func() ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
+		return o.OnRampReader.GetSendRequestsForSeqNums(ctx, seqNums, finalized)
 	})
 }
 

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

@@ -35,6 +35,7 @@ func TestOnRampObservedMethods(t *testing.T) {
 	excludedMethods := []string{
 		"Address",
 		"Close",
+		"GetSendRequestsBetweenSeqNums", // It's deprecated, going to be removed soon
 	}
 
 	// Defines the overridden method calls to test.
@@ -45,9 +46,9 @@ func TestOnRampObservedMethods(t *testing.T) {
 		Arguments:  []interface{}{testutils.Context(t)},
 		Returns:    []interface{}{cciptypes.OnRampDynamicConfig{}, nil},
 	}
-	methodCalls["GetSendRequestsBetweenSeqNums"] = MethodCall{
-		MethodName: "GetSendRequestsBetweenSeqNums",
-		Arguments:  []interface{}{testutils.Context(t), uint64(0), uint64(100), true},
+	methodCalls["GetSendRequestsForSeqNums"] = MethodCall{
+		MethodName: "GetSendRequestsForSeqNums",
+		Arguments:  []interface{}{testutils.Context(t), []cciptypes.SequenceNumberRange{{Min: 0, Max: 100}}, true},
 		Returns:    []interface{}{nil, nil},
 	}
 	methodCalls["IsSourceChainHealthy"] = MethodCall{