concurrent get blocks

scroll-tech · Dec 12, 2023 · 55ff91f · 55ff91f
1 parent b52f665
commit 55ff91f
Show file tree

Hide file tree

Showing 3 changed files with 91 additions and 20 deletions.
diff --git a/bridge-history-api/internal/controller/fetcher/l1_fetcher.go b/bridge-history-api/internal/controller/fetcher/l1_fetcher.go
@@ -127,13 +127,13 @@ func (c *L1MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 	log.Info("fetch and save L1 events", "from", from, "to", to)
 	var l1FailedGatewayRouterTxs []*orm.CrossMessage
 	blockTimestampsMap := make(map[uint64]uint64)
-	for number := from; number <= to; number++ {
-		blockNumber := new(big.Int).SetUint64(number)
-		block, err := c.client.BlockByNumber(ctx, blockNumber)
-		if err != nil {
-			log.Error("failed to get block by number", "number", blockNumber.String(), "err", err)
-			return err
-		}
+	blocks, err := utils.GetL1BlocksInRange(c.ctx, c.client, from, to)
+	if err != nil {
+		log.Error("failed to get L1 blocks in range", "from", from, "to", to, "err", err)
+		return err
+	}
+	for i := from; i <= to; i++ {
+		block := blocks[i-from]
 		blockTimestampsMap[block.NumberU64()] = block.Time()
 
 		for _, tx := range block.Transactions() {
@@ -143,7 +143,8 @@ func (c *L1MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 			}
 			toAddress := to.String()
 			if toAddress == c.cfg.GatewayRouterAddr {
-				receipt, err := c.client.TransactionReceipt(ctx, tx.Hash())
+				var receipt *types.Receipt
+				receipt, err = c.client.TransactionReceipt(ctx, tx.Hash())
 				if err != nil {
 					log.Error("Failed to get transaction receipt", "txHash", tx.Hash().String(), "err", err)
 					return err
@@ -152,7 +153,8 @@ func (c *L1MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 				// Check if the transaction failed
 				if receipt.Status == types.ReceiptStatusFailed {
 					signer := types.NewLondonSigner(new(big.Int).SetUint64(tx.ChainId().Uint64()))
-					sender, err := signer.Sender(tx)
+					var sender common.Address
+					sender, err = signer.Sender(tx)
 					if err != nil {
 						log.Error("get sender failed", "chain id", tx.ChainId().Uint64(), "tx hash", tx.Hash().String(), "err", err)
 						return err

diff --git a/bridge-history-api/internal/controller/fetcher/l2_fetcher.go b/bridge-history-api/internal/controller/fetcher/l2_fetcher.go
@@ -12,7 +12,6 @@ import (
 	"github.com/scroll-tech/go-ethereum/crypto"
 	"github.com/scroll-tech/go-ethereum/ethclient"
 	"github.com/scroll-tech/go-ethereum/log"
-	"github.com/scroll-tech/go-ethereum/rpc"
 	"gorm.io/gorm"
 
 	backendabi "scroll-tech/bridge-history-api/abi"
@@ -120,13 +119,13 @@ func (c *L2MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 	var l2FailedGatewayRouterTxs []*orm.CrossMessage
 	var l2RevertedRelayedMessages []*orm.CrossMessage
 	blockTimestampsMap := make(map[uint64]uint64)
-	for number := from; number <= to; number++ {
-		blockNumber := new(big.Int).SetUint64(number)
-		block, err := c.client.GetBlockByNumberOrHash(ctx, rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(number)))
-		if err != nil {
-			log.Error("failed to get block by number", "number", blockNumber.String(), "err", err)
-			return err
-		}
+	blocks, err := utils.GetL2BlocksInRange(c.ctx, c.client, from, to)
+	if err != nil {
+		log.Error("failed to get L2 blocks in range", "from", from, "to", to, "err", err)
+		return err
+	}
+	for i := from; i <= to; i++ {
+		block := blocks[i-from]
 		blockTimestampsMap[block.NumberU64()] = block.Time()
 
 		for _, tx := range block.Transactions() {
@@ -136,7 +135,8 @@ func (c *L2MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 			}
 			toAddress := to.String()
 			if toAddress == c.cfg.GatewayRouterAddr {
-				receipt, err := c.client.TransactionReceipt(ctx, tx.Hash())
+				var receipt *types.Receipt
+				receipt, err = c.client.TransactionReceipt(ctx, tx.Hash())
 				if err != nil {
 					log.Error("Failed to get transaction receipt", "txHash", tx.Hash().String(), "err", err)
 					return err
@@ -145,7 +145,8 @@ func (c *L2MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 				// Check if the transaction failed
 				if receipt.Status == types.ReceiptStatusFailed {
 					signer := types.NewLondonSigner(new(big.Int).SetUint64(tx.ChainId().Uint64()))
-					sender, err := signer.Sender(tx)
+					var sender common.Address
+					sender, err = signer.Sender(tx)
 					if err != nil {
 						log.Error("get sender failed", "chain id", tx.ChainId().Uint64(), "tx hash", tx.Hash().String(), "err", err)
 						return err
@@ -162,7 +163,8 @@ func (c *L2MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 				}
 			}
 			if tx.Type() == types.L1MessageTxType {
-				receipt, err := c.client.TransactionReceipt(ctx, tx.Hash())
+				var receipt *types.Receipt
+				receipt, err = c.client.TransactionReceipt(ctx, tx.Hash())
 				if err != nil {
 					log.Error("Failed to get transaction receipt", "txHash", tx.Hash().String(), "err", err)
 					return err

diff --git a/bridge-history-api/internal/utils/utils.go b/bridge-history-api/internal/utils/utils.go
@@ -12,6 +12,9 @@ import (
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/crypto"
 	"github.com/scroll-tech/go-ethereum/ethclient"
+	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rpc"
+	"golang.org/x/sync/errgroup"
 
 	backendabi "scroll-tech/bridge-history-api/abi"
 )
@@ -111,3 +114,67 @@ func GetBatchRangeFromCalldata(calldata []byte) (uint64, uint64, error) {
 
 	return startBlock, finishBlock, err
 }
+
+// GetL1BlocksInRange gets a batch of blocks for a block range [start, end] inclusive.
+func GetL1BlocksInRange(ctx context.Context, cli *ethclient.Client, start, end uint64) ([]*types.Block, error) {
+	var (
+		eg          errgroup.Group
+		blocks      = make([]*types.Block, end-start+1)
+		concurrency = 32
+		sem         = make(chan struct{}, concurrency)
+	)
+
+	for i := start; i <= end; i++ {
+		sem <- struct{}{} // Acquire a slot in the semaphore
+		blockNum := int64(i)
+		index := i - start
+		eg.Go(func() error {
+			defer func() { <-sem }() // Release the slot when done
+			block, err := cli.BlockByNumber(ctx, big.NewInt(blockNum))
+			if err != nil {
+				log.Error("Failed to fetch block number", "number", blockNum, "error", err)
+				return err
+			}
+			blocks[index] = block
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		log.Error("Error waiting for block fetching routines", "error", err)
+		return nil, err
+	}
+	return blocks, nil
+}
+
+// GetL2BlocksInRange gets a batch of blocks for a block range [start, end] inclusive.
+func GetL2BlocksInRange(ctx context.Context, cli *ethclient.Client, start, end uint64) ([]*types.BlockWithRowConsumption, error) {
+	var (
+		eg          errgroup.Group
+		blocks      = make([]*types.BlockWithRowConsumption, end-start+1)
+		concurrency = 32
+		sem         = make(chan struct{}, concurrency)
+	)
+
+	for i := start; i <= end; i++ {
+		sem <- struct{}{} // Acquire a slot in the semaphore
+		blockNum := rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(i))
+		index := i - start
+		eg.Go(func() error {
+			defer func() { <-sem }() // Release the slot when done
+			block, err := cli.GetBlockByNumberOrHash(ctx, blockNum)
+			if err != nil {
+				log.Error("Failed to fetch block number", "number", blockNum, "error", err)
+				return err
+			}
+			blocks[index] = block
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		log.Error("Error waiting for block fetching routines", "error", err)
+		return nil, err
+	}
+	return blocks, nil
+}