feat(verifier): add worker pool for verifying proofs (#357)

Co-authored-by: maskpp <maskpp266@gmail.com>
Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>
Co-authored-by: Péter Garamvölgyi <peter@scroll.io>
Co-authored-by: colinlyguo <651734127@qq.com>

.github/workflows/contracts.yaml

@@ -2,26 +2,10 @@ name: Contracts
 
 on:
   push:
-    branches:
-      - master
-      - main
-      - prod
-      - release/*
-      - staging
-      - develop
-      - alpha
     paths:
       - 'contracts/**'
       - '.github/workflows/contracts.yaml'
   pull_request:
-    branches:
-      - master
-      - main
-      - prod
-      - release/*
-      - staging
-      - develop
-      - alpha
     paths:
       - 'contracts/**'
       - '.github/workflows/contracts.yaml'
@@ -44,6 +28,9 @@ jobs:
         uses: foundry-rs/foundry-toolchain@v1
         with:
           version: nightly
+      
+      - name: Setup LCOV
+        uses: hrishikesh-kadam/setup-lcov@v1
 
       - name: Install Node.js 14
         uses: actions/setup-node@v2
@@ -80,6 +67,22 @@ jobs:
       - name: Run foundry tests
         run: forge test -vvv
 
+      - name: Run foundry coverage
+        run : forge coverage --report lcov
+
+      - name : Prune coverage
+        run : lcov --remove ./lcov.info -o ./lcov.info.pruned 'src/mocks/*' 'src/test/*' 'scripts/*' 'node_modules/*' 'lib/*'
+
+      - name: Report code coverage
+        uses: zgosalvez/github-actions-report-lcov@v3
+        with:
+          coverage-files: contracts/lcov.info.pruned
+          minimum-coverage: 0
+          artifact-name: code-coverage-report
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          working-directory: contracts
+          update-comment: true
+
   hardhat:
     runs-on: ubuntu-latest
 

bridge/config/relayer_config.go

@@ -51,6 +51,8 @@ type RelayerConfig struct {
 	GasOracleConfig *GasOracleConfig `json:"gas_oracle_config"`
 	// The interval in which we send finalize batch transactions.
 	FinalizeBatchIntervalSec uint64 `json:"finalize_batch_interval_sec"`
+	// MessageRelayMinGasLimit to avoid OutOfGas error
+	MessageRelayMinGasLimit uint64 `json:"message_relay_min_gas_limit,omitempty"`
 	// The private key of the relayer
 	MessageSenderPrivateKeys   []*ecdsa.PrivateKey `json:"-"`
 	GasOracleSenderPrivateKeys []*ecdsa.PrivateKey `json:"-"`

bridge/l1/relayer.go

@@ -35,6 +35,8 @@ const (
 	gasPriceDiffPrecision = 1000000
 
 	defaultGasPriceDiff = 50000 // 5%
+
+	defaultMessageRelayMinGasLimit = 130000 // should be enough for both ERC20 and ETH relay
 )
 
 // Layer1Relayer is responsible for
@@ -58,6 +60,8 @@ type Layer1Relayer struct {
 	gasOracleCh     <-chan *sender.Confirmation
 	l1GasOracleABI  *abi.ABI
 
+	minGasLimitForMessageRelay uint64
+
 	lastGasPrice uint64
 	minGasPrice  uint64
 	gasPriceDiff uint64
@@ -92,6 +96,11 @@ func NewLayer1Relayer(ctx context.Context, db database.OrmFactory, cfg *config.R
 		gasPriceDiff = defaultGasPriceDiff
 	}
 
+	minGasLimitForMessageRelay := uint64(defaultMessageRelayMinGasLimit)
+	if cfg.MessageRelayMinGasLimit != 0 {
+		minGasLimitForMessageRelay = cfg.MessageRelayMinGasLimit
+	}
+
 	return &Layer1Relayer{
 		ctx: ctx,
 		db:  db,
@@ -104,6 +113,8 @@ func NewLayer1Relayer(ctx context.Context, db database.OrmFactory, cfg *config.R
 		gasOracleCh:     gasOracleSender.ConfirmChan(),
 		l1GasOracleABI:  bridge_abi.L1GasPriceOracleABI,
 
+		minGasLimitForMessageRelay: minGasLimitForMessageRelay,
+
 		minGasPrice:  minGasPrice,
 		gasPriceDiff: gasPriceDiff,
 
@@ -138,7 +149,7 @@ func (r *Layer1Relayer) ProcessSavedEvents() {
 func (r *Layer1Relayer) processSavedEvent(msg *types.L1Message) error {
 	calldata := common.Hex2Bytes(msg.Calldata)
 
-	hash, err := r.messageSender.SendTransaction(msg.MsgHash, &r.cfg.MessengerContractAddress, big.NewInt(0), calldata)
+	hash, err := r.messageSender.SendTransaction(msg.MsgHash, &r.cfg.MessengerContractAddress, big.NewInt(0), calldata, r.minGasLimitForMessageRelay)
 	if err != nil && err.Error() == "execution reverted: Message expired" {
 		return r.db.UpdateLayer1Status(r.ctx, msg.MsgHash, types.MsgExpired)
 	}
@@ -190,7 +201,7 @@ func (r *Layer1Relayer) ProcessGasPriceOracle() {
 				return
 			}
 
-			hash, err := r.gasOracleSender.SendTransaction(block.Hash, &r.cfg.GasPriceOracleContractAddress, big.NewInt(0), data)
+			hash, err := r.gasOracleSender.SendTransaction(block.Hash, &r.cfg.GasPriceOracleContractAddress, big.NewInt(0), data, 0)
 			if err != nil {
 				if !errors.Is(err, sender.ErrNoAvailableAccount) {
 					log.Error("Failed to send setL1BaseFee tx to layer2 ", "block.Hash", block.Hash, "block.Height", block.Number, "err", err)

bridge/l2/relayer.go

@@ -44,6 +44,8 @@ const (
 	gasPriceDiffPrecision = 1000000
 
 	defaultGasPriceDiff = 50000 // 5%
+
+	defaultMessageRelayMinGasLimit = 200000 // should be enough for both ERC20 and ETH relay
 )
 
 // Layer2Relayer is responsible for
@@ -72,6 +74,8 @@ type Layer2Relayer struct {
 	gasOracleCh     <-chan *sender.Confirmation
 	l2GasOracleABI  *abi.ABI
 
+	minGasLimitForMessageRelay uint64
+
 	lastGasPrice uint64
 	minGasPrice  uint64
 	gasPriceDiff uint64
@@ -122,6 +126,11 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db databa
 		gasPriceDiff = defaultGasPriceDiff
 	}
 
+	minGasLimitForMessageRelay := uint64(defaultMessageRelayMinGasLimit)
+	if cfg.MessageRelayMinGasLimit != 0 {
+		minGasLimitForMessageRelay = cfg.MessageRelayMinGasLimit
+	}
+
 	return &Layer2Relayer{
 		ctx: ctx,
 		db:  db,
@@ -140,6 +149,8 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db databa
 		gasOracleCh:     gasOracleSender.ConfirmChan(),
 		l2GasOracleABI:  bridge_abi.L2GasPriceOracleABI,
 
+		minGasLimitForMessageRelay: minGasLimitForMessageRelay,
+
 		minGasPrice:  minGasPrice,
 		gasPriceDiff: gasPriceDiff,
 
@@ -232,7 +243,7 @@ func (r *Layer2Relayer) processSavedEvent(msg *types.L2Message) error {
 		return err
 	}
 
-	hash, err := r.messageSender.SendTransaction(msg.MsgHash, &r.cfg.MessengerContractAddress, big.NewInt(0), data)
+	hash, err := r.messageSender.SendTransaction(msg.MsgHash, &r.cfg.MessengerContractAddress, big.NewInt(0), data, r.minGasLimitForMessageRelay)
 	if err != nil && err.Error() == "execution reverted: Message expired" {
 		return r.db.UpdateLayer2Status(r.ctx, msg.MsgHash, types.MsgExpired)
 	}
@@ -284,7 +295,7 @@ func (r *Layer2Relayer) ProcessGasPriceOracle() {
 				return
 			}
 
-			hash, err := r.gasOracleSender.SendTransaction(batch.Hash, &r.cfg.GasPriceOracleContractAddress, big.NewInt(0), data)
+			hash, err := r.gasOracleSender.SendTransaction(batch.Hash, &r.cfg.GasPriceOracleContractAddress, big.NewInt(0), data, 0)
 			if err != nil {
 				if !errors.Is(err, sender.ErrNoAvailableAccount) {
 					log.Error("Failed to send setL2BaseFee tx to layer2 ", "batch.Hash", batch.Hash, "err", err)
@@ -330,7 +341,7 @@ func (r *Layer2Relayer) SendCommitTx(batchData []*types.BatchData) error {
 		bytes = append(bytes, batch.Hash().Bytes()...)
 	}
 	txID := crypto.Keccak256Hash(bytes).String()
-	txHash, err := r.rollupSender.SendTransaction(txID, &r.cfg.RollupContractAddress, big.NewInt(0), calldata)
+	txHash, err := r.rollupSender.SendTransaction(txID, &r.cfg.RollupContractAddress, big.NewInt(0), calldata, 0)
 	if err != nil {
 		if !errors.Is(err, sender.ErrNoAvailableAccount) {
 			log.Error("Failed to send commitBatches tx to layer1 ", "err", err)
@@ -479,7 +490,7 @@ func (r *Layer2Relayer) ProcessCommittedBatches() {
 
 		txID := hash + "-finalize"
 		// add suffix `-finalize` to avoid duplication with commit tx in unit tests
-		txHash, err := r.rollupSender.SendTransaction(txID, &r.cfg.RollupContractAddress, big.NewInt(0), data)
+		txHash, err := r.rollupSender.SendTransaction(txID, &r.cfg.RollupContractAddress, big.NewInt(0), data, 0)
 		finalizeTxHash := &txHash
 		if err != nil {
 			if !errors.Is(err, sender.ErrNoAvailableAccount) {

bridge/l2/watcher_test.go

@@ -45,7 +45,7 @@ func testCreateNewWatcherAndStop(t *testing.T) {
 	numTransactions := 3
 	toAddress := common.HexToAddress("0x4592d8f8d7b001e72cb26a73e4fa1806a51ac79d")
 	for i := 0; i < numTransactions; i++ {
-		_, err = newSender.SendTransaction(strconv.Itoa(1000+i), &toAddress, big.NewInt(1000000000), nil)
+		_, err = newSender.SendTransaction(strconv.Itoa(1000+i), &toAddress, big.NewInt(1000000000), nil, 0)
 		assert.NoError(t, err)
 		<-newSender.ConfirmChan()
 	}

bridge/sender/estimategas.go

@@ -9,12 +9,12 @@ import (
 	"github.com/scroll-tech/go-ethereum/common"
 )
 
-func (s *Sender) estimateLegacyGas(auth *bind.TransactOpts, contract *common.Address, value *big.Int, input []byte) (*FeeData, error) {
+func (s *Sender) estimateLegacyGas(auth *bind.TransactOpts, contract *common.Address, value *big.Int, input []byte, minGasLimit uint64) (*FeeData, error) {
 	gasPrice, err := s.client.SuggestGasPrice(s.ctx)
 	if err != nil {
 		return nil, err
 	}
-	gasLimit, err := s.estimateGasLimit(auth, contract, input, gasPrice, nil, nil, value)
+	gasLimit, err := s.estimateGasLimit(auth, contract, input, gasPrice, nil, nil, value, minGasLimit)
 	if err != nil {
 		return nil, err
 	}
@@ -24,7 +24,7 @@ func (s *Sender) estimateLegacyGas(auth *bind.TransactOpts, contract *common.Add
 	}, nil
 }
 
-func (s *Sender) estimateDynamicGas(auth *bind.TransactOpts, contract *common.Address, value *big.Int, input []byte) (*FeeData, error) {
+func (s *Sender) estimateDynamicGas(auth *bind.TransactOpts, contract *common.Address, value *big.Int, input []byte, minGasLimit uint64) (*FeeData, error) {
 	gasTipCap, err := s.client.SuggestGasTipCap(s.ctx)
 	if err != nil {
 		return nil, err
@@ -38,7 +38,7 @@ func (s *Sender) estimateDynamicGas(auth *bind.TransactOpts, contract *common.Ad
 		gasTipCap,
 		new(big.Int).Mul(baseFee, big.NewInt(2)),
 	)
-	gasLimit, err := s.estimateGasLimit(auth, contract, input, nil, gasTipCap, gasFeeCap, value)
+	gasLimit, err := s.estimateGasLimit(auth, contract, input, nil, gasTipCap, gasFeeCap, value, minGasLimit)
 	if err != nil {
 		return nil, err
 	}
@@ -49,7 +49,7 @@ func (s *Sender) estimateDynamicGas(auth *bind.TransactOpts, contract *common.Ad
 	}, nil
 }
 
-func (s *Sender) estimateGasLimit(opts *bind.TransactOpts, contract *common.Address, input []byte, gasPrice, gasTipCap, gasFeeCap, value *big.Int) (uint64, error) {
+func (s *Sender) estimateGasLimit(opts *bind.TransactOpts, contract *common.Address, input []byte, gasPrice, gasTipCap, gasFeeCap, value *big.Int, minGasLimit uint64) (uint64, error) {
 	msg := ethereum.CallMsg{
 		From:      opts.From,
 		To:        contract,
@@ -63,6 +63,10 @@ func (s *Sender) estimateGasLimit(opts *bind.TransactOpts, contract *common.Addr
 	if err != nil {
 		return 0, err
 	}
+	if minGasLimit > gasLimit {
+		gasLimit = minGasLimit
+	}
+
 	gasLimit = gasLimit * 15 / 10 // 50% extra gas to void out of gas error
 
 	return gasLimit, nil

bridge/sender/sender.go

@@ -150,15 +150,15 @@ func (s *Sender) NumberOfAccounts() int {
 	return len(s.auths.accounts)
 }
 
-func (s *Sender) getFeeData(auth *bind.TransactOpts, target *common.Address, value *big.Int, data []byte) (*FeeData, error) {
+func (s *Sender) getFeeData(auth *bind.TransactOpts, target *common.Address, value *big.Int, data []byte, minGasLimit uint64) (*FeeData, error) {
 	if s.config.TxType == DynamicFeeTxType {
-		return s.estimateDynamicGas(auth, target, value, data)
+		return s.estimateDynamicGas(auth, target, value, data, minGasLimit)
 	}
-	return s.estimateLegacyGas(auth, target, value, data)
+	return s.estimateLegacyGas(auth, target, value, data, minGasLimit)
 }
 
 // SendTransaction send a signed L2tL1 transaction.
-func (s *Sender) SendTransaction(ID string, target *common.Address, value *big.Int, data []byte) (hash common.Hash, err error) {
+func (s *Sender) SendTransaction(ID string, target *common.Address, value *big.Int, data []byte, minGasLimit uint64) (hash common.Hash, err error) {
 	// We occupy the ID, in case some other threads call with the same ID in the same time
 	if _, loaded := s.pendingTxs.LoadOrStore(ID, nil); loaded {
 		return common.Hash{}, fmt.Errorf("has the repeat tx ID, ID: %s", ID)
@@ -182,7 +182,7 @@ func (s *Sender) SendTransaction(ID string, target *common.Address, value *big.I
 		tx      *types.Transaction
 	)
 	// estimate gas fee
-	if feeData, err = s.getFeeData(auth, target, value, data); err != nil {
+	if feeData, err = s.getFeeData(auth, target, value, data, minGasLimit); err != nil {
 		return
 	}
 	if tx, err = s.createAndSendTx(auth, feeData, target, value, data, nil); err == nil {

bridge/sender/sender_test.go

@@ -14,6 +14,7 @@ import (
 	cmap "github.com/orcaman/concurrent-map"
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/crypto"
+	"github.com/scroll-tech/go-ethereum/ethclient"
 	"github.com/scroll-tech/go-ethereum/rpc"
 	"github.com/stretchr/testify/assert"
 
@@ -56,11 +57,38 @@ func TestSender(t *testing.T) {
 	// Setup
 	setupEnv(t)
 
+	t.Run("test min gas limit", func(t *testing.T) { testMinGasLimit(t) })
+
 	t.Run("test 1 account sender", func(t *testing.T) { testBatchSender(t, 1) })
 	t.Run("test 3 account sender", func(t *testing.T) { testBatchSender(t, 3) })
 	t.Run("test 8 account sender", func(t *testing.T) { testBatchSender(t, 8) })
 }
 
+func testMinGasLimit(t *testing.T) {
+	senderCfg := cfg.L1Config.RelayerConfig.SenderConfig
+	senderCfg.Confirmations = rpc.LatestBlockNumber
+	newSender, err := sender.NewSender(context.Background(), senderCfg, privateKeys)
+	assert.NoError(t, err)
+	defer newSender.Stop()
+
+	client, err := ethclient.Dial(senderCfg.Endpoint)
+	assert.NoError(t, err)
+
+	// MinGasLimit = 0
+	txHash0, err := newSender.SendTransaction("0", &common.Address{}, big.NewInt(1), nil, 0)
+	assert.NoError(t, err)
+	tx0, _, err := client.TransactionByHash(context.Background(), txHash0)
+	assert.NoError(t, err)
+	assert.Greater(t, tx0.Gas(), uint64(0))
+
+	// MinGasLimit = 100000
+	txHash1, err := newSender.SendTransaction("1", &common.Address{}, big.NewInt(1), nil, 100000)
+	assert.NoError(t, err)
+	tx1, _, err := client.TransactionByHash(context.Background(), txHash1)
+	assert.NoError(t, err)
+	assert.Equal(t, tx1.Gas(), uint64(150000))
+}
+
 func testBatchSender(t *testing.T, batchSize int) {
 	for len(privateKeys) < batchSize {
 		priv, err := crypto.GenerateKey()
@@ -90,7 +118,7 @@ func testBatchSender(t *testing.T, batchSize int) {
 			for i := 0; i < TXBatch; i++ {
 				toAddr := common.HexToAddress("0x4592d8f8d7b001e72cb26a73e4fa1806a51ac79d")
 				id := strconv.Itoa(i + index*1000)
-				_, err := newSender.SendTransaction(id, &toAddr, big.NewInt(1), nil)
+				_, err := newSender.SendTransaction(id, &toAddr, big.NewInt(1), nil, 0)
 				if errors.Is(err, sender.ErrNoAvailableAccount) {
 					<-time.After(time.Second)
 					continue

common/utils/workerpool/workerpool.go

@@ -0,0 +1,51 @@
+package workerpool
+
+import (
+	"sync"
+)
+
+// WorkerPool is responsible for creating workers and managing verify proof task between them
+type WorkerPool struct {
+	maxWorker     int
+	taskQueueChan chan func()
+	wg            sync.WaitGroup
+}
+
+// NewWorkerPool creates new worker pool with given amount of workers
+func NewWorkerPool(maxWorker int) *WorkerPool {
+	return &WorkerPool{
+		maxWorker:     maxWorker,
+		taskQueueChan: nil,
+		wg:            sync.WaitGroup{},
+	}
+}
+
+// Run runs WorkerPool
+func (vwp *WorkerPool) Run() {
+	vwp.taskQueueChan = make(chan func())
+	for i := 0; i < vwp.maxWorker; i++ {
+		go func() {
+			for task := range vwp.taskQueueChan {
+				if task != nil {
+					task()
+					vwp.wg.Done()
+				} else {
+					return
+				}
+			}
+		}()
+	}
+}
+
+// Stop stop WorkerPool
+func (vwp *WorkerPool) Stop() {
+	vwp.wg.Wait()
+	// close task queue channel, so that all goruotines listening from it stop
+	close(vwp.taskQueueChan)
+}
+
+// AddTask adds a task to WorkerPool
+func (vwp *WorkerPool) AddTask(task func()) {
+	vwp.wg.Add(1)
+	vwp.taskQueueChan <- task
+}

common/utils/workerpool/workerpool_test.go

@@ -0,0 +1,57 @@
+package workerpool_test
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"scroll-tech/common/utils/workerpool"
+)
+
+func TestWorkerPool(t *testing.T) {
+	as := assert.New(t)
+
+	vwp := workerpool.NewWorkerPool(2)
+	vwp.Run()
+	var cnt int32 = 3
+
+	task := func() {
+		time.Sleep(500 * time.Millisecond)
+		atomic.AddInt32(&cnt, -1)
+	}
+
+	go vwp.AddTask(task)
+	go vwp.AddTask(task)
+	go vwp.AddTask(task)
+
+	time.Sleep(600 * time.Millisecond)
+	as.Equal(int32(1), atomic.LoadInt32(&cnt))
+	vwp.Stop()
+	as.Equal(int32(0), atomic.LoadInt32(&cnt))
+
+}
+
+func TestWorkerPoolStopAndStart(t *testing.T) {
+	as := assert.New(t)
+	vwp := workerpool.NewWorkerPool(1)
+	var cnt int32 = 3
+
+	task := func() {
+		time.Sleep(500 * time.Millisecond)
+		atomic.AddInt32(&cnt, -1)
+	}
+
+	vwp.Run()
+	vwp.AddTask(task)
+	vwp.AddTask(task)
+	vwp.Stop()
+	as.Equal(int32(1), atomic.LoadInt32(&cnt))
+
+	vwp.Run()
+	vwp.AddTask(task)
+	vwp.Stop()
+	as.Equal(int32(0), atomic.LoadInt32(&cnt))
+
+}

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "alpha-v2.0"
+var tag = "alpha-v2.2"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {

contracts/docs/apis/L1ScrollMessenger.md

@@ -163,17 +163,6 @@ function owner() external view returns (address)
 |---|---|---|
 | _0 | address | undefined |
 
-### pause
-
-```solidity
-function pause() external nonpayable
-```
-
-Pause the contract
-
-*This function can only called by contract owner.*
-
-
 ### paused
 
 ```solidity
@@ -264,6 +253,26 @@ The address of Rollup contract.
 
 ### sendMessage
 
+```solidity
+function sendMessage(address _to, uint256 _value, bytes _message, uint256 _gasLimit, address _refundAddress) external payable
+```
+
+Send cross chain message from L1 to L2 or L2 to L1.
+
+
+
+#### Parameters
+
+| Name | Type | Description |
+|---|---|---|
+| _to | address | undefined |
+| _value | uint256 | undefined |
+| _message | bytes | undefined |
+| _gasLimit | uint256 | undefined |
+| _refundAddress | address | undefined |
+
+### sendMessage
+
 ```solidity
 function sendMessage(address _to, uint256 _value, bytes _message, uint256 _gasLimit) external payable
 ```
@@ -281,6 +290,22 @@ Send cross chain message from L1 to L2 or L2 to L1.
 | _message | bytes | undefined |
 | _gasLimit | uint256 | undefined |
 
+### setPause
+
+```solidity
+function setPause(bool _status) external nonpayable
+```
+
+Pause the contract
+
+*This function can only called by contract owner.*
+
+#### Parameters
+
+| Name | Type | Description |
+|---|---|---|
+| _status | bool | The pause status to update. |
+
 ### transferOwnership
 
 ```solidity

contracts/docs/apis/L2ScrollMessenger.md

@@ -212,17 +212,6 @@ function owner() external view returns (address)
 |---|---|---|
 | _0 | address | undefined |
 
-### pause
-
-```solidity
-function pause() external nonpayable
-```
-
-Pause the contract
-
-*This function can only called by contract owner.*
-
-
 ### paused
 
 ```solidity
@@ -294,6 +283,26 @@ function retryMessageWithProof(address _from, address _to, uint256 _value, uint2
 
 ### sendMessage
 
+```solidity
+function sendMessage(address _to, uint256 _value, bytes _message, uint256 _gasLimit, address _refundAddress) external payable
+```
+
+Send cross chain message from L1 to L2 or L2 to L1.
+
+
+
+#### Parameters
+
+| Name | Type | Description |
+|---|---|---|
+| _to | address | undefined |
+| _value | uint256 | undefined |
+| _message | bytes | undefined |
+| _gasLimit | uint256 | undefined |
+| _refundAddress | address | undefined |
+
+### sendMessage
+
 ```solidity
 function sendMessage(address _to, uint256 _value, bytes _message, uint256 _gasLimit) external payable
 ```
@@ -311,6 +320,22 @@ Send cross chain message from L1 to L2 or L2 to L1.
 | _message | bytes | undefined |
 | _gasLimit | uint256 | undefined |
 
+### setPause
+
+```solidity
+function setPause(bool _status) external nonpayable
+```
+
+Pause the contract
+
+*This function can only called by contract owner.*
+
+#### Parameters
+
+| Name | Type | Description |
+|---|---|---|
+| _status | bool | The pause status to update. |
+
 ### transferOwnership
 
 ```solidity

contracts/src/L1/L1ScrollMessenger.sol

@@ -44,6 +44,28 @@ contract L1ScrollMessenger is PausableUpgradeable, ScrollMessengerBase, IL1Scrol
   /// @notice The address of L1MessageQueue contract.
   address public messageQueue;
 
+  // @note move to ScrollMessengerBase in next big refactor
+  /// @dev The status of for non-reentrant check.
+  uint256 private _lock_status;
+
+  /**********************
+   * Function Modifiers *
+   **********************/
+
+  modifier nonReentrant() {
+    // On the first call to nonReentrant, _notEntered will be true
+    require(_lock_status != _ENTERED, "ReentrancyGuard: reentrant call");
+
+    // Any calls to nonReentrant after this point will fail
+    _lock_status = _ENTERED;
+
+    _;
+
+    // By storing the original value once again, a refund is triggered (see
+    // https://eips.ethereum.org/EIPS/eip-2200)
+    _lock_status = _NOT_ENTERED;
+  }
+
   /***************
    * Constructor *
    ***************/
@@ -80,46 +102,18 @@ contract L1ScrollMessenger is PausableUpgradeable, ScrollMessengerBase, IL1Scrol
     bytes memory _message,
     uint256 _gasLimit
   ) external payable override whenNotPaused {
-    address _messageQueue = messageQueue; // gas saving
-    address _counterpart = counterpart; // gas saving
+    _sendMessage(_to, _value, _message, _gasLimit, tx.origin);
+  }
 
