refactor: compose DA types for more cohesion, maintainability and cod…

…e reuse (#925)

* refactor: compose DA types for more cohesion, maintainability and code reuse

* feat: simultaneous L1 message sync and L1 sync from DA instead of consecutive (#929)

* feat: add exponential backoff utility

* feat: add exponential backoff to pipeline

* feat: run pipeline and L1 message sync in parallel

* Update rollup/da_syncer/syncing_pipeline.go

init waitgroup

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>

* update da v2 to use correct blob decoding method

* introduce NewCommitBatchDAV1WithBlobDecodeFunc

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>

common/backoff/exponential.go

@@ -0,0 +1,50 @@
+package backoff
+
+import (
+	"math"
+	"math/rand"
+	"time"
+)
+
+type Exponential struct {
+	attempt int
+
+	maxJitter time.Duration
+
+	min time.Duration
+	max time.Duration
+}
+
+func NewExponential(minimum, maximum, maxJitter time.Duration) *Exponential {
+	return &Exponential{
+		min:       minimum,
+		max:       maximum,
+		maxJitter: maxJitter,
+	}
+}
+
+func (e *Exponential) NextDuration() time.Duration {
+	var jitter time.Duration
+	if e.maxJitter > 0 {
+		jitter = time.Duration(rand.Int63n(e.maxJitter.Nanoseconds()))
+	}
+
+	minFloat := float64(e.min)
+	duration := math.Pow(2, float64(e.attempt)) * minFloat
+
+	// limit at configured maximum
+	if duration > float64(e.max) {
+		duration = float64(e.max)
+	}
+
+	e.attempt++
+	return time.Duration(duration) + jitter
+}
+
+func (e *Exponential) Reset() {
+	e.attempt = 0
+}
+
+func (e *Exponential) Attempt() int {
+	return e.attempt
+}

common/backoff/exponential_test.go

@@ -0,0 +1,39 @@
+package backoff
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestExponentialBackoff(t *testing.T) {
+	t.Run("Multiple attempts", func(t *testing.T) {
+		e := NewExponential(100*time.Millisecond, 10*time.Second, 0)
+		expectedDurations := []time.Duration{
+			100 * time.Millisecond,
+			200 * time.Millisecond,
+			400 * time.Millisecond,
+			800 * time.Millisecond,
+			1600 * time.Millisecond,
+			3200 * time.Millisecond,
+			6400 * time.Millisecond,
+			10 * time.Second, // capped at max
+		}
+		for i, expected := range expectedDurations {
+			require.Equal(t, expected, e.NextDuration(), "attempt %d", i)
+		}
+	})
+
+	t.Run("Jitter added", func(t *testing.T) {
+		e := NewExponential(1*time.Second, 10*time.Second, 1*time.Second)
+		duration := e.NextDuration()
+		require.GreaterOrEqual(t, duration, 1*time.Second)
+		require.Less(t, duration, 2*time.Second)
+	})
+
+	t.Run("Edge case: min > max", func(t *testing.T) {
+		e := NewExponential(10*time.Second, 5*time.Second, 0)
+		require.Equal(t, 5*time.Second, e.NextDuration())
+	})
+}

eth/backend.go

@@ -221,6 +221,18 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
 	}
 	eth.txPool = core.NewTxPool(config.TxPool, chainConfig, eth.blockchain)
 
+	// Initialize and start DA syncing pipeline before SyncService as SyncService is blocking until all L1 messages are loaded.
+	// We need SyncService to load the L1 messages for DA syncing, but since both sync from last known L1 state, we can
+	// simply let them run simultaneously. If messages are missing in DA syncing, it will be handled by the syncing pipeline
+	// by waiting and retrying.
+	if config.EnableDASyncing {
+		eth.syncingPipeline, err = da_syncer.NewSyncingPipeline(context.Background(), eth.blockchain, chainConfig, eth.chainDb, l1Client, stack.Config().L1DeploymentBlock, config.DA)
+		if err != nil {
+			return nil, fmt.Errorf("cannot initialize da syncer: %w", err)
+		}
+		eth.syncingPipeline.Start()
+	}
+
 	// initialize and start L1 message sync service
 	eth.syncService, err = sync_service.NewSyncService(context.Background(), chainConfig, stack.Config(), eth.chainDb, l1Client)
 	if err != nil {
@@ -237,14 +249,6 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
 		eth.rollupSyncService.Start()
 	}
 
-	if config.EnableDASyncing {
-		eth.syncingPipeline, err = da_syncer.NewSyncingPipeline(context.Background(), eth.blockchain, chainConfig, eth.chainDb, l1Client, stack.Config().L1DeploymentBlock, config.DA)
-		if err != nil {
-			return nil, fmt.Errorf("cannot initialize da syncer: %w", err)
-		}
-		eth.syncingPipeline.Start()
-	}
-
 	// Permit the downloader to use the trie cache allowance during fast sync
 	cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit
 	checkpoint := config.Checkpoint

go.mod

@@ -50,7 +50,7 @@ require (
 	github.com/prometheus/tsdb v0.7.1
 	github.com/rjeczalik/notify v0.9.1
 	github.com/rs/cors v1.7.0
-	github.com/scroll-tech/da-codec v0.1.1-0.20240708084945-cb02d638c45f
+	github.com/scroll-tech/da-codec v0.1.1-0.20240729153040-31de3201bd1a
 	github.com/scroll-tech/zktrie v0.8.4
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4

go.sum

@@ -394,8 +394,8 @@ github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncj
 github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/scroll-tech/da-codec v0.1.1-0.20240708084945-cb02d638c45f h1:ZKPhn674+2AgBdIn2ZLGePsUZdM2823m2tJp+JlQf/Y=
-github.com/scroll-tech/da-codec v0.1.1-0.20240708084945-cb02d638c45f/go.mod h1:O9jsbQGNnTEfyfZg7idevq6jGGSQshX70elX+TRH8vU=
+github.com/scroll-tech/da-codec v0.1.1-0.20240729153040-31de3201bd1a h1:2+g6p5dxRQYFjDMBZgBJkem2HpIJdXEIlXzxreTCd4w=
+github.com/scroll-tech/da-codec v0.1.1-0.20240729153040-31de3201bd1a/go.mod h1:O9jsbQGNnTEfyfZg7idevq6jGGSQshX70elX+TRH8vU=
 github.com/scroll-tech/zktrie v0.8.4 h1:UagmnZ4Z3ITCk+aUq9NQZJNAwnWl4gSxsLb2Nl7IgRE=
 github.com/scroll-tech/zktrie v0.8.4/go.mod h1:XvNo7vAk8yxNyTjBDj5WIiFzYW4bx/gJ78+NK6Zn6Uk=
 github.com/segmentio/kafka-go v0.1.0/go.mod h1:X6itGqS9L4jDletMsxZ7Dz+JFWxM6JHfPOCvTvk+EJo=

rollup/da_syncer/batch_queue.go

@@ -7,27 +7,28 @@ import (
 
 	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/ethdb"
+	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/da"
 )
 
 type BatchQueue struct {
 	// batches is map from batchIndex to batch blocks
-	batches                 map[uint64]DAEntry
+	batches                 map[uint64]da.Entry
 	DAQueue                 *DAQueue
 	db                      ethdb.Database
 	lastFinalizedBatchIndex uint64
 }
 
 func NewBatchQueue(DAQueue *DAQueue, db ethdb.Database) *BatchQueue {
 	return &BatchQueue{
-		batches:                 make(map[uint64]DAEntry),
+		batches:                 make(map[uint64]da.Entry),
 		DAQueue:                 DAQueue,
 		db:                      db,
 		lastFinalizedBatchIndex: 0,
 	}
 }
 
 // NextBatch finds next finalized batch and returns data, that was committed in that batch
-func (bq *BatchQueue) NextBatch(ctx context.Context) (DAEntry, error) {
+func (bq *BatchQueue) NextBatch(ctx context.Context) (da.Entry, error) {
 	if batch, ok := bq.getFinalizedBatch(); ok {
 		return batch, nil
 	}
@@ -36,18 +37,14 @@ func (bq *BatchQueue) NextBatch(ctx context.Context) (DAEntry, error) {
 		if err != nil {
 			return nil, err
 		}
-		switch daEntry := daEntry.(type) {
-		case *CommitBatchDAV0:
-			bq.batches[daEntry.BatchIndex] = daEntry
-		case *CommitBatchDAV1:
-			bq.batches[daEntry.BatchIndex] = daEntry
-		case *CommitBatchDAV2:
-			bq.batches[daEntry.BatchIndex] = daEntry
-		case *RevertBatchDA:
-			bq.deleteBatch(daEntry.BatchIndex)
-		case *FinalizeBatchDA:
-			if daEntry.BatchIndex > bq.lastFinalizedBatchIndex {
-				bq.lastFinalizedBatchIndex = daEntry.BatchIndex
+		switch daEntry.Type() {
+		case da.CommitBatchV0Type, da.CommitBatchV1Type, da.CommitBatchV2Type:
+			bq.batches[daEntry.BatchIndex()] = daEntry
+		case da.RevertBatchType:
+			bq.deleteBatch(daEntry.BatchIndex())
+		case da.FinalizeBatchType:
+			if daEntry.BatchIndex() > bq.lastFinalizedBatchIndex {
+				bq.lastFinalizedBatchIndex = daEntry.BatchIndex()
 			}
 			ret, ok := bq.getFinalizedBatch()
 			if ok {
@@ -62,7 +59,7 @@ func (bq *BatchQueue) NextBatch(ctx context.Context) (DAEntry, error) {
 }
 
 // getFinalizedBatch returns next finalized batch if there is available
-func (bq *BatchQueue) getFinalizedBatch() (DAEntry, bool) {
+func (bq *BatchQueue) getFinalizedBatch() (da.Entry, bool) {
 	if len(bq.batches) == 0 {
 		return nil, false
 	}
@@ -93,7 +90,7 @@ func (bq *BatchQueue) deleteBatch(batchIndex uint64) {
 	if !ok {
 		return
 	}
-	curBatchL1Height := batch.GetL1BlockNumber()
+	curBatchL1Height := batch.L1BlockNumber()
 	delete(bq.batches, batchIndex)
 	if len(bq.batches) == 0 {
 		rawdb.WriteDASyncedL1BlockNumber(bq.db, curBatchL1Height)
@@ -102,10 +99,10 @@ func (bq *BatchQueue) deleteBatch(batchIndex uint64) {
 	// we store here min height of currently loaded batches to be able to start syncing from the same place in case of restart
 	var minBatchL1Height uint64 = math.MaxUint64
 	for _, val := range bq.batches {
-		if val.GetL1BlockNumber() < minBatchL1Height {
-			minBatchL1Height = val.GetL1BlockNumber()
+		if val.L1BlockNumber() < minBatchL1Height {
+			minBatchL1Height = val.L1BlockNumber()
 		}
 	}
-	rawdb.WriteDASyncedL1BlockNumber(bq.db, minBatchL1Height-1)
 
+	rawdb.WriteDASyncedL1BlockNumber(bq.db, minBatchL1Height-1)
 }

rollup/da_syncer/block_queue.go

@@ -3,9 +3,9 @@ package da_syncer
 import (
 	"context"
 	"fmt"
-	"math/big"
 
 	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/da"
 )
 
 type BlockQueue struct {
@@ -37,120 +37,17 @@ func (bq *BlockQueue) getBlocksFromBatch(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	switch daEntry := daEntry.(type) {
-	case *CommitBatchDAV0:
-		bq.blocks, err = bq.processDAV0ToBlocks(daEntry)
-		if err != nil {
-			return err
-		}
-	case *CommitBatchDAV1:
-		bq.blocks, err = bq.processDAV1ToBlocks(daEntry)
-		if err != nil {
-			return err
-		}
-	case *CommitBatchDAV2:
-		bq.blocks, err = bq.processDAV2ToBlocks(daEntry)
-		if err != nil {
-			return err
-		}
-	default:
-		return fmt.Errorf("unexpected type of daEntry: %T", daEntry)
-	}
-	return nil
-}
 
-func (bq *BlockQueue) processDAV0ToBlocks(daEntry *CommitBatchDAV0) ([]*types.Block, error) {
-	var blocks []*types.Block
-	l1TxPointer := 0
-	var curL1TxIndex uint64 = daEntry.ParentTotalL1MessagePopped
-	for _, chunk := range daEntry.Chunks {
-		for blockId, daBlock := range chunk.Blocks {
-			// create header
-			header := types.Header{
-				Number:   big.NewInt(0).SetUint64(daBlock.BlockNumber),
-				Time:     daBlock.Timestamp,
-				BaseFee:  daBlock.BaseFee,
-				GasLimit: daBlock.GasLimit,
-			}
-			// create txs
-			// var txs types.Transactions
-			txs := make(types.Transactions, 0, daBlock.NumTransactions)
-			// insert l1 msgs
-			for l1TxPointer < len(daEntry.L1Txs) && daEntry.L1Txs[l1TxPointer].QueueIndex < curL1TxIndex+uint64(daBlock.NumL1Messages) {
-				l1Tx := types.NewTx(daEntry.L1Txs[l1TxPointer])
-				txs = append(txs, l1Tx)
-				l1TxPointer++
-			}
-			curL1TxIndex += uint64(daBlock.NumL1Messages)
-			// insert l2 txs
-			txs = append(txs, chunk.Transactions[blockId]...)
-			block := types.NewBlockWithHeader(&header).WithBody(txs, make([]*types.Header, 0))
-			blocks = append(blocks, block)
-		}
+	entryWithBlocks, ok := daEntry.(da.EntryWithBlocks)
+	// this should never happen because we only receive CommitBatch entries
+	if !ok {
+		return fmt.Errorf("unexpected type of daEntry: %T", daEntry)
 	}
-	return blocks, nil
-}
 
-func (bq *BlockQueue) processDAV1ToBlocks(daEntry *CommitBatchDAV1) ([]*types.Block, error) {
-	var blocks []*types.Block
-	l1TxPointer := 0
-	var curL1TxIndex uint64 = daEntry.ParentTotalL1MessagePopped
-	for _, chunk := range daEntry.Chunks {
-		for blockId, daBlock := range chunk.Blocks {
-			// create header
-			header := types.Header{
-				Number:   big.NewInt(0).SetUint64(daBlock.BlockNumber),
-				Time:     daBlock.Timestamp,
-				BaseFee:  daBlock.BaseFee,
-				GasLimit: daBlock.GasLimit,
-			}
-			// create txs
-			// var txs types.Transactions
-			txs := make(types.Transactions, 0, daBlock.NumTransactions)
-			// insert l1 msgs
-			for l1TxPointer < len(daEntry.L1Txs) && daEntry.L1Txs[l1TxPointer].QueueIndex < curL1TxIndex+uint64(daBlock.NumL1Messages) {
-				l1Tx := types.NewTx(daEntry.L1Txs[l1TxPointer])
-				txs = append(txs, l1Tx)
-				l1TxPointer++
-			}
-			curL1TxIndex += uint64(daBlock.NumL1Messages)
-			// insert l2 txs
-			txs = append(txs, chunk.Transactions[blockId]...)
-			block := types.NewBlockWithHeader(&header).WithBody(txs, make([]*types.Header, 0))
-			blocks = append(blocks, block)
-		}
+	bq.blocks, err = entryWithBlocks.Blocks()
+	if err != nil {
+		return fmt.Errorf("failed to get blocks from daEntry: %w", err)
 	}
-	return blocks, nil
-}
 
-func (bq *BlockQueue) processDAV2ToBlocks(daEntry *CommitBatchDAV2) ([]*types.Block, error) {
-	var blocks []*types.Block
-	l1TxPointer := 0
-	var curL1TxIndex uint64 = daEntry.ParentTotalL1MessagePopped
-	for _, chunk := range daEntry.Chunks {
-		for blockId, daBlock := range chunk.Blocks {
-			// create header
-			header := types.Header{
-				Number:   big.NewInt(0).SetUint64(daBlock.BlockNumber),
-				Time:     daBlock.Timestamp,
-				BaseFee:  daBlock.BaseFee,
-				GasLimit: daBlock.GasLimit,
-			}
-			// create txs
-			// var txs types.Transactions
-			txs := make(types.Transactions, 0, daBlock.NumTransactions)
-			// insert l1 msgs
-			for l1TxPointer < len(daEntry.L1Txs) && daEntry.L1Txs[l1TxPointer].QueueIndex < curL1TxIndex+uint64(daBlock.NumL1Messages) {
-				l1Tx := types.NewTx(daEntry.L1Txs[l1TxPointer])
-				txs = append(txs, l1Tx)
-				l1TxPointer++
-			}
-			curL1TxIndex += uint64(daBlock.NumL1Messages)
-			// insert l2 txs
-			txs = append(txs, chunk.Transactions[blockId]...)
-			block := types.NewBlockWithHeader(&header).WithBody(txs, make([]*types.Header, 0))
-			blocks = append(blocks, block)
-		}
-	}
-	return blocks, nil
+	return nil
 }