synchronizer: check l1blocks (#3546)

* wip

* run on background L1block checker

* fix lint and documentation

* fix conflict

* add unittest

* more unittest

* fix lint

* increase timeout for async unittest

* fix unittest

* rename GetResponse for GetResult and fix uniitest

* add a second gorutines for check the newest blocks

* more unittest

* add unittest and run also preCheck on launch

* by default Precheck from FINALIZED and SAFE

* fix unittest, apply PR comments

* changes suggested by ARR552 in integration method

* fix documentation

* import new network-l1-mock from PR#3553

* import new network-l1-mock from PR#3553

* import new network-l1-mock from PR#3553

* import new network-l1-mock from PR#3553

* fix unittest

* fix PR comments

* fix error

* checkReorgAndExecuteReset can't be call with lastEthBlockSynced=nil

* add parentHash to error

* fix error

* merge 3553 fix unittest

* fix unittest

* fix wrong merge

* adapt parallel reorg detection to flow

* fix unit tests

* fix log

* allow use sync parallel mode

---------

Co-authored-by: Alonso <ARR551@protonmail.com>

docs/config-file/node-config-schema.json

@@ -537,6 +537,57 @@
 					"description": "L1SyncCheckL2BlockNumberModulus is the modulus used to choose the l2block to check\na modules 5, for instance, means check all l2block multiples of 5 (10,15,20,...)",
 					"default": 600
 				},
