From a4ed392269e015314e339e15030f478931029f6b Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 12 Jun 2026 19:43:41 +0200 Subject: [PATCH] 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)