Skip to content

Upgrade error handler #408

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.

Sign up for GitHub

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

Draft
dimriou wants to merge 1 commit into
base: develop
Choose a base branch
from
Draft

Upgrade error handler #408

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
20 changes: 15 additions & 5 deletions pkg/txm/clientwrappers/dualbroadcast/meta_error_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@ package dualbroadcast
import (
"context"
"errors"
"fmt"

"github.com/ethereum/go-ethereum/common"

"github.com/smartcontractkit/chainlink-common/pkg/logger"
"github.com/smartcontractkit/chainlink-evm/pkg/txm"
"github.com/smartcontractkit/chainlink-evm/pkg/txm/types"
)
Expand All @@ -17,15 +17,25 @@ func NewErrorHandler() *errorHandler {
return &errorHandler{}
}

func (e *errorHandler) HandleError(ctx context.Context, tx *types.Transaction, txErr error, txStore txm.TxStore, setNonce func(common.Address, uint64), isFromBroadcastMethod bool) error {
func (e *errorHandler) HandleError(
ctx context.Context,
lggr logger.Logger,
tx *types.Transaction,
txErr error,
txStore txm.TxStore,
setNonce func(common.Address, uint64),
isFromBroadcastMethod bool,
) (noTransmission bool, err 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) || errors.Is(txErr, ErrAuction)) && tx.AttemptCount == 1 {
if err := txStore.MarkTxFatal(ctx, tx, tx.FromAddress); err != nil {
return err
return false, err
}
setNonce(tx.FromAddress, *tx.Nonce)
return fmt.Errorf("transaction with txID: %d marked as fatal", tx.ID)
lggr.Infof("transaction with txID: %d marked as fatal", tx.ID)
return true, nil
}

return txErr
// If the error message is not recognized, we don't know if we didn't transmit so return false and continue with the standard execution path.
return false, nil
}
28 changes: 17 additions & 11 deletions pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,10 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/zap"

"github.com/smartcontractkit/chainlink-common/pkg/logger"
"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
"github.com/smartcontractkit/chainlink-evm/pkg/assets"
"github.com/smartcontractkit/chainlink-evm/pkg/gas"
"github.com/smartcontractkit/chainlink-evm/pkg/testutils"
Expand All @@ -21,6 +23,7 @@ func TestMetaErrorHandler(t *testing.T) {
require.NotNil(t, errorHandler)

t.Run("handles no bids error for first attempt", func(t *testing.T) {
lggr, observedLogs := logger.TestObserved(t, zap.InfoLevel)
nonce := uint64(1)
address := testutils.NewAddress()
txRequest := &types.TxRequest{
Expand All @@ -29,7 +32,7 @@ func TestMetaErrorHandler(t *testing.T) {
ToAddress: testutils.NewAddress(),
}
setNonce := func(address common.Address, nonce uint64) {}
txStoreManager := storage.NewInMemoryStoreManager(logger.Test(t), testutils.FixtureChainID)
txStoreManager := storage.NewInMemoryStoreManager(lggr, testutils.FixtureChainID)
require.NoError(t, txStoreManager.Add(address))
txStore := txStoreManager.InMemoryStoreMap[address]
_ = txStore.CreateTransaction(txRequest)
Expand All @@ -44,9 +47,10 @@ func TestMetaErrorHandler(t *testing.T) {
_, 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)
require.ErrorContains(t, err, "transaction with txID: 0 marked as fatal")
noTransmission, err := errorHandler.HandleError(t.Context(), lggr, tx, ErrNoBids, txStoreManager, setNonce, false)
require.NoError(t, err)
require.True(t, noTransmission)
tests.AssertLogEventually(t, observedLogs, "transaction with txID: 0 marked as fatal")
_, unconfirmedCount := txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
assert.Equal(t, 0, unconfirmedCount)
})
Expand Down Expand Up @@ -78,14 +82,15 @@ func TestMetaErrorHandler(t *testing.T) {
_, 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)
noTransmission, err := errorHandler.HandleError(t.Context(), logger.Test(t), tx, txErr, txStoreManager, setNonce, false)
require.NoError(t, err)
require.False(t, noTransmission)
_, unconfirmedCount := txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
assert.Equal(t, 1, unconfirmedCount)
})

