Async tx for rollup and finalize

ops/devnet-morph/devnet/__init__.py

@@ -214,7 +214,7 @@ def devnet_deploy(paths, args):
         env_content = envfile.readlines()
         for line in env_content:
             line = line.strip()
-            if line and not line.startswith('#'):  # 忽略空行和注释行
+            if line and not line.startswith('#'):
                 key, value = line.split('=')
                 env_data[key.strip()] = value.strip()
         env_data['L1_CROSS_DOMAIN_MESSENGER'] = addresses['Proxy__L1CrossDomainMessenger']
@@ -231,6 +231,8 @@ def devnet_deploy(paths, args):
 
     log.info('Bringing up L2.')
 
+
+
     run_command(['docker', 'compose', '-f', 'docker-compose-4nodes.yml', 'up',
                  '-d'], check=False, cwd=paths.ops_dir,
                 env={
@@ -243,7 +245,7 @@ def devnet_deploy(paths, args):
                     'GENESIS_FILE_PATH': '/genesis.json',
                     'L1_ETH_RPC': 'http://l1:8545',
                     'L1_BEACON_CHAIN_RPC': 'http://beacon-chain:3500',
-                    'BUILD_GETH': build_geth_target
+                    'BUILD_GETH': build_geth_target,  
                 })
     wait_up(8545)
     wait_for_rpc_server('127.0.0.1:8545')

tx-submitter/.gitignore

@@ -3,4 +3,6 @@
 build
 tx-submitter
 .DS_Store
-.vscode
+.vscode
+*.log
+*debug_bin*

tx-submitter/entry.go

@@ -5,10 +5,12 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"os/signal"
 
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/ethclient"
 	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rpc"
 	"github.com/urfave/cli"
 	"gopkg.in/natefinch/lumberjack.v2"
 
@@ -75,12 +77,14 @@ func Main() func(ctx *cli.Context) error {
 			return err
 		}
 
-		// Connect to L1 and L2 providers. Perform these last since they are the
-		// most expensive.
-		l1Client, err := ethclient.DialContext(ctx, cfg.L1EthRpc)
+		l1RpcClient, err := rpc.Dial(cfg.L1EthRpc)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to connect to L1 provider: %w", err)
 		}