-    // compute the actual cross domain message calldata.
-    uint256 _messageNonce = IL1MessageQueue(_messageQueue).nextCrossDomainMessageIndex();
-    bytes memory _xDomainCalldata = _encodeXDomainCalldata(msg.sender, _to, _value, _messageNonce, _message);
-
-    // compute and deduct the messaging fee to fee vault.
-    uint256 _fee = IL1MessageQueue(_messageQueue).estimateCrossDomainMessageFee(
-      address(this),
-      _counterpart,
-      _xDomainCalldata,
-      _gasLimit
-    );
-    require(msg.value >= _fee + _value, "Insufficient msg.value");
-    if (_fee > 0) {
-      (bool _success, ) = feeVault.call{ value: _fee }("");
-      require(_success, "Failed to deduct the fee");
-    }
-
-    // append message to L1MessageQueue
-    IL1MessageQueue(_messageQueue).appendCrossDomainMessage(_counterpart, _gasLimit, _xDomainCalldata);
-
-    // record the message hash for future use.
-    bytes32 _xDomainCalldataHash = keccak256(_xDomainCalldata);
-
-    // normally this won't happen, since each message has different nonce, but just in case.
-    require(!isL1MessageSent[_xDomainCalldataHash], "Duplicated message");
-    isL1MessageSent[_xDomainCalldataHash] = true;
-
-    emit SentMessage(msg.sender, _to, _value, _messageNonce, _gasLimit, _message);
-
-    // refund fee to tx.origin
-    unchecked {
-      uint256 _refund = msg.value - _fee - _value;
-      if (_refund > 0) {
-        (bool _success, ) = tx.origin.call{ value: _refund }("");
-        require(_success, "Failed to refund the fee");
-      }
-    }
+  /// @inheritdoc IScrollMessenger
+  function sendMessage(
+    address _to,
+    uint256 _value,
+    bytes calldata _message,
+    uint256 _gasLimit,
+    address _refundAddress
+  ) external payable override whenNotPaused {
+    _sendMessage(_to, _value, _message, _gasLimit, _refundAddress);
   }
 
   /// @inheritdoc IL1ScrollMessenger
