feat(rollup_sync_service): handle multiple batches in FinalizeBatch event (#865)

* feat: now FinalizeBatch event will influence one or more batches

* fix bug && add unit test

* fix typo

* Update core/rawdb/accessors_rollup_event_test.go

Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com>

* trivial fix

* chore: auto version bump [bot]

* trivial change: move write finalized things logic out of loop

* utilize code && comment

* remove redundant check

* two log level changing case

---------

Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com>
Co-authored-by: Thegaram <Thegaram@users.noreply.github.com>

core/rawdb/accessors_rollup_event.go

@@ -144,3 +144,30 @@ func ReadFinalizedL2BlockNumber(db ethdb.Reader) *uint64 {
 	finalizedL2BlockNumber := number.Uint64()
 	return &finalizedL2BlockNumber
 }
+
+// WriteLastFinalizedBatchIndex stores the last finalized batch index in the database.
+func WriteLastFinalizedBatchIndex(db ethdb.KeyValueWriter, lastFinalizedBatchIndex uint64) {
+	value := big.NewInt(0).SetUint64(lastFinalizedBatchIndex).Bytes()
+	if err := db.Put(lastFinalizedBatchIndexKey, value); err != nil {
+		log.Crit("failed to store last finalized batch index for rollup event", "batch index", lastFinalizedBatchIndex, "value", value, "err", err)
+	}
+}
+
+// ReadLastFinalizedBatchIndex fetches the last finalized batch index from the database.
+func ReadLastFinalizedBatchIndex(db ethdb.Reader) *uint64 {
+	data, err := db.Get(lastFinalizedBatchIndexKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("failed to read last finalized batch index from database", "key", lastFinalizedBatchIndexKey, "err", err)
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("unexpected finalized batch index in database", "data", data, "number", number)
+	}
+
+	lastFinalizedBatchIndex := number.Uint64()
+	return &lastFinalizedBatchIndex
+}

core/rawdb/accessors_rollup_event_test.go

@@ -45,7 +45,7 @@ func TestFinalizedL2BlockNumber(t *testing.T) {
 
 	// read non-existing value
 	if got := ReadFinalizedL2BlockNumber(db); got != nil {
-		t.Fatal("Expected 0 for non-existing value", "got", *got)
+		t.Fatal("Expected nil for non-existing value", "got", *got)
 	}
 
 	for _, num := range blockNumbers {
@@ -58,6 +58,32 @@ func TestFinalizedL2BlockNumber(t *testing.T) {
 	}
 }
 
+func TestLastFinalizedBatchIndex(t *testing.T) {
+	batchIndxes := []uint64{
+		1,
+		1 << 2,
+		1 << 8,
+		1 << 16,
+		1 << 32,
+	}
+
+	db := NewMemoryDatabase()
+
+	// read non-existing value
+	if got := ReadLastFinalizedBatchIndex(db); got != nil {
+		t.Fatal("Expected nil for non-existing value", "got", *got)
+	}
+
+	for _, num := range batchIndxes {
+		WriteLastFinalizedBatchIndex(db, num)
+		got := ReadLastFinalizedBatchIndex(db)
+
+		if *got != num {
+			t.Fatal("Batch index mismatch", "expected", num, "got", got)
+		}
+	}
+}
+
 func TestFinalizedBatchMeta(t *testing.T) {
 	batches := []*FinalizedBatchMeta{
 		{

core/rawdb/schema.go

@@ -115,6 +115,7 @@ var (
 	batchChunkRangesPrefix            = []byte("R-bcr")
 	batchMetaPrefix                   = []byte("R-bm")
 	finalizedL2BlockNumberKey         = []byte("R-finalized")
+	lastFinalizedBatchIndexKey        = []byte("R-finalizedBatchIndex")
 
 	// Row consumption
 	rowConsumptionPrefix = []byte("rc") // rowConsumptionPrefix + hash -> row consumption by block

params/version.go

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

rollup/rollup_sync_service/rollup_sync_service.go

@@ -225,9 +225,26 @@ func (s *RollupSyncService) parseAndUpdateRollupEventLogs(logs []types.Log, endB
 			batchIndex := event.BatchIndex.Uint64()
 			log.Trace("found new FinalizeBatch event", "batch index", batchIndex)
 
-			parentBatchMeta, chunks, err := s.getLocalInfoForBatch(batchIndex)
+			lastFinalizedBatchIndex := rawdb.ReadLastFinalizedBatchIndex(s.db)
+
+			// After darwin, FinalizeBatch event emitted every bundle, which contains multiple batches.
+			// Therefore there are a range of finalized batches need to be saved into db.
+			//
+			// The range logic also applies to the batches before darwin when FinalizeBatch event emitted
+			// per single batch. In this situation, `batchIndex` just equals to `*lastFinalizedBatchIndex + 1`
+			// and only one batch is processed through the for loop.
+			startBatchIndex := batchIndex
+			if lastFinalizedBatchIndex != nil {
+				startBatchIndex = *lastFinalizedBatchIndex + 1
+			} else {
+				log.Warn("got nil when reading last finalized batch index. This should happen only once.")
+			}
+
+			var highestFinalizedBlockNumber uint64
+			for index := startBatchIndex; index <= batchIndex; index++ {
+				parentBatchMeta, chunks, err := s.getLocalInfoForBatch(index)
 				if err != nil {
-				return fmt.Errorf("failed to get local node info, batch index: %v, err: %w", batchIndex, err)
+					return fmt.Errorf("failed to get local node info, batch index: %v, err: %w", index, err)
 				}
 
 				endBlock, finalizedBatchMeta, err := validateBatch(event, parentBatchMeta, chunks, s.bc.Config(), s.stack)
@@ -235,13 +252,19 @@ func (s *RollupSyncService) parseAndUpdateRollupEventLogs(logs []types.Log, endB
 					return fmt.Errorf("fatal: validateBatch failed: finalize event: %v, err: %w", event, err)
 				}
 
-			rawdb.WriteFinalizedL2BlockNumber(s.db, endBlock)
+				rawdb.WriteFinalizedBatchMeta(s.db, index, finalizedBatchMeta)
-			rawdb.WriteFinalizedBatchMeta(s.db, batchIndex, finalizedBatchMeta)
 
-			if batchIndex%100 == 0 {
+				if index%100 == 0 {
-				log.Info("finalized batch progress", "batch index", batchIndex, "finalized l2 block height", endBlock)
+					log.Info("finalized batch progress", "batch index", index, "finalized l2 block height", endBlock)
 				}
 
+				highestFinalizedBlockNumber = endBlock
+			}
+
+			rawdb.WriteFinalizedL2BlockNumber(s.db, highestFinalizedBlockNumber)
+			rawdb.WriteLastFinalizedBatchIndex(s.db, batchIndex)
+			log.Debug("write finalized l2 block number", "batch index", batchIndex, "finalized l2 block height", highestFinalizedBlockNumber)
+
 		default:
 			return fmt.Errorf("unknown event, topic: %v, tx hash: %v", vLog.Topics[0].Hex(), vLog.TxHash.Hex())
 		}