/* Copyright 2022 The Erigon contributors Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package txpool import ( "bytes" "container/heap" "context" "encoding/binary" "encoding/hex" "encoding/json" "errors" "fmt" "math" "math/big" "runtime" "sort" "sync" "sync/atomic" "time" gokzg4844 "github.com/crate-crypto/go-kzg-4844" mapset "github.com/deckarep/golang-set/v2" "github.com/go-stack/stack" "github.com/google/btree" "github.com/hashicorp/golang-lru/v2/simplelru" "github.com/holiman/uint256" "github.com/ledgerwatch/log/v3" "github.com/ledgerwatch/erigon-lib/chain" "github.com/ledgerwatch/erigon-lib/common" "github.com/ledgerwatch/erigon-lib/common/assert" "github.com/ledgerwatch/erigon-lib/common/cmp" "github.com/ledgerwatch/erigon-lib/common/dbg" "github.com/ledgerwatch/erigon-lib/common/fixedgas" "github.com/ledgerwatch/erigon-lib/common/u256" libkzg "github.com/ledgerwatch/erigon-lib/crypto/kzg" "github.com/ledgerwatch/erigon-lib/gointerfaces" "github.com/ledgerwatch/erigon-lib/gointerfaces/grpcutil" "github.com/ledgerwatch/erigon-lib/gointerfaces/remote" proto_txpool "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool" "github.com/ledgerwatch/erigon-lib/kv" "github.com/ledgerwatch/erigon-lib/kv/kvcache" "github.com/ledgerwatch/erigon-lib/kv/mdbx" "github.com/ledgerwatch/erigon-lib/metrics" "github.com/ledgerwatch/erigon-lib/txpool/txpoolcfg" "github.com/ledgerwatch/erigon-lib/types" ) const DefaultBlockGasLimit = uint64(30000000) var ( processBatchTxsTimer = metrics.NewSummary(`pool_process_remote_txs`) addRemoteTxsTimer = metrics.NewSummary(`pool_add_remote_txs`) newBlockTimer = metrics.NewSummary(`pool_new_block`) writeToDBTimer = metrics.NewSummary(`pool_write_to_db`) propagateToNewPeerTimer = metrics.NewSummary(`pool_propagate_to_new_peer`) propagateNewTxsTimer = metrics.NewSummary(`pool_propagate_new_txs`) writeToDBBytesCounter = metrics.GetOrCreateGauge(`pool_write_to_db_bytes`) pendingSubCounter = metrics.GetOrCreateGauge(`txpool_pending`) queuedSubCounter = metrics.GetOrCreateGauge(`txpool_queued`) basefeeSubCounter = metrics.GetOrCreateGauge(`txpool_basefee`) ) var TraceAll = false // Pool is interface for the transaction pool // This interface exists for the convenience of testing, and not yet because // there are multiple implementations type Pool interface { ValidateSerializedTxn(serializedTxn []byte) error // Handle 3 main events - new remote txs from p2p, new local txs from RPC, new blocks from execution layer AddRemoteTxs(ctx context.Context, newTxs types.TxSlots) AddLocalTxs(ctx context.Context, newTxs types.TxSlots, tx kv.Tx) ([]txpoolcfg.DiscardReason, error) OnNewBlock(ctx context.Context, stateChanges *remote.StateChangeBatch, unwindTxs, unwindBlobTxs, minedTxs types.TxSlots, tx kv.Tx) error // IdHashKnown check whether transaction with given Id hash is known to the pool IdHashKnown(tx kv.Tx, hash []byte) (bool, error) FilterKnownIdHashes(tx kv.Tx, hashes types.Hashes) (unknownHashes types.Hashes, err error) Started() bool GetRlp(tx kv.Tx, hash []byte) ([]byte, error) AddNewGoodPeer(peerID types.PeerID) } var _ Pool = (*TxPool)(nil) // compile-time interface check // SubPoolMarker is an ordered bitset of five bits that's used to sort transactions into sub-pools. Bits meaning: // 1. Absence of nonce gaps. Set to 1 for transactions whose nonce is N, state nonce for the sender is M, and there are transactions for all nonces between M and N from the same sender. Set to 0 is the transaction's nonce is divided from the state nonce by one or more nonce gaps. // 2. Sufficient balance for gas. Set to 1 if the balance of sender's account in the state is B, nonce of the sender in the state is M, nonce of the transaction is N, and the sum of feeCap x gasLimit + transferred_value of all transactions from this sender with nonces N+1 ... M is no more than B. Set to 0 otherwise. In other words, this bit is set if there is currently a guarantee that the transaction and all its required prior transactions will be able to pay for gas. // 3. Not too much gas: Set to 1 if the transaction doesn't use too much gas // 4. Dynamic fee requirement. Set to 1 if feeCap of the transaction is no less than baseFee of the currently pending block. Set to 0 otherwise. // 5. Local transaction. Set to 1 if transaction is local. type SubPoolMarker uint8 const ( NoNonceGaps = 0b010000 EnoughBalance = 0b001000 NotTooMuchGas = 0b000100 EnoughFeeCapBlock = 0b000010 IsLocal = 0b000001 BaseFeePoolBits = NoNonceGaps + EnoughBalance + NotTooMuchGas ) // metaTx holds transaction and some metadata type metaTx struct { Tx *types.TxSlot minFeeCap uint256.Int nonceDistance uint64 // how far their nonces are from the state's nonce for the sender cumulativeBalanceDistance uint64 // how far their cumulativeRequiredBalance are from the state's balance for the sender minTip uint64 bestIndex int worstIndex int timestamp uint64 // when it was added to pool subPool SubPoolMarker currentSubPool SubPoolType minedBlockNum uint64 } func newMetaTx(slot *types.TxSlot, isLocal bool, timestamp uint64) *metaTx { mt := &metaTx{Tx: slot, worstIndex: -1, bestIndex: -1, timestamp: timestamp} if isLocal { mt.subPool = IsLocal } return mt } type SubPoolType uint8 const PendingSubPool SubPoolType = 1 const BaseFeeSubPool SubPoolType = 2 const QueuedSubPool SubPoolType = 3 func (sp SubPoolType) String() string { switch sp { case PendingSubPool: return "Pending" case BaseFeeSubPool: return "BaseFee" case QueuedSubPool: return "Queued" } return fmt.Sprintf("Unknown:%d", sp) } // sender - immutable structure which stores only nonce and balance of account type sender struct { balance uint256.Int nonce uint64 } func newSender(nonce uint64, balance uint256.Int) *sender { return &sender{nonce: nonce, balance: balance} } var emptySender = newSender(0, *uint256.NewInt(0)) func SortByNonceLess(a, b *metaTx) bool { if a.Tx.SenderID != b.Tx.SenderID { return a.Tx.SenderID < b.Tx.SenderID } return a.Tx.Nonce < b.Tx.Nonce } // TxPool - holds all pool-related data structures and lock-based tiny methods // most of logic implemented by pure tests-friendly functions // // txpool doesn't start any goroutines - "leave concurrency to user" design // txpool has no DB-TX fields - "leave db transactions management to user" design // txpool has _chainDB field - but it must maximize local state cache hit-rate - and perform minimum _chainDB transactions // // It preserve TxSlot objects immutable type TxPool struct { _chainDB kv.RoDB // remote db - use it wisely _stateCache kvcache.Cache lock *sync.Mutex recentlyConnectedPeers *recentlyConnectedPeers // all txs will be propagated to this peers eventually, and clear list senders *sendersBatch // batch processing of remote transactions // handling is fast enough without batching, but batching allows: // - fewer _chainDB transactions // - batch notifications about new txs (reduced P2P spam to other nodes about txs propagation) // - and as a result reducing lock contention unprocessedRemoteTxs *types.TxSlots unprocessedRemoteByHash map[string]int // to reject duplicates byHash map[string]*metaTx // tx_hash => tx : only those records not committed to db yet discardReasonsLRU *simplelru.LRU[string, txpoolcfg.DiscardReason] // tx_hash => discard_reason : non-persisted pending *PendingPool baseFee *SubPool queued *SubPool minedBlobTxsByBlock map[uint64][]*metaTx // (blockNum => slice): cache of recently mined blobs minedBlobTxsByHash map[string]*metaTx // (hash => mt): map of recently mined blobs isLocalLRU *simplelru.LRU[string, struct{}] // tx_hash => is_local : to restore isLocal flag of unwinded transactions newPendingTxs chan types.Announcements // notifications about new txs in Pending sub-pool all *BySenderAndNonce // senderID => (sorted map of tx nonce => *metaTx) deletedTxs []*metaTx // list of discarded txs since last db commit promoted types.Announcements cfg txpoolcfg.Config chainID uint256.Int lastSeenBlock atomic.Uint64 lastSeenCond *sync.Cond lastFinalizedBlock atomic.Uint64 started atomic.Bool pendingBaseFee atomic.Uint64 pendingBlobFee atomic.Uint64 // For gas accounting for blobs, which has its own dimension blockGasLimit atomic.Uint64 shanghaiTime *uint64 isPostShanghai atomic.Bool agraBlock *uint64 isPostAgra atomic.Bool cancunTime *uint64 isPostCancun atomic.Bool maxBlobsPerBlock uint64 feeCalculator FeeCalculator logger log.Logger } type FeeCalculator interface { CurrentFees(chainConfig *chain.Config, db kv.Getter) (baseFee uint64, blobFee uint64, minBlobGasPrice, blockGasLimit uint64, err error) } func New(newTxs chan types.Announcements, coreDB kv.RoDB, cfg txpoolcfg.Config, cache kvcache.Cache, chainID uint256.Int, shanghaiTime, agraBlock, cancunTime *big.Int, maxBlobsPerBlock uint64, feeCalculator FeeCalculator, logger log.Logger, ) (*TxPool, error) { localsHistory, err := simplelru.NewLRU[string, struct{}](10_000, nil) if err != nil { return nil, err } discardHistory, err := simplelru.NewLRU[string, txpoolcfg.DiscardReason](10_000, nil) if err != nil { return nil, err } byNonce := &BySenderAndNonce{ tree: btree.NewG[*metaTx](32, SortByNonceLess), search: &metaTx{Tx: &types.TxSlot{}}, senderIDTxnCount: map[uint64]int{}, senderIDBlobCount: map[uint64]uint64{}, } tracedSenders := make(map[common.Address]struct{}) for _, sender := range cfg.TracedSenders { tracedSenders[common.BytesToAddress([]byte(sender))] = struct{}{} } lock := &sync.Mutex{} res := &TxPool{ lock: lock, lastSeenCond: sync.NewCond(lock), byHash: map[string]*metaTx{}, isLocalLRU: localsHistory, discardReasonsLRU: discardHistory, all: byNonce, recentlyConnectedPeers: &recentlyConnectedPeers{}, pending: NewPendingSubPool(PendingSubPool, cfg.PendingSubPoolLimit), baseFee: NewSubPool(BaseFeeSubPool, cfg.BaseFeeSubPoolLimit), queued: NewSubPool(QueuedSubPool, cfg.QueuedSubPoolLimit), newPendingTxs: newTxs, _stateCache: cache, senders: newSendersCache(tracedSenders), _chainDB: coreDB, cfg: cfg, chainID: chainID, unprocessedRemoteTxs: &types.TxSlots{}, unprocessedRemoteByHash: map[string]int{}, minedBlobTxsByBlock: map[uint64][]*metaTx{}, minedBlobTxsByHash: map[string]*metaTx{}, maxBlobsPerBlock: maxBlobsPerBlock, feeCalculator: feeCalculator, logger: logger, } if shanghaiTime != nil { if !shanghaiTime.IsUint64() { return nil, errors.New("shanghaiTime overflow") } shanghaiTimeU64 := shanghaiTime.Uint64() res.shanghaiTime = &shanghaiTimeU64 } if agraBlock != nil { if !agraBlock.IsUint64() { return nil, errors.New("agraBlock overflow") } agraBlockU64 := agraBlock.Uint64() res.agraBlock = &agraBlockU64 } if cancunTime != nil { if !cancunTime.IsUint64() { return nil, errors.New("cancunTime overflow") } cancunTimeU64 := cancunTime.Uint64() res.cancunTime = &cancunTimeU64 } return res, nil } func (p *TxPool) Start(ctx context.Context, db kv.RwDB) error { if p.started.Load() { return nil } return db.View(ctx, func(tx kv.Tx) error { coreDb, _ := p.coreDBWithCache() coreTx, err := coreDb.BeginRo(ctx) if err != nil { return err } defer coreTx.Rollback() if err := p.fromDB(ctx, tx, coreTx); err != nil { return fmt.Errorf("loading pool from DB: %w", err) } if p.started.CompareAndSwap(false, true) { p.logger.Info("[txpool] Started") } return nil }) } func (p *TxPool) OnNewBlock(ctx context.Context, stateChanges *remote.StateChangeBatch, unwindTxs, unwindBlobTxs, minedTxs types.TxSlots, tx kv.Tx) error { defer newBlockTimer.ObserveDuration(time.Now()) //t := time.Now() coreDB, cache := p.coreDBWithCache() cache.OnNewBlock(stateChanges) coreTx, err := coreDB.BeginRo(ctx) if err != nil { return err } defer coreTx.Rollback() block := stateChanges.ChangeBatch[len(stateChanges.ChangeBatch)-1].BlockHeight baseFee := stateChanges.PendingBlockBaseFee available := len(p.pending.best.ms) defer func() { p.logger.Debug("[txpool] New block", "block", block, "unwound", len(unwindTxs.Txs), "mined", len(minedTxs.Txs), "baseFee", baseFee, "pending-pre", available, "pending", p.pending.Len(), "baseFee", p.baseFee.Len(), "queued", p.queued.Len(), "err", err) }() if err = minedTxs.Valid(); err != nil { return err } cacheView, err := cache.View(ctx, coreTx) if err != nil { return err } p.lock.Lock() defer func() { if err == nil { p.lastSeenBlock.Store(block) p.lastSeenCond.Broadcast() } p.lock.Unlock() }() if assert.Enable { if _, err := kvcache.AssertCheckValues(ctx, coreTx, cache); err != nil { p.logger.Error("AssertCheckValues", "err", err, "stack", stack.Trace().String()) } } pendingBaseFee, baseFeeChanged := p.setBaseFee(baseFee) // Update pendingBase for all pool queues and slices if baseFeeChanged { p.pending.best.pendingBaseFee = pendingBaseFee p.pending.worst.pendingBaseFee = pendingBaseFee p.baseFee.best.pendingBastFee = pendingBaseFee p.baseFee.worst.pendingBaseFee = pendingBaseFee p.queued.best.pendingBastFee = pendingBaseFee p.queued.worst.pendingBaseFee = pendingBaseFee } pendingBlobFee := stateChanges.PendingBlobFeePerGas p.setBlobFee(pendingBlobFee) p.blockGasLimit.Store(stateChanges.BlockGasLimit) for i, txn := range unwindBlobTxs.Txs { if txn.Type == types.BlobTxType { knownBlobTxn, err := p.getCachedBlobTxnLocked(coreTx, txn.IDHash[:]) if err != nil { return err } if knownBlobTxn != nil { unwindTxs.Append(knownBlobTxn.Tx, unwindBlobTxs.Senders.At(i), false) } } } if err = p.senders.onNewBlock(stateChanges, unwindTxs, minedTxs, p.logger); err != nil { return err } _, unwindTxs, err = p.validateTxs(&unwindTxs, cacheView) if err != nil { return err } if assert.Enable { for _, txn := range unwindTxs.Txs { if txn.SenderID == 0 { panic(fmt.Errorf("onNewBlock.unwindTxs: senderID can't be zero")) } } for _, txn := range minedTxs.Txs { if txn.SenderID == 0 { panic(fmt.Errorf("onNewBlock.minedTxs: senderID can't be zero")) } } } if err = p.processMinedFinalizedBlobs(coreTx, minedTxs.Txs, stateChanges.FinalizedBlock); err != nil { return err } if err = p.removeMined(p.all, minedTxs.Txs); err != nil { return err } var announcements types.Announcements announcements, err = p.addTxsOnNewBlock(block, cacheView, stateChanges, p.senders, unwindTxs, /* newTxs */ pendingBaseFee, stateChanges.BlockGasLimit, p.logger) if err != nil { return err } p.pending.EnforceWorstInvariants() p.baseFee.EnforceInvariants() p.queued.EnforceInvariants() p.promote(pendingBaseFee, pendingBlobFee, &announcements, p.logger) p.pending.EnforceBestInvariants() p.promoted.Reset() p.promoted.AppendOther(announcements) if p.promoted.Len() > 0 { select { case p.newPendingTxs <- p.promoted.Copy(): default: } } return nil } func (p *TxPool) processRemoteTxs(ctx context.Context) error { if !p.Started() { return fmt.Errorf("txpool not started yet") } defer processBatchTxsTimer.ObserveDuration(time.Now()) coreDB, cache := p.coreDBWithCache() coreTx, err := coreDB.BeginRo(ctx) if err != nil { return err } defer coreTx.Rollback() cacheView, err := cache.View(ctx, coreTx) if err != nil { return err } //t := time.Now() p.lock.Lock() defer p.lock.Unlock() l := len(p.unprocessedRemoteTxs.Txs) if l == 0 { return nil } err = p.senders.registerNewSenders(p.unprocessedRemoteTxs, p.logger) if err != nil { return err } _, newTxs, err := p.validateTxs(p.unprocessedRemoteTxs, cacheView) if err != nil { return err } announcements, _, err := p.addTxs(p.lastSeenBlock.Load(), cacheView, p.senders, newTxs, p.pendingBaseFee.Load(), p.pendingBlobFee.Load(), p.blockGasLimit.Load(), true, p.logger) if err != nil { return err } p.promoted.Reset() p.promoted.AppendOther(announcements) if p.promoted.Len() > 0 { select { case <-ctx.Done(): return nil case p.newPendingTxs <- p.promoted.Copy(): default: } } p.unprocessedRemoteTxs.Resize(0) p.unprocessedRemoteByHash = map[string]int{} //p.logger.Info("[txpool] on new txs", "amount", len(newPendingTxs.txs), "in", time.Since(t)) return nil } func (p *TxPool) getRlpLocked(tx kv.Tx, hash []byte) (rlpTxn []byte, sender common.Address, isLocal bool, err error) { txn, ok := p.byHash[string(hash)] if ok && txn.Tx.Rlp != nil { return txn.Tx.Rlp, p.senders.senderID2Addr[txn.Tx.SenderID], txn.subPool&IsLocal > 0, nil } v, err := tx.GetOne(kv.PoolTransaction, hash) if err != nil { return nil, common.Address{}, false, err } if v == nil { return nil, common.Address{}, false, nil } return v[20:], *(*[20]byte)(v[:20]), txn != nil && txn.subPool&IsLocal > 0, nil } func (p *TxPool) GetRlp(tx kv.Tx, hash []byte) ([]byte, error) { p.lock.Lock() defer p.lock.Unlock() rlpTx, _, _, err := p.getRlpLocked(tx, hash) return common.Copy(rlpTx), err } func (p *TxPool) AppendLocalAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte) { p.lock.Lock() defer p.lock.Unlock() for hash, txn := range p.byHash { if txn.subPool&IsLocal == 0 { continue } types = append(types, txn.Tx.Type) sizes = append(sizes, txn.Tx.Size) hashes = append(hashes, hash...) } return types, sizes, hashes } func (p *TxPool) AppendRemoteAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte) { p.lock.Lock() defer p.lock.Unlock() for hash, txn := range p.byHash { if txn.subPool&IsLocal != 0 { continue } types = append(types, txn.Tx.Type) sizes = append(sizes, txn.Tx.Size) hashes = append(hashes, hash...) } for hash, txIdx := range p.unprocessedRemoteByHash { txSlot := p.unprocessedRemoteTxs.Txs[txIdx] types = append(types, txSlot.Type) sizes = append(sizes, txSlot.Size) hashes = append(hashes, hash...) } return types, sizes, hashes } func (p *TxPool) AppendAllAnnouncements(types []byte, sizes []uint32, hashes []byte) ([]byte, []uint32, []byte) { types, sizes, hashes = p.AppendLocalAnnouncements(types, sizes, hashes) types, sizes, hashes = p.AppendRemoteAnnouncements(types, sizes, hashes) return types, sizes, hashes } func (p *TxPool) idHashKnown(tx kv.Tx, hash []byte, hashS string) (bool, error) { if _, ok := p.unprocessedRemoteByHash[hashS]; ok { return true, nil } if _, ok := p.discardReasonsLRU.Get(hashS); ok { return true, nil } if _, ok := p.byHash[hashS]; ok { return true, nil } if _, ok := p.minedBlobTxsByHash[hashS]; ok { return true, nil } return tx.Has(kv.PoolTransaction, hash) } func (p *TxPool) IdHashKnown(tx kv.Tx, hash []byte) (bool, error) { hashS := string(hash) p.lock.Lock() defer p.lock.Unlock() return p.idHashKnown(tx, hash, hashS) } func (p *TxPool) FilterKnownIdHashes(tx kv.Tx, hashes types.Hashes) (unknownHashes types.Hashes, err error) { p.lock.Lock() defer p.lock.Unlock() for i := 0; i < len(hashes); i += 32 { known, err := p.idHashKnown(tx, hashes[i:i+32], string(hashes[i:i+32])) if err != nil { return unknownHashes, err } if !known { unknownHashes = append(unknownHashes, hashes[i:i+32]...) } } return unknownHashes, err } func (p *TxPool) getUnprocessedTxn(hashS string) (*types.TxSlot, bool) { if i, ok := p.unprocessedRemoteByHash[hashS]; ok { return p.unprocessedRemoteTxs.Txs[i], true } return nil, false } func (p *TxPool) getCachedBlobTxnLocked(tx kv.Tx, hash []byte) (*metaTx, error) { hashS := string(hash) if mt, ok := p.minedBlobTxsByHash[hashS]; ok { return mt, nil } if txn, ok := p.getUnprocessedTxn(hashS); ok { return newMetaTx(txn, false, 0), nil } if mt, ok := p.byHash[hashS]; ok { return mt, nil } has, err := tx.Has(kv.PoolTransaction, hash) if err != nil { return nil, err } if !has { return nil, nil } txn, err := tx.GetOne(kv.PoolTransaction, hash) if err != nil { return nil, err } parseCtx := types.NewTxParseContext(p.chainID, false) parseCtx.WithSender(false) txSlot := &types.TxSlot{} parseCtx.ParseTransaction(txn, 0, txSlot, nil, false, true, nil) return newMetaTx(txSlot, false, 0), nil } func (p *TxPool) IsLocal(idHash []byte) bool { hashS := string(idHash) p.lock.Lock() defer p.lock.Unlock() return p.isLocalLRU.Contains(hashS) } func (p *TxPool) AddNewGoodPeer(peerID types.PeerID) { p.recentlyConnectedPeers.AddPeer(peerID) } func (p *TxPool) Started() bool { return p.started.Load() } func (p *TxPool) best(n uint16, txs *types.TxsRlp, tx kv.Tx, onTopOf, availableGas, availableBlobGas uint64, yielded mapset.Set[[32]byte]) (bool, int, error) { p.lock.Lock() defer p.lock.Unlock() for last := p.lastSeenBlock.Load(); last < onTopOf; last = p.lastSeenBlock.Load() { p.logger.Debug("[txpool] Waiting for block", "expecting", onTopOf, "lastSeen", last, "txRequested", n, "pending", p.pending.Len(), "baseFee", p.baseFee.Len(), "queued", p.queued.Len()) p.lastSeenCond.Wait() } best := p.pending.best isShanghai := p.isShanghai() || p.isAgra() txs.Resize(uint(cmp.Min(int(n), len(best.ms)))) var toRemove []*metaTx count := 0 i := 0 defer func() { p.logger.Debug("[txpool] Processing best request", "last", onTopOf, "txRequested", n, "txAvailable", len(best.ms), "txProcessed", i, "txReturned", count) }() for ; count < int(n) && i < len(best.ms); i++ { // if we wouldn't have enough gas for a standard transaction then quit out early if availableGas < fixedgas.TxGas { break } mt := best.ms[i] if yielded.Contains(mt.Tx.IDHash) { continue } if mt.Tx.Gas >= p.blockGasLimit.Load() { // Skip transactions with very large gas limit continue } rlpTx, sender, isLocal, err := p.getRlpLocked(tx, mt.Tx.IDHash[:]) if err != nil { return false, count, err } if len(rlpTx) == 0 { toRemove = append(toRemove, mt) continue } // Skip transactions that require more blob gas than is available blobCount := uint64(len(mt.Tx.BlobHashes)) if blobCount*fixedgas.BlobGasPerBlob > availableBlobGas { continue } availableBlobGas -= blobCount * fixedgas.BlobGasPerBlob // make sure we have enough gas in the caller to add this transaction. // not an exact science using intrinsic gas but as close as we could hope for at // this stage intrinsicGas, _ := txpoolcfg.CalcIntrinsicGas(uint64(mt.Tx.DataLen), uint64(mt.Tx.DataNonZeroLen), nil, mt.Tx.Creation, true, true, isShanghai) if intrinsicGas > availableGas { // we might find another TX with a low enough intrinsic gas to include so carry on continue } availableGas -= intrinsicGas txs.Txs[count] = rlpTx copy(txs.Senders.At(count), sender.Bytes()) txs.IsLocal[count] = isLocal yielded.Add(mt.Tx.IDHash) count++ } txs.Resize(uint(count)) if len(toRemove) > 0 { for _, mt := range toRemove { p.pending.Remove(mt, "best", p.logger) } } return true, count, nil } func (p *TxPool) YieldBest(n uint16, txs *types.TxsRlp, tx kv.Tx, onTopOf, availableGas, availableBlobGas uint64, toSkip mapset.Set[[32]byte]) (bool, int, error) { return p.best(n, txs, tx, onTopOf, availableGas, availableBlobGas, toSkip) } func (p *TxPool) PeekBest(n uint16, txs *types.TxsRlp, tx kv.Tx, onTopOf, availableGas, availableBlobGas uint64) (bool, error) { set := mapset.NewThreadUnsafeSet[[32]byte]() onTime, _, err := p.YieldBest(n, txs, tx, onTopOf, availableGas, availableBlobGas, set) return onTime, err } func (p *TxPool) CountContent() (int, int, int) { p.lock.Lock() defer p.lock.Unlock() return p.pending.Len(), p.baseFee.Len(), p.queued.Len() } func (p *TxPool) AddRemoteTxs(_ context.Context, newTxs types.TxSlots) { if p.cfg.NoGossip { // if no gossip, then // disable adding remote transactions // consume remote tx from fetch return } defer addRemoteTxsTimer.ObserveDuration(time.Now()) p.lock.Lock() defer p.lock.Unlock() for i, txn := range newTxs.Txs { hashS := string(txn.IDHash[:]) _, ok := p.unprocessedRemoteByHash[hashS] if ok { continue } p.unprocessedRemoteByHash[hashS] = len(p.unprocessedRemoteTxs.Txs) p.unprocessedRemoteTxs.Append(txn, newTxs.Senders.At(i), false) } } func toBlobs(_blobs [][]byte) []gokzg4844.Blob { blobs := make([]gokzg4844.Blob, len(_blobs)) for i, _blob := range _blobs { var b gokzg4844.Blob copy(b[:], _blob) blobs[i] = b } return blobs } func (p *TxPool) validateTx(txn *types.TxSlot, isLocal bool, stateCache kvcache.CacheView) txpoolcfg.DiscardReason { isShanghai := p.isShanghai() || p.isAgra() if isShanghai { if txn.DataLen > fixedgas.MaxInitCodeSize { return txpoolcfg.InitCodeTooLarge } } if txn.Type == types.BlobTxType { if !p.isCancun() { return txpoolcfg.TypeNotActivated } if txn.Creation { return txpoolcfg.CreateBlobTxn } blobCount := uint64(len(txn.BlobHashes)) if blobCount == 0 { return txpoolcfg.NoBlobs } if blobCount > p.maxBlobsPerBlock { return txpoolcfg.TooManyBlobs } equalNumber := len(txn.BlobHashes) == len(txn.Blobs) && len(txn.Blobs) == len(txn.Commitments) && len(txn.Commitments) == len(txn.Proofs) if !equalNumber { return txpoolcfg.UnequalBlobTxExt } for i := 0; i < len(txn.Commitments); i++ { if libkzg.KZGToVersionedHash(txn.Commitments[i]) != libkzg.VersionedHash(txn.BlobHashes[i]) { return txpoolcfg.BlobHashCheckFail } } // https://github.com/ethereum/consensus-specs/blob/017a8495f7671f5fff2075a9bfc9238c1a0982f8/specs/deneb/polynomial-commitments.md#verify_blob_kzg_proof_batch kzgCtx := libkzg.Ctx() err := kzgCtx.VerifyBlobKZGProofBatch(toBlobs(txn.Blobs), txn.Commitments, txn.Proofs) if err != nil { return txpoolcfg.UnmatchedBlobTxExt } } // Drop non-local transactions under our own minimal accepted gas price or tip if !isLocal && uint256.NewInt(p.cfg.MinFeeCap).Cmp(&txn.FeeCap) == 1 { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx underpriced idHash=%x local=%t, feeCap=%d, cfg.MinFeeCap=%d", txn.IDHash, isLocal, txn.FeeCap, p.cfg.MinFeeCap)) } return txpoolcfg.UnderPriced } gas, reason := txpoolcfg.CalcIntrinsicGas(uint64(txn.DataLen), uint64(txn.DataNonZeroLen), nil, txn.Creation, true, true, isShanghai) if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx intrinsic gas idHash=%x gas=%d", txn.IDHash, gas)) } if reason != txpoolcfg.Success { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx intrinsic gas calculated failed idHash=%x reason=%s", txn.IDHash, reason)) } return reason } if gas > txn.Gas { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx intrinsic gas > txn.gas idHash=%x gas=%d, txn.gas=%d", txn.IDHash, gas, txn.Gas)) } return txpoolcfg.IntrinsicGas } if !isLocal && uint64(p.all.count(txn.SenderID)) > p.cfg.AccountSlots { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx marked as spamming idHash=%x slots=%d, limit=%d", txn.IDHash, p.all.count(txn.SenderID), p.cfg.AccountSlots)) } return txpoolcfg.Spammer } if !isLocal && p.all.blobCount(txn.SenderID) > p.cfg.BlobSlots { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx marked as spamming (too many blobs) idHash=%x slots=%d, limit=%d", txn.IDHash, p.all.count(txn.SenderID), p.cfg.AccountSlots)) } return txpoolcfg.Spammer } // check nonce and balance senderNonce, senderBalance, _ := p.senders.info(stateCache, txn.SenderID) if senderNonce > txn.Nonce { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx nonce too low idHash=%x nonce in state=%d, txn.nonce=%d", txn.IDHash, senderNonce, txn.Nonce)) } return txpoolcfg.NonceTooLow } // Transactor should have enough funds to cover the costs total := requiredBalance(txn) if senderBalance.Cmp(total) < 0 { if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: validateTx insufficient funds idHash=%x balance in state=%d, txn.gas*txn.tip=%d", txn.IDHash, senderBalance, total)) } return txpoolcfg.InsufficientFunds } return txpoolcfg.Success } var maxUint256 = new(uint256.Int).SetAllOne() // Sender should have enough balance for: gasLimit x feeCap + blobGas x blobFeeCap + transferred_value // See YP, Eq (61) in Section 6.2 "Execution" func requiredBalance(txn *types.TxSlot) *uint256.Int { // See https://github.com/ethereum/EIPs/pull/3594 total := uint256.NewInt(txn.Gas) _, overflow := total.MulOverflow(total, &txn.FeeCap) if overflow { return maxUint256 } // and https://eips.ethereum.org/EIPS/eip-4844#gas-accounting blobCount := uint64(len(txn.BlobHashes)) if blobCount != 0 { maxBlobGasCost := uint256.NewInt(fixedgas.BlobGasPerBlob) maxBlobGasCost.Mul(maxBlobGasCost, uint256.NewInt(blobCount)) _, overflow = maxBlobGasCost.MulOverflow(maxBlobGasCost, &txn.BlobFeeCap) if overflow { return maxUint256 } _, overflow = total.AddOverflow(total, maxBlobGasCost) if overflow { return maxUint256 } } _, overflow = total.AddOverflow(total, &txn.Value) if overflow { return maxUint256 } return total } func (p *TxPool) isShanghai() bool { // once this flag has been set for the first time we no longer need to check the timestamp set := p.isPostShanghai.Load() if set { return true } if p.shanghaiTime == nil { return false } shanghaiTime := *p.shanghaiTime // a zero here means Shanghai is always active if shanghaiTime == 0 { p.isPostShanghai.Swap(true) return true } now := time.Now().Unix() activated := uint64(now) >= shanghaiTime if activated { p.isPostShanghai.Swap(true) } return activated } func (p *TxPool) isAgra() bool { // once this flag has been set for the first time we no longer need to check the block set := p.isPostAgra.Load() if set { return true } if p.agraBlock == nil { return false } agraBlock := *p.agraBlock // a zero here means Agra is always active if agraBlock == 0 { p.isPostAgra.Swap(true) return true } tx, err := p._chainDB.BeginRo(context.Background()) if err != nil { return false } defer tx.Rollback() head_block, err := chain.CurrentBlockNumber(tx) if head_block == nil || err != nil { return false } // A new block is built on top of the head block, so when the head is agraBlock-1, // the new block should use the Agra rules. activated := (*head_block + 1) >= agraBlock if activated { p.isPostAgra.Swap(true) } return activated } func (p *TxPool) isCancun() bool { // once this flag has been set for the first time we no longer need to check the timestamp set := p.isPostCancun.Load() if set { return true } if p.cancunTime == nil { return false } cancunTime := *p.cancunTime // a zero here means Cancun is always active if cancunTime == 0 { p.isPostCancun.Swap(true) return true } now := time.Now().Unix() activated := uint64(now) >= cancunTime if activated { p.isPostCancun.Swap(true) } return activated } // Check that that the serialized txn should not exceed a certain max size func (p *TxPool) ValidateSerializedTxn(serializedTxn []byte) error { const ( // txSlotSize is used to calculate how many data slots a single transaction // takes up based on its size. The slots are used as DoS protection, ensuring // that validating a new transaction remains a constant operation (in reality // O(maxslots), where max slots are 4 currently). txSlotSize = 32 * 1024 // txMaxSize is the maximum size a single transaction can have. This field has // non-trivial consequences: larger transactions are significantly harder and // more expensive to propagate; larger transactions also take more resources // to validate whether they fit into the pool or not. txMaxSize = 4 * txSlotSize // 128KB // Should be enough for a transaction with 6 blobs blobTxMaxSize = 800_000 ) txType, err := types.PeekTransactionType(serializedTxn) if err != nil { return err } maxSize := txMaxSize if txType == types.BlobTxType { maxSize = blobTxMaxSize } if len(serializedTxn) > maxSize { return types.ErrRlpTooBig } return nil } func (p *TxPool) validateTxs(txs *types.TxSlots, stateCache kvcache.CacheView) (reasons []txpoolcfg.DiscardReason, goodTxs types.TxSlots, err error) { // reasons is pre-sized for direct indexing, with the default zero // value DiscardReason of NotSet reasons = make([]txpoolcfg.DiscardReason, len(txs.Txs)) if err := txs.Valid(); err != nil { return reasons, goodTxs, err } goodCount := 0 for i, txn := range txs.Txs { reason := p.validateTx(txn, txs.IsLocal[i], stateCache) if reason == txpoolcfg.Success { goodCount++ // Success here means no DiscardReason yet, so leave it NotSet continue } if reason == txpoolcfg.Spammer { p.punishSpammer(txn.SenderID) } reasons[i] = reason } goodTxs.Resize(uint(goodCount)) j := 0 for i, txn := range txs.Txs { if reasons[i] == txpoolcfg.NotSet { goodTxs.Txs[j] = txn goodTxs.IsLocal[j] = txs.IsLocal[i] copy(goodTxs.Senders.At(j), txs.Senders.At(i)) j++ } } return reasons, goodTxs, nil } // punishSpammer by drop half of it's transactions with high nonce func (p *TxPool) punishSpammer(spammer uint64) { count := p.all.count(spammer) / 2 if count > 0 { txsToDelete := make([]*metaTx, 0, count) p.all.descend(spammer, func(mt *metaTx) bool { txsToDelete = append(txsToDelete, mt) count-- return count > 0 }) for _, mt := range txsToDelete { switch mt.currentSubPool { case PendingSubPool: p.pending.Remove(mt, "punishSpammer", p.logger) case BaseFeeSubPool: p.baseFee.Remove(mt, "punishSpammer", p.logger) case QueuedSubPool: p.queued.Remove(mt, "punishSpammer", p.logger) default: //already removed } p.discardLocked(mt, txpoolcfg.Spammer) // can't call it while iterating by all } } } func fillDiscardReasons(reasons []txpoolcfg.DiscardReason, newTxs types.TxSlots, discardReasonsLRU *simplelru.LRU[string, txpoolcfg.DiscardReason]) []txpoolcfg.DiscardReason { for i := range reasons { if reasons[i] != txpoolcfg.NotSet { continue } reason, ok := discardReasonsLRU.Get(string(newTxs.Txs[i].IDHash[:])) if ok { reasons[i] = reason } else { reasons[i] = txpoolcfg.Success } } return reasons } func (p *TxPool) AddLocalTxs(ctx context.Context, newTransactions types.TxSlots, tx kv.Tx) ([]txpoolcfg.DiscardReason, error) { coreDb, cache := p.coreDBWithCache() coreTx, err := coreDb.BeginRo(ctx) if err != nil { return nil, err } defer coreTx.Rollback() cacheView, err := cache.View(ctx, coreTx) if err != nil { return nil, err } p.lock.Lock() defer p.lock.Unlock() if err = p.senders.registerNewSenders(&newTransactions, p.logger); err != nil { return nil, err } reasons, newTxs, err := p.validateTxs(&newTransactions, cacheView) if err != nil { return nil, err } announcements, addReasons, err := p.addTxs(p.lastSeenBlock.Load(), cacheView, p.senders, newTxs, p.pendingBaseFee.Load(), p.pendingBlobFee.Load(), p.blockGasLimit.Load(), true, p.logger) if err == nil { for i, reason := range addReasons { if reason != txpoolcfg.NotSet { reasons[i] = reason } } } else { return nil, err } p.promoted.Reset() p.promoted.AppendOther(announcements) reasons = fillDiscardReasons(reasons, newTxs, p.discardReasonsLRU) for i, reason := range reasons { if reason == txpoolcfg.Success { txn := newTxs.Txs[i] if txn.Traced { p.logger.Info(fmt.Sprintf("TX TRACING: AddLocalTxs promotes idHash=%x, senderId=%d", txn.IDHash, txn.SenderID)) } p.promoted.Append(txn.Type, txn.Size, txn.IDHash[:]) } } if p.promoted.Len() > 0 { select { case p.newPendingTxs <- p.promoted.Copy(): default: } } return reasons, nil } func (p *TxPool) coreDBWithCache() (kv.RoDB, kvcache.Cache) { p.lock.Lock() defer p.lock.Unlock() return p._chainDB, p._stateCache } func (p *TxPool) addTxs(blockNum uint64, cacheView kvcache.CacheView, senders *sendersBatch, newTxs types.TxSlots, pendingBaseFee, pendingBlobFee, blockGasLimit uint64, collect bool, logger log.Logger) (types.Announcements, []txpoolcfg.DiscardReason, error) { if assert.Enable { for _, txn := range newTxs.Txs { if txn.SenderID == 0 { panic(fmt.Errorf("senderID can't be zero")) } } } // This can be thought of a reverse operation from the one described before. // When a block that was deemed "the best" of its height, is no longer deemed "the best", the // transactions contained in it, are now viable for inclusion in other blocks, and therefore should // be returned into the transaction pool. // An interesting note here is that if the block contained any transactions local to the node, // by being first removed from the pool (from the "local" part of it), and then re-injected, // they effective lose their priority over the "remote" transactions. In order to prevent that, // somehow the fact that certain transactions were local, needs to be remembered for some // time (up to some "immutability threshold"). sendersWithChangedState := map[uint64]struct{}{} discardReasons := make([]txpoolcfg.DiscardReason, len(newTxs.Txs)) announcements := types.Announcements{} for i, txn := range newTxs.Txs { if found, ok := p.byHash[string(txn.IDHash[:])]; ok { discardReasons[i] = txpoolcfg.DuplicateHash // In case if the transition is stuck, "poke" it to rebroadcast if collect && newTxs.IsLocal[i] && (found.currentSubPool == PendingSubPool || found.currentSubPool == BaseFeeSubPool) { announcements.Append(found.Tx.Type, found.Tx.Size, found.Tx.IDHash[:]) } continue } mt := newMetaTx(txn, newTxs.IsLocal[i], blockNum) if reason := p.addLocked(mt, &announcements); reason != txpoolcfg.NotSet { discardReasons[i] = reason continue } discardReasons[i] = txpoolcfg.NotSet // unnecessary if txn.Traced { logger.Info(fmt.Sprintf("TX TRACING: schedule sendersWithChangedState idHash=%x senderId=%d", txn.IDHash, mt.Tx.SenderID)) } sendersWithChangedState[mt.Tx.SenderID] = struct{}{} } for senderID := range sendersWithChangedState { nonce, balance, err := senders.info(cacheView, senderID) if err != nil { return announcements, discardReasons, err } p.onSenderStateChange(senderID, nonce, balance, blockGasLimit, logger) } p.promote(pendingBaseFee, pendingBlobFee, &announcements, logger) p.pending.EnforceBestInvariants() return announcements, discardReasons, nil } // TODO: Looks like a copy of the above func (p *TxPool) addTxsOnNewBlock(blockNum uint64, cacheView kvcache.CacheView, stateChanges *remote.StateChangeBatch, senders *sendersBatch, newTxs types.TxSlots, pendingBaseFee uint64, blockGasLimit uint64, logger log.Logger) (types.Announcements, error) { if assert.Enable { for _, txn := range newTxs.Txs { if txn.SenderID == 0 { panic(fmt.Errorf("senderID can't be zero")) } } } // This can be thought of a reverse operation from the one described before. // When a block that was deemed "the best" of its height, is no longer deemed "the best", the // transactions contained in it, are now viable for inclusion in other blocks, and therefore should // be returned into the transaction pool. // An interesting note here is that if the block contained any transactions local to the node, // by being first removed from the pool (from the "local" part of it), and then re-injected, // they effective lose their priority over the "remote" transactions. In order to prevent that, // somehow the fact that certain transactions were local, needs to be remembered for some // time (up to some "immutability threshold"). sendersWithChangedState := map[uint64]struct{}{} announcements := types.Announcements{} for i, txn := range newTxs.Txs { if _, ok := p.byHash[string(txn.IDHash[:])]; ok { continue } mt := newMetaTx(txn, newTxs.IsLocal[i], blockNum) if reason := p.addLocked(mt, &announcements); reason != txpoolcfg.NotSet { p.discardLocked(mt, reason) continue } sendersWithChangedState[mt.Tx.SenderID] = struct{}{} } // add senders changed in state to `sendersWithChangedState` list for _, changesList := range stateChanges.ChangeBatch { for _, change := range changesList.Changes { switch change.Action { case remote.Action_UPSERT, remote.Action_UPSERT_CODE: if change.Incarnation > 0 { continue } addr := gointerfaces.ConvertH160toAddress(change.Address) id, ok := senders.getID(addr) if !ok { continue } sendersWithChangedState[id] = struct{}{} } } } for senderID := range sendersWithChangedState { nonce, balance, err := senders.info(cacheView, senderID) if err != nil { return announcements, err } p.onSenderStateChange(senderID, nonce, balance, blockGasLimit, logger) } return announcements, nil } func (p *TxPool) setBaseFee(baseFee uint64) (uint64, bool) { changed := false if baseFee > 0 { changed = baseFee != p.pendingBaseFee.Load() p.pendingBaseFee.Store(baseFee) } return p.pendingBaseFee.Load(), changed } func (p *TxPool) setBlobFee(blobFee uint64) { if blobFee > 0 { p.pendingBlobFee.Store(blobFee) } } func (p *TxPool) addLocked(mt *metaTx, announcements *types.Announcements) txpoolcfg.DiscardReason { // Insert to pending pool, if pool doesn't have txn with same Nonce and bigger Tip found := p.all.get(mt.Tx.SenderID, mt.Tx.Nonce) if found != nil { if found.Tx.Type == types.BlobTxType && mt.Tx.Type != types.BlobTxType { return txpoolcfg.BlobTxReplace } priceBump := p.cfg.PriceBump //Blob txn threshold checks for replace txn if mt.Tx.Type == types.BlobTxType { priceBump = p.cfg.BlobPriceBump blobFeeThreshold, overflow := (&uint256.Int{}).MulDivOverflow( &found.Tx.BlobFeeCap, uint256.NewInt(100+priceBump), uint256.NewInt(100), ) if mt.Tx.BlobFeeCap.Lt(blobFeeThreshold) && !overflow { if bytes.Equal(found.Tx.IDHash[:], mt.Tx.IDHash[:]) { return txpoolcfg.NotSet } return txpoolcfg.ReplaceUnderpriced // TODO: This is the same as NotReplaced } } //Regular txn threshold checks tipThreshold := uint256.NewInt(0) tipThreshold = tipThreshold.Mul(&found.Tx.Tip, uint256.NewInt(100+priceBump)) tipThreshold.Div(tipThreshold, u256.N100) feecapThreshold := uint256.NewInt(0) feecapThreshold.Mul(&found.Tx.FeeCap, uint256.NewInt(100+priceBump)) feecapThreshold.Div(feecapThreshold, u256.N100) if mt.Tx.Tip.Cmp(tipThreshold) < 0 || mt.Tx.FeeCap.Cmp(feecapThreshold) < 0 { // Both tip and feecap need to be larger than previously to replace the transaction // In case if the transition is stuck, "poke" it to rebroadcast if mt.subPool&IsLocal != 0 && (found.currentSubPool == PendingSubPool || found.currentSubPool == BaseFeeSubPool) { announcements.Append(found.Tx.Type, found.Tx.Size, found.Tx.IDHash[:]) } if bytes.Equal(found.Tx.IDHash[:], mt.Tx.IDHash[:]) { return txpoolcfg.NotSet } return txpoolcfg.NotReplaced } switch found.currentSubPool { case PendingSubPool: p.pending.Remove(found, "add", p.logger) case BaseFeeSubPool: p.baseFee.Remove(found, "add", p.logger) case QueuedSubPool: p.queued.Remove(found, "add", p.logger) default: //already removed } p.discardLocked(found, txpoolcfg.ReplacedByHigherTip) } // Don't add blob tx to queued if it's less than current pending blob base fee if mt.Tx.Type == types.BlobTxType && mt.Tx.BlobFeeCap.LtUint64(p.pendingBlobFee.Load()) { return txpoolcfg.FeeTooLow } hashStr := string(mt.Tx.IDHash[:]) p.byHash[hashStr] = mt if replaced := p.all.replaceOrInsert(mt, p.logger); replaced != nil { if assert.Enable { panic("must never happen") } } if mt.subPool&IsLocal != 0 { p.isLocalLRU.Add(hashStr, struct{}{}) } // All transactions are first added to the queued pool and then immediately promoted from there if required p.queued.Add(mt, "addLocked", p.logger) // Remove from mined cache as we are now "resurrecting" it to a sub-pool p.deleteMinedBlobTxn(hashStr) return txpoolcfg.NotSet } // dropping transaction from all sub-structures and from db // Important: don't call it while iterating by all func (p *TxPool) discardLocked(mt *metaTx, reason txpoolcfg.DiscardReason) { hashStr := string(mt.Tx.IDHash[:]) delete(p.byHash, hashStr) p.deletedTxs = append(p.deletedTxs, mt) p.all.delete(mt, reason, p.logger) p.discardReasonsLRU.Add(hashStr, reason) } // Cache recently mined blobs in anticipation of reorg, delete finalized ones func (p *TxPool) processMinedFinalizedBlobs(coreTx kv.Tx, minedTxs []*types.TxSlot, finalizedBlock uint64) error { p.lastFinalizedBlock.Store(finalizedBlock) // Remove blobs in the finalized block and older, loop through all entries for l := len(p.minedBlobTxsByBlock); l > 0 && finalizedBlock > 0; l-- { // delete individual hashes for _, mt := range p.minedBlobTxsByBlock[finalizedBlock] { delete(p.minedBlobTxsByHash, string(mt.Tx.IDHash[:])) } // delete the map entry for this block num delete(p.minedBlobTxsByBlock, finalizedBlock) // move on to older blocks, if present finalizedBlock-- } // Add mined blobs minedBlock := p.lastSeenBlock.Load() p.minedBlobTxsByBlock[minedBlock] = make([]*metaTx, 0) for _, txn := range minedTxs { if txn.Type == types.BlobTxType { mt := &metaTx{Tx: txn, minedBlockNum: minedBlock} p.minedBlobTxsByBlock[minedBlock] = append(p.minedBlobTxsByBlock[minedBlock], mt) mt.bestIndex = len(p.minedBlobTxsByBlock[minedBlock]) - 1 p.minedBlobTxsByHash[string(txn.IDHash[:])] = mt } } return nil } // Delete individual hash entries from minedBlobTxs cache func (p *TxPool) deleteMinedBlobTxn(hash string) { mt, exists := p.minedBlobTxsByHash[hash] if !exists { return } l := len(p.minedBlobTxsByBlock[mt.minedBlockNum]) if l > 1 { p.minedBlobTxsByBlock[mt.minedBlockNum][mt.bestIndex] = p.minedBlobTxsByBlock[mt.minedBlockNum][l-1] } p.minedBlobTxsByBlock[mt.minedBlockNum] = p.minedBlobTxsByBlock[mt.minedBlockNum][:l-1] delete(p.minedBlobTxsByHash, hash) } func (p *TxPool) NonceFromAddress(addr [20]byte) (nonce uint64, inPool bool) { p.lock.Lock() defer p.lock.Unlock() senderID, found := p.senders.getID(addr) if !found { return 0, false } return p.all.nonce(senderID) } // removeMined - apply new highest block (or batch of blocks) // // 1. New best block arrives, which potentially changes the balance and the nonce of some senders. // We use senderIds data structure to find relevant senderId values, and then use senders data structure to // modify state_balance and state_nonce, potentially remove some elements (if transaction with some nonce is // included into a block), and finally, walk over the transaction records and update SubPool fields depending on // the actual presence of nonce gaps and what the balance is. func (p *TxPool) removeMined(byNonce *BySenderAndNonce, minedTxs []*types.TxSlot) error { noncesToRemove := map[uint64]uint64{} for _, txn := range minedTxs { nonce, ok := noncesToRemove[txn.SenderID] if !ok || txn.Nonce > nonce { noncesToRemove[txn.SenderID] = txn.Nonce } } var toDel []*metaTx // can't delete items while iterate them discarded := 0 pendingRemoved := 0 baseFeeRemoved := 0 queuedRemoved := 0 for senderID, nonce := range noncesToRemove { byNonce.ascend(senderID, func(mt *metaTx) bool { if mt.Tx.Nonce > nonce { if mt.Tx.Traced { p.logger.Debug("[txpool] removing mined, cmp nonces", "tx.nonce", mt.Tx.Nonce, "sender.nonce", nonce) } return false } if mt.Tx.Traced { p.logger.Info("TX TRACING: removeMined", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "senderId", mt.Tx.SenderID, "nonce", mt.Tx.Nonce, "currentSubPool", mt.currentSubPool) } toDel = append(toDel, mt) // del from sub-pool switch mt.currentSubPool { case PendingSubPool: pendingRemoved++ p.pending.Remove(mt, "remove-mined", p.logger) case BaseFeeSubPool: baseFeeRemoved++ p.baseFee.Remove(mt, "remove-mined", p.logger) case QueuedSubPool: queuedRemoved++ p.queued.Remove(mt, "remove-mined", p.logger) default: //already removed } return true }) discarded += len(toDel) for _, mt := range toDel { p.discardLocked(mt, txpoolcfg.Mined) } toDel = toDel[:0] } if discarded > 0 { p.logger.Debug("Discarded transactions", "count", discarded, "pending", pendingRemoved, "baseFee", baseFeeRemoved, "queued", queuedRemoved) } return nil } // onSenderStateChange is the function that recalculates ephemeral fields of transactions and determines // which sub pool they will need to go to. Since this depends on other transactions from the same sender by with lower // nonces, and also affect other transactions from the same sender with higher nonce, it loops through all transactions // for a given senderID func (p *TxPool) onSenderStateChange(senderID uint64, senderNonce uint64, senderBalance uint256.Int, blockGasLimit uint64, logger log.Logger) { noGapsNonce := senderNonce cumulativeRequiredBalance := uint256.NewInt(0) minFeeCap := uint256.NewInt(0).SetAllOne() minTip := uint64(math.MaxUint64) var toDel []*metaTx // can't delete items while iterate them p.all.ascend(senderID, func(mt *metaTx) bool { deleteAndContinueReasonLog := "" if senderNonce > mt.Tx.Nonce { deleteAndContinueReasonLog = "low nonce" } else if mt.Tx.Nonce != noGapsNonce && mt.Tx.Type == types.BlobTxType { // Discard nonce-gapped blob txns deleteAndContinueReasonLog = "nonce-gapped blob txn" } if deleteAndContinueReasonLog != "" { if mt.Tx.Traced { logger.Info("TX TRACING: onSenderStateChange loop iteration remove", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "senderID", senderID, "senderNonce", senderNonce, "txn.nonce", mt.Tx.Nonce, "currentSubPool", mt.currentSubPool, "reason", deleteAndContinueReasonLog) } // del from sub-pool switch mt.currentSubPool { case PendingSubPool: p.pending.Remove(mt, deleteAndContinueReasonLog, p.logger) case BaseFeeSubPool: p.baseFee.Remove(mt, deleteAndContinueReasonLog, p.logger) case QueuedSubPool: p.queued.Remove(mt, deleteAndContinueReasonLog, p.logger) default: //already removed } toDel = append(toDel, mt) return true } if minFeeCap.Gt(&mt.Tx.FeeCap) { *minFeeCap = mt.Tx.FeeCap } mt.minFeeCap = *minFeeCap if mt.Tx.Tip.IsUint64() { minTip = cmp.Min(minTip, mt.Tx.Tip.Uint64()) } mt.minTip = minTip mt.nonceDistance = 0 if mt.Tx.Nonce > senderNonce { // no uint underflow mt.nonceDistance = mt.Tx.Nonce - senderNonce } needBalance := requiredBalance(mt.Tx) // 2. Absence of nonce gaps. Set to 1 for transactions whose nonce is N, state nonce for // the sender is M, and there are transactions for all nonces between M and N from the same // sender. Set to 0 is the transaction's nonce is divided from the state nonce by one or more nonce gaps. mt.subPool &^= NoNonceGaps if noGapsNonce == mt.Tx.Nonce { mt.subPool |= NoNonceGaps noGapsNonce++ } // 3. Sufficient balance for gas. Set to 1 if the balance of sender's account in the // state is B, nonce of the sender in the state is M, nonce of the transaction is N, and the // sum of feeCap x gasLimit + transferred_value of all transactions from this sender with // nonces N+1 ... M is no more than B. Set to 0 otherwise. In other words, this bit is // set if there is currently a guarantee that the transaction and all its required prior // transactions will be able to pay for gas. mt.subPool &^= EnoughBalance mt.cumulativeBalanceDistance = math.MaxUint64 if mt.Tx.Nonce >= senderNonce { cumulativeRequiredBalance = cumulativeRequiredBalance.Add(cumulativeRequiredBalance, needBalance) // already deleted all transactions with nonce <= sender.nonce if senderBalance.Gt(cumulativeRequiredBalance) || senderBalance.Eq(cumulativeRequiredBalance) { mt.subPool |= EnoughBalance } else { if cumulativeRequiredBalance.IsUint64() && senderBalance.IsUint64() { mt.cumulativeBalanceDistance = cumulativeRequiredBalance.Uint64() - senderBalance.Uint64() } } } mt.subPool &^= NotTooMuchGas if mt.Tx.Gas < blockGasLimit { mt.subPool |= NotTooMuchGas } if mt.Tx.Traced { logger.Info("TX TRACING: onSenderStateChange loop iteration update", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "senderId", mt.Tx.SenderID, "nonce", mt.Tx.Nonce, "subPool", mt.currentSubPool) } // Some fields of mt might have changed, need to fix the invariants in the subpool best and worst queues switch mt.currentSubPool { case PendingSubPool: p.pending.Updated(mt) case BaseFeeSubPool: p.baseFee.Updated(mt) case QueuedSubPool: p.queued.Updated(mt) } return true }) for _, mt := range toDel { p.discardLocked(mt, txpoolcfg.NonceTooLow) } logger.Trace("[txpool] onSenderStateChange", "sender", senderID, "count", p.all.count(senderID), "pending", p.pending.Len(), "baseFee", p.baseFee.Len(), "queued", p.queued.Len()) } // promote reasserts invariants of the subpool and returns the list of transactions that ended up // being promoted to the pending or basefee pool, for re-broadcasting func (p *TxPool) promote(pendingBaseFee uint64, pendingBlobFee uint64, announcements *types.Announcements, logger log.Logger) { // Demote worst transactions that do not qualify for pending sub pool anymore, to other sub pools, or discard for worst := p.pending.Worst(); p.pending.Len() > 0 && (worst.subPool < BaseFeePoolBits || worst.minFeeCap.LtUint64(pendingBaseFee) || (worst.Tx.Type == types.BlobTxType && worst.Tx.BlobFeeCap.LtUint64(pendingBlobFee))); worst = p.pending.Worst() { if worst.subPool >= BaseFeePoolBits { tx := p.pending.PopWorst() announcements.Append(tx.Tx.Type, tx.Tx.Size, tx.Tx.IDHash[:]) p.baseFee.Add(tx, "demote-pending", logger) } else { p.queued.Add(p.pending.PopWorst(), "demote-pending", logger) } } // Promote best transactions from base fee pool to pending pool while they qualify for best := p.baseFee.Best(); p.baseFee.Len() > 0 && best.subPool >= BaseFeePoolBits && best.minFeeCap.CmpUint64(pendingBaseFee) >= 0 && (best.Tx.Type != types.BlobTxType || best.Tx.BlobFeeCap.CmpUint64(pendingBlobFee) >= 0); best = p.baseFee.Best() { tx := p.baseFee.PopBest() announcements.Append(tx.Tx.Type, tx.Tx.Size, tx.Tx.IDHash[:]) p.pending.Add(tx, logger) } // Demote worst transactions that do not qualify for base fee pool anymore, to queued sub pool, or discard for worst := p.baseFee.Worst(); p.baseFee.Len() > 0 && worst.subPool < BaseFeePoolBits; worst = p.baseFee.Worst() { p.queued.Add(p.baseFee.PopWorst(), "demote-base", logger) } // Promote best transactions from the queued pool to either pending or base fee pool, while they qualify for best := p.queued.Best(); p.queued.Len() > 0 && best.subPool >= BaseFeePoolBits; best = p.queued.Best() { if best.minFeeCap.Cmp(uint256.NewInt(pendingBaseFee)) >= 0 { tx := p.queued.PopBest() announcements.Append(tx.Tx.Type, tx.Tx.Size, tx.Tx.IDHash[:]) p.pending.Add(tx, logger) } else { p.baseFee.Add(p.queued.PopBest(), "promote-queued", logger) } } // Discard worst transactions from the queued sub pool if they do not qualify // // Discard worst transactions from pending pool until it is within capacity limit for p.pending.Len() > p.pending.limit { p.discardLocked(p.pending.PopWorst(), txpoolcfg.PendingPoolOverflow) } // Discard worst transactions from pending sub pool until it is within capacity limits for p.baseFee.Len() > p.baseFee.limit { p.discardLocked(p.baseFee.PopWorst(), txpoolcfg.BaseFeePoolOverflow) } // Discard worst transactions from the queued sub pool until it is within its capacity limits for _ = p.queued.Worst(); p.queued.Len() > p.queued.limit; _ = p.queued.Worst() { p.discardLocked(p.queued.PopWorst(), txpoolcfg.QueuedPoolOverflow) } } // txMaxBroadcastSize is the max size of a transaction that will be broadcasted. // All transactions with a higher size will be announced and need to be fetched // by the peer. const txMaxBroadcastSize = 4 * 1024 // MainLoop - does: // send pending byHash to p2p: // - new byHash // - all pooled byHash to recently connected peers // - all local pooled byHash to random peers periodically // // promote/demote transactions // reorgs func MainLoop(ctx context.Context, db kv.RwDB, p *TxPool, newTxs chan types.Announcements, send *Send, newSlotsStreams *NewSlotsStreams, notifyMiningAboutNewSlots func()) { syncToNewPeersEvery := time.NewTicker(p.cfg.SyncToNewPeersEvery) defer syncToNewPeersEvery.Stop() processRemoteTxsEvery := time.NewTicker(p.cfg.ProcessRemoteTxsEvery) defer processRemoteTxsEvery.Stop() commitEvery := time.NewTicker(p.cfg.CommitEvery) defer commitEvery.Stop() logEvery := time.NewTicker(p.cfg.LogEvery) defer logEvery.Stop() err := p.Start(ctx, db) if err != nil { p.logger.Error("[txpool] Failed to start", "err", err) return } for { select { case <-ctx.Done(): _, _ = p.flush(ctx, db) return case <-logEvery.C: p.logStats() case <-processRemoteTxsEvery.C: if !p.Started() { continue } if err := p.processRemoteTxs(ctx); err != nil { if grpcutil.IsRetryLater(err) || grpcutil.IsEndOfStream(err) { time.Sleep(3 * time.Second) continue } p.logger.Error("[txpool] process batch remote txs", "err", err) } case <-commitEvery.C: if db != nil && p.Started() { t := time.Now() written, err := p.flush(ctx, db) if err != nil { p.logger.Error("[txpool] flush is local history", "err", err) continue } writeToDBBytesCounter.SetUint64(written) p.logger.Debug("[txpool] Commit", "written_kb", written/1024, "in", time.Since(t)) } case announcements := <-newTxs: go func() { for i := 0; i < 16; i++ { // drain more events from channel, then merge and dedup them select { case a := <-newTxs: announcements.AppendOther(a) continue default: } break } if announcements.Len() == 0 { return } defer propagateNewTxsTimer.ObserveDuration(time.Now()) announcements = announcements.DedupCopy() notifyMiningAboutNewSlots() if p.cfg.NoGossip { // drain newTxs for emptying newTx channel // newTx channel will be filled only with local transactions // early return to avoid outbound transaction propagation log.Debug("[txpool] tx gossip disabled", "state", "drain new transactions") return } var localTxTypes []byte var localTxSizes []uint32 var localTxHashes types.Hashes var localTxRlps [][]byte var remoteTxTypes []byte var remoteTxSizes []uint32 var remoteTxHashes types.Hashes var remoteTxRlps [][]byte var broadcastHashes types.Hashes slotsRlp := make([][]byte, 0, announcements.Len()) if err := db.View(ctx, func(tx kv.Tx) error { for i := 0; i < announcements.Len(); i++ { t, size, hash := announcements.At(i) slotRlp, err := p.GetRlp(tx, hash) if err != nil { return err } if len(slotRlp) == 0 { continue } // Empty rlp can happen if a transaction we want to broadcast has just been mined, for example slotsRlp = append(slotsRlp, slotRlp) if p.IsLocal(hash) { localTxTypes = append(localTxTypes, t) localTxSizes = append(localTxSizes, size) localTxHashes = append(localTxHashes, hash...) // "Nodes MUST NOT automatically broadcast blob transactions to their peers" - EIP-4844 if t != types.BlobTxType { localTxRlps = append(localTxRlps, slotRlp) broadcastHashes = append(broadcastHashes, hash...) } } else { remoteTxTypes = append(remoteTxTypes, t) remoteTxSizes = append(remoteTxSizes, size) remoteTxHashes = append(remoteTxHashes, hash...) // "Nodes MUST NOT automatically broadcast blob transactions to their peers" - EIP-4844 if t != types.BlobTxType && len(slotRlp) < txMaxBroadcastSize { remoteTxRlps = append(remoteTxRlps, slotRlp) } } } return nil }); err != nil { p.logger.Error("[txpool] collect info to propagate", "err", err) return } if newSlotsStreams != nil { // TODO(eip-4844) What is this for? Is it OK to broadcast blob transactions? newSlotsStreams.Broadcast(&proto_txpool.OnAddReply{RplTxs: slotsRlp}, p.logger) } // broadcast local transactions const localTxsBroadcastMaxPeers uint64 = 10 txSentTo := send.BroadcastPooledTxs(localTxRlps, localTxsBroadcastMaxPeers) for i, peer := range txSentTo { p.logger.Trace("Local tx broadcast", "txHash", hex.EncodeToString(broadcastHashes.At(i)), "to peer", peer) } hashSentTo := send.AnnouncePooledTxs(localTxTypes, localTxSizes, localTxHashes, localTxsBroadcastMaxPeers*2) for i := 0; i < localTxHashes.Len(); i++ { hash := localTxHashes.At(i) p.logger.Trace("Local tx announced", "txHash", hex.EncodeToString(hash), "to peer", hashSentTo[i], "baseFee", p.pendingBaseFee.Load()) } // broadcast remote transactions const remoteTxsBroadcastMaxPeers uint64 = 3 send.BroadcastPooledTxs(remoteTxRlps, remoteTxsBroadcastMaxPeers) send.AnnouncePooledTxs(remoteTxTypes, remoteTxSizes, remoteTxHashes, remoteTxsBroadcastMaxPeers*2) }() case <-syncToNewPeersEvery.C: // new peer newPeers := p.recentlyConnectedPeers.GetAndClean() if len(newPeers) == 0 { continue } if p.cfg.NoGossip { // avoid transaction gossiping for new peers log.Debug("[txpool] tx gossip disabled", "state", "sync new peers") continue } t := time.Now() var hashes types.Hashes var types []byte var sizes []uint32 types, sizes, hashes = p.AppendAllAnnouncements(types, sizes, hashes[:0]) go send.PropagatePooledTxsToPeersList(newPeers, types, sizes, hashes) propagateToNewPeerTimer.ObserveDuration(t) } } } func (p *TxPool) flushNoFsync(ctx context.Context, db kv.RwDB) (written uint64, err error) { p.lock.Lock() defer p.lock.Unlock() //it's important that write db tx is done inside lock, to make last writes visible for all read operations if err := db.UpdateNosync(ctx, func(tx kv.RwTx) error { err = p.flushLocked(tx) if err != nil { return err } written, _, err = tx.(*mdbx.MdbxTx).SpaceDirty() if err != nil { return err } return nil }); err != nil { return 0, err } return written, nil } func (p *TxPool) flush(ctx context.Context, db kv.RwDB) (written uint64, err error) { defer writeToDBTimer.ObserveDuration(time.Now()) // 1. get global lock on txpool and flush it to db, without fsync (to release lock asap) // 2. then fsync db without txpool lock written, err = p.flushNoFsync(ctx, db) if err != nil { return 0, err } // fsync if err := db.Update(ctx, func(tx kv.RwTx) error { return nil }); err != nil { return 0, err } return written, nil } func (p *TxPool) flushLocked(tx kv.RwTx) (err error) { for i, mt := range p.deletedTxs { id := mt.Tx.SenderID idHash := mt.Tx.IDHash[:] if !p.all.hasTxs(id) { addr, ok := p.senders.senderID2Addr[id] if ok { delete(p.senders.senderID2Addr, id) delete(p.senders.senderIDs, addr) } } //fmt.Printf("del:%d,%d,%d\n", mt.Tx.senderID, mt.Tx.nonce, mt.Tx.tip) has, err := tx.Has(kv.PoolTransaction, idHash) if err != nil { return err } if has { if err := tx.Delete(kv.PoolTransaction, idHash); err != nil { return err } } p.deletedTxs[i] = nil // for gc } txHashes := p.isLocalLRU.Keys() encID := make([]byte, 8) if err := tx.ClearBucket(kv.RecentLocalTransaction); err != nil { return err } for i, txHash := range txHashes { binary.BigEndian.PutUint64(encID, uint64(i)) if err := tx.Append(kv.RecentLocalTransaction, encID, []byte(txHash)); err != nil { return err } } v := make([]byte, 0, 1024) for txHash, metaTx := range p.byHash { if metaTx.Tx.Rlp == nil { continue } v = common.EnsureEnoughSize(v, 20+len(metaTx.Tx.Rlp)) addr, ok := p.senders.senderID2Addr[metaTx.Tx.SenderID] if !ok { p.logger.Warn("[txpool] flush: sender address not found by ID", "senderID", metaTx.Tx.SenderID) continue } copy(v[:20], addr.Bytes()) copy(v[20:], metaTx.Tx.Rlp) has, err := tx.Has(kv.PoolTransaction, []byte(txHash)) if err != nil { return err } if !has { if err := tx.Put(kv.PoolTransaction, []byte(txHash), v); err != nil { return err } } metaTx.Tx.Rlp = nil } binary.BigEndian.PutUint64(encID, p.pendingBaseFee.Load()) if err := tx.Put(kv.PoolInfo, PoolPendingBaseFeeKey, encID); err != nil { return err } binary.BigEndian.PutUint64(encID, p.pendingBlobFee.Load()) if err := tx.Put(kv.PoolInfo, PoolPendingBlobFeeKey, encID); err != nil { return err } if err := PutLastSeenBlock(tx, p.lastSeenBlock.Load(), encID); err != nil { return err } // clean - in-memory data structure as later as possible - because if during this Tx will happen error, // DB will stay consistent but some in-memory structures may be already cleaned, and retry will not work // failed write transaction must not create side-effects p.deletedTxs = p.deletedTxs[:0] return nil } func (p *TxPool) fromDB(ctx context.Context, tx kv.Tx, coreTx kv.Tx) error { if p.lastSeenBlock.Load() == 0 { lastSeenBlock, err := LastSeenBlock(tx) if err != nil { return err } p.lastSeenBlock.Store(lastSeenBlock) } // this is neccessary as otherwise best - which waits for sync events // may wait for ever if blocks have been process before the txpool // starts with an empty db lastSeenProgress, err := getExecutionProgress(coreTx) if err != nil { return err } if p.lastSeenBlock.Load() < lastSeenProgress { // TODO we need to process the blocks since the // last seen to make sure that the tx pool is in // sync with the processed blocks p.lastSeenBlock.Store(lastSeenProgress) } cacheView, err := p._stateCache.View(ctx, coreTx) if err != nil { return err } it, err := tx.Range(kv.RecentLocalTransaction, nil, nil) if err != nil { return err } for it.HasNext() { _, v, err := it.Next() if err != nil { return err } p.isLocalLRU.Add(string(v), struct{}{}) } txs := types.TxSlots{} parseCtx := types.NewTxParseContext(p.chainID, false) parseCtx.WithSender(false) i := 0 it, err = tx.Range(kv.PoolTransaction, nil, nil) if err != nil { return err } for it.HasNext() { k, v, err := it.Next() if err != nil { return err } addr, txRlp := *(*[20]byte)(v[:20]), v[20:] txn := &types.TxSlot{} // TODO(eip-4844) ensure wrappedWithBlobs when transactions are saved to the DB _, err = parseCtx.ParseTransaction(txRlp, 0, txn, nil, false /* hasEnvelope */, true /*wrappedWithBlobs*/, nil) if err != nil { err = fmt.Errorf("err: %w, rlp: %x", err, txRlp) p.logger.Warn("[txpool] fromDB: parseTransaction", "err", err) continue } txn.Rlp = nil // means that we don't need store it in db anymore txn.SenderID, txn.Traced = p.senders.getOrCreateID(addr, p.logger) binary.BigEndian.Uint64(v) // TODO - unnecessary line, remove isLocalTx := p.isLocalLRU.Contains(string(k)) if reason := p.validateTx(txn, isLocalTx, cacheView); reason != txpoolcfg.NotSet && reason != txpoolcfg.Success { return nil // TODO: Clarify - if one of the txs has the wrong reason, no pooled txs! } txs.Resize(uint(i + 1)) txs.Txs[i] = txn txs.IsLocal[i] = isLocalTx copy(txs.Senders.At(i), addr[:]) i++ } var pendingBaseFee, pendingBlobFee, minBlobGasPrice, blockGasLimit uint64 if p.feeCalculator != nil { if chainConfig, _ := ChainConfig(tx); chainConfig != nil { pendingBaseFee, pendingBlobFee, minBlobGasPrice, blockGasLimit, err = p.feeCalculator.CurrentFees(chainConfig, coreTx) if err != nil { return err } } } if pendingBaseFee == 0 { v, err := tx.GetOne(kv.PoolInfo, PoolPendingBaseFeeKey) if err != nil { return err } if len(v) > 0 { pendingBaseFee = binary.BigEndian.Uint64(v) } } if pendingBlobFee == 0 { v, err := tx.GetOne(kv.PoolInfo, PoolPendingBlobFeeKey) if err != nil { return err } if len(v) > 0 { pendingBlobFee = binary.BigEndian.Uint64(v) } } if pendingBlobFee == 0 { pendingBlobFee = minBlobGasPrice } if blockGasLimit == 0 { blockGasLimit = DefaultBlockGasLimit } err = p.senders.registerNewSenders(&txs, p.logger) if err != nil { return err } if _, _, err := p.addTxs(p.lastSeenBlock.Load(), cacheView, p.senders, txs, pendingBaseFee, pendingBlobFee, blockGasLimit, false, p.logger); err != nil { return err } p.pendingBaseFee.Store(pendingBaseFee) p.pendingBlobFee.Store(pendingBlobFee) p.blockGasLimit.Store(blockGasLimit) return nil } func getExecutionProgress(db kv.Getter) (uint64, error) { data, err := db.GetOne(kv.SyncStageProgress, []byte("Execution")) if err != nil { return 0, err } if len(data) == 0 { return 0, nil } if len(data) < 8 { return 0, fmt.Errorf("value must be at least 8 bytes, got %d", len(data)) } return binary.BigEndian.Uint64(data[:8]), nil } func LastSeenBlock(tx kv.Getter) (uint64, error) { v, err := tx.GetOne(kv.PoolInfo, PoolLastSeenBlockKey) if err != nil { return 0, err } if len(v) == 0 { return 0, nil } return binary.BigEndian.Uint64(v), nil } func PutLastSeenBlock(tx kv.Putter, n uint64, buf []byte) error { buf = common.EnsureEnoughSize(buf, 8) binary.BigEndian.PutUint64(buf, n) err := tx.Put(kv.PoolInfo, PoolLastSeenBlockKey, buf) if err != nil { return err } return nil } func ChainConfig(tx kv.Getter) (*chain.Config, error) { v, err := tx.GetOne(kv.PoolInfo, PoolChainConfigKey) if err != nil { return nil, err } if len(v) == 0 { return nil, nil } var config chain.Config if err := json.Unmarshal(v, &config); err != nil { return nil, fmt.Errorf("invalid chain config JSON in pool db: %w", err) } return &config, nil } func PutChainConfig(tx kv.Putter, cc *chain.Config, buf []byte) error { wr := bytes.NewBuffer(buf) if err := json.NewEncoder(wr).Encode(cc); err != nil { return fmt.Errorf("invalid chain config JSON in pool db: %w", err) } if err := tx.Put(kv.PoolInfo, PoolChainConfigKey, wr.Bytes()); err != nil { return err } return nil } // nolint func (p *TxPool) printDebug(prefix string) { fmt.Printf("%s.pool.byHash\n", prefix) for _, j := range p.byHash { fmt.Printf("\tsenderID=%d, nonce=%d, tip=%d\n", j.Tx.SenderID, j.Tx.Nonce, j.Tx.Tip) } fmt.Printf("%s.pool.queues.len: %d,%d,%d\n", prefix, p.pending.Len(), p.baseFee.Len(), p.queued.Len()) for _, mt := range p.pending.best.ms { mt.Tx.PrintDebug(fmt.Sprintf("%s.pending: %b,%d,%d,%d", prefix, mt.subPool, mt.Tx.SenderID, mt.Tx.Nonce, mt.Tx.Tip)) } for _, mt := range p.baseFee.best.ms { mt.Tx.PrintDebug(fmt.Sprintf("%s.baseFee : %b,%d,%d,%d", prefix, mt.subPool, mt.Tx.SenderID, mt.Tx.Nonce, mt.Tx.Tip)) } for _, mt := range p.queued.best.ms { mt.Tx.PrintDebug(fmt.Sprintf("%s.queued : %b,%d,%d,%d", prefix, mt.subPool, mt.Tx.SenderID, mt.Tx.Nonce, mt.Tx.Tip)) } } func (p *TxPool) logStats() { if !p.Started() { //p.logger.Info("[txpool] Not started yet, waiting for new blocks...") return } p.lock.Lock() defer p.lock.Unlock() var m runtime.MemStats dbg.ReadMemStats(&m) ctx := []interface{}{ //"block", p.lastSeenBlock.Load(), "pending", p.pending.Len(), "baseFee", p.baseFee.Len(), "queued", p.queued.Len(), } cacheKeys := p._stateCache.Len() if cacheKeys > 0 { ctx = append(ctx, "cache_keys", cacheKeys) } ctx = append(ctx, "alloc", common.ByteCount(m.Alloc), "sys", common.ByteCount(m.Sys)) p.logger.Info("[txpool] stat", ctx...) pendingSubCounter.SetInt(p.pending.Len()) basefeeSubCounter.SetInt(p.baseFee.Len()) queuedSubCounter.SetInt(p.queued.Len()) } // Deprecated need switch to streaming-like func (p *TxPool) deprecatedForEach(_ context.Context, f func(rlp []byte, sender common.Address, t SubPoolType), tx kv.Tx) { p.lock.Lock() defer p.lock.Unlock() p.all.ascendAll(func(mt *metaTx) bool { slot := mt.Tx slotRlp := slot.Rlp if slot.Rlp == nil { v, err := tx.GetOne(kv.PoolTransaction, slot.IDHash[:]) if err != nil { p.logger.Warn("[txpool] foreach: get tx from db", "err", err) return true } if v == nil { p.logger.Warn("[txpool] foreach: tx not found in db") return true } slotRlp = v[20:] } if sender, found := p.senders.senderID2Addr[slot.SenderID]; found { f(slotRlp, sender, mt.currentSubPool) } return true }) } var PoolChainConfigKey = []byte("chain_config") var PoolLastSeenBlockKey = []byte("last_seen_block") var PoolPendingBaseFeeKey = []byte("pending_base_fee") var PoolPendingBlobFeeKey = []byte("pending_blob_fee") // recentlyConnectedPeers does buffer IDs of recently connected good peers // then sync of pooled Transaction can happen to all of then at once // DoS protection and performance saving // it doesn't track if peer disconnected, it's fine type recentlyConnectedPeers struct { peers []types.PeerID lock sync.Mutex } func (l *recentlyConnectedPeers) AddPeer(p types.PeerID) { l.lock.Lock() defer l.lock.Unlock() l.peers = append(l.peers, p) } func (l *recentlyConnectedPeers) GetAndClean() []types.PeerID { l.lock.Lock() defer l.lock.Unlock() peers := l.peers l.peers = nil return peers } // nolint func (sc *sendersBatch) printDebug(prefix string) { fmt.Printf("%s.sendersBatch.sender\n", prefix) //for i, j := range sc.senderInfo { // fmt.Printf("\tid=%d,nonce=%d,balance=%d\n", i, j.nonce, j.balance.Uint64()) //} } // sendersBatch stores in-memory senders-related objects - which are different from DB (updated/dirty) // flushing to db periodically. it doesn't play as read-cache (because db is small and memory-mapped - doesn't need cache) // non thread-safe type sendersBatch struct { senderIDs map[common.Address]uint64 senderID2Addr map[uint64]common.Address tracedSenders map[common.Address]struct{} senderID uint64 } func newSendersCache(tracedSenders map[common.Address]struct{}) *sendersBatch { return &sendersBatch{senderIDs: map[common.Address]uint64{}, senderID2Addr: map[uint64]common.Address{}, tracedSenders: tracedSenders} } func (sc *sendersBatch) getID(addr common.Address) (uint64, bool) { id, ok := sc.senderIDs[addr] return id, ok } func (sc *sendersBatch) getOrCreateID(addr common.Address, logger log.Logger) (uint64, bool) { _, traced := sc.tracedSenders[addr] if !traced { traced = TraceAll } id, ok := sc.senderIDs[addr] if !ok { sc.senderID++ id = sc.senderID sc.senderIDs[addr] = id sc.senderID2Addr[id] = addr if traced { logger.Info(fmt.Sprintf("TX TRACING: allocated senderID %d to sender %x", id, addr)) } } return id, traced } func (sc *sendersBatch) info(cacheView kvcache.CacheView, id uint64) (nonce uint64, balance uint256.Int, err error) { addr, ok := sc.senderID2Addr[id] if !ok { panic("must not happen") } encoded, err := cacheView.Get(addr.Bytes()) if err != nil { return 0, emptySender.balance, err } if len(encoded) == 0 { return emptySender.nonce, emptySender.balance, nil } nonce, balance, err = types.DecodeSender(encoded) if err != nil { return 0, emptySender.balance, err } return nonce, balance, nil } func (sc *sendersBatch) registerNewSenders(newTxs *types.TxSlots, logger log.Logger) (err error) { for i, txn := range newTxs.Txs { txn.SenderID, txn.Traced = sc.getOrCreateID(newTxs.Senders.AddressAt(i), logger) } return nil } func (sc *sendersBatch) onNewBlock(stateChanges *remote.StateChangeBatch, unwindTxs, minedTxs types.TxSlots, logger log.Logger) error { for _, diff := range stateChanges.ChangeBatch { for _, change := range diff.Changes { // merge state changes addrB := gointerfaces.ConvertH160toAddress(change.Address) sc.getOrCreateID(addrB, logger) } for i, txn := range unwindTxs.Txs { txn.SenderID, txn.Traced = sc.getOrCreateID(unwindTxs.Senders.AddressAt(i), logger) } for i, txn := range minedTxs.Txs { txn.SenderID, txn.Traced = sc.getOrCreateID(minedTxs.Senders.AddressAt(i), logger) } } return nil } // BySenderAndNonce - designed to perform most expensive operation in TxPool: // "recalculate all ephemeral fields of all transactions" by algo // - for all senders - iterate over all transactions in nonce growing order // // Performances decisions: // - All senders stored inside 1 large BTree - because iterate over 1 BTree is faster than over map[senderId]BTree // - sortByNonce used as non-pointer wrapper - because iterate over BTree of pointers is 2x slower type BySenderAndNonce struct { tree *btree.BTreeG[*metaTx] search *metaTx senderIDTxnCount map[uint64]int // count of sender's txns in the pool - may differ from nonce senderIDBlobCount map[uint64]uint64 // count of sender's total number of blobs in the pool } func (b *BySenderAndNonce) nonce(senderID uint64) (nonce uint64, ok bool) { s := b.search s.Tx.SenderID = senderID s.Tx.Nonce = math.MaxUint64 b.tree.DescendLessOrEqual(s, func(mt *metaTx) bool { if mt.Tx.SenderID == senderID { nonce = mt.Tx.Nonce ok = true } return false }) return nonce, ok } func (b *BySenderAndNonce) ascendAll(f func(*metaTx) bool) { b.tree.Ascend(func(mt *metaTx) bool { return f(mt) }) } func (b *BySenderAndNonce) ascend(senderID uint64, f func(*metaTx) bool) { s := b.search s.Tx.SenderID = senderID s.Tx.Nonce = 0 b.tree.AscendGreaterOrEqual(s, func(mt *metaTx) bool { if mt.Tx.SenderID != senderID { return false } return f(mt) }) } func (b *BySenderAndNonce) descend(senderID uint64, f func(*metaTx) bool) { s := b.search s.Tx.SenderID = senderID s.Tx.Nonce = math.MaxUint64 b.tree.DescendLessOrEqual(s, func(mt *metaTx) bool { if mt.Tx.SenderID != senderID { return false } return f(mt) }) } func (b *BySenderAndNonce) count(senderID uint64) int { return b.senderIDTxnCount[senderID] } func (b *BySenderAndNonce) blobCount(senderID uint64) uint64 { return b.senderIDBlobCount[senderID] } func (b *BySenderAndNonce) hasTxs(senderID uint64) bool { has := false b.ascend(senderID, func(*metaTx) bool { has = true return false }) return has } func (b *BySenderAndNonce) get(senderID, txNonce uint64) *metaTx { s := b.search s.Tx.SenderID = senderID s.Tx.Nonce = txNonce if found, ok := b.tree.Get(s); ok { return found } return nil } // nolint func (b *BySenderAndNonce) has(mt *metaTx) bool { return b.tree.Has(mt) } func (b *BySenderAndNonce) delete(mt *metaTx, reason txpoolcfg.DiscardReason, logger log.Logger) { if _, ok := b.tree.Delete(mt); ok { if mt.Tx.Traced { logger.Info("TX TRACING: Deleted tx by nonce", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "sender", mt.Tx.SenderID, "nonce", mt.Tx.Nonce, "reason", reason) } senderID := mt.Tx.SenderID count := b.senderIDTxnCount[senderID] if count > 1 { b.senderIDTxnCount[senderID] = count - 1 } else { delete(b.senderIDTxnCount, senderID) } if mt.Tx.Type == types.BlobTxType && mt.Tx.Blobs != nil { accBlobCount := b.senderIDBlobCount[senderID] txnBlobCount := len(mt.Tx.Blobs) if txnBlobCount > 1 { b.senderIDBlobCount[senderID] = accBlobCount - uint64(txnBlobCount) } else { delete(b.senderIDBlobCount, senderID) } } } } func (b *BySenderAndNonce) replaceOrInsert(mt *metaTx, logger log.Logger) *metaTx { it, ok := b.tree.ReplaceOrInsert(mt) if ok { if mt.Tx.Traced { logger.Info("TX TRACING: Replaced tx by nonce", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "sender", mt.Tx.SenderID, "nonce", mt.Tx.Nonce) } return it } if mt.Tx.Traced { logger.Info("TX TRACING: Inserted tx by nonce", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "sender", mt.Tx.SenderID, "nonce", mt.Tx.Nonce) } b.senderIDTxnCount[mt.Tx.SenderID]++ if mt.Tx.Type == types.BlobTxType && mt.Tx.Blobs != nil { b.senderIDBlobCount[mt.Tx.SenderID] += uint64(len(mt.Tx.Blobs)) } return nil } // PendingPool - is different from other pools - it's best is Slice instead of Heap // It's more expensive to maintain "slice sort" invariant, but it allow do cheap copy of // pending.best slice for mining (because we consider txs and metaTx are immutable) type PendingPool struct { best *bestSlice worst *WorstQueue limit int t SubPoolType } func NewPendingSubPool(t SubPoolType, limit int) *PendingPool { return &PendingPool{limit: limit, t: t, best: &bestSlice{ms: []*metaTx{}}, worst: &WorstQueue{ms: []*metaTx{}}} } // bestSlice - is similar to best queue, but uses a linear structure with O(n log n) sort complexity and // it maintains element.bestIndex field type bestSlice struct { ms []*metaTx pendingBaseFee uint64 } func (s *bestSlice) Len() int { return len(s.ms) } func (s *bestSlice) Swap(i, j int) { s.ms[i], s.ms[j] = s.ms[j], s.ms[i] s.ms[i].bestIndex, s.ms[j].bestIndex = i, j } func (s *bestSlice) Less(i, j int) bool { return s.ms[i].better(s.ms[j], *uint256.NewInt(s.pendingBaseFee)) } func (s *bestSlice) UnsafeRemove(i *metaTx) { s.Swap(i.bestIndex, len(s.ms)-1) s.ms[len(s.ms)-1].bestIndex = -1 s.ms[len(s.ms)-1] = nil s.ms = s.ms[:len(s.ms)-1] } func (s *bestSlice) UnsafeAdd(i *metaTx) { i.bestIndex = len(s.ms) s.ms = append(s.ms, i) } func (p *PendingPool) EnforceWorstInvariants() { heap.Init(p.worst) } func (p *PendingPool) EnforceBestInvariants() { sort.Sort(p.best) } func (p *PendingPool) Best() *metaTx { //nolint if len(p.best.ms) == 0 { return nil } return p.best.ms[0] } func (p *PendingPool) Worst() *metaTx { //nolint if len(p.worst.ms) == 0 { return nil } return (p.worst.ms)[0] } func (p *PendingPool) PopWorst() *metaTx { //nolint i := heap.Pop(p.worst).(*metaTx) if i.bestIndex >= 0 { p.best.UnsafeRemove(i) } return i } func (p *PendingPool) Updated(mt *metaTx) { heap.Fix(p.worst, mt.worstIndex) } func (p *PendingPool) Len() int { return len(p.best.ms) } func (p *PendingPool) Remove(i *metaTx, reason string, logger log.Logger) { if i.Tx.Traced { logger.Info(fmt.Sprintf("TX TRACING: removed from subpool %s", p.t), "idHash", fmt.Sprintf("%x", i.Tx.IDHash), "sender", i.Tx.SenderID, "nonce", i.Tx.Nonce, "reason", reason) } if i.worstIndex >= 0 { heap.Remove(p.worst, i.worstIndex) } if i.bestIndex >= 0 { p.best.UnsafeRemove(i) } i.currentSubPool = 0 } func (p *PendingPool) Add(i *metaTx, logger log.Logger) { if i.Tx.Traced { logger.Info(fmt.Sprintf("TX TRACING: added to subpool %s, IdHash=%x, sender=%d, nonce=%d", p.t, i.Tx.IDHash, i.Tx.SenderID, i.Tx.Nonce)) } i.currentSubPool = p.t heap.Push(p.worst, i) p.best.UnsafeAdd(i) } func (p *PendingPool) DebugPrint(prefix string) { for i, it := range p.best.ms { fmt.Printf("%s.best: %d, %d, %d,%d\n", prefix, i, it.subPool, it.bestIndex, it.Tx.Nonce) } for i, it := range p.worst.ms { fmt.Printf("%s.worst: %d, %d, %d,%d\n", prefix, i, it.subPool, it.worstIndex, it.Tx.Nonce) } } type SubPool struct { best *BestQueue worst *WorstQueue limit int t SubPoolType } func NewSubPool(t SubPoolType, limit int) *SubPool { return &SubPool{limit: limit, t: t, best: &BestQueue{}, worst: &WorstQueue{}} } func (p *SubPool) EnforceInvariants() { heap.Init(p.worst) heap.Init(p.best) } func (p *SubPool) Best() *metaTx { //nolint if len(p.best.ms) == 0 { return nil } return p.best.ms[0] } func (p *SubPool) Worst() *metaTx { //nolint if len(p.worst.ms) == 0 { return nil } return p.worst.ms[0] } func (p *SubPool) PopBest() *metaTx { //nolint i := heap.Pop(p.best).(*metaTx) heap.Remove(p.worst, i.worstIndex) return i } func (p *SubPool) PopWorst() *metaTx { //nolint i := heap.Pop(p.worst).(*metaTx) heap.Remove(p.best, i.bestIndex) return i } func (p *SubPool) Len() int { return p.best.Len() } func (p *SubPool) Add(i *metaTx, reason string, logger log.Logger) { if i.Tx.Traced { logger.Info(fmt.Sprintf("TX TRACING: added to subpool %s", p.t), "idHash", fmt.Sprintf("%x", i.Tx.IDHash), "sender", i.Tx.SenderID, "nonce", i.Tx.Nonce, "reason", reason) } i.currentSubPool = p.t heap.Push(p.best, i) heap.Push(p.worst, i) } func (p *SubPool) Remove(i *metaTx, reason string, logger log.Logger) { if i.Tx.Traced { logger.Info(fmt.Sprintf("TX TRACING: removed from subpool %s", p.t), "idHash", fmt.Sprintf("%x", i.Tx.IDHash), "sender", i.Tx.SenderID, "nonce", i.Tx.Nonce, "reason", reason) } heap.Remove(p.best, i.bestIndex) heap.Remove(p.worst, i.worstIndex) i.currentSubPool = 0 } func (p *SubPool) Updated(i *metaTx) { heap.Fix(p.best, i.bestIndex) heap.Fix(p.worst, i.worstIndex) } func (p *SubPool) DebugPrint(prefix string) { for i, it := range p.best.ms { fmt.Printf("%s.best: %d, %d, %d\n", prefix, i, it.subPool, it.bestIndex) } for i, it := range p.worst.ms { fmt.Printf("%s.worst: %d, %d, %d\n", prefix, i, it.subPool, it.worstIndex) } } type BestQueue struct { ms []*metaTx pendingBastFee uint64 } // Returns true if the txn "mt" is better than the parameter txn "than" // it first compares the subpool markers of the two meta txns, then, // (since they have the same subpool marker, and thus same pool) // depending on the pool - pending (P), basefee (B), queued (Q) - // it compares the effective tip (for P), nonceDistance (for both P,Q) // minFeeCap (for B), and cumulative balance distance (for P, Q) func (mt *metaTx) better(than *metaTx, pendingBaseFee uint256.Int) bool { subPool := mt.subPool thanSubPool := than.subPool if mt.minFeeCap.Cmp(&pendingBaseFee) >= 0 { subPool |= EnoughFeeCapBlock } if than.minFeeCap.Cmp(&pendingBaseFee) >= 0 { thanSubPool |= EnoughFeeCapBlock } if subPool != thanSubPool { return subPool > thanSubPool } switch mt.currentSubPool { case PendingSubPool: var effectiveTip, thanEffectiveTip uint256.Int if mt.minFeeCap.Cmp(&pendingBaseFee) >= 0 { difference := uint256.NewInt(0) difference.Sub(&mt.minFeeCap, &pendingBaseFee) if difference.Cmp(uint256.NewInt(mt.minTip)) <= 0 { effectiveTip = *difference } else { effectiveTip = *uint256.NewInt(mt.minTip) } } if than.minFeeCap.Cmp(&pendingBaseFee) >= 0 { difference := uint256.NewInt(0) difference.Sub(&than.minFeeCap, &pendingBaseFee) if difference.Cmp(uint256.NewInt(than.minTip)) <= 0 { thanEffectiveTip = *difference } else { thanEffectiveTip = *uint256.NewInt(than.minTip) } } if effectiveTip.Cmp(&thanEffectiveTip) != 0 { return effectiveTip.Cmp(&thanEffectiveTip) > 0 } // Compare nonce and cumulative balance. Just as a side note, it doesn't // matter if they're from same sender or not because we're comparing // nonce distance of the sender from state's nonce and not the actual // value of nonce. if mt.nonceDistance != than.nonceDistance { return mt.nonceDistance < than.nonceDistance } if mt.cumulativeBalanceDistance != than.cumulativeBalanceDistance { return mt.cumulativeBalanceDistance < than.cumulativeBalanceDistance } case BaseFeeSubPool: if mt.minFeeCap.Cmp(&than.minFeeCap) != 0 { return mt.minFeeCap.Cmp(&than.minFeeCap) > 0 } case QueuedSubPool: if mt.nonceDistance != than.nonceDistance { return mt.nonceDistance < than.nonceDistance } if mt.cumulativeBalanceDistance != than.cumulativeBalanceDistance { return mt.cumulativeBalanceDistance < than.cumulativeBalanceDistance } } return mt.timestamp < than.timestamp } func (mt *metaTx) worse(than *metaTx, pendingBaseFee uint256.Int) bool { subPool := mt.subPool thanSubPool := than.subPool if mt.minFeeCap.Cmp(&pendingBaseFee) >= 0 { subPool |= EnoughFeeCapBlock } if than.minFeeCap.Cmp(&pendingBaseFee) >= 0 { thanSubPool |= EnoughFeeCapBlock } if subPool != thanSubPool { return subPool < thanSubPool } switch mt.currentSubPool { case PendingSubPool: if mt.minFeeCap != than.minFeeCap { return mt.minFeeCap.Cmp(&than.minFeeCap) < 0 } if mt.nonceDistance != than.nonceDistance { return mt.nonceDistance > than.nonceDistance } if mt.cumulativeBalanceDistance != than.cumulativeBalanceDistance { return mt.cumulativeBalanceDistance > than.cumulativeBalanceDistance } case BaseFeeSubPool, QueuedSubPool: if mt.nonceDistance != than.nonceDistance { return mt.nonceDistance > than.nonceDistance } if mt.cumulativeBalanceDistance != than.cumulativeBalanceDistance { return mt.cumulativeBalanceDistance > than.cumulativeBalanceDistance } } return mt.timestamp > than.timestamp } func (p BestQueue) Len() int { return len(p.ms) } func (p BestQueue) Less(i, j int) bool { return p.ms[i].better(p.ms[j], *uint256.NewInt(p.pendingBastFee)) } func (p BestQueue) Swap(i, j int) { p.ms[i], p.ms[j] = p.ms[j], p.ms[i] p.ms[i].bestIndex = i p.ms[j].bestIndex = j } func (p *BestQueue) Push(x interface{}) { n := len(p.ms) item := x.(*metaTx) item.bestIndex = n p.ms = append(p.ms, item) } func (p *BestQueue) Pop() interface{} { old := p.ms n := len(old) item := old[n-1] old[n-1] = nil // avoid memory leak item.bestIndex = -1 // for safety item.currentSubPool = 0 // for safety p.ms = old[0 : n-1] return item } type WorstQueue struct { ms []*metaTx pendingBaseFee uint64 } func (p WorstQueue) Len() int { return len(p.ms) } func (p WorstQueue) Less(i, j int) bool { return p.ms[i].worse(p.ms[j], *uint256.NewInt(p.pendingBaseFee)) } func (p WorstQueue) Swap(i, j int) { p.ms[i], p.ms[j] = p.ms[j], p.ms[i] p.ms[i].worstIndex = i p.ms[j].worstIndex = j } func (p *WorstQueue) Push(x interface{}) { n := len(p.ms) item := x.(*metaTx) item.worstIndex = n p.ms = append(p.ms, x.(*metaTx)) } func (p *WorstQueue) Pop() interface{} { old := p.ms n := len(old) item := old[n-1] old[n-1] = nil // avoid memory leak item.worstIndex = -1 // for safety item.currentSubPool = 0 // for safety p.ms = old[0 : n-1] return item }