Bugfix/ccip 11890 executor OOF metrics

chains/txmgr/broadcaster.go

@@ -53,7 +53,6 @@ type TransmitCheckerFactory[CID chains.ID, ADDR chains.Hashable, THASH, BHASH ch
 
 // TransmitChecker determines whether a transaction should be submitted on-chain.
 type TransmitChecker[CID chains.ID, ADDR chains.Hashable, THASH, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {
-
 	// Check the given transaction. If the transaction should not be sent, an error indicating why
 	// is returned. Errors should only be returned if the checker can confirm that a transaction
 	// should not be sent, other errors (for example connection or other unexpected errors) should
@@ -64,6 +63,7 @@ type TransmitChecker[CID chains.ID, ADDR chains.Hashable, THASH, BHASH chains.Ha
 type broadcasterMetrics interface {
 	IncrementNumBroadcastedTxs(ctx context.Context)
 	RecordTimeUntilTxBroadcast(ctx context.Context, duration float64)
+	IncrementNumInsufficientFundsForTx(ctx context.Context, fromAddress string)
 }
 
 // Broadcaster monitors txes for transactions that need to
@@ -552,6 +552,8 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) handleInProgress
 		// replace the current attempt, and retry after the backoff duration.
 		// The new attempt must be replaced immediately because of a database constraint.
 		eb.SvcErrBuffer.Append(err)
+		eb.metrics.IncrementNumInsufficientFundsForTx(ctx, etx.FromAddress.String())
+		lgr.Warnw("Transaction rejected due to insufficient funds at sending address, will retry", "fromAddress", etx.FromAddress.String(), "err", err)
 		if _, _, replaceErr := eb.replaceAttemptWithNewEstimation(ctx, lgr, etx, attempt); replaceErr != nil {
 			return replaceErr, true
 		}
@@ -679,7 +681,8 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) nextUnstartedTra
 func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) replaceAttemptWithBumpedGas(ctx context.Context, lgr logger.Logger, txError error, etx types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], attempt types.TxAttempt[CID, ADDR, THASH, BHASH, SEQ, FEE]) (replacedAttempt types.TxAttempt[CID, ADDR, THASH, BHASH, SEQ, FEE], retryable bool, err error) {
 	// 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,
+		logger.With(
+			lgr,
 			"sendError", txError,
 			"attemptFee", attempt.TxFee,
 			"maxGasPriceConfig", eb.feeConfig.MaxFeePrice(),

chains/txmgr/confirmer.go

@@ -37,6 +37,7 @@ type confimerMetrics interface {
 	IncrementNumConfirmedTxs(ctx context.Context, confirmedTransactions int)
 	RecordTimeUntilTxConfirmed(ctx context.Context, duration float64)
 	RecordBlocksUntilTxConfirmed(ctx context.Context, blocksElapsed float64)
+	IncrementNumInsufficientFundsForTx(ctx context.Context, fromAddress string)
 }
 
 // Confirmer is a broad service which performs four different tasks in sequence on every new longest chain
@@ -754,6 +755,7 @@ func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) handleInProgres
 		timeout := ec.dbConfig.DefaultQueryTimeout()
 		return ec.txStore.SaveConfirmedAttempt(ctx, timeout, &attempt, now)
 	case multinode.InsufficientFunds:
+		ec.metrics.IncrementNumInsufficientFundsForTx(ctx, etx.FromAddress.String())
 		timeout := ec.dbConfig.DefaultQueryTimeout()
 		return ec.txStore.SaveInsufficientFundsAttempt(ctx, timeout, &attempt, now)
 	case multinode.Successful:

metrics/txm.go

@@ -71,6 +71,10 @@ var (
 			float64(100),
 		},
 	}, []string{"chainID"})
