feat: L1Reader (#1099)

* 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

common/heapmap.go

@@ -0,0 +1,90 @@
+package common
+
+type HeapMap[K comparable, T Comparable[T]] struct {
+	h              *Heap[T]
+	m              *ShrinkingMap[K, *HeapElement[T]]
+	keyFromElement func(T) K
+}
+
+func NewHeapMap[K comparable, T Comparable[T]](keyFromElement func(T) K) *HeapMap[K, T] {
+	return &HeapMap[K, T]{
+		h:              NewHeap[T](),
+		m:              NewShrinkingMap[K, *HeapElement[T]](1000),
+		keyFromElement: keyFromElement,
+	}
+}
+
+func (hm *HeapMap[K, T]) Len() int {
+	return hm.h.Len()
+}
+
+func (hm *HeapMap[K, T]) Push(element T) bool {
+	k := hm.keyFromElement(element)
+
+	if hm.m.Has(k) {
+		return false
+	}
+
+	heapElement := hm.h.Push(element)
+	hm.m.Set(k, heapElement)
+
+	return true
+}
+
+func (hm *HeapMap[K, T]) Pop() T {
+	element := hm.h.Pop()
+	k := hm.keyFromElement(element.Value())
+	hm.m.Delete(k)
+
+	return element.Value()
+}
+
+func (hm *HeapMap[K, T]) Peek() T {
+	return hm.h.Peek().Value()
+}
+
+func (hm *HeapMap[K, T]) RemoveByElement(element T) bool {
+	key := hm.keyFromElement(element)
+	heapElement, exists := hm.m.Get(key)
+	if !exists {
+		return false
+	}
+
+	hm.h.Remove(heapElement)
+	hm.m.Delete(key)
+
+	return true
+}
+
+func (hm *HeapMap[K, T]) RemoveByKey(key K) bool {
+	heapElement, exists := hm.m.Get(key)
+	if !exists {
+		return false
+	}
+
+	hm.h.Remove(heapElement)
+	hm.m.Delete(key)
+
+	return true
+}
+
+func (hm *HeapMap[K, T]) Clear() {
+	hm.h.Clear()
+	hm.m = NewShrinkingMap[K, *HeapElement[T]](1000)
+}
+
+func (hm *HeapMap[K, T]) Keys() []K {
+	return hm.m.Keys()
+}
+
+func (hm *HeapMap[K, T]) Elements() []T {
+	var elements []T
+	for _, element := range hm.m.Values() {
+		elements = append(elements, element.Value())
+	}
+	return elements
+}
+
+func (hm *HeapMap[K, T]) Has(element T) bool {
+	return hm.m.Has(hm.keyFromElement(element))
+}

common/shrinkingmap.go

@@ -47,6 +47,22 @@ func (s *ShrinkingMap[K, V]) Delete(key K) (deleted bool) {
 	return true
 }
 
+func (s *ShrinkingMap[K, V]) Keys() []K {
+	var keys []K
+	for k := range s.m {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+func (s *ShrinkingMap[K, V]) Values() []V {
+	var values []V
+	for _, v := range s.m {
+		values = append(values, v)
+	}
+	return values
+}
+
 func (s *ShrinkingMap[K, V]) Size() (size int) {
 	return len(s.m)
 }

eth/backend.go

@@ -57,6 +57,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/rlp"
 	"github.com/scroll-tech/go-ethereum/rollup/ccc"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
 	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 	"github.com/scroll-tech/go-ethereum/rpc"
@@ -109,7 +110,7 @@ type Ethereum struct {
 
 // New creates a new Ethereum object (including the
 // initialisation of the common Ethereum object)
-func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthClient) (*Ethereum, error) {
+func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ethereum, error) {
 	// Ensure configuration values are compatible and sane
 	if config.SyncMode == downloader.LightSync {
 		return nil, errors.New("can't run eth.Ethereum in light sync mode, use les.LightEthereum")

rollup/da_syncer/blob_client/beacon_node_client.go

@@ -12,12 +12,10 @@ import (
 
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/crypto/kzg4844"
-	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
 )
 
 type BeaconNodeClient struct {
 	apiEndpoint    string
-	l1Client       *rollup_sync_service.L1Client
 	genesisTime    uint64
 	secondsPerSlot uint64
 }
@@ -28,7 +26,7 @@ var (
 	beaconNodeBlobEndpoint    = "/eth/v1/beacon/blob_sidecars"
 )
 
-func NewBeaconNodeClient(apiEndpoint string, l1Client *rollup_sync_service.L1Client) (*BeaconNodeClient, error) {
+func NewBeaconNodeClient(apiEndpoint string) (*BeaconNodeClient, error) {
 	// get genesis time
 	genesisPath, err := url.JoinPath(apiEndpoint, beaconNodeGenesisEndpoint)
 	if err != nil {
@@ -94,19 +92,13 @@ func NewBeaconNodeClient(apiEndpoint string, l1Client *rollup_sync_service.L1Cli
 
 	return &BeaconNodeClient{
 		apiEndpoint:    apiEndpoint,
-		l1Client:       l1Client,
 		genesisTime:    genesisTime,
 		secondsPerSlot: secondsPerSlot,
 	}, nil
 }
 
-func (c *BeaconNodeClient) GetBlobByVersionedHashAndBlockNumber(ctx context.Context, versionedHash common.Hash, blockNumber uint64) (*kzg4844.Blob, error) {
+func (c *BeaconNodeClient) GetBlobByVersionedHashAndBlockTime(ctx context.Context, versionedHash common.Hash, blockTime uint64) (*kzg4844.Blob, error) {
-	// get block timestamp to calculate slot
+	slot := (blockTime - c.genesisTime) / c.secondsPerSlot
-	header, err := c.l1Client.GetHeaderByNumber(blockNumber)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get header by number, err: %w", err)
-	}
-	slot := (header.Time - c.genesisTime) / c.secondsPerSlot
 
 	// get blob sidecar for slot
 	blobSidecarPath, err := url.JoinPath(c.apiEndpoint, beaconNodeBlobEndpoint, fmt.Sprintf("%d", slot))
@@ -156,7 +148,7 @@ func (c *BeaconNodeClient) GetBlobByVersionedHashAndBlockNumber(ctx context.Cont
 		}
 	}
 
-	return nil, fmt.Errorf("missing blob %v in slot %d, block number %d", versionedHash, slot, blockNumber)
+	return nil, fmt.Errorf("missing blob %v in slot %d", versionedHash, slot)
 }
 
 type GenesisResp struct {

rollup/da_syncer/blob_client/blob_client.go

@@ -17,7 +17,7 @@ const (
 )
 
 type BlobClient interface {
-	GetBlobByVersionedHashAndBlockNumber(ctx context.Context, versionedHash common.Hash, blockNumber uint64) (*kzg4844.Blob, error)
+	GetBlobByVersionedHashAndBlockTime(ctx context.Context, versionedHash common.Hash, blockTime uint64) (*kzg4844.Blob, error)
 }
 
 type BlobClients struct {
@@ -32,13 +32,13 @@ func NewBlobClients(blobClients ...BlobClient) *BlobClients {
 	}
 }
 
-func (c *BlobClients) GetBlobByVersionedHashAndBlockNumber(ctx context.Context, versionedHash common.Hash, blockNumber uint64) (*kzg4844.Blob, error) {
+func (c *BlobClients) GetBlobByVersionedHashAndBlockTime(ctx context.Context, versionedHash common.Hash, blockTime uint64) (*kzg4844.Blob, error) {
 	if len(c.list) == 0 {
 		return nil, fmt.Errorf("BlobClients.GetBlobByVersionedHash: list of BlobClients is empty")
 	}
 
 	for i := 0; i < len(c.list); i++ {
-		blob, err := c.list[c.curPos].GetBlobByVersionedHashAndBlockNumber(ctx, versionedHash, blockNumber)
+		blob, err := c.list[c.curPos].GetBlobByVersionedHashAndBlockTime(ctx, versionedHash, blockTime)
 		if err == nil {
 			return blob, nil
 		}

rollup/da_syncer/blob_client/blob_scan_client.go

@@ -26,7 +26,7 @@ func NewBlobScanClient(apiEndpoint string) *BlobScanClient {
 	}
 }
 
-func (c *BlobScanClient) GetBlobByVersionedHashAndBlockNumber(ctx context.Context, versionedHash common.Hash, blockNumber uint64) (*kzg4844.Blob, error) {
+func (c *BlobScanClient) GetBlobByVersionedHashAndBlockTime(ctx context.Context, versionedHash common.Hash, blockTime uint64) (*kzg4844.Blob, error) {
 	// blobscan api docs https://api.blobscan.com/#/blobs/blob-getByBlobId
 	path, err := url.JoinPath(c.apiEndpoint, versionedHash.String())
 	if err != nil {

rollup/da_syncer/blob_client/block_native_client.go

@@ -24,7 +24,7 @@ func NewBlockNativeClient(apiEndpoint string) *BlockNativeClient {
 	}
 }
 
-func (c *BlockNativeClient) GetBlobByVersionedHashAndBlockNumber(ctx context.Context, versionedHash common.Hash, blockNumber uint64) (*kzg4844.Blob, error) {
+func (c *BlockNativeClient) GetBlobByVersionedHashAndBlockTime(ctx context.Context, versionedHash common.Hash, blockTime uint64) (*kzg4844.Blob, error) {
 	// blocknative api docs https://docs.blocknative.com/blocknative-data-archive/blob-archive
 	path, err := url.JoinPath(c.apiEndpoint, versionedHash.String())
 	if err != nil {

rollup/da_syncer/da/calldata_blob_source.go

@@ -8,25 +8,14 @@ import (
 	"github.com/scroll-tech/da-codec/encoding"
 
 	"github.com/scroll-tech/go-ethereum/accounts/abi"
-	"github.com/scroll-tech/go-ethereum/common"
-	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/ethdb"
-	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/blob_client"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/serrors"
-	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 )
 
 const (
-	callDataBlobSourceFetchBlockRange  uint64 = 500
+	callDataBlobSourceFetchBlockRange uint64 = 500
-	commitBatchEventName                      = "CommitBatch"
-	revertBatchEventName                      = "RevertBatch"
-	finalizeBatchEventName                    = "FinalizeBatch"
-	commitBatchMethodName                     = "commitBatch"
-	commitBatchWithBlobProofMethodName        = "commitBatchWithBlobProof"
-
-	// the length of method ID at the beginning of transaction data
-	methodIDLength = 4
 )
 
 var (
@@ -34,46 +23,40 @@ var (
 )
 
 type CalldataBlobSource struct {
-	ctx                           context.Context
+	ctx            context.Context
-	l1Client                      *rollup_sync_service.L1Client
+	l1Reader       *l1.Reader
-	blobClient                    blob_client.BlobClient
+	blobClient     blob_client.BlobClient
-	l1height                      uint64
+	l1Height       uint64
-	scrollChainABI                *abi.ABI
+	scrollChainABI *abi.ABI
-	l1CommitBatchEventSignature   common.Hash
+	db             ethdb.Database
-	l1RevertBatchEventSignature   common.Hash
-	l1FinalizeBatchEventSignature common.Hash
-	db                            ethdb.Database
 
 	l1Finalized uint64
 }
 
-func NewCalldataBlobSource(ctx context.Context, l1height uint64, l1Client *rollup_sync_service.L1Client, blobClient blob_client.BlobClient, db ethdb.Database) (*CalldataBlobSource, error) {
+func NewCalldataBlobSource(ctx context.Context, l1height uint64, l1Reader *l1.Reader, blobClient blob_client.BlobClient, db ethdb.Database) (*CalldataBlobSource, error) {
-	scrollChainABI, err := rollup_sync_service.ScrollChainMetaData.GetAbi()
+	scrollChainABI, err := l1.ScrollChainMetaData.GetAbi()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get scroll chain abi: %w", err)
 	}
 	return &CalldataBlobSource{
-		ctx:                           ctx,
+		ctx:            ctx,
-		l1Client:                      l1Client,
+		l1Reader:       l1Reader,
-		blobClient:                    blobClient,
+		blobClient:     blobClient,
-		l1height:                      l1height,
+		l1Height:       l1height,
-		scrollChainABI:                scrollChainABI,
+		scrollChainABI: scrollChainABI,
-		l1CommitBatchEventSignature:   scrollChainABI.Events[commitBatchEventName].ID,
+		db:             db,
-		l1RevertBatchEventSignature:   scrollChainABI.Events[revertBatchEventName].ID,
-		l1FinalizeBatchEventSignature: scrollChainABI.Events[finalizeBatchEventName].ID,
-		db:                            db,
 	}, nil
 }
 
 func (ds *CalldataBlobSource) NextData() (Entries, error) {
 	var err error
-	to := ds.l1height + callDataBlobSourceFetchBlockRange
+	to := ds.l1Height + callDataBlobSourceFetchBlockRange
 
 	// If there's not enough finalized blocks to request up to, we need to query finalized block number.
 	// Otherwise, we know that there's more finalized blocks than we want to request up to
 	// -> no need to query finalized block number
 	if to > ds.l1Finalized {
-		ds.l1Finalized, err = ds.l1Client.GetLatestFinalizedBlockNumber()
+		ds.l1Finalized, err = ds.l1Reader.GetLatestFinalizedBlockNumber()
 		if err != nil {
 			return nil, serrors.NewTemporaryError(fmt.Errorf("failed to query GetLatestFinalizedBlockNumber, error: %v", err))
 		}
@@ -81,69 +64,51 @@ func (ds *CalldataBlobSource) NextData() (Entries, error) {
 		to = min(to, ds.l1Finalized)
 	}
 
-	if ds.l1height > to {
+	if ds.l1Height > to {
 		return nil, ErrSourceExhausted
 	}
 
-	logs, err := ds.l1Client.FetchRollupEventsInRange(ds.l1height, to)
+	rollupEvents, err := ds.l1Reader.FetchRollupEventsInRange(ds.l1Height, to)
 	if err != nil {
-		return nil, serrors.NewTemporaryError(fmt.Errorf("cannot get events, l1height: %d, error: %v", ds.l1height, err))
+		return nil, serrors.NewTemporaryError(fmt.Errorf("cannot get rollup events, l1Height: %d, error: %v", ds.l1Height, err))
 	}
-	da, err := ds.processLogsToDA(logs)
+	da, err := ds.processRollupEventsToDA(rollupEvents)
 	if err != nil {
-		return nil, serrors.NewTemporaryError(fmt.Errorf("failed to process logs to DA, error: %v", err))
+		return nil, serrors.NewTemporaryError(fmt.Errorf("failed to process rollup events to DA, error: %v", err))
 	}
 
-	ds.l1height = to + 1
+	ds.l1Height = to + 1
 	return da, nil
 }
 
 func (ds *CalldataBlobSource) L1Height() uint64 {
-	return ds.l1height
+	return ds.l1Height
 }
 
-func (ds *CalldataBlobSource) processLogsToDA(logs []types.Log) (Entries, error) {
+func (ds *CalldataBlobSource) processRollupEventsToDA(rollupEvents l1.RollupEvents) (Entries, error) {
 	var entries Entries
 	var entry Entry
 	var err error
-
+	for _, rollupEvent := range rollupEvents {
-	for _, vLog := range logs {
+		switch rollupEvent.Type() {
-		switch vLog.Topics[0] {
+		case l1.CommitEventType:
-		case ds.l1CommitBatchEventSignature:
+			commitEvent, ok := rollupEvent.(*l1.CommitBatchEvent)
-			event := &rollup_sync_service.L1CommitBatchEvent{}
+			// this should never happen because we just check event type
-			if err = rollup_sync_service.UnpackLog(ds.scrollChainABI, event, commitBatchEventName, vLog); err != nil {
+			if !ok {
-				return nil, fmt.Errorf("failed to unpack commit rollup event log, err: %w", err)
+				return nil, fmt.Errorf("unexpected type of rollup event: %T", rollupEvent)
+			}
+			if entry, err = ds.getCommitBatchDA(commitEvent); err != nil {
+				return nil, fmt.Errorf("failed to get commit batch da: %v, err: %w", rollupEvent.BatchIndex().Uint64(), err)
 			}
 
-			batchIndex := event.BatchIndex.Uint64()
+		case l1.RevertEventType:
-			log.Trace("found new CommitBatch event", "batch index", batchIndex)
+			entry = NewRevertBatch(rollupEvent.BatchIndex().Uint64())
 
-			if entry, err = ds.getCommitBatchDA(batchIndex, &vLog); err != nil {
+		case l1.FinalizeEventType:
-				return nil, fmt.Errorf("failed to get commit batch da: %v, err: %w", batchIndex, err)
+			entry = NewFinalizeBatch(rollupEvent.BatchIndex().Uint64())
-			}
-
-		case ds.l1RevertBatchEventSignature:
-			event := &rollup_sync_service.L1RevertBatchEvent{}
-			if err = rollup_sync_service.UnpackLog(ds.scrollChainABI, event, revertBatchEventName, vLog); err != nil {
-				return nil, fmt.Errorf("failed to unpack revert rollup event log, err: %w", err)
-			}
-
-			batchIndex := event.BatchIndex.Uint64()
-			log.Trace("found new RevertBatchType event", "batch index", batchIndex)
-			entry = NewRevertBatch(batchIndex)
-
-		case ds.l1FinalizeBatchEventSignature:
-			event := &rollup_sync_service.L1FinalizeBatchEvent{}
-			if err = rollup_sync_service.UnpackLog(ds.scrollChainABI, event, finalizeBatchEventName, vLog); err != nil {
-				return nil, fmt.Errorf("failed to unpack finalized rollup event log, err: %w", err)
-			}
-
-			batchIndex := event.BatchIndex.Uint64()
-			log.Trace("found new FinalizeBatchType event", "batch index", event.BatchIndex.Uint64())
-			entry = NewFinalizeBatch(batchIndex)
 
 		default:
-			return nil, fmt.Errorf("unknown event, topic: %v, tx hash: %v", vLog.Topics[0].Hex(), vLog.TxHash.Hex())
+			return nil, fmt.Errorf("unknown rollup event, type: %v", rollupEvent.Type())
 		}
 
 		entries = append(entries, entry)
@@ -151,97 +116,27 @@ func (ds *CalldataBlobSource) processLogsToDA(logs []types.Log) (Entries, error)
 	return entries, nil
 }
 
-type commitBatchArgs struct {
+func (ds *CalldataBlobSource) getCommitBatchDA(commitEvent *l1.CommitBatchEvent) (Entry, error) {
-	Version                uint8
+	if commitEvent.BatchIndex().Uint64() == 0 {
-	ParentBatchHeader      []byte
-	Chunks                 [][]byte
-	SkippedL1MessageBitmap []byte
-}
-
-func newCommitBatchArgs(method *abi.Method, values []interface{}) (*commitBatchArgs, error) {
-	var args commitBatchArgs
-	err := method.Inputs.Copy(&args, values)
-	return &args, err
-}
-
-func newCommitBatchArgsFromCommitBatchWithProof(method *abi.Method, values []interface{}) (*commitBatchArgs, error) {
-	var args commitBatchWithBlobProofArgs
-	err := method.Inputs.Copy(&args, values)
-	if err != nil {
-		return nil, err
-	}
-	return &commitBatchArgs{
-		Version:                args.Version,
-		ParentBatchHeader:      args.ParentBatchHeader,
-		Chunks:                 args.Chunks,
-		SkippedL1MessageBitmap: args.SkippedL1MessageBitmap,
-	}, nil
-}
-
-type commitBatchWithBlobProofArgs struct {
-	Version                uint8
-	ParentBatchHeader      []byte
-	Chunks                 [][]byte
-	SkippedL1MessageBitmap []byte
-	BlobDataProof          []byte
-}
-
-func (ds *CalldataBlobSource) getCommitBatchDA(batchIndex uint64, vLog *types.Log) (Entry, error) {
-	if batchIndex == 0 {
 		return NewCommitBatchDAV0Empty(), nil
 	}
 
-	txData, err := ds.l1Client.FetchTxData(vLog)
+	args, err := ds.l1Reader.FetchCommitTxData(commitEvent)
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch tx data, tx hash: %v, err: %w", vLog.TxHash.Hex(), err)
+		return nil, fmt.Errorf("failed to fetch commit tx data of batch %d, tx hash: %v, err: %w", commitEvent.BatchIndex().Uint64(), commitEvent.TxHash().Hex(), err)
-	}
-	if len(txData) < methodIDLength {
-		return nil, fmt.Errorf("transaction data is too short, length of tx data: %v, minimum length required: %v", len(txData), methodIDLength)
 	}
 
-	method, err := ds.scrollChainABI.MethodById(txData[:methodIDLength])
+	codec, err := encoding.CodecFromVersion(encoding.CodecVersion(args.Version))
 	if err != nil {
-		return nil, fmt.Errorf("failed to get method by ID, ID: %v, err: %w", txData[:methodIDLength], err)
+		return nil, fmt.Errorf("unsupported codec version: %v, batch index: %v, err: %w", args.Version, commitEvent.BatchIndex().Uint64(), err)
-	}
-	values, err := method.Inputs.Unpack(txData[methodIDLength:])
-	if err != nil {
-		return nil, fmt.Errorf("failed to unpack transaction data using ABI, tx data: %v, err: %w", txData, err)
-	}
-	if method.Name == commitBatchMethodName {
-		args, err := newCommitBatchArgs(method, values)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode calldata into commitBatch args, values: %+v, err: %w", values, err)
-		}
-		codecVersion := encoding.CodecVersion(args.Version)
-		codec, err := encoding.CodecFromVersion(codecVersion)
-		if err != nil {
-			return nil, fmt.Errorf("unsupported codec version: %v, batch index: %v, err: %w", codecVersion, batchIndex, err)
-		}
-		switch args.Version {
-		case 0:
-			return NewCommitBatchDAV0(ds.db, codec, args.Version, batchIndex, args.ParentBatchHeader, args.Chunks, args.SkippedL1MessageBitmap, vLog.BlockNumber)
-		case 1, 2:
-			return NewCommitBatchDAWithBlob(ds.ctx, ds.db, codec, ds.l1Client, ds.blobClient, vLog, args.Version, batchIndex, args.ParentBatchHeader, args.Chunks, args.SkippedL1MessageBitmap)
-		default:
-			return nil, fmt.Errorf("failed to decode DA, codec version is unknown: codec version: %d", args.Version)
-		}
-	} else if method.Name == commitBatchWithBlobProofMethodName {
-		args, err := newCommitBatchArgsFromCommitBatchWithProof(method, values)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode calldata into commitBatch args, values: %+v, err: %w", values, err)
-		}
-		codecVersion := encoding.CodecVersion(args.Version)
-		codec, err := encoding.CodecFromVersion(codecVersion)
-		if err != nil {
-			return nil, fmt.Errorf("unsupported codec version: %v, batch index: %v, err: %w", codecVersion, batchIndex, err)
-		}
-		switch args.Version {
-		case 3, 4:
-			return NewCommitBatchDAWithBlob(ds.ctx, ds.db, codec, ds.l1Client, ds.blobClient, vLog, args.Version, batchIndex, args.ParentBatchHeader, args.Chunks, args.SkippedL1MessageBitmap)
-		default:
-			return nil, fmt.Errorf("failed to decode DA, codec version is unknown: codec version: %d", args.Version)
-		}
 	}
 
-	return nil, fmt.Errorf("unknown method name: %s", method.Name)
+	switch codec.Version() {
+	case 0:
+		return NewCommitBatchDAV0(ds.db, codec, commitEvent, args.ParentBatchHeader, args.Chunks, args.SkippedL1MessageBitmap)
+	case 1, 2, 3, 4:
+		return NewCommitBatchDAWithBlob(ds.ctx, ds.db, ds.l1Reader, ds.blobClient, codec, commitEvent, args.ParentBatchHeader, args.Chunks, args.SkippedL1MessageBitmap)
+	default:
+		return nil, fmt.Errorf("failed to decode DA, codec version is unknown: codec version: %d", args.Version)
+	}
 }

rollup/da_syncer/da/commitV0.go

@@ -10,6 +10,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/serrors"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 )
 
 type CommitBatchDAV0 struct {
@@ -25,19 +26,17 @@ type CommitBatchDAV0 struct {
 
 func NewCommitBatchDAV0(db ethdb.Database,
 	codec encoding.Codec,
-	version uint8,
+	commitEvent *l1.CommitBatchEvent,
-	batchIndex uint64,
 	parentBatchHeader []byte,
 	chunks [][]byte,
 	skippedL1MessageBitmap []byte,
-	l1BlockNumber uint64,
 ) (*CommitBatchDAV0, error) {
 	decodedChunks, err := codec.DecodeDAChunksRawTx(chunks)
 	if err != nil {
-		return nil, fmt.Errorf("failed to unpack chunks: %d, err: %w", batchIndex, err)
+		return nil, fmt.Errorf("failed to unpack chunks: %d, err: %w", commitEvent.BatchIndex().Uint64(), err)
 	}
 
-	return NewCommitBatchDAV0WithChunks(db, version, batchIndex, parentBatchHeader, decodedChunks, skippedL1MessageBitmap, l1BlockNumber)
+	return NewCommitBatchDAV0WithChunks(db, uint8(codec.Version()), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent.BlockNumber())
 }
 
 func NewCommitBatchDAV0WithChunks(db ethdb.Database,
@@ -141,6 +140,7 @@ func getTotalMessagesPoppedFromChunks(decodedChunks []*encoding.DAChunkRawTx) in
 
 func getL1Messages(db ethdb.Database, parentTotalL1MessagePopped uint64, skippedBitmap []byte, totalL1MessagePopped int) ([]*types.L1MessageTx, error) {
 	var txs []*types.L1MessageTx
+
 	decodedSkippedBitmap, err := encoding.DecodeBitmap(skippedBitmap, totalL1MessagePopped)
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode skipped message bitmap: err: %w", err)

rollup/da_syncer/da/commitV1.go

@@ -8,10 +8,9 @@ import (
 	"github.com/scroll-tech/da-codec/encoding"
 
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/blob_client"
-	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 
 	"github.com/scroll-tech/go-ethereum/common"
-	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/crypto/kzg4844"
 	"github.com/scroll-tech/go-ethereum/ethdb"
 )
@@ -21,32 +20,34 @@ type CommitBatchDAV1 struct {
 }
 
 func NewCommitBatchDAWithBlob(ctx context.Context, db ethdb.Database,
-	codec encoding.Codec,
+	l1Reader *l1.Reader,
-	l1Client *rollup_sync_service.L1Client,
 	blobClient blob_client.BlobClient,
-	vLog *types.Log,
+	codec encoding.Codec,
-	version uint8,
+	commitEvent *l1.CommitBatchEvent,
-	batchIndex uint64,
 	parentBatchHeader []byte,
 	chunks [][]byte,
 	skippedL1MessageBitmap []byte,
 ) (*CommitBatchDAV1, error) {
 	decodedChunks, err := codec.DecodeDAChunksRawTx(chunks)
 	if err != nil {
-		return nil, fmt.Errorf("failed to unpack chunks: %v, err: %w", batchIndex, err)
+		return nil, fmt.Errorf("failed to unpack chunks: %v, err: %w", commitEvent.BatchIndex().Uint64(), err)
 	}
 
-	versionedHash, err := l1Client.FetchTxBlobHash(vLog)
+	versionedHash, err := l1Reader.FetchTxBlobHash(commitEvent.TxHash(), commitEvent.BlockHash())
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch blob hash, err: %w", err)
 	}
 
-	blob, err := blobClient.GetBlobByVersionedHashAndBlockNumber(ctx, versionedHash, vLog.BlockNumber)
+	header, err := l1Reader.FetchBlockHeaderByNumber(commitEvent.BlockNumber())
+	if err != nil {
+		return nil, fmt.Errorf("failed to get header by number, err: %w", err)
+	}
+	blob, err := blobClient.GetBlobByVersionedHashAndBlockTime(ctx, versionedHash, header.Time)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch blob from blob client, err: %w", err)
 	}
 	if blob == nil {
-		return nil, fmt.Errorf("unexpected, blob == nil and err != nil, batch index: %d, versionedHash: %s, blobClient: %T", batchIndex, versionedHash.String(), blobClient)
+		return nil, fmt.Errorf("unexpected, blob == nil and err != nil, batch index: %d, versionedHash: %s, blobClient: %T", commitEvent.BatchIndex().Uint64(), versionedHash.String(), blobClient)
 	}
 
 	// compute blob versioned hash and compare with one from tx
@@ -69,7 +70,7 @@ func NewCommitBatchDAWithBlob(ctx context.Context, db ethdb.Database,
 		return nil, fmt.Errorf("decodedChunks is nil after decoding")
 	}
 
-	v0, err := NewCommitBatchDAV0WithChunks(db, version, batchIndex, parentBatchHeader, decodedChunks, skippedL1MessageBitmap, vLog.BlockNumber)
+	v0, err := NewCommitBatchDAV0WithChunks(db, uint8(codec.Version()), commitEvent.BatchIndex().Uint64(), parentBatchHeader, decodedChunks, skippedL1MessageBitmap, commitEvent.BlockNumber())
 	if err != nil {
 		return nil, err
 	}

rollup/da_syncer/data_source.go

@@ -8,7 +8,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/blob_client"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/da"
-	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 )
 
 type DataSource interface {
@@ -19,21 +19,21 @@ type DataSource interface {
 type DataSourceFactory struct {
 	config        Config
 	genesisConfig *params.ChainConfig
-	l1Client      *rollup_sync_service.L1Client
+	l1Reader      *l1.Reader
 	blobClient    blob_client.BlobClient
 	db            ethdb.Database
 }
 
-func NewDataSourceFactory(blockchain *core.BlockChain, genesisConfig *params.ChainConfig, config Config, l1Client *rollup_sync_service.L1Client, blobClient blob_client.BlobClient, db ethdb.Database) *DataSourceFactory {
+func NewDataSourceFactory(blockchain *core.BlockChain, genesisConfig *params.ChainConfig, config Config, l1Reader *l1.Reader, blobClient blob_client.BlobClient, db ethdb.Database) *DataSourceFactory {
 	return &DataSourceFactory{
 		config:        config,
 		genesisConfig: genesisConfig,
-		l1Client:      l1Client,
+		l1Reader:      l1Reader,
 		blobClient:    blobClient,
 		db:            db,
 	}
 }
 
 func (ds *DataSourceFactory) OpenDataSource(ctx context.Context, l1height uint64) (DataSource, error) {
-	return da.NewCalldataBlobSource(ctx, l1height, ds.l1Client, ds.blobClient, ds.db)
+	return da.NewCalldataBlobSource(ctx, l1height, ds.l1Reader, ds.blobClient, ds.db)
 }

rollup/da_syncer/syncing_pipeline.go

@@ -15,8 +15,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/blob_client"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/serrors"
-	"github.com/scroll-tech/go-ethereum/rollup/rollup_sync_service"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
-	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 )
 
 // Config is the configuration parameters of data availability syncing.
@@ -42,20 +41,18 @@ type SyncingPipeline struct {
 	daSyncer   *DASyncer
 }
 
-func NewSyncingPipeline(ctx context.Context, blockchain *core.BlockChain, genesisConfig *params.ChainConfig, db ethdb.Database, ethClient sync_service.EthClient, l1DeploymentBlock uint64, config Config) (*SyncingPipeline, error) {
+func NewSyncingPipeline(ctx context.Context, blockchain *core.BlockChain, genesisConfig *params.ChainConfig, db ethdb.Database, ethClient l1.Client, l1DeploymentBlock uint64, config Config) (*SyncingPipeline, error) {
-	scrollChainABI, err := rollup_sync_service.ScrollChainMetaData.GetAbi()
+	l1Reader, err := l1.NewReader(ctx, l1.Config{
+		ScrollChainAddress:    genesisConfig.Scroll.L1Config.ScrollChainAddress,
+		L1MessageQueueAddress: genesisConfig.Scroll.L1Config.L1MessageQueueAddress,
+	}, ethClient)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get scroll chain abi: %w", err)
+		return nil, fmt.Errorf("failed to initialize l1.Reader, err = %w", err)
-	}
-
-	l1Client, err := rollup_sync_service.NewL1Client(ctx, ethClient, genesisConfig.Scroll.L1Config.L1ChainId, genesisConfig.Scroll.L1Config.ScrollChainAddress, scrollChainABI)
-	if err != nil {
-		return nil, err
 	}
 
 	blobClientList := blob_client.NewBlobClients()
 	if config.BeaconNodeAPIEndpoint != "" {
-		beaconNodeClient, err := blob_client.NewBeaconNodeClient(config.BeaconNodeAPIEndpoint, l1Client)
+		beaconNodeClient, err := blob_client.NewBeaconNodeClient(config.BeaconNodeAPIEndpoint)
 		if err != nil {
 			log.Warn("failed to create BeaconNodeClient", "err", err)
 		} else {
@@ -72,7 +69,7 @@ func NewSyncingPipeline(ctx context.Context, blockchain *core.BlockChain, genesi
 		return nil, errors.New("DA syncing is enabled but no blob client is configured. Please provide at least one blob client via command line flag")
 	}
 
-	dataSourceFactory := NewDataSourceFactory(blockchain, genesisConfig, config, l1Client, blobClientList, db)
+	dataSourceFactory := NewDataSourceFactory(blockchain, genesisConfig, config, l1Reader, blobClientList, db)
 	syncedL1Height := l1DeploymentBlock - 1
 	from := rawdb.ReadDASyncedL1BlockNumber(db)
 	if from != nil {

rollup/l1/abi_test.go

@@ -0,0 +1,79 @@
+package l1
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/crypto"
+)
+
+func TestEventSignatures(t *testing.T) {
+	assert.Equal(t, crypto.Keccak256Hash([]byte("CommitBatch(uint256,bytes32)")), ScrollChainABI.Events["CommitBatch"].ID)
+	assert.Equal(t, crypto.Keccak256Hash([]byte("RevertBatch(uint256,bytes32)")), ScrollChainABI.Events["RevertBatch"].ID)
+	assert.Equal(t, crypto.Keccak256Hash([]byte("FinalizeBatch(uint256,bytes32,bytes32,bytes32)")), ScrollChainABI.Events["FinalizeBatch"].ID)
+}
+
+func TestUnpackLog(t *testing.T) {
+	mockBatchIndex := big.NewInt(123)
+	mockBatchHash := crypto.Keccak256Hash([]byte("mockBatch"))
+	mockStateRoot := crypto.Keccak256Hash([]byte("mockStateRoot"))
+	mockWithdrawRoot := crypto.Keccak256Hash([]byte("mockWithdrawRoot"))
+
+	tests := []struct {
+		eventName string
+		mockLog   types.Log
+		expected  interface{}
+		out       interface{}
+	}{
+		{
+			commitBatchEventName,
+			types.Log{
+				Data:   nil,
+				Topics: []common.Hash{ScrollChainABI.Events[commitBatchEventName].ID, common.BigToHash(mockBatchIndex), mockBatchHash},
+			},
+			&CommitBatchEventUnpacked{
+				BatchIndex: mockBatchIndex,
+				BatchHash:  mockBatchHash,
+			},
+			&CommitBatchEventUnpacked{},
+		},
+		{
+			revertBatchEventName,
+			types.Log{
+				Data:   nil,
+				Topics: []common.Hash{ScrollChainABI.Events[revertBatchEventName].ID, common.BigToHash(mockBatchIndex), mockBatchHash},
+			},
+			&RevertBatchEventUnpacked{
+				BatchIndex: mockBatchIndex,
+				BatchHash:  mockBatchHash,
+			},
+			&RevertBatchEventUnpacked{},
+		},
+		{
+			finalizeBatchEventName,
+			types.Log{
+				Data:   append(mockStateRoot.Bytes(), mockWithdrawRoot.Bytes()...),
+				Topics: []common.Hash{ScrollChainABI.Events[finalizeBatchEventName].ID, common.BigToHash(mockBatchIndex), mockBatchHash},
+			},
+			&FinalizeBatchEventUnpacked{
+				BatchIndex:   mockBatchIndex,
+				BatchHash:    mockBatchHash,
+				StateRoot:    mockStateRoot,
+				WithdrawRoot: mockWithdrawRoot,
+			},
+			&FinalizeBatchEventUnpacked{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.eventName, func(t *testing.T) {
+			err := UnpackLog(ScrollChainABI, tt.out, tt.eventName, tt.mockLog)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expected, tt.out)
+		})
+	}
+}

rollup/l1/reader.go

@@ -0,0 +1,380 @@
+package l1
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/big"
+
+	"github.com/scroll-tech/go-ethereum"
+	"github.com/scroll-tech/go-ethereum/accounts/abi"
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rpc"
+)
+
+const (
+	commitBatchEventName      = "CommitBatch"
+	revertBatchEventName      = "RevertBatch"
+	finalizeBatchEventName    = "FinalizeBatch"
+	nextUnfinalizedQueueIndex = "nextUnfinalizedQueueIndex"
+	lastFinalizedBatchIndex   = "lastFinalizedBatchIndex"
+
+	defaultRollupEventsFetchBlockRange = 100
+)
+
+type Reader struct {
+	ctx    context.Context
+	config Config
+	client Client
+
+	scrollChainABI                *abi.ABI
+	l1MessageQueueABI             *abi.ABI
+	l1CommitBatchEventSignature   common.Hash
+	l1RevertBatchEventSignature   common.Hash
+	l1FinalizeBatchEventSignature common.Hash
+}
+
+// Config is the configuration parameters of data availability syncing.
+type Config struct {
+	ScrollChainAddress    common.Address // address of ScrollChain contract
+	L1MessageQueueAddress common.Address // address of L1MessageQueue contract
+}
+
+// NewReader initializes a new Reader instance
+func NewReader(ctx context.Context, config Config, l1Client Client) (*Reader, error) {
+	if config.ScrollChainAddress == (common.Address{}) {
+		return nil, errors.New("must pass non-zero scrollChainAddress to L1Client")
+	}
+
+	if config.L1MessageQueueAddress == (common.Address{}) {
+		return nil, errors.New("must pass non-zero l1MessageQueueAddress to L1Client")
+	}
+
+	reader := Reader{
+		ctx:    ctx,
+		config: config,
+		client: l1Client,
+
+		scrollChainABI:                ScrollChainABI,
+		l1MessageQueueABI:             L1MessageQueueABIManual,
+		l1CommitBatchEventSignature:   ScrollChainABI.Events[commitBatchEventName].ID,
+		l1RevertBatchEventSignature:   ScrollChainABI.Events[revertBatchEventName].ID,
+		l1FinalizeBatchEventSignature: ScrollChainABI.Events[finalizeBatchEventName].ID,
+	}
+
+	return &reader, nil
+}
+
+func (r *Reader) FinalizedL1MessageQueueIndex(blockNumber uint64) (uint64, error) {
+	data, err := r.l1MessageQueueABI.Pack(nextUnfinalizedQueueIndex)
+	if err != nil {
+		return 0, fmt.Errorf("failed to pack %s: %w", nextUnfinalizedQueueIndex, err)
+	}
+
+	result, err := r.client.CallContract(r.ctx, ethereum.CallMsg{
+		To:   &r.config.L1MessageQueueAddress,
+		Data: data,
+	}, new(big.Int).SetUint64(blockNumber))
+	if err != nil {
+		return 0, fmt.Errorf("failed to call %s: %w", nextUnfinalizedQueueIndex, err)
+	}
+
+	var parsedResult *big.Int
+	if err = r.l1MessageQueueABI.UnpackIntoInterface(&parsedResult, nextUnfinalizedQueueIndex, result); err != nil {
+		return 0, fmt.Errorf("failed to unpack result: %w", err)
+	}
+
+	next := parsedResult.Uint64()
+	if next == 0 {
+		return 0, nil
+	}
+
+	return next - 1, nil
+}
+
+func (r *Reader) LatestFinalizedBatch(blockNumber uint64) (uint64, error) {
+	data, err := r.scrollChainABI.Pack(lastFinalizedBatchIndex)
+	if err != nil {
+		return 0, fmt.Errorf("failed to pack %s: %w", lastFinalizedBatchIndex, err)
+	}
+
+	result, err := r.client.CallContract(r.ctx, ethereum.CallMsg{
+		To:   &r.config.ScrollChainAddress,
+		Data: data,
+	}, new(big.Int).SetUint64(blockNumber))
+	if err != nil {
+		return 0, fmt.Errorf("failed to call %s: %w", lastFinalizedBatchIndex, err)
+	}
+
+	var parsedResult *big.Int
+	if err = r.scrollChainABI.UnpackIntoInterface(&parsedResult, lastFinalizedBatchIndex, result); err != nil {
+		return 0, fmt.Errorf("failed to unpack result: %w", err)
+	}
+
+	return parsedResult.Uint64(), nil
+}
+
+// GetLatestFinalizedBlockNumber fetches the block number of the latest finalized block from the L1 chain.
+func (r *Reader) GetLatestFinalizedBlockNumber() (uint64, error) {
+	header, err := r.client.HeaderByNumber(r.ctx, big.NewInt(int64(rpc.FinalizedBlockNumber)))
+	if err != nil {
+		return 0, err
+	}
+	if !header.Number.IsInt64() {
+		return 0, fmt.Errorf("received unexpected block number in L1Client: %v", header.Number)
+	}
+	return header.Number.Uint64(), nil
+}
+
+// FetchBlockHeaderByNumber fetches the block header by number
+func (r *Reader) FetchBlockHeaderByNumber(blockNumber uint64) (*types.Header, error) {
+	return r.client.HeaderByNumber(r.ctx, big.NewInt(int64(blockNumber)))
+}
+
+// FetchTxData fetches tx data corresponding to given event log
+func (r *Reader) FetchTxData(txHash, blockHash common.Hash) ([]byte, error) {
+	tx, err := r.fetchTx(txHash, blockHash)
+	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) {
+	tx, err := r.fetchTx(txHash, blockHash)
+	if err != nil {
+		return common.Hash{}, 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 blobHashes[0], nil
+}
+
+// FetchRollupEventsInRange retrieves and parses commit/revert/finalize rollup events between block numbers: [from, to].
+func (r *Reader) FetchRollupEventsInRange(from, to uint64) (RollupEvents, error) {
+	log.Trace("L1Client fetchRollupEventsInRange", "fromBlock", from, "toBlock", to)
+	var logs []types.Log
+
+	err := queryInBatches(r.ctx, from, to, defaultRollupEventsFetchBlockRange, func(from, to uint64) (bool, error) {
+		query := ethereum.FilterQuery{
+			FromBlock: big.NewInt(int64(from)), // inclusive
+			ToBlock:   big.NewInt(int64(to)),   // inclusive
+			Addresses: []common.Address{
+				r.config.ScrollChainAddress,
+			},
+			Topics: make([][]common.Hash, 1),
+		}
+		query.Topics[0] = make([]common.Hash, 3)
+		query.Topics[0][0] = r.l1CommitBatchEventSignature
+		query.Topics[0][1] = r.l1RevertBatchEventSignature
+		query.Topics[0][2] = r.l1FinalizeBatchEventSignature
+
+		logsBatch, err := r.client.FilterLogs(r.ctx, query)
+		if err != nil {
+			return false, fmt.Errorf("failed to filter logs, err: %w", err)
+		}
+		logs = append(logs, logsBatch...)
+		return true, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return r.processLogsToRollupEvents(logs)
+}
+
+// FetchRollupEventsInRangeWithCallback retrieves and parses commit/revert/finalize rollup events between block numbers: [from, to].
+func (r *Reader) FetchRollupEventsInRangeWithCallback(from, to uint64, callback func(event RollupEvent) bool) error {
+	log.Trace("L1Client fetchRollupEventsInRange", "fromBlock", from, "toBlock", to)
+
+	err := queryInBatches(r.ctx, from, to, defaultRollupEventsFetchBlockRange, func(from, to uint64) (bool, error) {
+		query := ethereum.FilterQuery{
+			FromBlock: big.NewInt(int64(from)), // inclusive
+			ToBlock:   big.NewInt(int64(to)),   // inclusive
+			Addresses: []common.Address{
+				r.config.ScrollChainAddress,
+			},
+			Topics: make([][]common.Hash, 1),
+		}
+		query.Topics[0] = make([]common.Hash, 3)
+		query.Topics[0][0] = r.l1CommitBatchEventSignature
+		query.Topics[0][1] = r.l1RevertBatchEventSignature
+		query.Topics[0][2] = r.l1FinalizeBatchEventSignature
+
+		logsBatch, err := r.client.FilterLogs(r.ctx, query)
+		if err != nil {
+			return false, fmt.Errorf("failed to filter logs, err: %w", err)
+		}
+
+		rollupEvents, err := r.processLogsToRollupEvents(logsBatch)
+		if err != nil {
+			return false, fmt.Errorf("failed to process logs to rollup events, err: %w", err)
+		}
+
+		for _, event := range rollupEvents {
+			if !callback(event) {
+				return false, nil
+			}
+		}
+
+		return true, nil
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *Reader) processLogsToRollupEvents(logs []types.Log) (RollupEvents, error) {
+	var rollupEvents RollupEvents
+	var rollupEvent RollupEvent
+	var err error
+
+	for _, vLog := range logs {
+		switch vLog.Topics[0] {
+		case r.l1CommitBatchEventSignature:
+			event := &CommitBatchEventUnpacked{}
+			if err = UnpackLog(r.scrollChainABI, event, commitBatchEventName, vLog); err != nil {
+				return nil, fmt.Errorf("failed to unpack commit rollup event log, err: %w", err)
+			}
+			log.Trace("found new CommitBatch event", "batch index", event.BatchIndex.Uint64())
+			rollupEvent = &CommitBatchEvent{
+				batchIndex:  event.BatchIndex,
+				batchHash:   event.BatchHash,
+				txHash:      vLog.TxHash,
+				blockHash:   vLog.BlockHash,
+				blockNumber: vLog.BlockNumber,
+			}
+
+		case r.l1RevertBatchEventSignature:
+			event := &RevertBatchEventUnpacked{}
+			if err = UnpackLog(r.scrollChainABI, event, revertBatchEventName, vLog); err != nil {
+				return nil, fmt.Errorf("failed to unpack revert rollup event log, err: %w", err)
+			}
+			log.Trace("found new RevertBatchType event", "batch index", event.BatchIndex.Uint64())
+			rollupEvent = &RevertBatchEvent{
+				batchIndex:  event.BatchIndex,
+				batchHash:   event.BatchHash,
+				txHash:      vLog.TxHash,
+				blockHash:   vLog.BlockHash,
+				blockNumber: vLog.BlockNumber,
+			}
+
+		case r.l1FinalizeBatchEventSignature:
+			event := &FinalizeBatchEventUnpacked{}
+			if err = UnpackLog(r.scrollChainABI, event, finalizeBatchEventName, vLog); err != nil {
+				return nil, fmt.Errorf("failed to unpack finalized rollup event log, err: %w", err)
+			}
+			log.Trace("found new FinalizeBatchType event", "batch index", event.BatchIndex.Uint64())
+			rollupEvent = &FinalizeBatchEvent{
+				batchIndex:   event.BatchIndex,
+				batchHash:    event.BatchHash,
+				stateRoot:    event.StateRoot,
+				withdrawRoot: event.WithdrawRoot,
+				txHash:       vLog.TxHash,
+				blockHash:    vLog.BlockHash,
+				blockNumber:  vLog.BlockNumber,
+			}
+
+		default:
+			return nil, fmt.Errorf("unknown event, topic: %v, tx hash: %v", vLog.Topics[0].Hex(), vLog.TxHash.Hex())
+		}
+
+		rollupEvents = append(rollupEvents, rollupEvent)
+	}
+	return rollupEvents, nil
+}
+
+func queryInBatches(ctx context.Context, fromBlock, toBlock uint64, batchSize uint64, queryFunc func(from, to uint64) (bool, error)) error {
+	for from := fromBlock; from <= toBlock; from += batchSize {
+		// check if context is done and return if it is
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		to := from + batchSize - 1
+		if to > toBlock {
+			to = toBlock
+		}
+		cont, err := queryFunc(from, to)
+		if err != nil {
+			return fmt.Errorf("error querying blocks %d to %d: %w", from, to, err)
+		}
+		if !cont {
+			break
+		}
+	}
+	return nil
+}
+
+// fetchTx fetches tx corresponding to given event log
+func (r *Reader) fetchTx(txHash, blockHash common.Hash) (*types.Transaction, error) {
+	tx, _, err := r.client.TransactionByHash(r.ctx, txHash)
+	if err != nil {
+		log.Debug("failed to get transaction by hash, probably an unindexed transaction, fetching the whole block to get the transaction",
+			"tx hash", txHash.Hex(), "block hash", blockHash.Hex(), "err", err)
+		block, err := r.client.BlockByHash(r.ctx, blockHash)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get block by hash, block hash: %v, err: %w", blockHash.Hex(), err)
+		}
+
+		found := false
+		for _, txInBlock := range block.Transactions() {
+			if txInBlock.Hash() == txHash {
+				tx = txInBlock
+				found = true
+				break
+			}
+		}
+		if !found {
+			return nil, fmt.Errorf("transaction not found in the block, tx hash: %v, block hash: %v", txHash.Hex(), blockHash.Hex())
+		}
+	}
+
+	return tx, nil
+}
+
+func (r *Reader) FetchCommitTxData(commitEvent *CommitBatchEvent) (*CommitBatchArgs, error) {
+	tx, err := r.fetchTx(commitEvent.TxHash(), commitEvent.BlockHash())
+	if err != nil {
+		return nil, err
+	}
+	txData := tx.Data()
+
+	if len(txData) < methodIDLength {
+		return nil, fmt.Errorf("transaction data is too short, length of tx data: %v, minimum length required: %v", len(txData), methodIDLength)
+	}
+
+	method, err := r.scrollChainABI.MethodById(txData[:methodIDLength])
+	if err != nil {
+		return nil, fmt.Errorf("failed to get method by ID, ID: %v, err: %w", txData[:methodIDLength], err)
+	}
+	values, err := method.Inputs.Unpack(txData[methodIDLength:])
+	if err != nil {
+		return nil, fmt.Errorf("failed to unpack transaction data using ABI, tx data: %v, err: %w", txData, err)
+	}
+
+	var args *CommitBatchArgs
+	if method.Name == commitBatchMethodName {
+		args, err = newCommitBatchArgs(method, values)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode calldata into commitBatch args %s, values: %+v, err: %w", commitBatchMethodName, values, err)
+		}
+	} else if method.Name == commitBatchWithBlobProofMethodName {
+		args, err = newCommitBatchArgsFromCommitBatchWithProof(method, values)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode calldata into commitBatch args %s, values: %+v, err: %w", commitBatchWithBlobProofMethodName, values, err)
+		}
+	} else {
+		return nil, fmt.Errorf("unknown method name for commit transaction: %s", method.Name)
+	}
+
+	return args, nil
+}

rollup/l1/reader_test.go

@@ -0,0 +1,125 @@
+package l1
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestQueryInBatches(t *testing.T) {
+	tests := []struct {
+		name          string
+		fromBlock     uint64
+		toBlock       uint64
+		batchSize     uint64
+		queryFunc     func(from, to uint64) (bool, error)
+		expectErr     bool
+		expectedErr   string
+		expectedCalls []struct {
+			from uint64
+			to   uint64
+		}
+	}{
+		{
+			name:      "Successful query in single batch",
+			fromBlock: 1,
+			toBlock:   10,
+			batchSize: 10,
+			queryFunc: func(from, to uint64) (bool, error) {
+				return true, nil
+			},
+			expectErr: false,
+			expectedCalls: []struct {
+				from uint64
+				to   uint64
+			}{
+				{from: 1, to: 10},
+			},
+		},
+		{
+			name:      "Successful query in multiple batches",
+			fromBlock: 1,
+			toBlock:   80,
+			batchSize: 10,
+			queryFunc: func(from, to uint64) (bool, error) {
+				return true, nil
+			},
+			expectErr: false,
+			expectedCalls: []struct {
+				from uint64
+				to   uint64
+			}{
+				{from: 1, to: 10},
+				{from: 11, to: 20},
+				{from: 21, to: 30},
+				{from: 31, to: 40},
+				{from: 41, to: 50},
+				{from: 51, to: 60},
+				{from: 61, to: 70},
+				{from: 71, to: 80},
+			},
+		},
+		{
+			name:      "Query function returns error",
+			fromBlock: 1,
+			toBlock:   10,
+			batchSize: 10,
+			queryFunc: func(from, to uint64) (bool, error) {
+				return false, errors.New("query error")
+			},
+			expectErr:   true,
+			expectedErr: "error querying blocks 1 to 10: query error",
+			expectedCalls: []struct {
+				from uint64
+				to   uint64
+			}{
+				{from: 1, to: 10},
+			},
+		},
+		{
+			name:      "Query function returns false to stop",
+			fromBlock: 1,
+			toBlock:   20,
+			batchSize: 10,
+			queryFunc: func(from, to uint64) (bool, error) {
+				if from == 1 {
+					return false, nil
+				}
+				return true, nil
+			},
+			expectErr: false,
+			expectedCalls: []struct {
+				from uint64
+				to   uint64
+			}{
+				{from: 1, to: 10},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var calls []struct {
+				from uint64
+				to   uint64
+			}
+			queryFunc := func(from, to uint64) (bool, error) {
+				calls = append(calls, struct {
+					from uint64
+					to   uint64
+				}{from, to})
+				return tt.queryFunc(from, to)
+			}
+			err := queryInBatches(context.Background(), tt.fromBlock, tt.toBlock, tt.batchSize, queryFunc)
+			if tt.expectErr {
+				require.Error(t, err)
+				require.EqualError(t, err, tt.expectedErr)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tt.expectedCalls, calls)
+		})
+	}
+}

rollup/l1/types.go

@@ -0,0 +1,22 @@
+package l1
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/scroll-tech/go-ethereum"
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+)
+
+type Client interface {
+	BlockNumber(ctx context.Context) (uint64, error)
+	ChainID(ctx context.Context) (*big.Int, error)
+	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
+	HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
+	HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error)
+	SubscribeFilterLogs(ctx context.Context, query ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error)
+	TransactionByHash(ctx context.Context, txHash common.Hash) (tx *types.Transaction, isPending bool, err error)
+	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
+	CallContract(ctx context.Context, msg ethereum.CallMsg, blockNumber *big.Int) ([]byte, error)
+}