Bci 3621/try new estimation for insufficient fund error instead of retry

.changeset/pretty-trees-smile.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+use new estimation for insufficient fund instead of retry to overcome gas spike #added

common/txmgr/broadcaster.go

@@ -64,6 +64,26 @@ var ErrTxRemoved = errors.New("tx removed")
 
 type ProcessUnstartedTxs[ADDR types.Hashable] func(ctx context.Context, fromAddress ADDR) (retryable bool, err error)
 
+type saveTryAgainAttemptParams[
+	CHAIN_ID types.ID,
+	HEAD types.Head[BLOCK_HASH],
+	ADDR types.Hashable,
+	TX_HASH types.Hashable,
+	BLOCK_HASH types.Hashable,
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+] struct {
+	etx                txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	attempt            txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	replacementAttempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	initialBroadcastAt time.Time
+	newFee             FEE
+	newFeeLimit        uint64
+	retry              int
+	txError            error
+	errType            client.SendTxReturnCode
+}
+
 // TransmitCheckerFactory creates a transmit checker based on a spec.
 type TransmitCheckerFactory[
 	CHAIN_ID types.ID,
@@ -558,21 +578,20 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) hand
 		eb.sequenceTracker.GenerateNextSequence(etx.FromAddress, *etx.Sequence)
 		return err, true
 	case client.Underpriced:
-		return eb.tryAgainBumpingGas(ctx, lgr, err, etx, attempt, initialBroadcastAt, retryCount+1)
+		return eb.tryAgainBumpingGas(ctx, lgr, err, errType, etx, attempt, initialBroadcastAt, retryCount+1)
 	case client.InsufficientFunds:
-		// NOTE: This bails out of the entire cycle and essentially "blocks" on
-		// any transaction that gets insufficient_funds. This is OK if a
-		// transaction with a large VALUE blocks because this always comes last
-		// in the processing list.
-		// If it blocks because of a transaction that is expensive due to large
-		// gas limit, we could have smaller transactions "above" it that could
-		// theoretically be sent, but will instead be blocked.
+		// NOTE:
+		// This can potentially happen during gas spike.
+		// We want to re-estimate the tx, and save the replaced attempt,
+		// and process it after the back off duration.
+		// This is done by tryAgainWithNewEstimation return retryable after saved the tx attempt,
+		// instead of calling handleInProgressTx() again
 		eb.SvcErrBuffer.Append(err)
-		fallthrough
+		return eb.tryAgainWithNewEstimation(ctx, lgr, err, errType, etx, attempt, initialBroadcastAt)
 	case client.Retryable:
 		return err, true
 	case client.FeeOutOfValidRange:
-		return eb.tryAgainWithNewEstimation(ctx, lgr, err, etx, attempt, initialBroadcastAt)
+		return eb.tryAgainWithNewEstimation(ctx, lgr, err, errType, etx, attempt, initialBroadcastAt)
 	case client.Unsupported:
 		return err, false
 	case client.ExceedsMaxFee:
@@ -680,7 +699,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) next
 	return etx, nil
 }
 