+				"L1BlockCheck": {
+					"properties": {
+						"Enable": {
+							"type": "boolean",
+							"description": "Enable if is true then the check l1 Block Hash is active",
+							"default": true
+						},
+						"L1SafeBlockPoint": {
+							"type": "string",
+							"enum": [
+								"finalized",
+								"safe",
+								"latest"
+							],
+							"description": "L1SafeBlockPoint is the point that a block is considered safe enough to be checked\nit can be: finalized, safe,pending or latest",
+							"default": "finalized"
+						},
+						"L1SafeBlockOffset": {
+							"type": "integer",
+							"description": "L1SafeBlockOffset is the offset to add to L1SafeBlockPoint as a safe point\nit can be positive or negative\nExample: L1SafeBlockPoint= finalized, L1SafeBlockOffset= -10, then the safe block ten blocks before the finalized block",
+							"default": 0
+						},
+						"ForceCheckBeforeStart": {
+							"type": "boolean",
+							"description": "ForceCheckBeforeStart if is true then the first time the system is started it will force to check all pending blocks",
+							"default": true
+						},
+						"PreCheckEnable": {
+							"type": "boolean",
+							"description": "PreCheckEnable if is true then the pre-check is active, will check blocks between L1SafeBlock and L1PreSafeBlock",
+							"default": true
+						},
+						"L1PreSafeBlockPoint": {
+							"type": "string",
+							"enum": [
+								"finalized",
+								"safe",
+								"latest"
+							],
+							"description": "L1PreSafeBlockPoint is the point that a block is considered safe enough to be checked\nit can be: finalized, safe,pending or latest",
+							"default": "safe"
+						},
+						"L1PreSafeBlockOffset": {
+							"type": "integer",
+							"description": "L1PreSafeBlockOffset is the offset to add to L1PreSafeBlockPoint as a safe point\nit can be positive or negative\nExample: L1PreSafeBlockPoint= finalized, L1PreSafeBlockOffset= -10, then the safe block ten blocks before the finalized block",
+							"default": 0
+						}
+					},
+					"additionalProperties": false,
+					"type": "object"
+				},
 				"L1SynchronizationMode": {
 					"type": "string",
 					"enum": [

state/interfaces.go

@@ -25,6 +25,7 @@ type storage interface {
 	GetLastBlock(ctx context.Context, dbTx pgx.Tx) (*Block, error)
 	GetPreviousBlock(ctx context.Context, offset uint64, dbTx pgx.Tx) (*Block, error)
 	GetFirstUncheckedBlock(ctx context.Context, fromBlockNumber uint64, dbTx pgx.Tx) (*Block, error)
+	GetUncheckedBlocks(ctx context.Context, fromBlockNumber uint64, toBlockNumber uint64, dbTx pgx.Tx) ([]*Block, error)
 	UpdateCheckedBlockByNumber(ctx context.Context, blockNumber uint64, newCheckedStatus bool, dbTx pgx.Tx) error
 	AddGlobalExitRoot(ctx context.Context, exitRoot *GlobalExitRoot, dbTx pgx.Tx) error
 	GetLatestGlobalExitRoot(ctx context.Context, maxBlockNumber uint64, dbTx pgx.Tx) (GlobalExitRoot, time.Time, error)
@@ -160,6 +161,7 @@ type storage interface {
 	UpdateBatchAsChecked(ctx context.Context, batchNumber uint64, dbTx pgx.Tx) error
 	GetNotCheckedBatches(ctx context.Context, dbTx pgx.Tx) ([]*Batch, error)
 	GetLastL2BlockByBatchNumber(ctx context.Context, batchNumber uint64, dbTx pgx.Tx) (*L2Block, error)
+	GetPreviousBlockToBlockNumber(ctx context.Context, blockNumber uint64, dbTx pgx.Tx) (*Block, error)
 	AddL1InfoTreeRecursiveRootToExitRoot(ctx context.Context, exitRoot *L1InfoTreeRecursiveExitRootStorageEntry, dbTx pgx.Tx) error
 	GetAllL1InfoTreeRecursiveRootEntries(ctx context.Context, dbTx pgx.Tx) ([]L1InfoTreeRecursiveExitRootStorageEntry, error)
 	GetLatestL1InfoTreeRecursiveRoot(ctx context.Context, maxBlockNumber uint64, dbTx pgx.Tx) (L1InfoTreeRecursiveExitRootStorageEntry, error)

state/pgstatestorage/block.go

@@ -63,6 +63,35 @@ func (p *PostgresStorage) GetFirstUncheckedBlock(ctx context.Context, fromBlockN
 	return &block, err
 }
 
+func (p *PostgresStorage) GetUncheckedBlocks(ctx context.Context, fromBlockNumber uint64, toBlockNumber uint64, dbTx pgx.Tx) ([]*state.Block, error) {
+	const getUncheckedBlocksSQL = "SELECT block_num, block_hash, parent_hash, received_at, checked FROM state.block WHERE block_num>=$1 AND block_num<=$2 AND checked=false ORDER BY block_num"
+
+	q := p.getExecQuerier(dbTx)
+
+	rows, err := q.Query(ctx, getUncheckedBlocksSQL, fromBlockNumber, toBlockNumber)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var blocks []*state.Block
+	for rows.Next() {
+		var (
+			blockHash  string
+			parentHash string
+			block      state.Block
+		)
+		err := rows.Scan(&block.BlockNumber, &blockHash, &parentHash, &block.ReceivedAt, &block.Checked)
+		if err != nil {
+			return nil, err
+		}
+		block.BlockHash = common.HexToHash(blockHash)
+		block.ParentHash = common.HexToHash(parentHash)
+		blocks = append(blocks, &block)
+	}
+	return blocks, nil
+}
+
 // GetPreviousBlock gets the offset previous L1 block respect to latest.
 func (p *PostgresStorage) GetPreviousBlock(ctx context.Context, offset uint64, dbTx pgx.Tx) (*state.Block, error) {
 	var (
@@ -83,6 +112,26 @@ func (p *PostgresStorage) GetPreviousBlock(ctx context.Context, offset uint64, d
 	return &block, err
 }
 
+// GetPreviousBlockToBlockNumber gets the previous L1 block respect blockNumber.
+func (p *PostgresStorage) GetPreviousBlockToBlockNumber(ctx context.Context, blockNumber uint64, dbTx pgx.Tx) (*state.Block, error) {
+	var (
+		blockHash  string
+		parentHash string
+		block      state.Block
+	)
+	const getPreviousBlockSQL = "SELECT block_num, block_hash, parent_hash, received_at,checked FROM state.block WHERE block_num < $1 ORDER BY block_num DESC LIMIT 1 "
+
+	q := p.getExecQuerier(dbTx)
+
+	err := q.QueryRow(ctx, getPreviousBlockSQL, blockNumber).Scan(&block.BlockNumber, &blockHash, &parentHash, &block.ReceivedAt, &block.Checked)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, state.ErrNotFound
+	}
+	block.BlockHash = common.HexToHash(blockHash)
+	block.ParentHash = common.HexToHash(parentHash)
+	return &block, err
+}
+
 // GetBlockByNumber returns the L1 block with the given number.
 func (p *PostgresStorage) GetBlockByNumber(ctx context.Context, blockNumber uint64, dbTx pgx.Tx) (*state.Block, error) {
 	var (

state/pgstatestorage/pgstatestorage_test.go

@@ -1703,3 +1703,24 @@ func TestUpdateCheckedBlockByNumber(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, b1.Checked)
 }
+
+func TestGetUncheckedBlocks(t *testing.T) {
+	var err error
+	blockNumber := uint64(61001)
+	err = testState.AddBlock(context.Background(), &state.Block{BlockNumber: blockNumber, Checked: true}, nil)
+	require.NoError(t, err)
+	err = testState.AddBlock(context.Background(), &state.Block{BlockNumber: blockNumber + 1, Checked: false}, nil)
+	require.NoError(t, err)
+	err = testState.AddBlock(context.Background(), &state.Block{BlockNumber: blockNumber + 2, Checked: true}, nil)
+	require.NoError(t, err)
+	err = testState.AddBlock(context.Background(), &state.Block{BlockNumber: blockNumber + 3, Checked: false}, nil)
+	require.NoError(t, err)
+	err = testState.AddBlock(context.Background(), &state.Block{BlockNumber: blockNumber + 4, Checked: false}, nil)
+	require.NoError(t, err)
+
+	blocks, err := testState.GetUncheckedBlocks(context.Background(), blockNumber, blockNumber+3, nil)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(blocks))
+	require.Equal(t, uint64(blockNumber+1), blocks[0].BlockNumber)
+	require.Equal(t, uint64(blockNumber+3), blocks[1].BlockNumber)
+}

synchronizer/common/reorg_error.go

@@ -0,0 +1,44 @@
+package common
+
+import "fmt"
+
+// ReorgError is an error that is raised when a reorg is detected
+type ReorgError struct {
+	// BlockNumber is the block number that caused the reorg
+	BlockNumber uint64
+	Err         error
+}
+
+// NewReorgError creates a new ReorgError
+func NewReorgError(blockNumber uint64, err error) *ReorgError {
+	return &ReorgError{
+		BlockNumber: blockNumber,
+		Err:         err,
+	}
+}
+
+func (e *ReorgError) Error() string {
+	return fmt.Sprintf("%s blockNumber: %d", e.Err.Error(), e.BlockNumber)
+}
+
+// IsReorgError checks if an error is a ReorgError
+func IsReorgError(err error) bool {
+	_, ok := err.(*ReorgError)
+	return ok
+}
+
+// GetReorgErrorBlockNumber returns the block number that caused the reorg
+func GetReorgErrorBlockNumber(err error) uint64 {
+	if reorgErr, ok := err.(*ReorgError); ok {
+		return reorgErr.BlockNumber
+	}
+	return 0
+}
+
+// GetReorgError returns the error that caused the reorg
+func GetReorgError(err error) error {
+	if reorgErr, ok := err.(*ReorgError); ok {
+		return reorgErr.Err
+	}
+	return nil
+}

synchronizer/common/syncinterfaces/async_l1_block_checker.go

@@ -0,0 +1,40 @@
+package syncinterfaces
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/0xPolygonHermez/zkevm-node/state"
+)
+
+type IterationResult struct {
+	Err           error
+	ReorgDetected bool
+	BlockNumber   uint64
+	ReorgMessage  string
+}
+
+func (ir *IterationResult) String() string {
+	if ir.Err == nil {
+		if ir.ReorgDetected {
+			return fmt.Sprintf("IterationResult{ReorgDetected: %v, BlockNumber: %d ReorgMessage:%s}", ir.ReorgDetected, ir.BlockNumber, ir.ReorgMessage)
+		} else {
+			return "IterationResult{None}"
+		}
+	} else {
+		return fmt.Sprintf("IterationResult{Err: %s, ReorgDetected: %v, BlockNumber: %d ReorgMessage:%s}", ir.Err.Error(), ir.ReorgDetected, ir.BlockNumber, ir.ReorgMessage)
+	}
+}
+
+type AsyncL1BlockChecker interface {
+	Run(ctx context.Context, onFinish func())
+	RunSynchronous(ctx context.Context) IterationResult
+	Stop()
+	GetResult() *IterationResult
+}
+
+type L1BlockCheckerIntegrator interface {
+	OnStart(ctx context.Context) error
+	OnResetState(ctx context.Context)
+	CheckReorgWrapper(ctx context.Context, reorgFirstBlockOk *state.Block, errReportedByReorgFunc error) (*state.Block, error)
+}

synchronizer/common/syncinterfaces/etherman.go

@@ -14,10 +14,11 @@ type EthermanFullInterface interface {
 	HeaderByNumber(ctx context.Context, number *big.Int) (*ethTypes.Header, error)
 	GetRollupInfoByBlockRange(ctx context.Context, fromBlock uint64, toBlock *uint64) ([]etherman.Block, map[common.Hash][]etherman.Order, error)
 	EthBlockByNumber(ctx context.Context, blockNumber uint64) (*ethTypes.Block, error)
-	GetLatestBatchNumber() (uint64, error)
 	GetTrustedSequencerURL() (string, error)
 	VerifyGenBlockNumber(ctx context.Context, genBlockNumber uint64) (bool, error)
 	GetLatestVerifiedBatchNum() (uint64, error)
+
+	EthermanGetLatestBatchNumber
 	GetFinalizedBlockNumber(ctx context.Context) (uint64, error)
 	EthermanPreRollup
 }