+	promNumInsufficientFunds = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "tx_manager_insufficient_funds_tx_count",
+		Help: "Number of transaction broadcast attempts rejected by an RPC node because the sending address had insufficient funds. Increments on every retry while the address remains underfunded, so a sustained rate indicates the address needs topping up.",
+	}, []string{"chainID", "senderAddress"})
 )
 
 type GenericTXMMetrics interface {
@@ -81,17 +85,19 @@ type GenericTXMMetrics interface {
 	IncrementNumConfirmedTxs(ctx context.Context, confirmedTransactions int)
 	RecordTimeUntilTxConfirmed(ctx context.Context, duration float64)
 	RecordBlocksUntilTxConfirmed(ctx context.Context, blocksElapsed float64)
+	IncrementNumInsufficientFundsForTx(ctx context.Context, fromAddress string)
 }
 
 type txmMetrics struct {
-	chainID                string
-	numBroadcastedTxs      metric.Int64Counter
-	timeUntilBroadcast     metric.Float64Histogram
-	numGasBumps            metric.Int64Counter
-	gasBumpExceedsLimit    metric.Int64Counter
-	numConfirmedTxs        metric.Int64Counter
-	timeUntilTxConfirmed   metric.Float64Histogram
-	blocksUntilTxConfirmed metric.Float64Histogram
+	chainID                 string
+	numBroadcastedTxs       metric.Int64Counter
+	timeUntilBroadcast      metric.Float64Histogram
+	numGasBumps             metric.Int64Counter
+	gasBumpExceedsLimit     metric.Int64Counter
+	numConfirmedTxs         metric.Int64Counter
+	timeUntilTxConfirmed    metric.Float64Histogram
+	blocksUntilTxConfirmed  metric.Float64Histogram
+	numInsufficientFundsTxs metric.Int64Counter
 }
 
 func NewGenericTxmMetrics(chainID string) (GenericTXMMetrics, error) {
@@ -130,15 +136,21 @@ func NewGenericTxmMetrics(chainID string) (GenericTXMMetrics, error) {
 		return nil, fmt.Errorf("failed to register blocks until tx confirmed metric: %w", err)
 	}
 
+	numInsufficientFundsTxs, err := beholder.GetMeter().Int64Counter("tx_manager_insufficient_funds_tx_count")
+	if err != nil {
+		return nil, fmt.Errorf("failed to register insufficient funds txs metric: %w", err)
+	}
+
 	return &txmMetrics{
-		chainID:                chainID,
-		numBroadcastedTxs:      numBroadcastedTxs,
-		timeUntilBroadcast:     timeUntilBroadcast,
-		numGasBumps:            numGasBumps,
-		gasBumpExceedsLimit:    gasBumpExceedsLimit,
-		numConfirmedTxs:        numConfirmedTxs,
-		timeUntilTxConfirmed:   timeUntilTxConfirmed,
-		blocksUntilTxConfirmed: blocksUntilTxConfirmed,
+		chainID:                 chainID,
+		numBroadcastedTxs:       numBroadcastedTxs,
+		timeUntilBroadcast:      timeUntilBroadcast,
+		numGasBumps:             numGasBumps,
+		gasBumpExceedsLimit:     gasBumpExceedsLimit,
+		numConfirmedTxs:         numConfirmedTxs,
+		timeUntilTxConfirmed:    timeUntilTxConfirmed,
+		blocksUntilTxConfirmed:  blocksUntilTxConfirmed,
+		numInsufficientFundsTxs: numInsufficientFundsTxs,
 	}, nil
 }
 
@@ -176,3 +188,11 @@ func (m *txmMetrics) RecordBlocksUntilTxConfirmed(ctx context.Context, blocksEla
 	promBlocksUntilTxConfirmed.WithLabelValues(m.chainID).Observe(blocksElapsed)
 	m.blocksUntilTxConfirmed.Record(ctx, blocksElapsed, metric.WithAttributes(attribute.String("chainID", m.chainID)))
 }
+
+func (m *txmMetrics) IncrementNumInsufficientFundsForTx(ctx context.Context, fromAddress string) {
+	promNumInsufficientFunds.WithLabelValues(m.chainID, fromAddress).Add(1)
+	m.numInsufficientFundsTxs.Add(ctx, 1, metric.WithAttributes(
+		attribute.String("chainID", m.chainID),
+		attribute.String("senderAddress", fromAddress),
+	))
+}

metrics/txm_test.go

@@ -90,3 +90,16 @@ func TestTxmMetrics_RecordBlocksUntilTxConfirmed(t *testing.T) {
 		testutil.CollectAndCount(promBlocksUntilTxConfirmed),
 	)
 }
+
+func TestTxmMetrics_IncrementNumInsufficientFundsForTx(t *testing.T) {
+	m := setupTestTxmMetrics(t)
+
+	m.IncrementNumInsufficientFundsForTx(t.Context(), "0xSenderAddress")
+	m.IncrementNumInsufficientFundsForTx(t.Context(), "0xSenderAddress")
+
+	require.InEpsilon(t,
+		2.0,
+		testutil.ToFloat64(promNumInsufficientFunds.WithLabelValues("1", "0xSenderAddress")),
+		0.001,
+	)
+}