Cherry-Pick: Handle terminally stuck transactions on send (#1298)

Cherry-pick of smartcontractkit/chainlink#14127

Required to successfully handle zk overflows on Polygon zkEVM and X
Layer.

Co-authored-by: amit-momin <[email protected]>
Co-authored-by: Rens Rooimans <[email protected]>

.changeset/yellow-cougars-act.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Added client error classification for terminally stuck transactions in the TXM #internal

common/txmgr/broadcaster.go

@@ -478,16 +478,16 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) hand
 	lgr.Infow("Sending transaction", "txAttemptID", attempt.ID, "txHash", attempt.Hash, "meta", etx.Meta, "feeLimit", attempt.ChainSpecificFeeLimit, "callerProvidedFeeLimit", etx.FeeLimit, "attempt", attempt, "etx", etx)
 	errType, err := eb.client.SendTransactionReturnCode(ctx, etx, attempt, lgr)
 
-	if errType != client.Fatal {
-		etx.InitialBroadcastAt = &initialBroadcastAt
-		etx.BroadcastAt = &initialBroadcastAt
-	}
-
-	switch errType {
-	case client.Fatal:
+	if errType == client.Fatal || errType == client.TerminallyStuck {
 		eb.SvcErrBuffer.Append(err)
 		etx.Error = null.StringFrom(err.Error())
 		return eb.saveFatallyErroredTransaction(lgr, &etx), true
+	}
+
+	etx.InitialBroadcastAt = &initialBroadcastAt
+	etx.BroadcastAt = &initialBroadcastAt
+
+	switch errType {
 	case client.TransactionAlreadyKnown:
 		fallthrough
 	case client.Successful:

common/txmgr/confirmer.go

@@ -471,7 +471,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Pro
 		go func(tx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) {
 			defer wg.Done()
 			lggr := tx.GetLogger(ec.lggr)
-			// Create an purge attempt for tx
+			// Create a purge attempt for tx
 			purgeAttempt, err := ec.TxAttemptBuilder.NewPurgeTxAttempt(ctx, tx, lggr)
 			if err != nil {
 				errMu.Lock()
@@ -975,6 +975,21 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
 		ec.SvcErrBuffer.Append(sendError)
 		// This will loop continuously on every new head so it must be handled manually by the node operator!
 		return ec.txStore.DeleteInProgressAttempt(ctx, attempt)
+	case client.TerminallyStuck:
+		// A transaction could broadcast successfully but then be considered terminally stuck on another attempt
+		// Even though the transaction can succeed under different circumstances, we want to purge this transaction as soon as we get this error
+		lggr.Warnw("terminally stuck transaction detected", "err", sendError.Error())
+		ec.SvcErrBuffer.Append(sendError)
+		// Create a purge attempt for tx
+		purgeAttempt, err := ec.TxAttemptBuilder.NewPurgeTxAttempt(ctx, etx, lggr)
+		if err != nil {
+			return fmt.Errorf("NewPurgeTxAttempt failed: %w", err)
+		}
+		// Replace the in progress attempt with the purge attempt
+		if err := ec.txStore.SaveReplacementInProgressAttempt(ctx, attempt, &purgeAttempt); err != nil {
+			return fmt.Errorf("saveReplacementInProgressAttempt failed: %w", err)
+		}
+		return ec.handleInProgressAttempt(ctx, lggr, etx, purgeAttempt, blockHeight)
 	case client.TransactionAlreadyKnown:
 		// Sequence too low indicated that a transaction at this sequence was confirmed already.
 		// Mark confirmed_missing_receipt and wait for the next cycle to try to get a receipt

core/chains/evm/client/errors.go

@@ -583,6 +583,11 @@ func ClassifySendError(err error, clientErrors config.ClientErrors, lggr logger.
 		)
 		return commonclient.ExceedsMaxFee
 	}
+	if sendError.IsTerminallyStuckConfigError(configErrors) {
+		lggr.Warnw("Transaction that would have been terminally stuck in the mempool detected on send. Marking as fatal error.", "err", sendError, "etx", tx)
+		// Attempt is thrown away in this case; we don't need it since it never got accepted by a node
+		return commonclient.TerminallyStuck
+	}
 	lggr.Criticalw("Unknown error encountered when sending transaction", "err", err, "etx", tx)
 	return commonclient.Unknown
 }

core/chains/evm/client/errors_test.go

@@ -285,7 +285,7 @@ func Test_Eth_Errors(t *testing.T) {
 	})
 
 	t.Run("Metis gas price errors", func(t *testing.T) {
-		err := evmclient.NewSendErrorS("primary websocket (wss://ws-mainnet.metis.io) call failed: gas price too low: 18000000000 wei, use at least tx.gasPrice = 19500000000 wei")
+		err = evmclient.NewSendErrorS("primary websocket (wss://ws-mainnet.metis.io) call failed: gas price too low: 18000000000 wei, use at least tx.gasPrice = 19500000000 wei")
 		assert.True(t, err.L2FeeTooLow(clientErrors))
 		err = newSendErrorWrapped("primary websocket (wss://ws-mainnet.metis.io) call failed: gas price too low: 18000000000 wei, use at least tx.gasPrice = 19500000000 wei")
 		assert.True(t, err.L2FeeTooLow(clientErrors))
@@ -297,7 +297,7 @@ func Test_Eth_Errors(t *testing.T) {
 	})
 
 	t.Run("moonriver errors", func(t *testing.T) {
-		err := evmclient.NewSendErrorS("primary http (http://***REDACTED***:9933) call failed: submit transaction to pool failed: Pool(Stale)")
+		err = evmclient.NewSendErrorS("primary http (http://***REDACTED***:9933) call failed: submit transaction to pool failed: Pool(Stale)")
 		assert.True(t, err.IsNonceTooLowError(clientErrors))
 		assert.False(t, err.IsTransactionAlreadyInMempool(clientErrors))
 		assert.False(t, err.Fatal(clientErrors))
@@ -306,6 +306,24 @@ func Test_Eth_Errors(t *testing.T) {
 		assert.False(t, err.IsNonceTooLowError(clientErrors))
 		assert.False(t, err.Fatal(clientErrors))
 	})
+
+	t.Run("IsTerminallyStuck", func(t *testing.T) {
+		tests := []errorCase{
+			{"failed to add tx to the pool: not enough step counters to continue the execution", true, "zkEVM"},
+			{"failed to add tx to the pool: not enough step counters to continue the execution", true, "Xlayer"},
+			{"failed to add tx to the pool: not enough keccak counters to continue the execution", true, "zkEVM"},
+			{"failed to add tx to the pool: not enough keccak counters to continue the execution", true, "Xlayer"},
+		}
+
+		for _, test := range tests {
+			t.Run(test.network, func(t *testing.T) {
+				err = evmclient.NewSendErrorS(test.message)
+				assert.Equal(t, err.IsTerminallyStuckConfigError(clientErrors), test.expect)
+				err = newSendErrorWrapped(test.message)
+				assert.Equal(t, err.IsTerminallyStuckConfigError(clientErrors), test.expect)
+			})
+		}
+	})
 }
 
 func Test_Eth_Errors_Fatal(t *testing.T) {

core/chains/evm/txmgr/broadcaster_test.go

@@ -518,6 +518,25 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Success(t *testing.T) {
 			assert.True(t, ethTx.Error.Valid)
 			assert.Equal(t, "transaction reverted during simulation: json-rpc error { Code = 42, Message = 'oh no, it reverted', Data = 'KqYi' }", ethTx.Error.String)
 		})
+
+		t.Run("terminally stuck transaction is marked as fatal", func(t *testing.T) {
+			terminallyStuckError := "failed to add tx to the pool: not enough step counters to continue the execution"
+			etx := mustCreateUnstartedTx(t, txStore, fromAddress, toAddress, []byte{42, 42, 0}, gasLimit, big.Int(assets.NewEthValue(243)), testutils.FixtureChainID)
+			ethClient.On("SendTransactionReturnCode", mock.Anything, mock.MatchedBy(func(tx *gethTypes.Transaction) bool {
+				return tx.Nonce() == uint64(346) && tx.Value().Cmp(big.NewInt(243)) == 0
+			}), fromAddress).Return(commonclient.Fatal, errors.New(terminallyStuckError)).Once()
+
+			// Start processing unstarted transactions
+			retryable, err := eb.ProcessUnstartedTxs(tests.Context(t), fromAddress)
+			assert.NoError(t, err)
+			assert.False(t, retryable)
+
+			dbTx, err := txStore.FindTxWithAttempts(ctx, etx.ID)
+			require.NoError(t, err)
+			assert.Equal(t, txmgrcommon.TxFatalError, dbTx.State)
+			assert.True(t, dbTx.Error.Valid)
+			assert.Equal(t, terminallyStuckError, dbTx.Error.String)
+		})
 	})
 }
 

