fix(transaction): prevent concurrent map access in monitor

Author: mfw78

pkg/transaction/monitor.go

@@ -18,8 +18,10 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 )
 
-var ErrTransactionCancelled = errors.New("transaction cancelled")
-var ErrMonitorClosed = errors.New("monitor closed")
+var (
+	ErrTransactionCancelled = errors.New("transaction cancelled")
+	ErrMonitorClosed        = errors.New("monitor closed")
+)
 
 // Monitor is a nonce-based watcher for transaction confirmations.
 // Instead of watching transactions individually, the senders nonce is monitored and transactions are checked based on this.
@@ -190,31 +192,56 @@ func watchStart(watches []transactionWatch) time.Time {
 	return start
 }
 
+// txToCheck holds snapshot data for a transaction that needs to be checked.
+// This allows releasing the lock during slow RPC calls.
+type txToCheck struct {
+	nonce      uint64
+	txHash     common.Hash
+	watchStart time.Time
+}
+
 // check pending checks the given block (number) for confirmed or cancelled transactions
 func (tm *transactionMonitor) checkPending(block uint64) error {
-	confirmedNonces := make(map[uint64]*types.Receipt)
-	var cancelledNonces []uint64
+	// Phase 1: Snapshot the transactions we need to check under the lock.
+	// This allows us to release the lock during slow RPC calls.
+	tm.lock.Lock()
+	var txsToCheck []txToCheck
+	noncesToCheck := make(map[uint64]struct{})
 	for nonceGroup, watchMap := range tm.watchesByNonce {
+		noncesToCheck[nonceGroup] = struct{}{}
 		for txHash, watches := range watchMap {
-			receipt, err := tm.backend.TransactionReceipt(tm.ctx, txHash)
-			if err != nil {
-				// wait for a few blocks to be mined before considering a transaction not existing
-				transactionWatchNotFoundTimeout := 5 * tm.pollingInterval
-				if errors.Is(err, ethereum.NotFound) && watchStart(watches).Before(time.Now().Add(transactionWatchNotFoundTimeout)) {
-					// if both err and receipt are nil, there is no receipt
-					// the reason why we consider this only potentially cancelled is to catch cases where after a reorg the original transaction wins
-					continue
-				}
-				return err
-			}
-			if receipt != nil {
-				// if we have a receipt we have a confirmation
-				confirmedNonces[nonceGroup] = receipt
+			txsToCheck = append(txsToCheck, txToCheck{
+				nonce:      nonceGroup,
+				txHash:     txHash,
+				watchStart: watchStart(watches),
+			})
+		}
+	}
+	tm.lock.Unlock()
+
+	// Phase 2: Make RPC calls without holding the lock.
+	// TransactionReceipt and NonceAt can be slow (100-500ms each).
+	confirmedNonces := make(map[uint64]*types.Receipt)
+	for _, tx := range txsToCheck {
+		receipt, err := tm.backend.TransactionReceipt(tm.ctx, tx.txHash)
+		if err != nil {
+			// wait for a few blocks to be mined before considering a transaction not existing
+			transactionWatchNotFoundTimeout := 5 * tm.pollingInterval
+			if errors.Is(err, ethereum.NotFound) && tx.watchStart.Before(time.Now().Add(transactionWatchNotFoundTimeout)) {
+				// if both err and receipt are nil, there is no receipt
+				// the reason why we consider this only potentially cancelled is to catch cases where after a reorg the original transaction wins
+				continue
 			}
+			return err
+		}
+		if receipt != nil {
+			// if we have a receipt we have a confirmation
+			confirmedNonces[tx.nonce] = receipt
 		}
 	}
 
-	for nonceGroup := range tm.watchesByNonce {
+	var cancelledNonces []uint64
+	for nonceGroup := range noncesToCheck {
 		if _, ok := confirmedNonces[nonceGroup]; ok {
 			continue
 		}
@@ -229,12 +256,18 @@ func (tm *transactionMonitor) checkPending(block uint64) error {
 		}
 	}
 
-	// notify the subscribers and remove watches for confirmed or cancelled transactions
+	// Phase 3: Notify subscribers and cleanup under the lock.
 	tm.lock.Lock()
 	defer tm.lock.Unlock()
 
+	// notify the subscribers and remove watches for confirmed or cancelled transactions
 	for nonce, receipt := range confirmedNonces {
-		for txHash, watches := range tm.watchesByNonce[nonce] {
+		watchMap, ok := tm.watchesByNonce[nonce]
+		if !ok {
+			// nonce was already processed (shouldn't happen but be defensive)
+			continue
+		}
+		for txHash, watches := range watchMap {
 			if receipt.TxHash == txHash {
 				for _, watch := range watches {
 					select {
@@ -255,7 +288,12 @@ func (tm *transactionMonitor) checkPending(block uint64) error {
 	}
 
 	for _, nonce := range cancelledNonces {
-		for _, watches := range tm.watchesByNonce[nonce] {
+		watchMap, ok := tm.watchesByNonce[nonce]
+		if !ok {
+			// nonce was already processed
+			continue
+		}
+		for _, watches := range watchMap {
 			for _, watch := range watches {
 				select {
 				case watch.errC <- ErrTransactionCancelled: