Compare commits

..

2 Commits

Author SHA1 Message Date
colin
968a396b5e refactor(sender): split signer address checks into gas-oracle and rollup-relayer (#1568)
Co-authored-by: colinlyguo <colinlyguo@users.noreply.github.com>
2024-12-06 13:22:39 +08:00
colin
fa2401c081 fix(sender): graceful restart (#1567)
Co-authored-by: colinlyguo <colinlyguo@users.noreply.github.com>
2024-12-03 00:51:01 +08:00
5 changed files with 91 additions and 27 deletions

View File

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

View File

@@ -71,28 +71,9 @@ type Layer2Relayer struct {
// NewLayer2Relayer will return a new instance of Layer2RelayerClient
func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm.DB, cfg *config.RelayerConfig, chainCfg *params.ChainConfig, initGenesis bool, serviceType ServiceType, reg prometheus.Registerer) (*Layer2Relayer, error) {
var gasOracleSender, commitSender, finalizeSender *sender.Sender
var err error
// check that all 3 signer addresses are different, because there will be a problem in managing nonce for different senders
gasOracleSenderAddr, err := addrFromSignerConfig(cfg.GasOracleSenderSignerConfig)
if err != nil {
return nil, fmt.Errorf("failed to parse addr from gas oracle signer config, err: %v", err)
}
commitSenderAddr, err := addrFromSignerConfig(cfg.CommitSenderSignerConfig)
if err != nil {
return nil, fmt.Errorf("failed to parse addr from commit sender config, err: %v", err)
}
finalizeSenderAddr, err := addrFromSignerConfig(cfg.FinalizeSenderSignerConfig)
if err != nil {
return nil, fmt.Errorf("failed to parse addr from finalize sender config, err: %v", err)
}
if gasOracleSenderAddr == commitSenderAddr || gasOracleSenderAddr == finalizeSenderAddr || commitSenderAddr == finalizeSenderAddr {
return nil, fmt.Errorf("gas oracle, commit, and finalize sender addresses must be different. Got: Gas Oracle=%s, Commit=%s, Finalize=%s",
gasOracleSenderAddr.Hex(), commitSenderAddr.Hex(), finalizeSenderAddr.Hex())
}
switch serviceType {
case ServiceTypeL2GasOracle:
gasOracleSender, err = sender.NewSender(ctx, cfg.SenderConfig, cfg.GasOracleSenderSignerConfig, "l2_relayer", "gas_oracle_sender", types.SenderTypeL2GasOracle, db, reg)
@@ -106,6 +87,18 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm.
}
case ServiceTypeL2RollupRelayer:
commitSenderAddr, err := addrFromSignerConfig(cfg.CommitSenderSignerConfig)
if err != nil {
return nil, fmt.Errorf("failed to parse addr from commit sender config, err: %v", err)
}
finalizeSenderAddr, err := addrFromSignerConfig(cfg.FinalizeSenderSignerConfig)
if err != nil {
return nil, fmt.Errorf("failed to parse addr from finalize sender config, err: %v", err)
}
if commitSenderAddr == finalizeSenderAddr {
return nil, fmt.Errorf("commit and finalize sender addresses must be different. Got: Commit=%s, Finalize=%s", commitSenderAddr.Hex(), finalizeSenderAddr.Hex())
}
commitSender, err = sender.NewSender(ctx, cfg.SenderConfig, cfg.CommitSenderSignerConfig, "l2_relayer", "commit_sender", types.SenderTypeCommitBatch, db, reg)
if err != nil {
return nil, fmt.Errorf("new commit sender failed, err: %w", err)

View File

@@ -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
}

View File

@@ -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
}

View File

@@ -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
}