@@ -199,4 +193,57 @@ contract L1ScrollMessenger is PausableUpgradeable, ScrollMessengerBase, IL1Scrol
       _unpause();
     }
   }
+
+  /**********************
+   * Internal Functions *
+   **********************/
+
+  function _sendMessage(
+    address _to,
+    uint256 _value,
+    bytes memory _message,
+    uint256 _gasLimit,
+    address _refundAddress
+  ) internal nonReentrant {
+    address _messageQueue = messageQueue; // gas saving
+    address _counterpart = counterpart; // gas saving
+
+    // compute the actual cross domain message calldata.
+    uint256 _messageNonce = IL1MessageQueue(_messageQueue).nextCrossDomainMessageIndex();
+    bytes memory _xDomainCalldata = _encodeXDomainCalldata(msg.sender, _to, _value, _messageNonce, _message);
+
+    // compute and deduct the messaging fee to fee vault.
+    uint256 _fee = IL1MessageQueue(_messageQueue).estimateCrossDomainMessageFee(
+      address(this),
+      _counterpart,
+      _xDomainCalldata,
+      _gasLimit
+    );
+    require(msg.value >= _fee + _value, "Insufficient msg.value");
+    if (_fee > 0) {
+      (bool _success, ) = feeVault.call{ value: _fee }("");
+      require(_success, "Failed to deduct the fee");
+    }
+
+    // append message to L1MessageQueue
+    IL1MessageQueue(_messageQueue).appendCrossDomainMessage(_counterpart, _gasLimit, _xDomainCalldata);
+
+    // record the message hash for future use.
+    bytes32 _xDomainCalldataHash = keccak256(_xDomainCalldata);
+
+    // normally this won't happen, since each message has different nonce, but just in case.
+    require(!isL1MessageSent[_xDomainCalldataHash], "Duplicated message");
+    isL1MessageSent[_xDomainCalldataHash] = true;
+
+    emit SentMessage(msg.sender, _to, _value, _messageNonce, _gasLimit, _message);
+
+    // refund fee to tx.origin
+    unchecked {
+      uint256 _refund = msg.value - _fee - _value;
+      if (_refund > 0) {
+        (bool _success, ) = _refundAddress.call{ value: _refund }("");
+        require(_success, "Failed to refund the fee");
+      }
+    }
+  }
 }

