feat(permissionless batches): recovery mode after permissionless batches

cmd/geth/main.go

@@ -25,6 +25,8 @@ import (
 	"strings"
 	"time"
 
+	"go.uber.org/automaxprocs/maxprocs"
+
 	"github.com/scroll-tech/go-ethereum/accounts"
 	"github.com/scroll-tech/go-ethereum/accounts/keystore"
 	"github.com/scroll-tech/go-ethereum/cmd/utils"
@@ -39,7 +41,6 @@ import (
 	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/metrics"
 	"github.com/scroll-tech/go-ethereum/node"
-	"go.uber.org/automaxprocs/maxprocs"
 
 	// Force-load the tracer engines to trigger registration
 	_ "github.com/scroll-tech/go-ethereum/eth/tracers/js"
@@ -160,6 +161,11 @@ var (
 		utils.DABlockNativeAPIEndpointFlag,
 		utils.DABlobScanAPIEndpointFlag,
 		utils.DABeaconNodeAPIEndpointFlag,
+		utils.DARecoveryModeFlag,
+		utils.DARecoveryInitialL1BlockFlag,
+		utils.DARecoveryInitialBatchFlag,
+		utils.DARecoverySignBlocksFlag,
+		utils.DARecoveryL2EndBlockFlag,
 	}, utils.NetworkFlags, utils.DatabaseFlags)
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -1053,6 +1053,26 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Name:  "da.blob.beaconnode",
 		Usage: "Beacon node API endpoint",
 	}
+	DARecoveryModeFlag = &cli.BoolFlag{
+		Name:  "da.recovery",
+		Usage: "Enable recovery mode for DA syncing",
+	}
+	DARecoveryInitialL1BlockFlag = &cli.Uint64Flag{
+		Name:  "da.recovery.initiall1block",
+		Usage: "Initial L1 block to start recovery from",
+	}
+	DARecoveryInitialBatchFlag = &cli.Uint64Flag{
+		Name:  "da.recovery.initialbatch",
+		Usage: "Initial batch to start recovery from",
+	}
+	DARecoverySignBlocksFlag = &cli.BoolFlag{
+		Name:  "da.recovery.signblocks",
+		Usage: "Sign blocks during recovery (requires correct Clique signer key and history of blocks with Clique signatures)",
+	}
+	DARecoveryL2EndBlockFlag = &cli.Uint64Flag{
+		Name:  "da.recovery.l2endblock",
+		Usage: "End L2 block to recover to",
+	}
 )
 
 var (
@@ -1816,6 +1836,21 @@ func setDA(ctx *cli.Context, cfg *ethconfig.Config) {
 		if ctx.IsSet(DABeaconNodeAPIEndpointFlag.Name) {
 			cfg.DA.BeaconNodeAPIEndpoint = ctx.String(DABeaconNodeAPIEndpointFlag.Name)
 		}
+		if ctx.IsSet(DARecoveryModeFlag.Name) {
+			cfg.DA.RecoveryMode = ctx.Bool(DARecoveryModeFlag.Name)
+		}
+		if ctx.IsSet(DARecoveryInitialL1BlockFlag.Name) {
+			cfg.DA.InitialL1Block = ctx.Uint64(DARecoveryInitialL1BlockFlag.Name)
+		}
+		if ctx.IsSet(DARecoveryInitialBatchFlag.Name) {
+			cfg.DA.InitialBatch = ctx.Uint64(DARecoveryInitialBatchFlag.Name)
+		}
+		if ctx.IsSet(DARecoverySignBlocksFlag.Name) {
+			cfg.DA.SignBlocks = ctx.Bool(DARecoverySignBlocksFlag.Name)
+		}
+		if ctx.IsSet(DARecoveryL2EndBlockFlag.Name) {
+			cfg.DA.L2EndBlock = ctx.Uint64(DARecoveryL2EndBlockFlag.Name)
+		}
 	}
 }
 

core/blockchain.go

@@ -2021,15 +2021,15 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 	return it.index, err
 }
 
