Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(sender): graceful restart #1567

Merged
merged 9 commits into from
Dec 2, 2024
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion common/version/version.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
36 changes: 34 additions & 2 deletions rollup/internal/controller/sender/sender.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
colinlyguo marked this conversation as resolved.
Show resolved Hide resolved
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
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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 {
Expand All @@ -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
colinlyguo marked this conversation as resolved.
Show resolved Hide resolved
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
}
Expand Down
4 changes: 2 additions & 2 deletions rollup/internal/orm/orm_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down
29 changes: 26 additions & 3 deletions rollup/internal/orm/pending_transaction.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -150,8 +151,30 @@ 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 deletes a transaction record from the database by transaction hash.
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{})

result := db.Where("hash = ?", hash.String()).Delete(&PendingTransaction{})
if result.Error != nil {
return fmt.Errorf("failed to delete pending transaction, err: %w", result.Error)
}
if result.RowsAffected == 0 {
return fmt.Errorf("no pending transaction found with hash: %s", hash.String())
}
if result.RowsAffected > 0 {
log.Warn("Successfully deleted pending 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]
Expand All @@ -160,7 +183,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
}
Expand Down