feat(tx-pool): fast reject known skipped txs (#1004) (#1014)

* feat(tx-pool): fast reject known skipped txs  (#1004)

* feat(tx-pool): fast reject known skipped txs (#1001)

* feat(tx-pool): fast reject known skipped txs

* fix make lint

* add miner flag to tx pool

* bump version

* fix

---------

Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com>

cmd/geth/main.go

@@ -452,6 +452,7 @@ func startNode(ctx *cli.Context, stack *node.Node, backend ethapi.Backend, isCon
 		// Set the gas price to the limits from the CLI and start mining
 		gasprice := flags.GlobalBig(ctx, utils.MinerGasPriceFlag.Name)
 		ethBackend.TxPool().SetGasTip(gasprice)
+		ethBackend.TxPool().SetIsMiner(true)
 		if err := ethBackend.StartMining(); err != nil {
 			utils.Fatalf("Failed to start mining: %v", err)
 		}

core/rawdb/accessors_skipped_txs.go

@@ -158,6 +158,16 @@ func ReadSkippedTransaction(db ethdb.Reader, txHash common.Hash) *SkippedTransac
 	return &stxV2
 }
 
+// IsSkippedTransaction checks if a transaction exists as a skipped transaction in the database.
+func IsSkippedTransaction(db ethdb.Reader, txHash common.Hash) bool {
+	exists, err := db.Has(SkippedTransactionKey(txHash))
+	if err != nil {
+		log.Error("Failed to check skipped transaction", "hash", txHash.String(), "err", err)
+		return false
+	}
+	return exists
+}
+
 // writeSkippedTransactionHash writes the hash of a skipped transaction to the database.
 func writeSkippedTransactionHash(db ethdb.KeyValueWriter, index uint64, txHash common.Hash) {
 	if err := db.Put(SkippedTransactionHashKey(index), txHash[:]); err != nil {

core/txpool/blobpool/blobpool.go

@@ -1584,6 +1584,10 @@ func (p *BlobPool) RemoveTx(hash common.Hash, outofbound bool, unreserve bool) i
 func (p *BlobPool) PauseReorgs() {
 	log.Debug("skip BlobPool `PauseReorgs`")
 }
+
 func (p *BlobPool) ResumeReorgs() {
 	log.Debug("skip BlobPool `ResumeReorgs`")
 }
+
+func (pool *BlobPool) SetIsMiner(isMiner bool) {
+}

core/txpool/legacypool/legacypool.go

@@ -30,9 +30,11 @@ import (
 	"github.com/scroll-tech/go-ethereum/common/prque"
 	"github.com/scroll-tech/go-ethereum/consensus/misc/eip1559"
 	"github.com/scroll-tech/go-ethereum/core"
+	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/core/state"
 	"github.com/scroll-tech/go-ethereum/core/txpool"
 	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/event"
 	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/metrics"
@@ -81,6 +83,7 @@ var (
 
 	// General tx metrics
 	knownTxMeter        = metrics.NewRegisteredMeter("txpool/known", nil)
+	knownSkippedTxMeter = metrics.NewRegisteredMeter("txpool/known/skipped", nil)
 	validTxMeter        = metrics.NewRegisteredMeter("txpool/valid", nil)
 	invalidTxMeter      = metrics.NewRegisteredMeter("txpool/invalid", nil)
 	underpricedTxMeter  = metrics.NewRegisteredMeter("txpool/underpriced", nil)
@@ -122,6 +125,8 @@ type BlockChain interface {
 
 	// StateAt returns a state database for a given root hash (generally the head).
 	StateAt(root common.Hash) (*state.StateDB, error)
+
+	Database() ethdb.Database
 }
 
 // Config are the configuration parameters of the transaction pool.
@@ -239,6 +244,7 @@ type LegacyPool struct {
 
 	reorgPauseCh             chan bool // requests to pause scheduleReorgLoop
 	realTxActivityShutdownCh chan struct{}
+	isMiner                  atomic.Bool
 }
 
 type txpoolResetRequest struct {
@@ -480,6 +486,17 @@ func (pool *LegacyPool) SetGasTip(tip *big.Int) {
 	log.Info("Legacy pool tip threshold updated", "tip", tip)
 }
 
+// SetIsMiner updates the miner status of the node.
+func (pool *LegacyPool) SetIsMiner(isMiner bool) {
+	pool.isMiner.Store(isMiner)
+	log.Info("Transaction pool miner status updated", "isMiner", isMiner)
+}
+
+// IsMiner returns the current miner status of the node.
+func (pool *LegacyPool) IsMiner() bool {
+	return pool.isMiner.Load()
+}
+
 // Nonce returns the next nonce of an account, with all transactions executable
 // by the pool already applied on top.
 func (pool *LegacyPool) Nonce(addr common.Address) uint64 {
@@ -742,6 +759,13 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
 		knownTxMeter.Mark(1)
 		return false, txpool.ErrAlreadyKnown
 	}
+
+	if pool.IsMiner() && rawdb.IsSkippedTransaction(pool.chain.Database(), hash) {
+		log.Trace("Discarding already known skipped transaction", "hash", hash)
+		knownSkippedTxMeter.Mark(1)
+		return false, txpool.ErrAlreadyKnown
+	}
+
 	// Make the local flag. If it's from local source or it's from the network but
 	// the sender is marked as local previously, treat it as the local transaction.
 	isLocal := local || pool.locals.containsTx(tx)

core/txpool/legacypool/legacypool_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/core/txpool"
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/crypto"
+	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/event"
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/trie"
@@ -98,6 +99,10 @@ func (bc *testBlockChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent)
 	return bc.chainHeadFeed.Subscribe(ch)
 }
 
+func (bc *testBlockChain) Database() ethdb.Database {
+	return nil
+}
+
 func transaction(nonce uint64, gaslimit uint64, key *ecdsa.PrivateKey) *types.Transaction {
 	return pricedTransaction(nonce, gaslimit, big.NewInt(1), key)
 }

core/txpool/subpool.go

@@ -147,4 +147,5 @@ type SubPool interface {
 
 	PauseReorgs()
 	ResumeReorgs()
+	SetIsMiner(isMiner bool)
 }

core/txpool/txpool.go

@@ -455,3 +455,9 @@ func (pool *TxPool) StatsWithMinBaseFee(minBaseFee *big.Int) (pending int, queue
 	}
 	return pending, queued
 }
+
+func (pool *TxPool) SetIsMiner(isMiner bool) {
+	for _, subpool := range pool.subpools {
+		subpool.SetIsMiner(isMiner)
+	}
+}

miner/miner_test.go

@@ -111,6 +111,10 @@ func (bc *testBlockChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent)
 	return bc.chainHeadFeed.Subscribe(ch)
 }
 
+func (bc *testBlockChain) Database() ethdb.Database {
+	return nil
+}
+
 func TestMiner(t *testing.T) {
 	miner, mux, cleanup := createMiner(t)
 	defer cleanup(false)