diff --git a/core/txpool/locals/errors.go b/core/txpool/locals/errors.go index de4d199490a..2149749e675 100644 --- a/core/txpool/locals/errors.go +++ b/core/txpool/locals/errors.go @@ -26,6 +26,12 @@ import ( // IsTemporaryReject determines whether the given error indicates a temporary // reason to reject a transaction from being included in the txpool. The result // may change if the txpool's state changes later. +// +// ErrUnderMinGasPrice must stay out of this set: the floor only rises as the +// chain advances, so retrying the same transaction cannot help while it stays +// where it is. It must not be used to drop a tracked transaction either -- a +// rollback or a reorg past the fork lowers the floor again, and the local +// tracker is the only thing that can bring those transactions back. func IsTemporaryReject(err error) bool { switch { case errors.Is(err, legacypool.ErrOutOfOrderTxFromDelegated): diff --git a/core/txpool/locals/errors_test.go b/core/txpool/locals/errors_test.go new file mode 100644 index 00000000000..7517f9e83fa --- /dev/null +++ b/core/txpool/locals/errors_test.go @@ -0,0 +1,41 @@ +// Copyright 2025 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 locals + +import ( + "errors" + "testing" + + "github.com/XinFinOrg/XDPoSChain/core/txpool" +) + +// TestIsTemporaryRejectExcludesMinGasPrice pins that a transaction priced below +// the gas schedule floor is not a temporary reject: the floor only rises as the +// chain advances, so retrying it cannot help and AddLocal must not track it. +func TestIsTemporaryRejectExcludesMinGasPrice(t *testing.T) { + if IsTemporaryReject(txpool.ErrUnderMinGasPrice) { + t.Error("ErrUnderMinGasPrice must not be a temporary reject") + } + // The pool's own price limit is a node local setting that can change, so it + // stays retryable. The contrast is what the assertion above is about. + if !IsTemporaryReject(txpool.ErrUnderpriced) { + t.Error("ErrUnderpriced must remain a temporary reject") + } + if IsTemporaryReject(errors.New("unrelated failure")) { + t.Error("an unrelated error must not be a temporary reject") + } +} diff --git a/core/txpool/locals/tx_tracker.go b/core/txpool/locals/tx_tracker.go index a3fee1a35e7..285291b77a2 100644 --- a/core/txpool/locals/tx_tracker.go +++ b/core/txpool/locals/tx_tracker.go @@ -35,6 +35,16 @@ import ( var ( recheckInterval = time.Minute localGauge = metrics.GetOrRegisterGauge("txpool/local", nil) + + // Tracked transactions a gas schedule fork priced out of the pool. They stay + // tracked so a rollback past the fork can pick them up again, but recheck + // holds them back from resubmission, so they are broken out here. + // + // Only transactions missing from the pool are counted: a tracked transaction + // the pool still has is counted as ok instead. local minus belowfloor is + // therefore the tracked transactions the pool holds plus the ones recheck + // resubmits this round, not just the latter. + belowFloorGauge = metrics.GetOrRegisterGauge("txpool/local/belowfloor", nil) ) // TxTracker is a struct used to track priority transactions; it will check from @@ -101,11 +111,78 @@ func (tracker *TxTracker) TrackAll(txs []*types.Transaction) { if err != nil { // Ignore this tx continue } - tracker.all[tx.Hash()] = tx - if tracker.byAddr[addr] == nil { - tracker.byAddr[addr] = legacypool.NewSortedMap() + list := tracker.byAddr[addr] + if list == nil { + list = legacypool.NewSortedMap() + tracker.byAddr[addr] = list } - tracker.byAddr[addr].Put(tx) + // A transaction tracked for a nonce that is already taken supersedes the + // one it replaces. SortedMap.Put overwrites silently, so the replaced + // transaction has to be dropped here: it is never returned by Forward + // again, and leaving it in `all` would keep journaling it forever, + // where it can win the nonce on the next load and resurrect a + // transaction the user already replaced. + // + // Which of the two supersedes the other is decided the way the pool + // decides it: + // + // 1. The pool still holds one of them. It occupies a nonce with at + // most one transaction, so whichever it holds won the nonce. This + // is checked first because the order transactions reach TrackAll + // is not the order they were accepted in: AddLocal tracks a local + // transaction only after Add has released the subpool lock, so two + // concurrent submissions can be accepted in one order and reach + // here in the other. + // 2. The pool holds neither -- two submissions it has since + // discarded, or a journal an older version wrote. Fall back to the + // substitution rules legacypool applies in list.Add: a special + // transaction always claims its nonce, a regular one must not + // evict a pending special one, and otherwise a replacement has to + // beat the transaction it replaces on both fee cap and tip. The + // price bump is deliberately not repeated here: it is a pool + // policy, and a transaction that beats the old one but misses the + // bump behaves exactly as it did before, so no case gets worse. + // + // Dropping the loser here also converges a journal written by an older + // version, which can hold both a transaction and its replacement: load + // feeds the file straight into TrackAll, so the replacement wins the + // nonce instead of whichever entry the older rotation happened to sort + // last, and the next rotation rewrites the journal from the converged + // set. + if replaced := list.Get(tx.Nonce()); replaced != nil { + switch { + case tracker.pool.Has(tx.Hash()): + // The pool accepted this one, so it holds the nonce. A tracked + // transaction that is dearer does not change that: it is one an + // older version journalled after the pool rejected it as + // retryable, and resubmitting it would evict this one. + case tracker.pool.Has(replaced.Hash()): + // The pool still holds the transaction this one would replace: + // it lost the race for the nonce there. + log.Debug("Ignoring tracked local transaction the pool superseded", "nonce", tx.Nonce(), + "kept", replaced.Hash(), "ignored", tx.Hash()) + continue + case tx.IsSpecialTransaction(): + // A special transaction always claims its nonce. + case replaced.IsSpecialTransaction(): + // A regular transaction must not evict a pending special one, + // however much dearer it is. + log.Debug("Ignoring tracked local transaction that would evict a special one", "nonce", tx.Nonce(), + "kept", replaced.Hash(), "ignored", tx.Hash()) + continue + case replaced.GasFeeCapCmp(tx) >= 0 || replaced.GasTipCapCmp(tx) >= 0: + // Not a replacement the pool would accept: keep the one that is + // already tracked. + log.Debug("Ignoring tracked local transaction the tracked one outbids", "nonce", tx.Nonce(), + "kept", replaced.Hash(), "ignored", tx.Hash()) + continue + } + delete(tracker.all, replaced.Hash()) + log.Debug("Replaced tracked local transaction", "nonce", tx.Nonce(), + "replaced", replaced.Hash(), "replacement", tx.Hash()) + } + list.Put(tx) + tracker.all[tx.Hash()] = tx if tracker.journal != nil { _ = tracker.journal.insert(tx) @@ -120,10 +197,15 @@ func (tracker *TxTracker) recheck(journalCheck bool) []*types.Transaction { defer tracker.mu.Unlock() var ( - numStales = 0 - numOk = 0 - resubmits []*types.Transaction + numStales = 0 + numOk = 0 + numBelowFloor = 0 + resubmits []*types.Transaction ) + // Resolved once: it can only change on a chain head update, and a change + // mid-recheck would make the counters below inconsistent. + floor := tracker.pool.MinGasPrice() + for sender, txs := range tracker.byAddr { // Wipe the stales stales := txs.Forward(tracker.pool.Nonce(sender)) @@ -138,6 +220,17 @@ func (tracker *TxTracker) recheck(journalCheck bool) []*types.Transaction { numOk++ continue } + // A gas schedule fork can raise the floor above transactions that + // were admitted under the previous tier. Re-submitting them only + // yields ErrUnderMinGasPrice, so hold them back until the floor + // drops again: a reorg or a set-head rollback past the fork + // reinstates them, and the pool does not bring back what it swept. + // They stay tracked, so recheck picks them up again on its own. + // Special transactions are exempt, exactly as during admission. + if !tx.IsSpecialTransaction() && tx.GasPriceIntCmp(floor) < 0 { + numBelowFloor++ + continue + } resubmits = append(resubmits, tx) } } @@ -165,7 +258,9 @@ func (tracker *TxTracker) recheck(journalCheck bool) []*types.Transaction { } } localGauge.Update(int64(len(tracker.all))) - log.Debug("Tx tracker status", "need-resubmit", len(resubmits), "stale", numStales, "ok", numOk) + belowFloorGauge.Update(int64(numBelowFloor)) + log.Debug("Tx tracker status", "need-resubmit", len(resubmits), "stale", numStales, + "ok", numOk, "below-floor", numBelowFloor) return resubmits } diff --git a/core/txpool/locals/tx_tracker_test.go b/core/txpool/locals/tx_tracker_test.go index 41c058592c4..96d081d37d5 100644 --- a/core/txpool/locals/tx_tracker_test.go +++ b/core/txpool/locals/tx_tracker_test.go @@ -37,6 +37,7 @@ import ( "github.com/XinFinOrg/XDPoSChain/crypto" "github.com/XinFinOrg/XDPoSChain/ethdb" "github.com/XinFinOrg/XDPoSChain/params" + "github.com/XinFinOrg/XDPoSChain/rlp" ) var ( @@ -50,7 +51,6 @@ var ( }, BaseFee: big.NewInt(params.InitialBaseFee), } - signer = types.LatestSigner(gspec.Config) ) type testEnv struct { @@ -58,9 +58,23 @@ type testEnv struct { pool *txpool.TxPool tracker *TxTracker genDb ethdb.Database + signer types.Signer } func newTestEnv(t *testing.T, n int, gasTip uint64, journal string) *testEnv { + return newTestEnvWithConfig(t, n, gasTip, journal, params.TestChainConfig) +} + +// newTestEnvWithConfig builds an environment around cfg. It builds its own +// genesis and signer instead of touching the package level ones, which the +// tests in this file share. +func newTestEnvWithConfig(t *testing.T, n int, gasTip uint64, journal string, cfg *params.ChainConfig) *testEnv { + gspec := &core.Genesis{ + Config: cfg, + Alloc: types.GenesisAlloc{address: {Balance: new(big.Int).Set(funds)}}, + BaseFee: big.NewInt(params.InitialBaseFee), + } + signer := types.LatestSigner(cfg) genDb, blocks, _ := core.GenerateChainWithGenesis(gspec, ethash.NewFaker(), n, func(i int, gen *core.BlockGen) { gasPrice := big.NewInt(params.InitialBaseFee) if baseFee := gen.BaseFee(); baseFee != nil { @@ -95,9 +109,25 @@ func newTestEnv(t *testing.T, n int, gasTip uint64, journal string) *testEnv { pool: pool, tracker: New(journal, time.Minute, gspec.Config, pool), genDb: genDb, + signer: signer, } } +// nonce returns the next nonce the test account can spend at the current head. +func (env *testEnv) nonce() uint64 { + head := env.chain.CurrentHeader() + state, _ := env.chain.StateAt(head.Root) + return state.GetNonce(address) +} + +// gasPrice returns a gas price the gas schedule of the current head admits. +func (env *testEnv) gasPrice() *big.Int { + if baseFee := env.chain.CurrentHeader().BaseFee; baseFee != nil { + return new(big.Int).Set(baseFee) + } + return big.NewInt(params.InitialBaseFee) +} + func (env *testEnv) close() { if err := env.pool.Close(); err != nil { panic(fmt.Sprintf("failed to close tx pool: %v", err)) @@ -116,7 +146,7 @@ func (env *testEnv) makeTxs(n int) []*types.Transaction { var txs []*types.Transaction for i := 0; i < n; i++ { - tx, _ := types.SignTx(types.NewTransaction(nonce+uint64(i), common.Address{0x00}, big.NewInt(1000), params.TxGas, gasPrice, nil), signer, key) + tx, _ := types.SignTx(types.NewTransaction(nonce+uint64(i), common.Address{0x00}, big.NewInt(1000), params.TxGas, gasPrice, nil), env.signer, key) txs = append(txs, tx) } return txs @@ -225,3 +255,508 @@ func TestStartContinuesOnCorruptedJournal(t *testing.T) { t.Fatal("Journal writer should be initialized even if journal load fails") } } + +// bumpedPrice returns the lowest gas price that clears the pool's price bump +// over price, the price a replacement has to carry to take a nonce from the +// transaction it replaces. +func bumpedPrice(price *big.Int) *big.Int { + bumped := new(big.Int).Div( + new(big.Int).Mul(price, big.NewInt(int64(100+legacypool.DefaultConfig.PriceBump))), + big.NewInt(100)) + return bumped.Add(bumped, big.NewInt(1)) +} + +// replacementPair returns two transactions sharing a nonce: the one tracked +// first, and the one that replaces it. The replacement clears the pool's price +// bump, so the pair stands for the substitution the pool accepts; two equally +// priced transactions are not a substitution, the pool rejects them. +func replacementPair(env *testEnv) (replaced, replacement *types.Transaction) { + nonce := env.nonce() + mk := func(to common.Address, gasPrice *big.Int) *types.Transaction { + tx, _ := types.SignTx(types.NewTransaction(nonce, to, big.NewInt(1000), params.TxGas, gasPrice, nil), env.signer, key) + return tx + } + return mk(common.Address{0x00}, env.gasPrice()), mk(common.Address{0x01}, bumpedPrice(env.gasPrice())) +} + +// writeReplacementJournal writes a journal holding two transactions that share +// a nonce, the way an older version could have left it behind: a transaction +// together with the one that replaced it, in the order selected by reverse. +func writeReplacementJournal(t *testing.T, reverse bool) (path string, replaced, replacement *types.Transaction) { + t.Helper() + + env := newTestEnv(t, 10, 0, "") + defer env.close() + + replaced, replacement = replacementPair(env) + order := []*types.Transaction{replaced, replacement} + if reverse { + order = []*types.Transaction{replacement, replaced} + } + var journal []byte + for _, tx := range order { + blob, err := rlp.EncodeToBytes(tx) + if err != nil { + t.Fatalf("Failed to encode transaction: %v", err) + } + journal = append(journal, blob...) + } + path = filepath.Join(t.TempDir(), fmt.Sprintf("%d", rand.Int63())) + if err := os.WriteFile(path, journal, 0o644); err != nil { + t.Fatalf("Failed to write journal: %v", err) + } + return path, replaced, replacement +} + +// TestTrackAllKeepsTransactionHeldByPool pins which transaction wins a nonce +// when the pool already holds one of them: the pool decides, because the order +// transactions reach TrackAll is not the order they were accepted in. Add +// tracks a local transaction only after SubPool.Add has released its lock, so +// a replacement can arrive here before the transaction it replaced. +func TestTrackAllKeepsTransactionHeldByPool(t *testing.T) { + for _, tc := range []struct { + name string + reversed bool + }{ + {name: "pooled first"}, + {name: "replacement first", reversed: true}, + } { + t.Run(tc.name, func(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + pooled, replacement := replacementPair(env) + if err := env.pool.Add([]*types.Transaction{pooled}, true)[0]; err != nil { + t.Fatalf("failed to add the transaction the pool must hold: %v", err) + } + pair := []*types.Transaction{pooled, replacement} + if tc.reversed { + pair = []*types.Transaction{replacement, pooled} + } + env.tracker.TrackAll(pair) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("tracked set must hold a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[pooled.Hash()]; !ok { + t.Fatalf("the transaction the pool holds must stay tracked: %v", pooled.Hash()) + } + if kept := env.tracker.byAddr[address].Get(pooled.Nonce()); kept == nil || kept.Hash() != pooled.Hash() { + t.Fatalf("nonce %d must hold the pooled transaction, got %v", pooled.Nonce(), kept) + } + }) + } +} + +// TestConcurrentReplacementKeepsPoolTransaction pins the tracker's view of two +// concurrent submissions for the same nonce: the original is added first and +// reaches the tracker last, the window AddLocal leaves between SubPool.Add +// releasing its lock and Track acquiring this one. The transaction the pool +// holds must win the nonce even though it is tracked last, or recheck would +// keep resubmitting the superseded one and the live one would lose the local +// resubmit protection entirely. +func TestConcurrentReplacementKeepsPoolTransaction(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + nonce := env.nonce() + price := env.gasPrice() + mk := func(to common.Address, gasPrice *big.Int) *types.Transaction { + tx, _ := types.SignTx(types.NewTransaction(nonce, to, big.NewInt(1000), params.TxGas, gasPrice, nil), env.signer, key) + return tx + } + var ( + original = mk(common.Address{0x00}, price) + replacement = mk(common.Address{0x01}, bumpedPrice(price)) + added = make(chan struct{}) + tracked = make(chan struct{}) + finished = make(chan error, 2) + ) + go func() { + err := env.pool.Add([]*types.Transaction{original}, true)[0] + close(added) + <-tracked + env.tracker.Track(original) + finished <- err + }() + go func() { + <-added + err := env.pool.Add([]*types.Transaction{replacement}, true)[0] + env.tracker.Track(replacement) + close(tracked) + finished <- err + }() + for i := 0; i < 2; i++ { + if err := <-finished; err != nil { + t.Fatalf("failed to submit the transaction: %v", err) + } + } + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("tracked set must hold a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[replacement.Hash()]; !ok { + t.Fatalf("the replacement the pool holds must stay tracked: %v", replacement.Hash()) + } +} + +func TestTrackAllDropsReplacedTransaction(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + replaced, replacement := replacementPair(env) + env.tracker.TrackAll([]*types.Transaction{replaced, replacement}) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("replaced transaction must be dropped: tracking %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[replaced.Hash()]; ok { + t.Fatalf("replaced transaction still tracked: %v", replaced.Hash()) + } + kept := env.tracker.byAddr[address].Get(replacement.Nonce()) + if kept == nil || kept.Hash() != replacement.Hash() { + t.Fatalf("nonce %d must hold the replacement, got %v", replacement.Nonce(), kept) + } +} + +func TestRecheckDoesNotResubmitReplacedTransaction(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + replaced, replacement := replacementPair(env) + env.tracker.TrackAll([]*types.Transaction{replaced, replacement}) + + resubmits := env.tracker.recheck(false) + if len(resubmits) != 1 || resubmits[0].Hash() != replacement.Hash() { + t.Fatalf("unexpected transactions to resubmit: %v", resubmits) + } +} + +func TestJournalRotationDropsReplacedTransaction(t *testing.T) { + journalPath := filepath.Join(t.TempDir(), fmt.Sprintf("%d", rand.Int63())) + env := newTestEnv(t, 10, 0, journalPath) + defer env.close() + + if err := env.tracker.Start(); err != nil { + t.Fatalf("Failed to start tracker: %v", err) + } + defer env.tracker.Stop() + + replaced, replacement := replacementPair(env) + env.tracker.TrackAll([]*types.Transaction{replaced, replacement}) + + // Rotate the journal from the tracked set: the replaced transaction has + // been dropped from it and must not come back. + env.tracker.recheck(true) + + reloaded := New(journalPath, time.Minute, params.TestChainConfig, env.pool) + if err := reloaded.journal.load(func(transactions []*types.Transaction) []error { + reloaded.TrackAll(transactions) + return nil + }); err != nil { + t.Fatalf("Failed to load journal: %v", err) + } + + reloaded.mu.Lock() + defer reloaded.mu.Unlock() + + if len(reloaded.all) != 1 { + t.Fatalf("rotated journal must hold a single transaction, got %d", len(reloaded.all)) + } + if _, ok := reloaded.all[replacement.Hash()]; !ok { + t.Fatalf("rotated journal must hold the replacement: %v", replacement.Hash()) + } +} + +func TestJournalLoadDropsReplacedTransaction(t *testing.T) { + journalPath, _, replacement := writeReplacementJournal(t, false) + env := newTestEnv(t, 10, 0, journalPath) + defer env.close() + + if err := env.tracker.Start(); err != nil { + t.Fatalf("Failed to start tracker: %v", err) + } + defer env.tracker.Stop() + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("loading must leave a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[replacement.Hash()]; !ok { + t.Fatalf("the replacement must survive the load: %v", replacement.Hash()) + } +} + +// TestJournalLoadKeepsReplacementPerNonce pins the load path for the reverse +// file order. A journal entry carries no timestamp or sequence number, so an +// older rotation could have written the transaction the user replaced after +// its replacement. Load feeds the file straight into TrackAll, which decides +// the nonce the way the pool would, so the replacement survives whatever order +// the entries come back in. +func TestJournalLoadKeepsReplacementPerNonce(t *testing.T) { + journalPath, _, replacement := writeReplacementJournal(t, true) + env := newTestEnv(t, 10, 0, journalPath) + defer env.close() + + if err := env.tracker.Start(); err != nil { + t.Fatalf("Failed to start tracker: %v", err) + } + defer env.tracker.Stop() + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("loading must leave a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[replacement.Hash()]; !ok { + t.Fatalf("the replacement must survive the load whatever the file order is: %v", replacement.Hash()) + } +} + +// TestTrackAllKeepsTransactionAcceptedByPool pins the case a dearer tracked +// transaction must not win: the pool accepted this one, so it holds the nonce. +// The dearer one is what an older version journalled after the pool rejected it +// as retryable, and resubmitting it would evict a transaction the user just +// submitted successfully. +func TestTrackAllKeepsTransactionAcceptedByPool(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + nonce := env.nonce() + mk := func(to common.Address, gasPrice *big.Int) *types.Transaction { + tx, _ := types.SignTx(types.NewTransaction(nonce, to, big.NewInt(1000), params.TxGas, gasPrice, nil), env.signer, key) + return tx + } + tracked := mk(common.Address{0x00}, new(big.Int).Mul(env.gasPrice(), big.NewInt(4))) + accepted := mk(common.Address{0x01}, env.gasPrice()) + + env.tracker.Track(tracked) + if err := env.pool.Add([]*types.Transaction{accepted}, true)[0]; err != nil { + t.Fatalf("failed to add the transaction the pool must accept: %v", err) + } + env.tracker.Track(accepted) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("tracked set must hold a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[accepted.Hash()]; !ok { + t.Fatalf("the transaction the pool accepted must stay tracked: %v", accepted.Hash()) + } +} + +// TestTrackAllKeepsReplacementWhenPoolHoldsNeither pins the fallback for two +// transactions the pool has discarded, seen in the order a replacement can +// reach the tracker in: the replacement first, the transaction it replaces +// last. Keeping the replacement is what stops recheck from resubmitting a +// transaction the user already superseded. +// +// This exercises the fallback rule directly rather than the race that reaches +// it: the concurrent interleaving is pinned by +// TestConcurrentReplacementKeepsPoolTransaction, and the pool-discards-both +// state needs no fork to construct. +func TestTrackAllKeepsReplacementWhenPoolHoldsNeither(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + replaced, replacement := replacementPair(env) + // Neither is submitted to the pool: what is left once the pool discards + // both, and the only state the fallback can be decided on. + env.tracker.Track(replacement) + env.tracker.Track(replaced) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 { + t.Fatalf("tracked set must hold a single transaction, got %d", len(env.tracker.all)) + } + if _, ok := env.tracker.all[replacement.Hash()]; !ok { + t.Fatalf("the replacement must stay tracked: %v", replacement.Hash()) + } + if kept := env.tracker.byAddr[address].Get(replacement.Nonce()); kept == nil || kept.Hash() != replacement.Hash() { + t.Fatalf("nonce %d must hold the replacement, got %v", replacement.Nonce(), kept) + } +} + +// TestTrackAllKeepsTrackedSpecialTransaction pins the exemption that mirrors +// the pool: a regular transaction must not evict a pending special one, so a +// dearer regular transaction must not take the nonce of a tracked special one. +func TestTrackAllKeepsTrackedSpecialTransaction(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + nonce := env.nonce() + special, _ := types.SignTx(types.NewTransaction(nonce, common.BlockSignersBinary, big.NewInt(0), params.TxGas, common.Big0, nil), env.signer, key) + regular, _ := types.SignTx(types.NewTransaction(nonce, common.Address{0x00}, big.NewInt(1000), params.TxGas, bumpedPrice(env.gasPrice()), nil), env.signer, key) + + env.tracker.Track(special) + env.tracker.Track(regular) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 || env.tracker.all[special.Hash()] == nil { + t.Fatalf("the special transaction must stay tracked: %v", env.tracker.all) + } +} + +// TestTrackAllLetsSpecialTransactionReplace pins the other half: a special +// transaction always claims its nonce, so it takes it from a dearer tracked +// transaction, exactly as it does in the pool. +func TestTrackAllLetsSpecialTransactionReplace(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + nonce := env.nonce() + regular, _ := types.SignTx(types.NewTransaction(nonce, common.Address{0x00}, big.NewInt(1000), params.TxGas, bumpedPrice(env.gasPrice()), nil), env.signer, key) + special, _ := types.SignTx(types.NewTransaction(nonce, common.BlockSignersBinary, big.NewInt(0), params.TxGas, common.Big0, nil), env.signer, key) + + env.tracker.Track(regular) + env.tracker.Track(special) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 || env.tracker.all[special.Hash()] == nil { + t.Fatalf("the special transaction must replace the tracked one: %v", env.tracker.all) + } +} + +// TestTrackAllKeepsFirstTransactionAtEqualPrice pins the tie: equally priced +// transactions are not a substitution, the pool rejects the later one, so the +// transaction already tracked keeps the nonce. +func TestTrackAllKeepsFirstTransactionAtEqualPrice(t *testing.T) { + env := newTestEnv(t, 10, 0, "") + defer env.close() + + nonce := env.nonce() + mk := func(to common.Address) *types.Transaction { + tx, _ := types.SignTx(types.NewTransaction(nonce, to, big.NewInt(1000), params.TxGas, env.gasPrice(), nil), env.signer, key) + return tx + } + first, second := mk(common.Address{0x00}), mk(common.Address{0x01}) + + env.tracker.TrackAll([]*types.Transaction{first, second}) + + env.tracker.mu.Lock() + defer env.tracker.mu.Unlock() + + if len(env.tracker.all) != 1 || env.tracker.all[first.Hash()] == nil { + t.Fatalf("the first of two equally priced transactions must stay tracked: %v", env.tracker.all) + } +} + +func TestRecheckHoldsBackBelowFloorTransactions(t *testing.T) { + // Push the gas tier fork past the test chain so the floor resolves to the + // baseline tier: params.TestChainConfig schedules Gas50x at block 0, which + // would put the floor at 50x from genesis on. The field cannot be cleared + // instead, CheckConfigForkOrder requires it. Gas2500xBlock stays nil, which + // CheckConfigForkOrder accepts; the assignment states it explicitly. + cfg := *params.TestChainConfig + cfg.Gas50xBlock = big.NewInt(1000) + cfg.Gas2500xBlock = nil + + env := newTestEnvWithConfig(t, 1, 0, "", &cfg) + defer env.close() + + floor := big.NewInt(common.DefaultMinGasPrice) + nonce := env.nonce() + mk := func(n uint64, gasPrice *big.Int) *types.Transaction { + tx, _ := types.SignTx(types.NewTransaction(n, common.Address{0x00}, big.NewInt(1000), params.TxGas, gasPrice, nil), env.signer, key) + return tx + } + // Priced at the floor and one wei below it: the comparison is + // GasPriceIntCmp(floor) < 0, so only the latter is held back. + atFloor := mk(nonce, floor) + belowFloor := mk(nonce+1, new(big.Int).Sub(floor, big.NewInt(1))) + + env.tracker.TrackAll([]*types.Transaction{atFloor, belowFloor}) + + resubmits := env.tracker.recheck(false) + if len(resubmits) != 1 || resubmits[0].Hash() != atFloor.Hash() { + t.Fatalf("unexpected transactions to resubmit: %v", resubmits) + } + // Held back, not dropped: it stays tracked so a lower floor picks it up. + if len(env.tracker.all) != 2 { + t.Fatalf("below-floor transaction must stay tracked, got %d", len(env.tracker.all)) + } +} + +func TestRecheckResumesAfterFloorDrops(t *testing.T) { + cfg := *params.TestChainConfig + cfg.Gas50xBlock = big.NewInt(5) + + env := newTestEnvWithConfig(t, 10, 0, "", &cfg) + defer env.close() + + // head=10, so the floor resolves at block 11: the 50x tier is active and a + // baseline-priced transaction is held back. + tx, _ := types.SignTx(types.NewTransaction( + env.nonce(), common.Address{0x00}, big.NewInt(1000), params.TxGas, + big.NewInt(common.DefaultMinGasPrice), nil), env.signer, key) + + env.tracker.Track(tx) + if resubmits := env.tracker.recheck(false); len(resubmits) != 0 { + t.Fatalf("transaction below the floor must not be resubmitted: %v", resubmits) + } + if len(env.tracker.all) != 1 { + t.Fatalf("transaction below the floor must stay tracked, got %d", len(env.tracker.all)) + } + + // Roll the head back before the fork: the floor drops to the baseline tier, + // which is exactly the transaction's price, so it is let through. recheck + // only filters on Forward(pool.Nonce(sender)), which leaves nonce 10 alone + // now that the state nonce has fallen back to 3. + if err := env.chain.SetHead(3); err != nil { + t.Fatalf("failed to roll back the chain: %v", err) + } + if err := env.pool.Sync(); err != nil { + t.Fatalf("failed to sync the txpool: %v", err) + } + + resubmits := env.tracker.recheck(false) + if len(resubmits) != 1 || resubmits[0].Hash() != tx.Hash() { + t.Fatalf("transaction must be resubmitted once the floor drops: %v", resubmits) + } +} + +// TestRecheckResubmitsSpecialTransactionBelowFloor pins the exemption that keeps +// recheck aligned with admission: special transactions are not priced against +// the floor there either, so a gas tier fork must not strand the ones submitted +// locally, which is what the tracker exists to guard. +func TestRecheckResubmitsSpecialTransactionBelowFloor(t *testing.T) { + cfg := *params.TestChainConfig + cfg.Gas50xBlock = big.NewInt(5) + + env := newTestEnvWithConfig(t, 10, 0, "", &cfg) + defer env.close() + + // head=10, so the floor resolves at block 11 on the 50x tier. The + // transaction is priced at the baseline tier, which a plain transaction + // would be held back for. + tx, _ := types.SignTx(types.NewTransaction( + env.nonce(), common.BlockSignersBinary, big.NewInt(0), params.TxGas, + big.NewInt(common.DefaultMinGasPrice), nil), env.signer, key) + + env.tracker.Track(tx) + + resubmits := env.tracker.recheck(false) + if len(resubmits) != 1 || resubmits[0].Hash() != tx.Hash() { + t.Fatalf("special transaction below the floor must still be resubmitted: %v", resubmits) + } +} diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 4f99cca47b1..30ae599fd14 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -437,6 +437,24 @@ func (p *TxPool) Nonce(addr common.Address) uint64 { return p.state.GetNonce(addr) } +// MinGasPrice returns the gas price floor the pool enforces on the transactions +// it admits. A pooled transaction can only be included from the next block +// onwards, so the floor is resolved at the height of the block pending on top +// of the current head, the height admission validation prices a transaction at. +// +// The head here is the chain's, which advances on block insertion, while +// admission resolves it against the subpool's, which follows on head events. +// The two drift for as long as a head event is in flight, so a floor read in +// that window can differ from the one admission applies. It self-corrects on +// the next recheck, whose period is orders of magnitude longer than the window. +func (p *TxPool) MinGasPrice() *big.Int { + var number *big.Int + if head := p.chain.CurrentBlock(); head != nil { + number = head.Number + } + return params.GetMinGasPrice(pendingBlockNumber(number), p.chain.Config()) +} + // Stats retrieves the current pool stats, namely the number of pending and the // number of queued (non-executable) transactions. func (p *TxPool) Stats() (int, int) { diff --git a/core/txpool/txpool_local_test.go b/core/txpool/txpool_local_test.go index 2882451160c..cf7ad79e4c6 100644 --- a/core/txpool/txpool_local_test.go +++ b/core/txpool/txpool_local_test.go @@ -337,3 +337,36 @@ func TestAddLocalTemporaryRejectWithoutTrackerReturnsError(t *testing.T) { t.Fatalf("unexpected call order: have %v", events) } } + +// TestAddLocalDoesNotTrackMinGasPriceRejectedTx checks that a local transaction +// the pool rejects for being priced below the gas schedule floor is not tracked: +// the floor only rises as the chain advances, so re-submitting the very same +// transaction could never succeed. The error is still surfaced to the caller. +func TestAddLocalDoesNotTrackMinGasPriceRejectedTx(t *testing.T) { + events := []string{} + tracker := &testLocalTracker{events: &events} + subpool := &testSubPool{ + events: &events, + addErrs: []error{ErrUnderMinGasPrice}, + } + + pool, err := New(0, testChain{}, []SubPool{subpool}) + if err != nil { + t.Fatalf("failed to create txpool: %v", err) + } + defer pool.Close() + + pool.SetLocalTracker(tracker) + + tx := types.NewTransaction(0, common.Address{0x1}, big.NewInt(1), 21000, big.NewInt(1), nil) + err = pool.AddLocal(tx, true) + if !errors.Is(err, ErrUnderMinGasPrice) { + t.Fatalf("unexpected error: have %v, want %v", err, ErrUnderMinGasPrice) + } + if len(tracker.tracked) != 0 { + t.Fatalf("tracker must not receive a tx rejected by the gas price floor: %v", tracker.tracked) + } + if !reflect.DeepEqual(events, []string{"add"}) { + t.Fatalf("unexpected call order: have %v", events) + } +} diff --git a/core/txpool/txpool_test.go b/core/txpool/txpool_test.go new file mode 100644 index 00000000000..6800abcb40e --- /dev/null +++ b/core/txpool/txpool_test.go @@ -0,0 +1,158 @@ +// Copyright 2025 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_test + +import ( + "math/big" + "testing" + + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/consensus/ethash" + "github.com/XinFinOrg/XDPoSChain/core" + "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/state" + "github.com/XinFinOrg/XDPoSChain/core/txpool" + "github.com/XinFinOrg/XDPoSChain/core/txpool/legacypool" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/core/vm" + "github.com/XinFinOrg/XDPoSChain/event" + "github.com/XinFinOrg/XDPoSChain/params" +) + +// TestMinGasPriceResolvesAtPendingBlock checks that MinGasPrice resolves the gas +// schedule at the block pending on top of the head, the height admission +// validation prices a transaction at. +func TestMinGasPriceResolvesAtPendingBlock(t *testing.T) { + cfg := *params.TestChainConfig + cfg.Gas50xBlock = big.NewInt(100) + cfg.Gas2500xBlock = big.NewInt(200) + + chain, pool := newMinGasPriceTestEnv(t, &cfg, 200) + defer chain.Stop() + defer pool.Close() + + var ( + baseline = big.NewInt(common.DefaultMinGasPrice) + gas50x = new(big.Int).Mul(baseline, big.NewInt(50)) + gas2500x = new(big.Int).Mul(baseline, big.NewInt(2500)) + ) + + // The chain is rewound from high to low: SetHead only goes backwards, and + // each step is followed by Sync so the pool has processed the new head. + if err := pool.Sync(); err != nil { + t.Fatalf("failed to sync the txpool: %v", err) + } + if have := pool.MinGasPrice(); have.Cmp(gas2500x) != 0 { + t.Fatalf("head 200: MinGasPrice = %v, want %v", have, gas2500x) + } + for _, tc := range []struct { + head uint64 + want *big.Int + }{ + {head: 199, want: gas2500x}, // pending 200, the fork fires + {head: 198, want: gas50x}, // pending 199 + {head: 99, want: gas50x}, // pending 100, the fork fires + {head: 98, want: baseline}, // pending 99 + } { + if err := chain.SetHead(tc.head); err != nil { + t.Fatalf("failed to rewind to %d: %v", tc.head, err) + } + if err := pool.Sync(); err != nil { + t.Fatalf("failed to sync the txpool at %d: %v", tc.head, err) + } + if have := pool.MinGasPrice(); have.Cmp(tc.want) != 0 { + t.Errorf("head %d: MinGasPrice = %v, want %v", tc.head, have, tc.want) + } + } +} + +// TestMinGasPriceFallsBackWithoutHeadNumber checks the floor a head without a +// block number resolves to. Admission validation passes a nil number to the gas +// schedule in that state too, so the fallback is what keeps the tracker pricing +// transactions against the same floor the pool admits them at. +func TestMinGasPriceFallsBackWithoutHeadNumber(t *testing.T) { + // The tiers are scheduled low enough that a nil number and the number the + // guard would otherwise fall into resolve to different tiers, so the + // assertion below cannot pass by accident. + cfg := *params.TestChainConfig + cfg.Gas50xBlock = big.NewInt(1) + cfg.Gas2500xBlock = big.NewInt(200) + + pool, err := txpool.New(0, numberlessChain{cfg: &cfg}, nil) + if err != nil { + t.Fatalf("failed to create tx pool: %v", err) + } + defer pool.Close() + + // No block number means no tier is scheduled, so the fallback price is the + // baseline tier. Pinned here so a new tier cannot silently change what the + // assertion below is about. + want := params.GetMinGasPrice(nil, &cfg) + if baseline := big.NewInt(common.DefaultMinGasPrice); want.Cmp(baseline) != 0 { + t.Fatalf("baseline tier changed: have %v, want %v", want, baseline) + } + if have := pool.MinGasPrice(); have.Cmp(want) != 0 { + t.Fatalf("MinGasPrice = %v, want %v", have, want) + } +} + +// numberlessChain reports a head that carries no block number, the state a +// chain is in before its first block is known. +type numberlessChain struct{ cfg *params.ChainConfig } + +func (c numberlessChain) Config() *params.ChainConfig { return c.cfg } + +func (c numberlessChain) CurrentBlock() *types.Header { return &types.Header{} } + +func (c numberlessChain) StateAt(common.Hash) (*state.StateDB, error) { + return state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase())) +} + +func (c numberlessChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription { + return event.NewSubscription(func(quit <-chan struct{}) error { + <-quit + return nil + }) +} + +// newMinGasPriceTestEnv builds a chain of n empty blocks over cfg and a txpool +// layered on top of it. The blocks carry no transactions: only the head height +// matters for the gas schedule. +func newMinGasPriceTestEnv(t *testing.T, cfg *params.ChainConfig, n int) (*core.BlockChain, *txpool.TxPool) { + t.Helper() + + genesis := &core.Genesis{ + Config: cfg, + BaseFee: big.NewInt(params.InitialBaseFee), + } + _, blocks, _ := core.GenerateChainWithGenesis(genesis, ethash.NewFaker(), n, nil) + + db := rawdb.NewMemoryDatabase() + chain, err := core.NewBlockChain(db, nil, genesis, ethash.NewFaker(), vm.Config{}) + if err != nil { + t.Fatalf("failed to create blockchain: %v", err) + } + legacyPool := legacypool.New(legacypool.DefaultConfig, chain) + pool, err := txpool.New(0, chain, []txpool.SubPool{legacyPool}) + if err != nil { + t.Fatalf("failed to create tx pool: %v", err) + } + if _, err := chain.InsertChain(blocks); err != nil { + t.Fatalf("failed to insert blocks: %v", err) + } + return chain, pool +} diff --git a/core/txpool/validation.go b/core/txpool/validation.go index 802672b3bd2..b584bb5c33d 100644 --- a/core/txpool/validation.go +++ b/core/txpool/validation.go @@ -30,6 +30,18 @@ import ( "github.com/XinFinOrg/XDPoSChain/params" ) +// pendingBlockNumber returns the height a pooled transaction is priced at: it +// can only be included from the next block onwards, so gas schedule lookups +// resolve the fork tier one past the given head number. A nil input means no +// head is known and resolves to nil. Admission validation and the local +// tracker's price floor must both go through here so they cannot drift apart. +func pendingBlockNumber(number *big.Int) *big.Int { + if number == nil { + return nil + } + return new(big.Int).Add(number, common.Big1) +} + // ValidationOptions define certain differences between transaction validation // across the different pools without having to duplicate those checks. type ValidationOptions struct { @@ -238,10 +250,7 @@ func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, op ) // A pooled tx can only be included from the next block onwards, so gas // schedule lookups below resolve the fork tier at that height. - pendingNumber := number - if number != nil { - pendingNumber = new(big.Int).Add(number, common.Big1) - } + pendingNumber := pendingBlockNumber(number) if to != nil { if value, ok := opts.Trc21FeeCapacity[*to]; ok { feeCapacity = value