From db0b68bd85fcf47dfe1e767111bf3a6990b95cd2 Mon Sep 17 00:00:00 2001 From: healthykim Date: Tue, 7 Jul 2026 11:28:29 +0200 Subject: [PATCH] core/txpool: add blocked transaction size cap --- core/txpool/blobpool/blobpool.go | 61 +++++++++++++++++++++++--- core/txpool/blobpool/config.go | 8 ++++ core/txpool/blobpool/evictheap.go | 19 ++++++-- core/txpool/blobpool/evictheap_test.go | 6 +-- core/txpool/blobpool/metrics.go | 2 + 5 files changed, 83 insertions(+), 13 deletions(-) diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index c013b7b1e3..d2d9f4fad2 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -352,6 +352,31 @@ func newBlobTxMeta(id uint64, storageSize uint32, ptx *BlobTxForPool) *blobTxMet return meta } +// updateBlocked updates the total size of transactions blocked by a partial +// transaction from the given account. It should be called after every p.index +// modification. +func (p *BlobPool) updateBlocked(addr common.Address) { + blockIndex := -1 + for i, m := range p.index[addr] { + if m.custody.OneCount() < kzg4844.DataPerBlob { + blockIndex = i + break + } + } + if blockIndex < 0 { + p.blocked -= p.blockedAccount[addr] + delete(p.blockedAccount, addr) + return + } + var bytes uint64 + for _, m := range p.index[addr][blockIndex:] { + bytes += uint64(m.storageSize) + } + + p.blocked = p.blocked - p.blockedAccount[addr] + bytes + p.blockedAccount[addr] = bytes +} + // BlobPool is the transaction pool dedicated to EIP-4844 blob transactions. // // Blob transactions are special snowflakes that are designed for a very specific @@ -543,6 +568,17 @@ type BlobPool struct { stored uint64 // Useful data size of all transactions on disk limbo *limbo // Persistent data store for the non-finalized blobs + // A partial blob transaction (custody count below DataPerBlob) cannot be included + // by our block. Since an account's transactions are included in nonce order, the first + // partial transaction blocks all following transactions in one account from + // inclusion. That transaction and all transactions after it are called the + // account's "blocked" transactions. To prevent DoS, the total size of blocked transactions + // is capped by blockedCap separately from the overall data cap. + + blocked uint64 // Data size of blocked transactions across all accounts + blockedAccount map[common.Address]uint64 // Per-account blocked transaction bytes + blockedCap uint64 // Maximum blocked data size (Datacap * BlockedRatio) + cQueue *conversionQueue gapped map[common.Address][]*BlobTxForPool // Transactions that are currently gapped (nonce too high) @@ -581,6 +617,8 @@ func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bo lookup: newLookup(), index: make(map[common.Address][]*blobTxMeta), spent: make(map[common.Address]*uint256.Int), + blockedAccount: make(map[common.Address]uint64), + blockedCap: uint64(float64(config.Datacap) * config.BlockedRatio), gapped: make(map[common.Address][]*BlobTxForPool), gappedSource: make(map[common.Hash]common.Address), } @@ -686,7 +724,7 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser if head.ExcessBlobGas != nil { blobfee = uint256.MustFromBig(eip4844.CalcBlobFee(p.chain.Config(), head)) } - p.evict = newPriceHeap(basefee, blobfee, p.index) + p.evict = newPriceHeap(basefee, blobfee, p.index, p.blockedAccount) // Guess what was announced. This is needed because we don't want to // participate in the diffusion of transactions where inclusion is blocked by @@ -718,7 +756,7 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser // Since the user might have modified their pool's capacity, evict anything // above the current allowance - for p.stored > p.config.Datacap { + for p.stored > p.config.Datacap || p.blocked > p.blockedCap { p.drop() } // Update the metrics and return the constructed pool @@ -935,6 +973,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6 } delete(p.index, addr) delete(p.spent, addr) + p.updateBlocked(addr) if inclusions != nil { // only during reorgs will the heap be initialized heap.Remove(p.evict, p.evict.index[addr]) } @@ -1136,6 +1175,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6 } } } + p.updateBlocked(addr) // Included cheap transactions might have left the remaining ones better from // an eviction point, fix any potential issues in the heap. if _, ok := p.index[addr]; ok && inclusions != nil { @@ -1521,11 +1561,13 @@ func (p *BlobPool) SetGasTip(tip *big.Int) { // Clear out the dropped transactions from the index if i > 0 { p.index[addr] = txs[:i] + p.updateBlocked(addr) heap.Fix(p.evict, p.evict.index[addr]) } else { delete(p.index, addr) delete(p.spent, addr) + p.updateBlocked(addr) heap.Remove(p.evict, p.evict.index[addr]) p.reserver.Release(addr) } @@ -2137,6 +2179,8 @@ func (p *BlobPool) addLocked(ptx *BlobTxForPool, checkGapped bool) (err error) { txs[i].evictionBlobFeeJumps = txs[i].blobfeeJumps } } + isBlocked := p.blockedAccount[from] > 0 + p.updateBlocked(from) // Update the eviction heap with the new information: // - If the transaction is from a new account, add it to the heap // - If the account had a singleton tx replaced, update the heap (new price caps) @@ -2152,13 +2196,13 @@ func (p *BlobPool) addLocked(ptx *BlobTxForPool, checkGapped bool) (err error) { evictionExecFeeDiff := oldEvictionExecFeeJumps - txs[len(txs)-1].evictionExecFeeJumps evictionBlobFeeDiff := oldEvictionBlobFeeJumps - txs[len(txs)-1].evictionBlobFeeJumps - if math.Abs(evictionExecFeeDiff) > 0.001 || math.Abs(evictionBlobFeeDiff) > 0.001 { // need math.Abs, can go up and down + if math.Abs(evictionExecFeeDiff) > 0.001 || math.Abs(evictionBlobFeeDiff) > 0.001 || isBlocked != (p.blockedAccount[from] > 0) { // need math.Abs, can go up and down heap.Fix(p.evict, p.evict.index[from]) } } // If the pool went over the allowed data limit, evict transactions until // we're again below the threshold - for p.stored > p.config.Datacap { + for p.stored > p.config.Datacap || p.blocked > p.blockedCap { p.drop() } p.updateStorageMetrics() @@ -2256,6 +2300,8 @@ func (p *BlobPool) drop() { } p.stored -= uint64(drop.storageSize) p.lookup.untrack(drop) + isBlocked := p.blockedAccount[from] > 0 + p.updateBlocked(from) // Remove the transaction from the pool's eviction heap: // - If the entire account was dropped, pop off the address @@ -2268,7 +2314,7 @@ func (p *BlobPool) drop() { evictionExecFeeDiff := tail.evictionExecFeeJumps - drop.evictionExecFeeJumps evictionBlobFeeDiff := tail.evictionBlobFeeJumps - drop.evictionBlobFeeJumps - if evictionExecFeeDiff > 0.001 || evictionBlobFeeDiff > 0.001 { // no need for math.Abs, monotonic decreasing + if evictionExecFeeDiff > 0.001 || evictionBlobFeeDiff > 0.001 || isBlocked != (p.blockedAccount[from] > 0) { // no need for math.Abs, monotonic decreasing heap.Fix(p.evict, 0) } } @@ -2401,6 +2447,7 @@ func (p *BlobPool) updateStorageMetrics() { datausedGauge.Update(int64(dataused)) datarealGauge.Update(int64(datareal)) slotusedGauge.Update(int64(slotused)) + blockedGauge.Update(int64(p.blocked)) oversizedDatausedGauge.Update(int64(oversizedDataused)) oversizedDatagapsGauge.Update(int64(oversizedDatagaps)) @@ -2608,9 +2655,11 @@ func (p *BlobPool) Clear() { p.lookup = newLookup() p.index = make(map[common.Address][]*blobTxMeta) p.spent = make(map[common.Address]*uint256.Int) + p.blockedAccount = make(map[common.Address]uint64) // Reset counters and the gapped buffer p.stored = 0 + p.blocked = 0 p.gapped = make(map[common.Address][]*BlobTxForPool) p.gappedSource = make(map[common.Hash]common.Address) @@ -2618,7 +2667,7 @@ func (p *BlobPool) Clear() { basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), p.head.Load())) blobfee = uint256.NewInt(params.BlobTxMinBlobGasprice) ) - p.evict = newPriceHeap(basefee, blobfee, p.index) + p.evict = newPriceHeap(basefee, blobfee, p.index, p.blockedAccount) } // GetCustody returns the custody bitmap for a given transaction hash. diff --git a/core/txpool/blobpool/config.go b/core/txpool/blobpool/config.go index d4c7d1c0f6..31cea47518 100644 --- a/core/txpool/blobpool/config.go +++ b/core/txpool/blobpool/config.go @@ -27,6 +27,8 @@ type Config struct { PriceBump uint64 // Minimum price bump percentage to replace an already existing nonce FetchProbability uint64 // EIP-8070: full blob fetch probability for sparse blobpool + + BlockedRatio float64 // Maximum fraction of Datacap the blocked (partial-gated) suffix may occupy } // DefaultConfig contains the default configurations for the transaction pool. @@ -34,6 +36,8 @@ var DefaultConfig = Config{ Datadir: "blobpool", Datacap: 10 * 1024 * 1024 * 1024 / 4, // TODO(karalabe): /4 handicap for rollout, gradually bump back up to 10GB PriceBump: 100, // either have patience or be aggressive, no mushy ground + + BlockedRatio: 0.5, } // sanitize checks the provided user configurations and changes anything that's @@ -48,5 +52,9 @@ func (config *Config) sanitize() Config { log.Warn("Sanitizing invalid blobpool price bump", "provided", conf.PriceBump, "updated", DefaultConfig.PriceBump) conf.PriceBump = DefaultConfig.PriceBump } + if conf.BlockedRatio <= 0 || conf.BlockedRatio > 1 { + log.Warn("Sanitizing invalid blobpool blocked cap ratio", "provided", conf.BlockedRatio, "updated", DefaultConfig.BlockedRatio) + conf.BlockedRatio = DefaultConfig.BlockedRatio + } return conf } diff --git a/core/txpool/blobpool/evictheap.go b/core/txpool/blobpool/evictheap.go index a46b8e9a6f..b4d93b2235 100644 --- a/core/txpool/blobpool/evictheap.go +++ b/core/txpool/blobpool/evictheap.go @@ -35,7 +35,8 @@ import ( // The goal of the heap is to decide which account has the worst bottleneck to // evict transactions from. type evictHeap struct { - metas map[common.Address][]*blobTxMeta // Pointer to the blob pool's index for price retrievals + metas map[common.Address][]*blobTxMeta // Pointer to the blob pool's index for price retrievals + blocked map[common.Address]uint64 // Pointer to the blob pool's per-account blocked byte counts basefeeJumps float64 // Pre-calculated absolute dynamic fee jumps for the base fee blobfeeJumps float64 // Pre-calculated absolute dynamic fee jumps for the blob fee @@ -46,10 +47,11 @@ type evictHeap struct { // newPriceHeap creates a new heap of cheapest accounts in the blob pool to evict // from in case of over saturation. -func newPriceHeap(basefee *uint256.Int, blobfee *uint256.Int, index map[common.Address][]*blobTxMeta) *evictHeap { +func newPriceHeap(basefee *uint256.Int, blobfee *uint256.Int, index map[common.Address][]*blobTxMeta, blocked map[common.Address]uint64) *evictHeap { heap := &evictHeap{ - metas: index, - index: make(map[common.Address]int, len(index)), + metas: index, + blocked: blocked, + index: make(map[common.Address]int, len(index)), } // Populate the heap in account sort order. Not really needed in practice, // but it makes the heap initialization deterministic and less annoying to @@ -88,6 +90,15 @@ func (h *evictHeap) Len() int { // Less implements sort.Interface as part of heap.Interface, returning which of // the two requested accounts has a cheaper bottleneck. func (h *evictHeap) Less(i, j int) bool { + blockedI := h.blocked[h.addrs[i]] > 0 + blockedJ := h.blocked[h.addrs[j]] > 0 + if blockedI != blockedJ { + // If one of the given account is a blocked account, that account + // should be considered as cheaper. + // Otherwise (if both are blocked or not blocked) they should be + // considered with fees + return blockedI + } txsI := h.metas[h.addrs[i]] txsJ := h.metas[h.addrs[j]] diff --git a/core/txpool/blobpool/evictheap_test.go b/core/txpool/blobpool/evictheap_test.go index 112fa77a01..328c4a8ff7 100644 --- a/core/txpool/blobpool/evictheap_test.go +++ b/core/txpool/blobpool/evictheap_test.go @@ -159,7 +159,7 @@ func TestPriceHeapSorting(t *testing.T) { }} } // Create a price heap and check the pop order - priceheap := newPriceHeap(uint256.NewInt(tt.basefee), uint256.NewInt(tt.blobfee), index) + priceheap := newPriceHeap(uint256.NewInt(tt.basefee), uint256.NewInt(tt.blobfee), index, nil) verifyHeapInternals(t, priceheap) for j := 0; j < len(tt.order); j++ { @@ -218,7 +218,7 @@ func benchmarkPriceHeapReinit(b *testing.B, datacap uint64) { }} } // Create a price heap and reinit it over and over - heap := newPriceHeap(uint256.NewInt(rnd.Uint64()), uint256.NewInt(rnd.Uint64()), index) + heap := newPriceHeap(uint256.NewInt(rnd.Uint64()), uint256.NewInt(rnd.Uint64()), index, nil) basefees := make([]*uint256.Int, b.N) blobfees := make([]*uint256.Int, b.N) @@ -294,7 +294,7 @@ func benchmarkPriceHeapOverflow(b *testing.B, datacap uint64) { }} } // Create a price heap and overflow it over and over - evict := newPriceHeap(uint256.NewInt(rnd.Uint64()), uint256.NewInt(rnd.Uint64()), index) + evict := newPriceHeap(uint256.NewInt(rnd.Uint64()), uint256.NewInt(rnd.Uint64()), index, nil) var ( addrs = make([]common.Address, b.N) metas = make([]*blobTxMeta, b.N) diff --git a/core/txpool/blobpool/metrics.go b/core/txpool/blobpool/metrics.go index 44e2098b22..7bac583791 100644 --- a/core/txpool/blobpool/metrics.go +++ b/core/txpool/blobpool/metrics.go @@ -29,6 +29,8 @@ var ( datarealGauge = metrics.NewRegisteredGauge("blobpool/datareal", nil) slotusedGauge = metrics.NewRegisteredGauge("blobpool/slotused", nil) + blockedGauge = metrics.NewRegisteredGauge("blobpool/blocked", nil) + limboDatausedGauge = metrics.NewRegisteredGauge("blobpool/limbo/dataused", nil) limboDatarealGauge = metrics.NewRegisteredGauge("blobpool/limbo/datareal", nil) limboSlotusedGauge = metrics.NewRegisteredGauge("blobpool/limbo/slotused", nil)