+		// Connect to L1 and L2 providers. Perform these last since they are the
+		// most expensive.
+		l1Client := ethclient.NewClient(l1RpcClient)
+
 		// l2 rpcs
 		var l2Clients []iface.L2Client
 		for _, rpc := range cfg.L2EthRpcs {
@@ -119,6 +123,7 @@ func Main() func(ctx *cli.Context) error {
 		sr := services.NewRollup(
 			ctx,
 			m,
+			l1RpcClient,
 			l1Client,
 			l2Clients,
 			l1Rollup,
@@ -160,6 +165,13 @@ func Main() func(ctx *cli.Context) error {
 		)
 		sr.Start()
 
+		// Catch CTRL-C to ensure a graceful shutdown.
+		interrupt := make(chan os.Signal, 1)
+		signal.Notify(interrupt, os.Interrupt)
+
+		// Wait until the interrupt signal is received from an OS signal.
+		<-interrupt
+
 		return nil
 	}
 }

tx-submitter/iface/rollup.go

@@ -15,9 +15,8 @@ type IRollup interface {
 	CommitBatch(opts *bind.TransactOpts, batchDataInput bindings.IRollupBatchDataInput, batchSignatureInput bindings.IRollupBatchSignatureInput) (*types.Transaction, error)
 	LastFinalizedBatchIndex(opts *bind.CallOpts) (*big.Int, error)
 	FinalizeBatch(opts *bind.TransactOpts, _batchIndex *big.Int) (*types.Transaction, error)
-	// will be used in next version
-	//FinalizationPeriodSeconds(opts *bind.CallOpts) (*big.Int, error)
 	BatchInsideChallengeWindow(opts *bind.CallOpts, batchIndex *big.Int) (bool, error)
+	BatchExist(opts *bind.CallOpts, batchIndex *big.Int) (bool, error)
 }
 
 // IL2Sequencer is the interface for the sequencer on L2

tx-submitter/services/pendingtx.go

@@ -0,0 +1,186 @@
+package services
+
+import (
+	"sort"
+	"sync"
+	"time"
+
+	"morph-l2/bindings/bindings"
+	"morph-l2/tx-submitter/utils"
+
+	"github.com/scroll-tech/go-ethereum/accounts/abi"
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+)
+
+type TxInfo struct {
+	sendTime uint64
+	tx       types.Transaction
+
+	queryTimes uint64
+}
+
+type PendingTxs struct {
+	mu sync.Mutex
+
+	txinfos map[common.Hash]TxInfo
+	pnonce  uint64 // pending nonce
+
+	failedIndex *uint64
+	pindex      uint64 // pending batch index
+
+	pfinalize       uint64
+	commitBatchId   []byte
+	finalizeBatchId []byte
+}
+
+func NewPendingTxs(cid []byte, fid []byte) *PendingTxs {
+	return &PendingTxs{
+		txinfos:         make(map[common.Hash]TxInfo),
+		commitBatchId:   cid,
+		finalizeBatchId: fid,
+	}
+}
+
+func (pt *PendingTxs) Add(tx types.Transaction) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	pt.txinfos[tx.Hash()] = TxInfo{
+		sendTime: uint64(time.Now().Unix()),
+		tx:       tx,
+	}
+}
+
+func (pt *PendingTxs) Remove(txHash common.Hash) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	delete(pt.txinfos, txHash)
+}
+
+func (pt *PendingTxs) Recover(txs []types.Transaction, a *abi.ABI) {
+	// restore state from mempool
+	if len(txs) > 0 {
+		var pbindex, pfindex uint64
+
+		for _, tx := range txs {
+			pt.Add(tx)
+
+			method := utils.ParseMethod(tx, a)
+			if method == "commitBatch" {
+
+				index := utils.ParseParentBatchIndex(tx.Data())
+				if index > pbindex {
+					pbindex = index
+				}
+			} else if method == "finalizeBatch" {
+				findex := utils.ParseFBatchIndex(tx.Data())
+				if findex > pfindex {
+					pfindex = findex
+				}
+			}
+		}
+
+		pt.SetPindex(pbindex)
+		pt.SetPFinalize(pfindex)
+		pt.SetNonce(txs[len(txs)-1].Nonce())
+	}
+}
+
+func (pt *PendingTxs) GetAll() []TxInfo {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	// copy txs and return
+	txs := make([]TxInfo, 0, len(pt.txinfos))
+	for _, tx := range pt.txinfos {
+		txs = append(txs, tx)
+	}
+
+	// sort by nonce
+	sort.SliceStable(txs, func(i, j int) bool {
+		return txs[i].tx.Nonce() < txs[j].tx.Nonce()
+	})
+
+	return txs
+}
+
+func (pt *PendingTxs) Get(txHash common.Hash) (TxInfo, bool) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	tx, ok := pt.txinfos[txHash]
+	return tx, ok
+}
+
+func (pt *PendingTxs) IncQueryTimes(txHash common.Hash) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	pt.txinfos[txHash] = TxInfo{tx: pt.txinfos[txHash].tx, queryTimes: pt.txinfos[txHash].queryTimes + 1, sendTime: pt.txinfos[txHash].sendTime}
+}
+
+func (pt *PendingTxs) SetFailedStatus(index uint64) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+
+	// failed index must be less than pindex
+	if pt.failedIndex != nil || index >= pt.pindex {
+		return
+	}
+
+	pt.failedIndex = &index
+}
+func (pt *PendingTxs) SetPindex(index uint64) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+
+	pt.pindex = index
+}
+
+func (pt *PendingTxs) SetNonce(nonce uint64) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	pt.pnonce = nonce
+}
+
+func (pt *PendingTxs) SetPFinalize(finalize uint64) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	pt.pfinalize = finalize
+}
+
+func (pt *PendingTxs) RemoveRollupRestriction() {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+
+	pt.failedIndex = nil
+}
+
+func (pt *PendingTxs) HaveFailed() bool {
+	return pt.failedIndex != nil
+}
+
+func (pt *PendingTxs) ExistedIndex(index uint64) bool {
+
+	txs := pt.GetAll()
+
+	abi, _ := bindings.RollupMetaData.GetAbi()
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	for i := len(txs) - 1; i >= 0; i-- {
+		tx := txs[i].tx
+		if utils.ParseMethod(tx, abi) == "commitBatch" {
+			pindex := utils.ParseParentBatchIndex(tx.Data()) + 1
+			if index == pindex {
+				return true
+			}
+
+		}
+
+	}
+	return false
+
+}
+
+func (pt *PendingTxs) ResetFailedIndex(index uint64) {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+	pt.failedIndex = &index
+}