add rollup_sync_service (#600)

* update cmd/geth/main.go

* update cmd/utils/flags.go

* add core/rawdb/accessors_rollup_event.go and tests

* update eth/ethconfig/gen_config.go

* update eth/ethconfig/config.go

* update core/rawdb/schema.go

* update eth/backend.go

* fix rollup/rollup_sync_service/rollup_sync_service.go

cmd/geth/main.go

@@ -147,6 +147,7 @@ var (
 		utils.L1EndpointFlag,
 		utils.L1ConfirmationsFlag,
 		utils.L1DeploymentBlockFlag,
+		utils.RollupVerifyEnabledFlag,
 	}, utils.NetworkFlags, utils.DatabaseFlags)
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -961,6 +961,12 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Usage: "L1 block height to start syncing from. Should be set to the L1 message queue deployment block number.",
 	}
 
+	// Rollup verify service settings
+	RollupVerifyEnabledFlag = &cli.BoolFlag{
+		Name:  "rollup.verify",
+		Usage: "Enable verification of batch consistency between L1 and L2 in rollup",
+	}
+
 	// Max block range for `eth_getLogs` method
 	MaxBlockRangeFlag = &cli.Int64Flag{
 		Name:  "rpc.getlogs.maxrange",
@@ -1664,6 +1670,12 @@ func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {
 	}
 }
 
