Skip to content

Commit

Permalink
Revert "core/txpool, eth/catalyst: ensure gas tip retains current val…
Browse files Browse the repository at this point in the history
…ue upon rollback" (#30521)

Reverts #30495

You are free to create a proper Clear method if that's the best way. But
one that does a proper cleanup, not some hacky call to set gas which
screws up logs, metrics and everything along the way. Also doesn't work
for legacy pool local transactions.

The current code had a hack in the simulated code, now we have a hack in
live txpooling code. No, that's not acceptable. I want the live code to
be proper, meaningful API, meaningful comments, meaningful
implementation.
  • Loading branch information
karalabe authored Sep 27, 2024
1 parent 52a9d89 commit 1df75db
Show file tree
Hide file tree
Showing 5 changed files with 61 additions and 86 deletions.
105 changes: 47 additions & 58 deletions core/txpool/blobpool/blobpool.go
Original file line number Diff line number Diff line change
Expand Up @@ -1011,56 +1011,6 @@ func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) error {
return nil
}

func (p *BlobPool) flushTransactionsBelowTip(tip *uint256.Int) {
for addr, txs := range p.index {
for i, tx := range txs {
if tx.execTipCap.Cmp(tip) < 0 {
// Drop the offending transaction
var (
ids = []uint64{tx.id}
nonces = []uint64{tx.nonce}
)
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].costCap)
p.stored -= uint64(tx.size)
delete(p.lookup, tx.hash)
txs[i] = nil

// Drop everything afterwards, no gaps allowed
for j, tx := range txs[i+1:] {
ids = append(ids, tx.id)
nonces = append(nonces, tx.nonce)

p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], tx.costCap)
p.stored -= uint64(tx.size)
delete(p.lookup, tx.hash)
txs[i+1+j] = nil
}
// Clear out the dropped transactions from the index
if i > 0 {
p.index[addr] = txs[:i]
heap.Fix(p.evict, p.evict.index[addr])
} else {
delete(p.index, addr)
delete(p.spent, addr)

heap.Remove(p.evict, p.evict.index[addr])
p.reserve(addr, false)
}
// Clear out the transactions from the data store
log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids)
dropUnderpricedMeter.Mark(int64(len(ids)))

for _, id := range ids {
if err := p.store.Delete(id); err != nil {
log.Error("Failed to delete dropped transaction", "id", id, "err", err)
}
}
break
}
}
}
}

// SetGasTip implements txpool.SubPool, allowing the blob pool's gas requirements
// to be kept in sync with the main transaction pool's gas requirements.
func (p *BlobPool) SetGasTip(tip *big.Int) {
Expand All @@ -1073,20 +1023,59 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {

// If the min miner fee increased, remove transactions below the new threshold
if old == nil || p.gasTip.Cmp(old) > 0 {
p.flushTransactionsBelowTip(p.gasTip)
for addr, txs := range p.index {
for i, tx := range txs {
if tx.execTipCap.Cmp(p.gasTip) < 0 {
// Drop the offending transaction
var (
ids = []uint64{tx.id}
nonces = []uint64{tx.nonce}
)
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].costCap)
p.stored -= uint64(tx.size)
delete(p.lookup, tx.hash)
txs[i] = nil

// Drop everything afterwards, no gaps allowed
for j, tx := range txs[i+1:] {
ids = append(ids, tx.id)
nonces = append(nonces, tx.nonce)

p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], tx.costCap)
p.stored -= uint64(tx.size)
delete(p.lookup, tx.hash)
txs[i+1+j] = nil
}
// Clear out the dropped transactions from the index
if i > 0 {
p.index[addr] = txs[:i]
heap.Fix(p.evict, p.evict.index[addr])
} else {
delete(p.index, addr)
delete(p.spent, addr)

heap.Remove(p.evict, p.evict.index[addr])
p.reserve(addr, false)
}
// Clear out the transactions from the data store
log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids)
dropUnderpricedMeter.Mark(int64(len(ids)))

for _, id := range ids {
if err := p.store.Delete(id); err != nil {
log.Error("Failed to delete dropped transaction", "id", id, "err", err)
}
}
break
}
}
}
}
log.Debug("Blobpool tip threshold updated", "tip", tip)
pooltipGauge.Update(tip.Int64())
p.updateStorageMetrics()
}

