diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index d99eba7..38fc118 100644 --- a/chains/txmgr/broadcaster.go +++ b/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(), diff --git a/chains/txmgr/confirmer.go b/chains/txmgr/confirmer.go index 68146f6..5922fcf 100644 --- a/chains/txmgr/confirmer.go +++ b/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: diff --git a/metrics/txm.go b/metrics/txm.go index ae8be59..0c70b95 100644 --- a/metrics/txm.go +++ b/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), + )) +} diff --git a/metrics/txm_test.go b/metrics/txm_test.go index 4121cb8..b4b2d4e 100644 --- a/metrics/txm_test.go +++ b/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, + ) +}