contracts/src/L2/L2ScrollMessenger.sol

@@ -63,15 +63,33 @@ contract L2ScrollMessenger is ScrollMessengerBase, PausableUpgradeable, IL2Scrol
   /// @notice The maximum number of times each L1 message can fail on L2.
   uint256 public maxFailedExecutionTimes;
 
+  // @note move to ScrollMessengerBase in next big refactor
+  /// @dev The status of for non-reentrant check.
+  uint256 private _lock_status;
+
+  /**********************
+   * Function Modifiers *
+   **********************/
+
+  modifier nonReentrant() {
+    // On the first call to nonReentrant, _notEntered will be true
+    require(_lock_status != _ENTERED, "ReentrancyGuard: reentrant call");
+
+    // Any calls to nonReentrant after this point will fail
+    _lock_status = _ENTERED;
+
+    _;
+
+    // By storing the original value once again, a refund is triggered (see
+    // https://eips.ethereum.org/EIPS/eip-2200)
+    _lock_status = _NOT_ENTERED;
+  }
+
   /***************
    * Constructor *
    ***************/
 
-  constructor(
-    address _blockContainer,
-    address _gasOracle,
-    address _messageQueue
-  ) {
+  constructor(address _blockContainer, address _gasOracle, address _messageQueue) {
     blockContainer = _blockContainer;
     gasOracle = _gasOracle;
     messageQueue = _messageQueue;
@@ -166,38 +184,18 @@ contract L2ScrollMessenger is ScrollMessengerBase, PausableUpgradeable, IL2Scrol
     bytes memory _message,
     uint256 _gasLimit
   ) external payable override whenNotPaused {
-    // by pass fee vault relay
-    if (feeVault != msg.sender) {
-      require(_gasLimit >= MIN_GAS_LIMIT, "gas limit too small");
-    }
+    _sendMessage(_to, _value, _message, _gasLimit, tx.origin);
+  }
 
-    // compute and deduct the messaging fee to fee vault.
-    uint256 _fee = _gasLimit * IL1GasPriceOracle(gasOracle).l1BaseFee();
-    require(msg.value >= _value + _fee, "Insufficient msg.value");
-    if (_fee > 0) {
-      (bool _success, ) = feeVault.call{ value: _fee }("");
-      require(_success, "Failed to deduct the fee");
-    }
-
-    uint256 _nonce = L2MessageQueue(messageQueue).nextMessageIndex();
-    bytes32 _xDomainCalldataHash = keccak256(_encodeXDomainCalldata(msg.sender, _to, _value, _nonce, _message));
-
-    // normally this won't happen, since each message has different nonce, but just in case.
-    require(!isL2MessageSent[_xDomainCalldataHash], "Duplicated message");
-    isL2MessageSent[_xDomainCalldataHash] = true;
-
-    L2MessageQueue(messageQueue).appendMessage(_xDomainCalldataHash);
-
-    emit SentMessage(msg.sender, _to, _value, _nonce, _gasLimit, _message);
-
-    // refund fee to tx.origin
-    unchecked {
-      uint256 _refund = msg.value - _fee - _value;
-      if (_refund > 0) {
-        (bool _success, ) = tx.origin.call{ value: _refund }("");
-        require(_success, "Failed to refund the fee");
-      }
-    }
+  /// @inheritdoc IScrollMessenger
+  function sendMessage(
+    address _to,
+    uint256 _value,
+    bytes calldata _message,
+    uint256 _gasLimit,
+    address _refundAddress
+  ) external payable override whenNotPaused {
+    _sendMessage(_to, _value, _message, _gasLimit, _refundAddress);
   }
 
   /// @inheritdoc IL2ScrollMessenger
@@ -270,6 +268,47 @@ contract L2ScrollMessenger is ScrollMessengerBase, PausableUpgradeable, IL2Scrol
    * Internal Functions *
    **********************/
 
+  function _sendMessage(
+    address _to,
+    uint256 _value,
+    bytes memory _message,
+    uint256 _gasLimit,
+    address _refundAddress
+  ) internal nonReentrant {
+    // by pass fee vault relay
+    if (feeVault != msg.sender) {
+      require(_gasLimit >= MIN_GAS_LIMIT, "gas limit too small");
+    }
+
+    // compute and deduct the messaging fee to fee vault.
+    uint256 _fee = _gasLimit * IL1GasPriceOracle(gasOracle).l1BaseFee();
+    require(msg.value >= _value + _fee, "Insufficient msg.value");
+    if (_fee > 0) {
+      (bool _success, ) = feeVault.call{ value: _fee }("");
+      require(_success, "Failed to deduct the fee");
+    }
+
+    uint256 _nonce = L2MessageQueue(messageQueue).nextMessageIndex();
+    bytes32 _xDomainCalldataHash = keccak256(_encodeXDomainCalldata(msg.sender, _to, _value, _nonce, _message));
+
+    // normally this won't happen, since each message has different nonce, but just in case.
+    require(!isL2MessageSent[_xDomainCalldataHash], "Duplicated message");
+    isL2MessageSent[_xDomainCalldataHash] = true;
+
+    L2MessageQueue(messageQueue).appendMessage(_xDomainCalldataHash);
+
+    emit SentMessage(msg.sender, _to, _value, _nonce, _gasLimit, _message);
+
+    // refund fee to tx.origin
+    unchecked {
+      uint256 _refund = msg.value - _fee - _value;
+      if (_refund > 0) {
+        (bool _success, ) = _refundAddress.call{ value: _refund }("");
+        require(_success, "Failed to refund the fee");
+      }
+    }
+  }
+
   function _executeMessage(
     address _from,
     address _to,

contracts/src/libraries/IScrollMessenger.sol

@@ -53,4 +53,18 @@ interface IScrollMessenger {
     bytes calldata message,
     uint256 gasLimit
   ) external payable;
+
+  /// @notice Send cross chain message from L1 to L2 or L2 to L1.
+  /// @param target The address of account who recieve the message.
+  /// @param value The amount of ether passed when call target contract.
+  /// @param message The content of the message.
+  /// @param gasLimit Gas limit required to complete the message relay on corresponding chain.
+  /// @param refundAddress The address of account who will receive the refunded fee.
+  function sendMessage(
+    address target,
+    uint256 value,
+    bytes calldata message,
+    uint256 gasLimit,
+    address refundAddress
+  ) external payable;
 }

contracts/src/libraries/ScrollMessengerBase.sol

@@ -23,6 +23,14 @@ abstract contract ScrollMessengerBase is OwnableUpgradeable, IScrollMessenger {
   /// @param _newFeeVault The address of new fee vault contract.
   event UpdateFeeVault(address _oldFeeVault, address _newFeeVault);
 
+  /*************
+   * Constants *
+   *************/
+
+  // https://github.com/OpenZeppelin/openzeppelin-contracts/blob/v4.5.0/contracts/security/ReentrancyGuard.sol
+  uint256 internal constant _NOT_ENTERED = 1;
+  uint256 internal constant _ENTERED = 2;
+
   /*************
    * Variables *
    *************/

contracts/src/test/L1ScrollMessengerTest.t.sol

@@ -30,6 +30,7 @@ contract L1ScrollMessengerTest is DSTestPlus {
 
     // Initialize L1 contracts
     l1Messenger.initialize(address(l2Messenger), feeVault, address(scrollChain), address(l1MessageQueue));
+    l1MessageQueue.initialize(address(l1Messenger), address(0));
     scrollChain.initialize(address(l1MessageQueue));
   }
 
@@ -49,4 +50,21 @@ contract L1ScrollMessengerTest is DSTestPlus {
     hevm.expectRevert("Forbid to call self");
     l1Messenger.relayMessageWithProof(address(this), address(l1Messenger), 0, 0, new bytes(0), proof);
   }
+
+  function testSendMessage(uint256 exceedValue, address refundAddress) external {
+    hevm.assume(refundAddress.code.length == 0);
+    hevm.assume(uint256(uint160(refundAddress)) > 100); // ignore some precompile contracts
+    hevm.assume(refundAddress != address(this));
+
+    exceedValue = bound(exceedValue, 1, address(this).balance / 2);
+
+    // Insufficient msg.value
+    hevm.expectRevert("Insufficient msg.value");
+    l1Messenger.sendMessage(address(0), 1, new bytes(0), 0, refundAddress);
+
+    // refund exceed fee
+    uint256 balanceBefore = refundAddress.balance;
+    l1Messenger.sendMessage{ value: 1 + exceedValue }(address(0), 1, new bytes(0), 0, refundAddress);
+    assertEq(balanceBefore + exceedValue, refundAddress.balance);
+  }
 }

contracts/src/test/L2ScrollMessenger.t.sol

@@ -47,4 +47,21 @@ contract L2ScrollMessengerTest is DSTestPlus {
     hevm.expectRevert("Forbid to call self");
     l2Messenger.relayMessage(address(this), address(l2Messenger), 0, 0, new bytes(0));
   }
+
+  function testSendMessage(uint256 exceedValue, address refundAddress) external {
+    hevm.assume(refundAddress.code.length == 0);
+    hevm.assume(uint256(uint160(refundAddress)) > 100); // ignore some precompile contracts
+    hevm.assume(refundAddress != address(this));
+
+    exceedValue = bound(exceedValue, 1, address(this).balance / 2);
+
+    // Insufficient msg.value
+    hevm.expectRevert("Insufficient msg.value");
+    l2Messenger.sendMessage(address(0), 1, new bytes(0), 21000, refundAddress);
+
+    // refund exceed fee
+    uint256 balanceBefore = refundAddress.balance;
+    l2Messenger.sendMessage{ value: 1 + exceedValue }(address(0), 1, new bytes(0), 21000, refundAddress);
+    assertEq(balanceBefore + exceedValue, refundAddress.balance);
+  }
 }

