Limit the number of items retrieved in one EventQuery

pkg/relay/ethereum/config.go

@@ -4,10 +4,11 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
-	"github.com/datachainlab/ethereum-ibc-relay-chain/pkg/utils"
 	"math/big"
 	"strings"
 
+	"github.com/datachainlab/ethereum-ibc-relay-chain/pkg/utils"
+
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/hyperledger-labs/yui-relayer/core"

pkg/relay/ethereum/events.go

@@ -10,12 +10,17 @@ import (
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/hyperledger-labs/yui-relayer/core"
 	"github.com/hyperledger-labs/yui-relayer/log"
 
 	"github.com/datachainlab/ethereum-ibc-relay-chain/pkg/contract/ibchandler"
 )
 
+const (
+	BlocksPerEventQueryDefault = 1000
+)
+
 var (
 	abiGeneratedClientIdentifier,
 	abiGeneratedConnectionIdentifier,
@@ -61,24 +66,11 @@ func (chain *Chain) findSentPackets(ctx core.QueryContext, fromHeight uint64) (c
 		dstPortID = channel.Counterparty.PortId
 		dstChannelID = channel.Counterparty.ChannelId
 	}
-
-	query := ethereum.FilterQuery{
-		FromBlock: big.NewInt(int64(fromHeight)),
-		ToBlock:   big.NewInt(int64(ctx.Height().GetRevisionHeight())),
-		Addresses: []common.Address{
-			chain.config.IBCAddress(),
-		},
-		Topics: [][]common.Hash{{
-			abiSendPacket.ID,
-		}},
-	}
-
-	logs, err := chain.client.FilterLogs(ctx.Context(), query)
+	logs, err := chain.filterLogs(ctx, fromHeight, abiSendPacket)
 	if err != nil {
 		logger.Error("failed to filter logs", err)
 		return nil, err
 	}
-
 	defer logger.TimeTrack(now, "findSentPackets", "num_logs", len(logs))
 
 	var packets core.PacketInfoList
@@ -158,22 +150,10 @@ func (chain *Chain) findReceivedPackets(ctx core.QueryContext, fromHeight uint64
 }
 
 func (chain *Chain) findRecvPacketEvents(ctx core.QueryContext, fromHeight uint64) ([]*ibchandler.IbchandlerRecvPacket, error) {
-	query := ethereum.FilterQuery{
-		FromBlock: big.NewInt(int64(fromHeight)),
-		ToBlock:   big.NewInt(int64(ctx.Height().GetRevisionHeight())),
-		Addresses: []common.Address{
-			chain.config.IBCAddress(),
-		},
-		Topics: [][]common.Hash{{
-			abiRecvPacket.ID,
-		}},
-	}
-
-	logs, err := chain.client.FilterLogs(ctx.Context(), query)
+	logs, err := chain.filterLogs(ctx, fromHeight, abiRecvPacket)
 	if err != nil {
 		return nil, err
 	}
-
 	var events []*ibchandler.IbchandlerRecvPacket
 	for _, log := range logs {
 		event, err := chain.ibcHandler.ParseRecvPacket(log)
@@ -187,22 +167,10 @@ func (chain *Chain) findRecvPacketEvents(ctx core.QueryContext, fromHeight uint6
 }
 
 func (chain *Chain) findWriteAckEvents(ctx core.QueryContext, fromHeight uint64) ([]*ibchandler.IbchandlerWriteAcknowledgement, error) {
-	query := ethereum.FilterQuery{
-		FromBlock: big.NewInt(int64(fromHeight)),
-		ToBlock:   big.NewInt(int64(ctx.Height().GetRevisionHeight())),
-		Addresses: []common.Address{
-			chain.config.IBCAddress(),
-		},
-		Topics: [][]common.Hash{{
-			abiWriteAcknowledgement.ID,
-		}},
-	}
-
-	logs, err := chain.client.FilterLogs(ctx.Context(), query)
+	logs, err := chain.filterLogs(ctx, fromHeight, abiWriteAcknowledgement)
 	if err != nil {
 		return nil, err
 	}
-
 	var events []*ibchandler.IbchandlerWriteAcknowledgement
 	for _, log := range logs {
 		event, err := chain.ibcHandler.ParseWriteAcknowledgement(log)
@@ -225,3 +193,44 @@ func (chain *Chain) GetChannelLogger() *log.RelayLogger {
 	channelID := chain.Path().ChannelID
 	return logger.WithChannel(chainID, portID, channelID)
 }
+
+func (chain *Chain) filterLogs(ctx core.QueryContext, fromHeight uint64, event abi.Event) ([]types.Log, error) {
+	blocksPerEventQuery := chain.config.BlocksPerEventQuery
+	if blocksPerEventQuery == 0 {
+		blocksPerEventQuery = BlocksPerEventQueryDefault
+	}
+
+	toHeight := ctx.Height().GetRevisionHeight()
+	totalBlocks := toHeight - fromHeight + 1
+	loopCount := totalBlocks / blocksPerEventQuery
+	if totalBlocks%blocksPerEventQuery != 0 {
+		loopCount++
+	}
+	var logs []types.Log
+	for i := uint64(0); i < loopCount; i++ {
+		var endBlockNum uint64
+		if i == loopCount-1 {
+			endBlockNum = toHeight
+		} else {
+			endBlockNum = fromHeight + (i+1)*blocksPerEventQuery - 1
+		}
+		startBlock := big.NewInt(int64(fromHeight + i*blocksPerEventQuery))
+		endBlock := big.NewInt(int64(endBlockNum))
+		query := ethereum.FilterQuery{
+			FromBlock: startBlock,
+			ToBlock:   endBlock,
+			Addresses: []common.Address{
+				chain.config.IBCAddress(),
+			},
+			Topics: [][]common.Hash{{
+				event.ID,
+			}},
+		}
+		filterLogs, err := chain.client.FilterLogs(ctx.Context(), query)
+		if err != nil {
+			return nil, err
+		}
+		logs = append(logs, filterLogs...)
+	}
+	return logs, nil
+}

proto/relayer/chains/ethereum/config/config.proto

@@ -34,6 +34,8 @@ message ChainConfig {
 
   string tx_type = 14;
   DynamicTxGasConfig dynamic_tx_gas_config = 15;
+
+  uint64 blocks_per_event_query = 16;
 }
 
 message AllowLCFunctionsConfig {