From ad52b364290ff196ebdd2eaaa3ecd89b3aff7816 Mon Sep 17 00:00:00 2001 From: Kashif Siddiqui Date: Thu, 2 Jul 2026 14:29:28 +0900 Subject: [PATCH 1/5] fix(txmgr): accept Hedera broadcast when pending nonce advances After a successful send, validateOnChainSequence only checked the mined (latest) nonce. On Hedera that count can lag mempool acceptance by several seconds, which triggered gas-bump retries on the same nonce. Also consult PendingSequenceAt so a tx accepted into the mempool is treated as successful even before latest advances. --- chains/txmgr/broadcaster.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index d99eba7..64b4b09 100644 --- a/chains/txmgr/broadcaster.go +++ b/chains/txmgr/broadcaster.go @@ -622,6 +622,17 @@ 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 + + // Hedera accepts txs into its mempool before the mined nonce (latest) advances. + // Pending reflects acceptance; latest can lag by several seconds. + nextSeqPending, err := eb.client.PendingSequenceAt(ctx, etx.FromAddress) + if err != nil { + return errType, err + } + if nextSeqPending.Int64() > txSeq.Int64() { + return multinode.Successful, nil + } + // Retrieve the latest mined sequence from on-chain nextSeqOnChain, err := eb.client.SequenceAt(ctx, etx.FromAddress, nil) if err != nil { From c0f8853a177b91388a7c817ab86c515b713ba0c1 Mon Sep 17 00:00:00 2001 From: Kashif Siddiqui Date: Fri, 3 Jul 2026 15:03:00 +0900 Subject: [PATCH 2/5] move to sleep and backoff approach --- chains/txmgr/broadcaster.go | 62 +++++++++++++++++++++++++++++-------- 1 file changed, 49 insertions(+), 13 deletions(-) diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index 64b4b09..b113b02 100644 --- a/chains/txmgr/broadcaster.go +++ b/chains/txmgr/broadcaster.go @@ -33,9 +33,17 @@ 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 after a successful send. + hederaDefaultSequencePollInterval = 10 * time.Second + + // hederaDefaultSequencePollRetries is the number of SequenceAt re-polls (after the initial check) + // before treating the broadcast as underpriced on Hedera. + hederaDefaultSequencePollRetries = 3 + // hederaChainType is the string representation of the Hedera chain type // Temporary solution until the Broadcaster is moved to the EVM code base hederaChainType = "hedera" @@ -623,18 +631,11 @@ 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 - // Hedera accepts txs into its mempool before the mined nonce (latest) advances. - // Pending reflects acceptance; latest can lag by several seconds. - nextSeqPending, err := eb.client.PendingSequenceAt(ctx, etx.FromAddress) - if err != nil { - return errType, err - } - if nextSeqPending.Int64() > txSeq.Int64() { - return multinode.Successful, nil - } - - // 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, + hederaDefaultSequencePollInterval, hederaDefaultSequencePollRetries, + ) if err != nil { return errType, err } @@ -664,6 +665,41 @@ 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, + pollInterval time.Duration, + maxPollRetries int, +) (SEQ, error) { + var nextSeqOnChain SEQ + for poll := 0; poll <= maxPollRetries; poll++ { + if poll > 0 { + lgr.Infow("Hedera mined sequence not yet advanced, waiting before re-check", + "delay", pollInterval, + "poll", poll, + "maxPolls", maxPollRetries, + ) + select { + case <-ctx.Done(): + return nextSeqOnChain, ctx.Err() + case <-time.After(pollInterval): + } + } + + var err error + nextSeqOnChain, err = eb.client.SequenceAt(ctx, fromAddress, nil) + if err != nil { + return nextSeqOnChain, err + } + if nextSeqOnChain.Int64() > txSeq.Int64() { + return nextSeqOnChain, nil + } + } + 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) { From b19d991f475af1887cfcc8bfb9aa89fd332579b3 Mon Sep 17 00:00:00 2001 From: Kashif Siddiqui Date: Fri, 3 Jul 2026 16:25:26 +0900 Subject: [PATCH 3/5] Add tests --- chains/txmgr/broadcaster.go | 28 ++++++- chains/txmgr/broadcaster_hedera_test.go | 97 +++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 2 deletions(-) create mode 100644 chains/txmgr/broadcaster_hedera_test.go diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index b113b02..bf8b1e0 100644 --- a/chains/txmgr/broadcaster.go +++ b/chains/txmgr/broadcaster.go @@ -674,6 +674,30 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterB maxPollRetries int, ) (SEQ, error) { var nextSeqOnChain SEQ + _, err := pollSequenceAtAfterBroadcast(ctx, lgr, txSeq.Int64(), pollInterval, maxPollRetries, + 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 +} + +// pollSequenceAtAfterBroadcast polls sequenceAt until the value exceeds txSeq, waiting pollInterval +// between attempts up to maxPollRetries times after the initial check. +func pollSequenceAtAfterBroadcast( + ctx context.Context, + lgr logger.SugaredLogger, + txSeq int64, + pollInterval time.Duration, + maxPollRetries int, + sequenceAt func(ctx context.Context) (int64, error), +) (int64, error) { + var nextSeqOnChain int64 for poll := 0; poll <= maxPollRetries; poll++ { if poll > 0 { lgr.Infow("Hedera mined sequence not yet advanced, waiting before re-check", @@ -689,11 +713,11 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterB } var err error - nextSeqOnChain, err = eb.client.SequenceAt(ctx, fromAddress, nil) + nextSeqOnChain, err = sequenceAt(ctx) if err != nil { return nextSeqOnChain, err } - if nextSeqOnChain.Int64() > txSeq.Int64() { + if nextSeqOnChain > txSeq { return nextSeqOnChain, nil } } diff --git a/chains/txmgr/broadcaster_hedera_test.go b/chains/txmgr/broadcaster_hedera_test.go new file mode 100644 index 0000000..4c8a6c7 --- /dev/null +++ b/chains/txmgr/broadcaster_hedera_test.go @@ -0,0 +1,97 @@ +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("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, 3, + 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, 3, + 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", func(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + start := time.Now() + + got, err := pollSequenceAtAfterBroadcast( + t.Context(), lgr, 85, pollInterval, 3, + 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(4), calls.Load()) + assert.GreaterOrEqual(t, time.Since(start), 3*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, 3, + func(context.Context) (int64, error) { + return 85, nil + }, + ) + + require.ErrorIs(t, err, context.Canceled) + }) +} From 1b094d60493bae56da4acaf65d13e0448b7684c8 Mon Sep 17 00:00:00 2001 From: Kashif Siddiqui Date: Tue, 7 Jul 2026 12:52:32 +0900 Subject: [PATCH 4/5] Inline Hedera sequence poll defaults in sequenceAtAfterBroadcastWithRetries. --- chains/txmgr/broadcaster.go | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index bf8b1e0..24873ab 100644 --- a/chains/txmgr/broadcaster.go +++ b/chains/txmgr/broadcaster.go @@ -632,10 +632,7 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) validateOnChainS txSeq := *etx.Sequence // Hedera can take several seconds before the mined nonce (latest) advances after a successful send. - nextSeqOnChain, err := eb.sequenceAtAfterBroadcastWithRetries( - ctx, lgr, etx.FromAddress, txSeq, - hederaDefaultSequencePollInterval, hederaDefaultSequencePollRetries, - ) + nextSeqOnChain, err := eb.sequenceAtAfterBroadcastWithRetries(ctx, lgr, etx.FromAddress, txSeq) if err != nil { return errType, err } @@ -670,11 +667,9 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterB lgr logger.SugaredLogger, fromAddress ADDR, txSeq SEQ, - pollInterval time.Duration, - maxPollRetries int, ) (SEQ, error) { var nextSeqOnChain SEQ - _, err := pollSequenceAtAfterBroadcast(ctx, lgr, txSeq.Int64(), pollInterval, maxPollRetries, + _, err := pollSequenceAtAfterBroadcast(ctx, lgr, txSeq.Int64(), hederaDefaultSequencePollInterval, hederaDefaultSequencePollRetries, func(ctx context.Context) (int64, error) { var err error nextSeqOnChain, err = eb.client.SequenceAt(ctx, fromAddress, nil) From 560bb813b45104c21ddebb35570e7a14b7adc81f Mon Sep 17 00:00:00 2001 From: Kashif Siddiqui Date: Wed, 8 Jul 2026 20:02:37 +0900 Subject: [PATCH 5/5] Make Hedera sequence polling opt-in via HederaBroadcastConfig. Polling is disabled unless txConfig implements the new optional interface with a positive timeout, preserving legacy single-check behavior for existing nodes. When enabled, poll every 2s by default until the configured overall timeout elapses instead of using a fixed retry count. --- chains/txmgr/broadcaster.go | 81 +++++++++++++++++-------- chains/txmgr/broadcaster_hedera_test.go | 33 +++++++--- chains/txmgr/types/config.go | 12 ++++ 3 files changed, 95 insertions(+), 31 deletions(-) diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index 24873ab..29630e1 100644 --- a/chains/txmgr/broadcaster.go +++ b/chains/txmgr/broadcaster.go @@ -37,12 +37,9 @@ const ( // when the mined sequence still has not advanced after polling. maxHederaBroadcastRetries = 3 - // hederaDefaultSequencePollInterval is the delay between Hedera SequenceAt re-polls after a successful send. - hederaDefaultSequencePollInterval = 10 * time.Second - - // hederaDefaultSequencePollRetries is the number of SequenceAt re-polls (after the initial check) - // before treating the broadcast as underpriced on Hedera. - hederaDefaultSequencePollRetries = 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 @@ -668,8 +665,9 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterB fromAddress ADDR, txSeq SEQ, ) (SEQ, error) { + pollInterval, pollTimeout := eb.hederaSequencePollConfig() var nextSeqOnChain SEQ - _, err := pollSequenceAtAfterBroadcast(ctx, lgr, txSeq.Int64(), hederaDefaultSequencePollInterval, hederaDefaultSequencePollRetries, + _, 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) @@ -682,32 +680,68 @@ func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) sequenceAtAfterB return nextSeqOnChain, err } -// pollSequenceAtAfterBroadcast polls sequenceAt until the value exceeds txSeq, waiting pollInterval -// between attempts up to maxPollRetries times after the initial check. +// 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, - maxPollRetries int, + pollTimeout time.Duration, sequenceAt func(ctx context.Context) (int64, error), ) (int64, error) { var nextSeqOnChain int64 - for poll := 0; poll <= maxPollRetries; poll++ { - if poll > 0 { - lgr.Infow("Hedera mined sequence not yet advanced, waiting before re-check", - "delay", pollInterval, - "poll", poll, - "maxPolls", maxPollRetries, - ) - select { - case <-ctx.Done(): - return nextSeqOnChain, ctx.Err() - case <-time.After(pollInterval): - } + 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): } - var err error nextSeqOnChain, err = sequenceAt(ctx) if err != nil { return nextSeqOnChain, err @@ -716,7 +750,6 @@ func pollSequenceAtAfterBroadcast( return nextSeqOnChain, nil } } - return nextSeqOnChain, nil } // Finds next transaction in the queue, assigns a sequence, and moves it to "in_progress" state ready for broadcast. diff --git a/chains/txmgr/broadcaster_hedera_test.go b/chains/txmgr/broadcaster_hedera_test.go index 4c8a6c7..9fa76e1 100644 --- a/chains/txmgr/broadcaster_hedera_test.go +++ b/chains/txmgr/broadcaster_hedera_test.go @@ -18,6 +18,24 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { 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() @@ -25,7 +43,7 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { start := time.Now() got, err := pollSequenceAtAfterBroadcast( - t.Context(), lgr, 85, pollInterval, 3, + t.Context(), lgr, 85, pollInterval, 20*time.Millisecond, func(context.Context) (int64, error) { calls.Add(1) return 86, nil @@ -46,7 +64,7 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { start := time.Now() got, err := pollSequenceAtAfterBroadcast( - t.Context(), lgr, 85, pollInterval, 3, + t.Context(), lgr, 85, pollInterval, 20*time.Millisecond, func(context.Context) (int64, error) { i := int(calls.Add(1)) - 1 return sequences[i], nil @@ -59,14 +77,15 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { assert.GreaterOrEqual(t, time.Since(start), 2*pollInterval) }) - t.Run("returns last sequence when it never advances", func(t *testing.T) { + 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, 3, + t.Context(), lgr, 85, pollInterval, pollTimeout, func(context.Context) (int64, error) { calls.Add(1) return 85, nil @@ -75,8 +94,8 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { require.NoError(t, err) assert.Equal(t, int64(85), got) - assert.Equal(t, int32(4), calls.Load()) - assert.GreaterOrEqual(t, time.Since(start), 3*pollInterval) + 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) { @@ -86,7 +105,7 @@ func TestPollSequenceAtAfterBroadcast(t *testing.T) { cancel() _, err := pollSequenceAtAfterBroadcast( - ctx, lgr, 85, pollInterval, 3, + ctx, lgr, 85, pollInterval, 20*time.Millisecond, func(context.Context) (int64, error) { return 85, nil }, diff --git a/chains/txmgr/types/config.go b/chains/txmgr/types/config.go index 1ab334b..01b35f6 100644 --- a/chains/txmgr/types/config.go +++ b/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 }