From 9527e04f482549dadb3fb1de6a9015c9ec0a81c8 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 18:01:17 +0200 Subject: [PATCH 1/6] core/txpool/blobpool: allow toggling cache mode on and off --- core/txpool/blobpool/cache.go | 42 +++++++++++++++++------------------ eth/catalyst/api.go | 19 +++++++--------- 2 files changed, 28 insertions(+), 33 deletions(-) diff --git a/core/txpool/blobpool/cache.go b/core/txpool/blobpool/cache.go index 23f53f0fc1..a6632ddb6c 100644 --- a/core/txpool/blobpool/cache.go +++ b/core/txpool/blobpool/cache.go @@ -81,11 +81,11 @@ type Cache struct { needCell bool // channels into loop - quit chan struct{} - topkRequest chan struct{} - topkTimer mclock.Timer - hasBlobsCh chan []common.Hash // list of tx hashes that should be pinned - enableCellCh chan struct{} // signals the loop to switch to cell mode + quit chan struct{} + topkRequest chan struct{} + topkTimer mclock.Timer + hasBlobsCh chan []common.Hash // list of tx hashes that should be pinned + cellModeCh chan bool // signals the loop to switch cell mode on/offo step func() // test hook fired after each loop iteration @@ -103,14 +103,14 @@ func NewCache(p *BlobPool) *Cache { // It allows injecting a clock and a step hook. func newCache(p *BlobPool, clock mclock.Clock, step func()) *Cache { c := &Cache{ - entries: make(map[common.Hash]*cachedBlob), - blobpool: p, - hasBlobsCh: make(chan []common.Hash, 1), - clock: clock, - step: step, - quit: make(chan struct{}), - topkRequest: make(chan struct{}, 1), - enableCellCh: make(chan struct{}, 1), + entries: make(map[common.Hash]*cachedBlob), + blobpool: p, + hasBlobsCh: make(chan []common.Hash, 1), + clock: clock, + step: step, + quit: make(chan struct{}), + topkRequest: make(chan struct{}, 1), + cellModeCh: make(chan bool, 1), } c.wg.Add(1) @@ -311,9 +311,9 @@ func (c *Cache) GetCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]* // blobs. This means we can also cache cells that lack enough blobs to // recover. It signals the loop to switch to cell mode and re-select // transactions from this wider pool. -func (c *Cache) EnableCell() { +func (c *Cache) SetCellMode(on bool) { select { - case c.enableCellCh <- struct{}{}: + case c.cellModeCh <- on: case <-c.quit: } } @@ -335,13 +335,11 @@ func (c *Cache) loop() { c.update(want) c.triggerTopKAfter(topKTimeout) - case <-c.enableCellCh: - // CL supports cell-based blob retrieval; switch to cell mode and - // re-select immediately over the now-wider pool. needCell is only - // touched here in the loop, so a fresh selection and update observe - // a consistent value. - if !c.needCell { - c.needCell = true + case on := <-c.cellModeCh: + // This runs when the CL signals (non-)support for cell proofs. Enable/disable + // cell mode and re-select immediately to force an update. + if c.needCell != on { + c.needCell = on c.triggerTopK() } diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go index f485633f0a..ba153b196d 100644 --- a/eth/catalyst/api.go +++ b/eth/catalyst/api.go @@ -22,6 +22,7 @@ import ( "errors" "fmt" "reflect" + "slices" "strconv" "sync" "sync/atomic" @@ -1198,17 +1199,13 @@ func (api *ConsensusAPI) checkFork(timestamp uint64, forks ...forks.Fork) bool { func (api *ConsensusAPI) ExchangeCapabilities(caps []string) []string { valueT := reflect.TypeOf(api) - for _, cap := range caps { - if cap == "engine_getBlobsV4" { - // If the CL supports getBlobsV4, we call EnableCell() on the - // blob cache to skip the blob recovery process. This is a - // one-directional toggle, which assumes that once the CL - // supports getBlobsV4, it will not fall back to getBlobsV3 - // again. - api.eth.BlobCache().EnableCell() - break - } - } + // If the CL supports getBlobsV4, we call EnableCell() on the + // blob cache to skip the blob recovery process. This is a + // one-directional toggle, which assumes that once the CL + // supports getBlobsV4, it will not fall back to getBlobsV3 + // again. + cellmode := slices.Contains(caps, "engine_getBlobsV4") + api.eth.BlobCache().SetCellMode(cellmode) ourCaps := make([]string, 0, valueT.NumMethod()) for i := 0; i < valueT.NumMethod(); i++ { From a4ed392269e015314e339e15030f478931029f6b Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 19:43:41 +0200 Subject: [PATCH 2/6] core/txpool/blobpool: refactor buffer callbacks --- core/txpool/blobpool/buffer.go | 29 +++++++++++++++-------------- core/txpool/blobpool/buffer_test.go | 20 ++++++++++---------- eth/handler.go | 6 +++++- 3 files changed, 30 insertions(+), 25 deletions(-) diff --git a/core/txpool/blobpool/buffer.go b/core/txpool/blobpool/buffer.go index 24437f4029..63109363dc 100644 --- a/core/txpool/blobpool/buffer.go +++ b/core/txpool/blobpool/buffer.go @@ -69,21 +69,22 @@ type BlobBuffer struct { txs map[common.Hash]*txEntry cells map[common.Hash]*cellEntry - addToPool func(*BlobTxForPool) error - validateTx func(*types.Transaction) error - dropPeer func(string) - completed []*BlobTxForPool completedCount atomic.Int32 + cb BlobBufferFunctions } -func NewBlobBuffer(validateTx func(*types.Transaction) error, addToPool func(*BlobTxForPool) error, dropPeer func(string)) *BlobBuffer { +type BlobBufferFunctions struct { + ValidateTx func(*types.Transaction) error + AddToPool func(*BlobTxForPool) error + DropPeer func(string) +} + +func NewBlobBuffer(cb BlobBufferFunctions) *BlobBuffer { return &BlobBuffer{ - txs: make(map[common.Hash]*txEntry), - cells: make(map[common.Hash]*cellEntry), - validateTx: validateTx, - addToPool: addToPool, - dropPeer: dropPeer, + txs: make(map[common.Hash]*txEntry), + cells: make(map[common.Hash]*cellEntry), + cb: cb, } } @@ -102,7 +103,7 @@ func (b *BlobBuffer) Flush() ([]common.Hash, []error) { errs := make([]error, len(b.completed)) for i, ptx := range b.completed { txs[i] = ptx.Tx.Hash() - errs[i] = b.addToPool(ptx) + errs[i] = b.cb.AddToPool(ptx) } b.completed = nil b.completedCount.Store(0) @@ -129,7 +130,7 @@ func (b *BlobBuffer) AddTx(txs []*types.Transaction, peer string) []error { } // tx validation (basic w/o lock) // error will be handled by tx fetcher - if err := b.validateTx(tx); err != nil { + if err := b.cb.ValidateTx(tx); err != nil { errs[i] = err continue } @@ -210,11 +211,11 @@ func (b *BlobBuffer) HasCells(hash common.Hash) bool { } func (b *BlobBuffer) dropPeers(peers []string) { - if b.dropPeer == nil { + if b.cb.DropPeer == nil { return } for _, p := range peers { - b.dropPeer(p) + b.cb.DropPeer(p) } } diff --git a/core/txpool/blobpool/buffer_test.go b/core/txpool/blobpool/buffer_test.go index ef2aadd3de..085cd793c7 100644 --- a/core/txpool/blobpool/buffer_test.go +++ b/core/txpool/blobpool/buffer_test.go @@ -39,11 +39,11 @@ func makePeerDelivery(t *testing.T, blobOffset, blobCount int, indices []uint64) func newTestBuffer(t *testing.T) *BlobBuffer { t.Helper() - return NewBlobBuffer( - func(tx *types.Transaction) error { return nil }, - func(ptx *BlobTxForPool) error { return nil }, - func(peer string) {}, - ) + return NewBlobBuffer(BlobBufferFunctions{ + ValidateTx: func(tx *types.Transaction) error { return nil }, + AddToPool: func(ptx *BlobTxForPool) error { return nil }, + DropPeer: func(peer string) {}, + }) } func TestSortCells(t *testing.T) { @@ -171,11 +171,11 @@ func TestBadCell(t *testing.T) { blobCount := 1 var dropped []string - buf := NewBlobBuffer( - func(tx *types.Transaction) error { return nil }, - func(ptx *BlobTxForPool) error { return nil }, - func(peer string) { dropped = append(dropped, peer) }, - ) + buf := NewBlobBuffer(BlobBufferFunctions{ + ValidateTx: func(tx *types.Transaction) error { return nil }, + AddToPool: func(ptx *BlobTxForPool) error { return nil }, + DropPeer: func(peer string) { dropped = append(dropped, peer) }, + }) tx := makeV1Tx(t, 0, blobCount, 0, key) hash := tx.Hash() diff --git a/eth/handler.go b/eth/handler.go index ab49b8a289..10edc5d7b5 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -190,7 +190,11 @@ func newHandler(config *handlerConfig) (*handler, error) { } // Construct the blob buffer for assembling blob txs from separate tx and cell deliveries. - blobBuffer := blobpool.NewBlobBuffer(h.blobpool.ValidateTxBasics, h.blobpool.AddPooledTx, h.removePeer) + blobBuffer := blobpool.NewBlobBuffer(blobpool.BlobBufferFunctions{ + ValidateTx: h.blobpool.ValidateTxBasics, + AddToPool: h.blobpool.AddPooledTx, + DropPeer: h.removePeer, + }) addTxs := func(txs []*types.Transaction) []error { return h.txpool.Add(txs, false) From 589f80ca56bd839b6c412f7d4ba5f05b1c3e09b8 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 19:54:37 +0200 Subject: [PATCH 3/6] core/txpool/blobpool: remove todo --- core/txpool/blobpool/buffer.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/txpool/blobpool/buffer.go b/core/txpool/blobpool/buffer.go index 63109363dc..c17ffd16b1 100644 --- a/core/txpool/blobpool/buffer.go +++ b/core/txpool/blobpool/buffer.go @@ -91,11 +91,13 @@ func NewBlobBuffer(cb BlobBufferFunctions) *BlobBuffer { // Flush adds all completed entries to the pool and returns the hashes // and corresponding errors (nil on success) for each attempted insert. func (b *BlobBuffer) Flush() ([]common.Hash, []error) { - // Read the count first and return early - // todo: increase threshold ? + // Read the count first and return early if there is nothing to do. + // Flush is called very frequently from the blob fetcher so this + // optimization is warranted. if b.completedCount.Load() == 0 { return nil, nil } + b.mu.Lock() defer b.mu.Unlock() From 4d107e77dc493ce072e5c2b4813fc8a3d4d188c2 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 19:54:51 +0200 Subject: [PATCH 4/6] core/txpool/blobpool: rename buffer.add into storeCompleted --- core/txpool/blobpool/buffer.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/core/txpool/blobpool/buffer.go b/core/txpool/blobpool/buffer.go index c17ffd16b1..140c05e574 100644 --- a/core/txpool/blobpool/buffer.go +++ b/core/txpool/blobpool/buffer.go @@ -137,7 +137,7 @@ func (b *BlobBuffer) AddTx(txs []*types.Transaction, peer string) []error { continue } if entry, ok := b.cells[hash]; ok { - b.add(hash, tx, entry) + b.storeCompleted(hash, tx, entry) continue } blobBufferTxFirstCounter.Inc(1) @@ -162,13 +162,14 @@ func (b *BlobBuffer) AddCells(hash common.Hash, deliveries map[string]*PeerDeliv added: time.Now(), } if txe, ok := b.txs[hash]; ok { - b.add(hash, txe.tx, b.cells[hash]) + b.storeCompleted(hash, txe.tx, b.cells[hash]) } blobBufferCellsFirstCounter.Inc(1) } -// add verifies cells per-peer, sorts them, and adds to the pool. -func (b *BlobBuffer) add(hash common.Hash, tx *types.Transaction, cells *cellEntry) { +// storeCompleted verifies cells per-peer, sorts them, and schedules them for +// addition into the pool. The actual addition happens in Flush(). +func (b *BlobBuffer) storeCompleted(hash common.Hash, tx *types.Transaction, cells *cellEntry) { sidecar := tx.BlobTxSidecar() // Per-peer cell verification From 63fe3ff7ffcf76fb32078b65d63fe4705bbc360f Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 19:57:04 +0200 Subject: [PATCH 5/6] eth/fetcher: blob fetcher style edits --- eth/fetcher/blob_fetcher.go | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/eth/fetcher/blob_fetcher.go b/eth/fetcher/blob_fetcher.go index 0513a6b785..6c0825fe0e 100644 --- a/eth/fetcher/blob_fetcher.go +++ b/eth/fetcher/blob_fetcher.go @@ -746,20 +746,20 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} if len(actives) == 0 { return } - // For each active peer, try to schedule some payload fetches - idle := len(f.requests) == 0 + wasIdle := len(f.requests) == 0 + + // For each active peer, try to schedule some payload fetches. f.forEachPeer(actives, func(peer string) { if len(f.announces[peer]) == 0 || len(f.requests[peer]) != 0 { return // continue } var ( - hashes = make([]common.Hash, 0, maxTxRetrievals) - custodies = make([]*types.CustodyBitmap, 0, maxTxRetrievals) + hashes []common.Hash + custodies []*types.CustodyBitmap ) f.forEachAnnounce(f.announces[peer], func(hash common.Hash, cells *types.CustodyBitmap) bool { var unfetched *types.CustodyBitmap - if f.fetches[hash] == nil { // tx is not being fetched unfetched = cells @@ -795,6 +795,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} return len(hashes) < maxPayloadRetrievals }) + // If any hashes were allocated, request them from the peer if len(hashes) > 0 { // Group hashes by custody bitmap @@ -817,7 +818,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} request = append(request, cr) } f.requests[peer] = request - go func(peer string, request []*cellRequest) { + go func() { for _, req := range request { blobRequestOutMeter.Mark(int64(len(req.txs))) if err := f.fn.FetchPayloads(peer, req.txs, req.cells); err != nil { @@ -826,11 +827,12 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} break } } - }(peer, request) + }() } }) + // If a new request was fired, schedule a timeout timer - if idle && len(f.requests) > 0 { + if wasIdle && len(f.requests) > 0 { f.rescheduleTimeout(timer, timeout) } } From 6489db71edddb7d02d216a280abe0e07272225bb Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 20:18:58 +0200 Subject: [PATCH 6/6] eth/fetcher: add comment --- eth/fetcher/tx_fetcher.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 6a43b2e238..55eef46e0c 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -379,6 +379,8 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction, poolTxs = batch } batch = append(poolTxs, blobTxs...) + + // Add regular tx to pool, blob tx to buffer. errs := append(f.addTxs(poolTxs), f.buffer.AddTx(blobTxs, peer)...) hashes := make([]common.Hash, len(batch))