fix(core/txpool/locals): drop the transaction a tracked local tx replaces

TrackAll only ever added to the tracked set, while the per-nonce SortedMap silently overwrote the entry a replacement displaced. The replaced transaction was therefore never returned by Forward again, stayed tracked forever, was rewritten into the journal on every rotation and could win the nonce on the next load, resurrecting a transaction the user had already replaced, because rotation writes the tracked set in map order through a non-stable sort.

Drop the replaced transaction from the tracked set when its nonce is taken over, unless the pool still holds it: the pool decides which transaction occupies a nonce, and Add tracks a local transaction only after SubPool.Add has released its lock, so two concurrent submissions can be accepted in one order and reach TrackAll in the other. Keeping the transaction the pool holds then stops a replacement from dropping the live transaction and pinning the superseded one, which recheck could never resubmit successfully and which would leave the live one without local protection.

This 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 entry later in the file wins the nonce and the tracked set stays consistent, and the next rotation rewrites the journal from it.

Build the test environment from an explicit chain config so tests can pin the gas schedule instead of sharing the package level genesis.

Cover both orders in which a replacement can reach the tracker, and the concurrent interleaving where the original is added first and tracked last.

Author: gzliudan

core/txpool/locals/tx_tracker.go

@@ -101,11 +101,43 @@ 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 the pool's call: it holds at
+		// most one transaction per sender and nonce, and Add tracks a local
+		// transaction only after SubPool.Add has released its lock, so two
+		// concurrent submissions can be accepted in one order and reach here in
+		// the other. Keeping the transaction the pool still holds in that case
+		// stops a replacement from dropping the live transaction and pinning
+		// the superseded one, which recheck could never resubmit successfully.
+		//
+		// Dropping it 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 entry later in the file wins the
+		// nonce and the tracked set stays consistent. The next rotation then
+		// rewrites the journal from the converged set.
+		if replaced := list.Get(tx.Nonce()); replaced != nil {
+			if tracker.pool.Has(replaced.Hash()) {
+				log.Debug("Ignoring tracked local transaction the pool superseded", "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)

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,17 +51,30 @@ var (
 		},
 		BaseFee: big.NewInt(params.InitialBaseFee),
 	}
-	signer = types.LatestSigner(gspec.Config)
 )
 
 type testEnv struct {
 	chain   *core.BlockChain
 	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,7 +109,23 @@ 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() {
@@ -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,265 @@ func TestStartContinuesOnCorruptedJournal(t *testing.T) {
 		t.Fatal("Journal writer should be initialized even if journal load fails")
 	}
 }
+
+// replacementPair returns two transactions sharing a nonce: the one tracked
+// first, and the one that replaces it.
+func replacementPair(env *testEnv) (replaced, replacement *types.Transaction) {
+	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
+	}
+	return mk(common.Address{0x00}), mk(common.Address{0x01})
+}
+
+// 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()
+	// The replacement has to clear the pool's price bump to take the nonce.
+	bumped := new(big.Int).Div(
+		new(big.Int).Mul(price, big.NewInt(int64(100+legacypool.DefaultConfig.PriceBump))),
+		big.NewInt(100))
+	bumped.Add(bumped, big.NewInt(1))
+	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}, bumped)
+		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())
+	}
+}
+
+// TestJournalLoadKeepsLastEntryPerNonce pins the load path for the reverse file
+// order. A journal entry carries no timestamp or sequence number, so when a
+// journal written by an older version holds both a transaction and the one that
+// replaced it, the entry later in the file wins the nonce -- even when that is
+// the transaction the user replaced. TrackAll drops the loser from the tracked
+// set either way; it cannot tell which one is newer.
+func TestJournalLoadKeepsLastEntryPerNonce(t *testing.T) {
+	journalPath, replaced, _ := 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[replaced.Hash()]; !ok {
+		t.Fatalf("the entry later in the file must survive the load: %v", replaced.Hash())
+	}
+}