From d99e85e7889afecc617eda3bfcc9cb9a687525a1 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 26 Aug 2026 10:06:46 +0100 Subject: [PATCH 01/10] Add lock to StuckTxDetector --- pkg/txm/stuck_tx_detector.go | 5 +++++ pkg/txm/stuck_tx_detector_test.go | 23 +++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/pkg/txm/stuck_tx_detector.go b/pkg/txm/stuck_tx_detector.go index 44725329aa..55f69e90cb 100644 --- a/pkg/txm/stuck_tx_detector.go +++ b/pkg/txm/stuck_tx_detector.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http" + "sync" "time" "github.com/ethereum/go-ethereum/common" @@ -27,6 +28,7 @@ type stuckTxDetector struct { lggr logger.Logger chainType chaintype.ChainType config StuckTxDetectorConfig + mu sync.Mutex lastPurgeMap map[common.Address]time.Time } @@ -60,6 +62,9 @@ func (s *stuckTxDetector) DetectStuckTransaction(ctx context.Context, tx *types. // so it is more likely to be picked up compared to a transaction that hasn't been broadcasted before. This would avoid slowing down TXM for sebsequent transactions // in case the current one is stuck. func (s *stuckTxDetector) timeBasedDetection(tx *types.Transaction) bool { + s.mu.Lock() + defer s.mu.Unlock() + threshold := (s.config.BlockTime * time.Duration(s.config.StuckTxBlockThreshold)) if tx.LastBroadcastAt == nil { if tx.AttemptCount >= maxAttemptsThreshold { diff --git a/pkg/txm/stuck_tx_detector_test.go b/pkg/txm/stuck_tx_detector_test.go index c57f5acb8e..611087059e 100644 --- a/pkg/txm/stuck_tx_detector_test.go +++ b/pkg/txm/stuck_tx_detector_test.go @@ -1,6 +1,7 @@ package txm import ( + "sync" "testing" "time" @@ -95,4 +96,26 @@ func TestTimeBasedDetection(t *testing.T) { assert.True(t, s.timeBasedDetection(tx1)) assert.False(t, s.timeBasedDetection(tx2)) }) + + t.Run("safe to call concurrently for different addresses", func(t *testing.T) { + config := StuckTxDetectorConfig{ + BlockTime: 1 * time.Second, + StuckTxBlockThreshold: 10, + } + s := NewStuckTxDetector(logger.Test(t), "", config) + lastBroadcastAt := time.Time{} + + var wg sync.WaitGroup + for range 10 { + wg.Go(func() { + tx := &types.Transaction{ + ID: 1, + LastBroadcastAt: &lastBroadcastAt, + FromAddress: testutils.NewAddress(), + } + s.timeBasedDetection(tx) + }) + } + wg.Wait() + }) } From f9caeb62a5055759ea71ea3d8aa6e58cea75c3b8 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 26 Aug 2026 10:13:12 +0100 Subject: [PATCH 02/10] Remove unused stuck tx detector method --- pkg/txm/stuck_tx_detector.go | 60 ------------------------------------ 1 file changed, 60 deletions(-) diff --git a/pkg/txm/stuck_tx_detector.go b/pkg/txm/stuck_tx_detector.go index 55f69e90cb..6f7e30b349 100644 --- a/pkg/txm/stuck_tx_detector.go +++ b/pkg/txm/stuck_tx_detector.go @@ -2,10 +2,6 @@ package txm import ( "context" - "encoding/json" - "fmt" - "io" - "net/http" "sync" "time" @@ -84,59 +80,3 @@ func (s *stuckTxDetector) timeBasedDetection(tx *types.Transaction) bool { } return false } - -type APIResponse struct { - Status string `json:"status,omitempty"` - Hash common.Hash `json:"hash,omitempty"` -} - -const ( - APIStatusPending = "PENDING" - APIStatusIncluded = "INCLUDED" - APIStatusFailed = "FAILED" - APIStatusCancelled = "CANCELLED" - APIStatusUnknown = "UNKNOWN" -) - -// Deprecated: DualBroadcastDetection doesn't provide any significant benefits in terms of speed and time -// based detection can replace it. -func (s *stuckTxDetector) DualBroadcastDetection(ctx context.Context, tx *types.Transaction) (bool, error) { - for _, attempt := range tx.Attempts { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.config.DetectionURL+attempt.Hash.String(), nil) - if err != nil { - return false, fmt.Errorf("failed to make request for txID: %v, attemptHash: %v - %w", tx.ID, attempt.Hash, err) - } - resp, err := http.DefaultClient.Do(req) - if err != nil { - return false, fmt.Errorf("failed to get transaction status for txID: %v, attemptHash: %v - %w", tx.ID, attempt.Hash, err) - } - if resp.StatusCode != http.StatusOK { - resp.Body.Close() - return false, fmt.Errorf("request %v failed with status: %d", req, resp.StatusCode) - } - body, err := io.ReadAll(resp.Body) - resp.Body.Close() - if err != nil { - return false, err - } - - var apiResponse APIResponse - err = json.Unmarshal(body, &apiResponse) - if err != nil { - return false, fmt.Errorf("failed to unmarshal response for txID: %v, attemptHash: %v - %w: %s", tx.ID, attempt.Hash, err, string(body)) - } - switch apiResponse.Status { - case APIStatusPending, APIStatusIncluded: - return false, nil - case APIStatusFailed, APIStatusCancelled: - s.lggr.Debugf("TxID: %v with attempHash: %v was marked as failed/cancelled by the RPC. Transaction is now considered stuck and will be purged.", - tx.ID, attempt.Hash) - return true, nil - case APIStatusUnknown: - continue - default: - continue - } - } - return false, nil -} From 25f5946d61d7c926caf27a3fac9dac81c829313d Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 26 Aug 2026 10:35:53 +0100 Subject: [PATCH 03/10] Fix address addition in storage --- pkg/txm/storage/inmemory_store_manager.go | 1 + pkg/txm/storage/inmemory_store_manager_test.go | 2 ++ 2 files changed, 3 insertions(+) diff --git a/pkg/txm/storage/inmemory_store_manager.go b/pkg/txm/storage/inmemory_store_manager.go index 4d6d421e18..25ca654fa2 100644 --- a/pkg/txm/storage/inmemory_store_manager.go +++ b/pkg/txm/storage/inmemory_store_manager.go @@ -41,6 +41,7 @@ func (m *InMemoryStoreManager) Add(addresses ...common.Address) (err error) { for _, address := range addresses { if _, exists := m.InMemoryStoreMap[address]; exists { err = errors.Join(err, fmt.Errorf("address %v already exists in store manager", address)) + continue } m.InMemoryStoreMap[address] = NewInMemoryStore(m.lggr, address, m.chainID) } diff --git a/pkg/txm/storage/inmemory_store_manager_test.go b/pkg/txm/storage/inmemory_store_manager_test.go index 48d664d096..b1ca580f22 100644 --- a/pkg/txm/storage/inmemory_store_manager_test.go +++ b/pkg/txm/storage/inmemory_store_manager_test.go @@ -20,10 +20,12 @@ func TestAdd(t *testing.T) { err := m.Add(fromAddress) require.NoError(t, err) assert.Len(t, m.InMemoryStoreMap, 1) + existingStore := m.InMemoryStoreMap[fromAddress] // Fails if address exists err = m.Add(fromAddress) require.Error(t, err) + assert.Same(t, existingStore, m.InMemoryStoreMap[fromAddress]) // Adds multiple addresses fromAddress1 := testutils.NewAddress() From aac34a49aa997828c067828951054419f74567be Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 26 Aug 2026 10:48:40 +0100 Subject: [PATCH 04/10] Add lock to InMemoryStoreManager --- .../dualbroadcast/meta_error_handler_test.go | 9 ++- pkg/txm/storage/inmemory_store_manager.go | 58 +++++++++++++------ .../storage/inmemory_store_manager_test.go | 13 +++-- pkg/txm/txm_test.go | 4 +- 4 files changed, 56 insertions(+), 28 deletions(-) diff --git a/pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go b/pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go index 59e934fc63..208cdc1b73 100644 --- a/pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go +++ b/pkg/txm/clientwrappers/dualbroadcast/meta_error_handler_test.go @@ -34,7 +34,8 @@ func TestMetaErrorHandler(t *testing.T) { setNonce := func(address common.Address, nonce uint64) {} txStoreManager := storage.NewInMemoryStoreManager(lggr, testutils.FixtureChainID) require.NoError(t, txStoreManager.Add(address)) - txStore := txStoreManager.InMemoryStoreMap[address] + txStore, exists := txStoreManager.GetStoreSafe(address) + require.True(t, exists) _ = txStore.CreateTransaction(txRequest) tx, err := txStore.UpdateUnstartedTransactionWithNonce(nonce) require.NoError(t, err) @@ -67,7 +68,8 @@ func TestMetaErrorHandler(t *testing.T) { setNonce := func(address common.Address, nonce uint64) {} txStoreManager := storage.NewInMemoryStoreManager(logger.Test(t), testutils.FixtureChainID) require.NoError(t, txStoreManager.Add(address)) - txStore := txStoreManager.InMemoryStoreMap[address] + txStore, exists := txStoreManager.GetStoreSafe(address) + require.True(t, exists) _ = txStore.CreateTransaction(txRequest) tx, err := txStore.UpdateUnstartedTransactionWithNonce(nonce) require.NoError(t, err) @@ -102,7 +104,8 @@ func TestMetaErrorHandler(t *testing.T) { setNonce := func(address common.Address, nonce uint64) {} txStoreManager := storage.NewInMemoryStoreManager(lggr, testutils.FixtureChainID) require.NoError(t, txStoreManager.Add(address)) - txStore := txStoreManager.InMemoryStoreMap[address] + txStore, exists := txStoreManager.GetStoreSafe(address) + require.True(t, exists) _ = txStore.CreateTransaction(txRequest) tx, err := txStore.UpdateUnstartedTransactionWithNonce(nonce) require.NoError(t, err) diff --git a/pkg/txm/storage/inmemory_store_manager.go b/pkg/txm/storage/inmemory_store_manager.go index 25ca654fa2..9b40dde0b7 100644 --- a/pkg/txm/storage/inmemory_store_manager.go +++ b/pkg/txm/storage/inmemory_store_manager.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "math/big" + "sync" "github.com/ethereum/go-ethereum/common" evmtypes "github.com/ethereum/go-ethereum/core/types" @@ -16,9 +17,10 @@ import ( const StoreNotFoundForAddress string = "InMemoryStore for address: %v not found" type InMemoryStoreManager struct { + mu sync.RWMutex lggr logger.Logger chainID *big.Int - InMemoryStoreMap map[common.Address]*InMemoryStore + inMemoryStoreMap map[common.Address]*InMemoryStore } func NewInMemoryStoreManager(lggr logger.Logger, chainID *big.Int) *InMemoryStoreManager { @@ -26,11 +28,19 @@ func NewInMemoryStoreManager(lggr logger.Logger, chainID *big.Int) *InMemoryStor return &InMemoryStoreManager{ lggr: lggr, chainID: chainID, - InMemoryStoreMap: inMemoryStoreMap} + inMemoryStoreMap: inMemoryStoreMap} +} + +func (m *InMemoryStoreManager) GetStoreSafe(fromAddress common.Address) (*InMemoryStore, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + + store, exists := m.inMemoryStoreMap[fromAddress] + return store, exists } func (m *InMemoryStoreManager) AbandonPendingTransactions(_ context.Context, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { store.AbandonPendingTransactions() return nil } @@ -38,46 +48,49 @@ func (m *InMemoryStoreManager) AbandonPendingTransactions(_ context.Context, fro } func (m *InMemoryStoreManager) Add(addresses ...common.Address) (err error) { + m.mu.Lock() + defer m.mu.Unlock() + for _, address := range addresses { - if _, exists := m.InMemoryStoreMap[address]; exists { + if _, exists := m.inMemoryStoreMap[address]; exists { err = errors.Join(err, fmt.Errorf("address %v already exists in store manager", address)) continue } - m.InMemoryStoreMap[address] = NewInMemoryStore(m.lggr, address, m.chainID) + m.inMemoryStoreMap[address] = NewInMemoryStore(m.lggr, address, m.chainID) } return } func (m *InMemoryStoreManager) AppendAttemptToTransaction(_ context.Context, txNonce uint64, fromAddress common.Address, attempt *types.Attempt) (attempts []*types.Attempt, err error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.AppendAttemptToTransaction(txNonce, attempt) } return nil, fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) CountUnstartedTransactions(fromAddress common.Address) (int, error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.CountUnstartedTransactions(), nil } return 0, fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) CreateEmptyUnconfirmedTransaction(_ context.Context, fromAddress common.Address, nonce uint64, gasLimit uint64) (*types.Transaction, error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.CreateEmptyUnconfirmedTransaction(nonce, gasLimit) } return nil, fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) CreateTransaction(_ context.Context, txRequest *types.TxRequest) (*types.Transaction, error) { - if store, exists := m.InMemoryStoreMap[txRequest.FromAddress]; exists { + if store, exists := m.GetStoreSafe(txRequest.FromAddress); exists { return store.CreateTransaction(txRequest), nil } return nil, fmt.Errorf(StoreNotFoundForAddress, txRequest.FromAddress) } func (m *InMemoryStoreManager) FetchUnconfirmedTransactionAtNonceWithCount(_ context.Context, nonce uint64, fromAddress common.Address) (tx *types.Transaction, count int, err error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { tx, count = store.FetchUnconfirmedTransactionAtNonceWithCount(nonce) return } @@ -85,14 +98,14 @@ func (m *InMemoryStoreManager) FetchUnconfirmedTransactionAtNonceWithCount(_ con } func (m *InMemoryStoreManager) FetchUnconfirmedTransactions(_ context.Context, fromAddress common.Address) ([]*types.Transaction, error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.FetchUnconfirmedTransactions() } return nil, fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) MarkConfirmedAndReorgedTransactions(_ context.Context, nonce uint64, fromAddress common.Address) (confirmedTxs []*types.Transaction, unconfirmedTxIDs []uint64, err error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { confirmedTxs, unconfirmedTxIDs, err = store.MarkConfirmedAndReorgedTransactions(nonce) return } @@ -100,49 +113,56 @@ func (m *InMemoryStoreManager) MarkConfirmedAndReorgedTransactions(_ context.Con } func (m *InMemoryStoreManager) MarkUnconfirmedTransactionPurgeable(_ context.Context, nonce uint64, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.MarkUnconfirmedTransactionPurgeable(nonce) } return fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) UpdateTransactionBroadcast(_ context.Context, txID uint64, nonce uint64, attemptHash common.Hash, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.UpdateTransactionBroadcast(txID, nonce, attemptHash) } return fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) UpdateUnstartedTransactionWithNonce(_ context.Context, fromAddress common.Address, nonce uint64) (*types.Transaction, error) { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.UpdateUnstartedTransactionWithNonce(nonce) } return nil, fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) DeleteAttemptForUnconfirmedTx(_ context.Context, nonce uint64, attempt *types.Attempt, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.DeleteAttemptForUnconfirmedTx(nonce, attempt) } return fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) MarkTxFatal(_ context.Context, tx *types.Transaction, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.MarkTxFatal(tx) } return fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) UpdateSignedAttempt(_ context.Context, txID uint64, attemptID uint64, signedTransaction *evmtypes.Transaction, fromAddress common.Address) error { - if store, exists := m.InMemoryStoreMap[fromAddress]; exists { + if store, exists := m.GetStoreSafe(fromAddress); exists { return store.UpdateSignedAttempt(txID, attemptID, signedTransaction) } return fmt.Errorf(StoreNotFoundForAddress, fromAddress) } func (m *InMemoryStoreManager) FindTxWithIdempotencyKey(_ context.Context, idempotencyKey string) (*types.Transaction, error) { - for _, store := range m.InMemoryStoreMap { + m.mu.RLock() + stores := make([]*InMemoryStore, 0, len(m.inMemoryStoreMap)) + for _, store := range m.inMemoryStoreMap { + stores = append(stores, store) + } + m.mu.RUnlock() + + for _, store := range stores { tx := store.FindTxWithIdempotencyKey(idempotencyKey) if tx != nil { return tx, nil diff --git a/pkg/txm/storage/inmemory_store_manager_test.go b/pkg/txm/storage/inmemory_store_manager_test.go index b1ca580f22..b5e8b4d118 100644 --- a/pkg/txm/storage/inmemory_store_manager_test.go +++ b/pkg/txm/storage/inmemory_store_manager_test.go @@ -19,13 +19,15 @@ func TestAdd(t *testing.T) { // Adds a new address err := m.Add(fromAddress) require.NoError(t, err) - assert.Len(t, m.InMemoryStoreMap, 1) - existingStore := m.InMemoryStoreMap[fromAddress] + existingStore, exists := m.GetStoreSafe(fromAddress) + require.True(t, exists) // Fails if address exists err = m.Add(fromAddress) require.Error(t, err) - assert.Same(t, existingStore, m.InMemoryStoreMap[fromAddress]) + store, exists := m.GetStoreSafe(fromAddress) + require.True(t, exists) + assert.Same(t, existingStore, store) // Adds multiple addresses fromAddress1 := testutils.NewAddress() @@ -33,5 +35,8 @@ func TestAdd(t *testing.T) { addresses := []common.Address{fromAddress1, fromAddress2} err = m.Add(addresses...) require.NoError(t, err) - assert.Len(t, m.InMemoryStoreMap, 3) + _, exists = m.GetStoreSafe(fromAddress1) + require.True(t, exists) + _, exists = m.GetStoreSafe(fromAddress2) + require.True(t, exists) } diff --git a/pkg/txm/txm_test.go b/pkg/txm/txm_test.go index e45c32a901..ace533bde7 100644 --- a/pkg/txm/txm_test.go +++ b/pkg/txm/txm_test.go @@ -455,8 +455,8 @@ func TestFlow_ResendTransaction(t *testing.T) { require.NoError(t, tm.BackfillTransactions(t.Context(), address)) // Set LastBroadcastAt to a time in the past to trigger retry condition - txStore := txStoreManager.InMemoryStoreMap[address] - require.NotNil(t, txStore) + txStore, exists := txStoreManager.GetStoreSafe(address) + require.True(t, exists) tx := txStore.UnconfirmedTransactions[initialNonce] require.NotNil(t, tx) pastTime := time.Now().Add(-(config.BlockTime*time.Duration(config.RetryBlockThreshold) + 1*time.Second)) From 40ac035ccb7f64f02754360b66e95c349aca4d28 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 26 Aug 2026 12:09:39 +0100 Subject: [PATCH 05/10] Add TXMv2 fixes --- pkg/txm/txm.go | 25 +++++++++++++------------ pkg/txm/txm_test.go | 6 ++---- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/pkg/txm/txm.go b/pkg/txm/txm.go index e4a70718a3..4edba53399 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -141,10 +141,10 @@ func (t *Txm) startAddress(address common.Address) { } 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 { @@ -278,7 +278,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,17 +349,18 @@ func (t *Txm) sendTransactionWithError(ctx context.Context, tx *types.Transactio return nil } } - 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. + // Best-effort check on the first transmission only. 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. + if tx.AttemptCount == 1 { + pendingNonce, pErr := t.client.PendingNonceAt(ctx, fromAddress) + if pErr != nil { + return pErr + } + if pendingNonce <= *tx.Nonce { 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) } - return fmt.Errorf("pending nonce for txID: %v didn't increase. PendingNonce: %d, TxNonce: %d. TxErr: %w", tx.ID, pendingNonce, *tx.Nonce, txErr) } } diff --git a/pkg/txm/txm_test.go b/pkg/txm/txm_test.go index ace533bde7..76089cb836 100644 --- a/pkg/txm/txm_test.go +++ b/pkg/txm/txm_test.go @@ -558,11 +558,9 @@ 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 doesn't fail the retry. 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) + require.NoError(t, err) tx, count, err = txStoreManager.FetchUnconfirmedTransactionAtNonceWithCount(t.Context(), 0, address) // same transaction is still in the store require.NoError(t, err) require.Equal(t, 1, count) From 468afd58f0fe68ec6cac28630a10957d803217e7 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Thu, 27 Aug 2026 19:24:09 +0100 Subject: [PATCH 06/10] More fixes --- .../dualbroadcast/meta_client.go | 7 ++++ .../dualbroadcast/meta_client_test.go | 25 +++++++++++++ pkg/txm/txm.go | 36 ++++++++++++------- pkg/txm/txm_test.go | 4 +-- 4 files changed, 58 insertions(+), 14 deletions(-) diff --git a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go index 40b08fc3e3..a6faba78ec 100644 --- a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go +++ b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go @@ -226,6 +226,13 @@ 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 meta, but even if it does, eventually it will still be purged. + 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/txm.go b/pkg/txm/txm.go index 4edba53399..6c4d6ba738 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -208,9 +208,19 @@ 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; in that case +// the update is dropped and the nonce gap left by the fatal tx is filled by +// BackfillTransactions' empty tx. func (t *Txm) SetNonce(address common.Address, nonce uint64) { t.nonceMapMu.Lock() defer t.nonceMapMu.Unlock() + 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 } @@ -349,18 +359,20 @@ func (t *Txm) sendTransactionWithError(ctx context.Context, tx *types.Transactio return nil } } - // Best-effort check on the first transmission only. 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. - if tx.AttemptCount == 1 { - pendingNonce, pErr := t.client.PendingNonceAt(ctx, fromAddress) - if pErr != nil { - return pErr - } - if pendingNonce <= *tx.Nonce { - 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) - } + // 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 { + 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) } } diff --git a/pkg/txm/txm_test.go b/pkg/txm/txm_test.go index 76089cb836..5ea50aa199 100644 --- a/pkg/txm/txm_test.go +++ b/pkg/txm/txm_test.go @@ -558,9 +558,9 @@ 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() - // The pending nonce is not checked after the first attempt, so the send error doesn't fail the retry. + // 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.NoError(t, err) + 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) require.Equal(t, 1, count) From 1fe70a2e1f6bde210afef6030c6045472391a1d8 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Tue, 1 Sep 2026 12:01:18 +0300 Subject: [PATCH 07/10] Update SetNonce comment --- pkg/txm/txm.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pkg/txm/txm.go b/pkg/txm/txm.go index 6c4d6ba738..6aa7758bb4 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -210,9 +210,7 @@ func (t *Txm) GetNonce(address common.Address) uint64 { // 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; in that case -// the update is dropped and the nonce gap left by the fatal tx is filled by -// BackfillTransactions' empty tx. +// 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() From adafa6a6a00094f34a2d942cdb7465704f2c3ef7 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 2 Sep 2026 12:54:23 +0300 Subject: [PATCH 08/10] Fix nits --- .../dualbroadcast/meta_client.go | 3 ++- pkg/txm/storage/inmemory_store.go | 2 +- pkg/txm/txm.go | 27 +++++++++++-------- pkg/txm/types/transaction.go | 2 +- 4 files changed, 20 insertions(+), 14 deletions(-) diff --git a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go index a6faba78ec..8a759970e8 100644 --- a/pkg/txm/clientwrappers/dualbroadcast/meta_client.go +++ b/pkg/txm/clientwrappers/dualbroadcast/meta_client.go @@ -229,7 +229,8 @@ func (a *MetaClient) SendTransaction(ctx context.Context, tx *types.Transaction, // 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 meta, but even if it does, eventually it will still be purged. + // 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) } 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 6aa7758bb4..4f99e4ed37 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, @@ -235,11 +234,15 @@ func (t *Txm) loop(address common.Address, triggerCh chan struct{}) { 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() @@ -250,17 +253,17 @@ func (t *Txm) loop(address common.Address, triggerCh chan struct{}) { t.lggr.Debug("Transaction broadcasting time elapsed: ", time.Since(start)) } if bo { - broadcastCh = time.After(broadcastWithBackoff.Duration()) + broadcastTimer.Reset(broadcastWithBackoff.Duration()) } else { broadcastWithBackoff.Reset() - broadcastCh = time.After(timeutil.JitterPct(0.1).Apply(broadcastInterval)) + broadcastTimer.Reset(timeutil.JitterPct(0.1).Apply(broadcastInterval)) } select { case <-ctx.Done(): return case <-triggerCh: continue - case <-broadcastCh: + case <-broadcastTimer.C: continue case <-backfillTicker.C: start := time.Now() @@ -411,7 +414,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 @@ -433,7 +436,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()) @@ -445,7 +448,9 @@ 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 { + if tx.LastBroadcastAt == nil || + time.Since(*tx.LastBroadcastAt) > (time.Duration(t.config.RetryBlockThreshold)*t.config.BlockTime) || + tx.IsPurgeable { t.lggr.Info("Rebroadcasting attempt for txID: ", tx.ID) return t.createAndSendAttempt(ctx, tx, address) } 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 From bb637018df1ae564dfc9c02e57d157eb177551eb Mon Sep 17 00:00:00 2001 From: Dimitris Date: Wed, 2 Sep 2026 17:18:17 +0300 Subject: [PATCH 09/10] Add transactionLifecycleID to rebroadcasting --- pkg/txm/txm.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/txm/txm.go b/pkg/txm/txm.go index 4f99e4ed37..d3d5cd471a 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -451,7 +451,7 @@ func (t *Txm) BackfillTransactions(ctx context.Context, address common.Address) if tx.LastBroadcastAt == nil || time.Since(*tx.LastBroadcastAt) > (time.Duration(t.config.RetryBlockThreshold)*t.config.BlockTime) || tx.IsPurgeable { - t.lggr.Info("Rebroadcasting attempt for txID: ", tx.ID) + t.lggr.Infow("Rebroadcasting attempt", "txID", tx.ID, "transactionLifecycleID", tx.GetTransactionLifecycleID(t.lggr)) return t.createAndSendAttempt(ctx, tx, address) } } From 7721d7e8cb4ff72304634f311ce674e278ec4708 Mon Sep 17 00:00:00 2001 From: Dimitris Date: Thu, 3 Sep 2026 12:04:17 +0300 Subject: [PATCH 10/10] Clean up loop method --- pkg/txm/txm.go | 50 +++++++++++++++++++++++++-------------------- pkg/txm/txm_test.go | 12 ++++++++--- 2 files changed, 37 insertions(+), 25 deletions(-) diff --git a/pkg/txm/txm.go b/pkg/txm/txm.go index d3d5cd471a..889c6843f6 100644 --- a/pkg/txm/txm.go +++ b/pkg/txm/txm.go @@ -135,8 +135,9 @@ 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) { @@ -221,21 +222,14 @@ func (t *Txm) SetNonce(address common.Address, nonce uint64) { t.nonceMap[address] = nonce } -func newBackoff(minDuration time.Duration) backoff.Backoff { - return backoff.Backoff{ - Min: minDuration, - Max: 1 * time.Minute, - Jitter: true, - } -} - 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) broadcastTimer := time.NewTimer(broadcastInterval) defer broadcastTimer.Stop() + backfillTicker := services.TickerConfig{Initial: t.config.BlockTime, JitterPct: services.DefaultJitter}.NewTicker(t.config.BlockTime) defer backfillTicker.Stop() @@ -246,18 +240,13 @@ func (t *Txm) loop(address common.Address, triggerCh chan struct{}) { 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 { - broadcastTimer.Reset(broadcastWithBackoff.Duration()) - } else { - broadcastWithBackoff.Reset() - broadcastTimer.Reset(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 @@ -270,9 +259,8 @@ func (t *Txm) loop(address common.Address, triggerCh chan struct{}) { 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)) } } } @@ -477,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 5ea50aa199..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) {