*: use txFeed in SubscribeQueuedTransactions

Author: yysung1123

core/txpool/blobpool/blobpool.go

@@ -353,8 +353,6 @@ type BlobPool struct {
 	discoverFeed event.Feed // Event feed to send out new tx events on pool discovery (reorg excluded)
 	insertFeed   event.Feed // Event feed to send out new tx events on pool inclusion (reorg included)
 
-	queuedTxFeed event.Feed
-
 	lock sync.RWMutex // Mutex protecting the pool during reorg handling
 }
 
@@ -1744,8 +1742,6 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
 		return err
 	}
 
-	go p.queuedTxFeed.Send(core.NewQueuedTxsEvent{Txs: types.Transactions{tx}})
-
 	// If the address is not yet known, request exclusivity to track the account
 	// only by this subpool until all transactions are evicted
 	from, _ := types.Sender(p.signer, tx) // already validated above
@@ -2090,11 +2086,6 @@ func (p *BlobPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool
 	}
 }
 
-// SubscribeQueuedTransactions subscribes to new queued transaction events.
-func (p *BlobPool) SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
-	return p.queuedTxFeed.Subscribe(ch)
-}
-
 // Nonce returns the next nonce of an account, with all transactions executable
 // by the pool already applied on top.
 func (p *BlobPool) Nonce(addr common.Address) uint64 {

core/txpool/legacypool/legacypool.go

@@ -234,8 +234,6 @@ type LegacyPool struct {
 	signer      types.Signer
 	mu          sync.RWMutex
 
-	queuedTxFeed event.Feed
-
 	currentHead   atomic.Pointer[types.Header] // Current head of the blockchain
 	currentState  *state.StateDB               // Current state in the blockchain head
 	pendingNonces *noncer                      // Pending state tracking virtual nonces
@@ -406,11 +404,6 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs
 	return pool.txFeed.Subscribe(ch)
 }
 
-// SubscribeQueuedTransactions subscribes to new queued transaction events.
-func (pool *LegacyPool) SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
-	return pool.queuedTxFeed.Subscribe(ch)
-}
-
 // SetGasTip updates the minimum gas tip required by the transaction pool for a
 // new transaction, and drops all transactions below this threshold.
 func (pool *LegacyPool) SetGasTip(tip *big.Int) {
@@ -677,9 +670,6 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 	// already validated by this point
 	from, _ := types.Sender(pool.signer, tx)
 
-	// Broadcast a new tx anyway if it's valid
-	go pool.queuedTxFeed.Send(core.NewQueuedTxsEvent{Txs: types.Transactions{tx}})
-
 	// If the address is not yet known, request exclusivity to track the account
 	// only by this subpool until all transactions are evicted
 	var (

core/txpool/legacypool/legacypool_test.go

@@ -320,34 +320,6 @@ func validateEvents(events chan core.NewTxsEvent, count int) error {
 	return nil
 }
 
-// validateQueuedEvents checks that the correct number of transaction addition events
-// were fired on the pool's event feed.
-func validateQueuedEvents(events chan core.NewQueuedTxsEvent, count int) error {
-	var received []*types.Transaction
-
-	for len(received) < count {
-		select {
-		case ev := <-events:
-			received = append(received, ev.Txs...)
-		case <-time.After(time.Second):
-			return fmt.Errorf("event #%d not fired", len(received))
-		}
-	}
-	if len(received) > count {
-		return fmt.Errorf("more than %d events fired: %v", count, received[count:])
-	}
-	select {
-	case ev := <-events:
-		return fmt.Errorf("more than %d events fired: %v", count, ev.Txs)
-
-	case <-time.After(50 * time.Millisecond):
-		// This branch should be "default", but it's a data race between goroutines,
-		// reading the event channel and pushing into it, so better wait a bit ensuring
-		// really nothing gets injected.
-	}
-	return nil
-}
-
 func deriveSender(tx *types.Transaction) (common.Address, error) {
 	return types.Sender(types.HomesteadSigner{}, tx)
 }
@@ -469,145 +441,6 @@ func TestInvalidTransactions(t *testing.T) {
 	}
 }
 
-func TestSubscribePendingAndQueuedTransactions(t *testing.T) {
-	t.Parallel()
-
-	// Create the pool to test the pricing enforcement with
-	statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
-	blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
-
-	pool := New(testTxPoolConfig, blockchain)
-	pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver())
-	defer pool.Close()
-
-	// Keep listening for authenticated transactions.
-	events := make(chan core.NewQueuedTxsEvent, 32)
-	sub := pool.SubscribeQueuedTransactions(events)
-	defer sub.Unsubscribe()
-
-	createAccounts := func(numbers int, txpool *LegacyPool) []*ecdsa.PrivateKey {
-		keys := []*ecdsa.PrivateKey{}
-		for i := 0; i < numbers; i++ {
-			key, _ := crypto.GenerateKey()
-			txpool.currentState.AddBalance(crypto.PubkeyToAddress(key.PublicKey), uint256.NewInt(1000000), tracing.BalanceChangeUnspecified)
-
-			keys = append(keys, key)
-		}
-		return keys
-	}
-
-	tests := []struct {
-		name        string
-		setup       func(txPool *LegacyPool)
-		validEvents int
-	}{
-		{
-			name:        "valid transaction",
-			validEvents: 10,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(4, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(2), keys[0]))
-				txs = append(txs, pricedTransaction(1, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(2, 100000, big.NewInt(2), keys[0]))
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[1]))
-				txs = append(txs, pricedTransaction(1, 100000, big.NewInt(2), keys[1]))
-				txs = append(txs, pricedTransaction(2, 100000, big.NewInt(2), keys[1]))
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(2), keys[2]))
-				txs = append(txs, pricedTransaction(1, 100000, big.NewInt(1), keys[2]))
-				txs = append(txs, pricedTransaction(2, 100000, big.NewInt(2), keys[2]))
-
-				ltx := pricedTransaction(0, 100000, big.NewInt(1), keys[3])
-
-				pool.addRemotes(txs)
-				pool.addRemoteSync(ltx)
-			},
-		},
-
-		{
-			name:        "duplicate transaction",
-			validEvents: 1,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(1, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-
-				pool.addRemotesSync(txs)
-			},
-		},
-		{
-			name:        "bump gas price",
-			validEvents: 2,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(1, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(2), keys[0]))
-
-				pool.addRemotesSync(txs)
-			},
-		},
-		{
-			name:        "duplicate nonce with different gas price",
-			validEvents: 2,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(1, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(2), keys[0]))
-
-				pool.addRemotesSync(txs)
-			},
-		},
-		{
-			name:        "duplicate nonce with different gas limit",
-			validEvents: 2,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(1, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(0, 80000, big.NewInt(1), keys[0]))
-
-				pool.addRemotesSync(txs)
-			},
-		},
-		{
-			name:        "discontinuous nonce",
-			validEvents: 2,
-			setup: func(txPool *LegacyPool) {
-				keys := createAccounts(1, txPool)
-				txs := types.Transactions{}
-
-				txs = append(txs, pricedTransaction(0, 100000, big.NewInt(1), keys[0]))
-				txs = append(txs, pricedTransaction(10000, 80000, big.NewInt(1), keys[0]))
-
-				pool.addRemotesSync(txs)
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tt.setup(pool)
-
-			if err := validateQueuedEvents(events, tt.validEvents); err != nil {
-				t.Fatalf("event firing failed: %v", err)
-			}
-			if err := validatePoolInternals(pool); err != nil {
-				t.Fatalf("pool internal state corrupted: %v", err)
-			}
-		})
-	}
-}
-
 func TestQueue(t *testing.T) {
 	t.Parallel()
 

core/txpool/subpool.go

@@ -158,9 +158,6 @@ type SubPool interface {
 	// or also for reorged out ones.
 	SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription
 
-	// SubscribeQueuedTransactions subscribes to new queued transaction events.
-	SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription
-
 	// Nonce returns the next nonce of an account, with all transactions executable
 	// by the pool already applied on top.
 	Nonce(addr common.Address) uint64

core/txpool/txpool.go

@@ -385,15 +385,6 @@ func (p *TxPool) SubscribePendingLocalTxsEvent(ch chan<- core.PendingLocalTxsEve
 	return p.subs.Track(event.JoinSubscriptions(subs...))
 }
 
-// SubscribeNewQueuedTxsEvent registers a subscription of NewQueuedTxsEvent and
-func (p *TxPool) SubscribeNewQueuedTxsEvent(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
-	subs := make([]event.Subscription, len(p.subpools))
-	for i, subpool := range p.subpools {
-		subs[i] = subpool.SubscribeQueuedTransactions(ch)
-	}
-	return p.subs.Track(event.JoinSubscriptions(subs...))
-}
-
 // SubscribeTransactions registers a subscription for new transaction events,
 // supporting feeding only newly seen or also resurrected transactions.
 func (p *TxPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription {

eth/api_backend.go

@@ -403,10 +403,6 @@ func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.S
 	return b.eth.txPool.SubscribeTransactions(ch, true)
 }
 
-func (b *EthAPIBackend) SubscribeNewQueuedTxsEvent(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
-	return b.eth.txPool.SubscribeNewQueuedTxsEvent(ch)
-}
-
 func (b *EthAPIBackend) SyncProgress(ctx context.Context) ethereum.SyncProgress {
 	prog := b.eth.Downloader().Progress()
 	if txProg, err := b.eth.blockchain.TxIndexProgress(); err == nil {

eth/filters/api.go

@@ -205,75 +205,6 @@ func (api *FilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool)
 	return rpcSub, nil
 }
 
-// NewQueuedTransactionFilter creates a filter that fetches queued transaction hashes
-// as transactions enter the pending state.
-//
-// It is part of the filter package because this filter can be used through the
-// `eth_getFilterChanges` polling method that is also used for log filters.
-func (api *FilterAPI) NewQueuedTransactionFilter() rpc.ID {
-	var (
-		queuedTxs   = make(chan []*types.Transaction)
-		queuedTxSub = api.events.SubscribeQueuedTxs(queuedTxs)
-		deadline    = 5 * time.Minute
-	)
-
-	api.filtersMu.Lock()
-	api.filters[queuedTxSub.ID] = &filter{typ: QueuedTransactionsSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: queuedTxSub}
-	api.filtersMu.Unlock()
-
-	go func() {
-		for {
-			select {
-			case ph := <-queuedTxs:
-				api.filtersMu.Lock()
-				if f, found := api.filters[queuedTxSub.ID]; found {
-					f.txs = append(f.txs, ph...)
-				}
-				api.filtersMu.Unlock()
-			case <-queuedTxSub.Err():
-				api.filtersMu.Lock()
-				delete(api.filters, queuedTxSub.ID)
-				api.filtersMu.Unlock()
-				return
-			}
-		}
-	}()
-
-	return queuedTxSub.ID
-}
-
-// NewQueuedTransactions creates a subscription that is triggered each time a transaction
-// enters the transaction pool and was signed from one of the transactions this nodes manages.
-func (api *FilterAPI) NewQueuedTransactions(ctx context.Context) (*rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
-	}
-
-	rpcSub := notifier.CreateSubscription()
-
-	go func() {
-		txs := make(chan []*types.Transaction, 128)
-		queuedTxSub := api.events.SubscribeQueuedTxs(txs)
-
-		for {
-			select {
-			case hashes := <-txs:
-				// To keep the original behaviour, send a single tx hash in one notification.
-				// TODO(rjl493456442) Send a batch of tx hashes in one notification
-				for _, h := range hashes {
-					notifier.Notify(rpcSub.ID, h)
-				}
-			case <-rpcSub.Err():
-				queuedTxSub.Unsubscribe()
-				return
-			}
-		}
-	}()
-
-	return rpcSub, nil
-}
-
 // NewBlockFilter creates a filter that fetches blocks that are imported into the chain.
 // It is part of the filter package since polling goes with eth_getFilterChanges.
 func (api *FilterAPI) NewBlockFilter() rpc.ID {
@@ -639,10 +570,6 @@ func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
 				f.txs = nil
 				return hashes, nil
 			}
-		case QueuedTransactionsSubscription:
-			txs := f.txs
-			f.txs = nil
-			return returnTransactions(txs), nil
 		case LogsSubscription:
 			logs := f.logs
 			f.logs = nil
@@ -671,15 +598,6 @@ func returnLogs(logs []*types.Log) []*types.Log {
 	return logs
 }
 
-// returnTxs is a helper that will return an empty transaction array case the given transationcs array is nil,
-// otherwise the given transactions array is returned.
-func returnTransactions(txs []*types.Transaction) []*types.Transaction {
-	if txs == nil {
-		return []*types.Transaction{}
-	}
-	return txs
-}
-
 // UnmarshalJSON sets *args fields with given data.
 func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
 	type input struct {