feat: parameterize sync service fetch block range (#1247)

* feat: parameterize sync service fetch block range

* chore: auto version bump [bot]

* fix type

Author: yiweichi

cmd/geth/main.go

@@ -171,6 +171,7 @@ var (
 		utils.L1ConfirmationsFlag,
 		utils.L1DeploymentBlockFlag,
 		utils.L1SyncIntervalFlag,
+		utils.L1FetchBlockRangeFlag,
 		utils.L1DisableMessageQueueV2Flag,
 		utils.CircuitCapacityCheckEnabledFlag,
 		utils.CircuitCapacityCheckWorkersFlag,

cmd/utils/flags.go

@@ -857,6 +857,10 @@ var (
 		Name:  "l1.sync.interval",
 		Usage: "Poll interval for L1 message syncing (e.g., 2s, 10s, 1m)",
 	}
+	L1FetchBlockRangeFlag = cli.Int64Flag{
+		Name:  "l1.sync.fetchblockrange",
+		Usage: "Block range for L1 message fetching in a single eth_getLogs query",
+	}
 	L1DisableMessageQueueV2Flag = &cli.BoolFlag{
 		Name:  "l1.disablemqv2",
 		Usage: "Disable L1 message queue v2",
@@ -1477,6 +1481,9 @@ func setL1(ctx *cli.Context, cfg *node.Config) {
 	if ctx.GlobalIsSet(L1SyncIntervalFlag.Name) {
 		cfg.L1SyncInterval = ctx.GlobalDuration(L1SyncIntervalFlag.Name)
 	}
+	if ctx.GlobalIsSet(L1FetchBlockRangeFlag.Name) {
+		cfg.L1FetchBlockRange = ctx.GlobalUint64(L1FetchBlockRangeFlag.Name)
+	}
 	if ctx.GlobalIsSet(L1DisableMessageQueueV2Flag.Name) {
 		cfg.L1DisableMessageQueueV2 = ctx.GlobalBool(L1DisableMessageQueueV2Flag.Name)
 	}

node/config.go

@@ -200,6 +200,8 @@ type Config struct {
 	L1DeploymentBlock uint64 `toml:",omitempty"`
 	// Poll interval for L1 message syncing
 	L1SyncInterval time.Duration `toml:",omitempty"`
+	// Block range for L1 message fetching in a single eth_getLogs query
+	L1FetchBlockRange uint64 `toml:",omitempty"`
 	// Explicitly disable L1 message queue V2 and only query from L1 message queue V1 (before EuclidV2)
 	L1DisableMessageQueueV2 bool `toml:",omitempty"`
 	// Is daSyncingEnabled

params/version.go

@@ -24,7 +24,7 @@ import (
 const (
 	VersionMajor = 5         // Major version component of the current release
 	VersionMinor = 9         // Minor version component of the current release
-	VersionPatch = 5         // Patch version component of the current release
+	VersionPatch = 6         // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )
 

rollup/sync_service/sync_service.go

@@ -51,6 +51,7 @@ type SyncService struct {
 	db                   ethdb.Database
 	msgCountFeed         event.Feed
 	pollInterval         time.Duration
+	fetchBlockRange      uint64
 	latestProcessedBlock uint64
 	scope                event.SubscriptionScope
 	stateMu              sync.Mutex
@@ -105,12 +106,18 @@ func NewSyncService(ctx context.Context, genesisConfig *params.ChainConfig, node
 		pollInterval = DefaultPollInterval
 	}
 
+	fetchBlockRange := nodeConfig.L1FetchBlockRange
+	if fetchBlockRange == 0 {
+		fetchBlockRange = DefaultFetchBlockRange
+	}
+
 	service := SyncService{
 		ctx:                  ctx,
 		cancel:               cancel,
 		client:               client,
 		db:                   db,
 		pollInterval:         pollInterval,
+		fetchBlockRange:      fetchBlockRange,
 		latestProcessedBlock: latestProcessedBlock,
 	}
 
@@ -236,7 +243,7 @@ func (s *SyncService) fetchMessages() {
 	numMsgsCollected := 0
 
 	// query in batches
-	for from := s.latestProcessedBlock + 1; from <= latestConfirmed; from += DefaultFetchBlockRange {
+	for from := s.latestProcessedBlock + 1; from <= latestConfirmed; from += s.fetchBlockRange {
 		select {
 		case <-s.ctx.Done():
 			// flush pending writes to database
@@ -250,7 +257,7 @@ func (s *SyncService) fetchMessages() {
 		default:
 		}
 
-		to := from + DefaultFetchBlockRange - 1
+		to := from + s.fetchBlockRange - 1
 		if to > latestConfirmed {
 			to = latestConfirmed
 		}