contracts/src/test/mocks/MockScrollMessenger.sol

@@ -35,4 +35,12 @@ contract MockScrollMessenger is IScrollMessenger {
     bytes memory _message,
     uint256 _gasLimit
   ) external payable {}
+
+  function sendMessage(
+    address _to,
+    uint256 _value,
+    bytes memory _message,
+    uint256 _gasLimit,
+    address _refundAddress
+  ) external payable {}
 }

coordinator/config.json

@@ -8,7 +8,8 @@
       "mock_mode": true,
       "params_path": "",
       "agg_vk_path": ""
-    }
+    },
+    "max_verifier_workers": 10
   },
   "db_config": {
     "driver_name": "postgres",

coordinator/config/config.go

@@ -10,6 +10,10 @@ import (
 	db_config "scroll-tech/database"
 )
 
+const (
+	defaultNumberOfVerifierWorkers = 10
+)
+
 // RollerManagerConfig loads sequencer configuration items.
 type RollerManagerConfig struct {
 	CompressionLevel int `json:"compression_level,omitempty"`
@@ -23,6 +27,8 @@ type RollerManagerConfig struct {
 	CollectionTime int `json:"collection_time"`
 	// Token time to live (in seconds)
 	TokenTimeToLive int `json:"token_time_to_live"`
+	// Max number of workers in verifier worker pool
+	MaxVerifierWorkers int `json:"max_verifier_workers,omitempty"`
 }
 
 // L2Config loads l2geth configuration items.
@@ -65,5 +71,9 @@ func NewConfig(file string) (*Config, error) {
 	}
 	cfg.RollerManagerConfig.OrderSession = order
 
+	if cfg.RollerManagerConfig.MaxVerifierWorkers == 0 {
+		cfg.RollerManagerConfig.MaxVerifierWorkers = defaultNumberOfVerifierWorkers
+	}
+
 	return cfg, nil
 }

coordinator/manager.go

@@ -23,6 +23,8 @@ import (
 	"scroll-tech/common/types"
 	"scroll-tech/database"
 
+	"scroll-tech/common/utils/workerpool"
+
 	"scroll-tech/coordinator/config"
 	"scroll-tech/coordinator/verifier"
 )
@@ -89,6 +91,9 @@ type Manager struct {
 	tokenCache *cache.Cache
 	// A mutex guarding registration
 	registerMu sync.RWMutex
+
+	// Verifier worker pool
+	verifierWorkerPool *workerpool.WorkerPool
 }
 
 // New returns a new instance of Manager. The instance will be not fully prepared,
@@ -110,6 +115,7 @@ func New(ctx context.Context, cfg *config.RollerManagerConfig, orm database.OrmF
 		orm:                orm,
 		Client:             client,
 		tokenCache:         cache.New(time.Duration(cfg.TokenTimeToLive)*time.Second, 1*time.Hour),
+		verifierWorkerPool: workerpool.NewWorkerPool(cfg.MaxVerifierWorkers),
 	}, nil
 }
 
