Merge pull request #92 from 0xPolygon/ccampbell/limit-unprocessed-query

Limit number of keys to process simultaneously
0xPolygon · Jun 21, 2024 · 30f1bf2 · 30f1bf2
2 parents d98a596 + d11e731
commit 30f1bf2
Show file tree

Hide file tree

Showing 7 changed files with 32 additions and 29 deletions.
diff --git a/db/db.go b/db/db.go
@@ -24,7 +24,7 @@ type DB interface {
 	GetLastProcessedBlock(ctx context.Context, task string) (uint64, error)
 
 	StoreUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error
-	GetUnresolvedBatchKeys(ctx context.Context) ([]types.BatchKey, error)
+	GetUnresolvedBatchKeys(ctx context.Context, limit uint) ([]types.BatchKey, error)
 	DeleteUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error
 
 	Exists(ctx context.Context, key common.Hash) bool
@@ -113,10 +113,10 @@ func (db *pgDB) StoreUnresolvedBatchKeys(ctx context.Context, bks []types.BatchK
 }
 
 // GetUnresolvedBatchKeys returns the unresolved batch keys from the database
-func (db *pgDB) GetUnresolvedBatchKeys(ctx context.Context) ([]types.BatchKey, error) {
-	const getUnresolvedBatchKeysSQL = "SELECT num, hash FROM data_node.unresolved_batches;"
+func (db *pgDB) GetUnresolvedBatchKeys(ctx context.Context, limit uint) ([]types.BatchKey, error) {
+	const getUnresolvedBatchKeysSQL = "SELECT num, hash FROM data_node.unresolved_batches LIMIT $1;"
 
-	rows, err := db.pg.QueryxContext(ctx, getUnresolvedBatchKeysSQL)
+	rows, err := db.pg.QueryxContext(ctx, getUnresolvedBatchKeysSQL, limit)
 	if err != nil {
 		return nil, err
 	}

diff --git a/db/db_test.go b/db/db_test.go
@@ -244,7 +244,8 @@ func Test_DB_GetUnresolvedBatchKeys(t *testing.T) {
 			// Seed data
 			seedUnresolvedBatchKeys(t, wdb, mock, tt.bks)
 
-			expected := mock.ExpectQuery(`SELECT num, hash FROM data_node\.unresolved_batches`)
+			var limit = uint(10)
+			expected := mock.ExpectQuery(`SELECT num, hash FROM data_node\.unresolved_batches LIMIT \$1\;`).WithArgs(limit)
 
 			if tt.returnErr != nil {
 				expected.WillReturnError(tt.returnErr)
@@ -256,7 +257,7 @@ func Test_DB_GetUnresolvedBatchKeys(t *testing.T) {
 
 			dbPG := New(wdb)
 
-			data, err := dbPG.GetUnresolvedBatchKeys(context.Background())
+			data, err := dbPG.GetUnresolvedBatchKeys(context.Background(), limit)
 			if tt.returnErr != nil {
 				require.ErrorIs(t, err, tt.returnErr)
 			} else {

diff --git a/mocks/db.generated.go b/mocks/db.generated.go
diff --git a/synchronizer/batches_test.go b/synchronizer/batches_test.go
@@ -643,7 +643,7 @@ func TestBatchSynchronizer_HandleUnresolvedBatches(t *testing.T) {
 		t.Parallel()
 
 		testFn(t, testConfig{
-			getUnresolvedBatchKeysArgs:    []interface{}{mock.Anything},
+			getUnresolvedBatchKeysArgs:    []interface{}{mock.Anything, uint(100)},
 			getUnresolvedBatchKeysReturns: []interface{}{nil, errors.New("error")},
 			isErrorExpected:               true,
 		})
@@ -653,7 +653,7 @@ func TestBatchSynchronizer_HandleUnresolvedBatches(t *testing.T) {
 		t.Parallel()
 
 		testFn(t, testConfig{
-			getUnresolvedBatchKeysArgs:    []interface{}{mock.Anything},
+			getUnresolvedBatchKeysArgs:    []interface{}{mock.Anything, uint(100)},
 			getUnresolvedBatchKeysReturns: []interface{}{nil, nil},
 			isErrorExpected:               false,
 		})
@@ -663,7 +663,7 @@ func TestBatchSynchronizer_HandleUnresolvedBatches(t *testing.T) {
 		t.Parallel()
 
 		testFn(t, testConfig{
-			getUnresolvedBatchKeysArgs: []interface{}{mock.Anything},
+			getUnresolvedBatchKeysArgs: []interface{}{mock.Anything, uint(100)},
 			getUnresolvedBatchKeysReturns: []interface{}{
 				[]types.BatchKey{{
 					Number: 10,
@@ -691,7 +691,7 @@ func TestBatchSynchronizer_HandleUnresolvedBatches(t *testing.T) {
 		t.Parallel()
 
 		testFn(t, testConfig{
-			getUnresolvedBatchKeysArgs: []interface{}{mock.Anything},
+			getUnresolvedBatchKeysArgs: []interface{}{mock.Anything, uint(100)},
 			getUnresolvedBatchKeysReturns: []interface{}{
 				[]types.BatchKey{{
 					Number: 10,

diff --git a/synchronizer/init.go b/synchronizer/init.go
@@ -12,8 +12,9 @@ import (
 )
 
 const (
-	initBlockTimeout = 15 * time.Second
-	minCodeLen       = 2
+	initBlockTimeout    = 15 * time.Second
+	minCodeLen          = 2
+	maxUnprocessedBatch = 100
 )
 
 // InitStartBlock initializes the L1 sync task by finding the inception block for the CDKValidium contract

diff --git a/synchronizer/store.go b/synchronizer/store.go
@@ -89,7 +89,7 @@ func getUnresolvedBatchKeys(parentCtx context.Context, db dbTypes.DB) ([]types.B
 	ctx, cancel := context.WithTimeout(parentCtx, dbTimeout)
 	defer cancel()
 
-	return db.GetUnresolvedBatchKeys(ctx)
+	return db.GetUnresolvedBatchKeys(ctx, maxUnprocessedBatch)
 }
 
 func deleteUnresolvedBatchKeys(parentCtx context.Context, db dbTypes.DB, keys []types.BatchKey) error {

diff --git a/synchronizer/store_test.go b/synchronizer/store_test.go
@@ -311,7 +311,7 @@ func Test_getUnresolvedBatchKeys(t *testing.T) {
 			db: func(t *testing.T) db.DB {
 				mockDB := mocks.NewDB(t)
 
-				mockDB.On("GetUnresolvedBatchKeys", mock.Anything).
+				mockDB.On("GetUnresolvedBatchKeys", mock.Anything, uint(100)).
 					Return(nil, testError)
 
 				return mockDB
@@ -323,7 +323,7 @@ func Test_getUnresolvedBatchKeys(t *testing.T) {
 			db: func(t *testing.T) db.DB {
 				mockDB := mocks.NewDB(t)
 
-				mockDB.On("GetUnresolvedBatchKeys", mock.Anything).Return(testData, nil)
+				mockDB.On("GetUnresolvedBatchKeys", mock.Anything, uint(100)).Return(testData, nil)
 
 				return mockDB
 			},