feat(MessageQueueV2): add reading of MessageQueueV2 and transition logic from V1 to V2 for EuclidV2 (#1108)

* 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

* implement first version of new da-codec and to handle multiple batches submitted in one transaction

* add CommitBatchDAV7 and handle multiple commit events submitted in a single transactions

* fix bug due to previous batch being empty when processing the first batch within a set of batches

* Allow using MPT

* implement reading of QueueTransaction from  L1MessageQueueV1 and V2

* implement access to V1 and V2 messages and replace usage so that V1 is used before EuclidV2 fork and V2 afterward

* add tests

* update to latest da-codec

* add field to CommittedBatchMeta to store LastL1MessageQueueHash for CodecV7 batches

* adjust rollup verifier to support CodecV7 batches

* address review comments

* consume all L1 messages before EuclidV2 fork

* address review comments

* address review comments

* fix issues after merge

* go mod tidy

* fix unit tests

* update da-codec

* add test TestValidateBatchCodecV7

* go mod tidy

* do not log error on shutdown

* add flag to explicitly disable L1MessageQueueV2

* add TestEuclidV2HardForkMessageQueue to scroll worker

* add sanity check for version to deserialization of committedBatchMetaV7

* add mechanism for nodes to retrieve V2 messages even if they upgrade after L1MessageQueueV2 is deployed

* chore: auto version bump [bot]

* address review comments

* Update core/rawdb/accessors_l1_message.go

Co-authored-by: colin <[email protected]>

* address review comments

---------

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

Author: 4 people

cmd/geth/main.go

@@ -169,6 +169,7 @@ var (
 		utils.L1EndpointFlag,
 		utils.L1ConfirmationsFlag,
 		utils.L1DeploymentBlockFlag,
+		utils.L1DisableMessageQueueV2Flag,
 		utils.CircuitCapacityCheckEnabledFlag,
 		utils.CircuitCapacityCheckWorkersFlag,
 		utils.RollupVerifyEnabledFlag,

cmd/utils/flags.go

@@ -854,6 +854,10 @@ var (
 		Name:  "l1.sync.startblock",
 		Usage: "L1 block height to start syncing from. Should be set to the L1 message queue deployment block number.",
 	}
+	L1DisableMessageQueueV2Flag = &cli.BoolFlag{
+		Name:  "l1.disablemqv2",
+		Usage: "Disable L1 message queue v2",
+	}
 
 	// Circuit capacity check settings
 	CircuitCapacityCheckEnabledFlag = cli.BoolFlag{
@@ -1408,6 +1412,9 @@ func setL1(ctx *cli.Context, cfg *node.Config) {
 	if ctx.GlobalIsSet(L1DeploymentBlockFlag.Name) {
 		cfg.L1DeploymentBlock = ctx.GlobalUint64(L1DeploymentBlockFlag.Name)
 	}
+	if ctx.GlobalIsSet(L1DisableMessageQueueV2Flag.Name) {
+		cfg.L1DisableMessageQueueV2 = ctx.GlobalBool(L1DisableMessageQueueV2Flag.Name)
+	}
 }
 
 func setSmartCard(ctx *cli.Context, cfg *node.Config) {

core/block_validator.go

@@ -139,8 +139,52 @@ func (v *BlockValidator) ValidateL1Messages(block *types.Block) error {
 	queueIndex := *nextQueueIndex
 
 	L1SectionOver := false
-	it := rawdb.IterateL1MessagesFrom(v.bc.db, queueIndex)
 
+	// From EuclidV2 onwards there can't be any skipped L1 messages, and we use a different L1MessageQueueV2.
+	if v.config.IsEuclidV2(block.Time()) {
+		it := rawdb.IterateL1MessagesV2From(v.bc.db, queueIndex)
+		for _, tx := range block.Transactions() {
+			if !tx.IsL1MessageTx() {
+				L1SectionOver = true
+				continue // we do not verify L2 transactions here
+			}
+
+			// check that L1 messages are before L2 transactions
+			if L1SectionOver {
+				return consensus.ErrInvalidL1MessageOrder
+			}
+
+			// queue index must be equal to the expected value
+			txQueueIndex := tx.AsL1MessageTx().QueueIndex
+			if txQueueIndex != queueIndex {
+				return consensus.ErrInvalidL1MessageOrder
+			}
+
+			if exists := it.Next(); !exists {
+				if err := it.Error(); err != nil {
+					log.Error("Unexpected DB error in ValidateL1Messages", "err", err, "queueIndex", txQueueIndex)
+				}
+				// the message in this block is not available in our local db.
+				// we'll reprocess this block at a later time.
+				return consensus.ErrMissingL1MessageData
+			}
+
+			// check that the L1 message in the block is the same that we collected from L1
+			msg := it.L1Message()
+			expectedHash := types.NewTx(&msg).Hash()
+
+			if tx.Hash() != expectedHash {
+				return consensus.ErrUnknownL1Message
+			}
+
+			// we expect L1 messages to be in order and contiguous
+			queueIndex++
+		}
+
+		return nil
+	}
+
+	it := rawdb.IterateL1MessagesV1From(v.bc.db, queueIndex)
 	for _, tx := range block.Transactions() {
 		if !tx.IsL1MessageTx() {
 			L1SectionOver = true

core/blockchain.go

@@ -1853,6 +1853,18 @@ func (bc *BlockChain) BuildAndWriteBlock(parentBlock *types.Block, header *types
 		l.BlockHash = blockHash
 	}
 
+	// Make sure the block body is valid e.g. ordering of L1 messages is correct and continuous.
+	if err = bc.validator.ValidateBody(fullBlock); err != nil {
+		bc.reportBlock(fullBlock, receipts, err)
+		return NonStatTy, fmt.Errorf("error validating block body %d: %w", fullBlock.Number().Uint64(), err)
+	}
+
+	// Double check: even though we just built the block, make sure it is valid.
+	if err = bc.validator.ValidateState(fullBlock, statedb, receipts, gasUsed); err != nil {
+		bc.reportBlock(fullBlock, receipts, err)
+		return NonStatTy, fmt.Errorf("error validating block %d: %w", fullBlock.Number().Uint64(), err)
+	}
+
 	return bc.writeBlockWithState(fullBlock, receipts, logs, statedb, false)
 }
 

core/rawdb/accessors_l1_message.go

@@ -141,9 +141,9 @@ type L1MessageIterator struct {
 	maxQueueIndex uint64
 }
 
-// IterateL1MessagesFrom creates an L1MessageIterator that iterates over
+// iterateL1MessagesFrom creates an L1MessageIterator that iterates over
 // all L1 message in the database starting at the provided enqueue index.
-func IterateL1MessagesFrom(db ethdb.Database, fromQueueIndex uint64) L1MessageIterator {
+func iterateL1MessagesFrom(db ethdb.Database, fromQueueIndex uint64) L1MessageIterator {
 	start := encodeBigEndian(fromQueueIndex)
 	it := db.NewIterator(l1MessagePrefix, start)
 	keyLength := len(l1MessagePrefix) + 8
@@ -208,10 +208,72 @@ func (it *L1MessageIterator) Error() error {
 	return it.inner.Error()
 }
 
-// ReadL1MessagesFrom retrieves up to `maxCount` L1 messages starting at `startIndex`.
-func ReadL1MessagesFrom(db ethdb.Database, startIndex, maxCount uint64) []types.L1MessageTx {
+// L1MessageV1Iterator is a wrapper around L1MessageIterator that allows us to iterate over L1 messages V1.
+type L1MessageV1Iterator struct {
+	db           ethdb.Database
+	v2StartIndex *uint64
+	L1MessageIterator
+}
+
+// IterateL1MessagesV1From yields a L1MessageV1Iterator with following behavior:
+// - If fromQueueIndex >= L1MessageV2StartIndex: yield 0 messages.
+// - Otherwise, simply yield all messages (guaranteed to be V1) starting from `fromQueueIndex` until `L1MessageV2StartIndex`.
+func IterateL1MessagesV1From(db ethdb.Database, fromQueueIndex uint64) L1MessageV1Iterator {
+	return L1MessageV1Iterator{
+		db:                db,
+		v2StartIndex:      ReadL1MessageV2StartIndex(db),
+		L1MessageIterator: iterateL1MessagesFrom(db, fromQueueIndex),
+	}
+}
+
+func (it *L1MessageV1Iterator) Next() bool {
+	for it.L1MessageIterator.Next() {
+		// L1MessageV2StartIndex is the first queue index of L1 messages that are from L1MessageQueueV2.
+		// Therefore, we stop reading L1 messages V1 when we reach this index.
+		// We need to check in every iteration if not yet set as the start index can be set in the meantime when we are reading L1 messages.
+		if it.v2StartIndex == nil {
+			it.v2StartIndex = ReadL1MessageV2StartIndex(it.db)
+		}
+
+		if it.v2StartIndex != nil && it.QueueIndex() >= *it.v2StartIndex {
+			return false
+		}
+		return true
+	}
+	return false
+}
+
+// L1MessageV2Iterator is a wrapper around L1MessageIterator that allows us to iterate over L1 messages V2.
+type L1MessageV2Iterator struct {
+	v2StartIndex *uint64
+	L1MessageIterator
+}
+
+// IterateL1MessagesV2From yields a L1MessageV2Iterator with following behavior:
+// - If fromQueueIndex < v2StartIndex: yield 0 messages.
+// - Otherwise, simply yield all messages (guaranteed to be v2) starting from `fromQueueIndex`.
+func IterateL1MessagesV2From(db ethdb.Database, fromQueueIndex uint64) L1MessageV2Iterator {
+	v2StartIndex := ReadL1MessageV2StartIndex(db)
+
+	return L1MessageV2Iterator{
+		v2StartIndex:      v2StartIndex,
+		L1MessageIterator: iterateL1MessagesFrom(db, fromQueueIndex),
+	}
+}
+
+func (it *L1MessageV2Iterator) Next() bool {
+	if it.v2StartIndex == nil {
+		return false
+	}
+
+	return it.L1MessageIterator.Next() && it.QueueIndex() >= *it.v2StartIndex
+}
+
+// ReadL1MessagesV1From retrieves up to `maxCount` L1 messages V1 starting at `startIndex`.
+// If startIndex is >= L1MessageV2StartIndex, this function returns an empty slice.
+func ReadL1MessagesV1From(db ethdb.Database, startIndex, maxCount uint64) []types.L1MessageTx {
 	msgs := make([]types.L1MessageTx, 0, maxCount)
-	it := IterateL1MessagesFrom(db, startIndex)
+	it := IterateL1MessagesV1From(db, startIndex)
 	defer it.Release()
 
 	index := startIndex
@@ -223,7 +285,50 @@ func ReadL1MessagesFrom(db ethdb.Database, startIndex, maxCount uint64) []types.
 		// sanity check
 		if msg.QueueIndex != index {
 			log.Crit(
-				"Unexpected QueueIndex in ReadL1MessagesFrom",
+				"Unexpected QueueIndex in ReadL1MessagesV1From",
+				"expected", index,
+				"got", msg.QueueIndex,
+				"startIndex", startIndex,
+				"maxCount", maxCount,
+			)
+		}
+
+		msgs = append(msgs, msg)
+		index += 1
+		count -= 1
+
+		iteratorL1MessageSizeGauge.Update(int64(unsafe.Sizeof(msg) + uintptr(cap(msg.Data))))
+
+		if msg.QueueIndex == it.maxQueueIndex {
+			break
+		}
+	}
+
+	if err := it.Error(); err != nil {
+		log.Crit("Failed to read L1 messages", "err", err)
+	}
+
+	return msgs
+}
+
+// ReadL1MessagesV2From retrieves up to `maxCount` L1 messages V2 starting at `startIndex`.
+// If startIndex is smaller than L1MessageV2StartIndex, this function returns an empty slice.
+func ReadL1MessagesV2From(db ethdb.Database, startIndex, maxCount uint64) []types.L1MessageTx {
+	msgs := make([]types.L1MessageTx, 0, maxCount)
+
+	it := IterateL1MessagesV2From(db, startIndex)
+	defer it.Release()
+
+	index := startIndex
+	count := maxCount
+
+	for count > 0 && it.Next() {
+		msg := it.L1Message()
+
+		// sanity check
+		if msg.QueueIndex != index {
+			log.Crit(
+				"Unexpected QueueIndex in ReadL1MessagesV2From",
 				"expected", index,
 				"got", msg.QueueIndex,
 				"startIndex", startIndex,
@@ -275,3 +380,65 @@ func ReadFirstQueueIndexNotInL2Block(db ethdb.Reader, l2BlockHash common.Hash) *
 	queueIndex := binary.BigEndian.Uint64(data)
 	return &queueIndex
 }
+
+// WriteL1MessageV2StartIndex writes the start index of L1 messages that are from L1MessageQueueV2.
+func WriteL1MessageV2StartIndex(db ethdb.KeyValueWriter, queueIndex uint64) {
+	value := big.NewInt(0).SetUint64(queueIndex).Bytes()
+
+	if err := db.Put(l1MessageV2StartIndexKey, value); err != nil {
+		log.Crit("Failed to update L1MessageV2 start index", "err", err)
+	}
+}
+
+// ReadL1MessageV2StartIndex retrieves the start index of L1 messages that are from L1MessageQueueV2.
+func ReadL1MessageV2StartIndex(db ethdb.Reader) *uint64 {
+	data, err := db.Get(l1MessageV2StartIndexKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("Failed to read L1MessageV2 start index from database", "err", err)
+	}
+	if len(data) == 0 {
+		return nil
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("Unexpected number for L1MessageV2 start index", "number", number)
+	}
+
+	res := number.Uint64()
+	return &res
+}
+
+// WriteL1MessageV2FirstL1BlockNumber writes the first synced L1 block number for L1MessageV2.
+func WriteL1MessageV2FirstL1BlockNumber(db ethdb.KeyValueWriter, l1BlockNumber uint64) {
+	value := big.NewInt(0).SetUint64(l1BlockNumber).Bytes()
+
+	if err := db.Put(l1MessageV2FirstL1BlockNumberKey, value); err != nil {
+		log.Crit("Failed to update L1MessageV2 start index", "err", err)
+	}
+}
+
+// ReadL1MessageV2FirstL1BlockNumber retrieves the first synced L1 block number for L1MessageV2.
+func ReadL1MessageV2FirstL1BlockNumber(db ethdb.Reader) *uint64 {
+	data, err := db.Get(l1MessageV2FirstL1BlockNumberKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("Failed to read L1MessageV2 first L1 block number from database", "err", err)
+	}
+	if len(data) == 0 {
+		return nil
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("Unexpected number for L1MessageV2 first L1 block number", "number", number)
+	}
+
+	res := number.Uint64()
+	return &res
+}