fix: limit DB query result count in chunk and batch proposer (#878)

Co-authored-by: Thegaram <Thegaram@users.noreply.github.com> Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com> Co-authored-by: colinlyguo <colinlyguo@users.noreply.github.com> Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com> Co-authored-by: HAOYUatHZ <HAOYUatHZ@users.noreply.github.com>
scroll-tech · Aug 25, 2023 · 87f18ef · 87f18ef
1 parent fecd129
commit 87f18ef
Show file tree

Hide file tree

Showing 10 changed files with 36 additions and 13 deletions.
diff --git a/bridge/internal/config/config.go b/bridge/internal/config/config.go
@@ -2,6 +2,7 @@ package config
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
 
@@ -15,6 +16,13 @@ type Config struct {
 	DBConfig *database.Config `json:"db_config"`
 }
 
+func (c *Config) validate() error {
+	if maxChunkPerBatch := c.L2Config.BatchProposerConfig.MaxChunkNumPerBatch; maxChunkPerBatch <= 0 {
+		return fmt.Errorf("Invalid max_chunk_num_per_batch configuration: %v", maxChunkPerBatch)
+	}
+	return nil
+}
+
 // NewConfig returns a new instance of Config.
 func NewConfig(file string) (*Config, error) {
 	buf, err := os.ReadFile(filepath.Clean(file))
@@ -28,5 +36,8 @@ func NewConfig(file string) (*Config, error) {
 		return nil, err
 	}
 
+	if err := cfg.validate(); err != nil {
+		return nil, err
+	}
 	return cfg, nil
 }
diff --git a/bridge/internal/controller/watcher/batch_proposer.go b/bridge/internal/controller/watcher/batch_proposer.go
@@ -148,7 +148,7 @@ func (p *BatchProposer) updateBatchInfoInDB(dbChunks []*orm.Chunk) error {
 }
 
 func (p *BatchProposer) proposeBatchChunks() ([]*orm.Chunk, error) {
-	dbChunks, err := p.chunkOrm.GetUnbatchedChunks(p.ctx)
+	dbChunks, err := p.chunkOrm.GetUnbatchedChunks(p.ctx, int(p.maxChunkNumPerBatch)+1)
 	if err != nil {
 		return nil, err
 	}

diff --git a/bridge/internal/controller/watcher/batch_proposer_test.go b/bridge/internal/controller/watcher/batch_proposer_test.go
@@ -40,7 +40,7 @@ func testBatchProposer(t *testing.T) {
 	bp.TryProposeBatch()
 
 	chunkOrm := orm.NewChunk(db)
-	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Empty(t, chunks)
 

diff --git a/bridge/internal/controller/watcher/chunk_proposer.go b/bridge/internal/controller/watcher/chunk_proposer.go
@@ -18,6 +18,10 @@ import (
 	"scroll-tech/bridge/internal/orm"
 )
 
+// maxNumBlockPerChunk is the maximum number of blocks we allow per chunk.
+// Normally we will pack much fewer blocks because of other limits.
+const maxNumBlockPerChunk int = 100
+
 // chunkRowConsumption is map(sub-circuit name => sub-circuit row count)
 type chunkRowConsumption map[string]uint64
 
@@ -182,7 +186,7 @@ func (p *ChunkProposer) updateChunkInfoInDB(chunk *types.Chunk) error {
 }
 
 func (p *ChunkProposer) proposeChunk() (*types.Chunk, error) {
-	blocks, err := p.l2BlockOrm.GetUnchunkedBlocks(p.ctx)
+	blocks, err := p.l2BlockOrm.GetUnchunkedBlocks(p.ctx, maxNumBlockPerChunk)
 	if err != nil {
 		return nil, err
 	}

diff --git a/bridge/internal/controller/watcher/chunk_proposer_test.go b/bridge/internal/controller/watcher/chunk_proposer_test.go
@@ -38,7 +38,7 @@ func testChunkProposer(t *testing.T) {
 	assert.NoError(t, err)
 
 	chunkOrm := orm.NewChunk(db)
-	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, chunks, 1)
 	assert.Equal(t, expectedHash.Hex(), chunks[0].Hash)
@@ -62,7 +62,7 @@ func testChunkProposerRowConsumption(t *testing.T) {
 	cp.TryProposeChunk()
 
 	chunkOrm := orm.NewChunk(db)
-	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, chunks, 0)
 }
diff --git a/bridge/internal/orm/chunk.go b/bridge/internal/orm/chunk.go
@@ -88,12 +88,16 @@ func (o *Chunk) GetChunksInRange(ctx context.Context, startIndex uint64, endInde
 }
 
 // GetUnbatchedChunks retrieves unbatched chunks from the database.
-func (o *Chunk) GetUnbatchedChunks(ctx context.Context) ([]*Chunk, error) {
+func (o *Chunk) GetUnbatchedChunks(ctx context.Context, limit int) ([]*Chunk, error) {
 	db := o.db.WithContext(ctx)
 	db = db.Model(&Chunk{})
 	db = db.Where("batch_hash IS NULL")
 	db = db.Order("index asc")
 
+	if limit > 0 {
+		db = db.Limit(limit)
+	}
+
 	var chunks []*Chunk
 	if err := db.Find(&chunks).Error; err != nil {
 		return nil, fmt.Errorf("Chunk.GetUnbatchedChunks error: %w", err)

diff --git a/bridge/internal/orm/l2_block.go b/bridge/internal/orm/l2_block.go
@@ -66,13 +66,17 @@ func (o *L2Block) GetL2BlocksLatestHeight(ctx context.Context) (uint64, error) {
 
 // GetUnchunkedBlocks get the l2 blocks that have not been put into a chunk.
 // The returned blocks are sorted in ascending order by their block number.
-func (o *L2Block) GetUnchunkedBlocks(ctx context.Context) ([]*types.WrappedBlock, error) {
+func (o *L2Block) GetUnchunkedBlocks(ctx context.Context, limit int) ([]*types.WrappedBlock, error) {
 	db := o.db.WithContext(ctx)
 	db = db.Model(&L2Block{})
 	db = db.Select("header, transactions, withdraw_root, row_consumption")
 	db = db.Where("chunk_hash IS NULL")
 	db = db.Order("number ASC")
 
+	if limit > 0 {
+		db = db.Limit(limit)
+	}
+
 	var l2Blocks []L2Block
 	if err := db.Find(&l2Blocks).Error; err != nil {
 		return nil, fmt.Errorf("L2Block.GetUnchunkedBlocks error: %w", err)

diff --git a/bridge/internal/orm/orm_test.go b/bridge/internal/orm/orm_test.go
@@ -101,7 +101,7 @@ func TestL2BlockOrm(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(3), height)
 
-	blocks, err := l2BlockOrm.GetUnchunkedBlocks(context.Background())
+	blocks, err := l2BlockOrm.GetUnchunkedBlocks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, blocks, 2)
 	assert.Equal(t, wrappedBlock1, blocks[0])
@@ -116,7 +116,7 @@ func TestL2BlockOrm(t *testing.T) {
 	err = l2BlockOrm.UpdateChunkHashInRange(context.Background(), 2, 2, "test hash")
 	assert.NoError(t, err)
 
-	blocks, err = l2BlockOrm.GetUnchunkedBlocks(context.Background())
+	blocks, err = l2BlockOrm.GetUnchunkedBlocks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, blocks, 1)
 	assert.Equal(t, wrappedBlock2, blocks[0])
@@ -135,7 +135,7 @@ func TestChunkOrm(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, dbChunk2.Hash, chunkHash2.Hex())
 
-	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, chunks, 2)
 	assert.Equal(t, chunkHash1.Hex(), chunks[0].Hash)
@@ -156,7 +156,7 @@ func TestChunkOrm(t *testing.T) {
 
 	err = chunkOrm.UpdateBatchHashInRange(context.Background(), 0, 0, "test hash")
 	assert.NoError(t, err)
-	chunks, err = chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err = chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, chunks, 1)
 }

diff --git a/bridge/tests/rollup_test.go b/bridge/tests/rollup_test.go
@@ -67,7 +67,7 @@ func testCommitBatchAndFinalizeBatch(t *testing.T) {
 	cp.TryProposeChunk()
 
 	chunkOrm := orm.NewChunk(db)
-	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background())
+	chunks, err := chunkOrm.GetUnbatchedChunks(context.Background(), 0)
 	assert.NoError(t, err)
 	assert.Len(t, chunks, 1)
 

diff --git a/common/version/version.go b/common/version/version.go
@@ -7,7 +7,7 @@ import (
 	"strings"
 )
 
-var tag = "v4.1.114"
+var tag = "v4.1.115"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {