feat(taiko-client): introduce TxMgrSelector for proposer / prover (#…

…17986)

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

packages/taiko-client/internal/utils/txmgr_selector.go

@@ -0,0 +1,62 @@
+package utils
+
+import (
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+)
+
+var (
+	defaultPrivateTxMgrRetryInterval = 5 * time.Minute
+)
+
+// TxMgrSelector is responsible for selecting the correct transaction manager,
+// it will choose the transaction manager for a private mempool if it is available and works well,
+// otherwise it will choose the normal transaction manager.
+type TxMgrSelector struct {
+	txMgr                     *txmgr.SimpleTxManager
+	privateTxMgr              *txmgr.SimpleTxManager
+	privateTxMgrFailedAt      *time.Time
+	privateTxMgrRetryInterval time.Duration
+}
+
+// NewTxMgrSelector creates a new TxMgrSelector instance.
+func NewTxMgrSelector(
+	txMgr *txmgr.SimpleTxManager,
+	privateTxMgr *txmgr.SimpleTxManager,
+	privateTxMgrRetryInterval *time.Duration,
+) *TxMgrSelector {
+	retryInterval := defaultPrivateTxMgrRetryInterval
+	if privateTxMgrRetryInterval != nil {
+		retryInterval = *privateTxMgrRetryInterval
+	}
+
+	return &TxMgrSelector{
+		txMgr:                     txMgr,
+		privateTxMgr:              privateTxMgr,
+		privateTxMgrFailedAt:      nil,
+		privateTxMgrRetryInterval: retryInterval,
+	}
+}
+
+// Select selects a transaction manager based on the current state.
+func (s *TxMgrSelector) Select() (*txmgr.SimpleTxManager, bool) {
+	// If there is no private transaction manager, return the normal transaction manager.
+	if s.privateTxMgr == nil {
+		return s.txMgr, false
+	}
+
+	// If the private transaction manager has failed, check if it is time to retry.
+	if s.privateTxMgrFailedAt == nil || time.Now().After(s.privateTxMgrFailedAt.Add(s.privateTxMgrRetryInterval)) {
+		return s.privateTxMgr, true
+	}
+
+	// Otherwise, return the normal transaction manager.
+	return s.txMgr, false
+}
+
+// RecordPrivateTxMgrFailed records the time when the private transaction manager has failed.
+func (s *TxMgrSelector) RecordPrivateTxMgrFailed() {
+	now := time.Now()
+	s.privateTxMgrFailedAt = &now
+}

packages/taiko-client/internal/utils/txmgr_selector_test.go

@@ -0,0 +1,29 @@
+package utils
+
+import (
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	testTxMgr    = &txmgr.SimpleTxManager{}
+	testSelector = NewTxMgrSelector(testTxMgr, nil, nil)
+)
+
+func TestNewTxMgrSelector(t *testing.T) {
+	require.Equal(t, defaultPrivateTxMgrRetryInterval, testSelector.privateTxMgrRetryInterval)
+}
+
+func TestSelect(t *testing.T) {
+	txMgr, isPrivate := testSelector.Select()
+	require.NotNil(t, txMgr)
+	require.False(t, isPrivate)
+}
+
+func TestRecordPrivateTxMgrFailed(t *testing.T) {
+	require.Nil(t, testSelector.privateTxMgrFailedAt)
+	testSelector.RecordPrivateTxMgrFailed()
+	require.NotNil(t, testSelector.privateTxMgrFailedAt)
+}

packages/taiko-client/proposer/proposer.go