@@ -119,6 +125,7 @@ func (m *Manager) Start() error {
 		return nil
 	}
 
+	m.verifierWorkerPool.Run()
 	m.restorePrevSessions()
 
 	atomic.StoreInt32(&m.running, 1)
@@ -132,6 +139,7 @@ func (m *Manager) Stop() {
 	if !m.isRunning() {
 		return
 	}
+	m.verifierWorkerPool.Stop()
 
 	atomic.StoreInt32(&m.running, 0)
 }
@@ -302,7 +310,7 @@ func (m *Manager) handleZkProof(pk string, msg *message.ProofDetail) error {
 	coordinatorProofsReceivedTotalCounter.Inc(1)
 
 	var err error
-	success, err = m.verifier.VerifyProof(msg.Proof)
+	success, err = m.verifyProof(msg.Proof)
 	if err != nil {
 		// TODO: this is only a temp workaround for testnet, we should return err in real cases
 		success = false
@@ -562,3 +570,26 @@ func (m *Manager) VerifyToken(authMsg *message.AuthMsg) (bool, error) {
 	}
 	return true, nil
 }
+
+func (m *Manager) addVerifyTask(proof *message.AggProof) chan verifyResult {
+	c := make(chan verifyResult, 1)
+	m.verifierWorkerPool.AddTask(func() {
+		result, err := m.verifier.VerifyProof(proof)
+		c <- verifyResult{result, err}
+	})
+	return c
+}
+
+func (m *Manager) verifyProof(proof *message.AggProof) (bool, error) {
+	if !m.isRunning() {
+		return false, errors.New("coordinator has stopped before verification")
+	}
+	verifyResultChan := m.addVerifyTask(proof)
+	verifyResult := <-verifyResultChan
+	return verifyResult.result, verifyResult.err
+}
+
+type verifyResult struct {
+	result bool
+	err    error
+}

coordinator/manager_test.go

@@ -501,10 +501,11 @@ func setupCoordinator(t *testing.T, dbCfg *database.DBConfig, rollersPerSession
 	assert.True(t, assert.NoError(t, err), "failed to get db handler.")
 
 	rollerManager, err = coordinator.New(context.Background(), &coordinator_config.RollerManagerConfig{
-		RollersPerSession: rollersPerSession,
-		Verifier:          &coordinator_config.VerifierConfig{MockMode: true},
-		CollectionTime:    1,
-		TokenTimeToLive:   5,
+		RollersPerSession:  rollersPerSession,
+		Verifier:           &coordinator_config.VerifierConfig{MockMode: true},
+		CollectionTime:     1,
+		TokenTimeToLive:    5,
+		MaxVerifierWorkers: 10,
 	}, db, nil)
 	assert.NoError(t, err)
 	assert.NoError(t, rollerManager.Start())