-func (bc *BlockChain) BuildAndWriteBlock(parentBlock *types.Block, header *types.Header, txs types.Transactions) (WriteStatus, error) {
+func (bc *BlockChain) BuildAndWriteBlock(parentBlock *types.Block, header *types.Header, txs types.Transactions, sign bool) (*types.Block, WriteStatus, error) {
 	if !bc.chainmu.TryLock() {
-		return NonStatTy, errInsertionInterrupted
+		return nil, NonStatTy, errInsertionInterrupted
 	}
 	defer bc.chainmu.Unlock()
 
 	statedb, err := state.New(parentBlock.Root(), bc.stateCache, bc.snaps)
 	if err != nil {
-		return NonStatTy, err
+		return nil, NonStatTy, err
 	}
 
 	statedb.StartPrefetcher("l1sync")
@@ -2040,18 +2040,51 @@ func (bc *BlockChain) BuildAndWriteBlock(parentBlock *types.Block, header *types
 	tempBlock := types.NewBlockWithHeader(header).WithBody(txs, nil)
 	receipts, logs, gasUsed, err := bc.processor.Process(tempBlock, statedb, bc.vmConfig)
 	if err != nil {
-		return NonStatTy, fmt.Errorf("error processing block: %w", err)
+		return nil, NonStatTy, fmt.Errorf("error processing block %d: %w", header.Number.Uint64(), err)
 	}
 
 	// TODO: once we have the extra and difficulty we need to verify the signature of the block with Clique
 	//  This should be done with https://github.com/scroll-tech/go-ethereum/pull/913.
 
-	// finalize and assemble block as fullBlock
+	if sign {
+		// remember the time as Clique will override it
+		originalTime := header.Time
+
+		err = bc.engine.Prepare(bc, header)
+		if err != nil {
+			return nil, NonStatTy, fmt.Errorf("error preparing block %d: %w", tempBlock.Number().Uint64(), err)
+		}
+
+		// we want to re-sign the block: set time to original value again.
+		header.Time = originalTime
+	}
+
+	// finalize and assemble block as fullBlock: replicates consensus.FinalizeAndAssemble()
 	header.GasUsed = gasUsed
 	header.Root = statedb.IntermediateRoot(bc.chainConfig.IsEIP158(header.Number))
 
 	fullBlock := types.NewBlock(header, txs, nil, receipts, trie.NewStackTrie(nil))
 
+	// Sign the block if requested
+	if sign {
+		resultCh, stopCh := make(chan *types.Block), make(chan struct{})
+		if err = bc.engine.Seal(bc, fullBlock, resultCh, stopCh); err != nil {
+			return nil, NonStatTy, fmt.Errorf("error sealing block %d: %w", fullBlock.Number().Uint64(), err)
+		}
+		// Clique.Seal() will only wait for a second before giving up on us. So make sure there is nothing computational heavy
+		// or a call that blocks between the call to Seal and the line below. Seal might introduce some delay, so we keep track of
+		// that artificially added delay and subtract it from overall runtime of commit().
+		fullBlock = <-resultCh
+		if fullBlock == nil {
+			return nil, NonStatTy, fmt.Errorf("sealing block failed %d: block is nil", header.Number.Uint64())
+		}
+
+		// verify the generated block with local consensus engine to make sure everything is as expected
+		if err = bc.engine.VerifyHeader(bc, fullBlock.Header()); err != nil {
+			return nil, NonStatTy, fmt.Errorf("error verifying signed block %d: %w", fullBlock.Number().Uint64(), err)
+		}
+	}
+
 	blockHash := fullBlock.Hash()
 	// manually replace the block hash in the receipts
 	for i, receipt := range receipts {
@@ -2068,7 +2101,14 @@ func (bc *BlockChain) BuildAndWriteBlock(parentBlock *types.Block, header *types
 		l.BlockHash = blockHash
 	}
 
-	return bc.writeBlockAndSetHead(fullBlock, receipts, logs, statedb, false)
+	// 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 nil, NonStatTy, fmt.Errorf("error validating block %d: %w", fullBlock.Number().Uint64(), err)
+	}
+
+	writeStatus, err := bc.writeBlockAndSetHead(fullBlock, receipts, logs, statedb, false)
+	return fullBlock, writeStatus, err
 }
 
 // insertSideChain is called when an import batch hits upon a pruned ancestor

eth/backend.go

@@ -254,6 +254,9 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthCl
 	// simply let them run simultaneously. If messages are missing in DA syncing, it will be handled by the syncing pipeline
 	// by waiting and retrying.
 	if config.EnableDASyncing {
+		// Enable CCC if flag is set so that row consumption can be generated.
+		config.DA.CCCEnable = config.CheckCircuitCapacity
+		config.DA.CCCNumWorkers = config.CCCMaxWorkers
 		eth.syncingPipeline, err = da_syncer.NewSyncingPipeline(context.Background(), eth.blockchain, chainConfig, eth.chainDb, l1Client, stack.Config().L1DeploymentBlock, config.DA)
 		if err != nil {
 			return nil, fmt.Errorf("cannot initialize da syncer: %w", err)

rollup/da_syncer/da_queue.go

@@ -10,14 +10,17 @@ import (
 
 // DAQueue is a pipeline stage that reads DA entries from a DataSource and provides them to the next stage.
 type DAQueue struct {
-	l1height          uint64
+	initialBatch uint64
+	l1height     uint64
+
 	dataSourceFactory *DataSourceFactory
 	dataSource        DataSource
 	da                da.Entries
 }
 
-func NewDAQueue(l1height uint64, dataSourceFactory *DataSourceFactory) *DAQueue {
+func NewDAQueue(l1height uint64, initialBatch uint64, dataSourceFactory *DataSourceFactory) *DAQueue {
 	return &DAQueue{
+		initialBatch:      initialBatch,
 		l1height:          l1height,
 		dataSourceFactory: dataSourceFactory,
 		dataSource:        nil,
@@ -26,15 +29,23 @@ func NewDAQueue(l1height uint64, dataSourceFactory *DataSourceFactory) *DAQueue
 }
 
 func (dq *DAQueue) NextDA(ctx context.Context) (da.Entry, error) {
-	for len(dq.da) == 0 {
-		err := dq.getNextData(ctx)
-		if err != nil {
-			return nil, err
+	for {
+		for len(dq.da) == 0 {
+			err := dq.getNextData(ctx)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		daEntry := dq.da[0]
+		dq.da = dq.da[1:]
+
+		if daEntry.BatchIndex() < dq.initialBatch {
+			continue
 		}
+
+		return daEntry, nil
 	}
-	daEntry := dq.da[0]
-	dq.da = dq.da[1:]
-	return daEntry, nil
 }
 
 func (dq *DAQueue) getNextData(ctx context.Context) error {

rollup/da_syncer/da_syncer.go

@@ -5,7 +5,9 @@ import (
 
 	"github.com/scroll-tech/go-ethereum/core"
 	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rollup/ccc"
 	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/da"
+	"github.com/scroll-tech/go-ethereum/rollup/da_syncer/serrors"
 )
 
 var (
@@ -14,35 +16,70 @@ var (
 )
 
 type DASyncer struct {
-	blockchain *core.BlockChain
+	asyncChecker *ccc.AsyncChecker
+	l2EndBlock   uint64
+	blockchain   *core.BlockChain
 }
 
-func NewDASyncer(blockchain *core.BlockChain) *DASyncer {
-	return &DASyncer{
+func NewDASyncer(blockchain *core.BlockChain, cccEnable bool, cccNumWorkers int, l2EndBlock uint64) *DASyncer {
+	s := &DASyncer{
+		l2EndBlock: l2EndBlock,
 		blockchain: blockchain,
 	}
+
+	if cccEnable {
+		s.asyncChecker = ccc.NewAsyncChecker(blockchain, cccNumWorkers, false)
+	}
+
+	return s
 }
 
 // SyncOneBlock receives a PartialBlock, makes sure it's the next block in the chain, executes it and inserts it to the blockchain.
-func (s *DASyncer) SyncOneBlock(block *da.PartialBlock) error {
+func (s *DASyncer) SyncOneBlock(block *da.PartialBlock, override bool, sign bool) error {
 	currentBlock := s.blockchain.CurrentBlock()
 
 	// we expect blocks to be consecutive. block.PartialHeader.Number == parentBlock.Number+1.
-	if block.PartialHeader.Number <= currentBlock.Number.Uint64() {
+	// if override is true, we allow blocks to be lower than the current block number and replace the blocks.
+	if !override && block.PartialHeader.Number <= currentBlock.Number.Uint64() {
 		log.Debug("block number is too low", "block number", block.PartialHeader.Number, "parent block number", currentBlock.Number.Uint64())
 		return ErrBlockTooLow
 	} else if block.PartialHeader.Number > currentBlock.Number.Uint64()+1 {
 		log.Debug("block number is too high", "block number", block.PartialHeader.Number, "parent block number", currentBlock.Number.Uint64())
 		return ErrBlockTooHigh
 	}
 
-	parentBlock := s.blockchain.GetBlockByNumber(currentBlock.Number.Uint64())
-	if _, err := s.blockchain.BuildAndWriteBlock(parentBlock, block.PartialHeader.ToHeader(), block.Transactions); err != nil {
+	parentBlockNumber := currentBlock.Number.Uint64()
+	if override {
+		parentBlockNumber = block.PartialHeader.Number - 1
+	}
+
+	parentBlock := s.blockchain.GetBlockByNumber(parentBlockNumber)
+	if parentBlock == nil {
+		return fmt.Errorf("failed getting parent block, number: %d", parentBlockNumber)
+	}
+
+	fullBlock, _, err := s.blockchain.BuildAndWriteBlock(parentBlock, block.PartialHeader.ToHeader(), block.Transactions, sign)
+	if err != nil {
 		return fmt.Errorf("failed building and writing block, number: %d, error: %v", block.PartialHeader.Number, err)
 	}
 
-	if s.blockchain.CurrentBlock().Number.Uint64()%1000 == 0 {
-		log.Info("L1 sync progress", "blockhain height", s.blockchain.CurrentBlock().Number.Uint64(), "block hash", s.blockchain.CurrentBlock().Hash(), "root", s.blockchain.CurrentBlock().Root)
+	if s.asyncChecker != nil {
+		_ = s.asyncChecker.Check(fullBlock)
+	}
+
+	currentBlock = s.blockchain.CurrentBlock()
+	if override && block.PartialHeader.Number != currentBlock.Number.Uint64() && block.PartialHeader.Number%100 == 0 {
+		newBlock := s.blockchain.GetHeaderByNumber(block.PartialHeader.Number)
+		log.Info("L1 sync progress", "processed block ", newBlock.Number.Uint64(), "block hash", newBlock.Hash(), "root", newBlock.Root)
+		log.Info("L1 sync progress", "blockhain height", currentBlock.Number.Uint64(), "block hash", currentBlock.Hash(), "root", currentBlock.Root)
+	} else if currentBlock.Number.Uint64()%100 == 0 {
+		log.Info("L1 sync progress", "blockhain height", currentBlock.Number.Uint64(), "block hash", currentBlock.Hash(), "root", currentBlock.Root)
+	}
+
+	if s.l2EndBlock > 0 && s.l2EndBlock == block.PartialHeader.Number {
+		newBlock := s.blockchain.GetHeaderByNumber(block.PartialHeader.Number)
+		log.Warn("L1 sync reached L2EndBlock: you can terminate recovery mode now", "L2EndBlock", newBlock.Number.Uint64(), "block hash", newBlock.Hash(), "root", newBlock.Root)
+		return serrors.Terminated
 	}
 
 	return nil

rollup/da_syncer/serrors/errors.go

@@ -12,6 +12,7 @@ const (
 var (
 	TemporaryError = NewTemporaryError(nil)
 	EOFError       = NewEOFError(nil)
+	Terminated     = fmt.Errorf("terminated")
 )
 
 type Type uint8