@@ -52,8 +52,7 @@ type Proposer struct {
 	lastProposedAt time.Time
 	totalEpochs    uint64
 
-	txmgr        *txmgr.SimpleTxManager
-	privateTxmgr *txmgr.SimpleTxManager
+	txmgrSelector *utils.TxMgrSelector
 
 	ctx context.Context
 	wg  sync.WaitGroup
@@ -94,10 +93,8 @@ func (p *Proposer) InitFromConfig(
 		return err
 	}
 
-	if txMgr != nil {
-		p.txmgr = txMgr
-	} else {
-		if p.txmgr, err = txmgr.NewSimpleTxManager(
+	if txMgr == nil {
+		if txMgr, err = txmgr.NewSimpleTxManager(
 			"proposer",
 			log.Root(),
 			&metrics.TxMgrMetrics,
@@ -106,22 +103,8 @@ func (p *Proposer) InitFromConfig(
 			return err
 		}
 	}
-	if privateTxMgr != nil {
-		p.privateTxmgr = privateTxMgr
-	} else {
-		if cfg.PrivateTxmgrConfigs != nil && len(cfg.PrivateTxmgrConfigs.L1RPCURL) > 0 {
-			if p.privateTxmgr, err = txmgr.NewSimpleTxManager(
-				"privateTxProposer",
-				log.Root(),
-				&metrics.TxMgrMetrics,
-				*cfg.PrivateTxmgrConfigs,
-			); err != nil {
-				return err
-			}
-		} else {
-			p.privateTxmgr = nil
-		}
-	}
+
+	p.txmgrSelector = utils.NewTxMgrSelector(txMgr, privateTxMgr, nil)
 
 	chainConfig := config.NewChainConfig(p.protocolConfigs)
 	p.chainConfig = chainConfig
@@ -410,20 +393,19 @@ func (p *Proposer) updateProposingTicker() {
 	p.proposingTimer = time.NewTimer(duration)
 }
 
+// sendTx is the internal function to send a transaction with a selected tx manager.
 func (p *Proposer) sendTx(ctx context.Context, txCandidate *txmgr.TxCandidate) error {
-	if p.privateTxmgr != nil {
-		receipt, err := p.privateTxmgr.Send(ctx, *txCandidate)
-		if err != nil || receipt.Status != types.ReceiptStatusSuccessful {
-			log.Warn("Failed to send TaikoL1.proposeBlock transaction by private tx manager",
-				"error", encoding.TryParsingCustomError(err),
-			)
-		} else {
-			return nil
-		}
-	}
-	receipt, err := p.txmgr.Send(ctx, *txCandidate)
+	txMgr, isPrivate := p.txmgrSelector.Select()
+	receipt, err := txMgr.Send(ctx, *txCandidate)
 	if err != nil {
-		log.Warn("Failed to send TaikoL1.proposeBlock transaction", "error", encoding.TryParsingCustomError(err))
+		log.Warn(
+			"Failed to send TaikoL1.proposeBlock / TaikoL1.proposeBlockV2 transaction by tx manager",
+			"isPrivateMempool", isPrivate,
+			"error", encoding.TryParsingCustomError(err),
+		)
+		if isPrivate {
+			p.txmgrSelector.RecordPrivateTxMgrFailed()
+		}
 		return err
 	}
 

packages/taiko-client/proposer/proposer_test.go

@@ -162,7 +162,8 @@ func (s *ProposerTestSuite) TestProposeTxLists() {
 	}
 
 	for _, txCandidate := range txCandidates {
-		receipt, err := p.txmgr.Send(ctx, txCandidate)
+		txMgr, _ := p.txmgrSelector.Select()
+		receipt, err := txMgr.Send(ctx, txCandidate)
 		s.Nil(err)
 		s.Nil(encoding.TryParsingCustomErrorFromReceipt(ctx, p.rpc.L1, p.proposerAddress, receipt))
 	}

packages/taiko-client/prover/proof_submitter/transaction/sender.go

@@ -14,15 +14,15 @@ import (
 
 	"github.com/taikoxyz/taiko-mono/packages/taiko-client/bindings/encoding"
 	"github.com/taikoxyz/taiko-mono/packages/taiko-client/internal/metrics"
+	"github.com/taikoxyz/taiko-mono/packages/taiko-client/internal/utils"
 	"github.com/taikoxyz/taiko-mono/packages/taiko-client/pkg/rpc"
 	producer "github.com/taikoxyz/taiko-mono/packages/taiko-client/prover/proof_producer"
 )
 
 // Sender is responsible for sending proof submission transactions with a backoff policy.
 type Sender struct {
 	rpc              *rpc.Client
-	txmgr            *txmgr.SimpleTxManager
-	privateTxmgr     *txmgr.SimpleTxManager
+	txmgrSelector    *utils.TxMgrSelector
 	proverSetAddress common.Address
 	gasLimit         uint64
 }
@@ -37,8 +37,7 @@ func NewSender(
 ) *Sender {
 	return &Sender{
 		rpc:              cli,
-		txmgr:            txmgr,
-		privateTxmgr:     privateTxmgr,
+		txmgrSelector:    utils.NewTxMgrSelector(txmgr, privateTxmgr, nil),
 		proverSetAddress: proverSetAddress,
 		gasLimit:         gasLimit,
 	}
@@ -78,38 +77,28 @@ func (s *Sender) Send(
 	}
 
 	// Send the transaction.
-	var (
-		next    = true
-		receipt *types.Receipt
-	)
-	if s.privateTxmgr != nil {
-		receipt, err = s.privateTxmgr.Send(ctx, *txCandidate)
-		if err != nil || receipt.Status != types.ReceiptStatusSuccessful {
-			log.Warn("Failed to send transaction by private tx manager in sender",
-				"error", encoding.TryParsingCustomError(err),
-			)
-		} else {
-			next = false
+	txMgr, isPrivate := s.txmgrSelector.Select()
+	receipt, err := txMgr.Send(ctx, *txCandidate)
+	if err != nil {
+		if isPrivate {
+			s.txmgrSelector.RecordPrivateTxMgrFailed()
 		}
+		return encoding.TryParsingCustomError(err)
 	}
-	if next {
-		receipt, err = s.txmgr.Send(ctx, *txCandidate)
-		if err != nil {
-			return encoding.TryParsingCustomError(err)
-		}
 
-		if receipt.Status != types.ReceiptStatusSuccessful {
-			log.Error(
-				"Failed to submit proof",
-				"blockID", proofWithHeader.BlockID,
-				"tier", proofWithHeader.Tier,
-				"txHash", receipt.TxHash,
-				"error", encoding.TryParsingCustomErrorFromReceipt(ctx, s.rpc.L1, s.txmgr.From(), receipt),
-			)
-			metrics.ProverSubmissionRevertedCounter.Add(1)
-			return ErrUnretryableSubmission
-		}
+	if receipt.Status != types.ReceiptStatusSuccessful {
+		log.Error(
+			"Failed to submit proof",
+			"blockID", proofWithHeader.BlockID,
+			"tier", proofWithHeader.Tier,
+			"txHash", receipt.TxHash,
+			"isPrivateMempool", isPrivate,
+			"error", encoding.TryParsingCustomErrorFromReceipt(ctx, s.rpc.L1, txMgr.From(), receipt),
+		)
+		metrics.ProverSubmissionRevertedCounter.Add(1)
+		return ErrUnretryableSubmission
 	}
+
 	log.Info(
 		"💰 Your block proof was accepted",
 		"blockID", proofWithHeader.BlockID,