fix: Prevent Hedera same-nonce spam retries for successful txs

chains/txmgr/broadcaster.go

@@ -33,9 +33,14 @@ const (
 	// spent on the transmit check.
 	TransmitCheckTimeout = 2 * time.Second
 
-	// maxBroadcastRetries is the number of times a transaction broadcast is retried when the sequence fails to increment on Hedera
+	// maxHederaBroadcastRetries is the number of times a transaction broadcast is retried with a bumped fee
+	// when the mined sequence still has not advanced after polling.
 	maxHederaBroadcastRetries = 3
 
+	// hederaDefaultSequencePollInterval is the delay between Hedera SequenceAt re-polls when polling is enabled
+	// and no interval is configured.
+	hederaDefaultSequencePollInterval = 2 * time.Second
+
 	// hederaChainType is the string representation of the Hedera chain type
 	// Temporary solution until the Broadcaster is moved to the EVM code base
 	hederaChainType = "hedera"
@@ -622,8 +627,9 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) validateOnChainS
 	}
 	// Transaction sequence cannot be nil here since a sequence is required to broadcast
 	txSeq := *etx.Sequence
-	// Retrieve the latest mined sequence from on-chain
-	nextSeqOnChain, err := eb.client.SequenceAt(ctx, etx.FromAddress, nil)
+
+	// Hedera can take several seconds before the mined nonce (latest) advances after a successful send.
+	nextSeqOnChain, err := eb.sequenceAtAfterBroadcastWithRetries(ctx, lgr, etx.FromAddress, txSeq)
 	if err != nil {
 		return errType, err
 	}
@@ -653,6 +659,99 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) validateOnChainS
 	return multinode.Successful, nil
 }
 
