fix: monitor-driven nonce reset for stuck transactions (#899)

* fix: replace maxNonceDrift with monitor-driven nonce reset

Instead of the transactor detecting nonce drift on every PendingNonceAt
call (which conflicts with maxPendingTxs=2048), move stuck detection to
the Monitor. When the confirmed nonce doesn't advance for 30s while
there are pending txs, the Monitor signals the transactor to reset its
local nonce via a NonceOverride channel.

This cleanly separates concerns: the Monitor already tracks confirmed
nonces and pending transactions, making it the right place for stuck
detection. The transactor simply reacts to the override signal.

Replaces the maxNonceDrift approach from PR #896 (bug 2 fix).
Retains the retry loop fall-through fix (bug 1) from PR #896.

* review: fix data race in log, add monitor stuck detection tests

- Fix data race: len(m.waitMap) was read without lock in the stuck
  detection log. Replaced hasPendingTxs() with pendingTxCount() that
  returns the count under lock, reused in both the condition and log.
- Add SetStuckDuration() for testing with short durations.
- Add TestStuckDetectionSendsNonceOverride: verifies monitor sends
  override when confirmed nonce doesn't advance with pending txs.
- Add TestNoOverrideWhenNonceAdvances: verifies no override fires
  when the confirmed nonce advances before stuckDuration.

Author: kant777

x/contracts/transactor/transactor.go

@@ -12,23 +12,19 @@ import (
 
 const (
 	txnRetriesLimit = 3
-
-	// defaultMaxNonceDrift is the maximum allowed difference between the
-	// local nonce counter and the chain's pending nonce before the transactor
-	// resets to the chain nonce. Since sends are serialized via a size-1
-	// channel, under normal operation the drift should be 0. A drift larger
-	// than this threshold indicates that previously sent transactions were
-	// lost (e.g. mempool cleared after a node restart).
-	defaultMaxNonceDrift uint64 = 5
 )
 
 // Watcher is an interface that is used to manage the lifecycle of a transaction.
 // The Allow method is used to determine if a transaction should be sent. The context
 // is passed to the method so that the watcher can determine this based on the context.
-// The Sent method is is used to notify the watcher that the transaction has been sent.
+// The Sent method is used to notify the watcher that the transaction has been sent.
+// The NonceOverride method returns a channel that signals the transactor to reset its
+// local nonce to the provided value. This is used by the monitor to recover from stuck
+// states where transactions were lost (e.g. mempool cleared after a node restart).
 type Watcher interface {
 	Allow(ctx context.Context, nonce uint64) bool
 	Sent(ctx context.Context, tx *types.Transaction)
+	NonceOverride() <-chan uint64
 }
 
 // Transactor is a wrapper around a bind.ContractBackend that ensures that
@@ -46,24 +42,14 @@ type Watcher interface {
 // of an error, the nonce is put back into the channel so that it can be reused.
 type Transactor struct {
 	bind.ContractBackend
-	nonceChan     chan uint64
-	watcher       Watcher
-	maxNonceDrift uint64
-	logger        *slog.Logger
+	nonceChan chan uint64
+	watcher   Watcher
+	logger    *slog.Logger
 }
 
 // Option is a functional option for configuring the Transactor.
 type Option func(*Transactor)
 
-// WithMaxNonceDrift sets the maximum allowed drift between the local nonce
-// counter and the chain's pending nonce before the transactor self-heals by
-// resetting to the chain nonce.
-func WithMaxNonceDrift(n uint64) Option {
-	return func(t *Transactor) {
-		t.maxNonceDrift = n
-	}
-}
-
 // WithLogger sets the logger for the transactor.
 func WithLogger(l *slog.Logger) Option {
 	return func(t *Transactor) {
@@ -85,7 +71,6 @@ func NewTransactor(
 		ContractBackend: backend,
 		watcher:         watcher,
 		nonceChan:       nonceChan,
-		maxNonceDrift:   defaultMaxNonceDrift,
 		logger:          slog.Default(),
 	}
 	for _, opt := range opts {
@@ -99,6 +84,17 @@ func (t *Transactor) PendingNonceAt(ctx context.Context, account common.Address)
 	case <-ctx.Done():
 		return 0, ctx.Err()
 	case nonce := <-t.nonceChan:
+		// Check if the monitor has signaled a nonce override (stuck detection).
+		select {
+		case override := <-t.watcher.NonceOverride():
+			t.logger.Warn("nonce override from monitor, resetting",
+				"localNonce", nonce,
+				"overrideNonce", override,
+			)
+			nonce = override
+		default:
+		}
+
 		pendingNonce, err := t.ContractBackend.PendingNonceAt(ctx, account)
 		if err != nil {
 			// this naked write is safe as only the SendTransaction writes to
@@ -110,19 +106,6 @@ func (t *Transactor) PendingNonceAt(ctx context.Context, account common.Address)
 		if pendingNonce > nonce {
 			return pendingNonce, nil
 		}
-		// Self-healing: if local nonce drifts too far ahead of the chain's
-		// pending nonce, transactions were likely lost (e.g. mempool cleared
-		// after a node restart). Reset to chain nonce to close the gap.
-		// This does not add any extra RPC calls — we already query the
-		// chain's pending nonce above.
-		if nonce > pendingNonce+t.maxNonceDrift {
-			t.logger.Warn("nonce drift detected, resetting to chain pending nonce",
-				"localNonce", nonce,
-				"chainPendingNonce", pendingNonce,
-				"drift", nonce-pendingNonce,
-			)
-			return pendingNonce, nil
-		}
 		return nonce, nil
 	}
 }

x/contracts/transactor/transactor_test.go

@@ -149,17 +149,44 @@ func (w *autoAllowWatcher) Sent(_ context.Context, tx *types.Transaction) {
 	}
 }
 
-func TestNonceDriftSelfHealing(t *testing.T) {
+func (w *autoAllowWatcher) NonceOverride() <-chan uint64 {
+	return make(chan uint64)
+}
+
+// nonceOverrideWatcher is a watcher that allows all txs and supports nonce override.
+type nonceOverrideWatcher struct {
+	txnChan       chan *types.Transaction
+	overrideChan  chan uint64
+}
+
+func (w *nonceOverrideWatcher) Allow(_ context.Context, _ uint64) bool {
+	return true
+}
+
+func (w *nonceOverrideWatcher) Sent(_ context.Context, tx *types.Transaction) {
+	if w.txnChan != nil {
+		w.txnChan <- tx
+	}
+}
+
+func (w *nonceOverrideWatcher) NonceOverride() <-chan uint64 {
+	return w.overrideChan
+}
+
+func TestNonceOverrideFromMonitor(t *testing.T) {
 	t.Parallel()
 
 	backend := &testBackend{
-		nonce: 100, // chain pending nonce
+		nonce: 100,
+	}
+	overrideChan := make(chan uint64, 1)
+	watcher := &nonceOverrideWatcher{
+		txnChan:      make(chan *types.Transaction, 16),
+		overrideChan: overrideChan,
 	}
-	watcher := &autoAllowWatcher{txnChan: make(chan *types.Transaction, 16)}
-	// maxNonceDrift = 5: drift of 6+ triggers reset
-	txnSender := transactor.NewTransactor(backend, watcher, transactor.WithMaxNonceDrift(5))
+	txnSender := transactor.NewTransactor(backend, watcher)
 
-	// First call: local nonce is 0 (initial), chain says 100 → returns 100
+	// First call: local nonce is 0, chain says 100 → returns 100
 	nonce, err := txnSender.PendingNonceAt(context.Background(), common.Address{})
 	if err != nil {
 		t.Fatal(err)
@@ -170,7 +197,7 @@ func TestNonceDriftSelfHealing(t *testing.T) {
 
 	// Send nonces 100-109 to advance local nonce to 110
 	for i := uint64(100); i <= 109; i++ {
-		backend.nonce = i + 1 // chain keeps up
+		backend.nonce = i + 1
 		if i > 100 {
 			nonce, err = txnSender.PendingNonceAt(context.Background(), common.Address{})
 			if err != nil {
@@ -185,28 +212,32 @@ func TestNonceDriftSelfHealing(t *testing.T) {
 		nonce = i + 1
 	}
 
-	// Local nonce is now 110. Simulate mempool wipe: chain nonce drops to 100.
-	// This simulates the node restart scenario where all pending txs are lost.
+	// Local nonce is now 110. Simulate monitor detecting stuck state and
+	// sending a nonce override to reset to the confirmed nonce (100).
 	backend.nonce = 100
+	overrideChan <- 100
 
-	// Drift = 110 - 100 = 10 > maxNonceDrift(5) → should reset to 100
 	nonce, err = txnSender.PendingNonceAt(context.Background(), common.Address{})
 	if err != nil {
 		t.Fatal(err)
 	}
 	if nonce != 100 {
-		t.Fatalf("expected nonce to reset to 100 (chain nonce), got %d", nonce)
+		t.Fatalf("expected nonce to reset to 100 via monitor override, got %d", nonce)
 	}
 }
 
-func TestNonceDriftWithinThreshold(t *testing.T) {
+func TestNonceOverrideNotTriggeredWithoutSignal(t *testing.T) {
 	t.Parallel()
 
 	backend := &testBackend{
 		nonce: 100,
 	}
-	watcher := &autoAllowWatcher{txnChan: make(chan *types.Transaction, 16)}
-	txnSender := transactor.NewTransactor(backend, watcher, transactor.WithMaxNonceDrift(5))
+	overrideChan := make(chan uint64, 1)
+	watcher := &nonceOverrideWatcher{
+		txnChan:      make(chan *types.Transaction, 16),
+		overrideChan: overrideChan,
+	}
+	txnSender := transactor.NewTransactor(backend, watcher)
 
 	// Get initial nonce from chain (100)
 	nonce, err := txnSender.PendingNonceAt(context.Background(), common.Address{})
@@ -216,7 +247,7 @@ func TestNonceDriftWithinThreshold(t *testing.T) {
 
 	// Send txs 100-104 to advance local nonce to 105
 	for i := uint64(100); i <= 104; i++ {
-		backend.nonce = i // chain nonce stays behind (simulating pending txs)
+		backend.nonce = i + 1
 		if i > 100 {
 			nonce, err = txnSender.PendingNonceAt(context.Background(), common.Address{})
 			if err != nil {
@@ -231,15 +262,18 @@ func TestNonceDriftWithinThreshold(t *testing.T) {
 		nonce = i + 1
 	}
 
-	// Local nonce = 105. Chain nonce = 104 (drift = 1, within threshold).
-	// Should use local nonce, NOT reset.
+	// No override signal sent — local nonce should be used even if
+	// chain nonce is lower, because the monitor hasn't detected a stuck state.
+	// After sending 100-104, local nonce is 105. The last PendingNonceAt in the
+	// loop saw chain=105, so transactor internal nonce is 105+1=106 after the
+	// last SendTransaction.
 	backend.nonce = 100
 	nonce, err = txnSender.PendingNonceAt(context.Background(), common.Address{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if nonce != 105 {
-		t.Fatalf("expected local nonce 105 (within drift threshold), got %d", nonce)
+	if nonce != 106 {
+		t.Fatalf("expected local nonce 106 (no override), got %d", nonce)
 	}
 }
 
@@ -305,6 +339,10 @@ func (w *testWatcher) Sent(ctx context.Context, tx *types.Transaction) {
 	w.txnChan <- tx
 }
 
+func (w *testWatcher) NonceOverride() <-chan uint64 {
+	return make(chan uint64)
+}
+
 type testBackend struct {
 	bind.ContractBackend
 	nonce           uint64

x/contracts/txmonitor/txmonitor.go

@@ -26,6 +26,13 @@ var (
 	ErrMonitorClosed = errors.New("monitor was closed")
 )
 
+const (
+	// defaultStuckDuration is how long the confirmed nonce must remain
+	// unchanged (while there are pending txs) before the monitor signals
+	// the transactor to reset its nonce.
+	defaultStuckDuration = 30 * time.Second
+)
+
 type TxnDetails struct {
 	Hash    common.Hash
 	Nonce   uint64
@@ -68,9 +75,11 @@ type Monitor struct {
 	newTxAdded         chan struct{}
 	nonceUpdate        chan struct{}
 	blockUpdate        chan waitCheck
+	nonceOverrideChan  chan uint64
 	logger             *slog.Logger
 	lastConfirmedNonce atomic.Uint64
 	maxPendingTxs      uint64
+	stuckDuration      time.Duration
 	metrics            *metrics
 }
 
@@ -86,17 +95,19 @@ func New(
 		saver = noopSaver{}
 	}
 	m := &Monitor{
-		owner:         owner,
-		client:        client,
-		logger:        logger,
-		helper:        helper,
-		saver:         saver,
-		maxPendingTxs: maxPendingTxs,
-		metrics:       newMetrics(),
-		waitMap:       make(map[uint64]map[common.Hash][]chan Result),
-		newTxAdded:    make(chan struct{}),
-		nonceUpdate:   make(chan struct{}),
-		blockUpdate:   make(chan waitCheck),
+		owner:             owner,
+		client:            client,
+		logger:            logger,
+		helper:            helper,
+		saver:             saver,
+		maxPendingTxs:     maxPendingTxs,
+		stuckDuration:     defaultStuckDuration,
+		metrics:           newMetrics(),
+		waitMap:           make(map[uint64]map[common.Hash][]chan Result),
+		newTxAdded:        make(chan struct{}),
+		nonceUpdate:       make(chan struct{}),
+		blockUpdate:       make(chan waitCheck),
+		nonceOverrideChan: make(chan uint64, 1),
 	}
 
 	pending, err := saver.PendingTxns()
@@ -111,10 +122,30 @@ func New(
 	return m
 }
 
+// SetStuckDuration overrides the default stuck detection duration.
+// This is intended for testing.
+func (m *Monitor) SetStuckDuration(d time.Duration) {
+	m.stuckDuration = d
+}
+
 func (m *Monitor) Metrics() []prometheus.Collector {
 	return m.metrics.Metrics()
 }
 
+// NonceOverride returns a channel that the transactor reads from to detect
+// nonce resets. When the monitor detects that the confirmed nonce has not
+// advanced for stuckDuration despite having pending transactions, it sends
+// the confirmed nonce on this channel to tell the transactor to reset.
+func (m *Monitor) NonceOverride() <-chan uint64 {
+	return m.nonceOverrideChan
+}
+
+func (m *Monitor) pendingTxCount() int {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
+	return len(m.waitMap)
+}
+
 func (m *Monitor) Start(ctx context.Context) <-chan struct{} {
 	wg := sync.WaitGroup{}
 	done := make(chan struct{})
@@ -142,6 +173,8 @@ func (m *Monitor) Start(ctx context.Context) <-chan struct{} {
 
 		m.logger.Info("monitor started")
 		lastBlock := uint64(0)
+		var lastNonceAdvance time.Time
+		var lastSeenNonce uint64
 		for {
 			newTx := false
 			select {
@@ -177,6 +210,28 @@ func (m *Monitor) Start(ctx context.Context) <-chan struct{} {
 			m.metrics.lastBlockNumber.Set(float64(currentBlock))
 			m.triggerNonceUpdate()
 
+			// Stuck detection: if the confirmed nonce hasn't advanced
+			// for stuckDuration while we have pending txs, signal
+			// the transactor to reset its local nonce.
+			if lastNonce > lastSeenNonce {
+				lastSeenNonce = lastNonce
+				lastNonceAdvance = time.Now()
+			} else if pendingCount := m.pendingTxCount(); pendingCount > 0 && !lastNonceAdvance.IsZero() &&
+				time.Since(lastNonceAdvance) >= m.stuckDuration {
+				m.logger.Warn("stuck detected: confirmed nonce not advancing, signaling nonce reset",
+					"confirmedNonce", lastNonce,
+					"stuckFor", time.Since(lastNonceAdvance).String(),
+					"pendingTxs", pendingCount,
+				)
+				select {
+				case m.nonceOverrideChan <- lastNonce:
+				default:
+					// channel already has a pending override
+				}
+				// Reset the timer so we don't spam overrides
+				lastNonceAdvance = time.Now()
+			}
+
 			select {
 			case m.blockUpdate <- waitCheck{lastNonce, currentBlock}:
 			default: