diff --git a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go index 40b08fc3e3..8a759970e8 100644 --- a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go +++ b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go @@ -226,6 +226,14 @@ func (a *MetaClient) SendTransaction(ctx context.Context, tx *types.Transaction, // #2 if !tx.IsPurgeable && tx.AttemptCount > 1 && len(tx.Attempts) > 0 { first := tx.Attempts[0] + // The first attempt is the only one signed with the auction's metacall payload. If it was pruned + // from the store, the remaining attempts carry the raw payload, which must not reach the public + // mempool, so refuse to rebroadcast. Ideally, the transaction will be purged by the stuck transaction + // detector before this condition is met, but even if it doesn't, eventually it will still be purged and + // the error will resolve. + if first.ID != 0 { + return fmt.Errorf("first attempt for transactionID(%d) was pruned, refusing to rebroadcast attemptID(%d)", tx.ID, first.ID) + } if first.SignedTransaction != nil { a.lggr.Infow("Intercepted attempt for tx(rebroadcasting first attempt)", "txID", tx.ID, "attempt", first) return a.c.SendTransaction(ctx, nil, first) diff --git a/pkg/txm/clientwrappers/dualbroadcast/meta_client_test.go b/pkg/txm/clientwrappers/dualbroadcast/meta_client_test.go index 802ac2c06c..b7c56e18e3 100644 --- a/pkg/txm/clientwrappers/dualbroadcast/meta_client_test.go +++ b/pkg/txm/clientwrappers/dualbroadcast/meta_client_test.go @@ -3,15 +3,40 @@ package dualbroadcast_test import ( "encoding/hex" "encoding/json" + "math/big" + "net/url" "testing" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/stretchr/testify/require" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + + "github.com/smartcontractkit/chainlink-evm/pkg/txm" "github.com/smartcontractkit/chainlink-evm/pkg/txm/clientwrappers/dualbroadcast" + "github.com/smartcontractkit/chainlink-evm/pkg/txm/types" ) +func TestMetaClient_SendTransaction_RefusesPrunedFirstAttempt(t *testing.T) { + customURL, err := url.Parse("https://test.url") + require.NoError(t, err) + client, err := dualbroadcast.NewMetaClient(logger.Test(t), nil, nil, customURL, big.NewInt(1337), nil, nil, txm.NewNoopTxmMetrics()) + require.NoError(t, err) + + nonce := uint64(0) + tx := &types.Transaction{ + ID: 1, + Nonce: &nonce, + AttemptCount: 11, + // The original first attempt (ID 0) was pruned from the store; the oldest remaining attempt + // carries the raw payload and must not be rebroadcast. + Attempts: []*types.Attempt{{ID: 1, TxID: 1}}, + } + err = client.SendTransaction(t.Context(), tx, tx.Attempts[0]) + require.ErrorContains(t, err, "was pruned") +} + func TestMetaClient_VerifyResponse(t *testing.T) { responseData := []byte(`{"jsonrpc":"2.0","result":{"userOperation":{"from":"0xb6065f79d99f29c3eda0ed1bda7ff88e7ee12f1e","to":"0x1b4cb47622705f0f67b6b18bbd1aa1a91fc77d37","value":"0x0","gas":"0x186a00","maxFeePerGas":"0x3b9aca00","nonce":"0xf","deadline":"0x9c834e6","dapp":"0xc38d38333687ea295753c214744e839eddc7aebb","control":"0xc38d38333687ea295753c214744e839eddc7aebb","callConfig":"0x2304","dappGasLimit":"0x1e8480","solverGasLimit":"0x5b8d80","bundlerSurchargeRate":"0x0","sessionKey":"0x0000000000000000000000000000000000000000","data":"0x02a688ed0000000000000000000000008ae79bb7cce2dc3d132c288971b0f74af02a3b4a000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000003646fadcf72000000000000000000000000b123c2e3a71b57f9678cf6212576f30d642ed89a000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000002e4ba0cb29e0001aeec9ec2ddb59123858e897b26f52e2173aca2c577dcce62e7ddc70f16020000000000000000000000000000000000000000000000000000000000418a059c756341541f2283d4f124ac39dc9e718f96453582f38acfa885e4ffaf26eb7800000000000000000000000000000000000000000000000000000000000000e0000000000000000000000000000000000000000000000000000000000000022000000000000000000000000000000000000000000000000000000000000002800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000012000000000000000000000000000000000000000000000000000000000678010b601000203000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000040000000000000000000000000000000000000000000000000000000075542db800000000000000000000000000000000000000000000000000000000755c8e4000000000000000000000000000000000000000000000000000000000756106c80000000000000000000000000000000000000000000000000000000075961a520000000000000000000000000000000000000000000000000000000000000002b974b422c8e712d3320994bf4ecd31e37b35a117ef4102285206c45aef4b8709ae407e7a604410ef1e622ebdb4ad302638ff0cf49ad5ab33486c206bf8ecd7dd0000000000000000000000000000000000000000000000000000000000000002671dd95393e5963a3842a5f95d5e6b544de4906b1c304d41aebbb7c2acf9cb065fb96b17223872656de8f05a17ffb6ccc3261b606179cc33db445158aa408b700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000","signature":"0xb6ae5c173d3c7577ea3b8b4d35a0dc33e0f2798d74173c5e6fee2f08b8ce20271287a87443e778a114890140a08be5faf05ebcdd59051f9747fdd35d0afe6aa01b"},"solverOperations":[{"from":"0x377136653944bdd5d9f0db22987b7432e76c354f","to":"0x1b4cb47622705f0f67b6b18bbd1aa1a91fc77d37","value":"0x0","gas":"0x30d40","maxFeePerGas":"0x3b9aca00","deadline":"0x9c834e6","solver":"0x4b548c6faf4a3c74571c3e194e71cbf2a5172501","control":"0xc38d38333687ea295753c214744e839eddc7aebb","userOpHash":"0x0648e7c2a3b705562efd0e06adb8a63916ba5a06ee92a8544369a431298bce47","bidToken":"0x0000000000000000000000000000000000000000","bidAmount":"0x186a0","data":"0xc000a702000000000000000000000000000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000326a27250470f3a518db1334faaa277f8282082d7f54fd9b0f6cfff8f403a74fcb27a12a3860a1146cf7306eb49aa9fa03ccb10000000000000000000000000000","signature":"0x91537f835005fcd243b52b7b6436df810a1ea0b18769b2b1417ee9dacc9f9dac309a8b9ef82d82d6572e5feeb7209d12739067afd315434001a2aadd35af77701c"}],"dAppOperation":{"from":"0x1f3c5ec2ef75a9e1e09b1d46f208669e81000ee6","to":"0x1b4cb47622705f0f67b6b18bbd1aa1a91fc77d37","nonce":"0x0","deadline":"0x9c834e6","control":"0xc38d38333687ea295753c214744e839eddc7aebb","bundler":"0x7daae72fc3d948b1fa90502f1b84b6a02cce7dfd","userOpHash":"0x0648e7c2a3b705562efd0e06adb8a63916ba5a06ee92a8544369a431298bce47","callChainHash":"0x25c222f384dd726dc226053a74775e0a491fd531d79c447da90ccfb40cedf3fe","signature":"0xd83608483c9649ecf98f3641342d24c864f2143d090b8d2610f3fcdfbd365b1a513bc40af478bcc5f254cc1899022cefacecdb54aff69b7c3e21ef11a4ed49df1b"},"metacallDestination":"0x1b4cb47622705f0f67b6b18bbd1aa1a91fc77d37","metacallGasLimit":"0x3c97b8","metacallMaxFeePerGas":"0x3b9aca00","metacallCallData":"0x4317ca01000000000000000000000000000000000000000000000000000000000000008000000000000000000000000000000000000000000000000000000000000007200000000000000000000000000000000000000000000000000000000000000a400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000b6065f79d99f29c3eda0ed1bda7ff88e7ee12f1e0000000000000000000000001b4cb47622705f0f67b6b18bbd1aa1a91fc77d3700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000186a00000000000000000000000000000000000000000000000000000000003b9aca00000000000000000000000000000000000000000000000000000000000000000f0000000000000000000000000000000000000000000000000000000009c834e6000000000000000000000000c38d38333687ea295753c214744e839eddc7aebb000000000000000000000000c38d38333687ea295753c214744e839eddc7aebb000000000000000000000000000000000000000000000000000000000000230400000000000000000000000000000000000000000000000000000000001e848000000000000000000000000000000000000000000000000000000000005b8d80000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000200000000000000000000000000000000000000000000000000000000000000062000000000000000000000000000000000000000000000000000000000000003e402a688ed0000000000000000000000008ae79bb7cce2dc3d132c288971b0f74af02a3b4a000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000003646fadcf72000000000000000000000000b123c2e3a71b57f9678cf6212576f30d642ed89a000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000002e4ba0cb29e0001aeec9ec2ddb59123858e897b26f52e2173aca2c577dcce62e7ddc70f16020000000000000000000000000000000000000000000000000000000000418a059c756341541f2283d4f124ac39dc9e718f96453582f38acfa885e4ffaf26eb7800000000000000000000000000000000000000000000000000000000000000e0000000000000000000000000000000000000000000000000000000000000022000000000000000000000000000000000000000000000000000000000000002800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000012000000000000000000000000000000000000000000000000000000000678010b601000203000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000040000000000000000000000000000000000000000000000000000000075542db800000000000000000000000000000000000000000000000000000000755c8e4000000000000000000000000000000000000000000000000000000000756106c80000000000000000000000000000000000000000000000000000000075961a520000000000000000000000000000000000000000000000000000000000000002b974b422c8e712d3320994bf4ecd31e37b35a117ef4102285206c45aef4b8709ae407e7a604410ef1e622ebdb4ad302638ff0cf49ad5ab33486c206bf8ecd7dd0000000000000000000000000000000000000000000000000000000000000002671dd95393e5963a3842a5f95d5e6b544de4906b1c304d41aebbb7c2acf9cb065fb96b17223872656de8f05a17ffb6ccc3261b606179cc33db445158aa408b700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000041b6ae5c173d3c7577ea3b8b4d35a0dc33e0f2798d74173c5e6fee2f08b8ce20271287a87443e778a114890140a08be5faf05ebcdd59051f9747fdd35d0afe6aa01b0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000020000000000000000000000000377136653944bdd5d9f0db22987b7432e76c354f0000000000000000000000001b4cb47622705f0f67b6b18bbd1aa1a91fc77d3700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000030d40000000000000000000000000000000000000000000000000000000003b9aca000000000000000000000000000000000000000000000000000000000009c834e60000000000000000000000004b548c6faf4a3c74571c3e194e71cbf2a5172501000000000000000000000000c38d38333687ea295753c214744e839eddc7aebb0648e7c2a3b705562efd0e06adb8a63916ba5a06ee92a8544369a431298bce47000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000186a000000000000000000000000000000000000000000000000000000000000001a000000000000000000000000000000000000000000000000000000000000002600000000000000000000000000000000000000000000000000000000000000084c000a702000000000000000000000000000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000326a27250470f3a518db1334faaa277f8282082d7f54fd9b0f6cfff8f403a74fcb27a12a3860a1146cf7306eb49aa9fa03ccb1000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004191537f835005fcd243b52b7b6436df810a1ea0b18769b2b1417ee9dacc9f9dac309a8b9ef82d82d6572e5feeb7209d12739067afd315434001a2aadd35af77701c000000000000000000000000000000000000000000000000000000000000000000000000000000000000001f3c5ec2ef75a9e1e09b1d46f208669e81000ee60000000000000000000000001b4cb47622705f0f67b6b18bbd1aa1a91fc77d3700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009c834e6000000000000000000000000c38d38333687ea295753c214744e839eddc7aebb0000000000000000000000007daae72fc3d948b1fa90502f1b84b6a02cce7dfd0648e7c2a3b705562efd0e06adb8a63916ba5a06ee92a8544369a431298bce4725c222f384dd726dc226053a74775e0a491fd531d79c447da90ccfb40cedf3fe00000000000000000000000000000000000000000000000000000000000001200000000000000000000000000000000000000000000000000000000000000041d83608483c9649ecf98f3641342d24c864f2143d090b8d2610f3fcdfbd365b1a513bc40af478bcc5f254cc1899022cefacecdb54aff69b7c3e21ef11a4ed49df1b00000000000000000000000000000000000000000000000000000000000000"},"id":1}`) txData, err := hex.DecodeString("6fadcf72000000000000000000000000b123c2e3a71b57f9678cf6212576f30d642ed89a000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000002e4ba0cb29e0001aeec9ec2ddb59123858e897b26f52e2173aca2c577dcce62e7ddc70f16020000000000000000000000000000000000000000000000000000000000418a059c756341541f2283d4f124ac39dc9e718f96453582f38acfa885e4ffaf26eb7800000000000000000000000000000000000000000000000000000000000000e0000000000000000000000000000000000000000000000000000000000000022000000000000000000000000000000000000000000000000000000000000002800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000012000000000000000000000000000000000000000000000000000000000678010b601000203000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000040000000000000000000000000000000000000000000000000000000075542db800000000000000000000000000000000000000000000000000000000755c8e4000000000000000000000000000000000000000000000000000000000756106c80000000000000000000000000000000000000000000000000000000075961a520000000000000000000000000000000000000000000000000000000000000002b974b422c8e712d3320994bf4ecd31e37b35a117ef4102285206c45aef4b8709ae407e7a604410ef1e622ebdb4ad302638ff0cf49ad5ab33486c206bf8ecd7dd0000000000000000000000000000000000000000000000000000000000000002671dd95393e5963a3842a5f95d5e6b544de4906b1c304d41aebbb7c2acf9cb065fb96b17223872656de8f05a17ffb6ccc3261b606179cc33db445158aa408b7000000000000000000000000000000000000000000000000000000000") diff --git a/pkg/txm/storage/inmemory_store.go b/pkg/txm/storage/inmemory_store.go index c0ae84c4f6..cda9982619 100644 --- a/pkg/txm/storage/inmemory_store.go +++ b/pkg/txm/storage/inmemory_store.go @@ -245,7 +245,7 @@ func (m *InMemoryStore) MarkConfirmedAndReorgedTransactions(latestNonce uint64) } existingTx, exists := m.UnconfirmedTransactions[*tx.Nonce] if exists { - m.lggr.Errorw("Another unconfirmed transaction with the same nonce exists. Transaction will overwritten.", + m.lggr.Errorw("Another unconfirmed transaction with the same nonce exists. Transaction will be overwritten.", "existingTx", existingTx, "newTx", tx) } if *tx.Nonce >= latestNonce { diff --git a/pkg/txm/txm.go b/pkg/txm/txm.go index e4a70718a3..889c6843f6 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -15,7 +15,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/timeutil" - "github.com/smartcontractkit/chainlink-evm/pkg/keys" "github.com/smartcontractkit/chainlink-evm/pkg/txm/types" ) @@ -68,7 +67,7 @@ type StuckTxDetector interface { } type Keystore interface { - EnabledAddressesForChain(ctx context.Context, chainID *big.Int) (addresses []common.Address, err error) + EnabledAddresses(ctx context.Context) (addresses []common.Address, err error) } type Config struct { @@ -87,7 +86,7 @@ type Txm struct { errorHandler ErrorHandler stuckTxDetector StuckTxDetector txStore TxStore - keystore keys.AddressLister + keystore Keystore config Config metrics Metrics @@ -99,7 +98,7 @@ type Txm struct { wg sync.WaitGroup } -func NewTxm(lggr logger.Logger, chainID *big.Int, client Client, attemptBuilder AttemptBuilder, txStore TxStore, stuckTxDetector StuckTxDetector, config Config, keystore keys.AddressLister, errorHandler ErrorHandler, metrics Metrics) *Txm { +func NewTxm(lggr logger.Logger, chainID *big.Int, client Client, attemptBuilder AttemptBuilder, txStore TxStore, stuckTxDetector StuckTxDetector, config Config, keystore Keystore, errorHandler ErrorHandler, metrics Metrics) *Txm { return &Txm{ lggr: logger.Sugared(logger.Named(lggr, "Txm")), keystore: keystore, @@ -136,15 +135,16 @@ func (t *Txm) startAddress(address common.Address) { triggerCh := make(chan struct{}, 1) t.triggerCh[address] = triggerCh - t.wg.Add(1) - go t.loop(address, triggerCh) + t.wg.Go(func() { + t.loop(address, triggerCh) + }) } func (t *Txm) initializeNonce(ctx context.Context, address common.Address) { - ctxWithTimeout, cancel := context.WithTimeout(ctx, pendingNonceDefaultTimeout) - defer cancel() for { + ctxWithTimeout, cancel := context.WithTimeout(ctx, pendingNonceDefaultTimeout) pendingNonce, err := t.client.PendingNonceAt(ctxWithTimeout, address) + cancel() if err != nil { t.lggr.Errorw("Error when fetching initial nonce", "address", address, "err", err) select { @@ -208,60 +208,59 @@ func (t *Txm) GetNonce(address common.Address) uint64 { return t.nonceMap[address] } +// SetNonce updates the local nonce map. Lowering the nonce is only allowed by exactly one, +// i.e. releasing the most recently assigned nonce after a failed transmission. Anything +// lower would collide with nonces already assigned to in-flight transactions. func (t *Txm) SetNonce(address common.Address, nonce uint64) { t.nonceMapMu.Lock() defer t.nonceMapMu.Unlock() - t.nonceMap[address] = nonce -} - -func newBackoff(minDuration time.Duration) backoff.Backoff { - return backoff.Backoff{ - Min: minDuration, - Max: 1 * time.Minute, - Jitter: true, + if current := t.nonceMap[address]; nonce+1 < current { + t.lggr.Criticalw("Rejected nonce update that would collide with in-flight transactions", + "address", address, "currentNonce", current, "requestedNonce", nonce) + return } + t.nonceMap[address] = nonce } func (t *Txm) loop(address common.Address, triggerCh chan struct{}) { - defer t.wg.Done() ctx, cancel := t.stopCh.NewCtx() defer cancel() + broadcastWithBackoff := newBackoff(1 * time.Second) - var broadcastCh <-chan time.Time + broadcastTimer := time.NewTimer(broadcastInterval) + defer broadcastTimer.Stop() + backfillTicker := services.TickerConfig{Initial: t.config.BlockTime, JitterPct: services.DefaultJitter}.NewTicker(t.config.BlockTime) defer backfillTicker.Stop() t.initializeNonce(ctx, address) + if ctx.Err() != nil { + return + } for { start := time.Now() - bo, err := t.BroadcastTransaction(ctx, address) + shouldBackoff, err := t.BroadcastTransaction(ctx, address) // use a backoff if transmission is being throttled. if err != nil { t.lggr.Errorw("Error during transaction broadcasting", "err", err) - } else { - t.lggr.Debug("Transaction broadcasting time elapsed: ", time.Since(start)) - } - if bo { - broadcastCh = time.After(broadcastWithBackoff.Duration()) - } else { - broadcastWithBackoff.Reset() - broadcastCh = time.After(timeutil.JitterPct(0.1).Apply(broadcastInterval)) } + t.lggr.Debug("Transaction broadcasting time elapsed: ", time.Since(start)) + resetBroadcastTimer(broadcastTimer, &broadcastWithBackoff, shouldBackoff) + select { case <-ctx.Done(): return case <-triggerCh: continue - case <-broadcastCh: + case <-broadcastTimer.C: continue case <-backfillTicker.C: start := time.Now() err := t.BackfillTransactions(ctx, address) if err != nil { t.lggr.Errorw("Error during backfill", "err", err) - } else { - t.lggr.Debug("Backfill time elapsed: ", time.Since(start)) } + t.lggr.Debug("Backfill time elapsed: ", time.Since(start)) } } } @@ -278,7 +277,7 @@ func (t *Txm) BroadcastTransaction(ctx context.Context, address common.Address) // to insufficient balance. We're making this trade-off to avoid storing stuck transactions and making unnecessary // RPC calls. The upper limit is always MaxInFlightTransactions regardless of the pending nonce. if unconfirmedCount >= MaxInFlightSubset { - if unconfirmedCount > MaxInFlightTransactions { + if unconfirmedCount >= MaxInFlightTransactions { t.metrics.IncrementLifecycleFailure(ctx, StageMaxInFlight) t.lggr.Warnf("Reached transaction limit: %d for unconfirmed transactions", MaxInFlightTransactions) return true, nil @@ -349,16 +348,19 @@ func (t *Txm) sendTransactionWithError(ctx context.Context, tx *types.Transactio return nil } } + // Best-effort check on the first transmission only: an increased pending nonce proves the transmission went + // through. After the first attempt, there is no guarantee an in-flight attempt in the mempool won't keep the + // pending nonce increased, i.e. transaction already known, so the result wouldn't tell us anything about this + // attempt's transmission and we assume it failed. + if tx.AttemptCount != 1 { + return fmt.Errorf("rebroadcast attempt for txID: %v failed: %w", tx.ID, txErr) + } pendingNonce, pErr := t.client.PendingNonceAt(ctx, fromAddress) if pErr != nil { return pErr } if pendingNonce <= *tx.Nonce { - if tx.AttemptCount == 1 { - // We increment the failure counter only during the first attempt to avoid overcounting. After the first attempt, there is no guarantee - // there isn't an in-flight transaction in the mempool that would prevent the nonce from increasing, i.e. transaction already known. - t.metrics.IncrementLifecycleFailure(ctx, StageBroadcast) - } + t.metrics.IncrementLifecycleFailure(ctx, StageBroadcast) return fmt.Errorf("pending nonce for txID: %v didn't increase. PendingNonce: %d, TxNonce: %d. TxErr: %w", tx.ID, pendingNonce, *tx.Nonce, txErr) } } @@ -400,7 +402,7 @@ func (t *Txm) BackfillTransactions(ctx context.Context, address common.Address) } if tx == nil || *tx.Nonce != latestNonce { - t.lggr.Warnf("Nonce gap at nonce: %d - address: %v. Creating a new transaction\n", latestNonce, address) + t.lggr.Warnf("Nonce gap at nonce: %d - address: %v. Creating a new transaction", latestNonce, address) t.metrics.IncrementNumNonceGaps(ctx) return t.createAndSendEmptyTx(ctx, latestNonce, address) } else { //nolint:revive //easier to read @@ -422,7 +424,7 @@ func (t *Txm) BackfillTransactions(ctx context.Context, address common.Address) if tx.AttemptCount >= maxAttemptsThreshold { t.metrics.ReachedMaxAttempts(ctx, true) - t.lggr.Warnf("Reached max attempts threshold for txID: %d. TXM will broadcast more attempts but if this"+ + t.lggr.Warnf("Reached max attempts threshold for txID: %d. TXM will broadcast more attempts but if this"+ " error persists, it means the transaction won't likely be confirmed and there is an issue with the transaction."+ "Look for any error messages from previous broadcasted attempts that may indicate why this happened, i.e. wallet is out of funds. Tx: %v", tx.ID, tx.PrintWithAttempts()) @@ -434,8 +436,10 @@ func (t *Txm) BackfillTransactions(ctx context.Context, address common.Address) // - The transaction has never been broadcasted successfully before // - The last broadcast was more than RetryBlockThreshold blocks ago // - The transaction is purgeable - if tx.LastBroadcastAt == nil || time.Since(*tx.LastBroadcastAt) > (t.config.BlockTime*time.Duration(t.config.RetryBlockThreshold)) || tx.IsPurgeable { - t.lggr.Info("Rebroadcasting attempt for txID: ", tx.ID) + if tx.LastBroadcastAt == nil || + time.Since(*tx.LastBroadcastAt) > (time.Duration(t.config.RetryBlockThreshold)*t.config.BlockTime) || + tx.IsPurgeable { + t.lggr.Infow("Rebroadcasting attempt", "txID", tx.ID, "transactionLifecycleID", tx.GetTransactionLifecycleID(t.lggr)) return t.createAndSendAttempt(ctx, tx, address) } } @@ -461,3 +465,21 @@ func (t *Txm) extractMetrics(ctx context.Context, txs []*types.Transaction) []ui } return confirmedTxIDs } + +func newBackoff(minDuration time.Duration) backoff.Backoff { + return backoff.Backoff{ + Min: minDuration, + Max: 1 * time.Minute, + Jitter: true, + } +} + +// resetBroadcastTimer resets the broadcast timer based on whether a backoff is needed. +func resetBroadcastTimer(timer *time.Timer, bo *backoff.Backoff, shouldBackoff bool) { + if shouldBackoff { + timer.Reset(bo.Duration()) + return + } + bo.Reset() + timer.Reset(timeutil.JitterPct(0.1).Apply(broadcastInterval)) +} diff --git a/pkg/txm/txm_test.go b/pkg/txm/txm_test.go index ace533bde7..ae9f4c02fc 100644 --- a/pkg/txm/txm_test.go +++ b/pkg/txm/txm_test.go @@ -311,7 +311,9 @@ func TestBackfillTransactions(t *testing.T) { client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() err = tm.BackfillTransactions(t.Context(), address) require.NoError(t, err) - tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("Rebroadcasting attempt for txID: %d", attempt.TxID)) + tests.AssertEventually(t, func() bool { + return observedLogs.FilterMessage("Rebroadcasting attempt").FilterField(zap.Uint64("txID", attempt.TxID)).Len() >= 1 + }) }) t.Run("retries instantly if the attempt is purgeable", func(t *testing.T) { @@ -352,7 +354,9 @@ func TestBackfillTransactions(t *testing.T) { client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() err = tm.BackfillTransactions(t.Context(), address) require.NoError(t, err) - tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("Rebroadcasting attempt for txID: %d", attempt.TxID)) + tests.AssertEventually(t, func() bool { + return observedLogs.FilterMessage("Rebroadcasting attempt").FilterField(zap.Uint64("txID", attempt.TxID)).Len() >= 1 + }) // Broadcasted once an empty transaction but it didn't get confirmed, so we need to broadcast again. client.On("NonceAt", mock.Anything, address, mock.Anything).Return(uint64(0), nil).Once() @@ -360,7 +364,9 @@ func TestBackfillTransactions(t *testing.T) { client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() err = tm.BackfillTransactions(t.Context(), address) require.NoError(t, err) - tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("Rebroadcasting attempt for txID: %d", attempt.TxID)) + tests.AssertEventually(t, func() bool { + return observedLogs.FilterMessage("Rebroadcasting attempt").FilterField(zap.Uint64("txID", attempt.TxID)).Len() >= 2 + }) }) t.Run("fetches the unconfirmed transaction for a given nonce, throws a warning for max limit and retries with a new attempt", func(t *testing.T) { @@ -558,10 +564,8 @@ func TestFlow_ErrorHandler(t *testing.T) { mockEstimator.On("BumpFee", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(gas.EvmFee{DynamicFee: gas.DynamicFee{GasTipCap: assets.NewWeiI(6), GasFeeCap: assets.NewWeiI(12)}}, defaultGasLimit, nil).Once() client.On("SendTransaction", mock.Anything, mock.Anything, mock.Anything).Return(dualbroadcast.ErrNoBids).Once() - client.On("PendingNonceAt", mock.Anything, address).Return(initialNonce, nil).Once() + // The pending nonce is not checked after the first attempt, so the send error is assumed to be a failed transmission and returned. err = tm.BackfillTransactions(t.Context(), address) // retry - require.Error(t, err) - require.ErrorContains(t, err, "pending nonce for txID: 1 didn't increase") require.ErrorIs(t, err, dualbroadcast.ErrNoBids) tx, count, err = txStoreManager.FetchUnconfirmedTransactionAtNonceWithCount(t.Context(), 0, address) // same transaction is still in the store require.NoError(t, err) diff --git a/pkg/txm/types/transaction.go b/pkg/txm/types/transaction.go index 70a789cd93..53175e6b5b 100644 --- a/pkg/txm/types/transaction.go +++ b/pkg/txm/types/transaction.go @@ -40,7 +40,7 @@ type Transaction struct { State commontypes.TxState IsPurgeable bool Attempts []*Attempt - AttemptCount uint16 // AttempCount is strictly kept in memory and prevents indefinite retrying + AttemptCount uint16 // AttemptCount is strictly kept in memory and prevents indefinite retrying Meta *sqlutil.JSON Subject uuid.NullUUID