func (p *BlobPool) FlushAllTransactions() {
maxUint256 := uint256.MustFromBig(new(big.Int).Sub(new(big.Int).Lsh(common.Big1, 256), common.Big1))
p.lock.Lock()
defer p.lock.Unlock()
p.flushTransactionsBelowTip(maxUint256)
}

// 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 (p *BlobPool) validateTx(tx *types.Transaction) error {
Expand Down
23 changes: 6 additions & 17 deletions core/txpool/legacypool/legacypool.go
Original file line number Diff line number Diff line change
Expand Up @@ -429,15 +429,6 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs
return pool.txFeed.Subscribe(ch)
}

func (pool *LegacyPool) flushTransactionsBelowTip(tip *big.Int) {
// pool.priced is sorted by GasFeeCap, so we have to iterate through pool.all instead
drop := pool.all.RemotesBelowTip(tip)
for _, tx := range drop {
pool.removeTx(tx.Hash(), false, true)
}
pool.priced.Removed(len(drop))
}

// SetGasTip updates the minimum gas tip required by the transaction pool for a
// new transaction, and drops all transactions below this threshold.
func (pool *LegacyPool) SetGasTip(tip *big.Int) {
Expand All @@ -451,18 +442,16 @@ func (pool *LegacyPool) SetGasTip(tip *big.Int) {
pool.gasTip.Store(newTip)
// If the min miner fee increased, remove transactions below the new threshold
if newTip.Cmp(old) > 0 {
pool.flushTransactionsBelowTip(tip)
// pool.priced is sorted by GasFeeCap, so we have to iterate through pool.all instead
drop := pool.all.RemotesBelowTip(tip)
for _, tx := range drop {
pool.removeTx(tx.Hash(), false, true)
}
pool.priced.Removed(len(drop))
}
log.Info("Legacy pool tip threshold updated", "tip", newTip)
}

func (pool *LegacyPool) FlushAllTransactions() {
maxUint256 := new(big.Int).Sub(new(big.Int).Lsh(common.Big1, 256), common.Big1)
pool.mu.Lock()
defer pool.mu.Unlock()
pool.flushTransactionsBelowTip(maxUint256)
}

// Nonce returns the next nonce of an account, with all transactions executable
// by the pool already applied on top.
func (pool *LegacyPool) Nonce(addr common.Address) uint64 {
Expand Down
3 changes: 0 additions & 3 deletions core/txpool/subpool.go
Original file line number Diff line number Diff line change
Expand Up @@ -116,9 +116,6 @@ type SubPool interface {
// transaction, and drops all transactions below this threshold.
SetGasTip(tip *big.Int)

// FlushAllTransactions drops all transactions in the pool.
FlushAllTransactions()

// Has returns an indicator whether subpool has a transaction cached with the
// given hash.
Has(hash common.Hash) bool
Expand Down
7 changes: 0 additions & 7 deletions core/txpool/txpool.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,13 +104,6 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
return pool, nil
}

// FlushAllTransactions removes all transactions from all subpools
func (p *TxPool) FlushAllTransactions() {
for _, subpool := range p.subpools {
subpool.FlushAllTransactions()
}
}

// reserver is a method to create an address reservation callback to exclusively
// assign/deassign addresses to/from subpools. This can ensure that at any point
// in time, only a single subpool is able to manage an account, avoiding cross
Expand Down
9 changes: 8 additions & 1 deletion eth/catalyst/simulated_beacon.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"crypto/sha256"
"errors"
"fmt"
"math/big"
"sync"
"time"

Expand All @@ -33,6 +34,7 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
)

Expand Down Expand Up @@ -284,7 +286,12 @@ func (c *SimulatedBeacon) Commit() common.Hash {

// Rollback un-sends previously added transactions.
func (c *SimulatedBeacon) Rollback() {
c.eth.TxPool().FlushAllTransactions()
// Flush all transactions from the transaction pools
maxUint256 := new(big.Int).Sub(new(big.Int).Lsh(common.Big1, 256), common.Big1)
c.eth.TxPool().SetGasTip(maxUint256)
// Set the gas tip back to accept new transactions
// TODO (Marius van der Wijden): set gas tip to parameter passed by config
c.eth.TxPool().SetGasTip(big.NewInt(params.GWei))
}

// Fork sets the head to the provided hash.
Expand Down

0 comments on commit 1df75db

Please sign in to comment.