diff --git a/chains/txmgr/broadcaster.go b/chains/txmgr/broadcaster.go index d99eba7..29630e1 100644 --- a/chains/txmgr/broadcaster.go +++ b/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) { diff --git a/chains/txmgr/broadcaster_hedera_test.go b/chains/txmgr/broadcaster_hedera_test.go new file mode 100644 index 0000000..9fa76e1 --- /dev/null +++ b/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) + }) +} 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 }