Switching to FilteredLogs method in LogPoller for send requests logs

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

@@ -8,9 +8,12 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/query"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/hashutil"
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip"
+
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/client"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/arm_contract"
@@ -165,6 +168,56 @@ 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(
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Min).Hex(), Operator: primitives.Gte},
+			}),
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Max).Hex(), Operator: primitives.Lte},
+			}),
+		))
+	}
+
+	sendRequestsQuery, err := query.Where(
+		o.address.String(),
+		logpoller.NewAddressFilter(o.address),
+		logpoller.NewEventSigFilter(o.sendRequestedEventSig),
+		query.Or(seqNrRanges...),
+		query.Confidence(ccipdata.LogsConfidence(finalized)),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	logs, err := o.lp.FilteredLogs(
+		ctx,
+		sendRequestsQuery,
+		query.NewLimitAndSort(query.Limit{}, query.NewSortBySequence(query.Asc)),
+		"GetSendRequestsBetweenSeqNums",
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	parsedLogs, err := ccipdata.ParseLogs[cciptypes.EVM2EVMMessage](logs, o.lggr, o.logToMessage)
+	if err != nil {
+		return nil, err
+	}
+
+	res := make([]cciptypes.EVM2EVMMessageWithTxMeta, 0, len(logs))
+	for _, log := range parsedLogs {
+		res = append(res, cciptypes.EVM2EVMMessageWithTxMeta{
+			TxMeta:         log.TxMeta,
+			EVM2EVMMessage: log.Data,
+		})
+	}
+
+	return res, nil
+}
+
 func (o *OnRamp) RouterAddress(context.Context) (cciptypes.Address, error) {
 	config, err := o.onRamp.GetDynamicConfig(nil)
 	if err != nil {

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

@@ -9,9 +9,12 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/query"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/hashutil"
 	cciptypes "github.com/smartcontractkit/chainlink-common/pkg/types/ccip"
+
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/client"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/arm_contract"
@@ -183,6 +186,56 @@ 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(
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Min).Hex(), Operator: primitives.Gte},
+			}),
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Max).Hex(), Operator: primitives.Lte},
+			}),
+		))
+	}
+
+	sendRequestsQuery, err := query.Where(
+		o.address.String(),
+		logpoller.NewAddressFilter(o.address),
+		logpoller.NewEventSigFilter(o.sendRequestedEventSig),
+		query.Or(seqNrRanges...),
+		query.Confidence(ccipdata.LogsConfidence(finalized)),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	logs, err := o.lp.FilteredLogs(
+		ctx,
+		sendRequestsQuery,
+		query.NewLimitAndSort(query.Limit{}, query.NewSortBySequence(query.Asc)),
+		"GetSendRequestsBetweenSeqNums",
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	parsedLogs, err := ccipdata.ParseLogs[cciptypes.EVM2EVMMessage](logs, o.lggr, o.logToMessage)
+	if err != nil {
+		return nil, err
+	}
+
+	res := make([]cciptypes.EVM2EVMMessageWithTxMeta, 0, len(logs))
+	for _, log := range parsedLogs {
+		res = append(res, cciptypes.EVM2EVMMessageWithTxMeta{
+			TxMeta:         log.TxMeta,
+			EVM2EVMMessage: log.Data,
+		})
+	}
+
+	return res, nil
+}
+
 func (o *OnRamp) RouterAddress(context.Context) (cciptypes.Address, error) {
 	config, err := o.onRamp.GetDynamicConfig(nil)
 	if err != nil {

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

@@ -161,16 +161,52 @@ func (o *OnRamp) SourcePriceRegistryAddress(ctx context.Context) (cciptypes.Addr
 }
 
 func (o *OnRamp) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, seqNumMax uint64, finalized bool) ([]cciptypes.EVM2EVMMessageWithTxMeta, error) {
+	logs, err := o.lp.LogsDataWordRange(
+		ctx,
+		o.sendRequestedEventSig,
+		o.address,
+		o.sendRequestedSeqNumberWord,
+		logpoller.EvmWord(seqNumMin),
+		logpoller.EvmWord(seqNumMax),
+		ccipdata.LogsConfirmations(finalized),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	parsedLogs, err := ccipdata.ParseLogs[cciptypes.EVM2EVMMessage](logs, o.lggr, o.logToMessage)
+	if err != nil {
+		return nil, err
+	}
+
+	res := make([]cciptypes.EVM2EVMMessageWithTxMeta, 0, len(logs))
+	for _, log := range parsedLogs {
+		res = append(res, cciptypes.EVM2EVMMessageWithTxMeta{
+			TxMeta:         log.TxMeta,
+			EVM2EVMMessage: log.Data,
+		})
+	}
+	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(
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Min).Hex(), Operator: primitives.Gte},
+			}),
+			logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
+				{Value: logpoller.EvmWord(seqNr.Max).Hex(), Operator: primitives.Lte},
+			}),
+		))
+	}
+
 	sendRequestsQuery, err := query.Where(
 		o.address.String(),
 		logpoller.NewAddressFilter(o.address),
 		logpoller.NewEventSigFilter(o.sendRequestedEventSig),
-		logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
-			{Value: logpoller.EvmWord(seqNumMin).Hex(), Operator: primitives.Gte},
-		}),
-		logpoller.NewEventByWordFilter(o.sendRequestedEventSig, uint8(o.sendRequestedSeqNumberWord), []primitives.ValueComparator{
-			{Value: logpoller.EvmWord(seqNumMax).Hex(), Operator: primitives.Lte},
-		}),
+		query.Or(seqNrRanges...),
 		query.Confidence(ccipdata.LogsConfidence(finalized)),
 	)
 	if err != nil {
@@ -180,7 +216,7 @@ func (o *OnRamp) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, s
 	logs, err := o.lp.FilteredLogs(
 		ctx,
 		sendRequestsQuery,
-		query.NewLimitAndSort(query.Limit{}, query.NewSortByBlock(query.Asc)),
+		query.NewLimitAndSort(query.Limit{}, query.NewSortBySequence(query.Asc)),
 		"GetSendRequestsBetweenSeqNums",
 	)
 	if err != nil {
@@ -199,6 +235,7 @@ func (o *OnRamp) GetSendRequestsBetweenSeqNums(ctx context.Context, seqNumMin, s
 			EVM2EVMMessage: log.Data,
 		})
 	}
