func (tm *transactionMonitor) watchPending() {
	defer tm.wg.Done()
	defer func() {
		tm.lock.Lock()
		defer tm.lock.Unlock()
		for watch := range tm.watches {
			watch.errC <- ErrMonitorClosed
		}
	}()
	var (
		lastBlock uint64 = 0
		added     bool    
	)
	for {
		added = false
		select {
		case <-tm.watchAdded:
			added = true
		case <-time.After(tm.pollingInterval):
		case <-tm.ctx.Done():
			return
		}
		if !tm.hasWatches() {
			continue
		}
		block, err := tm.backend.BlockNumber(tm.ctx)
		if err != nil {
			tm.logger.Errorf("could not get block number: %v", err)
			continue
		} else if block <= lastBlock && !added {
			continue
		}
		if err := tm.checkPending(block); err != nil {
			tm.logger.Tracef("error while checking pending transactions: %v", err)
			continue
		}
		lastBlock = block
	}
}
main watch loop
package transaction
import (
	"context"
	"errors"
	"io"
	"math/big"
	"sync"
	"time"
	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethsana/sana/pkg/logging"
)
var ErrTransactionCancelled = errors.New("transaction cancelled")
var ErrMonitorClosed = errors.New("monitor closed")
type Monitor interface {
	io.Closer
	WatchTransaction(txHash common.Hash, nonce uint64) (<-chan types.Receipt, <-chan error, error)
}
type transactionMonitor struct {
	lock       sync.Mutex
	ctx        context.Context     
	cancelFunc context.CancelFunc  
	wg         sync.WaitGroup
	logger  logging.Logger
	backend Backend
	sender  common.Address  
	pollingInterval   time.Duration  
	cancellationDepth uint64         
	watches    map[*transactionWatch]struct{}  
	watchAdded chan struct{}                   
}
type transactionWatch struct {
	receiptC chan types.Receipt  
	errC     chan error          
	txHash common.Hash  
	nonce  uint64       
}
func NewMonitor(logger logging.Logger, backend Backend, sender common.Address, pollingInterval time.Duration, cancellationDepth uint64) Monitor {
	ctx, cancelFunc := context.WithCancel(context.Background())
	t := &transactionMonitor{
		ctx:        ctx,
		cancelFunc: cancelFunc,
		logger:     logger,
		backend:    backend,
		sender:     sender,
		pollingInterval:   pollingInterval,
		cancellationDepth: cancellationDepth,
		watches:    make(map[*transactionWatch]struct{}),
		watchAdded: make(chan struct{}, 1),
	}
	t.wg.Add(1)
	go t.watchPending()
	return t
}
func (tm *transactionMonitor) WatchTransaction(txHash common.Hash, nonce uint64) (<-chan types.Receipt, <-chan error, error) {
	tm.lock.Lock()
	defer tm.lock.Unlock()
	receiptC := make(chan types.Receipt, 1)
	errC := make(chan error, 1)
	tm.watches[&transactionWatch{
		receiptC: receiptC,
		errC:     errC,
		txHash:   txHash,
		nonce:    nonce,
	}] = struct{}{}
	select {
	case tm.watchAdded <- struct{}{}:
	default:
	}
	tm.logger.Tracef("starting to watch transaction %x with nonce %d", txHash, nonce)
	return receiptC, errC, nil
}