feat(tx-submitter): add rollup loss control functionality

WorldDogs · WorldDogs · commit c34ea51b2a5d · 2024-11-21T16:46:58.000+08:00
- Implement batch cache to store and retrieve rollup batches
- Add logic to calculate and compare transaction fees with collected L1 fees
- Introduce new configuration options for rollup loss control
- Update transaction submission process to include fee checks
- Add unit tests for new functionality
diff --git a/tx-submitter/entry.go b/tx-submitter/entry.go
@@ -69,6 +69,9 @@ func Main() func(ctx *cli.Context) error {
 			"rough_estimate_per_l1_msg", cfg.RollupTxGasPerL1Msg,
 			"log_level", cfg.LogLevel,
 			"leveldb_pathname", cfg.LeveldbPathName,
+			"l1_stop_in_blocks", cfg.BlockNotIncreasedThreshold,
+			"rollup_loss_control", cfg.RollupLossControl,
+			"rolluo_loss_buffer", cfg.MaxRollupLossBuffer,
 		)
 
 		ctx, cancel := context.WithCancel(context.Background())
diff --git a/tx-submitter/flags/flags.go b/tx-submitter/flags/flags.go
@@ -229,37 +229,37 @@ var (
 	}
 
 	// external sign
-	ExternalSign = cli.BoolFlag{
+	ExternalSignFlag = cli.BoolFlag{
 		Name:   "external_sign",
 		Usage:  "Enable external sign",
 		EnvVar: prefixEnvVar("EXTERNAL_SIGN"),
 	}
 
 	// address
-	ExternalSignAddress = cli.StringFlag{
+	ExternalSignAddressFlag = cli.StringFlag{
 		Name:   "external_sign_address",
 		Usage:  "The address of the external sign",
 		EnvVar: prefixEnvVar("EXTERNAL_SIGN_ADDRESS"),
 	}
 	// appid
-	ExternalSignAppid = cli.StringFlag{
+	ExternalSignAppidFlag = cli.StringFlag{
 		Name:   "external_sign_appid",
 		Usage:  "The appid of the external sign",
 		EnvVar: prefixEnvVar("EXTERNAL_SIGN_APPID"),
 	}
 	// chain
-	ExternalSignChain = cli.StringFlag{
+	ExternalSignChainFlag = cli.StringFlag{
 		Name:   "external_sign_chain",
 		Usage:  "The chain of the external sign",
 		EnvVar: prefixEnvVar("EXTERNAL_SIGN_CHAIN"),
 	}
 	// url
-	ExternalSignUrl = cli.StringFlag{
+	ExternalSignUrlFlag = cli.StringFlag{
 		Name:   "external_sign_url",
 		Usage:  "The url of the external sign",
 		EnvVar: prefixEnvVar("EXTERNAL_SIGN_URL"),
 	}
-	ExternalSignRsaPriv = cli.StringFlag{
+	ExternalSignRsaPrivFlag = cli.StringFlag{
 		Name:   "external_rsa_priv",
 		Usage:  "The rsa private key of the external sign",
 		EnvVar: "SEQUENCER_EXTERNAL_SIGN_RSA_PRIV", // use sequencer rsa from xxx
@@ -300,12 +300,24 @@ var (
 	}
 
 	// l1 block not incremented threshold
-	BlockNotIncreasedThreshold = cli.Int64Flag{
+	BlockNotIncreasedThresholdFlag = cli.Int64Flag{
 		Name:   "block_not_increased_threshold",
 		Usage:  "The threshold for block not incremented",
 		Value:  5,
 		EnvVar: prefixEnvVar("BLOCK_NOT_INCREASED_THRESHOLD"),
 	}
+	RollupLossControlFlag = cli.BoolFlag{
+		Name:   "rollup_loss_control",
+		Usage:  "Enable rollup loss control",
+		EnvVar: prefixEnvVar("ROLLUP_LOSS_CONTROL"),
+	}
+	// max rollup loss buffer
+	MaxRollupLossBufferFlag = cli.Int64Flag{
+		Name:   "max_rollup_loss_buffer",
+		Usage:  "single transaction rollup maximum allowable loss buff",
+		Value:  50, // default 50% percent
+		EnvVar: prefixEnvVar("MAX_ROLLUP_LOSS_BUFFER"),
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -349,18 +361,19 @@ var optionalFlags = []cli.Flag{
 	MaxTxsInPendingPoolFlag,
 
 	// external sign
-	ExternalSign,
-	ExternalSignAddress,
-	ExternalSignAppid,
-	ExternalSignChain,
-	ExternalSignUrl,
-	ExternalSignRsaPriv,
+	ExternalSignFlag,
+	ExternalSignAddressFlag,
+	ExternalSignAppidFlag,
+	ExternalSignChainFlag,
+	ExternalSignUrlFlag,
+	ExternalSignRsaPrivFlag,
 	RoughEstimateGasFlag,
 	RotatorBufferFlag,
 	StakingEventStoreFileFlag,
 	EventIndexStepFlag,
 	LeveldbPathNameFlag,
-	BlockNotIncreasedThreshold,
+	BlockNotIncreasedThresholdFlag,
+	MaxRollupLossBufferFlag,
 }
 
 // Flags contains the list of configuration options available to the binary.
diff --git a/tx-submitter/services/batch_cache.go b/tx-submitter/services/batch_cache.go
@@ -0,0 +1,39 @@
+package services
+
+import (
+	"sync"
+
+	"github.com/morph-l2/go-ethereum/eth"
+)
+
+type BatchCache struct {
+	batchCache map[uint64]*eth.RPCRollupBatch
+	m          sync.Mutex
+}
+
+func NewBatchCache() *BatchCache {
+	return &BatchCache{
+		batchCache: make(map[uint64]*eth.RPCRollupBatch),
+	}
+}
+
+func (b *BatchCache) Get(batchIndex uint64) (*eth.RPCRollupBatch, bool) {
+	b.m.Lock()
+	defer b.m.Unlock()
+
+	batch, ok := b.batchCache[batchIndex]
+	return batch, ok
+}
+func (b *BatchCache) Set(batchIndex uint64, batch *eth.RPCRollupBatch) {
+	b.m.Lock()
+	defer b.m.Unlock()
+
+	b.batchCache[batchIndex] = batch
+}
+
+func (b *BatchCache) Delete(batchIndex uint64) {
+	b.m.Lock()
+	defer b.m.Unlock()
+
+	delete(b.batchCache, batchIndex)
+}
diff --git a/tx-submitter/services/batch_calche_test.go b/tx-submitter/services/batch_calche_test.go
@@ -0,0 +1,21 @@
+package services
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBatchCache(t *testing.T) {
+
+	cache := NewBatchCache()
+	cache.Set(1, nil)
+	_, ok := cache.Get(1)
+	require.True(t, ok)
+	cache.Delete(1)
+	_, ok = cache.Get(1)
+	require.False(t, ok)
+	_, ok = cache.Get(2)
+	require.False(t, ok)
+
+}
diff --git a/tx-submitter/services/rollup.go b/tx-submitter/services/rollup.go
@@ -75,7 +75,7 @@ type Rollup struct {
 	// collectedL1FeeSum
 	collectedL1FeeSum float64
 	// batchcache
-	batchCache map[uint64]*eth.RPCRollupBatch
+	batchCache *BatchCache
 	bm         *l1checker.BlockMonitor
 }
 
@@ -114,7 +114,7 @@ func NewRollup(
 		cfg:             cfg,
 		signer:          types.LatestSignerForChainID(chainId),
 		externalRsaPriv: rsaPriv,
-		batchCache:      make(map[uint64]*eth.RPCRollupBatch),
+		batchCache:      NewBatchCache(),
 		ldb:             ldb,
 		bm:              bm,
 	}
@@ -256,7 +256,7 @@ func (r *Rollup) ProcessTx() error {
 		if ispending {
 			if txRecord.sendTime+uint64(r.cfg.TxTimeout.Seconds()) < uint64(time.Now().Unix()) {
 				log.Info("tx timeout", "tx", rtx.Hash().Hex(), "nonce", rtx.Nonce(), "method", method)
-				newtx, err := r.ReSubmitTx(false, rtx)
+				newtx, err := r.ReSubmitTx(false, rtx, r.GetReSubmitBatchIndex(method, rtx.Data()))
 				if err != nil {
 					log.Error("resubmit tx", "error", err, "tx", rtx.Hash().Hex(), "nonce", rtx.Nonce())
 					return fmt.Errorf("resubmit tx error:%w", err)
@@ -281,7 +281,8 @@ func (r *Rollup) ProcessTx() error {
 						"nonce", rtx.Nonce(),
 						"query_times", txRecord.queryTimes,
 					)
-					replacedtx, err := r.ReSubmitTx(true, rtx)
+
+					replacedtx, err := r.ReSubmitTx(true, rtx, r.GetReSubmitBatchIndex(method, rtx.Data()))
 					if err != nil {
 						log.Error("resend discarded tx", "old_tx", rtx.Hash().String(), "nonce", rtx.Nonce(), "error", err)
 						if utils.ErrStringMatch(err, core.ErrNonceTooLow) {
@@ -358,7 +359,7 @@ func (r *Rollup) ProcessTx() error {
 					}
 					r.metrics.SetRollupCost(fee)
 					index := utils.ParseParentBatchIndex(rtx.Data()) + 1
-					batch, ok := r.batchCache[index]
+					batch, ok := r.batchCache.Get(index)
 					if ok {
 						collectedL1FeeFloat := ToEtherFloat((*big.Int)(batch.CollectedL1Fee))
 						r.collectedL1FeeSum += collectedL1FeeFloat
@@ -368,7 +369,7 @@ func (r *Rollup) ProcessTx() error {
 						}
 						r.metrics.SetCollectedL1Fee(ToEtherFloat((*big.Int)(batch.CollectedL1Fee)))
 						// remove batch from cache
-						delete(r.batchCache, index)
+						r.batchCache.Delete(index)
 					} else {
 						log.Warn("batch not found in batchCache while set collect fee metrics",
 							"index", index,
@@ -527,7 +528,7 @@ func (r *Rollup) finalize() error {
 		"size", signedTx.Size(),
 	)
 
-	err = r.SendTx(signedTx)
+	err = r.SendTx(signedTx, 0)
 	if err != nil {
 		log.Error("send finalize tx to mempool", "error", err.Error())
 		if utils.ErrStringMatch(err, core.ErrNonceTooLow) {
@@ -650,9 +651,13 @@ func (r *Rollup) rollup() error {
 		return nil
 	}
 
-	batch, err := GetRollupBatchByIndex(batchIndex, r.L2Clients)
-	if err != nil {
-		return fmt.Errorf("get rollup batch by index err:%v", err)
+	var batch *eth.RPCRollupBatch
+	batch, ok := r.batchCache.Get(batchIndex)
+	if !ok {
+		batch, err = GetRollupBatchByIndex(batchIndex, r.L2Clients)
+		if err != nil {
+			return fmt.Errorf("get rollup batch by index err:%v", err)
+		}
 	}
 
 	// check if the batch is valid
@@ -668,7 +673,7 @@ func (r *Rollup) rollup() error {
 
 	// set batch cache
 	// it shoud be removed after the batch is committed
-	r.batchCache[batchIndex] = batch
+	r.batchCache.Set(batchIndex, batch)
 
 	signature, err := r.buildSignatureInput(batch)
 	if err != nil {
@@ -708,7 +713,7 @@ func (r *Rollup) rollup() error {
 		}
 
 		if r.cfg.RoughEstimateGas {
-			msgcnt := utils.ParseL1MessageCnt(batch.BlockContexts)
+			msgcnt := r.ParseL1MessageCnt(batch.BlockContexts)
 			gas = r.RoughRollupGasEstimate(msgcnt)
 			log.Info("rough estimate rollup tx gas", "gas", gas, "msgcnt", msgcnt)
 		} else {
@@ -786,7 +791,7 @@ func (r *Rollup) rollup() error {
 		"blob_len", len(signedTx.BlobHashes()),
 	)
 
-	err = r.SendTx(signedTx)
+	err = r.SendTx(signedTx, batchIndex)
 	if err != nil {
 		log.Error("send tx to mempool", "error", err.Error())
 		if utils.ErrStringMatch(err, core.ErrNonceTooLow) {
@@ -1091,7 +1096,7 @@ func UpdateGasLimit(tx *types.Transaction) (*types.Transaction, error) {
 }
 
 // send tx to l1 with business logic check
-func (r *Rollup) SendTx(tx *types.Transaction) error {
+func (r *Rollup) SendTx(tx *types.Transaction, batchIndex uint64) error {
 
 	// judge tx info is valid
 	if tx == nil {
@@ -1101,7 +1106,30 @@ func (r *Rollup) SendTx(tx *types.Transaction) error {
 	if !r.bm.IsGrowth() {
 		return fmt.Errorf("block not growth in %d blocks time", r.cfg.BlockNotIncreasedThreshold)
 	}
+	// check batch loss
+	if r.cfg.RollupLossControl {
+		// calc fee
+		fee := utils.CalcFeeForTx(tx)
+		// get batch
+		var collectedL1Fee *big.Int
+		batch, ok := r.batchCache.Get(batchIndex)
+		if ok {
+			collectedL1Fee = batch.CollectedL1Fee.ToInt()
+			if collectedL1Fee == nil {
+				collectedL1Fee = big.NewInt(0)
+			}
+		} else {
+			log.Warn("batch not found in cache when calc fee before SendTx", "batchIndex", batchIndex)
+			collectedL1Fee = big.NewInt(0)
+		}
+		// targetFee = fee * (100 + cfg.RotatorBuffer)/100
+		targetFee := new(big.Int).Mul(fee, big.NewInt(100+r.cfg.RotatorBuffer))
+		targetFee.Div(targetFee, big.NewInt(100))
 
+		if collectedL1Fee.Cmp(targetFee) < 0 {
+			return fmt.Errorf("tx fee exceed collectedL1Fee: targetFee=%v,fee=%v,collectedL1Fee:=%v", targetFee, fee, collectedL1Fee)
+		}
+	}
 	err := sendTx(r.L1Client, r.cfg.TxFeeLimit, tx)
 	if err != nil {
 		return err
@@ -1138,7 +1166,7 @@ func sendTx(client iface.Client, txFeeLimit uint64, tx *types.Transaction) error
 	return client.SendTransaction(context.Background(), tx)
 }
 
-func (r *Rollup) ReSubmitTx(resend bool, tx *types.Transaction) (*types.Transaction, error) {
+func (r *Rollup) ReSubmitTx(resend bool, tx *types.Transaction, batchIndex uint64) (*types.Transaction, error) {
 	if tx == nil {
 		return nil, errors.New("nil tx")
 	}
@@ -1229,7 +1257,7 @@ func (r *Rollup) ReSubmitTx(resend bool, tx *types.Transaction) (*types.Transact
 		return nil, fmt.Errorf("sign tx error:%w", err)
 	}
 	// send tx
-	err = r.SendTx(newTx)
+	err = r.SendTx(newTx, batchIndex)
 	if err != nil {
 		return nil, fmt.Errorf("send tx error:%w", err)
 	}
@@ -1335,3 +1363,11 @@ func (r *Rollup) InitFeeMetricsSum() error {
 	r.metrics.CollectedL1FeeSum.Add(r.collectedL1FeeSum)
 	return nil
 }
+
+func (r *Rollup) GetReSubmitBatchIndex(method string, calldta []byte) uint64 {
+	if method == "commitBatch" {
+		return utils.ParseBatchIndex(method, calldta)
+	} else {
+		return 0
+	}
+}
diff --git a/tx-submitter/services/utils.go b/tx-submitter/services/utils.go
@@ -2,9 +2,11 @@ package services
 
 import (
 	"crypto/sha256"
+	"encoding/binary"
 	"math/big"
 
 	"github.com/morph-l2/go-ethereum/common"
+	"github.com/morph-l2/go-ethereum/common/hexutil"
 	"github.com/morph-l2/go-ethereum/core/types"
 	"github.com/morph-l2/go-ethereum/crypto/kzg4844"
 	"github.com/morph-l2/go-ethereum/params"
@@ -73,3 +75,16 @@ func ToEtherFloat(weiAmt *big.Int) float64 {
 	return fEtherAmt
 
 }
+
+func (r *Rollup) ParseL1MessageCnt(blockContexts hexutil.Bytes) uint64 {
+
+	var l1msgcnt uint64
+	blockNum := binary.BigEndian.Uint16(blockContexts[:2])
+	remainingBz := blockContexts[2:]
+	for i := 0; i < int(blockNum); i++ {
+		l1msgcnt += uint64(binary.BigEndian.Uint16(remainingBz[58:60]))
+		remainingBz = remainingBz[60:]
+	}
+
+	return l1msgcnt
+}
diff --git a/tx-submitter/utils/config.go b/tx-submitter/utils/config.go
diff --git a/tx-submitter/utils/utils.go b/tx-submitter/utils/utils.go
diff --git a/tx-submitter/utils/utils_test.go b/tx-submitter/utils/utils_test.go

Original file line number	Diff line number	Diff line change
`@@ -69,6 +69,9 @@ func Main() func(ctx *cli.Context) error {`
`69`	`69`	`"rough_estimate_per_l1_msg", cfg.RollupTxGasPerL1Msg,`
`70`	`70`	`"log_level", cfg.LogLevel,`
`71`	`71`	`"leveldb_pathname", cfg.LeveldbPathName,`
	`72`	`+ "l1_stop_in_blocks", cfg.BlockNotIncreasedThreshold,`
	`73`	`+ "rollup_loss_control", cfg.RollupLossControl,`
	`74`	`+ "rolluo_loss_buffer", cfg.MaxRollupLossBuffer,`
`72`	`75`	`)`
`73`	`76`
`74`	`77`	`ctx, cancel := context.WithCancel(context.Background())`