Split MetaClient errors from standard RPC errors

pkg/txm/clientwrappers/dualbroadcast/meta_client.go

@@ -116,9 +116,14 @@ const (
 )
 
 var ErrNoBids = errors.New("no bids")
+var ErrAuction = errors.New("auction error")
 
 var _ txm.Client = &MetaClient{}
 
+type MetaClientTxStore interface {
+	UpdateSignedAttempt(_ context.Context, txID uint64, attemptID uint64, signedTransaction *evmtypes.Transaction, fromAddress common.Address) error
+}
+
 type MetaClientKeystore interface {
 	SignMessage(ctx context.Context, address common.Address, data []byte) ([]byte, error)
 	SignTx(ctx context.Context, fromAddress common.Address, tx *evmtypes.Transaction) (*evmtypes.Transaction, error)
@@ -137,9 +142,10 @@ type MetaClient struct {
 	customURL *url.URL
 	chainID   *big.Int
 	metrics   *MetaMetrics
+	txStore   MetaClientTxStore
 }
 
-func NewMetaClient(lggr logger.Logger, c MetaClientRPC, ks MetaClientKeystore, customURL *url.URL, chainID *big.Int) (*MetaClient, error) {
+func NewMetaClient(lggr logger.Logger, c MetaClientRPC, ks MetaClientKeystore, customURL *url.URL, chainID *big.Int, txStore MetaClientTxStore) (*MetaClient, error) {
 	metrics, err := NewMetaMetrics(chainID.String(), lggr)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create Meta metrics: %w", err)
@@ -152,6 +158,7 @@ func NewMetaClient(lggr logger.Logger, c MetaClientRPC, ks MetaClientKeystore, c
 		customURL: customURL,
 		chainID:   chainID,
 		metrics:   metrics,
+		txStore:   txStore,
 	}, nil
 }
 
@@ -163,30 +170,49 @@ func (a *MetaClient) PendingNonceAt(ctx context.Context, address common.Address)
 	return a.c.PendingNonceAt(ctx, address)
 }
 
+// SendTransactions handles three different cases:
+// 1. Auctions & Sends an attempt if it's a meta transaction and it hasn't broadcasted before.
+// 2. Sends the first attempt if it's a meta transaction and it has broadcasted before. This covers RPC errors.
+// 3. Sends an empty transaction to the mempool to clear the nonce.
 func (a *MetaClient) SendTransaction(ctx context.Context, tx *types.Transaction, attempt *types.Attempt) error {
 	meta, err := tx.GetMeta()
 	if err != nil {
 		return err
 	}
 
-	if meta != nil && meta.DualBroadcast != nil && *meta.DualBroadcast && !tx.IsPurgeable && meta.DualBroadcastParams != nil && meta.FwdrDestAddress != nil {
+	// #1
+	if meta != nil &&
+		meta.DualBroadcast != nil && *meta.DualBroadcast && meta.DualBroadcastParams != nil && meta.FwdrDestAddress != nil &&
+		tx.AttemptCount == 1 && !tx.IsPurgeable {
+		// Auction & Validate
 		meta, err := a.SendRequest(ctx, tx, attempt, *meta.DualBroadcastParams, tx.ToAddress)
 		if err != nil {
 			a.metrics.RecordSendRequestError(ctx)
 			a.metrics.emitAtlasError(ctx, "send_request", a.customURL, err, tx)
-			return fmt.Errorf("error sending request for transactionID(%d): %w", tx.ID, err)
+			return fmt.Errorf("error sending request for transactionID(%d): %w", tx.ID, errors.Join(err, ErrAuction))
 		}
+		// Send Metacall
 		if meta != nil {
 			if err := a.SendOperation(ctx, tx, attempt, *meta); err != nil {
 				a.metrics.RecordSendOperationError(ctx)
 				a.metrics.emitAtlasError(ctx, "send_operation", a.customURL, err, tx)
-				return fmt.Errorf("failed to send operation for transactionID(%d): %w", tx.ID, err)
+				return fmt.Errorf("failed to send operation for transactionID(%d): %w", tx.ID, errors.Join(err, ErrAuction))
 			}
 			return nil
 		}
 		a.lggr.Infof("No bids for transactionID(%d): ", tx.ID)
 		return ErrNoBids
 	}
+	// #2
+	if !tx.IsPurgeable && tx.AttemptCount > 1 && len(tx.Attempts) > 0 {
+		first := tx.Attempts[0]
+		if first.SignedTransaction != nil {
+			a.lggr.Infow("Intercepted attempt for tx(rebroadcasting first attempt)", "txID", tx.ID, "attempt", first)
+			return a.c.SendTransaction(ctx, first.SignedTransaction)
+		}
+	}
+
+	// #3
 	a.lggr.Infow("Broadcasting attempt to public mempool", "tx", tx)
 	return a.c.SendTransaction(ctx, attempt.SignedTransaction)
 }
@@ -527,6 +553,9 @@ func (a *MetaClient) SendOperation(ctx context.Context, tx *types.Transaction, a
 	if err != nil {
 		return fmt.Errorf("failed to sign attempt for txID: %v, err: %w", tx.ID, err)
 	}
+	if err := a.txStore.UpdateSignedAttempt(ctx, tx.ID, attempt.ID, signedTx, tx.FromAddress); err != nil {
+		return fmt.Errorf("failed to update signed attempt for txID: %v, err: %w", tx.ID, err)
+	}
 	a.lggr.Infow("Intercepted attempt for tx", "txID", tx.ID, "hash", signedTx.Hash(), "toAddress", meta.ToAddress, "gasLimit", meta.GasLimit,
 		"TipCap", tip, "FeeCap", meta.MaxFeePerGas)
 	return a.c.SendTransaction(ctx, signedTx)

pkg/txm/clientwrappers/dualbroadcast/meta_error_handler.go

@@ -19,7 +19,7 @@ func NewErrorHandler() *errorHandler {
 
 func (e *errorHandler) HandleError(ctx context.Context, tx *types.Transaction, txErr error, txStore txm.TxStore, setNonce func(common.Address, uint64), isFromBroadcastMethod bool) error {
 	// Mark the tx as fatal only if this is the first broadcast. In any other case, other txs might be included on-chain.
-	if errors.Is(txErr, ErrNoBids) && tx.AttemptCount == 1 {
+	if (errors.Is(txErr, ErrNoBids) || errors.Is(txErr, ErrAuction)) && tx.AttemptCount == 1 {
 		if err := txStore.MarkTxFatal(ctx, tx, tx.FromAddress); err != nil {
 			return err
 		}

pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go

@@ -41,7 +41,8 @@ func TestMetaErrorHandler(t *testing.T) {
 			GasLimit: 22000,
 			Hash:     testutils.NewHash(),
 		}
-		require.NoError(t, txStore.AppendAttemptToTransaction(*tx.Nonce, attempt))
+		_, err = txStore.AppendAttemptToTransaction(*tx.Nonce, attempt)
+		require.NoError(t, err)
 		tx, _ = txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
 		err = errorHandler.HandleError(t.Context(), tx, ErrNoBids, txStoreManager, setNonce, false)
 		require.Error(t, err)
@@ -72,13 +73,47 @@ func TestMetaErrorHandler(t *testing.T) {
 			GasLimit: 22000,
 			Hash:     testutils.NewHash(),
 		}
-		require.NoError(t, txStore.AppendAttemptToTransaction(*tx.Nonce, attempt))
-		require.NoError(t, txStore.AppendAttemptToTransaction(*tx.Nonce, attempt))
+		_, err = txStore.AppendAttemptToTransaction(*tx.Nonce, attempt)
+		require.NoError(t, err)
+		_, err = txStore.AppendAttemptToTransaction(*tx.Nonce, attempt)
+		require.NoError(t, err)
 		tx, _ = txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
 		err = errorHandler.HandleError(t.Context(), tx, txErr, txStoreManager, setNonce, false)
 		require.Error(t, err)
 		require.ErrorIs(t, err, txErr)
 		_, unconfirmedCount := txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
 		assert.Equal(t, 1, unconfirmedCount)
 	})
+
+	t.Run("handles auction error for first attempt", func(t *testing.T) {
+		nonce := uint64(1)
+		address := testutils.NewAddress()
+		txRequest := &types.TxRequest{
+			ChainID:     testutils.FixtureChainID,
+			FromAddress: address,
+			ToAddress:   testutils.NewAddress(),
+		}
+		txErr := ErrAuction
+		setNonce := func(address common.Address, nonce uint64) {}
+		txStoreManager := storage.NewInMemoryStoreManager(logger.Test(t), testutils.FixtureChainID)
+		require.NoError(t, txStoreManager.Add(address))
+		txStore := txStoreManager.InMemoryStoreMap[address]
+		_ = txStore.CreateTransaction(txRequest)
+		tx, err := txStore.UpdateUnstartedTransactionWithNonce(nonce)
+		require.NoError(t, err)
+		attempt := &types.Attempt{
+			TxID:     tx.ID,
+			Fee:      gas.EvmFee{GasPrice: assets.NewWeiI(1)},
+			GasLimit: 22000,
+			Hash:     testutils.NewHash(),
+		}
+		_, err = txStore.AppendAttemptToTransaction(*tx.Nonce, attempt)
+		require.NoError(t, err)
+		tx, _ = txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
+		err = errorHandler.HandleError(t.Context(), tx, txErr, txStoreManager, setNonce, false)
+		require.Error(t, err)
+		require.ErrorContains(t, err, "transaction with txID: 0 marked as fatal")
+		_, unconfirmedCount := txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
+		assert.Equal(t, 0, unconfirmedCount)
+	})
 }

pkg/txm/clientwrappers/dualbroadcast/selector.go

@@ -12,13 +12,13 @@ import (
 	"github.com/smartcontractkit/chainlink-evm/pkg/txm"
 )
 
-func SelectClient(lggr logger.Logger, client client.Client, keyStore keys.ChainStore, url *url.URL, chainID *big.Int) (txm.Client, txm.ErrorHandler, error) {
+func SelectClient(lggr logger.Logger, client client.Client, keyStore keys.ChainStore, url *url.URL, chainID *big.Int, txStore MetaClientTxStore) (txm.Client, txm.ErrorHandler, error) {
 	urlString := url.String()
 	switch {
 	case strings.Contains(urlString, "flashbots"):
 		return NewFlashbotsClient(client, keyStore, url), nil, nil
 	default:
-		mc, err := NewMetaClient(lggr, client, keyStore, url, chainID)
+		mc, err := NewMetaClient(lggr, client, keyStore, url, chainID, txStore)
 		if err != nil {
 			return nil, nil, err
 		}

pkg/txm/storage/inmemory_store.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	evmtypes "github.com/ethereum/go-ethereum/core/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-evm/pkg/txm/types"
@@ -74,17 +75,17 @@ func (m *InMemoryStore) AbandonPendingTransactions() {
 	m.UnconfirmedTransactions = make(map[uint64]*types.Transaction)
 }
 
-func (m *InMemoryStore) AppendAttemptToTransaction(txNonce uint64, attempt *types.Attempt) error {
+func (m *InMemoryStore) AppendAttemptToTransaction(txNonce uint64, attempt *types.Attempt) (attempts []*types.Attempt, err error) {
 	m.Lock()
 	defer m.Unlock()
 
 	tx, exists := m.UnconfirmedTransactions[txNonce]
 	if !exists {
-		return fmt.Errorf("unconfirmed tx was not found for nonce: %d - txID: %v", txNonce, attempt.TxID)
+		return nil, fmt.Errorf("unconfirmed tx was not found for nonce: %d - txID: %v", txNonce, attempt.TxID)
 	}
 
 	if tx.ID != attempt.TxID {
-		return fmt.Errorf("unconfirmed tx with nonce exists but attempt points to a different txID. Found Tx: %v - txID: %v", m.UnconfirmedTransactions[txNonce], attempt.TxID)
+		return nil, fmt.Errorf("unconfirmed tx with nonce exists but attempt points to a different txID. Found Tx: %v - txID: %v", tx, attempt.TxID)
 	}
 
 	attempt.CreatedAt = time.Now()
@@ -97,7 +98,11 @@ func (m *InMemoryStore) AppendAttemptToTransaction(txNonce uint64, attempt *type
 	}
 	m.UnconfirmedTransactions[txNonce].Attempts = append(m.UnconfirmedTransactions[txNonce].Attempts, attempt.DeepCopy())
 
-	return nil
+	attempts = make([]*types.Attempt, len(tx.Attempts))
+	for i, a := range tx.Attempts {
+		attempts[i] = a.DeepCopy()
+	}
+	return attempts, nil
 }
 
 func (m *InMemoryStore) CountUnstartedTransactions() int {
@@ -377,6 +382,24 @@ func (m *InMemoryStore) MarkTxFatal(txToMark *types.Transaction) error {
 	return nil
 }
 
+func (m *InMemoryStore) UpdateSignedAttempt(txID uint64, attemptID uint64, signedTransaction *evmtypes.Transaction) error {
+	m.Lock()
+	defer m.Unlock()
+
+	tx, exists := m.Transactions[txID]
+	if !exists {
+		return fmt.Errorf("tx was not found for txID: %v", txID)
+	}
+
+	for _, attempt := range tx.Attempts {
+		if attempt.ID == attemptID {
+			attempt.SignedTransaction = signedTransaction
+			return nil
+		}
+	}
+	return fmt.Errorf("attempt was not found for attemptID: %v", attemptID)
+}
+
 // Orchestrator
 func (m *InMemoryStore) FindTxWithIdempotencyKey(idempotencyKey string) *types.Transaction {
 	m.RLock()

pkg/txm/storage/inmemory_store_manager.go

@@ -7,6 +7,7 @@ import (
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
+	evmtypes "github.com/ethereum/go-ethereum/core/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-evm/pkg/txm/types"
@@ -46,11 +47,11 @@ func (m *InMemoryStoreManager) Add(addresses ...common.Address) (err error) {
 	return
 }
 
-func (m *InMemoryStoreManager) AppendAttemptToTransaction(_ context.Context, txNonce uint64, fromAddress common.Address, attempt *types.Attempt) error {
+func (m *InMemoryStoreManager) AppendAttemptToTransaction(_ context.Context, txNonce uint64, fromAddress common.Address, attempt *types.Attempt) (attempts []*types.Attempt, err error) {
 	if store, exists := m.InMemoryStoreMap[fromAddress]; exists {
 		return store.AppendAttemptToTransaction(txNonce, attempt)
 	}
-	return fmt.Errorf(StoreNotFoundForAddress, fromAddress)
+	return nil, fmt.Errorf(StoreNotFoundForAddress, fromAddress)
 }
 
 func (m *InMemoryStoreManager) CountUnstartedTransactions(fromAddress common.Address) (int, error) {
@@ -125,6 +126,13 @@ func (m *InMemoryStoreManager) MarkTxFatal(_ context.Context, tx *types.Transact
 	return fmt.Errorf(StoreNotFoundForAddress, fromAddress)
 }
 
+func (m *InMemoryStoreManager) UpdateSignedAttempt(_ context.Context, txID uint64, attemptID uint64, signedTransaction *evmtypes.Transaction, fromAddress common.Address) error {
+	if store, exists := m.InMemoryStoreMap[fromAddress]; exists {
+		return store.UpdateSignedAttempt(txID, attemptID, signedTransaction)
+	}
+	return fmt.Errorf(StoreNotFoundForAddress, fromAddress)
+}
+
 func (m *InMemoryStoreManager) FindTxWithIdempotencyKey(_ context.Context, idempotencyKey string) (*types.Transaction, error) {
 	for _, store := range m.InMemoryStoreMap {
 		tx := store.FindTxWithIdempotencyKey(idempotencyKey)