diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 1ac4b9d2c3..6a5b7a759d 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/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 - state *state.StateDB // Current state at the head of the chain - gasTip *uint256.Int // Currently accepted minimum gas tip + head atomic.Pointer[types.Header] // Current head of the chain + state *state.StateDB // Current state at the head of the chain + 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 { - blobfee = uint256.MustFromBig(eip4844.CalcBlobFee(p.chain.Config(), p.head)) + if head.ExcessBlobGas != nil { + 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 - p.gasTip = uint256.MustFromBig(tip) + old := p.gasTip.Load() + 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 -// the legacy sidecars if Osaka fork has been activated. +// preCheck performs the static validation upon the provided txs and converts +// 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) { - head := p.chain.CurrentBlock() - if !p.chain.Config().IsOsaka(head.Number, head.Time) { - return txs, make([]error, len(txs)) +func (p *BlobPool) preCheck(txs []*types.Transaction) ([]*types.Transaction, []error) { + var ( + head = p.head.Load() + 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 - // procedure which takes a few seconds, even minutes. Fortunately it - // 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, err) - continue + // Before the Osaka fork, reject the blob txs with cell proofs + if !isOsaka { + if tx.BlobTxSidecar().Version == types.BlobSidecarVersion0 { + errs = append(errs, nil) + } else { + errs = append(errs, errors.New("cell proof is not supported yet")) } + continue } - errs = append(errs, nil) + // 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) diff --git a/core/txpool/blobpool/conversion.go b/core/txpool/blobpool/conversion.go index 8b257e2014..d759c8ba4f 100644 --- a/core/txpool/blobpool/conversion.go +++ b/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 } } diff --git a/core/txpool/blobpool/conversion_test.go b/core/txpool/blobpool/conversion_test.go new file mode 100644 index 0000000000..a9fd26dbaf --- /dev/null +++ b/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 . + +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 +}