fix(sender): graceful restart (#1567)

Co-authored-by: colinlyguo <[email protected]>

Author: colinlyguo

common/version/version.go

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

rollup/internal/controller/sender/sender.go

@@ -232,6 +232,12 @@ func (s *Sender) SendTransaction(contextID string, target *common.Address, data
 	}
 
 	if err := s.client.SendTransaction(s.ctx, signedTx); err != nil {
+		// Delete the transaction from the pending transaction table if it fails to send.
+		if updateErr := s.pendingTransactionOrm.DeleteTransactionByTxHash(s.ctx, signedTx.Hash()); updateErr != nil {
+			log.Error("failed to delete transaction", "tx hash", signedTx.Hash().String(), "from", s.transactionSigner.GetAddr().String(), "nonce", signedTx.Nonce(), "err", updateErr)
+			return common.Hash{}, fmt.Errorf("failed to delete transaction, err: %w", updateErr)
+		}
+
 		log.Error("failed to send tx", "tx hash", signedTx.Hash().String(), "from", s.transactionSigner.GetAddr().String(), "nonce", signedTx.Nonce(), "err", err)
 		// Check if contain nonce, and reset nonce
 		// only reset nonce when it is not from resubmit
@@ -458,6 +464,15 @@ func (s *Sender) createReplacingTransaction(tx *gethTypes.Transaction, baseFee,
 				blobGasFeeCap = maxBlobGasPrice
 			}
 
+			// Check if any fee cap is less than double
+			doubledTipCap := new(big.Int).Mul(originalGasTipCap, big.NewInt(2))
+			doubledFeeCap := new(big.Int).Mul(originalGasFeeCap, big.NewInt(2))
+			doubledBlobFeeCap := new(big.Int).Mul(originalBlobGasFeeCap, big.NewInt(2))
+			if gasTipCap.Cmp(doubledTipCap) < 0 || gasFeeCap.Cmp(doubledFeeCap) < 0 || blobGasFeeCap.Cmp(doubledBlobFeeCap) < 0 {
+				log.Error("gas fees must be at least double", "originalTipCap", originalGasTipCap, "currentTipCap", gasTipCap, "requiredTipCap", doubledTipCap, "originalFeeCap", originalGasFeeCap, "currentFeeCap", gasFeeCap, "requiredFeeCap", doubledFeeCap, "originalBlobFeeCap", originalBlobGasFeeCap, "currentBlobFeeCap", blobGasFeeCap, "requiredBlobFeeCap", doubledBlobFeeCap)
+				return nil, errors.New("gas fees must be at least double")
+			}
+
 			feeData.gasFeeCap = gasFeeCap
 			feeData.gasTipCap = gasTipCap
 			feeData.blobGasFeeCap = blobGasFeeCap
@@ -520,7 +535,7 @@ func (s *Sender) checkPendingTransaction() {
 			if receipt.BlockNumber.Uint64() <= confirmed {
 				if dbTxErr := s.db.Transaction(func(dbTX *gorm.DB) error {
 					// Update the status of the transaction to TxStatusConfirmed.
-					if updateErr := s.pendingTransactionOrm.UpdatePendingTransactionStatusByTxHash(s.ctx, originalTx.Hash(), types.TxStatusConfirmed, dbTX); updateErr != nil {
+					if updateErr := s.pendingTransactionOrm.UpdateTransactionStatusByTxHash(s.ctx, originalTx.Hash(), types.TxStatusConfirmed, dbTX); updateErr != nil {
 						log.Error("failed to update transaction status by tx hash", "hash", originalTx.Hash().String(), "sender meta", s.getSenderMeta(), "from", s.transactionSigner.GetAddr().String(), "nonce", originalTx.Nonce(), "err", updateErr)
 						return updateErr
 					}
@@ -595,7 +610,7 @@ func (s *Sender) checkPendingTransaction() {
 			// A corner case is that the transaction is inserted into the table but not sent to the chain, because the server is stopped in the middle.
 			// This case will be handled by the checkPendingTransaction function.
 			if dbTxErr := s.db.Transaction(func(dbTX *gorm.DB) error {
-				if updateErr := s.pendingTransactionOrm.UpdatePendingTransactionStatusByTxHash(s.ctx, originalTx.Hash(), types.TxStatusReplaced, dbTX); updateErr != nil {
+				if updateErr := s.pendingTransactionOrm.UpdateTransactionStatusByTxHash(s.ctx, originalTx.Hash(), types.TxStatusReplaced, dbTX); updateErr != nil {
 					return fmt.Errorf("failed to update status of transaction with hash %s to TxStatusReplaced, err: %w", newSignedTx.Hash().String(), updateErr)
 				}
 				if updateErr := s.pendingTransactionOrm.InsertPendingTransaction(s.ctx, txnToCheck.ContextID, s.getSenderMeta(), newSignedTx, blockNumber, dbTX); updateErr != nil {
@@ -608,6 +623,23 @@ func (s *Sender) checkPendingTransaction() {
 			}
 
 			if err := s.client.SendTransaction(s.ctx, newSignedTx); err != nil {
+				// SendTransaction failed, need to rollback the previous database changes
+				if rollbackErr := s.db.Transaction(func(tx *gorm.DB) error {
+					// Restore original transaction status back to pending
+					if updateErr := s.pendingTransactionOrm.UpdateTransactionStatusByTxHash(s.ctx, originalTx.Hash(), types.TxStatusPending, tx); updateErr != nil {
+						return fmt.Errorf("failed to rollback status of original transaction, err: %w", updateErr)
+					}
+					// Delete the new transaction that was inserted
+					if updateErr := s.pendingTransactionOrm.DeleteTransactionByTxHash(s.ctx, newSignedTx.Hash(), tx); updateErr != nil {
+						return fmt.Errorf("failed to delete new transaction, err: %w", updateErr)
+					}
+					return nil
+				}); rollbackErr != nil {
+					// Both SendTransaction and rollback failed
+					log.Error("failed to rollback database after SendTransaction failed", "tx hash", newSignedTx.Hash().String(), "from", s.transactionSigner.GetAddr().String(), "nonce", newSignedTx.Nonce(), "sendTxErr", err, "rollbackErr", rollbackErr)
+					return
+				}
+
 				log.Error("failed to send replacing tx", "tx hash", newSignedTx.Hash().String(), "from", s.transactionSigner.GetAddr().String(), "nonce", newSignedTx.Nonce(), "err", err)
 				return
 			}

rollup/internal/orm/orm_test.go

@@ -560,7 +560,7 @@ func TestPendingTransactionOrm(t *testing.T) {
 	err = pendingTransactionOrm.InsertPendingTransaction(context.Background(), "test", senderMeta, tx1, 0)
 	assert.NoError(t, err)
 
-	err = pendingTransactionOrm.UpdatePendingTransactionStatusByTxHash(context.Background(), tx0.Hash(), types.TxStatusReplaced)
+	err = pendingTransactionOrm.UpdateTransactionStatusByTxHash(context.Background(), tx0.Hash(), types.TxStatusReplaced)
 	assert.NoError(t, err)
 
 	txs, err := pendingTransactionOrm.GetPendingOrReplacedTransactionsBySenderType(context.Background(), senderMeta.Type, 2)
@@ -577,7 +577,7 @@ func TestPendingTransactionOrm(t *testing.T) {
 	assert.Equal(t, senderMeta.Address.String(), txs[1].SenderAddress)
 	assert.Equal(t, senderMeta.Type, txs[1].SenderType)
 
-	err = pendingTransactionOrm.UpdatePendingTransactionStatusByTxHash(context.Background(), tx1.Hash(), types.TxStatusConfirmed)
+	err = pendingTransactionOrm.UpdateTransactionStatusByTxHash(context.Background(), tx1.Hash(), types.TxStatusConfirmed)
 	assert.NoError(t, err)
 
 	txs, err = pendingTransactionOrm.GetPendingOrReplacedTransactionsBySenderType(context.Background(), senderMeta.Type, 2)
@@ -594,4 +594,17 @@ func TestPendingTransactionOrm(t *testing.T) {
 	status, err := pendingTransactionOrm.GetTxStatusByTxHash(context.Background(), tx0.Hash())
 	assert.NoError(t, err)
 	assert.Equal(t, types.TxStatusConfirmedFailed, status)
+
+	// Test DeleteTransactionByTxHash
+	err = pendingTransactionOrm.DeleteTransactionByTxHash(context.Background(), tx0.Hash())
+	assert.NoError(t, err)
+
+	// Verify the transaction is deleted
+	status, err = pendingTransactionOrm.GetTxStatusByTxHash(context.Background(), tx0.Hash())
+	assert.NoError(t, err)
+	assert.Equal(t, types.TxStatusUnknown, status) // Should return unknown status for deleted transaction
+
+	// Try to delete non-existent transaction
+	err = pendingTransactionOrm.DeleteTransactionByTxHash(context.Background(), common.HexToHash("0x123"))
+	assert.Error(t, err) // Should return error for non-existent transaction
 }

rollup/internal/orm/pending_transaction.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/scroll-tech/go-ethereum/common"
 	gethTypes "github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/log"
 	"gorm.io/gorm"
 
 	"scroll-tech/common/types"
@@ -150,8 +151,33 @@ func (o *PendingTransaction) InsertPendingTransaction(ctx context.Context, conte
 	return nil
 }
 
-// UpdatePendingTransactionStatusByTxHash updates the status of a transaction based on the transaction hash.
-func (o *PendingTransaction) UpdatePendingTransactionStatusByTxHash(ctx context.Context, hash common.Hash, status types.TxStatus, dbTX ...*gorm.DB) error {
+// DeleteTransactionByTxHash permanently deletes a transaction record from the database by transaction hash.
+// Using permanent delete (Unscoped) instead of soft delete to prevent database bloat, as repeated SendTransaction failures
+// could write a large number of transactions to the database.
+func (o *PendingTransaction) DeleteTransactionByTxHash(ctx context.Context, hash common.Hash, dbTX ...*gorm.DB) error {
+	db := o.db
+	if len(dbTX) > 0 && dbTX[0] != nil {
+		db = dbTX[0]
+	}
+	db = db.WithContext(ctx)
+	db = db.Model(&PendingTransaction{})
+
+	// Perform permanent delete by using Unscoped()
+	result := db.Where("hash = ?", hash.String()).Unscoped().Delete(&PendingTransaction{})
+	if result.Error != nil {
+		return fmt.Errorf("failed to delete transaction, err: %w", result.Error)
+	}
+	if result.RowsAffected == 0 {
+		return fmt.Errorf("no transaction found with hash: %s", hash.String())
+	}
+	if result.RowsAffected > 0 {
+		log.Warn("Successfully deleted transaction", "hash", hash.String())
+	}
+	return nil
+}
+
+// UpdateTransactionStatusByTxHash updates the status of a transaction based on the transaction hash.
+func (o *PendingTransaction) UpdateTransactionStatusByTxHash(ctx context.Context, hash common.Hash, status types.TxStatus, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
@@ -160,7 +186,7 @@ func (o *PendingTransaction) UpdatePendingTransactionStatusByTxHash(ctx context.
 	db = db.Model(&PendingTransaction{})
 	db = db.Where("hash = ?", hash.String())
 	if err := db.Update("status", status).Error; err != nil {
-		return fmt.Errorf("failed to UpdatePendingTransactionStatusByTxHash, txHash: %s, error: %w", hash, err)
+		return fmt.Errorf("failed to UpdateTransactionStatusByTxHash, txHash: %s, error: %w", hash, err)
 	}
 	return nil
 }