From 24d73042b301329fe448df8e8bb4742ab7c90ed1 Mon Sep 17 00:00:00 2001
From: Gary Rong <garyrong0905@gmail.com>
Date: Sun, 24 Sep 2023 18:02:13 +0800
Subject: [PATCH] core: manage txpool subscription in mainpool

---
 core/txpool/blobpool/blobpool.go     | 10 +++-------
 core/txpool/legacypool/legacypool.go |  6 +-----
 core/txpool/txpool.go                |  4 +++-
 trie/triedb/pathdb/database.go       |  2 +-
 4 files changed, 8 insertions(+), 14 deletions(-)

diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go
index 042ff3be20..2615cfa558 100644
--- a/core/txpool/blobpool/blobpool.go
+++ b/core/txpool/blobpool/blobpool.go
@@ -307,10 +307,8 @@ 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
-	eventScope event.SubscriptionScope // Event scope to track and mass unsubscribe on termination
-
-	lock sync.RWMutex // Mutex protecting the pool during reorg handling
+	eventFeed event.Feed   // Event feed to send out new tx events on pool inclusion
+	lock      sync.RWMutex // Mutex protecting the pool during reorg handling
 }
 
 // New creates a new blob transaction pool to gather, sort and filter inbound
@@ -430,8 +428,6 @@ 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
@@ -1465,7 +1461,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.eventScope.Track(p.eventFeed.Subscribe(ch))
+	return p.eventFeed.Subscribe(ch)
 }
 
 // Nonce returns the next nonce of an account, with all transactions executable
diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go
index 00e326c4b8..611daa9d7f 100644
--- a/core/txpool/legacypool/legacypool.go
+++ b/core/txpool/legacypool/legacypool.go
@@ -208,7 +208,6 @@ type LegacyPool struct {
 	chain       BlockChain
 	gasTip      atomic.Pointer[big.Int]
 	txFeed      event.Feed
-	scope       event.SubscriptionScope
 	signer      types.Signer
 	mu          sync.RWMutex
 
@@ -391,9 +390,6 @@ 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()
@@ -415,7 +411,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.scope.Track(pool.txFeed.Subscribe(ch))
+	return pool.txFeed.Subscribe(ch)
 }
 
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go
index 1aed986c88..5c6db20757 100644
--- a/core/txpool/txpool.go
+++ b/core/txpool/txpool.go
@@ -205,7 +205,6 @@ 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() {
 		for _, subpool := range p.subpools {
@@ -214,6 +213,9 @@ func (p *TxPool) Close() error {
 			}
 		}
 	}
+	// Terminate all the subpool subscriptions
+	p.subs.Close()
+
 	if len(errs) > 0 {
 		return fmt.Errorf("txpool close errors: %v", errs)
 	}
diff --git a/trie/triedb/pathdb/database.go b/trie/triedb/pathdb/database.go
index ad8db9045e..0e441c3161 100644
--- a/trie/triedb/pathdb/database.go
+++ b/trie/triedb/pathdb/database.go
@@ -261,7 +261,7 @@ func (db *Database) Deactivate() error {
 
 	// Write the initial sync flag to persist it across restarts.
 	rawdb.WriteSnapSyncStatusFlag(db.diskdb, rawdb.StateSyncRunning)
-	log.Info("Disabled trie database due to ongoing sync")
+	log.Info("Disabled trie database due to state sync")
 	return nil
 }