core/txpool: revert changes in txpool

core/txpool/blobpool/blobpool.go

@@ -307,8 +307,10 @@ type BlobPool struct {
 	spent  map[common.Address]*uint256.Int  // Expenditure tracking for individual accounts
 	evict  *evictHeap                       // Heap of cheapest accounts for eviction when full
 
-	eventFeed event.Feed   // Event feed to send out new tx events on pool inclusion
+	eventFeed  event.Feed              // Event feed to send out new tx events on pool inclusion
-	lock      sync.RWMutex // Mutex protecting the pool during reorg handling
+	eventScope event.SubscriptionScope // Event scope to track and mass unsubscribe on termination
+
+	lock sync.RWMutex // Mutex protecting the pool during reorg handling
 }
 
 // New creates a new blob transaction pool to gather, sort and filter inbound
@@ -353,7 +355,13 @@ func (p *BlobPool) Init(gasTip *big.Int, head *types.Header, reserve txpool.Addr
 			return err
 		}
 	}
+	// Initialize the state with head block, or fallback to empty one in
+	// case the head state is not available(might occur when node is not
+	// fully synced).
 	state, err := p.chain.StateAt(head.Root)
+	if err != nil {
+		state, err = p.chain.StateAt(types.EmptyRootHash)
+	}
 	if err != nil {
 		return err
 	}