t.Run("handles auction error for first attempt", func(t *testing.T) {
lggr, observedLogs := logger.TestObserved(t, zap.InfoLevel)
nonce := uint64(1)
address := testutils.NewAddress()
txRequest := &types.TxRequest{
Expand All @@ -95,7 +100,7 @@ func TestMetaErrorHandler(t *testing.T) {
}
txErr := ErrAuction
setNonce := func(address common.Address, nonce uint64) {}
txStoreManager := storage.NewInMemoryStoreManager(logger.Test(t), testutils.FixtureChainID)
txStoreManager := storage.NewInMemoryStoreManager(lggr, testutils.FixtureChainID)
require.NoError(t, txStoreManager.Add(address))
txStore := txStoreManager.InMemoryStoreMap[address]
_ = txStore.CreateTransaction(txRequest)
Expand All @@ -110,9 +115,10 @@ func TestMetaErrorHandler(t *testing.T) {
_, 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")
noTransmission, err := errorHandler.HandleError(t.Context(), lggr, tx, txErr, txStoreManager, setNonce, false)
require.NoError(t, err)
require.True(t, noTransmission)
tests.AssertLogEventually(t, observedLogs, "transaction with txID: 0 marked as fatal")
_, unconfirmedCount := txStore.FetchUnconfirmedTransactionAtNonceWithCount(nonce)
assert.Equal(t, 0, unconfirmedCount)
})
Expand Down
18 changes: 13 additions & 5 deletions pkg/txm/txm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,9 @@ type AttemptBuilder interface {
}

type ErrorHandler interface {
HandleError(context.Context, *types.Transaction, error, TxStore, func(common.Address, uint64), bool) (err error)
// HandleError tries to decide if there was a successful transmission by parsing the error message. If it can't decide, it returns control to
// the standard execution path.
HandleError(context.Context, logger.Logger, *types.Transaction, error, TxStore, func(common.Address, uint64), bool) (noTransmission bool, err error)
}

type StuckTxDetector interface {
Expand Down Expand Up @@ -337,11 +339,17 @@ func (t *Txm) sendTransactionWithError(ctx context.Context, tx *types.Transactio
start := time.Now()
txErr := t.client.SendTransaction(ctx, tx, attempt)
t.lggr.Infow("Broadcasted attempt", "tx", tx, "attempt", attempt, "transactionLifecycleID", tx.GetTransactionLifecycleID(t.lggr), "duration", time.Since(start), "txErr", txErr)
if txErr != nil && t.errorHandler != nil {
if err = t.errorHandler.HandleError(ctx, tx, txErr, t.txStore, t.SetNonce, false); err != nil {
return
if txErr != nil {
// If ErrorHandler is set, try to decide if there was a successful transmission by parsing the error message. If there wasn't, we return early.
if t.errorHandler != nil {
noTransmission, hErr := t.errorHandler.HandleError(ctx, t.lggr, tx, txErr, t.txStore, t.SetNonce, false)
if hErr != nil {
return hErr
}
if noTransmission {
return nil
}
}
} else if txErr != nil {
pendingNonce, pErr := t.client.PendingNonceAt(ctx, fromAddress)
if pErr != nil {
return pErr
Expand Down
18 changes: 12 additions & 6 deletions pkg/txm/txm_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -512,16 +512,17 @@ func TestFlow_ErrorHandler(t *testing.T) {
t.Parallel()

client := txm.NewMockClient(t)
txStoreManager := storage.NewInMemoryStoreManager(logger.Test(t), testutils.FixtureChainID)
lggr, observedLogs := logger.TestObserved(t, zap.InfoLevel)
txStoreManager := storage.NewInMemoryStoreManager(lggr, testutils.FixtureChainID)
address := testutils.NewAddress()
require.NoError(t, txStoreManager.Add(address))
config := txm.Config{EIP1559: true, EmptyTxLimitDefault: 22000, RetryBlockThreshold: 0, BlockTime: 2 * time.Second}
mockEstimator := mocks.NewEvmFeeEstimator(t)
keystore := &keystest.FakeChainStore{}
attemptBuilder := txm.NewAttemptBuilder(func(address common.Address) *assets.Wei { return assets.NewWeiI(1) }, mockEstimator, keystore, 22000)
stuckTxDetector := txm.NewStuckTxDetector(logger.Test(t), "", txm.StuckTxDetectorConfig{BlockTime: config.BlockTime, StuckTxBlockThreshold: uint32(config.RetryBlockThreshold + 1)})
stuckTxDetector := txm.NewStuckTxDetector(lggr, "", txm.StuckTxDetectorConfig{BlockTime: config.BlockTime, StuckTxBlockThreshold: uint32(config.RetryBlockThreshold + 1)})
errorHandler := dualbroadcast.NewErrorHandler()
tm := txm.NewTxm(logger.Test(t), testutils.FixtureChainID, client, attemptBuilder, txStoreManager, stuckTxDetector, config, keystore, errorHandler)
tm := txm.NewTxm(lggr, testutils.FixtureChainID, client, attemptBuilder, txStoreManager, stuckTxDetector, config, keystore, errorHandler)
metrics, err := txm.NewTxmMetrics(testutils.FixtureChainID)
require.NoError(t, err)
tm.Metrics = metrics
Expand All @@ -544,8 +545,11 @@ func TestFlow_ErrorHandler(t *testing.T) {
Return(gas.EvmFee{DynamicFee: gas.DynamicFee{GasTipCap: assets.NewWeiI(5), GasFeeCap: assets.NewWeiI(10)}}, defaultGasLimit, nil).Once()
client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(dualbroadcast.ErrNoBids).Once()
_, err = tm.BroadcastTransaction(t.Context(), address)
require.Error(t, err)
require.ErrorContains(t, err, "transaction with txID: 0 marked as fatal")
require.NoError(t, err)
tests.AssertLogEventually(t, observedLogs, "transaction with txID: 0 marked as fatal")
_, count, err := txStoreManager.FetchUnconfirmedTransactionAtNonceWithCount(t.Context(), 0, address)
require.NoError(t, err)
require.Equal(t, 0, count)

// Create transaction 2
IDK2 := "IDK2"
Expand Down Expand Up @@ -575,9 +579,11 @@ func TestFlow_ErrorHandler(t *testing.T) {
mockEstimator.On("BumpFee", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
Return(gas.EvmFee{DynamicFee: gas.DynamicFee{GasTipCap: assets.NewWeiI(6), GasFeeCap: assets.NewWeiI(12)}}, defaultGasLimit, nil).Once()
client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(dualbroadcast.ErrNoBids).Once()
client.On("PendingNonceAt", mock.Anything, address).Return(initialNonce, nil).Once()
err = tm.BackfillTransactions(t.Context(), address) // retry
require.Error(t, err)
require.ErrorContains(t, err, dualbroadcast.ErrNoBids.Error())
require.ErrorContains(t, err, "pending nonce for txID: 1 didn't increase")
require.ErrorIs(t, err, dualbroadcast.ErrNoBids)
tx, count, err = txStoreManager.FetchUnconfirmedTransactionAtNonceWithCount(t.Context(), 0, address) // same transaction is still in the store
require.NoError(t, err)
require.Equal(t, 1, count)
Expand Down
Loading