+func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterBroadcastWithRetries(
+	ctx context.Context,
+	lgr logger.SugaredLogger,
+	fromAddress ADDR,
+	txSeq SEQ,
+) (SEQ, error) {
+	pollInterval, pollTimeout := eb.hederaSequencePollConfig()
+	var nextSeqOnChain SEQ
+	_, err := pollSequenceAtAfterBroadcast(ctx, lgr, txSeq.Int64(), pollInterval, pollTimeout,
+		func(ctx context.Context) (int64, error) {
+			var err error
+			nextSeqOnChain, err = eb.client.SequenceAt(ctx, fromAddress, nil)
+			if err != nil {
+				return 0, err
+			}
+			return nextSeqOnChain.Int64(), nil
+		},
+	)
+	return nextSeqOnChain, err
+}
+
+// hederaSequencePollConfig returns polling settings for Hedera post-send sequence validation.
+// Polling is disabled unless txConfig implements HederaBroadcastConfig with a positive timeout.
+func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) hederaSequencePollConfig() (pollInterval time.Duration, pollTimeout time.Duration) {
+	if eb.chainType != hederaChainType {
+		return 0, 0
+	}
+	cfg, ok := eb.txConfig.(types.HederaBroadcastConfig)
+	if !ok {
+		return 0, 0
+	}
+	timeout := cfg.HederaSequencePollTimeout()
+	if timeout == nil || *timeout <= 0 {
+		return 0, 0
+	}
+
+	interval := hederaDefaultSequencePollInterval
+	if v := cfg.HederaSequencePollInterval(); v != nil && *v > 0 {
+		interval = *v
+	}
+	return interval, *timeout
+}
+
+// pollSequenceAtAfterBroadcast polls sequenceAt until the value exceeds txSeq or pollTimeout elapses.
+// When pollTimeout is zero, only a single immediate check is performed (legacy behavior).
+func pollSequenceAtAfterBroadcast(
+	ctx context.Context,
+	lgr logger.SugaredLogger,
+	txSeq int64,
+	pollInterval time.Duration,
+	pollTimeout time.Duration,
+	sequenceAt func(ctx context.Context) (int64, error),
+) (int64, error) {
+	var nextSeqOnChain int64
+	nextSeqOnChain, err := sequenceAt(ctx)
+	if err != nil {
+		return nextSeqOnChain, err
+	}
+	if nextSeqOnChain > txSeq || pollTimeout <= 0 {
+		return nextSeqOnChain, nil
+	}
+
+	deadline := time.Now().Add(pollTimeout)
+	for {
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			return nextSeqOnChain, nil
+		}
+
+		wait := pollInterval
+		if wait > remaining {
+			wait = remaining
+		}
+		lgr.Infow("Hedera mined sequence not yet advanced, waiting before re-check",
+			"delay", wait,
+			"remaining", remaining,
+		)
+		select {
+		case <-ctx.Done():
+			return nextSeqOnChain, ctx.Err()
+		case <-time.After(wait):
+		}
+
+		nextSeqOnChain, err = sequenceAt(ctx)
+		if err != nil {
+			return nextSeqOnChain, err
+		}
+		if nextSeqOnChain > txSeq {
+			return nextSeqOnChain, nil
+		}
+	}
+}
+
 // Finds next transaction in the queue, assigns a sequence, and moves it to "in_progress" state ready for broadcast.
 // Returns nil if no transactions are in queue
 func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) nextUnstartedTransactionWithSequence(fromAddress ADDR) (*types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], error) {

chains/txmgr/broadcaster_hedera_test.go

@@ -0,0 +1,116 @@
+package txmgr
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+func TestPollSequenceAtAfterBroadcast(t *testing.T) {
+	t.Parallel()
+
+	lgr := logger.Sugared(logger.Test(t))
+	pollInterval := 5 * time.Millisecond
+
+	t.Run("legacy single check when polling disabled", func(t *testing.T) {
+		t.Parallel()
+
+		var calls atomic.Int32
+
+		got, err := pollSequenceAtAfterBroadcast(
+			t.Context(), lgr, 85, pollInterval, 0,
+			func(context.Context) (int64, error) {
+				calls.Add(1)
+				return 85, nil
+			},
+		)
+
+		require.NoError(t, err)
+		assert.Equal(t, int64(85), got)
+		assert.Equal(t, int32(1), calls.Load())
+	})
+
+	t.Run("returns immediately when sequence already advanced", func(t *testing.T) {
+		t.Parallel()
+
+		var calls atomic.Int32
+		start := time.Now()
+
+		got, err := pollSequenceAtAfterBroadcast(
+			t.Context(), lgr, 85, pollInterval, 20*time.Millisecond,
+			func(context.Context) (int64, error) {
+				calls.Add(1)
+				return 86, nil
+			},
+		)
+
+		require.NoError(t, err)
+		assert.Equal(t, int64(86), got)
+		assert.Equal(t, int32(1), calls.Load())
+		assert.Less(t, time.Since(start), pollInterval)
+	})
+
+	t.Run("waits and polls until sequence advances", func(t *testing.T) {
+		t.Parallel()
+
+		sequences := []int64{85, 85, 86}
+		var calls atomic.Int32
+		start := time.Now()
+
+		got, err := pollSequenceAtAfterBroadcast(
+			t.Context(), lgr, 85, pollInterval, 20*time.Millisecond,
+			func(context.Context) (int64, error) {
+				i := int(calls.Add(1)) - 1
+				return sequences[i], nil
+			},
+		)
+
+		require.NoError(t, err)
+		assert.Equal(t, int64(86), got)
+		assert.Equal(t, int32(3), calls.Load())
+		assert.GreaterOrEqual(t, time.Since(start), 2*pollInterval)
+	})
+
+	t.Run("returns last sequence when it never advances before timeout", func(t *testing.T) {
+		t.Parallel()
+
+		var calls atomic.Int32
+		start := time.Now()
+		pollTimeout := 15 * time.Millisecond
+
+		got, err := pollSequenceAtAfterBroadcast(
+			t.Context(), lgr, 85, pollInterval, pollTimeout,
+			func(context.Context) (int64, error) {
+				calls.Add(1)
+				return 85, nil
+			},
+		)
+
+		require.NoError(t, err)
+		assert.Equal(t, int64(85), got)
+		assert.GreaterOrEqual(t, calls.Load(), int32(2))
+		assert.LessOrEqual(t, time.Since(start), pollTimeout+pollInterval)
+	})
+
+	t.Run("returns context error while waiting", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+
+		_, err := pollSequenceAtAfterBroadcast(
+			ctx, lgr, 85, pollInterval, 20*time.Millisecond,
+			func(context.Context) (int64, error) {
+				return 85, nil
+			},
+		)
+
+		require.ErrorIs(t, err, context.Canceled)
+	})
+}

chains/txmgr/types/config.go

@@ -34,6 +34,18 @@ type BroadcasterTransactionsConfig interface {
 	MaxInFlight() uint32
 }
 
+// HederaBroadcastConfig is an optional interface implemented by txConfig.
+// When HederaSequencePollTimeout is unset or zero, the broadcaster keeps legacy behavior:
+// a single immediate SequenceAt check after a successful send with no polling backoff.
+type HederaBroadcastConfig interface {
+	// HederaSequencePollTimeout is the total time to wait for the mined nonce to advance.
+	// Nil or zero disables polling (legacy behavior).
+	HederaSequencePollTimeout() *time.Duration
+	// HederaSequencePollInterval is the delay between SequenceAt checks while polling.
+	// Nil uses the framework default. Ignored when polling is disabled.
+	HederaSequencePollInterval() *time.Duration
+}
+
 type BroadcasterListenerConfig interface {
 	FallbackPollInterval() time.Duration
 }