feat(L1 follower): adjust to recent CodecV7 and contract changes

core/rawdb/accessors_da_syncer.go

@@ -1,23 +1,33 @@
 package rawdb
 
 import (
+	"bytes"
 	"math/big"
 
 	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rlp"
 )
 
-// WriteDASyncedL1BlockNumber writes the highest synced L1 block number to the database.
-func WriteDASyncedL1BlockNumber(db ethdb.KeyValueWriter, L1BlockNumber uint64) {
-	value := big.NewInt(0).SetUint64(L1BlockNumber).Bytes()
+type DAProcessedBatchMeta struct {
+	BatchIndex            uint64
+	L1BlockNumber         uint64
+	TotalL1MessagesPopped uint64
+}
 
+// WriteDAProcessedBatchMeta writes the batch metadata of the latest processed DA batch.
+func WriteDAProcessedBatchMeta(db ethdb.KeyValueWriter, daProcessedBatchMeta *DAProcessedBatchMeta) {
+	value, err := rlp.EncodeToBytes(daProcessedBatchMeta)
+	if err != nil {
+		log.Crit("failed to RLP encode committed batch metadata", "batch index", daProcessedBatchMeta.BatchIndex, "committed batch meta", daProcessedBatchMeta, "err", err)
+	}
 	if err := db.Put(daSyncedL1BlockNumberKey, value); err != nil {
-		log.Crit("Failed to update DA synced L1 block number", "err", err)
+		log.Crit("Failed to update DAProcessedBatchMeta", "err", err)
 	}
 }
 
-// ReadDASyncedL1BlockNumber retrieves the highest synced L1 block number.
-func ReadDASyncedL1BlockNumber(db ethdb.Reader) *uint64 {
+// ReadDAProcessedBatchMeta retrieves the batch metadata of the latest processed DA batch.
+func ReadDAProcessedBatchMeta(db ethdb.Reader) *DAProcessedBatchMeta {
 	data, err := db.Get(daSyncedL1BlockNumberKey)
 	if err != nil && isNotFoundErr(err) {
 		return nil
@@ -29,11 +39,25 @@ func ReadDASyncedL1BlockNumber(db ethdb.Reader) *uint64 {
 		return nil
 	}
 
-	number := new(big.Int).SetBytes(data)
-	if !number.IsUint64() {
-		log.Crit("Unexpected DA synced L1 block number in database", "number", number)
+	// Try decoding from the newest format for future proofness, then the older one for old data.
+	daProcessedBatchMeta := new(DAProcessedBatchMeta)
+	if err = rlp.Decode(bytes.NewReader(data), daProcessedBatchMeta); err == nil {
+		return daProcessedBatchMeta
 	}
 
-	value := number.Uint64()
-	return &value
+	// Before storing DAProcessedBatchMeta we used to store a single uint64 value for the L1 block number.
+	l1BlockNumber := new(big.Int).SetBytes(data)
+	if !l1BlockNumber.IsUint64() {
+		log.Crit("Unexpected DA synced L1 block number in database", "number", l1BlockNumber)
+	}
+
+	// We can simply set only the L1BlockNumber because carrying forward the totalL1MessagesPopped is not required before EuclidV2 (CodecV7)
+	// (the parentTotalL1MessagePopped is given via the parentBatchHeader).
+	// Nodes need to update to the new version to be able to continue syncing after EuclidV2 (CodecV7). Therefore,
+	// the only nodes that might read a uint64 value are nodes that were running L1 follower before the EuclidV2.
+	return &DAProcessedBatchMeta{
+		BatchIndex:            0,
+		L1BlockNumber:         l1BlockNumber.Uint64(),
+		TotalL1MessagesPopped: 0,
+	}
 }

core/rawdb/accessors_l1_message.go

@@ -383,9 +383,10 @@ func ReadFirstQueueIndexNotInL2Block(db ethdb.Reader, l2BlockHash common.Hash) *
 
 // 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()
+	var buf [8]byte
+	binary.BigEndian.PutUint64(buf[:], queueIndex)
 
-	if err := db.Put(l1MessageV2StartIndexKey, value); err != nil {
+	if err := db.Put(l1MessageV2StartIndexKey, buf[:]); err != nil {
 		log.Crit("Failed to update L1MessageV2 start index", "err", err)
 	}
 }
@@ -402,13 +403,11 @@ func ReadL1MessageV2StartIndex(db ethdb.Reader) *uint64 {
 	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)
+	if len(data) != 8 {
+		return nil
 	}
+	res := binary.BigEndian.Uint64(data)
 
-	res := number.Uint64()
 	return &res
 }
 

core/rawdb/accessors_rollup_event.go

@@ -25,7 +25,7 @@ type CommittedBatchMeta struct {
 	ChunkBlockRanges []*ChunkBlockRange
 
 	// introduced with CodecV7
-	LastL1MessageQueueHash common.Hash
+	PostL1MessageQueueHash common.Hash
 }
 
 type committedBatchMetaV0 struct {
@@ -170,7 +170,7 @@ func WriteCommittedBatchMeta(db ethdb.KeyValueWriter, batchIndex uint64, committ
 		committedBatchMetaToStore = &committedBatchMetaV7{
 			Version:                committedBatchMeta.Version,
 			ChunkBlockRanges:       committedBatchMeta.ChunkBlockRanges,
-			LastL1MessageQueueHash: committedBatchMeta.LastL1MessageQueueHash,
+			LastL1MessageQueueHash: committedBatchMeta.PostL1MessageQueueHash,
 		}
 	}
 
@@ -202,7 +202,7 @@ func ReadCommittedBatchMeta(db ethdb.Reader, batchIndex uint64) (*CommittedBatch
 		return &CommittedBatchMeta{
 			Version:                cbm7.Version,
 			ChunkBlockRanges:       cbm7.ChunkBlockRanges,
-			LastL1MessageQueueHash: cbm7.LastL1MessageQueueHash,
+			PostL1MessageQueueHash: cbm7.LastL1MessageQueueHash,
 		}, nil
 	}
 
@@ -214,7 +214,7 @@ func ReadCommittedBatchMeta(db ethdb.Reader, batchIndex uint64) (*CommittedBatch
 	return &CommittedBatchMeta{
 		Version:                cbm0.Version,
 		ChunkBlockRanges:       cbm0.ChunkBlockRanges,
-		LastL1MessageQueueHash: common.Hash{},
+		PostL1MessageQueueHash: common.Hash{},
 	}, nil
 }
 

core/rawdb/accessors_rollup_event_test.go

@@ -182,15 +182,15 @@ func TestWriteReadDeleteCommittedBatchMeta(t *testing.T) {
 			meta: &CommittedBatchMeta{
 				Version:                7,
 				ChunkBlockRanges:       []*ChunkBlockRange{{StartBlockNumber: 0, EndBlockNumber: 10}},
-				LastL1MessageQueueHash: common.Hash{1, 2, 3, 4, 5, 6, 7},
+				PostL1MessageQueueHash: common.Hash{1, 2, 3, 4, 5, 6, 7},
 			},
 		},
 		{
 			batchIndex: 255,
 			meta: &CommittedBatchMeta{
 				Version:                255,
 				ChunkBlockRanges:       []*ChunkBlockRange{{StartBlockNumber: 0, EndBlockNumber: 10}, {StartBlockNumber: 11, EndBlockNumber: 20}},
-				LastL1MessageQueueHash: common.Hash{255},
+				PostL1MessageQueueHash: common.Hash{255},
 			},
 		},
 	}
@@ -237,7 +237,7 @@ func TestOverwriteCommittedBatchMeta(t *testing.T) {
 	newMeta := &CommittedBatchMeta{
 		Version:                255,
 		ChunkBlockRanges:       []*ChunkBlockRange{{StartBlockNumber: 0, EndBlockNumber: 20}, {StartBlockNumber: 21, EndBlockNumber: 30}},
-		LastL1MessageQueueHash: common.Hash{255},
+		PostL1MessageQueueHash: common.Hash{255},
 	}
 
 	// write initial meta
@@ -282,5 +282,5 @@ func compareCommittedBatchMeta(a, b *CommittedBatchMeta) bool {
 		}
 	}
 
-	return a.LastL1MessageQueueHash == b.LastL1MessageQueueHash
+	return a.PostL1MessageQueueHash == b.PostL1MessageQueueHash
 }

go.mod

@@ -51,7 +51,7 @@ require (
 	github.com/prometheus/tsdb v0.7.1
 	github.com/rjeczalik/notify v0.9.1
 	github.com/rs/cors v1.7.0
-	github.com/scroll-tech/da-codec v0.1.3-0.20250210041951-d028c537b995
+	github.com/scroll-tech/da-codec v0.1.3-0.20250226072559-f8a8d3898f54
 	github.com/scroll-tech/zktrie v0.8.4
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/sourcegraph/conc v0.3.0

go.sum

@@ -396,8 +396,8 @@ github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncj
 github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/scroll-tech/da-codec v0.1.3-0.20250210041951-d028c537b995 h1:Zo1p42CUS9pADSKoDD0ZoDxf4dQ3gttqWZlV+RSeImk=
-github.com/scroll-tech/da-codec v0.1.3-0.20250210041951-d028c537b995/go.mod h1:UZhhjzqYsyEhcvY0Y+SP+oMdeOUqFn/UXpbAYuPGzg0=
+github.com/scroll-tech/da-codec v0.1.3-0.20250226072559-f8a8d3898f54 h1:qVpsVu1J91opTn6HYeuzWcBRVhQmPR8g05i+PlOjlI4=
+github.com/scroll-tech/da-codec v0.1.3-0.20250226072559-f8a8d3898f54/go.mod h1:xECEHZLVzbdUn+tNbRJhRIjLGTOTmnFQuTgUTeVLX58=
 github.com/scroll-tech/zktrie v0.8.4 h1:UagmnZ4Z3ITCk+aUq9NQZJNAwnWl4gSxsLb2Nl7IgRE=
 github.com/scroll-tech/zktrie v0.8.4/go.mod h1:XvNo7vAk8yxNyTjBDj5WIiFzYW4bx/gJ78+NK6Zn6Uk=
 github.com/segmentio/kafka-go v0.1.0/go.mod h1:X6itGqS9L4jDletMsxZ7Dz+JFWxM6JHfPOCvTvk+EJo=

params/version.go

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

rollup/da_syncer/batch_queue.go

@@ -7,7 +7,9 @@ import (
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/ethdb"
+	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/da"
+	"github.com/scroll-tech/go-ethereum/rollup/l1"
 )
 
 // BatchQueue is a pipeline stage that reads all batch events from DAQueue and provides only finalized batches to the next stage.
@@ -17,21 +19,24 @@ type BatchQueue struct {
 	lastFinalizedBatchIndex uint64
 	batches                 *common.Heap[da.Entry]
 	batchesMap              *common.ShrinkingMap[uint64, *common.HeapElement[da.Entry]]
+
+	previousBatch *rawdb.DAProcessedBatchMeta
 }
 
-func NewBatchQueue(DAQueue *DAQueue, db ethdb.Database) *BatchQueue {
+func NewBatchQueue(DAQueue *DAQueue, db ethdb.Database, lastProcessedBatch *rawdb.DAProcessedBatchMeta) *BatchQueue {
 	return &BatchQueue{
 		DAQueue:                 DAQueue,
 		db:                      db,
-		lastFinalizedBatchIndex: 0,
+		lastFinalizedBatchIndex: lastProcessedBatch.BatchIndex,
 		batches:                 common.NewHeap[da.Entry](),
 		batchesMap:              common.NewShrinkingMap[uint64, *common.HeapElement[da.Entry]](1000),
+		previousBatch:           lastProcessedBatch,
 	}
 }
 
 // NextBatch finds next finalized batch and returns data, that was committed in that batch
-func (bq *BatchQueue) NextBatch(ctx context.Context) (da.Entry, error) {
-	if batch := bq.getFinalizedBatch(); batch != nil {
+func (bq *BatchQueue) NextBatch(ctx context.Context) (da.EntryWithBlocks, error) {
+	if batch := bq.nextFinalizedBatch(); batch != nil {
 		return batch, nil
 	}
 
@@ -50,13 +55,15 @@ func (bq *BatchQueue) NextBatch(ctx context.Context) (da.Entry, error) {
 		case da.CommitBatchV0Type, da.CommitBatchWithBlobType:
 			bq.addBatch(daEntry)
 		case da.RevertBatchType:
-			bq.deleteBatch(daEntry)
+			if err = bq.handleRevertEvent(daEntry.Event()); err != nil {
+				return nil, fmt.Errorf("failed to handle revert event: %w", err)
+			}
 		case da.FinalizeBatchType:
 			if daEntry.BatchIndex() > bq.lastFinalizedBatchIndex {
 				bq.lastFinalizedBatchIndex = daEntry.BatchIndex()
 			}
 
-			if batch := bq.getFinalizedBatch(); batch != nil {
+			if batch := bq.nextFinalizedBatch(); batch != nil {
 				return batch, nil
 			}
 		default:
@@ -65,16 +72,17 @@ func (bq *BatchQueue) NextBatch(ctx context.Context) (da.Entry, error) {
 	}
 }
 
-// getFinalizedBatch returns next finalized batch if there is available
-func (bq *BatchQueue) getFinalizedBatch() da.Entry {
+// nextFinalizedBatch returns next finalized batch if there is available
+func (bq *BatchQueue) nextFinalizedBatch() da.EntryWithBlocks {
 	if bq.batches.Len() == 0 {
 		return nil
 	}
 
 	batch := bq.batches.Peek().Value()
+	// we process all batches smaller or equal to the last finalized batch index -> this reflects bundles of multiple batches
+	// where we only receive the finalize event for the last batch of the bundle.
 	if batch.BatchIndex() <= bq.lastFinalizedBatchIndex {
-		bq.deleteBatch(batch)
-		return batch
+		return bq.processAndDeleteBatch(batch)
 	} else {
 		return nil
 	}
@@ -85,25 +93,84 @@ func (bq *BatchQueue) addBatch(batch da.Entry) {
 	bq.batchesMap.Set(batch.BatchIndex(), heapElement)
 }
 
-// deleteBatch deletes data committed in the batch from map, because this batch is reverted or finalized
-// updates DASyncedL1BlockNumber
-func (bq *BatchQueue) deleteBatch(batch da.Entry) {
-	batchHeapElement, exists := bq.batchesMap.Get(batch.BatchIndex())
+func (bq *BatchQueue) handleRevertEvent(event l1.RollupEvent) error {
+	switch event.Type() {
+	case l1.RevertEventV0Type:
+		revertBatch, ok := event.(*l1.RevertBatchEventV0)
+		if !ok {
+			return fmt.Errorf("unexpected type of revert event: %T, expected RevertEventV0Type", event)
+		}
+
+		bq.deleteBatch(revertBatch.BatchIndex().Uint64())
+	case l1.RevertEventV7Type:
+		revertBatch, ok := event.(*l1.RevertBatchEventV7)
+		if !ok {
+			return fmt.Errorf("unexpected type of revert event: %T, expected RevertEventV7Type", event)
+		}
+
+		// delete all batches from revertBatch.StartBatchIndex (inclusive) to revertBatch.FinishBatchIndex (inclusive)
+		for i := revertBatch.StartBatchIndex().Uint64(); i <= revertBatch.FinishBatchIndex().Uint64(); i++ {
+			bq.deleteBatch(i)
+		}
+	default:
+		return fmt.Errorf("unexpected type of revert event: %T", event)
+	}
+
+	return nil
+}
+
+func (bq *BatchQueue) deleteBatch(batchIndex uint64) (deleted bool) {
+	batchHeapElement, exists := bq.batchesMap.Get(batchIndex)
 	if !exists {
-		return
+		return false
 	}
 
-	bq.batchesMap.Delete(batch.BatchIndex())
+	bq.batchesMap.Delete(batchIndex)
 	bq.batches.Remove(batchHeapElement)
 
-	// we store here min height of currently loaded batches to be able to start syncing from the same place in case of restart
-	// TODO: we should store this information when the batch is done being processed to avoid inconsistencies
-	rawdb.WriteDASyncedL1BlockNumber(bq.db, batch.L1BlockNumber()-1)
+	return true
+}
+
+// processAndDeleteBatch processes a batch and deletes the batch from map. Stores the syncing progress on disk.
+func (bq *BatchQueue) processAndDeleteBatch(batch da.Entry) da.EntryWithBlocks {
+	if !bq.deleteBatch(batch.BatchIndex()) {
+		return nil
+	}
+
+	entryWithBlocks, ok := batch.(da.EntryWithBlocks)
+	if !ok {
+		// this should only happen if we delete a reverted batch
+		return nil
+	}
+
+	// sanity check that the next batch is the one we expect. If not, we skip the batch.
+	if bq.previousBatch.BatchIndex > 0 && bq.previousBatch.BatchIndex+1 != entryWithBlocks.BatchIndex() {
+		log.Info("BatchQueue: skipping batch ", "currentBatch", entryWithBlocks.BatchIndex(), "previousBatch", bq.previousBatch.BatchIndex)
+		return nil
+	}
+
+	// carry forward the total L1 messages popped from the previous batch
+	entryWithBlocks.SetParentTotalL1MessagePopped(bq.previousBatch.TotalL1MessagesPopped)
+
+	// we store the previous batch as it has been completely processed which we know because the next batch is requested within the pipeline.
+	// In case of a restart or crash we can continue from the last processed batch (and its metadata).
+	rawdb.WriteDAProcessedBatchMeta(bq.db, bq.previousBatch)
+
+	log.Info("processing batch", "batchIndex", entryWithBlocks.BatchIndex(), "L1BlockNumber", entryWithBlocks.L1BlockNumber(), "totalL1MessagesPopped", entryWithBlocks.TotalL1MessagesPopped(), "previousBatch", bq.previousBatch.BatchIndex, "previousL1BlockNumber", bq.previousBatch.L1BlockNumber, "previous TotalL1MessagesPopped", bq.previousBatch.TotalL1MessagesPopped)
+
+	bq.previousBatch = &rawdb.DAProcessedBatchMeta{
+		L1BlockNumber:         entryWithBlocks.L1BlockNumber(),
+		BatchIndex:            entryWithBlocks.BatchIndex(),
+		TotalL1MessagesPopped: entryWithBlocks.TotalL1MessagesPopped(),
+	}
+
+	return entryWithBlocks
 }
 
-func (bq *BatchQueue) Reset(height uint64) {
+func (bq *BatchQueue) Reset(lastProcessedBatchMeta *rawdb.DAProcessedBatchMeta) {
 	bq.batches.Clear()
 	bq.batchesMap.Clear()
-	bq.lastFinalizedBatchIndex = 0
-	bq.DAQueue.Reset(height)
+	bq.lastFinalizedBatchIndex = lastProcessedBatchMeta.BatchIndex
+	bq.previousBatch = lastProcessedBatchMeta
+	bq.DAQueue.Reset(lastProcessedBatchMeta)
 }