eth, cmd: add flag for fetch probability (blobpool.fetchprobability)

cmd/geth/main.go

@@ -75,6 +75,7 @@ var (
 		utils.BlobPoolDataDirFlag,
 		utils.BlobPoolDataCapFlag,
 		utils.BlobPoolPriceBumpFlag,
+		utils.BlobPoolFetchProbabilityFlag,
 		utils.SyncModeFlag,
 		utils.SyncTargetFlag,
 		utils.ExitWhenSyncedFlag,

cmd/utils/flags.go

@@ -48,6 +48,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
+	"github.com/ethereum/go-ethereum/eth/fetcher"
 	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	"github.com/ethereum/go-ethereum/eth/syncer"
@@ -497,6 +498,12 @@ var (
 		Value:    ethconfig.Defaults.BlobPool.PriceBump,
 		Category: flags.BlobPoolCategory,
 	}
+	BlobPoolFetchProbabilityFlag = &cli.Uint64Flag{
+		Name:     "blobpool.fetchprobability",
+		Usage:    "Probability of fetching the full blob payload for sparse blobpool",
+		Value:    fetcher.DefaultFetchProbability,
+		Category: flags.BlobPoolCategory,
+	}
 	// Performance tuning settings
 	CacheFlag = &cli.IntFlag{
 		Name:     "cache",
@@ -1669,6 +1676,10 @@ func setBlobPool(ctx *cli.Context, cfg *blobpool.Config) {
 	if ctx.IsSet(BlobPoolPriceBumpFlag.Name) {
 		cfg.PriceBump = ctx.Uint64(BlobPoolPriceBumpFlag.Name)
 	}
+	if ctx.IsSet(BlobPoolFetchProbabilityFlag.Name) {
+		v := ctx.Uint64(BlobPoolFetchProbabilityFlag.Name)
+		cfg.FetchProbability = &v
+	}
 }
 
 func setMiner(ctx *cli.Context, cfg *miner.Config) {

core/txpool/blobpool/config.go

@@ -25,6 +25,8 @@ type Config struct {
 	Datadir   string // Data directory containing the currently executable blobs
 	Datacap   uint64 // Soft-cap of database storage (hard cap is larger due to overhead)
 	PriceBump uint64 // Minimum price bump percentage to replace an already existing nonce
+
+	FetchProbability *uint64 // EIP-8070: full blob fetch probability for sparse blobpool
 }
 
 // DefaultConfig contains the default configurations for the transaction pool.

eth/backend.go

@@ -344,20 +344,27 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		stack.RegisterLifecycle(eth.localTxTracker)
 	}
 
+	var fetchProb uint64
+	if config.BlobPool.FetchProbability != nil {
+		fetchProb = *config.BlobPool.FetchProbability
+	} else {
+		fetchProb = fetcher.DefaultFetchProbability
+	}
+
 	// Permit the downloader to use the trie cache allowance during fast sync
 	cacheLimit := options.TrieCleanLimit + options.TrieDirtyLimit + options.SnapshotLimit
 	if eth.handler, err = newHandler(&handlerConfig{
-		NodeID:         eth.p2pServer.Self().ID(),
+		NodeID:           eth.p2pServer.Self().ID(),
-		Database:       chainDb,
+		Database:         chainDb,
-		Chain:          eth.blockchain,
+		Chain:            eth.blockchain,
-		TxPool:         eth.txPool,
+		TxPool:           eth.txPool,
-		BlobPool:       eth.blobTxPool,
+		BlobPool:         eth.blobTxPool,
-		Network:        networkID,
+		Network:          networkID,
-		Sync:           config.SyncMode,
+		Sync:             config.SyncMode,
-		BloomCache:     uint64(cacheLimit),
+		BloomCache:       uint64(cacheLimit),
-		RequiredBlocks: config.RequiredBlocks,
+		RequiredBlocks:   config.RequiredBlocks,
-		Custody:        types.CustodyBitmapAll,
+		SnapV2:           config.SnapV2,
-		SnapV2:         config.SnapV2,
+		FetchProbability: fetchProb,
 	}); err != nil {
 		return nil, err
 	}

eth/fetcher/blob_fetcher.go

@@ -41,12 +41,14 @@ type random interface {
 var blobFetchTimeout = 5 * time.Second
 var blobAvailabilityTimeout = 2 * time.Second
 
+// DefaultFetchProbability is the default probability of fetching the full blob
+// payload for the sparse blobpool.
+const DefaultFetchProbability = 15
+
 const (
-	availabilityThreshold         = 2
+	availabilityThreshold = 2
-	maxPayloadRetrievals          = 128
+	maxPayloadRetrievals  = 128
-	maxPayloadAnnounces           = 4096
+	maxPayloadAnnounces   = 4096
-	fetchProbability              = 15
-	MAX_CELLS_PER_PARTIAL_REQUEST = 8
 
 	// maxCellRequests caps the burst of cell requests we can issue at once
 	// to a single peer. Worst case 256 * 6 = 1536 cells (~3 MB)
@@ -140,7 +142,8 @@ type BlobFetcher struct {
 	requests   map[string][]*cellRequest                      // In-flight transaction retrievals
 	alternates map[common.Hash]map[string]types.CustodyBitmap // In-flight transaction alternate origins (in case the peer is dropped)
 
-	fn BlobFetcherFunctions // callbacks
+	fn               BlobFetcherFunctions // callbacks
+	fetchProbability uint64
 
 	// peerTokens tracks each peer's remaining cell request token.
 	peerTokens map[string]*token
@@ -157,28 +160,29 @@ type token struct {
 	last   mclock.AbsTime
 }
 
-func NewBlobFetcher(fn BlobFetcherFunctions, custody types.CustodyBitmap, rand random) *BlobFetcher {
+func NewBlobFetcher(fn BlobFetcherFunctions, custody types.CustodyBitmap, rand random, fetchProbability uint64) *BlobFetcher {
 	return &BlobFetcher{
-		notify:     make(chan *blobTxAnnounce),
+		notify:           make(chan *blobTxAnnounce),
-		cleanup:    make(chan *payloadDelivery),
+		cleanup:          make(chan *payloadDelivery),
-		drop:       make(chan *txDrop),
+		drop:             make(chan *txDrop),
-		custodyCh:  make(chan types.CustodyBitmap),
+		custodyCh:        make(chan types.CustodyBitmap),
-		quit:       make(chan struct{}),
+		quit:             make(chan struct{}),
-		full:       make(map[common.Hash]struct{}),
+		full:             make(map[common.Hash]struct{}),
-		partial:    make(map[common.Hash]struct{}),
+		partial:          make(map[common.Hash]struct{}),
-		waitlist:   make(map[common.Hash]map[string]struct{}),
+		waitlist:         make(map[common.Hash]map[string]struct{}),
-		waittime:   make(map[common.Hash]mclock.AbsTime),
+		waittime:         make(map[common.Hash]mclock.AbsTime),
-		waitslots:  make(map[string]map[common.Hash]struct{}),
+		waitslots:        make(map[string]map[common.Hash]struct{}),
-		announces:  make(map[string]map[common.Hash]*cellWithSeq),
+		announces:        make(map[string]map[common.Hash]*cellWithSeq),
-		fetches:    make(map[common.Hash]*fetchStatus),
+		fetches:          make(map[common.Hash]*fetchStatus),
-		requests:   make(map[string][]*cellRequest),
+		requests:         make(map[string][]*cellRequest),
-		alternates: make(map[common.Hash]map[string]types.CustodyBitmap),
+		alternates:       make(map[common.Hash]map[string]types.CustodyBitmap),
-		peerTokens: make(map[string]*token),
+		peerTokens:       make(map[string]*token),
-		fn:         fn,
+		fn:               fn,
-		custody:    custody,
+		fetchProbability: fetchProbability,
-		clock:      mclock.System{},
+		custody:          custody,
-		realTime:   time.Now,
+		clock:            mclock.System{},
-		rand:       rand,
+		realTime:         time.Now,
+		rand:             rand,
 	}
 }
 
@@ -295,7 +299,7 @@ func (f *BlobFetcher) loop() {
 							randomValue = f.rand.Intn(100)
 						}
 						// For eager mode, always fetch immediately
-						if randomValue < fetchProbability || f.custody.OneCount() >= kzg4844.DataPerBlob {
+						if uint64(randomValue) < f.fetchProbability || f.custody.OneCount() >= kzg4844.DataPerBlob {
 							f.full[hash] = struct{}{}
 						} else {
 							f.partial[hash] = struct{}{}

eth/fetcher/blob_fetcher_test.go

@@ -162,6 +162,7 @@ func TestBlobFetcherFullFetch(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 5}, // Force full requests (5 < fetchProbability)
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -253,6 +254,7 @@ func TestBlobFetcherPartialFetch(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 60}, // Force partial requests (20 >= 15)
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -345,6 +347,7 @@ func TestBlobFetcherFullDelivery(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 5}, // Force full requests for simplicity
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -393,6 +396,7 @@ func TestBlobFetcherPartialDelivery(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 60},
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -529,6 +533,7 @@ func TestBlobFetcherAvailabilityTimeout(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 60},
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -571,6 +576,7 @@ func TestBlobFetcherPeerDrop(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 5},
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -648,6 +654,7 @@ func TestBlobFetcherFetchTimeout(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 5},
+				15,
 			)
 		},
 		steps: []interface{}{
@@ -1044,6 +1051,7 @@ func TestMultiBlobDeliveryVerification(t *testing.T) {
 				},
 				custody,
 				&mockRand{value: 60}, // Force partial requests (60 >= fetchProbability)
+				15,
 			)
 		},
 		steps: []interface{}{

eth/handler.go

@@ -113,17 +113,17 @@ type blobPool interface {
 // handlerConfig is the collection of initialization parameters to create a full
 // node network handler.
 type handlerConfig struct {
-	NodeID         enode.ID               // P2P node ID used for tx propagation topology
+	NodeID           enode.ID               // P2P node ID used for tx propagation topology
-	Database       ethdb.Database         // Database for direct sync insertions
+	Database         ethdb.Database         // Database for direct sync insertions
-	Chain          *core.BlockChain       // Blockchain to serve data from
+	Chain            *core.BlockChain       // Blockchain to serve data from
-	TxPool         txPool                 // Transaction pool to propagate from
+	TxPool           txPool                 // Transaction pool to propagate from
-	BlobPool       blobPool               // Blob pool for cell-based blob data availability
+	BlobPool         blobPool               // Blob pool for cell-based blob data availability
-	Network        uint64                 // Network identifier to advertise
+	Network          uint64                 // Network identifier to advertise
-	Sync           ethconfig.SyncMode     // Whether to snap or full sync
+	Sync             ethconfig.SyncMode     // Whether to snap or full sync
-	BloomCache     uint64                 // Megabytes to alloc for snap sync bloom
+	BloomCache       uint64                 // Megabytes to alloc for snap sync bloom
-	RequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
+	RequiredBlocks   map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges
-	Custody        types.CustodyBitmap
+	SnapV2           bool                   // Whether to advertise and sync via the snap/2 protocol
-	SnapV2         bool // Whether to advertise and sync via the snap/2 protocol
+	FetchProbability uint64                 // Full blob fetch probability for sparse blobpool (blobFetcher)
 }
 
 type handler struct {
@@ -231,7 +231,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		},
 		DropPeer: h.removePeer,
 	}
-	h.blobFetcher = fetcher.NewBlobFetcher(blobCallbacks, config.Custody, nil)
+	h.blobFetcher = fetcher.NewBlobFetcher(blobCallbacks, types.CustodyBitmapAll, nil, config.FetchProbability)
 	return h, nil
 }