feat: limit number of pending blob-carrying transactions (#1442)

Co-authored-by: colinlyguo <colinlyguo@users.noreply.github.com>
Co-authored-by: NazariiDenha <NazariiDenha@users.noreply.github.com>

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "v4.4.38"
+var tag = "v4.4.39"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {

rollup/conf/config.json

@@ -41,7 +41,8 @@
         "max_blob_gas_price": 10000000000000,
         "tx_type": "DynamicFeeTx",
         "check_pending_time": 1,
-        "min_gas_tip": 100000000
+        "min_gas_tip": 100000000,
+        "max_pending_blob_txs": 3
       },
       "gas_oracle_config": {
         "min_gas_price": 0,

rollup/internal/config/relayer.go

@@ -32,6 +32,8 @@ type SenderConfig struct {
 	MaxBlobGasPrice uint64 `json:"max_blob_gas_price"`
 	// The transaction type to use: LegacyTx, DynamicFeeTx, BlobTx
 	TxType string `json:"tx_type"`
+	// The maximum number of pending blob-carrying transactions
+	MaxPendingBlobTxs int64 `json:"max_pending_blob_txs"`
 }
 
 // ChainMonitor this config is used to get batch status from chain_monitor API.

rollup/internal/controller/relayer/l2_relayer.go

@@ -414,6 +414,15 @@ func (r *Layer2Relayer) ProcessPendingBatches() {
 
 		txHash, err := r.commitSender.SendTransaction(dbBatch.Hash, &r.cfg.RollupContractAddress, calldata, blob, fallbackGasLimit)
 		if err != nil {
+			if errors.Is(err, sender.ErrTooManyPendingBlobTxs) {
+				r.metrics.rollupL2RelayerProcessPendingBatchErrTooManyPendingBlobTxsTotal.Inc()
+				log.Debug(
+					"Skipped sending commitBatch tx to L1: too many pending blob txs",
+					"maxPending", r.cfg.SenderConfig.MaxPendingBlobTxs,
+					"err", err,
+				)
+				return
+			}
 			log.Error(
 				"Failed to send commitBatch tx to layer1",
 				"index", dbBatch.Index,

rollup/internal/controller/relayer/l2_relayer_metrics.go

@@ -8,26 +8,27 @@ import (
 )
 
 type l2RelayerMetrics struct {
-	rollupL2RelayerProcessPendingBatchTotal                     prometheus.Counter
-	rollupL2RelayerProcessPendingBatchSuccessTotal              prometheus.Counter
-	rollupL2RelayerGasPriceOraclerRunTotal                      prometheus.Counter
-	rollupL2RelayerLastGasPrice                                 prometheus.Gauge
-	rollupL2RelayerProcessCommittedBatchesTotal                 prometheus.Counter
-	rollupL2RelayerProcessCommittedBatchesFinalizedTotal        prometheus.Counter
-	rollupL2RelayerProcessCommittedBatchesFinalizedSuccessTotal prometheus.Counter
-	rollupL2BatchesCommittedConfirmedTotal                      prometheus.Counter
-	rollupL2BatchesCommittedConfirmedFailedTotal                prometheus.Counter
-	rollupL2BatchesFinalizedConfirmedTotal                      prometheus.Counter
-	rollupL2BatchesFinalizedConfirmedFailedTotal                prometheus.Counter
-	rollupL2UpdateGasOracleConfirmedTotal                       prometheus.Counter
-	rollupL2UpdateGasOracleConfirmedFailedTotal                 prometheus.Counter
-	rollupL2ChainMonitorLatestFailedCall                        prometheus.Counter
-	rollupL2ChainMonitorLatestFailedBatchStatus                 prometheus.Counter
-	rollupL2RelayerProcessPendingBundlesTotal                   prometheus.Counter
-	rollupL2RelayerProcessPendingBundlesFinalizedTotal          prometheus.Counter
-	rollupL2RelayerProcessPendingBundlesFinalizedSuccessTotal   prometheus.Counter
-	rollupL2BundlesFinalizedConfirmedTotal                      prometheus.Counter
-	rollupL2BundlesFinalizedConfirmedFailedTotal                prometheus.Counter
+	rollupL2RelayerProcessPendingBatchTotal                         prometheus.Counter
+	rollupL2RelayerProcessPendingBatchSuccessTotal                  prometheus.Counter
+	rollupL2RelayerProcessPendingBatchErrTooManyPendingBlobTxsTotal prometheus.Counter
+	rollupL2RelayerGasPriceOraclerRunTotal                          prometheus.Counter
+	rollupL2RelayerLastGasPrice                                     prometheus.Gauge
+	rollupL2RelayerProcessCommittedBatchesTotal                     prometheus.Counter
+	rollupL2RelayerProcessCommittedBatchesFinalizedTotal            prometheus.Counter
+	rollupL2RelayerProcessCommittedBatchesFinalizedSuccessTotal     prometheus.Counter
+	rollupL2BatchesCommittedConfirmedTotal                          prometheus.Counter
+	rollupL2BatchesCommittedConfirmedFailedTotal                    prometheus.Counter
+	rollupL2BatchesFinalizedConfirmedTotal                          prometheus.Counter
+	rollupL2BatchesFinalizedConfirmedFailedTotal                    prometheus.Counter
+	rollupL2UpdateGasOracleConfirmedTotal                           prometheus.Counter
+	rollupL2UpdateGasOracleConfirmedFailedTotal                     prometheus.Counter
+	rollupL2ChainMonitorLatestFailedCall                            prometheus.Counter
+	rollupL2ChainMonitorLatestFailedBatchStatus                     prometheus.Counter
+	rollupL2RelayerProcessPendingBundlesTotal                       prometheus.Counter
+	rollupL2RelayerProcessPendingBundlesFinalizedTotal              prometheus.Counter
+	rollupL2RelayerProcessPendingBundlesFinalizedSuccessTotal       prometheus.Counter
+	rollupL2BundlesFinalizedConfirmedTotal                          prometheus.Counter
+	rollupL2BundlesFinalizedConfirmedFailedTotal                    prometheus.Counter
 }
 
 var (
@@ -46,6 +47,10 @@ func initL2RelayerMetrics(reg prometheus.Registerer) *l2RelayerMetrics {
 				Name: "rollup_layer2_process_pending_batch_success_total",
 				Help: "The total number of layer2 process pending success batch",
 			}),
+			rollupL2RelayerProcessPendingBatchErrTooManyPendingBlobTxsTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+				Name: "rollup_layer2_process_pending_batch_err_too_many_pending_blob_txs_total",
+				Help: "The total number of layer2 process pending batch failed on too many pending blob txs",
+			}),
 			rollupL2RelayerGasPriceOraclerRunTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 				Name: "rollup_layer2_gas_price_oracler_total",
 				Help: "The total number of layer2 gas price oracler run total",

rollup/internal/controller/sender/sender.go

@@ -39,6 +39,11 @@ const (
 	DynamicFeeTxType = "DynamicFeeTx"
 )
 
+var (
+	// ErrTooManyPendingBlobTxs
+	ErrTooManyPendingBlobTxs = errors.New("the limit of pending blob-carrying transactions has been exceeded")
+)
+
 // Confirmation struct used to indicate transaction confirmation details
 type Confirmation struct {
 	ContextID    string
@@ -180,6 +185,23 @@ func (s *Sender) SendTransaction(contextID string, target *common.Address, data
 	)
 
 	if blob != nil {
+		// check that number of pending blob-carrying txs is not too big
+		if s.senderType == types.SenderTypeCommitBatch {
+			var numPendingTransactions int64
+			// We should count here only blob-carrying txs, but due to check that blob != nil, we know that we already switched to blobs.
+			// Now all txs with SenderTypeCommitBatch will be blob-carrying, but some of previous pending txs could still be non-blob.
+			// But this can happen only once at the moment of switching from non-blob to blob (pre-Bernoulli and post-Bernoulli) and it doesn't break anything.
+			// So don't need to add check that tx carries blob
+			numPendingTransactions, err = s.pendingTransactionOrm.GetCountPendingTransactionsBySenderType(s.ctx, s.senderType)
+			if err != nil {
+				log.Error("failed to count pending transactions", "err: %w", err)
+				return common.Hash{}, fmt.Errorf("failed to count pending transactions, err: %w", err)
+			}
+			if numPendingTransactions >= s.config.MaxPendingBlobTxs {
+				return common.Hash{}, ErrTooManyPendingBlobTxs
+			}
+
+		}
 		sidecar, err = makeSidecar(blob)
 		if err != nil {
 			log.Error("failed to make sidecar for blob transaction", "error", err)

rollup/internal/controller/sender/sender_test.go

@@ -121,7 +121,6 @@ func setupEnv(t *testing.T) {
 
 func TestSender(t *testing.T) {
 	setupEnv(t)
-
 	t.Run("test new sender", testNewSender)
 	t.Run("test send and retrieve transaction", testSendAndRetrieveTransaction)
 	t.Run("test fallback gas limit", testFallbackGasLimit)
@@ -137,6 +136,7 @@ func TestSender(t *testing.T) {
 	t.Run("test check pending transaction replaced tx confirmed", testCheckPendingTransactionReplacedTxConfirmed)
 	t.Run("test check pending transaction multiple times with only one transaction pending", testCheckPendingTransactionTxMultipleTimesWithOnlyOneTxPending)
 	t.Run("test blob transaction with blobhash op contract call", testBlobTransactionWithBlobhashOpContractCall)
+	t.Run("test test send blob-carrying tx over limit", testSendBlobCarryingTxOverLimit)
 }
 
 func testNewSender(t *testing.T) {
@@ -881,3 +881,22 @@ func randFieldElement() [32]byte {
 
 	return gokzg4844.SerializeScalar(r)
 }
+
+func testSendBlobCarryingTxOverLimit(t *testing.T) {
+	cfgCopy := *cfg.L2Config.RelayerConfig.SenderConfig
+	cfgCopy.TxType = "DynamicFeeTx"
+
+	sqlDB, err := db.DB()
+	assert.NoError(t, err)
+	assert.NoError(t, migrate.ResetDB(sqlDB))
+	s, err := NewSender(context.Background(), &cfgCopy, privateKey, "test", "test", types.SenderTypeCommitBatch, db, nil)
+	assert.NoError(t, err)
+
+	for i := 0; i < int(cfgCopy.MaxPendingBlobTxs); i++ {
+		_, err = s.SendTransaction("0", &common.Address{}, nil, randBlob(), 0)
+		assert.NoError(t, err)
+	}
+	_, err = s.SendTransaction("0", &common.Address{}, nil, randBlob(), 0)
+	assert.ErrorIs(t, err, ErrTooManyPendingBlobTxs)
+	s.Stop()
+}

rollup/internal/orm/pending_transaction.go

@@ -85,6 +85,19 @@ func (o *PendingTransaction) GetPendingOrReplacedTransactionsBySenderType(ctx co
 	return transactions, nil
 }
 
+// GetCountPendingTransactionsBySenderType retrieves number of pending transactions filtered by sender type
+func (o *PendingTransaction) GetCountPendingTransactionsBySenderType(ctx context.Context, senderType types.SenderType) (int64, error) {
+	var count int64
+	db := o.db.WithContext(ctx)
+	db = db.Model(&PendingTransaction{})
+	db = db.Where("sender_type = ?", senderType)
+	db = db.Where("status = ?", types.TxStatusPending)
+	if err := db.Count(&count).Error; err != nil {
+		return 0, fmt.Errorf("failed to count pending transactions by sender type, error: %w", err)
+	}
+	return count, nil
+}
+
 // GetConfirmedTransactionsBySenderType retrieves confirmed transactions filtered by sender type, limited to a specified count.
 // for unit test
 func (o *PendingTransaction) GetConfirmedTransactionsBySenderType(ctx context.Context, senderType types.SenderType, limit int) ([]PendingTransaction, error) {