Fix/2.61 upgrade fork changes (#1630)

* sequencer to go through stage loop whilst halted * updates to L1 processing when network is in a halted state
0xPolygonHermez · Jan 14, 2025 · dc7ceae · dc7ceae
1 parent a57c00b
commit dc7ceae
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 11 deletions.
diff --git a/zk/stages/stage_batches.go b/zk/stages/stage_batches.go
@@ -5,28 +5,28 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
-	"sync/atomic"
-	"time"
 	"os"
+	"sync/atomic"
 	"syscall"
+	"time"
 
 	"github.com/ledgerwatch/erigon-lib/chain"
 	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/kv"
 
+	"github.com/ledgerwatch/erigon/core/rawdb"
+	"github.com/ledgerwatch/erigon/core/state"
 	ethTypes "github.com/ledgerwatch/erigon/core/types"
+	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/eth/stagedsync"
 	"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
 	"github.com/ledgerwatch/erigon/params"
 	"github.com/ledgerwatch/erigon/zk"
+	"github.com/ledgerwatch/erigon/zk/datastream/client"
 	"github.com/ledgerwatch/erigon/zk/datastream/types"
 	"github.com/ledgerwatch/erigon/zk/erigon_db"
 	"github.com/ledgerwatch/erigon/zk/hermez_db"
 	"github.com/ledgerwatch/erigon/zk/sequencer"
-	"github.com/ledgerwatch/erigon/core/rawdb"
-	"github.com/ledgerwatch/erigon/core/state"
-	"github.com/ledgerwatch/erigon/eth/ethconfig"
-	"github.com/ledgerwatch/erigon/zk/datastream/client"
 	"github.com/ledgerwatch/log/v3"
 )
 
@@ -201,6 +201,7 @@ func SpawnStageBatches(
 
 	var highestDSL2Block uint64
 	newBlockCheckStartTIme := time.Now()
+	newBlockCheckCounter := 0
 	for {
 		select {
 		case <-ctx.Done():
@@ -227,6 +228,11 @@ func SpawnStageBatches(
 		if time.Since(newBlockCheckStartTIme) > 10*time.Second {
 			log.Info(fmt.Sprintf("[%s] Waiting for at least one new block in datastream", logPrefix), "datastreamBlock", highestDSL2Block, "last processed block", stageProgressBlockNo)
 			newBlockCheckStartTIme = time.Now()
+			newBlockCheckCounter++
+			if newBlockCheckCounter > 3 {
+				log.Info(fmt.Sprintf("[%s] No new blocks in datastream, entering stage loop", logPrefix))
+				return nil
+			}
 		}
 		time.Sleep(50 * time.Millisecond)
 	}

diff --git a/zk/stages/stage_sequence_execute.go b/zk/stages/stage_sequence_execute.go
@@ -162,10 +162,15 @@ func sequencingBatchStep(
 		shouldCheckForExecutionAndDataStreamAlignment = false
 	}
 
-	needsUnwind, err := tryHaltSequencer(batchContext, batchState, streamWriter, u, executionAt)
+	needsUnwind, exitStage, err := tryHaltSequencer(batchContext, batchState, streamWriter, u, executionAt)
 	if needsUnwind || err != nil {
 		return err
 	}
+	if exitStage {
+		log.Info(fmt.Sprintf("[%s] Exiting stage during halted sequencer", logPrefix))
+		// commit the tx so any updates to the stream etc are persisted
+		return sdb.tx.Commit()
+	}
 
 	if err := utils.UpdateZkEVMBlockCfg(cfg.chainConfig, sdb.hermezDb, logPrefix); err != nil {
 		return err

diff --git a/zk/stages/stage_sequence_execute_utils.go b/zk/stages/stage_sequence_execute_utils.go
@@ -427,7 +427,7 @@ func updateSequencerProgress(tx kv.RwTx, newHeight uint64, newBatch uint64, unwi
 	return nil
 }
 
-func tryHaltSequencer(batchContext *BatchContext, batchState *BatchState, streamWriter *SequencerBatchStreamWriter, u stagedsync.Unwinder, latestBlock uint64) (bool, error) {
+func tryHaltSequencer(batchContext *BatchContext, batchState *BatchState, streamWriter *SequencerBatchStreamWriter, u stagedsync.Unwinder, latestBlock uint64) (bool, bool, error) {
 	if batchContext.cfg.zk.SequencerHaltOnBatchNumber != 0 && batchContext.cfg.zk.SequencerHaltOnBatchNumber == batchState.batchNumber {
 		log.Info(fmt.Sprintf("[%s] Attempting to halt on batch %v, checking for pending verifications", batchContext.s.LogPrefix(), batchState.batchNumber))
 
@@ -439,7 +439,7 @@ func tryHaltSequencer(batchContext *BatchContext, batchState *BatchState, stream
 				time.Sleep(2 * time.Second)
 				needsUnwind, err := updateStreamAndCheckRollback(batchContext, batchState, streamWriter, u)
 				if needsUnwind || err != nil {
-					return needsUnwind, err
+					return needsUnwind, false, err
 				}
 			} else {
 				log.Info(fmt.Sprintf("[%s] No pending verifications, halting sequencer...", batchContext.s.LogPrefix()))
@@ -449,16 +449,21 @@ func tryHaltSequencer(batchContext *BatchContext, batchState *BatchState, stream
 
 		// we need to ensure the batch is also sealed in the datastream at this point
 		if err := finalizeLastBatchInDatastreamIfNotFinalized(batchContext, batchState.batchNumber-1, latestBlock); err != nil {
-			return false, err
+			return false, false, err
 		}
 
+		haltedCount := 0
 		for {
 			log.Info(fmt.Sprintf("[%s] Halt sequencer on batch %d...", batchContext.s.LogPrefix(), batchState.batchNumber))
 			time.Sleep(5 * time.Second) //nolint:gomnd
+			haltedCount++
+			if haltedCount > 3 {
+				return false, true, nil
+			}
 		}
 	}
 
-	return false, nil
+	return false, false, nil
 }
 
 type batchChecker interface {