core: Add feeds for dropped transactions

This adds a feed for dropped transactions, and the key scenarios
where transactions can be dropped to the feed as transactions are
dropped.

I'll be reviewing this in more depth to make sure I haven't missed
any scenarios, but this is the first pass.

core/events.go

@@ -24,6 +24,18 @@ import (
 // NewTxsEvent is posted when a batch of transactions enter the transaction pool.
 type NewTxsEvent struct{ Txs []*types.Transaction }
 
+// DropTxsEvent is posted when a batch of transactions are removed from the transaction pool
+type DropTxsEvent struct{
+	Txs []*types.Transaction
+	Reason string
+}
+
+// RejectedTxsEvent is posted when a batch of transactions are removed from the transaction pool
+type RejectedTxsEvent struct{
+	Txs []*types.Transaction
+	Reason string
+}
+
 // NewMinedBlockEvent is posted when a block has been imported.
 type NewMinedBlockEvent struct{ Block *types.Block }
 
@@ -41,3 +53,17 @@ type ChainSideEvent struct {
 }
 
 type ChainHeadEvent struct{ Block *types.Block }
+
+
+const (
+	dropUnderpriced = "underpriced-txs"
+	dropLowNonce = "low-nonce-txs"
+	dropUnpayable = "unpayable-txs"
+
+	dropAccountCap = "account-cap-txs" // Accounts exceeding txpool.accountslots transactions
+	dropReplaced = "replaced-txs"
+	dropUnexecutable = "unexecutable-txs"
+	dropTruncating = "truncating-txs"
+	dropOld = "old-txs"
+	dropGasPriceUpdated = "updated-gas-price"
+)

core/tx_pool.go

@@ -225,14 +225,16 @@ func (config *TxPoolConfig) sanitize() TxPoolConfig {
 // current state) and future transactions. Transactions move between those
 // two states over time as they are received and processed.
 type TxPool struct {
-	config      TxPoolConfig
+	config       TxPoolConfig
-	chainconfig *params.ChainConfig
+	chainconfig  *params.ChainConfig
-	chain       blockChain
+	chain        blockChain
-	gasPrice    *big.Int
+	gasPrice     *big.Int
-	txFeed      event.Feed
+	txFeed       event.Feed
-	scope       event.SubscriptionScope
+	dropTxFeed   event.Feed
-	signer      types.Signer
+	rejectTxFeed event.Feed
-	mu          sync.RWMutex
+	scope        event.SubscriptionScope
+	signer       types.Signer
+	mu           sync.RWMutex
 
 	istanbul bool // Fork indicator whether we are in the istanbul stage.
 
@@ -374,9 +376,14 @@ func (pool *TxPool) loop() {
 				}
 				// Any non-locals old enough should be removed
 				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
-					for _, tx := range pool.queue[addr].Flatten() {
+					txs := pool.queue[addr].Flatten()
+					for _, tx := range txs {
 						pool.removeTx(tx.Hash(), true)
 					}
+					pool.dropTxFeed.Send(DropTxsEvent{
+						Txs: txs,
+						Reason: dropOld,
+					})
 				}
 			}
 			pool.mu.Unlock()
@@ -415,6 +422,18 @@ func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- NewTxsEvent) event.Subscripti
 	return pool.scope.Track(pool.txFeed.Subscribe(ch))
 }
 
+// SubscribeDropTxsEvent registers a subscription of DropTxsEvent and
+// starts sending event to the given channel.
+func (pool *TxPool) SubscribeDropTxsEvent(ch chan<- DropTxsEvent) event.Subscription {
+	return pool.scope.Track(pool.dropTxFeed.Subscribe(ch))
+}
+
+// SubscribeRejectedTxsEvent registers a subscription of RejectedTxsEvent and
+// starts sending event to the given channel.
+func (pool *TxPool) SubscribeRejectedTxsEvent(ch chan<- RejectedTxsEvent) event.Subscription {
+	return pool.scope.Track(pool.rejectTxFeed.Subscribe(ch))
+}
+
 // GasPrice returns the current gas price enforced by the transaction pool.
 func (pool *TxPool) GasPrice() *big.Int {
 	pool.mu.RLock()
@@ -430,9 +449,14 @@ func (pool *TxPool) SetGasPrice(price *big.Int) {
 	defer pool.mu.Unlock()
 
 	pool.gasPrice = price
-	for _, tx := range pool.priced.Cap(price, pool.locals) {
+	txs := pool.priced.Cap(price, pool.locals)
+	for _, tx := range txs {
 		pool.removeTx(tx.Hash(), false)
 	}
+	pool.dropTxFeed.Send(DropTxsEvent{
+		Txs: txs,
+		Reason: dropGasPriceUpdated,
+	})
 	log.Info("Transaction pool price threshold updated", "price", price)
 }
 
@@ -605,6 +629,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 			underpricedTxMeter.Mark(1)
 			pool.removeTx(tx.Hash(), false)
 		}
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: drop,
+			Reason: dropUnderpriced,
+		})
 	}
 	// Try to replace an existing transaction in the pending pool
 	from, _ := types.Sender(pool.signer, tx) // already validated
@@ -620,6 +648,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (replaced bool, err e
 			pool.all.Remove(old.Hash())
 			pool.priced.Removed(1)
 			pendingReplaceMeter.Mark(1)
+			pool.dropTxFeed.Send(DropTxsEvent{
+				Txs: []*types.Transaction{old},
+				Reason: dropReplaced,
+			})
 		}
 		pool.all.Add(tx)
 		pool.priced.Put(tx)