+
 	return res, nil
 }
 

go.mod

@@ -75,7 +75,7 @@ require (
 	github.com/smartcontractkit/chain-selectors v1.0.17
 	github.com/smartcontractkit/chainlink-automation v1.0.4
 	github.com/smartcontractkit/chainlink-ccip v0.0.0-20240702134133-a6d2f429671f
-	github.com/smartcontractkit/chainlink-common v0.1.7-0.20240625144005-556a7ba7b11c
+	github.com/smartcontractkit/chainlink-common v0.1.7-0.20240703131331-f2ea6168ae2b
 	github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240621143432-85370a54b141
 	github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240220203239-09be0ea34540
 	github.com/smartcontractkit/chainlink-feeds v0.0.0-20240523182126-1784bd4f3294

go.sum

@@ -1088,6 +1088,8 @@ github.com/smartcontractkit/chainlink-ccip v0.0.0-20240702134133-a6d2f429671f h1
 github.com/smartcontractkit/chainlink-ccip v0.0.0-20240702134133-a6d2f429671f/go.mod h1:vy2vEF1K61khqGcbOCrHBKFPvSQW0O1eg19Sv74Xq/4=
 github.com/smartcontractkit/chainlink-common v0.1.7-0.20240625144005-556a7ba7b11c h1:dR2y3uzuZ8cJOaMSij8LQIdySuImwxkiZFjZ62ML5S0=
 github.com/smartcontractkit/chainlink-common v0.1.7-0.20240625144005-556a7ba7b11c/go.mod h1:L32xvCpk84Nglit64OhySPMP1tM3TTBK7Tw0qZl7Sd4=
+github.com/smartcontractkit/chainlink-common v0.1.7-0.20240703131331-f2ea6168ae2b h1:zROL/MaEg2FspTxwX0uXdO7KocuLxs9zTeNopoaBQPY=
+github.com/smartcontractkit/chainlink-common v0.1.7-0.20240703131331-f2ea6168ae2b/go.mod h1:EWvSuqIJUYXZLEHewC7WCaPylM2jyjF3Q36BZPS4MoI=
 github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240621143432-85370a54b141 h1:TMOoYaeSDkkI3jkCH7lKHOZaLkeDuxFTNC+XblD6M0M=
 github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240621143432-85370a54b141/go.mod h1:0UNuO3nDt9MFsZPaHJBEUolxVkN0iC69j1ccDp95e8k=
 github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240220203239-09be0ea34540 h1:xFSv8561jsLtF6gYZr/zW2z5qUUAkcFkApin2mnbYTo=