Merge db0b68bd85 into dddbaa4bf3

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.

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
 }

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]]
 

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)

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)