core/txpool/blobpool: fix static check

core/txpool/blobpool/blobpool.go

@@ -27,6 +27,7 @@ import (
 	"path/filepath"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -93,6 +94,11 @@ const (
 	// storeVersion is the current slotter layout used for the billy.Database
 	// store.
 	storeVersion = 1
+
+	// conversionTimeWindow defines the period after the Osaka fork during which
+	// the pool will still accept and convert legacy blob transactions. After this
+	// window, all legacy blob transactions will be rejected.
+	conversionTimeWindow = time.Hour * 2
 )
 
 // blobTxMeta is the minimal subset of types.BlobTx necessary to validate and
@@ -333,9 +339,9 @@ type BlobPool struct {
 	chain  BlockChain       // Chain object to access the state through
 	cQueue *conversionQueue // The queue for performing legacy sidecar conversion
 
-	head   *types.Header  // Current head of the chain
+	head   atomic.Pointer[types.Header] // Current head of the chain
 	state  *state.StateDB               // Current state at the head of the chain
-	gasTip *uint256.Int   // Currently accepted minimum gas tip
+	gasTip atomic.Pointer[uint256.Int]  // Currently accepted minimum gas tip
 
 	lookup *lookup                          // Lookup table mapping blobs to txs and txs to billy entries
 	index  map[common.Address][]*blobTxMeta // Blob transactions grouped by accounts, sorted by nonce
@@ -360,7 +366,7 @@ func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bo
 		hasPendingAuth: hasPendingAuth,
 		signer:         types.LatestSigner(chain.Config()),
 		chain:          chain,
-		cQueue:         newConversionQueue(),
+		cQueue:         newConversionQueue(), // Deprecate it after the osaka fork
 		lookup:         newLookup(),
 		index:          make(map[common.Address][]*blobTxMeta),
 		spent:          make(map[common.Address]*uint256.Int),
@@ -402,7 +408,8 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 	if err != nil {
 		return err
 	}
-	p.head, p.state = head, state
+	p.head.Store(head)
+	p.state = state
 
 	// Create new slotter for pre-Osaka blob configuration.
 	slotter := newSlotter(eip4844.LatestMaxBlobsPerBlock(p.chain.Config()))
@@ -442,11 +449,11 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 		p.recheck(addr, nil)
 	}
 	var (
-		basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), p.head))
+		basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), head))
 		blobfee = uint256.NewInt(params.BlobTxMinBlobGasprice)
 	)
