diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index b918925bd2..2a2e1080b0 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -543,6 +543,8 @@ type BlobPool struct { stored uint64 // Useful data size of all transactions on disk limbo *limbo // Persistent data store for the non-finalized blobs + cQueue *conversionQueue + gapped map[common.Address][]*BlobTxForPool // Transactions that are currently gapped (nonce too high) gappedSource map[common.Hash]common.Address // Source of gapped transactions to allow rechecking on inclusion @@ -642,11 +644,14 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser ) index := func(id uint64, size uint32, blob []byte) { err := p.parseTransaction(id, size, blob) - if err != nil { - toDelete = append(toDelete, id) - } + // Transactions in legacy format will be queued for cell computation. + // This entry will be swapped from the store after the conversion. if errors.Is(err, errLegacyTx) { convertTxs = append(convertTxs, id) + return + } + if err != nil { + toDelete = append(toDelete, id) } } store, err := billy.Open(billy.Options{Path: queuedir, Repair: true}, slotter, index) @@ -655,52 +660,7 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser } p.store = store - // Migrate legacy transactions (types.Transaction) to pooledBlobTx format. - if len(convertTxs) > 0 { - for _, id := range convertTxs { - var tx types.Transaction - data, err := p.store.Get(id) - if err != nil { - continue - } - err = rlp.DecodeBytes(data, &tx) - if err != nil { - continue - } - if tx.BlobTxSidecar() == nil { - continue - } - ptx, err := newBlobTxForPool(&tx) - // Note that we skip errors here. - // Just like parseTransaction failure does not abort the blobpool creation, - // conversion process also cannot abort the entire process. - if err != nil { - log.Error("Failed to convert legacy tx to pooledBlobTx", "hash", tx.Hash(), "err", err) - continue - } - blob, err := rlp.EncodeToBytes(ptx) - if err != nil { - continue - } - id, err := p.store.Put(blob) - if err != nil { - continue - } - meta := newBlobTxMeta(id, p.store.Size(id), ptx) - - // If the newly inserted transaction fails to be tracked, - // it should also be removed with those in `toDelete` - sender, err := types.Sender(p.signer, ptx.Tx) - if err != nil { - toDelete = append(toDelete, id) - continue - } - if err := p.trackTransaction(meta, sender); err != nil { - toDelete = append(toDelete, id) - continue - } - } - } + p.cQueue = newConversionQueue() if len(toDelete) > 0 { log.Warn("Dropping invalidated blob transactions", "ids", toDelete) @@ -744,7 +704,8 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser // Pool initialized, attach the blob limbo to it to track blobs included // recently but not yet finalized - p.limbo, err = newLimbo(p.chain.Config(), limbodir) + var convertLimbo []uint64 + p.limbo, convertLimbo, err = newLimbo(p.chain.Config(), limbodir) if err != nil { p.Close() return err @@ -763,12 +724,22 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser // Update the metrics and return the constructed pool datacapGauge.Update(int64(p.config.Datacap)) p.updateStorageMetrics() + + if len(convertTxs) > 0 { + p.cQueue.launchConversion(func() { p.convertLegacyTxs(convertTxs) }) + } + if len(convertLimbo) > 0 { + p.cQueue.launchConversion(func() { p.convertLegacyLimbo(convertLimbo) }) + } return nil } // Close closes down the underlying persistent store. func (p *BlobPool) Close() error { var errs []error + if p.cQueue != nil { + p.cQueue.close() + } 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) @@ -817,6 +788,91 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error { return p.trackTransaction(meta, sender) } +func (p *BlobPool) convertLegacyTxs(ids []uint64) { + start := time.Now() + var converted, discarded int + + for _, id := range ids { + p.lock.Lock() + data, err := p.store.Get(id) + if err != nil { + p.lock.Unlock() + continue + } + if derr := p.store.Delete(id); derr != nil { + log.Error("Failed to delete legacy blob tx", "id", id, "err", derr) + p.lock.Unlock() + continue + } + p.lock.Unlock() + var tx types.Transaction + if err := rlp.DecodeBytes(data, &tx); err != nil { + log.Error("Failed to decode legacy blob tx", "id", id, "err", err) + continue + } + ptx, err := newBlobTxForPool(&tx) + if err != nil { + log.Error("Failed to convert legacy blob tx", "hash", tx.Hash(), "err", err) + continue + } + if err := p.AddPooledTx(ptx); err != nil { + log.Debug("Discarded converted blob tx", "hash", tx.Hash(), "err", err) + discarded++ + } else { + converted++ + } + } + + log.Info("Completed blob transaction conversion", "converted", converted, "discarded", discarded, "elapsed", common.PrettyDuration(time.Since(start))) +} + +func (p *BlobPool) convertLegacyLimbo(ids []uint64) { + start := time.Now() + var converted, discarded int + + for _, id := range ids { + p.lock.Lock() + data, err := p.limbo.store.Get(id) + if err != nil { + p.lock.Unlock() + continue + } + if derr := p.limbo.store.Delete(id); derr != nil { + log.Error("Failed to delete legacy blob tx", "id", id, "err", derr) + p.lock.Unlock() + continue + } + p.lock.Unlock() + var legacy struct { + TxHash common.Hash + Block uint64 + Tx *types.Transaction + } + if err := rlp.DecodeBytes(data, &legacy); err != nil { + log.Error("Failed to decode legacy limbo entry", "id", id, "err", err) + continue + } + if legacy.Tx == nil || legacy.Tx.BlobTxSidecar() == nil { + continue + } + ptx, err := newBlobTxForPool(legacy.Tx) + if err != nil { + log.Error("Failed to convert legacy limbo entry", "hash", legacy.TxHash, "err", err) + continue + } + p.lock.Lock() + if err := p.limbo.setAndIndex(ptx, legacy.Block); err != nil { + log.Error("Failed to re-store converted limbo entry", "hash", legacy.TxHash, "err", err) + discarded++ + } else { + converted++ + } + p.lock.Unlock() + } + + log.Info("Completed limbo blob conversion", "converted", converted, "discarded", discarded, "elapsed", common.PrettyDuration(time.Since(start))) +} + // trackTransaction registers a transaction's metadata in the pool's indices. func (p *BlobPool) trackTransaction(meta *blobTxMeta, sender common.Address) error { if p.lookup.exists(meta.hash) { @@ -1904,7 +1960,7 @@ func (p *BlobPool) Add(txs []*types.Transaction, sync bool) []error { if errs[i] = p.ValidateTxBasics(tx); errs[i] != nil { continue } - ptx, err := newBlobTxForPool(tx) + ptx, err := p.cQueue.convert(tx) if err != nil { errs[i] = err continue diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 394b9e31d4..4e5f4f5e5a 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -31,6 +31,7 @@ import ( "slices" "sync" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/misc/eip1559" @@ -1334,6 +1335,14 @@ func TestLegacyTxConversion(t *testing.T) { } defer pool.Close() + deadline := time.Now().Add(30 * time.Second) + for pool.Get(tx1.Hash()) == nil || pool.Get(tx2.Hash()) == nil { + if time.Now().After(deadline) { + t.Fatalf("legacy txs were not converted in time") + } + time.Sleep(10 * time.Millisecond) + } + // Both transactions should be retrievable. for _, want := range []*types.Transaction{tx1, tx2} { got := pool.Get(want.Hash()) @@ -1359,6 +1368,75 @@ func TestLegacyTxConversion(t *testing.T) { verifyPoolInternals(t, pool) } +func TestLegacyLimboConversion(t *testing.T) { + storage := t.TempDir() + os.MkdirAll(filepath.Join(storage, pendingTransactionStore), 0700) + limbodir := filepath.Join(storage, limboedTransactionStore) + os.MkdirAll(limbodir, 0700) + + key, _ := crypto.GenerateKey() + tx := makeMultiBlobTx(0, 1, 1000, 100, 2, 0, key) + + store, err := billy.Open(billy.Options{Path: limbodir}, newSlotterEIP7594(testMaxBlobsPerBlock), nil) + if err != nil { + t.Fatalf("failed to open limbo billy: %v", err) + } + data, err := rlp.EncodeToBytes(&struct { + TxHash common.Hash + Block uint64 + Tx *types.Transaction + }{TxHash: tx.Hash(), Block: 42, Tx: tx}) + if err != nil { + t.Fatalf("failed to legacy-encode limbo entry: %v", err) + } + if _, err := store.Put(data); err != nil { + t.Fatalf("failed to put legacy limbo entry: %v", err) + } + store.Close() + + statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) + statedb.Commit(0, true, false) + chain := &testBlockChain{ + config: params.MainnetChainConfig, + basefee: uint256.NewInt(params.InitialBaseFee), + blobfee: uint256.NewInt(params.BlobTxMinBlobGasprice), + statedb: statedb, + } + pool := New(Config{Datadir: storage}, chain, nil) + if err := pool.Init(1, chain.CurrentBlock(), newReserver()); err != nil { + t.Fatalf("failed to create blob pool: %v", err) + } + defer pool.Close() + + deadline := time.Now().Add(30 * time.Second) + for { + pool.lock.RLock() + _, ok := pool.limbo.index[tx.Hash()] + pool.lock.RUnlock() + if ok { + break + } + if time.Now().After(deadline) { + t.Fatalf("legacy limbo entry was not converted in time") + } + time.Sleep(10 * time.Millisecond) + } + + pool.lock.Lock() + ptx, err := pool.limbo.pull(tx.Hash()) + pool.lock.Unlock() + if err != nil { + t.Fatalf("failed to pull converted limbo entry: %v", err) + } + full, err := ptx.toTx() + if err != nil { + t.Fatalf("failed to reconstruct tx from converted limbo entry: %v", err) + } + if full.Hash() != tx.Hash() { + t.Fatalf("converted limbo tx hash mismatch: have %s, want %s", full.Hash(), tx.Hash()) + } +} + // TestBlobCountLimit tests the blobpool enforced limits on the max blob count. func TestBlobCountLimit(t *testing.T) { var ( diff --git a/core/txpool/blobpool/conversion.go b/core/txpool/blobpool/conversion.go new file mode 100644 index 0000000000..714975cce3 --- /dev/null +++ b/core/txpool/blobpool/conversion.go @@ -0,0 +1,207 @@ +// Copyright 2026 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 ( + "errors" + "slices" + "sync/atomic" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/log" +) + +// 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 + +type convertResult struct { + ptx *BlobTxForPool + err error +} + +// txConvert represents a conversion task with an attached legacy blob transaction. +type txConvert struct { + tx *types.Transaction // Legacy blob transaction + done chan convertResult // Channel for signaling back if the conversion succeeds +} + +// conversionQueue is a dedicated queue for converting legacy blob transactions +// received from the network after the Osaka fork. Since conversion is expensive, +// it is performed in the background by a single thread, ensuring the main Geth +// process is not overloaded. +type conversionQueue struct { + tasks chan *txConvert + startConversion chan func() + quit chan struct{} + closed chan struct{} + + queue []func() + taskDone chan struct{} +} + +// newConversionQueue constructs the conversion queue. +func newConversionQueue() *conversionQueue { + q := &conversionQueue{ + tasks: make(chan *txConvert), + startConversion: make(chan func()), + quit: make(chan struct{}), + closed: make(chan struct{}), + } + go q.loop() + return q +} + +// convert accepts a legacy blob transaction with version-0 blobs and queues it +// for conversion. +// +// This function may block for a long time until the transaction is processed. +func (q *conversionQueue) convert(tx *types.Transaction) (*BlobTxForPool, error) { + done := make(chan convertResult, 1) + select { + case q.tasks <- &txConvert{tx: tx, done: done}: + res := <-done + return res.ptx, res.err + case <-q.closed: + return nil, errors.New("conversion queue closed") + } +} + +// launchConversion starts a conversion task in the background. +func (q *conversionQueue) launchConversion(fn func()) error { + select { + case q.startConversion <- fn: + return nil + case <-q.closed: + return errors.New("conversion queue closed") + } +} + +// close terminates the conversion queue. +func (q *conversionQueue) close() { + select { + case <-q.closed: + return + default: + close(q.quit) + <-q.closed + } +} + +// run converts a batch of legacy blob txs to the new cell proof format. +func (q *conversionQueue) run(tasks []*txConvert, done chan struct{}, interrupt *atomic.Int32) { + defer close(done) + + for _, t := range tasks { + if interrupt != nil && interrupt.Load() != 0 { + t.done <- convertResult{err: errors.New("conversion is interrupted")} + continue + } + // Run the conversion, the original sidecar will be mutated in place + start := time.Now() + ptx, err := newBlobTxForPool(t.tx) + t.done <- convertResult{ptx: ptx, err: err} + log.Trace("Converted legacy blob tx", "hash", t.tx.Hash(), "err", err, "elapsed", common.PrettyDuration(time.Since(start))) + } +} + +func (q *conversionQueue) loop() { + defer close(q.closed) + + var ( + 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. + txTasks []*txConvert + ) + + for { + select { + case t := <-q.tasks: + if len(txTasks) >= maxPendingConversionTasks { + t.done <- convertResult{err: errors.New("conversion queue is overloaded")} + continue + } + txTasks = append(txTasks, t) + + // Launch the background conversion thread if it's idle + if done == nil { + done, interrupt = make(chan struct{}), new(atomic.Int32) + + tasks := slices.Clone(txTasks) + txTasks = txTasks[:0] + go q.run(tasks, done, interrupt) + } + + case <-done: + done, interrupt = nil, nil + if len(txTasks) > 0 { + done, interrupt = make(chan struct{}), new(atomic.Int32) + tasks := slices.Clone(txTasks) + txTasks = txTasks[:0] + go q.run(tasks, done, interrupt) + } + + case fn := <-q.startConversion: + q.queue = append(q.queue, fn) + q.runNextTask() + + case <-q.taskDone: + q.runNextTask() + + case <-q.quit: + if done != nil { + log.Debug("Waiting for blob proof conversion to exit") + interrupt.Store(1) + <-done + } + if q.taskDone != nil { + log.Debug("Waiting for blobpool billy conversion to exit") + <-q.taskDone + } + // Signal any tasks that were queued for the next batch but never started + // so callers blocked in convert() receive an error instead of hanging. + for _, t := range txTasks { + // Best-effort notify; t.done is a buffered channel of size 1 + // created by convert(), and we send exactly once per task. + t.done <- convertResult{err: errors.New("conversion queue closed")} + } + // Drop references to allow GC of the backing array. + txTasks = txTasks[:0] + return + } + } +} + +func (q *conversionQueue) runNextTask() { + if len(q.queue) == 0 { + q.taskDone = nil + return + } + fn := q.queue[0] + q.queue = append(q.queue[:0], q.queue[1:]...) + + done := make(chan struct{}) + go func() { defer close(done); fn() }() + q.taskDone = done +} diff --git a/core/txpool/blobpool/conversion_test.go b/core/txpool/blobpool/conversion_test.go new file mode 100644 index 0000000000..b52239641e --- /dev/null +++ b/core/txpool/blobpool/conversion_test.go @@ -0,0 +1,114 @@ +// Copyright 2026 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 ( + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/params" +) + +func TestConversionQueueBasic(t *testing.T) { + queue := newConversionQueue() + defer queue.close() + + key, _ := crypto.GenerateKey() + tx := makeTx(0, 1, 1, 1, key) + + ptx, err := queue.convert(tx) + if err != nil { + t.Fatalf("Expected successful conversion, got error: %v", err) + } + if ptx == nil { + t.Fatal("Expected a converted transaction, got nil") + } + if ptx.Tx.Hash() != tx.Hash() { + t.Errorf("Converted tx hash mismatch: have %s, want %s", ptx.Tx.Hash(), tx.Hash()) + } + if len(ptx.CellSidecar.Cells) == 0 { + t.Error("Expected cells to be computed during conversion") + } +} + +func TestConversionQueueClosed(t *testing.T) { + queue := newConversionQueue() + queue.close() + + key, _ := crypto.GenerateKey() + tx := makeTx(0, 1, 1, 1, key) + + if _, err := queue.convert(tx); 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 +} + +func TestConversionQueueAutoRestartBatch(t *testing.T) { + queue := newConversionQueue() + defer queue.close() + + key, _ := crypto.GenerateKey() + + // Create a heavy transaction to ensure the first batch runs long enough + // for subsequent tasks to be queued while it is active. + heavy := makeMultiBlobTx(0, 1, 1, 1, int(params.BlobTxMaxBlobs), 0, key) + + var wg sync.WaitGroup + wg.Add(1) + heavyDone := make(chan error, 1) + go func() { + defer wg.Done() + _, err := queue.convert(heavy) + heavyDone <- err + }() + + // Give the conversion worker a head start so that the following tasks are + // enqueued while the first batch is running. + time.Sleep(200 * time.Millisecond) + + tx1 := makeTx(1, 1, 1, 1, key) + tx2 := makeTx(2, 1, 1, 1, key) + + wg.Add(2) + done1 := make(chan error, 1) + done2 := make(chan error, 1) + go func() { defer wg.Done(); _, err := queue.convert(tx1); done1 <- err }() + go func() { defer wg.Done(); _, err := queue.convert(tx2); done2 <- err }() + + for _, c := range []struct { + name string + done chan error + }{{"tx1", done1}, {"tx2", done2}, {"heavy", heavyDone}} { + select { + case err := <-c.done: + if err != nil { + t.Fatalf("%s conversion error: %v", c.name, err) + } + case <-time.After(30 * time.Second): + t.Fatalf("timeout waiting for %s conversion", c.name) + } + } + wg.Wait() +} diff --git a/core/txpool/blobpool/limbo.go b/core/txpool/blobpool/limbo.go index dc091132bb..8e44f51fdd 100644 --- a/core/txpool/blobpool/limbo.go +++ b/core/txpool/blobpool/limbo.go @@ -49,7 +49,7 @@ type limbo struct { } // newLimbo opens and indexes a set of limboed blob transactions. -func newLimbo(config *params.ChainConfig, datadir string) (*limbo, error) { +func newLimbo(config *params.ChainConfig, datadir string) (*limbo, []uint64, error) { l := &limbo{ index: make(map[common.Hash]uint64), groups: make(map[uint64]map[uint64]common.Hash), @@ -61,19 +61,27 @@ func newLimbo(config *params.ChainConfig, datadir string) (*limbo, error) { // See if we need to migrate the limbo after fusaka. slotter, err := tryMigrate(config, slotter, datadir) if err != nil { - return nil, err + return nil, nil, err } // Index all limboed blobs on disk and delete anything unprocessable - var fails []uint64 + var ( + fails []uint64 + convert []uint64 + ) index := func(id uint64, size uint32, data []byte) { - if l.parseBlob(id, data) != nil { + err := l.parseBlob(id, data) + if errors.Is(err, errLegacyTx) { + convert = append(convert, id) + return + } + if err != nil { fails = append(fails, id) } } store, err := billy.Open(billy.Options{Path: datadir, Repair: true}, slotter, index) if err != nil { - return nil, err + return nil, nil, err } l.store = store @@ -82,11 +90,11 @@ func newLimbo(config *params.ChainConfig, datadir string) (*limbo, error) { for _, id := range fails { if err := l.store.Delete(id); err != nil { l.Close() - return nil, err + return nil, nil, err } } } - return l, nil + return l, convert, nil } // Close closes down the underlying persistent store. @@ -99,6 +107,9 @@ func (l *limbo) Close() error { func (l *limbo) parseBlob(id uint64, data []byte) error { item := new(limboBlob) if err := rlp.DecodeBytes(data, item); err != nil { + if isLegacyLimboBlob(data) { + return errLegacyTx + } // This path is impossible unless the disk data representation changes // across restarts. For that ever improbable case, recover gracefully // by ignoring this data entry. @@ -122,6 +133,17 @@ func (l *limbo) parseBlob(id uint64, data []byte) error { return nil } +// isLegacyLimboBlob returns true if the data is encoded in legacy limboBlob type. +// It checks whether the first byte of third element is blobTxType. +func isLegacyLimboBlob(data []byte) bool { + elems, err := rlp.SplitListValues(data) + if err != nil || len(elems) < 3 { + return false + } + kind, content, _, err := rlp.Split(elems[2]) + return err == nil && kind == rlp.String && len(content) > 1 && content[0] == types.BlobTxType +} + // finalize evicts all blobs belonging to a recently finalized block or older. func (l *limbo) finalize(final *types.Header) { // Just in case there's no final block yet (network not yet merged, weird