add tx validation in buffer

core/txpool/blobpool/blobpool.go

@@ -1485,7 +1485,7 @@ func (p *BlobPool) ValidateTxBasics(tx *types.Transaction) error {
 		Accept:       1 << types.BlobTxType,
 		MaxSize:      txMaxSize,
 		MinTip:       p.gasTip.Load().ToBig(),
-		MaxBlobCount: maxBlobsPerTx,
+		MaxBlobCount: maxBlobsPerTx, //todo this field is currently not being used
 	}
 	return txpool.ValidateTransaction(tx, p.head.Load(), p.signer, opts)
 }

core/txpool/blobpool/buffer.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 )
 
+// todo: per-peer size limit
 var (
 	blobBufferTxFirstCounter    = metrics.NewRegisteredCounter("blobpool/buffer/txfirst", nil)
 	blobBufferCellsFirstCounter = metrics.NewRegisteredCounter("blobpool/buffer/cellsfirst", nil)
@@ -47,7 +48,9 @@ type PeerDelivery struct {
 }
 
 type txEntry struct {
-	tx    *types.Transaction
+	tx *types.Transaction
+	// Technically it is not required to store peer information to drop properly.
+	// This is mainly for per peer size limit check.
 	peer  string
 	added time.Time
 }
@@ -62,16 +65,18 @@ type BlobBuffer struct {
 	txs   map[common.Hash]*txEntry
 	cells map[common.Hash]*cellEntry
 
-	addToPool func(*BlobTxForPool) error
+	addToPool  func(*BlobTxForPool) error
-	dropPeer  func(string)
+	validateTx func(*types.Transaction) error
+	dropPeer   func(string)
 }
 
-func NewBlobBuffer(addToPool func(*BlobTxForPool) error, dropPeer func(string)) *BlobBuffer {
+func NewBlobBuffer(validateTx func(*types.Transaction) error, addToPool func(*BlobTxForPool) error, dropPeer func(string)) *BlobBuffer {
 	return &BlobBuffer{
-		txs:       make(map[common.Hash]*txEntry),
+		txs:        make(map[common.Hash]*txEntry),
-		cells:     make(map[common.Hash]*cellEntry),
+		cells:      make(map[common.Hash]*cellEntry),
-		addToPool: addToPool,
+		validateTx: validateTx,
-		dropPeer:  dropPeer,
+		addToPool:  addToPool,
+		dropPeer:   dropPeer,
 	}
 }
 
@@ -88,6 +93,12 @@ func (b *BlobBuffer) AddTx(tx *types.Transaction, peer string) error {
 	if sidecar == nil {
 		return fmt.Errorf("blob transaction without sidecar")
 	}
+	// tx validation
+	if err := b.validateTx(tx); err != nil {
+		log.Warn("Transaction validation failed, dropping peer", "peer", peer, "err", err)
+		b.dropPeer(peer)
+		return err
+	}
 	// vhash check
 	if err := sidecar.ValidateBlobCommitmentHashes(tx.BlobHashes()); err != nil {
 		log.Warn("Commitment hash mismatch, dropping peer", "peer", peer, "err", err)
@@ -99,13 +110,6 @@ func (b *BlobBuffer) AddTx(tx *types.Transaction, peer string) error {
 		b.dropPeer(peer)
 		return fmt.Errorf("insufficient proofs in sidecar")
 	}
-	// todo: I also considered performing additional validation for the metrics of the
-	// tx_fetcher. This could be used to avoid sending GetCells requests when the
-	// nonce is too low or the transaction is underpriced. However, doing so would
-	// require taking buffered transactions into account as well, and would require
-	// allowing the buffer to be part of the fetcher’s scheduling logic.
-	// Therefore, I will leave this as a TODO for now.
-
 	if entry, ok := b.cells[hash]; ok {
 		return b.add(hash, tx, entry)
 	}

core/txpool/blobpool/buffer_test.go

@@ -43,6 +43,7 @@ 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) {},
 	)
@@ -180,6 +181,7 @@ func TestBadCell(t *testing.T) {
 
 	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) },
 	)
@@ -220,6 +222,7 @@ func TestBadTx(t *testing.T) {
 
 	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) },
 	)

eth/handler.go

@@ -107,6 +107,7 @@ type blobPool interface {
 	GetBlobCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]*kzg4844.Cell, [][]*kzg4844.Proof, error)
 	GetCustody(hash common.Hash) *types.CustodyBitmap
 	AddPooledTx(pooledTx *blobpool.BlobTxForPool) error
+	ValidateTxBasics(pooledTx *types.Transaction) error
 }
 
 // handlerConfig is the collection of initialization parameters to create a full
@@ -189,7 +190,7 @@ func newHandler(config *handlerConfig) (*handler, error) {
 	}
 
 	// Construct the blob buffer for assembling blob txs from separate tx and cell deliveries.
-	h.blobBuffer = blobpool.NewBlobBuffer(h.blobpool.AddPooledTx, h.removePeer)
+	h.blobBuffer = blobpool.NewBlobBuffer(h.blobpool.ValidateTxBasics, h.blobpool.AddPooledTx, h.removePeer)
 
 	addTxs := func(peer string, txs []*types.Transaction) []error {
 		errs := make([]error, len(txs))

eth/handler_test.go

@@ -271,6 +271,10 @@ func (p *testTxPool) FilterType(kind byte) bool {
 	return false
 }
 
+func (p *testTxPool) ValidateTxBasics(_ *types.Transaction) error {
+	return nil
+}
+
 // testHandler is a live implementation of the Ethereum protocol handler, just
 // preinitialized with some sane testing defaults and the transaction pool mocked
 // out.