From 44fed2423779edd85460c1e038317b9a13250a95 Mon Sep 17 00:00:00 2001 From: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com> Date: Thu, 21 Dec 2023 16:17:47 +0800 Subject: [PATCH] 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 | 1 + cmd/utils/flags.go | 13 ++ core/rawdb/accessors_rollup_event.go | 146 ++++++++++++++ core/rawdb/accessors_rollup_event_test.go | 186 ++++++++++++++++++ core/rawdb/schema.go | 16 ++ eth/backend.go | 17 +- eth/ethconfig/config.go | 3 + eth/ethconfig/gen_config.go | 6 + .../rollup_sync_service.go | 4 +- 9 files changed, 388 insertions(+), 4 deletions(-) create mode 100644 core/rawdb/accessors_rollup_event.go create mode 100644 core/rawdb/accessors_rollup_event_test.go diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 3e57e59a1b..180d7bc0ea 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -147,6 +147,7 @@ var ( utils.L1EndpointFlag, utils.L1ConfirmationsFlag, utils.L1DeploymentBlockFlag, + utils.RollupVerifyEnabledFlag, }, utils.NetworkFlags, utils.DatabaseFlags) rpcFlags = []cli.Flag{ diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index b47b83ec00..c4121e5894 100644 --- a/cmd/utils/flags.go +++ b/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 diff --git a/core/rawdb/accessors_rollup_event.go b/core/rawdb/accessors_rollup_event.go new file mode 100644 index 0000000000..0ec9b16d07 --- /dev/null +++ b/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 +} diff --git a/core/rawdb/accessors_rollup_event_test.go b/core/rawdb/accessors_rollup_event_test.go new file mode 100644 index 0000000000..939e28468b --- /dev/null +++ b/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)) +} diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index c644e5043a..078d830006 100644 --- a/core/rawdb/schema.go +++ b/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)...) +} diff --git a/eth/backend.go b/eth/backend.go index 4a757d4bdd..c8eea074c7 100644 --- a/eth/backend.go +++ b/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" ) @@ -70,8 +71,9 @@ type Ethereum struct { config *ethconfig.Config // Handlers - txPool *txpool.TxPool - syncService *sync_service.SyncService + 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() diff --git a/eth/ethconfig/config.go b/eth/ethconfig/config.go index 76aa27ccb3..3ec771f218 100644 --- a/eth/ethconfig/config.go +++ b/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 } diff --git a/eth/ethconfig/gen_config.go b/eth/ethconfig/gen_config.go index d8f4694927..aed74f2db9 100644 --- a/eth/ethconfig/gen_config.go +++ b/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 } diff --git a/rollup/rollup_sync_service/rollup_sync_service.go b/rollup/rollup_sync_service/rollup_sync_service.go index 5f46c23895..94fa90383b 100644 --- a/rollup/rollup_sync_service/rollup_sync_service.go +++ b/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) }