-	if p.head.ExcessBlobGas != nil {
+	if head.ExcessBlobGas != nil {
-		blobfee = uint256.MustFromBig(eip4844.CalcBlobFee(p.chain.Config(), p.head))
+		blobfee = uint256.MustFromBig(eip4844.CalcBlobFee(p.chain.Config(), head))
 	}
 	p.evict = newPriceHeap(basefee, blobfee, p.index)
 
@@ -476,8 +483,10 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 
 // Close closes down the underlying persistent store.
 func (p *BlobPool) Close() error {
-	var errs []error
+	// Terminate the conversion queue
 	p.cQueue.close()
+
+	var errs []error
 	if p.limbo != nil { // Close might be invoked due to error in constructor, before p,limbo is set
 		if err := p.limbo.Close(); err != nil {
 			errs = append(errs, err)
@@ -835,7 +844,7 @@ func (p *BlobPool) Reset(oldHead, newHead *types.Header) {
 		log.Error("Failed to reset blobpool state", "err", err)
 		return
 	}
-	p.head = newHead
+	p.head.Store(newHead)
 	p.state = statedb
 
 	// Run the reorg between the old and new head and figure out which accounts
@@ -858,7 +867,7 @@ func (p *BlobPool) Reset(oldHead, newHead *types.Header) {
 		}
 	}
 	// Flush out any blobs from limbo that are older than the latest finality
-	if p.chain.Config().IsCancun(p.head.Number, p.head.Time) {
+	if p.chain.Config().IsCancun(newHead.Number, newHead.Time) {
 		p.limbo.finalize(p.chain.CurrentFinalBlock())
 	}
 	// Reset the price heap for the new set of basefee/blobfee pairs
@@ -1059,14 +1068,15 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
 	defer p.lock.Unlock()
 
 	// Store the new minimum gas tip
-	old := p.gasTip
+	old := p.gasTip.Load()
-	p.gasTip = uint256.MustFromBig(tip)
+	newTip := uint256.MustFromBig(tip)
+	p.gasTip.Store(newTip)
 
 	// If the min miner fee increased, remove transactions below the new threshold
-	if old == nil || p.gasTip.Cmp(old) > 0 {
+	if old == nil || newTip.Cmp(old) > 0 {
 		for addr, txs := range p.index {
 			for i, tx := range txs {
-				if tx.execTipCap.Cmp(p.gasTip) < 0 {
+				if tx.execTipCap.Cmp(newTip) < 0 {
 					// Drop the offending transaction
 					var (
 						ids    = []uint64{tx.id}
@@ -1126,10 +1136,10 @@ func (p *BlobPool) ValidateTxBasics(tx *types.Transaction) error {
 		Config:       p.chain.Config(),
 		Accept:       1 << types.BlobTxType,
 		MaxSize:      txMaxSize,
-		MinTip:       p.gasTip.ToBig(),
+		MinTip:       p.gasTip.Load().ToBig(),
 		MaxBlobCount: maxBlobsPerTx,
 	}
-	return txpool.ValidateTransaction(tx, p.head, p.signer, opts)
+	return txpool.ValidateTransaction(tx, p.head.Load(), p.signer, opts)
 }
 
 // checkDelegationLimit determines if the tx sender is delegated or has a
@@ -1435,34 +1445,51 @@ func (p *BlobPool) AvailableBlobs(vhashes []common.Hash) int {
 	return available
 }
 
-// preprocess performs the static validation upon the provided txs and converts
+// preCheck performs the static validation upon the provided txs and converts
-// the legacy sidecars if Osaka fork has been activated.
+// the legacy sidecars if Osaka fork has been activated with a short time window.
 //
 // This function is pure static and lock free.
-func (p *BlobPool) preprocess(txs []*types.Transaction) ([]*types.Transaction, []error) {
+func (p *BlobPool) preCheck(txs []*types.Transaction) ([]*types.Transaction, []error) {
-	head := p.chain.CurrentBlock()
+	var (
-	if !p.chain.Config().IsOsaka(head.Number, head.Time) {
+		head     = p.head.Load()
-		return txs, make([]error, len(txs))
+		isOsaka  = p.chain.Config().IsOsaka(head.Number, head.Time)
+		deadline time.Time
+	)
+	if isOsaka {
+		deadline = time.Unix(int64(*p.chain.Config().OsakaTime), 0).Add(conversionTimeWindow)
 	}
 	var errs []error
 	for _, tx := range txs {
 		// Validate the transaction statically at first to avoid unnecessary
-		// conversion. This step doesn't require lock.
+		// conversion. This step doesn't require lock protection.
 		if err := p.ValidateTxBasics(tx); err != nil {
 			errs = append(errs, err)
 			continue
 		}
-		// Convert the legacy sidecar after Osaka fork. This could be a long
+		// Before the Osaka fork, reject the blob txs with cell proofs
-		// procedure which takes a few seconds, even minutes. Fortunately it
+		if !isOsaka {
-		// will only block the routine of the source peer without affecting
-		// other parts.
 			if tx.BlobTxSidecar().Version == types.BlobSidecarVersion0 {
-			if err := p.cQueue.convert(tx); err != nil {
+				errs = append(errs, nil)
-				errs = append(errs, err)
+			} else {
+				errs = append(errs, errors.New("cell proof is not supported yet"))
+			}
 			continue
 		}
-		}
+		// After the Osaka fork, reject the legacy blob txs if the conversion
+		// time window is passed.
+		if tx.BlobTxSidecar().Version == types.BlobSidecarVersion1 {
 			errs = append(errs, nil)
+			continue
+		}
+		if time.Now().After(deadline) {
+			errs = append(errs, errors.New("legacy blob tx is not supported"))
+			continue
+		}
+		// Convert the legacy sidecar after Osaka fork. This could be a long
+		// procedure which takes a few seconds, even minutes if there is a long
+		// queue. Fortunately it will only block the routine of the source peer
+		// announcing the tx, without affecting other parts.
+		errs = append(errs, p.cQueue.convert(tx))
 	}
 	return txs, errs
 }
@@ -1474,7 +1501,7 @@ func (p *BlobPool) Add(txs []*types.Transaction, sync bool) []error {
 		errs []error
 		adds = make([]*types.Transaction, 0, len(txs))
 	)
-	txs, errs = p.preprocess(txs)
+	txs, errs = p.preCheck(txs)
 	for i, tx := range txs {
 		if errs[i] != nil {
 			continue
@@ -1973,7 +2000,7 @@ func (p *BlobPool) Clear() {
 	p.spent = make(map[common.Address]*uint256.Int)
 
 	var (
-		basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), p.head))
+		basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), p.head.Load()))
 		blobfee = uint256.NewInt(params.BlobTxMinBlobGasprice)
 	)
 	p.evict = newPriceHeap(basefee, blobfee, p.index)

core/txpool/blobpool/conversion.go

@@ -18,13 +18,22 @@ package blobpool
 
 import (
 	"errors"
+	"sync/atomic"
+	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 )
 
-// cTask represents a conversion task with an attached result channel.
+// maxPendingConversionTasks caps the number of pending conversion tasks. This
+// prevents excessive memory usage; the worst-case scenario (2k transactions
+// with 6 blobs each) would consume approximately 1.5GB of memory.
+const maxPendingConversionTasks = 2048
+
+// cTask represents a conversion task with an attached legacy blob transaction.
 type cTask struct {
-	tx   *types.Transaction // Blob transaction, sidecar is expected
+	tx   *types.Transaction // Legacy blob transaction
 	done chan error         // Channel for signaling back if the conversion succeeds
 }
 
@@ -74,17 +83,25 @@ func (q *conversionQueue) close() {
 	}
 }
 
-func (q *conversionQueue) run(tasks []*cTask, done chan struct{}) {
+// run converts a batch of legacy blob txs to the new cell proof format.
+func (q *conversionQueue) run(tasks []*cTask, done chan struct{}, interrupt *atomic.Int32) {
 	defer close(done)
 
 	for _, t := range tasks {
+		if interrupt != nil && interrupt.Load() != 0 {
+			log.Info("Legacy blob tx conversion is interrupted")
+			return
+		}
 		sidecar := t.tx.BlobTxSidecar()
 		if sidecar == nil {
 			t.done <- errors.New("tx without sidecar")
 			continue
 		}
 		// Run the conversion, the original sidecar will be mutated in place
-		t.done <- sidecar.ToV1()
+		start := time.Now()
+		err := sidecar.ToV1()
+		t.done <- err
+		log.Trace("Converted legacy blob tx", "hash", t.tx.Hash(), "err", err, "elapsed", common.PrettyDuration(time.Since(start)))
 	}
 }
 
@@ -92,24 +109,41 @@ func (q *conversionQueue) loop() {
 	defer close(q.closed)
 
 	var (
-		done   = make(chan struct{})
+		done      chan struct{} // Non-nil if background routine is active
+		interrupt *atomic.Int32 // Flag to signal conversion interruption
+
+		// The pending tasks for sidecar conversion. We assume the number of legacy
+		// blob transactions requiring conversion will not be excessive. However,
+		// a hard cap is applied as a protective measure.
 		cTasks []*cTask
 	)
 	for {
 		select {
 		case t := <-q.tasks:
+			if len(cTasks) >= maxPendingConversionTasks {
+				t.done <- errors.New("conversion queue is overloaded")
+				continue
+			}
 			cTasks = append(cTasks, t)
+
+			// Launch the background conversion thread if it's idle
 			if done == nil {
-				done = make(chan struct{})
+				done, interrupt = make(chan struct{}), new(atomic.Int32)
 
 				tasks := cTasks
 				cTasks = cTasks[:0]
-				go q.run(tasks, done)
+				go q.run(tasks, done, interrupt)
 			}
 		case <-done:
-			done = nil
+			done, interrupt = nil, nil
 
 		case <-q.quit:
+			if done == nil {
+				return
+			}
+			interrupt.Store(1)
+			log.Info("Waiting background converter to exit")
+			<-done
 			return
 		}
 	}

core/txpool/blobpool/conversion_test.go

@@ -0,0 +1,101 @@
+// Copyright 2025 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package blobpool
+
+import (
+	"crypto/ecdsa"
+	"crypto/sha256"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/holiman/uint256"
+)
+
+// createV1BlobTx creates a blob transaction with version 1 sidecar for testing.
+func createV1BlobTx(nonce uint64, key *ecdsa.PrivateKey) *types.Transaction {
+	blob := &kzg4844.Blob{byte(nonce)}
+	commitment, _ := kzg4844.BlobToCommitment(blob)
+	cellProofs, _ := kzg4844.ComputeCellProofs(blob)
+
+	blobtx := &types.BlobTx{
+		ChainID:    uint256.MustFromBig(params.MainnetChainConfig.ChainID),
+		Nonce:      nonce,
+		GasTipCap:  uint256.NewInt(1),
+		GasFeeCap:  uint256.NewInt(1000),
+		Gas:        21000,
+		BlobFeeCap: uint256.NewInt(100),
+		BlobHashes: []common.Hash{kzg4844.CalcBlobHashV1(sha256.New(), &commitment)},
+		Value:      uint256.NewInt(100),
+		Sidecar:    types.NewBlobTxSidecar(types.BlobSidecarVersion1, []kzg4844.Blob{*blob}, []kzg4844.Commitment{commitment}, cellProofs),
+	}
+	return types.MustSignNewTx(key, types.LatestSigner(params.MainnetChainConfig), blobtx)
+}
+
+func TestConversionQueueBasic(t *testing.T) {
+	queue := newConversionQueue()
+	defer queue.close()
+
+	key, _ := crypto.GenerateKey()
+	tx := makeTx(0, 1, 1, 1, key)
+	if err := queue.convert(tx); err != nil {
+		t.Fatalf("Expected successful conversion, got error: %v", err)
+	}
+	if tx.BlobTxSidecar().Version != types.BlobSidecarVersion1 {
+		t.Errorf("Expected sidecar version to be %d, got %d", types.BlobSidecarVersion1, tx.BlobTxSidecar().Version)
+	}
+}
+
+func TestConversionQueueV1BlobTx(t *testing.T) {
+	queue := newConversionQueue()
+	defer queue.close()
+
+	key, _ := crypto.GenerateKey()
+	tx := createV1BlobTx(0, key)
+	version := tx.BlobTxSidecar().Version
+
+	err := queue.convert(tx)
+	if err != nil {
+		t.Fatalf("Expected successful conversion, got error: %v", err)
+	}
+	if tx.BlobTxSidecar().Version != version {
+		t.Errorf("Expected sidecar version to remain %d, got %d", version, tx.BlobTxSidecar().Version)
+	}
+}
+
+func TestConversionQueueClosed(t *testing.T) {
+	queue := newConversionQueue()
+
+	// Close the queue first
+	queue.close()
+	key, _ := crypto.GenerateKey()
+	tx := makeTx(0, 1, 1, 1, key)
+
+	err := queue.convert(tx)
+	if err == nil {
+		t.Fatal("Expected error when converting on closed queue, got nil")
+	}
+}
+
+func TestConversionQueueDoubleClose(t *testing.T) {
+	queue := newConversionQueue()
+	queue.close()
+	queue.close() // Should not panic
+}