core/chains/evm/txmgr/confirmer_test.go

@@ -2656,6 +2656,58 @@ func TestEthConfirmer_RebroadcastWhereNecessary_WhenOutOfEth(t *testing.T) {
 	})
 }
 
+func TestEthConfirmer_RebroadcastWhereNecessary_TerminallyStuckError(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	cfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+		c.EVM[0].GasEstimator.PriceMax = assets.GWei(500)
+	})
+	txStore := cltest.NewTestTxStore(t, db)
+	ctx := tests.Context(t)
+
+	ethClient := testutils.NewEthClientMockWithDefaultChain(t)
+	ethKeyStore := cltest.NewKeyStore(t, db).Eth()
+	_, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore)
+
+	evmcfg := evmtest.NewChainScopedConfig(t, cfg)
+
+	// Use a mock keystore for this test
+	ec := newEthConfirmer(t, txStore, ethClient, cfg, evmcfg, ethKeyStore, nil)
+	currentHead := int64(30)
+	oldEnough := int64(19)
+	nonce := int64(0)
+	terminallyStuckError := "failed to add tx to the pool: not enough step counters to continue the execution"
+
+	t.Run("terminally stuck transaction replaced with purge attempt", func(t *testing.T) {
+		originalBroadcastAt := time.Unix(1616509100, 0)
+		etx := cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, nonce, fromAddress, originalBroadcastAt)
+		nonce++
+		attempt1_1 := etx.TxAttempts[0]
+		var dbAttempt txmgr.DbEthTxAttempt
+		require.NoError(t, db.Get(&dbAttempt, `UPDATE evm.tx_attempts SET broadcast_before_block_num=$1 WHERE id=$2 RETURNING *`, oldEnough, attempt1_1.ID))
+
+		// Return terminally stuck error on first rebroadcast
+		ethClient.On("SendTransactionReturnCode", mock.Anything, mock.MatchedBy(func(tx *types.Transaction) bool {
+			return tx.Nonce() == uint64(*etx.Sequence)
+		}), fromAddress).Return(commonclient.TerminallyStuck, errors.New(terminallyStuckError)).Once()
+		// Return successful for purge attempt
+		ethClient.On("SendTransactionReturnCode", mock.Anything, mock.MatchedBy(func(tx *types.Transaction) bool {
+			return tx.Nonce() == uint64(*etx.Sequence)
+		}), fromAddress).Return(commonclient.Successful, nil).Once()
+
+		// Start processing transactions for rebroadcast
+		require.NoError(t, ec.RebroadcastWhereNecessary(tests.Context(t), currentHead))
+		var err error
+		etx, err = txStore.FindTxWithAttempts(ctx, etx.ID)
+		require.NoError(t, err)
+
+		require.Len(t, etx.TxAttempts, 2)
+		purgeAttempt := etx.TxAttempts[0]
+		require.True(t, purgeAttempt.IsPurgeAttempt)
+	})
+}
+
 func TestEthConfirmer_EnsureConfirmedTransactionsInLongestChain(t *testing.T) {
 	t.Parallel()
 

core/chains/evm/txmgr/txmgr_test.go

@@ -747,6 +747,7 @@ func TestTxm_GetTransactionStatus(t *testing.T) {
 	t.Run("returns fatal for fatal error state with terminally stuck error", func(t *testing.T) {
 		idempotencyKey := uuid.New().String()
 		_, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore)
+		// Test the internal terminally stuck error returns Fatal
 		nonce := evmtypes.Nonce(0)
 		broadcast := time.Now()
 		tx := &txmgr.Tx{
@@ -765,6 +766,27 @@ func TestTxm_GetTransactionStatus(t *testing.T) {
 		state, err := txm.GetTransactionStatus(ctx, idempotencyKey)
 		require.Equal(t, commontypes.Fatal, state)
 		require.Error(t, err, evmclient.TerminallyStuckMsg)
+
+		// Test a terminally stuck client error returns Fatal
+		nonce = evmtypes.Nonce(1)
+		idempotencyKey = uuid.New().String()
+		terminallyStuckClientError := "failed to add tx to the pool: not enough step counters to continue the execution"
+		tx = &txmgr.Tx{
+			Sequence:           &nonce,
+			IdempotencyKey:     &idempotencyKey,
+			FromAddress:        fromAddress,
+			EncodedPayload:     []byte{1, 2, 3},
+			FeeLimit:           feeLimit,
+			State:              txmgrcommon.TxFatalError,
+			Error:              null.NewString(terminallyStuckClientError, true),
+			BroadcastAt:        &broadcast,
+			InitialBroadcastAt: &broadcast,
+		}
+		err = txStore.InsertTx(ctx, tx)
+		require.NoError(t, err)
+		state, err = txm.GetTransactionStatus(ctx, idempotencyKey)
+		require.Equal(t, commontypes.Fatal, state)
+		require.Error(t, err, evmclient.TerminallyStuckMsg)
 	})
 
 	t.Run("returns failed for fatal error state with other error", func(t *testing.T) {