@@ -428,6 +436,8 @@ func (p *BlobPool) Close() error {
 	if err := p.store.Close(); err != nil {
 		errs = append(errs, err)
 	}
+	p.eventScope.Close()
+
 	switch {
 	case errs == nil:
 		return nil
@@ -1461,7 +1471,7 @@ func (p *BlobPool) updateLimboMetrics() {
 // SubscribeTransactions registers a subscription of NewTxsEvent and
 // starts sending event to the given channel.
 func (p *BlobPool) SubscribeTransactions(ch chan<- core.NewTxsEvent) event.Subscription {
-	return p.eventFeed.Subscribe(ch)
+	return p.eventScope.Track(p.eventFeed.Subscribe(ch))
 }
 
 // Nonce returns the next nonce of an account, with all transactions executable

core/txpool/legacypool/legacypool.go

@@ -208,6 +208,7 @@ type LegacyPool struct {
 	chain       BlockChain
 	gasTip      atomic.Pointer[big.Int]
 	txFeed      event.Feed
+	scope       event.SubscriptionScope
 	signer      types.Signer
 	mu          sync.RWMutex
 
@@ -297,7 +298,20 @@ func (pool *LegacyPool) Init(gasTip *big.Int, head *types.Header, reserve txpool
 
 	// Set the basic pool parameters
 	pool.gasTip.Store(gasTip)
-	pool.reset(nil, head)
+
+	// Initialize the state with head block, or fallback to empty one in
+	// case the head state is not available(might occur when node is not
+	// fully synced).
+	statedb, err := pool.chain.StateAt(head.Root)
+	if err != nil {
+		statedb, err = pool.chain.StateAt(types.EmptyRootHash)
+	}
+	if err != nil {
+		return err
+	}
+	pool.currentHead.Store(head)
+	pool.currentState = statedb
+	pool.pendingNonces = newNoncer(statedb)
 
 	// Start the reorg loop early, so it can handle requests generated during
 	// journal loading.
@@ -390,6 +404,9 @@ func (pool *LegacyPool) loop() {
 
 // Close terminates the transaction pool.
 func (pool *LegacyPool) Close() error {
+	// Unsubscribe all subscriptions registered from txpool
+	pool.scope.Close()
+
 	// Terminate the pool reorger and return
 	close(pool.reorgShutdownCh)
 	pool.wg.Wait()
@@ -411,7 +428,7 @@ func (pool *LegacyPool) Reset(oldHead, newHead *types.Header) {
 // SubscribeTransactions registers a subscription of NewTxsEvent and
 // starts sending event to the given channel.
 func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent) event.Subscription {
-	return pool.txFeed.Subscribe(ch)
+	return pool.scope.Track(pool.txFeed.Subscribe(ch))
 }
 
 // SetGasTip updates the minimum gas tip required by the transaction pool for a

core/txpool/txpool.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
-	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -56,9 +55,6 @@ type BlockChain interface {
 	// CurrentBlock returns the current head of the chain.
 	CurrentBlock() *types.Header
 
-	// HasState checks if state is present in the database or not.
-	HasState(common.Hash) bool
-
 	// SubscribeChainHeadEvent subscribes to new blocks being added to the chain.
 	SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
 }
@@ -69,8 +65,7 @@ type BlockChain interface {
 // They exit the pool when they are included in the blockchain or evicted due to
 // resource constraints.
 type TxPool struct {
-	inited   atomic.Bool // Flag whether the subpools are initialized
+	subpools []SubPool // List of subpools for specialized transaction handling
-	subpools []SubPool   // List of subpools for specialized transaction handling
 
 	reservations map[common.Address]SubPool // Map with the account to pool reservations
 	reserveLock  sync.Mutex                 // Lock protecting the account reservations
@@ -92,63 +87,18 @@ func New(gasTip *big.Int, chain BlockChain, subpools []SubPool) (*TxPool, error)
 		reservations: make(map[common.Address]SubPool),
 		quit:         make(chan chan error),
 	}
-	if chain.HasState(head.Root) {
+	for i, subpool := range subpools {
-		if err := pool.init(gasTip, head); err != nil {
+		if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil {
+			for j := i - 1; j >= 0; j-- {
+				subpools[j].Close()
+			}
 			return nil, err
 		}
-		go pool.loop(head, chain)
-	} else {
-		go pool.lazyInit(gasTip, chain)
 	}
+	go pool.loop(head, chain)
 	return pool, nil
 }
 
-// init performs the initialization for subpools.
-func (p *TxPool) init(gasTip *big.Int, head *types.Header) error {
-	for i, subpool := range p.subpools {
-		if err := subpool.Init(gasTip, head, p.reserver(i, subpool)); err != nil {
-			for j := i - 1; j >= 0; j-- {
-				p.subpools[j].Close()
-			}
-			return err
-		}
-	}
-	p.inited.Store(true)
-	log.Info("Initialized subpools", "head", head.Number, "hash", head.Hash())
-	return nil
-}
-
-// lazyInit waits the signal that state sync is completed and initializes the subpools.
-func (p *TxPool) lazyInit(gasTip *big.Int, chain BlockChain) {
-	var (
-		newHeadCh  = make(chan core.ChainHeadEvent)
-		newHeadSub = chain.SubscribeChainHeadEvent(newHeadCh)
-	)
-	defer newHeadSub.Unsubscribe()
-
-	var errc chan error
-	for errc == nil {
-		select {
-		case event := <-newHeadCh:
-			head := event.Block.Header()
-			if !chain.HasState(head.Root) {
-				continue // shouldn't happen
-			}
-			if err := p.init(gasTip, head); err != nil {
-				// TODO(rjl493456442) can we shutdown the node gracefully?
-				log.Crit("Failed to lazy init subpools", "err", err)
-			}
-			go p.loop(head, chain)
-			return
-
-		case errc = <-p.quit:
-			// Termination requested, break out on the next loop round
-		}
-	}
-	// Notify the closer of termination (no error possible for now)
-	errc <- nil
-}
-
 // reserver is a method to create an address reservation callback to exclusively
 // assign/deassign addresses to/from subpools. This can ensure that at any point
 // in time, only a single subpool is able to manage an account, avoiding cross
@@ -205,19 +155,15 @@ func (p *TxPool) Close() error {
 	if err := <-errc; err != nil {
 		errs = append(errs, err)
 	}
-	// Terminate each subpool if they are initialized
+
-	if p.inited.Load() {
+	// Terminate each subpool
-		for _, subpool := range p.subpools {
+	for _, subpool := range p.subpools {
-			if err := subpool.Close(); err != nil {
+		if err := subpool.Close(); err != nil {
-				errs = append(errs, err)
+			errs = append(errs, err)
-			}
 		}
 	}
-	// Terminate all the subpool subscriptions
-	p.subs.Close()
-
 	if len(errs) > 0 {
-		return fmt.Errorf("txpool close errors: %v", errs)
+		return fmt.Errorf("subpool close errors: %v", errs)
 	}
 	return nil
 }
@@ -286,10 +232,6 @@ func (p *TxPool) loop(head *types.Header, chain BlockChain) {
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
 // new transaction, and drops all transactions below this threshold.
 func (p *TxPool) SetGasTip(tip *big.Int) {
-	if !p.inited.Load() {
-		log.Info("Skip tip adjustment as txpool hasn't been initialized", "provided", tip)
-		return
-	}
 	for _, subpool := range p.subpools {
 		subpool.SetGasTip(tip)
 	}
@@ -298,9 +240,6 @@ func (p *TxPool) SetGasTip(tip *big.Int) {
 // Has returns an indicator whether the pool has a transaction cached with the
 // given hash.
 func (p *TxPool) Has(hash common.Hash) bool {
-	if !p.inited.Load() {
-		return false
-	}
 	for _, subpool := range p.subpools {
 		if subpool.Has(hash) {
 			return true
@@ -311,9 +250,6 @@ func (p *TxPool) Has(hash common.Hash) bool {
 
 // Get returns a transaction if it is contained in the pool, or nil otherwise.
 func (p *TxPool) Get(hash common.Hash) *types.Transaction {
-	if !p.inited.Load() {
-		return nil
-	}
 	for _, subpool := range p.subpools {
 		if tx := subpool.Get(hash); tx != nil {
 			return tx
@@ -326,13 +262,6 @@ func (p *TxPool) Get(hash common.Hash) *types.Transaction {
 // to the large transaction churn, add may postpone fully integrating the tx
 // to a later point to batch multiple ones together.
 func (p *TxPool) Add(txs []*types.Transaction, local bool, sync bool) []error {
-	if !p.inited.Load() {
-		errs := make([]error, len(txs))
-		for i := 0; i < len(errs); i++ {
-			errs[i] = errors.New("txpool is not initialized")
-		}
-		return errs
-	}
 	// Split the input transactions between the subpools. It shouldn't really
 	// happen that we receive merged batches, but better graceful than strange
 	// errors.
@@ -378,9 +307,6 @@ func (p *TxPool) Add(txs []*types.Transaction, local bool, sync bool) []error {
 // Pending retrieves all currently processable transactions, grouped by origin
 // account and sorted by nonce.
 func (p *TxPool) Pending(enforceTips bool) map[common.Address][]*LazyTransaction {
-	if !p.inited.Load() {
-		return nil
-	}
 	txs := make(map[common.Address][]*LazyTransaction)
 	for _, subpool := range p.subpools {
 		for addr, set := range subpool.Pending(enforceTips) {
@@ -403,9 +329,6 @@ func (p *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscrip
 // Nonce returns the next nonce of an account, with all transactions executable
 // by the pool already applied on top.
 func (p *TxPool) Nonce(addr common.Address) uint64 {
-	if !p.inited.Load() {
-		return 0
-	}
 	// Since (for now) accounts are unique to subpools, only one pool will have
 	// (at max) a non-state nonce. To avoid stateful lookups, just return the
 	// highest nonce for now.
@@ -421,9 +344,6 @@ func (p *TxPool) Nonce(addr common.Address) uint64 {
 // Stats retrieves the current pool stats, namely the number of pending and the
 // number of queued (non-executable) transactions.
 func (p *TxPool) Stats() (int, int) {
-	if !p.inited.Load() {
-		return 0, 0
-	}
 	var runnable, blocked int
 	for _, subpool := range p.subpools {
 		run, block := subpool.Stats()
@@ -437,9 +357,6 @@ func (p *TxPool) Stats() (int, int) {
 // Content retrieves the data content of the transaction pool, returning all the
 // pending as well as queued transactions, grouped by account and sorted by nonce.
 func (p *TxPool) Content() (map[common.Address][]*types.Transaction, map[common.Address][]*types.Transaction) {
-	if !p.inited.Load() {
-		return nil, nil
-	}
 	var (
 		runnable = make(map[common.Address][]*types.Transaction)
 		blocked  = make(map[common.Address][]*types.Transaction)
@@ -460,9 +377,6 @@ func (p *TxPool) Content() (map[common.Address][]*types.Transaction, map[common.
 // ContentFrom retrieves the data content of the transaction pool, returning the
 // pending as well as queued transactions of this address, grouped by nonce.
 func (p *TxPool) ContentFrom(addr common.Address) ([]*types.Transaction, []*types.Transaction) {
-	if !p.inited.Load() {
-		return nil, nil
-	}
 	for _, subpool := range p.subpools {
 		run, block := subpool.ContentFrom(addr)
 		if len(run) != 0 || len(block) != 0 {
@@ -474,9 +388,6 @@ func (p *TxPool) ContentFrom(addr common.Address) ([]*types.Transaction, []*type
 
 // Locals retrieves the accounts currently considered local by the pool.
 func (p *TxPool) Locals() []common.Address {
-	if !p.inited.Load() {
-		return nil
-	}
 	// Retrieve the locals from each subpool and deduplicate them
 	locals := make(map[common.Address]struct{})
 	for _, subpool := range p.subpools {
@@ -495,9 +406,6 @@ func (p *TxPool) Locals() []common.Address {
 // Status returns the known status (unknown/pending/queued) of a transaction
 // identified by their hashes.
 func (p *TxPool) Status(hash common.Hash) TxStatus {
-	if !p.inited.Load() {
-		return TxStatusUnknown
-	}
 	for _, subpool := range p.subpools {
 		if status := subpool.Status(hash); status != TxStatusUnknown {
 			return status
@@ -505,8 +413,3 @@ func (p *TxPool) Status(hash common.Hash) TxStatus {
 	}
 	return TxStatusUnknown
 }
-
-// Inited returns the indicator if txpool is fully initialized.
-func (p *TxPool) Inited() bool {
-	return p.inited.Load()
-}