diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 3d66803fd7..06690431d0 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -1009,6 +1009,19 @@ func (pool *LegacyPool) get(hash common.Hash) *types.Transaction { return pool.all.Get(hash) } +type ownerReserver interface { + Owns(common.Address) bool +} + +func (pool *LegacyPool) releaseReservation(addr common.Address) { + if ownerAware, ok := pool.reserver.(ownerReserver); ok { + if !ownerAware.Owns(addr) { + return + } + } + _ = pool.reserver.Release(addr) +} + // GetRLP returns a RLP-encoded transaction if it is contained in the pool. func (pool *LegacyPool) GetRLP(hash common.Hash) []byte { tx := pool.all.Get(hash) @@ -1069,7 +1082,7 @@ func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool, unreserve bo _, hasQueued = pool.queue.get(addr) ) if !hasPending && !hasQueued { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } }() } @@ -1425,7 +1438,7 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T for _, addr := range removedAddresses { _, hasPending := pool.pending[addr] if !hasPending { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } } return promoted @@ -1526,7 +1539,7 @@ func (pool *LegacyPool) truncateQueue() { for _, addr := range removedAddresses { _, hasPending := pool.pending[addr] if !hasPending { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } } } @@ -1587,7 +1600,7 @@ func (pool *LegacyPool) demoteUnexecutables() { delete(pool.pending, addr) pendingAddrsGauge.Dec(1) if _, ok := pool.queue.get(addr); !ok { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } } } @@ -1834,11 +1847,11 @@ func (pool *LegacyPool) Clear() { for addr := range pool.pending { if _, ok := pool.queue.get(addr); !ok { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } } for _, addr := range pool.queue.addresses() { - pool.reserver.Release(addr) + pool.releaseReservation(addr) } pool.all.Clear() pool.priced.Reheap() diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index f8592ba001..3e50c97e8e 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -233,6 +233,13 @@ func (r *reserver) Has(address common.Address) bool { return false // reserver only supports a single pool } +func (r *reserver) Owns(address common.Address) bool { + r.lock.RLock() + defer r.lock.RUnlock() + _, exists := r.accounts[address] + return exists +} + func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) blockchain := newTestBlockChain(config, 10000000, statedb, new(event.Feed)) @@ -397,6 +404,46 @@ func TestStateChangeDuringReset(t *testing.T) { } } +func TestPromoteExecutablesQueueEmptyWithoutReservation(t *testing.T) { + t.Parallel() + + statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) + chain := newTestBlockChain(params.TestChainConfig, 10000000, statedb, new(event.Feed)) + + key, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("failed to generate key: %v", err) + } + addr := crypto.PubkeyToAddress(key.PublicKey) + statedb.SetBalance(addr, uint256.NewInt(params.Ether), tracing.BalanceChangeUnspecified) + + r := &reserver{accounts: make(map[common.Address]struct{})} + pool := New(testTxPoolConfig, chain) + if err := pool.Init(testTxPoolConfig.PriceLimit, chain.CurrentBlock(), r); err != nil { + t.Fatalf("failed to init pool: %v", err) + } + defer pool.Close() + <-pool.initDoneCh + + queuedTx := pricedTransaction(5, 100000, big.NewInt(params.GWei), key) + if err := pool.addRemoteSync(queuedTx); err != nil { + t.Fatalf("failed to add queued tx: %v", err) + } + + r.lock.Lock() + delete(r.accounts, addr) + r.lock.Unlock() + + pool.mu.Lock() + defer pool.mu.Unlock() + pool.currentState.SetNonce(addr, 10, tracing.NonceChangeUnspecified) + pool.promoteExecutables([]common.Address{addr}) + + if _, ok := pool.queue.get(addr); ok { + t.Fatal("queue should be empty after stale tx is dropped") + } +} + func testAddBalance(pool *LegacyPool, addr common.Address, amount *big.Int) { pool.mu.Lock() pool.currentState.AddBalance(addr, uint256.MustFromBig(amount), tracing.BalanceChangeUnspecified) diff --git a/core/txpool/reserver.go b/core/txpool/reserver.go index b6ecef9f1a..0d18a7af95 100644 --- a/core/txpool/reserver.go +++ b/core/txpool/reserver.go @@ -136,3 +136,12 @@ func (h *ReservationHandle) Has(address common.Address) bool { id, exists := h.tracker.accounts[address] return exists && id != h.id } + +// Owns reports whether this handle currently owns the reservation for address. +func (h *ReservationHandle) Owns(address common.Address) bool { + h.tracker.lock.RLock() + defer h.tracker.lock.RUnlock() + + id, exists := h.tracker.accounts[address] + return exists && id == h.id +}