+func setEnableRollupVerify(ctx *cli.Context, cfg *ethconfig.Config) {
+	if ctx.IsSet(RollupVerifyEnabledFlag.Name) {
+		cfg.EnableRollupVerify = ctx.Bool(RollupVerifyEnabledFlag.Name)
+	}
+}
+
 func setMaxBlockRange(ctx *cli.Context, cfg *ethconfig.Config) {
 	if ctx.IsSet(MaxBlockRangeFlag.Name) {
 		cfg.MaxBlockRange = ctx.Int64(MaxBlockRangeFlag.Name)
@@ -1727,6 +1739,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	setMiner(ctx, &cfg.Miner)
 	setRequiredBlocks(ctx, cfg)
 	setLes(ctx, cfg)
+	setEnableRollupVerify(ctx, cfg)
 	setMaxBlockRange(ctx, cfg)
 
 	// Cap the cache allowance and tune the garbage collector

core/rawdb/accessors_rollup_event.go

@@ -0,0 +1,146 @@
+package rawdb
+
+import (
+	"bytes"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+// ChunkBlockRange represents the range of blocks within a chunk.
+type ChunkBlockRange struct {
+	StartBlockNumber uint64
+	EndBlockNumber   uint64
+}
+
+// FinalizedBatchMeta holds metadata for finalized batches.
+type FinalizedBatchMeta struct {
+	BatchHash            common.Hash
+	TotalL1MessagePopped uint64 // total number of L1 messages popped before and in this batch.
+	StateRoot            common.Hash
+	WithdrawRoot         common.Hash
+}
+
+// WriteRollupEventSyncedL1BlockNumber stores the latest synced L1 block number related to rollup events in the database.
+func WriteRollupEventSyncedL1BlockNumber(db ethdb.KeyValueWriter, l1BlockNumber uint64) {
+	value := big.NewInt(0).SetUint64(l1BlockNumber).Bytes()
+	if err := db.Put(rollupEventSyncedL1BlockNumberKey, value); err != nil {
+		log.Crit("failed to store rollup event synced L1 block number for rollup event", "err", err)
+	}
+}
+
+// ReadRollupEventSyncedL1BlockNumber fetches the highest synced L1 block number associated with rollup events from the database.
+func ReadRollupEventSyncedL1BlockNumber(db ethdb.Reader) *uint64 {
+	data, err := db.Get(rollupEventSyncedL1BlockNumberKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("failed to read rollup event synced L1 block number from database", "err", err)
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("unexpected rollup event synced L1 block number in database", "number", number)
+	}
+
+	rollupEventSyncedL1BlockNumber := number.Uint64()
+	return &rollupEventSyncedL1BlockNumber
+}
+
+// WriteBatchChunkRanges writes the block ranges for each chunk within a batch to the database.
+// It serializes the chunk ranges using RLP and stores them under a key derived from the batch index.
+func WriteBatchChunkRanges(db ethdb.KeyValueWriter, batchIndex uint64, chunkBlockRanges []*ChunkBlockRange) {
+	bytes, err := rlp.EncodeToBytes(chunkBlockRanges)
+	if err != nil {
+		log.Crit("failed to RLP encode batch chunk ranges", "batch index", batchIndex, "err", err)
+	}
+	if err := db.Put(batchChunkRangesKey(batchIndex), bytes); err != nil {
+		log.Crit("failed to store batch chunk ranges", "batch index", batchIndex, "err", err)
+	}
+}
+
+// DeleteBatchChunkRanges removes the block ranges of all chunks associated with a specific batch from the database.
+// Note: Only non-finalized batches can be reverted.
+func DeleteBatchChunkRanges(db ethdb.KeyValueWriter, batchIndex uint64) {
+	if err := db.Delete(batchChunkRangesKey(batchIndex)); err != nil {
+		log.Crit("failed to delete batch chunk ranges", "batch index", batchIndex, "err", err)
+	}
+}
+
+// ReadBatchChunkRanges retrieves the block ranges of all chunks associated with a specific batch from the database.
+// It returns a list of ChunkBlockRange pointers, or nil if no chunk ranges are found for the given batch index.
+func ReadBatchChunkRanges(db ethdb.Reader, batchIndex uint64) []*ChunkBlockRange {
+	data, err := db.Get(batchChunkRangesKey(batchIndex))
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("failed to read batch chunk ranges from database", "err", err)
+	}
+
+	cr := new([]*ChunkBlockRange)
+	if err := rlp.Decode(bytes.NewReader(data), cr); err != nil {
+		log.Crit("Invalid ChunkBlockRange RLP", "batch index", batchIndex, "data", data, "err", err)
+	}
+	return *cr
+}
+
+// WriteFinalizedBatchMeta stores the metadata of a finalized batch in the database.
+func WriteFinalizedBatchMeta(db ethdb.KeyValueWriter, batchIndex uint64, finalizedBatchMeta *FinalizedBatchMeta) {
+	var err error
+	bytes, err := rlp.EncodeToBytes(finalizedBatchMeta)
+	if err != nil {
+		log.Crit("failed to RLP encode batch metadata", "batch index", batchIndex, "err", err)
+	}
+	if err := db.Put(batchMetaKey(batchIndex), bytes); err != nil {
+		log.Crit("failed to store batch metadata", "batch index", batchIndex, "err", err)
+	}
+}
+
+// ReadFinalizedBatchMeta fetches the metadata of a finalized batch from the database.
+func ReadFinalizedBatchMeta(db ethdb.Reader, batchIndex uint64) *FinalizedBatchMeta {
+	data, err := db.Get(batchMetaKey(batchIndex))
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("failed to read finalized batch metadata from database", "err", err)
+	}
+
+	fbm := new(FinalizedBatchMeta)
+	if err := rlp.Decode(bytes.NewReader(data), fbm); err != nil {
+		log.Crit("Invalid FinalizedBatchMeta RLP", "batch index", batchIndex, "data", data, "err", err)
+	}
+	return fbm
+}
+
+// WriteFinalizedL2BlockNumber stores the highest finalized L2 block number in the database.
+func WriteFinalizedL2BlockNumber(db ethdb.KeyValueWriter, l2BlockNumber uint64) {
+	value := big.NewInt(0).SetUint64(l2BlockNumber).Bytes()
+	if err := db.Put(finalizedL2BlockNumberKey, value); err != nil {
+		log.Crit("failed to store finalized L2 block number for rollup event", "err", err)
+	}
+}
+
+// ReadFinalizedL2BlockNumber fetches the highest finalized L2 block number from the database.
+func ReadFinalizedL2BlockNumber(db ethdb.Reader) *uint64 {
+	data, err := db.Get(finalizedL2BlockNumberKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("failed to read finalized L2 block number from database", "err", err)
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("unexpected finalized L2 block number in database", "number", number)
+	}
+
+	finalizedL2BlockNumber := number.Uint64()
+	return &finalizedL2BlockNumber
+}

core/rawdb/accessors_rollup_event_test.go

@@ -0,0 +1,186 @@
+package rawdb
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+func TestWriteRollupEventSyncedL1BlockNumber(t *testing.T) {
+	blockNumbers := []uint64{
+		1,
+		1 << 2,
+		1 << 8,
+		1 << 16,
+		1 << 32,
+	}
+
+	db := NewMemoryDatabase()
+
+	// read non-existing value
+	if got := ReadRollupEventSyncedL1BlockNumber(db); got != nil {
+		t.Fatal("Expected 0 for non-existing value", "got", *got)
+	}
+
+	for _, num := range blockNumbers {
+		WriteRollupEventSyncedL1BlockNumber(db, num)
+		got := ReadRollupEventSyncedL1BlockNumber(db)
+
+		if *got != num {
+			t.Fatal("Block number mismatch", "expected", num, "got", got)
+		}
+	}
+}
+
+func TestFinalizedL2BlockNumber(t *testing.T) {
+	blockNumbers := []uint64{
+		1,
+		1 << 2,
+		1 << 8,
+		1 << 16,
+		1 << 32,
+	}
+
+	db := NewMemoryDatabase()
+
+	// read non-existing value
+	if got := ReadFinalizedL2BlockNumber(db); got != nil {
+		t.Fatal("Expected 0 for non-existing value", "got", *got)
+	}
+
+	for _, num := range blockNumbers {
+		WriteFinalizedL2BlockNumber(db, num)
+		got := ReadFinalizedL2BlockNumber(db)
+
+		if *got != num {
+			t.Fatal("Block number mismatch", "expected", num, "got", got)
+		}
+	}
+}
+
+func TestFinalizedBatchMeta(t *testing.T) {
+	batches := []*FinalizedBatchMeta{
+		{
+			BatchHash:            common.BytesToHash([]byte("batch1")),
+			TotalL1MessagePopped: 123,
+			StateRoot:            common.BytesToHash([]byte("stateRoot1")),
+			WithdrawRoot:         common.BytesToHash([]byte("withdrawRoot1")),
+		},
+		{
+			BatchHash:            common.BytesToHash([]byte("batch2")),
+			TotalL1MessagePopped: 456,
+			StateRoot:            common.BytesToHash([]byte("stateRoot2")),
+			WithdrawRoot:         common.BytesToHash([]byte("withdrawRoot2")),
+		},
+		{
+			BatchHash:            common.BytesToHash([]byte("batch3")),
+			TotalL1MessagePopped: 789,
+			StateRoot:            common.BytesToHash([]byte("stateRoot3")),
+			WithdrawRoot:         common.BytesToHash([]byte("withdrawRoot3")),
+		},
+	}
+
+	db := NewMemoryDatabase()
+
+	for i, batch := range batches {
+		batchIndex := uint64(i)
+		WriteFinalizedBatchMeta(db, batchIndex, batch)
+	}
+
+	for i, batch := range batches {
+		batchIndex := uint64(i)
+		readBatch := ReadFinalizedBatchMeta(db, batchIndex)
+		if readBatch == nil {
+			t.Fatal("Failed to read batch from database")
+		}
+		if readBatch.BatchHash != batch.BatchHash || readBatch.TotalL1MessagePopped != batch.TotalL1MessagePopped ||
+			readBatch.StateRoot != batch.StateRoot || readBatch.WithdrawRoot != batch.WithdrawRoot {
+			t.Fatal("Mismatch in read batch", "expected", batch, "got", readBatch)
+		}
+	}
+
+	// over-write
+	newBatch := &FinalizedBatchMeta{
+		BatchHash:            common.BytesToHash([]byte("newBatch")),
+		TotalL1MessagePopped: 999,
+		StateRoot:            common.BytesToHash([]byte("newStateRoot")),
+		WithdrawRoot:         common.BytesToHash([]byte("newWithdrawRoot")),
+	}
+	WriteFinalizedBatchMeta(db, 0, newBatch) // over-writing the batch with index 0
+	readBatch := ReadFinalizedBatchMeta(db, 0)
+	if readBatch.BatchHash != newBatch.BatchHash || readBatch.TotalL1MessagePopped != newBatch.TotalL1MessagePopped ||
+		readBatch.StateRoot != newBatch.StateRoot || readBatch.WithdrawRoot != newBatch.WithdrawRoot {
+		t.Fatal("Mismatch after over-writing batch", "expected", newBatch, "got", readBatch)
+	}
+
+	// read non-existing value
+	nonExistingIndex := uint64(len(batches) + 1)
+	readBatch = ReadFinalizedBatchMeta(db, nonExistingIndex)
+	if readBatch != nil {
+		t.Fatal("Expected nil for non-existing value", "got", readBatch)
+	}
+}
+
+func TestBatchChunkRanges(t *testing.T) {
+	chunks := [][]*ChunkBlockRange{
+		{
+			{StartBlockNumber: 1, EndBlockNumber: 100},
+			{StartBlockNumber: 101, EndBlockNumber: 200},
+		},
+		{
+			{StartBlockNumber: 201, EndBlockNumber: 300},
+			{StartBlockNumber: 301, EndBlockNumber: 400},
+		},
+		{
+			{StartBlockNumber: 401, EndBlockNumber: 500},
+		},
+	}
+
+	db := NewMemoryDatabase()
+
+	for i, chunkRange := range chunks {
+		batchIndex := uint64(i)
+		WriteBatchChunkRanges(db, batchIndex, chunkRange)
+	}
+
+	for i, chunkRange := range chunks {
+		batchIndex := uint64(i)
+		readChunkRange := ReadBatchChunkRanges(db, batchIndex)
+		if len(readChunkRange) != len(chunkRange) {
+			t.Fatal("Mismatch in number of chunk ranges", "expected", len(chunkRange), "got", len(readChunkRange))
+		}
+
+		for j, cr := range readChunkRange {
+			if cr.StartBlockNumber != chunkRange[j].StartBlockNumber || cr.EndBlockNumber != chunkRange[j].EndBlockNumber {
+				t.Fatal("Mismatch in chunk range", "batch index", batchIndex, "expected", chunkRange[j], "got", cr)
+			}
+		}
+	}
+
+	// over-write
+	newRange := []*ChunkBlockRange{{StartBlockNumber: 1001, EndBlockNumber: 1100}}
+	WriteBatchChunkRanges(db, 0, newRange)
+	readChunkRange := ReadBatchChunkRanges(db, 0)
+	if len(readChunkRange) != 1 || readChunkRange[0].StartBlockNumber != 1001 || readChunkRange[0].EndBlockNumber != 1100 {
+		t.Fatal("Over-write failed for chunk range", "expected", newRange, "got", readChunkRange)
+	}
+
+	// read non-existing value
+	if readChunkRange = ReadBatchChunkRanges(db, uint64(len(chunks)+1)); readChunkRange != nil {
+		t.Fatal("Expected nil for non-existing value", "got", readChunkRange)
+	}
+
+	// delete: revert batch
+	for i := range chunks {
+		batchIndex := uint64(i)
+		DeleteBatchChunkRanges(db, batchIndex)
+
+		readChunkRange := ReadBatchChunkRanges(db, batchIndex)
+		if readChunkRange != nil {
+			t.Fatal("Chunk range was not deleted", "batch index", batchIndex)
+		}
+	}
+
+	// delete non-existing value: ensure the delete operation handles non-existing values without errors.
+	DeleteBatchChunkRanges(db, uint64(len(chunks)+1))
+}

core/rawdb/schema.go

@@ -145,6 +145,12 @@ var (
 	l1MessagePrefix                   = []byte("L1") // l1MessagePrefix + queueIndex (uint64 big endian) -> L1MessageTx
 	firstQueueIndexNotInL2BlockPrefix = []byte("q")  // firstQueueIndexNotInL2BlockPrefix + L2 block hash -> enqueue index
 	highestSyncedQueueIndexKey        = []byte("HighestSyncedQueueIndex")
+
+	// Scroll rollup event store
+	rollupEventSyncedL1BlockNumberKey = []byte("R-LastRollupEventSyncedL1BlockNumber")
+	batchChunkRangesPrefix            = []byte("R-bcr")
+	batchMetaPrefix                   = []byte("R-bm")
+	finalizedL2BlockNumberKey         = []byte("R-finalized")
 )
 
 // Use the updated "L1" prefix on all new networks
@@ -372,3 +378,13 @@ func L1MessageKey(queueIndex uint64) []byte {
 func FirstQueueIndexNotInL2BlockKey(l2BlockHash common.Hash) []byte {
 	return append(firstQueueIndexNotInL2BlockPrefix, l2BlockHash.Bytes()...)
 }
+
+// batchChunkRangesKey = batchChunkRangesPrefix + batch index (uint64 big endian)
+func batchChunkRangesKey(batchIndex uint64) []byte {
+	return append(batchChunkRangesPrefix, encodeBigEndian(batchIndex)...)
+}
+
+// batchMetaKey = batchMetaPrefix + batch index (uint64 big endian)
+func batchMetaKey(batchIndex uint64) []byte {
+	return append(batchMetaPrefix, encodeBigEndian(batchIndex)...)
+}

eth/backend.go

@@ -57,6 +57,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/rollup/rollup_sync_service"
 	"github.com/ethereum/go-ethereum/rollup/sync_service"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -72,6 +73,7 @@ type Ethereum struct {
 	// Handlers
 	txPool            *txpool.TxPool
 	syncService       *sync_service.SyncService
+	rollupSyncService *rollup_sync_service.RollupSyncService
 
 	blockchain         *core.BlockChain
 	handler            *handler
@@ -239,6 +241,14 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
 		return nil, fmt.Errorf("cannot initialize L1 sync service: %w", err)
 	}
 	eth.syncService.Start()
+	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.Config().L1DeploymentBlock)
+		if err != nil {
+			return nil, fmt.Errorf("cannot initialize rollup event sync service: %w", err)
+		}
+		eth.rollupSyncService.Start()
+	}
 
 	// Permit the downloader to use the trie cache allowance during fast sync
 	cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit
@@ -547,6 +557,9 @@ func (s *Ethereum) Stop() error {
 	close(s.closeBloomHandler)
 	s.txPool.Close()
 	s.syncService.Stop()
+	if s.config.EnableRollupVerify {
+		s.rollupSyncService.Stop()
+	}
 	s.miner.Close()
 	s.blockchain.Stop()
 	s.engine.Close()

eth/ethconfig/config.go

@@ -170,6 +170,9 @@ type Config struct {
 	// OverrideVerkle (TODO: remove after the fork)
 	OverrideVerkle *uint64 `toml:",omitempty"`
 
+	// Enable verification of batch consistency between L1 and L2 in rollup
+	EnableRollupVerify bool
+
 	// Max block range for eth_getLogs api method
 	MaxBlockRange int64
 }

eth/ethconfig/gen_config.go

@@ -56,6 +56,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
 		RPCTxFeeCap             float64
 		OverrideCancun          *uint64 `toml:",omitempty"`
 		OverrideVerkle          *uint64 `toml:",omitempty"`
+		EnableRollupVerify      bool
 		MaxBlockRange           int64
 	}
 	var enc Config
@@ -98,6 +99,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
 	enc.RPCTxFeeCap = c.RPCTxFeeCap
 	enc.OverrideCancun = c.OverrideCancun
 	enc.OverrideVerkle = c.OverrideVerkle