@@ -669,6 +701,10 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
 		pool.all.Remove(old.Hash())
 		pool.priced.Removed(1)
 		queuedReplaceMeter.Mark(1)
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: []*types.Transaction{old},
+			Reason: dropReplaced,
+		})
 	} else {
 		// Nothing was replaced, bump the queued counter
 		queuedGauge.Inc(1)
@@ -718,6 +754,10 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 		pool.priced.Removed(1)
 
 		pendingReplaceMeter.Mark(1)
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: []*types.Transaction{old},
+			Reason: dropReplaced,
+		})
 	} else {
 		// Nothing was replaced, bump the pending counter
 		pendingGauge.Inc(1)
@@ -907,6 +947,10 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 			pool.pendingNonces.setIfLower(addr, tx.Nonce())
 			// Reduce the pending counter
 			pendingGauge.Dec(int64(1 + len(invalids)))
+			pool.dropTxFeed.Send(DropTxsEvent{
+				Txs: invalids,
+				Reason: dropUnexecutable,
+			})
 			return
 		}
 	}
@@ -1194,6 +1238,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 			pool.all.Remove(hash)
 			log.Trace("Removed old queued transaction", "hash", hash)
 		}
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: forwards,
+			Reason: dropLowNonce,
+		})
 		// Drop all transactions that are too costly (low balance or out of gas)
 		drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
 		for _, tx := range drops {
@@ -1202,6 +1250,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 			log.Trace("Removed unpayable queued transaction", "hash", hash)
 		}
 		queuedNofundsMeter.Mark(int64(len(drops)))
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: drops,
+			Reason: dropUnpayable,
+		})
 
 		// Gather all executable transactions and promote them
 		readies := list.Ready(pool.pendingNonces.get(addr))
@@ -1224,6 +1276,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
 				log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
 			}
 			queuedRateLimitMeter.Mark(int64(len(caps)))
+			pool.dropTxFeed.Send(DropTxsEvent{
+				Txs: caps,
+				Reason: dropAccountCap,
+			})
 		}
 		// Mark all the items dropped as removed
 		pool.priced.Removed(len(forwards) + len(drops) + len(caps))
@@ -1287,6 +1343,10 @@ func (pool *TxPool) truncatePending() {
 						pool.pendingNonces.setIfLower(offenders[i], tx.Nonce())
 						log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
 					}
+					pool.dropTxFeed.Send(DropTxsEvent{
+						Txs: caps,
+						Reason: dropAccountCap,
+					})
 					pool.priced.Removed(len(caps))
 					pendingGauge.Dec(int64(len(caps)))
 					if pool.locals.contains(offenders[i]) {
@@ -1314,6 +1374,10 @@ func (pool *TxPool) truncatePending() {
 					pool.pendingNonces.setIfLower(addr, tx.Nonce())
 					log.Trace("Removed fairness-exceeding pending transaction", "hash", hash)
 				}
+				pool.dropTxFeed.Send(DropTxsEvent{
+					Txs: caps,
+					Reason: dropAccountCap,
+				})
 				pool.priced.Removed(len(caps))
 				pendingGauge.Dec(int64(len(caps)))
 				if pool.locals.contains(addr) {
@@ -1354,9 +1418,14 @@ func (pool *TxPool) truncateQueue() {
 
 		// Drop all transactions if they are less than the overflow
 		if size := uint64(list.Len()); size <= drop {
-			for _, tx := range list.Flatten() {
+			txs := list.Flatten()
+			for _, tx := range txs {
 				pool.removeTx(tx.Hash(), true)
 			}
+			pool.dropTxFeed.Send(DropTxsEvent{
+				Txs: txs,
+				Reason: dropTruncating,
+			})
 			drop -= size
 			queuedRateLimitMeter.Mark(int64(size))
 			continue
@@ -1367,6 +1436,10 @@ func (pool *TxPool) truncateQueue() {
 			pool.removeTx(txs[i].Hash(), true)
 			drop--
 			queuedRateLimitMeter.Mark(1)
+			pool.dropTxFeed.Send(DropTxsEvent{
+				Txs: []*types.Transaction{txs[i]},
+				Reason: dropTruncating,
+			})
 		}
 	}
 }
@@ -1386,6 +1459,10 @@ func (pool *TxPool) demoteUnexecutables() {
 			pool.all.Remove(hash)
 			log.Trace("Removed old pending transaction", "hash", hash)
 		}
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: olds,
+			Reason: dropLowNonce,
+		})
 		// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
 		drops, invalids := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
 		for _, tx := range drops {
@@ -1393,6 +1470,10 @@ func (pool *TxPool) demoteUnexecutables() {
 			log.Trace("Removed unpayable pending transaction", "hash", hash)
 			pool.all.Remove(hash)
 		}
+		pool.dropTxFeed.Send(DropTxsEvent{
+			Txs: drops,
+			Reason: dropUnpayable,
+		})
 		pool.priced.Removed(len(olds) + len(drops))
 		pendingNofundsMeter.Mark(int64(len(drops)))
 

tests/testdata

@@ -1 +1 @@
-Subproject commit 7497b116a019beb26215cbea4028df068dea06be
+Subproject commit 6b85703b568f4456582a00665d8a3e5c3b20b484