-func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryAgainBumpingGas(ctx context.Context, lgr logger.Logger, txError error, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time, retry int) (err error, retryable bool) {
+func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryAgainBumpingGas(ctx context.Context, lgr logger.Logger, txError error, errType client.SendTxReturnCode, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time, retry int) (err error, retryable bool) {
 	// This log error is not applicable to Hedera since the action required would not be needed for its gas estimator
 	if eb.chainType != hederaChainType {
 		logger.With(lgr,
@@ -698,33 +717,54 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryA
 	if err != nil {
 		return fmt.Errorf("tryAgainBumpFee failed: %w", err), retryable
 	}
-
-	return eb.saveTryAgainAttempt(ctx, lgr, etx, attempt, replacementAttempt, initialBroadcastAt, bumpedFee, bumpedFeeLimit, retry)
+	params := saveTryAgainAttemptParams[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{
+		etx:                etx,
+		attempt:            attempt,
+		replacementAttempt: replacementAttempt,
+		initialBroadcastAt: initialBroadcastAt,
+		newFee:             bumpedFee,
+		newFeeLimit:        bumpedFeeLimit,
+		retry:              0,
+		txError:            txError,
+		errType:            errType,
+	}
+	return eb.saveTryAgainAttempt(ctx, lgr, params)
 }
 
-func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryAgainWithNewEstimation(ctx context.Context, lgr logger.Logger, txError error, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time) (err error, retryable bool) {
-	if attempt.TxType == 0x2 {
-		err = fmt.Errorf("re-estimation is not supported for EIP-1559 transactions. Node returned error: %v. This is a bug", txError.Error())
-		logger.Sugared(eb.lggr).AssumptionViolation(err.Error())
-		return err, false
-	}
-
+func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryAgainWithNewEstimation(ctx context.Context, lgr logger.Logger, txError error, errType client.SendTxReturnCode, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time) (err error, retryable bool) {
 	replacementAttempt, fee, feeLimit, retryable, err := eb.NewTxAttemptWithType(ctx, etx, lgr, attempt.TxType, feetypes.OptForceRefetch)
 	if err != nil {
 		return fmt.Errorf("tryAgainWithNewEstimation failed to build new attempt: %w", err), retryable
 	}
 	lgr.Warnw("L2 rejected transaction due to incorrect fee, re-estimated and will try again",
 		"etxID", etx.ID, "err", err, "newGasPrice", fee, "newGasLimit", feeLimit)
 
-	return eb.saveTryAgainAttempt(ctx, lgr, etx, attempt, replacementAttempt, initialBroadcastAt, fee, feeLimit, 0)
+	params := saveTryAgainAttemptParams[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{
+		etx:                etx,
+		attempt:            attempt,
+		replacementAttempt: replacementAttempt,
+		initialBroadcastAt: initialBroadcastAt,
+		newFee:             fee,
+		newFeeLimit:        feeLimit,
+		retry:              0,
+		txError:            txError,
+		errType:            errType,
+	}
+	return eb.saveTryAgainAttempt(ctx, lgr, params)
 }
 
-func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) saveTryAgainAttempt(ctx context.Context, lgr logger.Logger, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], replacementAttempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time, newFee FEE, newFeeLimit uint64, retry int) (err error, retyrable bool) {
-	if err = eb.txStore.SaveReplacementInProgressAttempt(ctx, attempt, &replacementAttempt); err != nil {
+func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) saveTryAgainAttempt(ctx context.Context, lgr logger.Logger, param saveTryAgainAttemptParams[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) (err error, retyrable bool) {
+	if err = eb.txStore.SaveReplacementInProgressAttempt(ctx, param.attempt, &param.replacementAttempt); err != nil {
 		return fmt.Errorf("tryAgainWithNewFee failed: %w", err), true
 	}
-	lgr.Debugw("Bumped fee on initial send", "oldFee", attempt.TxFee.String(), "newFee", newFee.String(), "newFeeLimit", newFeeLimit)
-	return eb.handleInProgressTx(ctx, etx, replacementAttempt, initialBroadcastAt, retry)
+	lgr.Debugw("Bumped fee on initial send", "oldFee", param.attempt.TxFee.String(), "newFee", param.newFee.String(), "newFeeLimit", param.newFeeLimit)
+
+	// this avoids re-estimated insufficient fund tx gets processed immediately, we want to back off the tx when gas spikes
+	if param.errType == client.InsufficientFunds {
+		return param.txError, true
+	}
+
+	return eb.handleInProgressTx(ctx, param.etx, param.replacementAttempt, param.initialBroadcastAt, param.retry)
 }
 
 func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) saveFatallyErroredTransaction(lgr logger.Logger, etx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error {

core/chains/evm/txmgr/broadcaster_test.go

@@ -1524,7 +1524,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) {
 		pgtest.MustExec(t, db, `DELETE FROM evm.txes WHERE nonce = $1`, localNextNonce)
 	})
 
-	t.Run("eth tx is left in progress if eth node returns insufficient eth", func(t *testing.T) {
+	t.Run("eth tx is replaced with new re-estimated tx if eth node returns insufficient eth", func(t *testing.T) {
 		insufficientEthError := "insufficient funds for transfer"
 		localNextNonce := getLocalNextNonce(t, nonceTracker, fromAddress)
 		etx := mustCreateUnstartedTx(t, txStore, fromAddress, toAddress, encodedPayload, gasLimit, value, testutils.FixtureChainID)
@@ -1537,17 +1537,20 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) {
 		assert.Contains(t, err.Error(), "insufficient funds for transfer")
 		assert.True(t, retryable)
 
-		// Check it was saved correctly with its attempt
-		etx, err = txStore.FindTxWithAttempts(ctx, etx.ID)
+		// Check it was saved correctly with replaced attempt
+		updated_etx, err := txStore.FindTxWithAttempts(ctx, etx.ID)
 		require.NoError(t, err)
+		assert.Nil(t, updated_etx.BroadcastAt)
+		assert.Nil(t, updated_etx.InitialBroadcastAt)
+		require.NotNil(t, updated_etx.Sequence)
+		assert.False(t, updated_etx.Error.Valid)
+		assert.Equal(t, txmgrcommon.TxUnstarted, etx.State)
+		require.Len(t, updated_etx.TxAttempts, 1)
 
-		assert.Nil(t, etx.BroadcastAt)
-		assert.Nil(t, etx.InitialBroadcastAt)
-		require.NotNil(t, etx.Sequence)
-		assert.False(t, etx.Error.Valid)
-		assert.Equal(t, txmgrcommon.TxInProgress, etx.State)
-		require.Len(t, etx.TxAttempts, 1)
-		attempt := etx.TxAttempts[0]
+		// check new attempt created
+		assert.NotEqual(t, updated_etx.CreatedAt, etx.CreatedAt)
+
+		attempt := updated_etx.TxAttempts[0]
 		assert.Equal(t, txmgrtypes.TxAttemptInProgress, attempt.State)
 		assert.Nil(t, attempt.BroadcastBeforeBlockNum)
 	})