From 950d5643b1926682ac7362df127d07f53e138350 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 6 Jun 2023 12:53:29 +0300 Subject: [PATCH] core/txpool: make transaction validation reusable across packages (pools) (#27429) * core/txpool: abstraction prep work for secondary pools (blob pool) * core/txpool: leave subpool concepts to a followup pr * les: fix tests using hard coded errors * core/txpool: use bitmaps instead of maps for tx type filtering --- cmd/geth/main.go | 2 +- core/txpool/txpool.go | 178 +++++++---------------- core/txpool/txpool2_test.go | 8 +- core/txpool/txpool_test.go | 79 ++++++----- core/txpool/validation.go | 225 ++++++++++++++++++++++++++++++ eth/api_miner.go | 2 +- eth/backend.go | 4 +- eth/protocols/eth/handler_test.go | 2 +- les/handler_test.go | 6 +- les/test_helper.go | 2 +- miner/miner_test.go | 2 +- miner/worker_test.go | 2 +- tests/fuzzers/les/les-fuzzer.go | 2 +- 13 files changed, 335 insertions(+), 179 deletions(-) create mode 100644 core/txpool/validation.go diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 92bb3852d..2289a72a1 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -419,7 +419,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().SetGasPrice(gasprice) + ethBackend.TxPool().SetGasTip(gasprice) if err := ethBackend.StartMining(); err != nil { utils.Fatalf("Failed to start mining: %v", err) } diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index ba71dfa88..139572540 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -18,7 +18,6 @@ package txpool import ( "errors" - "fmt" "math" "math/big" "sort" @@ -91,10 +90,6 @@ var ( // ErrFutureReplacePending is returned if a future transaction replaces a pending // transaction. Future transactions should only be able to replace other future transactions. ErrFutureReplacePending = errors.New("future transaction tries to replace pending") - - // ErrOverdraft is returned if a transaction would cause the senders balance to go negative - // thus invalidating a potential large number of transactions. - ErrOverdraft = errors.New("transaction would cause overdraft") ) var ( @@ -178,8 +173,7 @@ type Config struct { Lifetime time.Duration // Maximum amount of time non-executable transaction are queued } -// DefaultConfig contains the default configurations for the transaction -// pool. +// DefaultConfig contains the default configurations for the transaction pool. var DefaultConfig = Config{ Journal: "transactions.rlp", Rejournal: time.Hour, @@ -245,20 +239,15 @@ type TxPool struct { config Config chainconfig *params.ChainConfig chain blockChain - gasPrice *big.Int + gasTip atomic.Pointer[big.Int] txFeed event.Feed scope event.SubscriptionScope signer types.Signer mu sync.RWMutex - istanbul atomic.Bool // Fork indicator whether we are in the istanbul stage. - eip2718 atomic.Bool // Fork indicator whether we are using EIP-2718 type transactions. - eip1559 atomic.Bool // Fork indicator whether we are using EIP-1559 type transactions. - shanghai atomic.Bool // Fork indicator whether we are in the Shanghai stage. - - currentState *state.StateDB // Current state in the blockchain head - pendingNonces *noncer // Pending state tracking virtual nonces - currentMaxGas atomic.Uint64 // Current gas limit for transaction caps + 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 locals *accountSet // Set of local transaction to exempt from eviction rules journal *journal // Journal of local transaction to back up to disk @@ -286,9 +275,9 @@ type txpoolResetRequest struct { oldHead, newHead *types.Header } -// NewTxPool creates a new transaction pool to gather, sort and filter inbound +// New creates a new transaction pool to gather, sort and filter inbound // transactions from the network. -func NewTxPool(config Config, chainconfig *params.ChainConfig, chain blockChain) *TxPool { +func New(config Config, chainconfig *params.ChainConfig, chain blockChain) *TxPool { // Sanitize the input to ensure no vulnerable gas prices are set config = (&config).sanitize() @@ -309,8 +298,8 @@ func NewTxPool(config Config, chainconfig *params.ChainConfig, chain blockChain) reorgDoneCh: make(chan chan struct{}), reorgShutdownCh: make(chan struct{}), initDoneCh: make(chan struct{}), - gasPrice: new(big.Int).SetUint64(config.PriceLimit), } + pool.gasTip.Store(new(big.Int).SetUint64(config.PriceLimit)) pool.locals = newAccountSet(pool.signer) for _, addr := range config.Locals { log.Info("Setting new local account", "address", addr) @@ -443,33 +432,25 @@ func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subsc return pool.scope.Track(pool.txFeed.Subscribe(ch)) } -// GasPrice returns the current gas price enforced by the transaction pool. -func (pool *TxPool) GasPrice() *big.Int { - pool.mu.RLock() - defer pool.mu.RUnlock() - - return new(big.Int).Set(pool.gasPrice) -} - -// SetGasPrice updates the minimum price required by the transaction pool for a +// SetGasTip updates the minimum gas tip required by the transaction pool for a // new transaction, and drops all transactions below this threshold. -func (pool *TxPool) SetGasPrice(price *big.Int) { +func (pool *TxPool) SetGasTip(tip *big.Int) { pool.mu.Lock() defer pool.mu.Unlock() - old := pool.gasPrice - pool.gasPrice = price - // if the min miner fee increased, remove transactions below the new threshold - if price.Cmp(old) > 0 { + old := pool.gasTip.Load() + pool.gasTip.Store(new(big.Int).Set(tip)) + + // If the min miner fee increased, remove transactions below the new threshold + if tip.Cmp(old) > 0 { // pool.priced is sorted by GasFeeCap, so we have to iterate through pool.all instead - drop := pool.all.RemotesBelowTip(price) + drop := pool.all.RemotesBelowTip(tip) for _, tx := range drop { pool.removeTx(tx.Hash(), false) } pool.priced.Removed(len(drop)) } - - log.Info("Transaction pool price threshold updated", "price", price) + log.Info("Transaction pool tip threshold updated", "tip", tip) } // Nonce returns the next nonce of an account, with all transactions executable @@ -556,7 +537,7 @@ func (pool *TxPool) Pending(enforceTips bool) map[common.Address]types.Transacti // If the miner requests tip enforcement, cap the lists now if enforceTips && !pool.locals.contains(addr) { for i, tx := range txs { - if tx.EffectiveGasTipIntCmp(pool.gasPrice, pool.priced.urgent.baseFee) < 0 { + if tx.EffectiveGasTipIntCmp(pool.gasTip.Load(), pool.priced.urgent.baseFee) < 0 { txs = txs[:i] break } @@ -598,93 +579,48 @@ func (pool *TxPool) local() map[common.Address]types.Transactions { // This check is meant as an early check which only needs to be performed once, // and does not require the pool mutex to be held. func (pool *TxPool) validateTxBasics(tx *types.Transaction, local bool) error { - // Accept only legacy transactions until EIP-2718/2930 activates. - if !pool.eip2718.Load() && tx.Type() != types.LegacyTxType { - return core.ErrTxTypeNotSupported + opts := &ValidationOptions{ + Config: pool.chainconfig, + Accept: 0 | + 1< txMaxSize { - return ErrOversizedData - } - // Check whether the init code size has been exceeded. - if pool.shanghai.Load() && tx.To() == nil && len(tx.Data()) > params.MaxInitCodeSize { - return fmt.Errorf("%w: code size %v limit %v", core.ErrMaxInitCodeSizeExceeded, len(tx.Data()), params.MaxInitCodeSize) - } - // Transactions can't be negative. This may never happen using RLP decoded - // transactions but may occur if you create a transaction using the RPC. - if tx.Value().Sign() < 0 { - return ErrNegativeValue - } - // Ensure the transaction doesn't exceed the current block limit gas. - if pool.currentMaxGas.Load() < tx.Gas() { - return ErrGasLimit - } - // Sanity check for extremely large numbers - if tx.GasFeeCap().BitLen() > 256 { - return core.ErrFeeCapVeryHigh - } - if tx.GasTipCap().BitLen() > 256 { - return core.ErrTipVeryHigh - } - // Ensure gasFeeCap is greater than or equal to gasTipCap. - if tx.GasFeeCapIntCmp(tx.GasTipCap()) < 0 { - return core.ErrTipAboveFeeCap - } - // Make sure the transaction is signed properly. - if _, err := types.Sender(pool.signer, tx); err != nil { - return ErrInvalidSender - } - // Drop non-local transactions under our own minimal accepted gas price or tip - if !local && tx.GasTipCapIntCmp(pool.gasPrice) < 0 { - return ErrUnderpriced - } - // Ensure the transaction has more gas than the basic tx fee. - intrGas, err := core.IntrinsicGas(tx.Data(), tx.AccessList(), tx.To() == nil, true, pool.istanbul.Load(), pool.shanghai.Load()) - if err != nil { + if err := ValidateTransaction(tx, nil, nil, nil, pool.currentHead.Load(), pool.signer, opts); err != nil { return err } - if tx.Gas() < intrGas { - return core.ErrIntrinsicGas - } return nil } // validateTx checks whether a transaction is valid according to the consensus // rules and adheres to some heuristic limits of the local node (price and size). func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { - // Signature has been checked already, this cannot error. - from, _ := types.Sender(pool.signer, tx) - // Ensure the transaction adheres to nonce ordering - if pool.currentState.GetNonce(from) > tx.Nonce() { - return core.ErrNonceTooLow - } - // Transactor should have enough funds to cover the costs - // cost == V + GP * GL - balance := pool.currentState.GetBalance(from) - if balance.Cmp(tx.Cost()) < 0 { - return core.ErrInsufficientFunds - } + opts := &ValidationOptionsWithState{ + State: pool.currentState, - // Verify that replacing transactions will not result in overdraft - list := pool.pending[from] - if list != nil { // Sender already has pending txs - sum := new(big.Int).Add(tx.Cost(), list.totalcost) - if repl := list.txs.Get(tx.Nonce()); repl != nil { - // Deduct the cost of a transaction replaced by this - sum.Sub(sum, repl.Cost()) - } - if balance.Cmp(sum) < 0 { - log.Trace("Replacing transactions would overdraft", "sender", from, "balance", pool.currentState.GetBalance(from), "required", sum) - return ErrOverdraft - } + FirstNonceGap: nil, // Pool allows arbitrary arrival order, don't invalidate nonce gaps + ExistingExpenditure: func(addr common.Address) *big.Int { + if list := pool.pending[addr]; list != nil { + return list.totalcost + } + return new(big.Int) + }, + ExistingCost: func(addr common.Address, nonce uint64) *big.Int { + if list := pool.pending[addr]; list != nil { + if tx := list.txs.Get(nonce); tx != nil { + return tx.Cost() + } + } + return nil + }, + } + if err := ValidateTransactionWithState(tx, pool.signer, opts); err != nil { + return err } return nil } @@ -995,7 +931,6 @@ func (pool *TxPool) addTxs(txs []*types.Transaction, local, sync bool) []error { // Exclude transactions with basic errors, e.g invalid signatures and // insufficient intrinsic gas as soon as possible and cache senders // in transactions before obtaining lock - if err := pool.validateTxBasics(tx, local); err != nil { errs[i] = err invalidTxMeter.Mark(1) @@ -1385,21 +1320,14 @@ func (pool *TxPool) reset(oldHead, newHead *types.Header) { log.Error("Failed to reset txpool state", "err", err) return } + pool.currentHead.Store(newHead) pool.currentState = statedb pool.pendingNonces = newNoncer(statedb) - pool.currentMaxGas.Store(newHead.GasLimit) // Inject any transactions discarded due to reorgs log.Debug("Reinjecting stale transactions", "count", len(reinject)) core.SenderCacher.Recover(pool.signer, reinject) pool.addTxsLocked(reinject, false) - - // Update all fork indicator by next pending block number. - next := new(big.Int).Add(newHead.Number, big.NewInt(1)) - pool.istanbul.Store(pool.chainconfig.IsIstanbul(next)) - pool.eip2718.Store(pool.chainconfig.IsBerlin(next)) - pool.eip1559.Store(pool.chainconfig.IsLondon(next)) - pool.shanghai.Store(pool.chainconfig.IsShanghai(next, uint64(time.Now().Unix()))) } // promoteExecutables moves transactions that have become processable from the @@ -1410,6 +1338,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans var promoted []*types.Transaction // Iterate over all accounts and promote any executable transactions + gasLimit := pool.currentHead.Load().GasLimit for _, addr := range accounts { list := pool.queue[addr] if list == nil { @@ -1423,7 +1352,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans } log.Trace("Removed old queued transactions", "count", len(forwards)) // Drop all transactions that are too costly (low balance or out of gas) - drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas.Load()) + drops, _ := list.Filter(pool.currentState.GetBalance(addr), gasLimit) for _, tx := range drops { hash := tx.Hash() pool.all.Remove(hash) @@ -1609,6 +1538,7 @@ func (pool *TxPool) truncateQueue() { // to trigger a re-heap is this function func (pool *TxPool) demoteUnexecutables() { // Iterate over all accounts and demote any non-executable transactions + gasLimit := pool.currentHead.Load().GasLimit for addr, list := range pool.pending { nonce := pool.currentState.GetNonce(addr) @@ -1620,7 +1550,7 @@ func (pool *TxPool) demoteUnexecutables() { log.Trace("Removed old pending transaction", "hash", hash) } // 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.Load()) + drops, invalids := list.Filter(pool.currentState.GetBalance(addr), gasLimit) for _, tx := range drops { hash := tx.Hash() log.Trace("Removed unpayable pending transaction", "hash", hash) diff --git a/core/txpool/txpool2_test.go b/core/txpool/txpool2_test.go index d7ae3bdb1..b67905056 100644 --- a/core/txpool/txpool2_test.go +++ b/core/txpool/txpool2_test.go @@ -83,7 +83,7 @@ func TestTransactionFutureAttack(t *testing.T) { config := testTxPoolConfig config.GlobalQueue = 100 config.GlobalSlots = 100 - pool := NewTxPool(config, eip1559Config, blockchain) + pool := New(config, eip1559Config, blockchain) defer pool.Stop() fillPool(t, pool) pending, _ := pool.Stats() @@ -116,7 +116,7 @@ func TestTransactionFuture1559(t *testing.T) { // Create the pool to test the pricing enforcement with statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, eip1559Config, blockchain) + pool := New(testTxPoolConfig, eip1559Config, blockchain) defer pool.Stop() // Create a number of test accounts, fund them and make transactions @@ -148,7 +148,7 @@ func TestTransactionZAttack(t *testing.T) { // Create the pool to test the pricing enforcement with statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, eip1559Config, blockchain) + pool := New(testTxPoolConfig, eip1559Config, blockchain) defer pool.Stop() // Create a number of test accounts, fund them and make transactions fillPool(t, pool) @@ -218,7 +218,7 @@ func BenchmarkFutureAttack(b *testing.B) { config := testTxPoolConfig config.GlobalQueue = 100 config.GlobalSlots = 100 - pool := NewTxPool(config, eip1559Config, blockchain) + pool := New(config, eip1559Config, blockchain) defer pool.Stop() fillPool(b, pool) diff --git a/core/txpool/txpool_test.go b/core/txpool/txpool_test.go index 22e106aaf..319e25bea 100644 --- a/core/txpool/txpool_test.go +++ b/core/txpool/txpool_test.go @@ -130,7 +130,7 @@ func setupPoolWithConfig(config *params.ChainConfig) (*TxPool, *ecdsa.PrivateKey blockchain := newTestBlockChain(10000000, statedb, new(event.Feed)) key, _ := crypto.GenerateKey() - pool := NewTxPool(testTxPoolConfig, config, blockchain) + pool := New(testTxPoolConfig, config, blockchain) // wait for the pool to initialize <-pool.initDoneCh @@ -247,7 +247,7 @@ func TestStateChangeDuringReset(t *testing.T) { tx0 := transaction(0, 100000, key) tx1 := transaction(1, 100000, key) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() nonce := pool.Nonce(address) @@ -313,7 +313,7 @@ func TestInvalidTransactions(t *testing.T) { } tx = transaction(1, 100000, key) - pool.gasPrice = big.NewInt(1000) + pool.gasTip.Store(big.NewInt(1000)) if err, want := pool.AddRemote(tx), ErrUnderpriced; !errors.Is(err, want) { t.Errorf("want %v have %v", want, err) } @@ -666,7 +666,7 @@ func TestPostponing(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() // Create two test accounts to produce different gap profiles with @@ -882,7 +882,7 @@ func testQueueGlobalLimiting(t *testing.T, nolocals bool) { config.NoLocals = nolocals config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible) - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Create a number of test accounts and fund them (last one will be the local) @@ -974,7 +974,7 @@ func testQueueTimeLimiting(t *testing.T, nolocals bool) { config.Lifetime = time.Second config.NoLocals = nolocals - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Create two test accounts to ensure remotes expire but locals do not @@ -1158,7 +1158,7 @@ func TestPendingGlobalLimiting(t *testing.T) { config := testTxPoolConfig config.GlobalSlots = config.AccountSlots * 10 - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Create a number of test accounts and fund them @@ -1210,7 +1210,7 @@ func TestAllowedTxSize(t *testing.T) { // // It is assumed the fields in the transaction (except of the data) are: // - nonce <= 32 bytes - // - gasPrice <= 32 bytes + // - gasTip <= 32 bytes // - gasLimit <= 32 bytes // - recipient == 20 bytes // - value <= 32 bytes @@ -1218,22 +1218,21 @@ func TestAllowedTxSize(t *testing.T) { // All those fields are summed up to at most 213 bytes. baseSize := uint64(213) dataSize := txMaxSize - baseSize - maxGas := pool.currentMaxGas.Load() // Try adding a transaction with maximal allowed size - tx := pricedDataTransaction(0, maxGas, big.NewInt(1), key, dataSize) + tx := pricedDataTransaction(0, pool.currentHead.Load().GasLimit, big.NewInt(1), key, dataSize) if err := pool.addRemoteSync(tx); err != nil { t.Fatalf("failed to add transaction of size %d, close to maximal: %v", int(tx.Size()), err) } // Try adding a transaction with random allowed size - if err := pool.addRemoteSync(pricedDataTransaction(1, maxGas, big.NewInt(1), key, uint64(rand.Intn(int(dataSize))))); err != nil { + if err := pool.addRemoteSync(pricedDataTransaction(1, pool.currentHead.Load().GasLimit, big.NewInt(1), key, uint64(rand.Intn(int(dataSize))))); err != nil { t.Fatalf("failed to add transaction of random allowed size: %v", err) } // Try adding a transaction of minimal not allowed size - if err := pool.addRemoteSync(pricedDataTransaction(2, maxGas, big.NewInt(1), key, txMaxSize)); err == nil { + if err := pool.addRemoteSync(pricedDataTransaction(2, pool.currentHead.Load().GasLimit, big.NewInt(1), key, txMaxSize)); err == nil { t.Fatalf("expected rejection on slightly oversize transaction") } // Try adding a transaction of random not allowed size - if err := pool.addRemoteSync(pricedDataTransaction(2, maxGas, big.NewInt(1), key, dataSize+1+uint64(rand.Intn(10*txMaxSize)))); err == nil { + if err := pool.addRemoteSync(pricedDataTransaction(2, pool.currentHead.Load().GasLimit, big.NewInt(1), key, dataSize+1+uint64(rand.Intn(10*txMaxSize)))); err == nil { t.Fatalf("expected rejection on oversize transaction") } // Run some sanity checks on the pool internals @@ -1262,7 +1261,7 @@ func TestCapClearsFromAll(t *testing.T) { config.AccountQueue = 2 config.GlobalSlots = 8 - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Create a number of test accounts and fund them @@ -1294,7 +1293,7 @@ func TestPendingMinimumAllowance(t *testing.T) { config := testTxPoolConfig config.GlobalSlots = 1 - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Create a number of test accounts and fund them @@ -1339,7 +1338,7 @@ func TestRepricing(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() // Keep track of transaction events to ensure all executables get announced @@ -1388,7 +1387,7 @@ func TestRepricing(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Reprice the pool and check that underpriced transactions get dropped - pool.SetGasPrice(big.NewInt(2)) + pool.SetGasTip(big.NewInt(2)) pending, queued = pool.Stats() if pending != 2 { @@ -1404,13 +1403,13 @@ func TestRepricing(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Check that we can't add the old transactions back - if err := pool.AddRemote(pricedTransaction(1, 100000, big.NewInt(1), keys[0])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(1, 100000, big.NewInt(1), keys[0])); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } - if err := pool.AddRemote(pricedTransaction(0, 100000, big.NewInt(1), keys[1])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(0, 100000, big.NewInt(1), keys[1])); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } - if err := pool.AddRemote(pricedTransaction(2, 100000, big.NewInt(1), keys[2])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(2, 100000, big.NewInt(1), keys[2])); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced queued transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } if err := validateEvents(events, 0); err != nil { @@ -1509,7 +1508,7 @@ func TestRepricingDynamicFee(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Reprice the pool and check that underpriced transactions get dropped - pool.SetGasPrice(big.NewInt(2)) + pool.SetGasTip(big.NewInt(2)) pending, queued = pool.Stats() if pending != 2 { @@ -1526,15 +1525,15 @@ func TestRepricingDynamicFee(t *testing.T) { } // Check that we can't add the old transactions back tx := pricedTransaction(1, 100000, big.NewInt(1), keys[0]) - if err := pool.AddRemote(tx); err != ErrUnderpriced { + if err := pool.AddRemote(tx); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } tx = dynamicFeeTx(0, 100000, big.NewInt(2), big.NewInt(1), keys[1]) - if err := pool.AddRemote(tx); err != ErrUnderpriced { + if err := pool.AddRemote(tx); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } tx = dynamicFeeTx(2, 100000, big.NewInt(1), big.NewInt(1), keys[2]) - if err := pool.AddRemote(tx); err != ErrUnderpriced { + if err := pool.AddRemote(tx); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced queued transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } if err := validateEvents(events, 0); err != nil { @@ -1587,7 +1586,7 @@ func TestRepricingKeepsLocals(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, eip1559Config, blockchain) + pool := New(testTxPoolConfig, eip1559Config, blockchain) defer pool.Stop() // Create a number of test accounts and fund them @@ -1638,13 +1637,13 @@ func TestRepricingKeepsLocals(t *testing.T) { validate() // Reprice the pool and check that nothing is dropped - pool.SetGasPrice(big.NewInt(2)) + pool.SetGasTip(big.NewInt(2)) validate() - pool.SetGasPrice(big.NewInt(2)) - pool.SetGasPrice(big.NewInt(4)) - pool.SetGasPrice(big.NewInt(8)) - pool.SetGasPrice(big.NewInt(100)) + pool.SetGasTip(big.NewInt(2)) + pool.SetGasTip(big.NewInt(4)) + pool.SetGasTip(big.NewInt(8)) + pool.SetGasTip(big.NewInt(100)) validate() } @@ -1664,7 +1663,7 @@ func TestUnderpricing(t *testing.T) { config.GlobalSlots = 2 config.GlobalQueue = 2 - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Keep track of transaction events to ensure all executables get announced @@ -1706,7 +1705,7 @@ func TestUnderpricing(t *testing.T) { t.Fatalf("pool internal state corrupted: %v", err) } // Ensure that adding an underpriced transaction on block limit fails - if err := pool.AddRemote(pricedTransaction(0, 100000, big.NewInt(1), keys[1])); err != ErrUnderpriced { + if err := pool.AddRemote(pricedTransaction(0, 100000, big.NewInt(1), keys[1])); !errors.Is(err, ErrUnderpriced) { t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } // Replace a future transaction with a future transaction @@ -1778,7 +1777,7 @@ func TestStableUnderpricing(t *testing.T) { config.GlobalSlots = 128 config.GlobalQueue = 0 - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) defer pool.Stop() // Keep track of transaction events to ensure all executables get announced @@ -1886,7 +1885,7 @@ func TestUnderpricingDynamicFee(t *testing.T) { // Ensure that adding an underpriced transaction fails tx := dynamicFeeTx(0, 100000, big.NewInt(2), big.NewInt(1), keys[1]) - if err := pool.AddRemote(tx); err != ErrUnderpriced { // Pend K0:0, K0:1, K2:0; Que K1:1 + if err := pool.AddRemote(tx); !errors.Is(err, ErrUnderpriced) { // Pend K0:0, K0:1, K2:0; Que K1:1 t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced) } @@ -2006,7 +2005,7 @@ func TestDeduplication(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() // Create a test account to add transactions with @@ -2072,7 +2071,7 @@ func TestReplacement(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() // Keep track of transaction events to ensure all executables get announced @@ -2282,7 +2281,7 @@ func testJournaling(t *testing.T, nolocals bool) { config.Journal = journal config.Rejournal = time.Second - pool := NewTxPool(config, params.TestChainConfig, blockchain) + pool := New(config, params.TestChainConfig, blockchain) // Create two test accounts to ensure remotes expire but locals do not local, _ := crypto.GenerateKey() @@ -2319,7 +2318,7 @@ func testJournaling(t *testing.T, nolocals bool) { statedb.SetNonce(crypto.PubkeyToAddress(local.PublicKey), 1) blockchain = newTestBlockChain(1000000, statedb, new(event.Feed)) - pool = NewTxPool(config, params.TestChainConfig, blockchain) + pool = New(config, params.TestChainConfig, blockchain) pending, queued = pool.Stats() if queued != 0 { @@ -2345,7 +2344,7 @@ func testJournaling(t *testing.T, nolocals bool) { statedb.SetNonce(crypto.PubkeyToAddress(local.PublicKey), 1) blockchain = newTestBlockChain(1000000, statedb, new(event.Feed)) - pool = NewTxPool(config, params.TestChainConfig, blockchain) + pool = New(config, params.TestChainConfig, blockchain) pending, queued = pool.Stats() if pending != 0 { @@ -2375,7 +2374,7 @@ func TestStatusCheck(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()), nil) blockchain := newTestBlockChain(1000000, statedb, new(event.Feed)) - pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) + pool := New(testTxPoolConfig, params.TestChainConfig, blockchain) defer pool.Stop() // Create the test accounts to check various transaction statuses with diff --git a/core/txpool/validation.go b/core/txpool/validation.go new file mode 100644 index 000000000..af678277f --- /dev/null +++ b/core/txpool/validation.go @@ -0,0 +1,225 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package txpool + +import ( + "crypto/sha256" + "fmt" + "math/big" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto/kzg4844" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/params" +) + +// ValidationOptions define certain differences between transaction validation +// across the different pools without having to duplicate those checks. +type ValidationOptions struct { + Config *params.ChainConfig // Chain configuration to selectively validate based on current fork rules + + Accept uint8 // Bitmap of transaction types that should be accepted for the calling pool + MaxSize uint64 // Maximum size of a transaction that the caller can meaningfully handle + MinTip *big.Int // Minimum gas tip needed to allow a transaction into the caller pool +} + +// ValidateTransaction is a helper method to check whether a transaction is valid +// according to the consensus rules, but does not check state-dependent validation +// (balance, nonce, etc). +// +// This check is public to allow different transaction pools to check the basic +// rules without duplicating code and running the risk of missed updates. +func ValidateTransaction(tx *types.Transaction, blobs []kzg4844.Blob, commits []kzg4844.Commitment, proofs []kzg4844.Proof, head *types.Header, signer types.Signer, opts *ValidationOptions) error { + // Ensure transactions not implemented by the calling pool are rejected + if opts.Accept&(1< opts.MaxSize { + return fmt.Errorf("%w: transaction size %v, limit %v", ErrOversizedData, tx.Size(), opts.MaxSize) + } + // Ensure only transactions that have been enabled are accepted + if !opts.Config.IsBerlin(head.Number) && tx.Type() != types.LegacyTxType { + return fmt.Errorf("%w: type %d rejected, pool not yet in Berlin", core.ErrTxTypeNotSupported, tx.Type()) + } + if !opts.Config.IsLondon(head.Number) && tx.Type() == types.DynamicFeeTxType { + return fmt.Errorf("%w: type %d rejected, pool not yet in London", core.ErrTxTypeNotSupported, tx.Type()) + } + if !opts.Config.IsCancun(head.Number, head.Time) && tx.Type() == types.BlobTxType { + return fmt.Errorf("%w: type %d rejected, pool not yet in Cancun", core.ErrTxTypeNotSupported, tx.Type()) + } + // Check whether the init code size has been exceeded + if opts.Config.IsShanghai(head.Number, head.Time) && tx.To() == nil && len(tx.Data()) > params.MaxInitCodeSize { + return fmt.Errorf("%w: code size %v, limit %v", core.ErrMaxInitCodeSizeExceeded, len(tx.Data()), params.MaxInitCodeSize) + } + // Transactions can't be negative. This may never happen using RLP decoded + // transactions but may occur for transactions created using the RPC. + if tx.Value().Sign() < 0 { + return ErrNegativeValue + } + // Ensure the transaction doesn't exceed the current block limit gas + if head.GasLimit < tx.Gas() { + return ErrGasLimit + } + // Sanity check for extremely large numbers (supported by RLP or RPC) + if tx.GasFeeCap().BitLen() > 256 { + return core.ErrFeeCapVeryHigh + } + if tx.GasTipCap().BitLen() > 256 { + return core.ErrTipVeryHigh + } + // Ensure gasFeeCap is greater than or equal to gasTipCap + if tx.GasFeeCapIntCmp(tx.GasTipCap()) < 0 { + return core.ErrTipAboveFeeCap + } + // Make sure the transaction is signed properly + if _, err := types.Sender(signer, tx); err != nil { + return ErrInvalidSender + } + // Ensure the transaction has more gas than the bare minimum needed to cover + // the transaction metadata + intrGas, err := core.IntrinsicGas(tx.Data(), tx.AccessList(), tx.To() == nil, true, opts.Config.IsIstanbul(head.Number), opts.Config.IsShanghai(head.Number, head.Time)) + if err != nil { + return err + } + if tx.Gas() < intrGas { + return fmt.Errorf("%w: needed %v, allowed %v", core.ErrIntrinsicGas, intrGas, tx.Gas()) + } + // Ensure the gasprice is high enough to cover the requirement of the calling + // pool and/or block producer + if tx.GasTipCapIntCmp(opts.MinTip) < 0 { + return fmt.Errorf("%w: tip needed %v, tip permitted %v", ErrUnderpriced, opts.MinTip, tx.GasTipCap()) + } + // Ensure blob transactions have valid commitments + if tx.Type() == types.BlobTxType { + // Ensure the number of items in the blob transaction and vairous side + // data match up before doing any expensive validations + hashes := tx.BlobHashes() + if len(hashes) == 0 { + return fmt.Errorf("blobless blob transaction") + } + if len(hashes) > params.BlobTxMaxDataGasPerBlock/params.BlobTxDataGasPerBlob { + return fmt.Errorf("too many blobs in transaction: have %d, permitted %d", len(hashes), params.BlobTxMaxDataGasPerBlock/params.BlobTxDataGasPerBlob) + } + if len(blobs) != len(hashes) { + return fmt.Errorf("invalid number of %d blobs compared to %d blob hashes", len(blobs), len(hashes)) + } + if len(commits) != len(hashes) { + return fmt.Errorf("invalid number of %d blob commitments compared to %d blob hashes", len(commits), len(hashes)) + } + if len(proofs) != len(hashes) { + return fmt.Errorf("invalid number of %d blob proofs compared to %d blob hashes", len(proofs), len(hashes)) + } + // Blob quantities match up, validate that the provers match with the + // transaction hash before getting to the cryptography + hasher := sha256.New() + for i, want := range hashes { + hasher.Write(commits[i][:]) + hash := hasher.Sum(nil) + hasher.Reset() + + var vhash common.Hash + vhash[0] = params.BlobTxHashVersion + copy(vhash[1:], hash[1:]) + + if vhash != want { + return fmt.Errorf("blob %d: computed hash %#x mismatches transaction one %#x", i, vhash, want) + } + } + // Blob commitments match with the hashes in the transaction, verify the + // blobs themselves via KZG + for i := range blobs { + if err := kzg4844.VerifyBlobProof(blobs[i], commits[i], proofs[i]); err != nil { + return fmt.Errorf("invalid blob %d: %v", i, err) + } + } + } + return nil +} + +// ValidationOptionsWithState define certain differences between stateful transaction +// validation across the different pools without having to duplicate those checks. +type ValidationOptionsWithState struct { + State *state.StateDB // State database to check nonces and balances against + + // FirstNonceGap is an optional callback to retrieve the first nonce gap in + // the list of pooled transactions of a specific account. If this method is + // set, nonce gaps will be checked and forbidden. If this method is not set, + // nonce gaps will be ignored and permitted. + FirstNonceGap func(addr common.Address) uint64 + + // ExistingExpenditure is a mandatory callback to retrieve the cummulative + // cost of the already pooled transactions to check for overdrafts. + ExistingExpenditure func(addr common.Address) *big.Int + + // ExistingCost is a mandatory callback to retrieve an already pooled + // transaction's cost with the given nonce to check for overdrafts. + ExistingCost func(addr common.Address, nonce uint64) *big.Int +} + +// ValidateTransactionWithState is a helper method to check whether a transaction +// is valid according to the pool's internal state checks (balance, nonce, gaps). +// +// This check is public to allow different transaction pools to check the stateful +// rules without duplicating code and running the risk of missed updates. +func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, opts *ValidationOptionsWithState) error { + // Ensure the transaction adheres to nonce ordering + from, err := signer.Sender(tx) // already validated (and cached), but cleaner to check + if err != nil { + log.Error("Transaction sender recovery failed", "err", err) + return err + } + next := opts.State.GetNonce(from) + if next > tx.Nonce() { + return fmt.Errorf("%w: next nonce %v, tx nonce %v", core.ErrNonceTooLow, next, tx.Nonce()) + } + // Ensure the transaction doesn't produce a nonce gap in pools that do not + // support arbitrary orderings + if opts.FirstNonceGap != nil { + if gap := opts.FirstNonceGap(from); gap < tx.Nonce() { + return fmt.Errorf("%w: tx nonce %v, gapped nonce %v", core.ErrNonceTooHigh, tx.Nonce(), gap) + } + } + // Ensure the transactor has enough funds to cover the transaction costs + var ( + balance = opts.State.GetBalance(from) + cost = tx.Cost() + ) + if balance.Cmp(cost) < 0 { + return fmt.Errorf("%w: balance %v, tx cost %v, overshot %v", core.ErrInsufficientFunds, balance, cost, new(big.Int).Sub(cost, balance)) + } + // Ensure the transactor has enough funds to cover for replacements or nonce + // expansions without overdrafts + spent := opts.ExistingExpenditure(from) + if prev := opts.ExistingCost(from, tx.Nonce()); prev != nil { + bump := new(big.Int).Sub(cost, prev) + need := new(big.Int).Add(spent, bump) + if balance.Cmp(need) < 0 { + return fmt.Errorf("%w: balance %v, queued cost %v, tx bumped %v, overshot %v", core.ErrInsufficientFunds, balance, spent, bump, new(big.Int).Sub(need, balance)) + } + } else { + need := new(big.Int).Add(spent, cost) + if balance.Cmp(need) < 0 { + return fmt.Errorf("%w: balance %v, queued cost %v, tx cost %v, overshot %v", core.ErrInsufficientFunds, balance, spent, cost, new(big.Int).Sub(need, balance)) + } + } + return nil +} diff --git a/eth/api_miner.go b/eth/api_miner.go index cd0c1e5c0..477531d49 100644 --- a/eth/api_miner.go +++ b/eth/api_miner.go @@ -63,7 +63,7 @@ func (api *MinerAPI) SetGasPrice(gasPrice hexutil.Big) bool { api.e.gasPrice = (*big.Int)(&gasPrice) api.e.lock.Unlock() - api.e.txPool.SetGasPrice((*big.Int)(&gasPrice)) + api.e.txPool.SetGasTip((*big.Int)(&gasPrice)) return true } diff --git a/eth/backend.go b/eth/backend.go index 4ba8df951..b3338918b 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -206,7 +206,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { if config.TxPool.Journal != "" { config.TxPool.Journal = stack.ResolvePath(config.TxPool.Journal) } - eth.txPool = txpool.NewTxPool(config.TxPool, eth.blockchain.Config(), eth.blockchain) + eth.txPool = txpool.New(config.TxPool, eth.blockchain.Config(), eth.blockchain) // Permit the downloader to use the trie cache allowance during fast sync cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit @@ -399,7 +399,7 @@ func (s *Ethereum) StartMining() error { s.lock.RLock() price := s.gasPrice s.lock.RUnlock() - s.txPool.SetGasPrice(price) + s.txPool.SetGasTip(price) // Configure the local mining address eb, err := s.Etherbase() diff --git a/eth/protocols/eth/handler_test.go b/eth/protocols/eth/handler_test.go index bbb9866bd..21782103c 100644 --- a/eth/protocols/eth/handler_test.go +++ b/eth/protocols/eth/handler_test.go @@ -119,7 +119,7 @@ func newTestBackendWithGenerator(blocks int, shanghai bool, generator func(int, return &testBackend{ db: db, chain: chain, - txpool: txpool.NewTxPool(txconfig, params.TestChainConfig, chain), + txpool: txpool.New(txconfig, params.TestChainConfig, chain), } } diff --git a/les/handler_test.go b/les/handler_test.go index 6d0b17144..a3854dff9 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -611,6 +611,8 @@ func testTransactionStatus(t *testing.T, protocol int) { var reqID uint64 test := func(tx *types.Transaction, send bool, expStatus light.TxStatus) { + t.Helper() + reqID++ if send { sendRequest(rawPeer.app, SendTxV2Msg, reqID, types.Transactions{tx}) @@ -618,14 +620,14 @@ func testTransactionStatus(t *testing.T, protocol int) { sendRequest(rawPeer.app, GetTxStatusMsg, reqID, []common.Hash{tx.Hash()}) } if err := expectResponse(rawPeer.app, TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil { - t.Errorf("transaction status mismatch") + t.Errorf("transaction status mismatch: %v", err) } } signer := types.HomesteadSigner{} // test error status by sending an underpriced transaction tx0, _ := types.SignTx(types.NewTransaction(0, userAddr1, big.NewInt(10000), params.TxGas, nil, nil), signer, bankKey) - test(tx0, true, light.TxStatus{Status: txpool.TxStatusUnknown, Error: txpool.ErrUnderpriced.Error()}) + test(tx0, true, light.TxStatus{Status: txpool.TxStatusUnknown, Error: "transaction underpriced: tip needed 1, tip permitted 0"}) tx1, _ := types.SignTx(types.NewTransaction(0, userAddr1, big.NewInt(10000), params.TxGas, big.NewInt(100000000000), nil), signer, bankKey) test(tx1, false, light.TxStatus{Status: txpool.TxStatusUnknown}) // query before sending, should be unknown diff --git a/les/test_helper.go b/les/test_helper.go index ead97ddd1..44a454eae 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -234,7 +234,7 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da txpoolConfig := txpool.DefaultConfig txpoolConfig.Journal = "" - txpool := txpool.NewTxPool(txpoolConfig, gspec.Config, simulation.Blockchain()) + txpool := txpool.New(txpoolConfig, gspec.Config, simulation.Blockchain()) server := &LesServer{ lesCommons: lesCommons{ diff --git a/miner/miner_test.go b/miner/miner_test.go index 5ea6e1830..67d038d68 100644 --- a/miner/miner_test.go +++ b/miner/miner_test.go @@ -267,7 +267,7 @@ func createMiner(t *testing.T) (*Miner, *event.TypeMux, func(skipMiner bool)) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(chainDB), nil) blockchain := &testBlockChain{statedb, 10000000, new(event.Feed)} - pool := txpool.NewTxPool(testTxPoolConfig, chainConfig, blockchain) + pool := txpool.New(testTxPoolConfig, chainConfig, blockchain) backend := NewMockBackend(bc, pool) // Create event Mux mux := new(event.TypeMux) diff --git a/miner/worker_test.go b/miner/worker_test.go index fb15d365a..d58382e1e 100644 --- a/miner/worker_test.go +++ b/miner/worker_test.go @@ -135,7 +135,7 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine return &testWorkerBackend{ db: db, chain: chain, - txPool: txpool.NewTxPool(testTxPoolConfig, chainConfig, chain), + txPool: txpool.New(testTxPoolConfig, chainConfig, chain), genesis: gspec, } } diff --git a/tests/fuzzers/les/les-fuzzer.go b/tests/fuzzers/les/les-fuzzer.go index 926de0458..c203c87f8 100644 --- a/tests/fuzzers/les/les-fuzzer.go +++ b/tests/fuzzers/les/les-fuzzer.go @@ -138,7 +138,7 @@ func newFuzzer(input []byte) *fuzzer { chtKeys: chtKeys, bloomKeys: bloomKeys, nonce: uint64(len(txHashes)), - pool: txpool.NewTxPool(txpool.DefaultConfig, params.TestChainConfig, chain), + pool: txpool.New(txpool.DefaultConfig, params.TestChainConfig, chain), input: bytes.NewReader(input), } }