mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-21 20:26:41 +00:00
core/txpool: add conversion queue
This commit is contained in:
parent
91b643be83
commit
1b50f1a275
5 changed files with 535 additions and 58 deletions
|
|
@ -543,6 +543,8 @@ type BlobPool struct {
|
||||||
stored uint64 // Useful data size of all transactions on disk
|
stored uint64 // Useful data size of all transactions on disk
|
||||||
limbo *limbo // Persistent data store for the non-finalized blobs
|
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)
|
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
|
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) {
|
index := func(id uint64, size uint32, blob []byte) {
|
||||||
err := p.parseTransaction(id, size, blob)
|
err := p.parseTransaction(id, size, blob)
|
||||||
if err != nil {
|
// Transactions in legacy format will be queued for cell computation.
|
||||||
toDelete = append(toDelete, id)
|
// This entry will be swapped from the store after the conversion.
|
||||||
}
|
|
||||||
if errors.Is(err, errLegacyTx) {
|
if errors.Is(err, errLegacyTx) {
|
||||||
convertTxs = append(convertTxs, id)
|
convertTxs = append(convertTxs, id)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
toDelete = append(toDelete, id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
store, err := billy.Open(billy.Options{Path: queuedir, Repair: true}, slotter, index)
|
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
|
p.store = store
|
||||||
|
|
||||||
// Migrate legacy transactions (types.Transaction) to pooledBlobTx format.
|
p.cQueue = newConversionQueue()
|
||||||
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
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(toDelete) > 0 {
|
if len(toDelete) > 0 {
|
||||||
log.Warn("Dropping invalidated blob transactions", "ids", toDelete)
|
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
|
// Pool initialized, attach the blob limbo to it to track blobs included
|
||||||
// recently but not yet finalized
|
// 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 {
|
if err != nil {
|
||||||
p.Close()
|
p.Close()
|
||||||
return err
|
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
|
// Update the metrics and return the constructed pool
|
||||||
datacapGauge.Update(int64(p.config.Datacap))
|
datacapGauge.Update(int64(p.config.Datacap))
|
||||||
p.updateStorageMetrics()
|
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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Close closes down the underlying persistent store.
|
// Close closes down the underlying persistent store.
|
||||||
func (p *BlobPool) Close() error {
|
func (p *BlobPool) Close() error {
|
||||||
var errs []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 p.limbo != nil { // Close might be invoked due to error in constructor, before p,limbo is set
|
||||||
if err := p.limbo.Close(); err != nil {
|
if err := p.limbo.Close(); err != nil {
|
||||||
errs = append(errs, err)
|
errs = append(errs, err)
|
||||||
|
|
@ -817,6 +788,91 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
|
||||||
return p.trackTransaction(meta, sender)
|
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.
|
// trackTransaction registers a transaction's metadata in the pool's indices.
|
||||||
func (p *BlobPool) trackTransaction(meta *blobTxMeta, sender common.Address) error {
|
func (p *BlobPool) trackTransaction(meta *blobTxMeta, sender common.Address) error {
|
||||||
if p.lookup.exists(meta.hash) {
|
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 {
|
if errs[i] = p.ValidateTxBasics(tx); errs[i] != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
ptx, err := newBlobTxForPool(tx)
|
ptx, err := p.cQueue.convert(tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errs[i] = err
|
errs[i] = err
|
||||||
continue
|
continue
|
||||||
|
|
|
||||||
|
|
@ -31,6 +31,7 @@ import (
|
||||||
"slices"
|
"slices"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/common"
|
"github.com/ethereum/go-ethereum/common"
|
||||||
"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
|
"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
|
||||||
|
|
@ -1334,6 +1335,14 @@ func TestLegacyTxConversion(t *testing.T) {
|
||||||
}
|
}
|
||||||
defer pool.Close()
|
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.
|
// Both transactions should be retrievable.
|
||||||
for _, want := range []*types.Transaction{tx1, tx2} {
|
for _, want := range []*types.Transaction{tx1, tx2} {
|
||||||
got := pool.Get(want.Hash())
|
got := pool.Get(want.Hash())
|
||||||
|
|
@ -1359,6 +1368,75 @@ func TestLegacyTxConversion(t *testing.T) {
|
||||||
verifyPoolInternals(t, pool)
|
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.
|
// TestBlobCountLimit tests the blobpool enforced limits on the max blob count.
|
||||||
func TestBlobCountLimit(t *testing.T) {
|
func TestBlobCountLimit(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
|
|
|
||||||
207
core/txpool/blobpool/conversion.go
Normal file
207
core/txpool/blobpool/conversion.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
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
|
||||||
|
}
|
||||||
114
core/txpool/blobpool/conversion_test.go
Normal file
114
core/txpool/blobpool/conversion_test.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
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()
|
||||||
|
}
|
||||||
|
|
@ -49,7 +49,7 @@ type limbo struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
// newLimbo opens and indexes a set of limboed blob transactions.
|
// 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{
|
l := &limbo{
|
||||||
index: make(map[common.Hash]uint64),
|
index: make(map[common.Hash]uint64),
|
||||||
groups: make(map[uint64]map[uint64]common.Hash),
|
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.
|
// See if we need to migrate the limbo after fusaka.
|
||||||
slotter, err := tryMigrate(config, slotter, datadir)
|
slotter, err := tryMigrate(config, slotter, datadir)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Index all limboed blobs on disk and delete anything unprocessable
|
// 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) {
|
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)
|
fails = append(fails, id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
store, err := billy.Open(billy.Options{Path: datadir, Repair: true}, slotter, index)
|
store, err := billy.Open(billy.Options{Path: datadir, Repair: true}, slotter, index)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
l.store = store
|
l.store = store
|
||||||
|
|
||||||
|
|
@ -82,11 +90,11 @@ func newLimbo(config *params.ChainConfig, datadir string) (*limbo, error) {
|
||||||
for _, id := range fails {
|
for _, id := range fails {
|
||||||
if err := l.store.Delete(id); err != nil {
|
if err := l.store.Delete(id); err != nil {
|
||||||
l.Close()
|
l.Close()
|
||||||
return nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return l, nil
|
return l, convert, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Close closes down the underlying persistent store.
|
// 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 {
|
func (l *limbo) parseBlob(id uint64, data []byte) error {
|
||||||
item := new(limboBlob)
|
item := new(limboBlob)
|
||||||
if err := rlp.DecodeBytes(data, item); err != nil {
|
if err := rlp.DecodeBytes(data, item); err != nil {
|
||||||
|
if isLegacyLimboBlob(data) {
|
||||||
|
return errLegacyTx
|
||||||
|
}
|
||||||
// This path is impossible unless the disk data representation changes
|
// This path is impossible unless the disk data representation changes
|
||||||
// across restarts. For that ever improbable case, recover gracefully
|
// across restarts. For that ever improbable case, recover gracefully
|
||||||
// by ignoring this data entry.
|
// by ignoring this data entry.
|
||||||
|
|
@ -122,6 +133,17 @@ func (l *limbo) parseBlob(id uint64, data []byte) error {
|
||||||
return nil
|
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.
|
// finalize evicts all blobs belonging to a recently finalized block or older.
|
||||||
func (l *limbo) finalize(final *types.Header) {
|
func (l *limbo) finalize(final *types.Header) {
|
||||||
// Just in case there's no final block yet (network not yet merged, weird
|
// Just in case there's no final block yet (network not yet merged, weird
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue