refactor(rollup sync service): use CalldataBlobSource to retrieve data from L1 (#1103)

* port changes from #1013

* port changes from #1068

* go.mod tidy

* fix compile error

* fix goimports

* fix log

* address review comments

* upgrade golang.org/x/net to 0.23.0

* port changes from #1018

* fix tests and linter errors

* address review comments

* refactor rollup sync service / verifier to use CalldataBlobSource to retrieve data from L1

* add configuration and initialize blob clients

* fix unit tests

* remove unused code

* address review comments

* address more review comments

* Allow using MPT

* fix issues after merge

* bump version

---------

Co-authored-by: Ömer Faruk Irmak <[email protected]>
Co-authored-by: Péter Garamvölgyi <[email protected]>

Author: 3 people

cmd/utils/flags.go

@@ -1641,15 +1641,15 @@ func setEnableRollupVerify(ctx *cli.Context, cfg *ethconfig.Config) {
 func setDA(ctx *cli.Context, cfg *ethconfig.Config) {
 	if ctx.IsSet(DASyncEnabledFlag.Name) {
 		cfg.EnableDASyncing = ctx.Bool(DASyncEnabledFlag.Name)
-		if ctx.IsSet(DABlobScanAPIEndpointFlag.Name) {
-			cfg.DA.BlobScanAPIEndpoint = ctx.String(DABlobScanAPIEndpointFlag.Name)
-		}
-		if ctx.IsSet(DABlockNativeAPIEndpointFlag.Name) {
-			cfg.DA.BlockNativeAPIEndpoint = ctx.String(DABlockNativeAPIEndpointFlag.Name)
-		}
-		if ctx.IsSet(DABeaconNodeAPIEndpointFlag.Name) {
-			cfg.DA.BeaconNodeAPIEndpoint = ctx.String(DABeaconNodeAPIEndpointFlag.Name)
-		}
+	}
+	if ctx.IsSet(DABlobScanAPIEndpointFlag.Name) {
+		cfg.DA.BlobScanAPIEndpoint = ctx.String(DABlobScanAPIEndpointFlag.Name)
+	}
+	if ctx.IsSet(DABlockNativeAPIEndpointFlag.Name) {
+		cfg.DA.BlockNativeAPIEndpoint = ctx.String(DABlockNativeAPIEndpointFlag.Name)
+	}
+	if ctx.IsSet(DABeaconNodeAPIEndpointFlag.Name) {
+		cfg.DA.BeaconNodeAPIEndpoint = ctx.String(DABeaconNodeAPIEndpointFlag.Name)
 	}
 }
 

core/rawdb/accessors_rollup_event.go

@@ -18,7 +18,8 @@ type ChunkBlockRange struct {
 
 // CommittedBatchMeta holds metadata for committed batches.
 type CommittedBatchMeta struct {
-	Version             uint8
+	Version uint8
+	// BlobVersionedHashes are the versioned hashes of the blobs in the batch. Currently unused. Left for compatibility.
 	BlobVersionedHashes []common.Hash
 	ChunkBlockRanges    []*ChunkBlockRange
 }

eth/backend.go

@@ -245,7 +245,7 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 
 	if config.EnableRollupVerify {
 		// initialize and start rollup event sync service
-		eth.rollupSyncService, err = rollup_sync_service.NewRollupSyncService(context.Background(), chainConfig, eth.chainDb, l1Client, eth.blockchain, stack)
+		eth.rollupSyncService, err = rollup_sync_service.NewRollupSyncService(context.Background(), chainConfig, eth.chainDb, l1Client, eth.blockchain, stack, config.DA)
 		if err != nil {
 			return nil, fmt.Errorf("cannot initialize rollup event sync service: %w", err)
 		}

params/version.go

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

rollup/da_syncer/da/calldata_blob_source.go

@@ -81,6 +81,10 @@ func (ds *CalldataBlobSource) NextData() (Entries, error) {
 	return da, nil
 }
 
+func (ds *CalldataBlobSource) SetL1Height(l1Height uint64) {
+	ds.l1Height = l1Height
+}
+
 func (ds *CalldataBlobSource) L1Height() uint64 {
 	return ds.l1Height
 }
@@ -106,10 +110,22 @@ func (ds *CalldataBlobSource) processRollupEventsToDA(rollupEvents l1.RollupEven
 			}
 
 		case l1.RevertEventType:
-			entry = NewRevertBatch(rollupEvent.BatchIndex().Uint64())
+			revertEvent, ok := rollupEvent.(*l1.RevertBatchEvent)
+			// this should never happen because we just check event type
+			if !ok {
+				return nil, fmt.Errorf("unexpected type of rollup event: %T", rollupEvent)
+			}
+
+			entry = NewRevertBatch(revertEvent)
 
 		case l1.FinalizeEventType:
-			entry = NewFinalizeBatch(rollupEvent.BatchIndex().Uint64())
+			finalizeEvent, ok := rollupEvent.(*l1.FinalizeBatchEvent)
+			// this should never happen because we just check event type
+			if !ok {
+				return nil, fmt.Errorf("unexpected type of rollup event: %T", rollupEvent)
+			}
+
+			entry = NewFinalizeBatch(finalizeEvent)
 
 		default:
 			return nil, fmt.Errorf("unknown rollup event, type: %v", rollupEvent.Type())

rollup/da_syncer/da/commitV0.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/scroll-tech/da-codec/encoding"
 
+	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/ethdb"
@@ -14,14 +15,14 @@ import (
 )
 
 type CommitBatchDAV0 struct {
-	version                    uint8
+	version                    encoding.CodecVersion
 	batchIndex                 uint64
 	parentTotalL1MessagePopped uint64
 	skippedL1MessageBitmap     []byte
 	chunks                     []*encoding.DAChunkRawTx
 	l1Txs                      []*types.L1MessageTx
 
-	l1BlockNumber uint64
+	event *l1.CommitBatchEvent
 }
 
 func NewCommitBatchDAV0(db ethdb.Database,
@@ -36,16 +37,16 @@ func NewCommitBatchDAV0(db ethdb.Database,
 		return nil, fmt.Errorf("failed to unpack chunks: %d, err: %w", commitEvent.BatchIndex().Uint64(), err)
 	}
 
-	return NewCommitBatchDAV0WithChunks(db, uint8(codec.Version()), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent.BlockNumber())
+	return NewCommitBatchDAV0WithChunks(db, codec.Version(), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent)
 }
 
 func NewCommitBatchDAV0WithChunks(db ethdb.Database,
-	version uint8,
+	version encoding.CodecVersion,
 	batchIndex uint64,
 	parentBatchHeader []byte,
 	decodedChunks []*encoding.DAChunkRawTx,
 	skippedL1MessageBitmap []byte,
-	l1BlockNumber uint64,
+	event *l1.CommitBatchEvent,
 ) (*CommitBatchDAV0, error) {
 	parentTotalL1MessagePopped := getBatchTotalL1MessagePopped(parentBatchHeader)
 	l1Txs, err := getL1Messages(db, parentTotalL1MessagePopped, skippedL1MessageBitmap, getTotalMessagesPoppedFromChunks(decodedChunks))
@@ -60,7 +61,7 @@ func NewCommitBatchDAV0WithChunks(db ethdb.Database,
 		skippedL1MessageBitmap:     skippedL1MessageBitmap,
 		chunks:                     decodedChunks,
 		l1Txs:                      l1Txs,
-		l1BlockNumber:              l1BlockNumber,
+		event:                      event,
 	}, nil
 }
 
@@ -70,12 +71,28 @@ func NewCommitBatchDAV0Empty() *CommitBatchDAV0 {
 	}
 }
 
+func (c *CommitBatchDAV0) Version() encoding.CodecVersion {
+	return c.version
+}
+
+func (c *CommitBatchDAV0) Chunks() []*encoding.DAChunkRawTx {
+	return c.chunks
+}
+
+func (c *CommitBatchDAV0) BlobVersionedHashes() []common.Hash {
+	return nil
+}
+
 func (c *CommitBatchDAV0) Type() Type {
 	return CommitBatchV0Type
 }
 
 func (c *CommitBatchDAV0) L1BlockNumber() uint64 {
-	return c.l1BlockNumber
+	return c.event.BlockNumber()
+}
+
+func (c *CommitBatchDAV0) Event() l1.RollupEvent {
+	return c.event
 }
 
 func (c *CommitBatchDAV0) BatchIndex() uint64 {

rollup/da_syncer/da/commitV1.go

@@ -17,6 +17,8 @@ import (
 
 type CommitBatchDAV1 struct {
 	*CommitBatchDAV0
+
+	versionedHashes []common.Hash
 }
 
 func NewCommitBatchDAWithBlob(ctx context.Context, db ethdb.Database,
@@ -33,11 +35,17 @@ func NewCommitBatchDAWithBlob(ctx context.Context, db ethdb.Database,
 		return nil, fmt.Errorf("failed to unpack chunks: %v, err: %w", commitEvent.BatchIndex().Uint64(), err)
 	}
 
-	versionedHash, err := l1Reader.FetchTxBlobHash(commitEvent.TxHash(), commitEvent.BlockHash())
+	versionedHashes, err := l1Reader.FetchTxBlobHashes(commitEvent.TxHash(), commitEvent.BlockHash())
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch blob hash, err: %w", err)
 	}
 
+	// with CommitBatchDAV1 we expect only one versioned hash as we commit only one blob per batch submission
+	if len(versionedHashes) != 1 {
+		return nil, fmt.Errorf("unexpected number of versioned hashes: %d", len(versionedHashes))
+	}
+	versionedHash := versionedHashes[0]
+
 	header, err := l1Reader.FetchBlockHeaderByNumber(commitEvent.BlockNumber())
 	if err != nil {
 		return nil, fmt.Errorf("failed to get header by number, err: %w", err)
@@ -70,14 +78,21 @@ func NewCommitBatchDAWithBlob(ctx context.Context, db ethdb.Database,
 		return nil, fmt.Errorf("decodedChunks is nil after decoding")
 	}
 
-	v0, err := NewCommitBatchDAV0WithChunks(db, uint8(codec.Version()), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent.BlockNumber())
+	v0, err := NewCommitBatchDAV0WithChunks(db, codec.Version(), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent)
 	if err != nil {
 		return nil, err
 	}
 
-	return &CommitBatchDAV1{v0}, nil
+	return &CommitBatchDAV1{
+		CommitBatchDAV0: v0,
+		versionedHashes: versionedHashes,
+	}, nil
 }
 
 func (c *CommitBatchDAV1) Type() Type {
 	return CommitBatchWithBlobType
 }
+
+func (c *CommitBatchDAV1) BlobVersionedHashes() []common.Hash {
+	return c.versionedHashes
+}

rollup/da_syncer/da/da.go

@@ -3,7 +3,11 @@ package da
 import (
 	"math/big"
 
+	"github.com/scroll-tech/da-codec/encoding"
+
+	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 )
 
 type Type int
@@ -25,11 +29,15 @@ type Entry interface {
 	BatchIndex() uint64
 	L1BlockNumber() uint64
 	CompareTo(Entry) int
+	Event() l1.RollupEvent
 }
 
 type EntryWithBlocks interface {
 	Entry
 	Blocks() []*PartialBlock
+	Version() encoding.CodecVersion
+	Chunks() []*encoding.DAChunkRawTx
+	BlobVersionedHashes() []common.Hash
 }
 
 type Entries []Entry

rollup/da_syncer/da/finalize.go

@@ -1,14 +1,16 @@
 package da
 
-type FinalizeBatch struct {
-	batchIndex uint64
+import (
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
+)
 
-	l1BlockNumber uint64
+type FinalizeBatch struct {
+	event *l1.FinalizeBatchEvent
 }
 
-func NewFinalizeBatch(batchIndex uint64) *FinalizeBatch {
+func NewFinalizeBatch(event *l1.FinalizeBatchEvent) *FinalizeBatch {
 	return &FinalizeBatch{
-		batchIndex: batchIndex,
+		event: event,
 	}
 }
 
@@ -17,11 +19,15 @@ func (f *FinalizeBatch) Type() Type {
 }
 
 func (f *FinalizeBatch) L1BlockNumber() uint64 {
-	return f.l1BlockNumber
+	return f.event.BlockNumber()
 }
 
 func (f *FinalizeBatch) BatchIndex() uint64 {
-	return f.batchIndex
+	return f.event.BatchIndex().Uint64()
+}
+
+func (f *FinalizeBatch) Event() l1.RollupEvent {
+	return f.event
 }
 
 func (f *FinalizeBatch) CompareTo(other Entry) int {

rollup/da_syncer/da/revert.go

@@ -1,14 +1,16 @@
 package da
 
-type RevertBatch struct {
-	batchIndex uint64
+import (
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
+)
 
-	l1BlockNumber uint64
+type RevertBatch struct {
+	event *l1.RevertBatchEvent
 }
 
-func NewRevertBatch(batchIndex uint64) *RevertBatch {
+func NewRevertBatch(event *l1.RevertBatchEvent) *RevertBatch {
 	return &RevertBatch{
-		batchIndex: batchIndex,
+		event: event,
 	}
 }
 
@@ -17,10 +19,14 @@ func (r *RevertBatch) Type() Type {
 }
 
 func (r *RevertBatch) L1BlockNumber() uint64 {
-	return r.l1BlockNumber
+	return r.event.BlockNumber()
 }
 func (r *RevertBatch) BatchIndex() uint64 {
-	return r.batchIndex
+	return r.event.BatchIndex().Uint64()
+}
+
+func (r *RevertBatch) Event() l1.RollupEvent {
+	return r.event
 }
 
 func (r *RevertBatch) CompareTo(other Entry) int {

rollup/l1/abi.go

@@ -158,6 +158,26 @@ type FinalizeBatchEvent struct {
 	blockNumber  uint64
 }
 
+func NewFinalizeBatchEvent(
+	batchIndex *big.Int,
+	batchHash common.Hash,
+	stateRoot common.Hash,
+	withdrawRoot common.Hash,
+	txHash common.Hash,
+	blockHash common.Hash,
+	blockNumber uint64,
+) *FinalizeBatchEvent {
+	return &FinalizeBatchEvent{
+		batchIndex:   batchIndex,
+		batchHash:    batchHash,
+		stateRoot:    stateRoot,
+		withdrawRoot: withdrawRoot,
+		txHash:       txHash,
+		blockHash:    blockHash,
+		blockNumber:  blockNumber,
+	}
+}
+
 func (f *FinalizeBatchEvent) TxHash() common.Hash {
 	return f.txHash
 }

rollup/l1/reader.go

@@ -139,20 +139,23 @@ func (r *Reader) FetchTxData(txHash, blockHash common.Hash) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
+
 	return tx.Data(), nil
 }
 
-// FetchTxBlobHash fetches tx blob hash corresponding to given event log
-func (r *Reader) FetchTxBlobHash(txHash, blockHash common.Hash) (common.Hash, error) {
+// FetchTxBlobHashes fetches tx blob hash corresponding to given event log
+func (r *Reader) FetchTxBlobHashes(txHash, blockHash common.Hash) ([]common.Hash, error) {
 	tx, err := r.fetchTx(txHash, blockHash)
 	if err != nil {
-		return common.Hash{}, err
+		return nil, fmt.Errorf("failed to fetch tx, tx hash: %v, block hash: %v, err: %w", txHash.Hex(), blockHash.Hex(), err)
 	}
+
 	blobHashes := tx.BlobHashes()
 	if len(blobHashes) == 0 {
-		return common.Hash{}, fmt.Errorf("transaction does not contain any blobs, tx hash: %v", txHash.Hex())
+		return nil, fmt.Errorf("transaction does not contain any blobs, tx hash: %v", txHash.Hex())
 	}
-	return blobHashes[0], nil
+
+	return blobHashes, nil
 }
 
 // FetchRollupEventsInRange retrieves and parses commit/revert/finalize rollup events between block numbers: [from, to].