+	enc.EnableRollupVerify = c.EnableRollupVerify
 	enc.MaxBlockRange = c.MaxBlockRange
 	return &enc, nil
 }
@@ -144,6 +146,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
 		RPCTxFeeCap             *float64
 		OverrideCancun          *uint64 `toml:",omitempty"`
 		OverrideVerkle          *uint64 `toml:",omitempty"`
+		EnableRollupVerify      *bool
 		MaxBlockRange           *int64
 	}
 	var dec Config
@@ -267,6 +270,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
 	if dec.OverrideVerkle != nil {
 		c.OverrideVerkle = dec.OverrideVerkle
 	}
+	if dec.EnableRollupVerify != nil {
+		c.EnableRollupVerify = *dec.EnableRollupVerify
+	}
 	if dec.MaxBlockRange != nil {
 		c.MaxBlockRange = *dec.MaxBlockRange
 	}

rollup/rollup_sync_service/rollup_sync_service.go

@@ -260,7 +260,7 @@ func (s *RollupSyncService) getLocalInfoForBatch(batchIndex uint64) (*rawdb.Fina
 			return nil, nil, s.ctx.Err()
 		}
 
-		localSyncedBlockHeight := s.bc.CurrentBlock().Number().Uint64()
+		localSyncedBlockHeight := s.bc.CurrentBlock().Number.Uint64()
 		if localSyncedBlockHeight >= endBlockNumber {
 			break // ready to proceed, exit retry loop
 		}
@@ -270,7 +270,7 @@ func (s *RollupSyncService) getLocalInfoForBatch(batchIndex uint64) (*rawdb.Fina
 		time.Sleep(defaultGetBlockInRangeRetryDelay)
 	}
 
-	localSyncedBlockHeight := s.bc.CurrentBlock().Number().Uint64()
+	localSyncedBlockHeight := s.bc.CurrentBlock().Number.Uint64()
 	if localSyncedBlockHeight < endBlockNumber {
 		return nil, nil, fmt.Errorf("local node is not synced up to the required block height: %v, local synced block height: %v", endBlockNumber, localSyncedBlockHeight)
 	}