core/txpool/blobpool: convert sidecars in-place after fork

core/txpool/blobpool/blobpool.go

@@ -344,7 +344,8 @@ type BlobPool struct {
 	discoverFeed event.Feed // Event feed to send out new tx events on pool discovery (reorg excluded)
 	insertFeed   event.Feed // Event feed to send out new tx events on pool inclusion (reorg included)
 
-	lock sync.RWMutex // Mutex protecting the pool during reorg handling
+	closeCh chan struct{} // Channel used by conversion thread to shutdown.
+	lock    sync.RWMutex  // Mutex protecting the pool during reorg handling
 }
 
 // New creates a new blob transaction pool to gather, sort and filter inbound
@@ -474,6 +475,7 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 
 // Close closes down the underlying persistent store.
 func (p *BlobPool) Close() error {
+	close(p.closeCh)
 	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 {
@@ -858,6 +860,20 @@ func (p *BlobPool) Reset(oldHead, newHead *types.Header) {
 	if p.chain.Config().IsCancun(p.head.Number, p.head.Time) {
 		p.limbo.finalize(p.chain.CurrentFinalBlock())
 	}
+
+	// At the Osaka transition, begin converting all v0 sidecars to v1.
+	osaka := p.chain.Config().IsOsaka
+	if osaka(p.head.Number, p.head.Time) && !osaka(oldHead.Number, oldHead.Time) {
+		// Collect all indexed tx metadatas at point of fork. Future transactions
+		// added are required to already be v1.
+		var all []*blobTxMeta
+		for _, m := range p.index {
+			all = append(all, m...)
+		}
+		p.closeCh = make(chan struct{})
+		go p.convertSidecars(all, p.closeCh)
+	}
+
 	// Reset the price heap for the new set of basefee/blobfee pairs
 	var (
 		basefee = uint256.MustFromBig(eip1559.CalcBaseFee(p.chain.Config(), newHead))
@@ -873,6 +889,92 @@ func (p *BlobPool) Reset(oldHead, newHead *types.Header) {
 	p.updateStorageMetrics()
 }
 
+// convertSidecars runs on a separate thread and manages the pool lock as needed
+// during the conversion. Any txs that fail to convert are dropped.
+func (p *BlobPool) convertSidecars(txs []*blobTxMeta, cancel chan struct{}) {
+	convert := func(m *blobTxMeta) error {
+		if m.version != types.BlobSidecarVersion0 {
+			return nil
+		}
+		p.lock.RLock()
+		data, err := p.store.Get(m.id)
+		if err != nil {
+			// This will most likely occur when a tx gets included before it can be
+			// converted locally.
+			p.lock.RUnlock()
+			return fmt.Errorf("unable to load tx for conversion: %w", err)
+		}
+		p.lock.RUnlock()
+
+		tx := new(types.Transaction)
+		if err := rlp.DecodeBytes(data, tx); err != nil {
+			return fmt.Errorf("failed to decode stored blob transaction: %w", err)
+		}
+		// Perform conversion.
+		sidecar := tx.BlobTxSidecar()
+		if sidecar == nil {
+			return fmt.Errorf("missing sidecar")
+		}
+		if err := sidecar.ToV1(); err != nil {
+			return fmt.Errorf("failed to convert sidecar: %w", err)
+		}
+
+		// Reencode and insert into store.
+		blob, err := rlp.EncodeToBytes(tx)
+		if err != nil {
+			return fmt.Errorf("failed to encode new sidecar: %w", err)
+		}
+		p.lock.Lock()
+		defer p.lock.Unlock()
+		id, err := p.store.Put(blob)
+		if err != nil {
+			return fmt.Errorf("failed to write tx with converted sidecar: %w", err)
+		}
+
+		// Update metadata and trackers for the tx.
+		m.id = id
+		m.version = types.BlobSidecarVersion1
+		p.lookup.track(m)
+
+		return nil
+	}
+
+	log.Info("Beginning blob sidecar conversion", "count", len(txs))
+
+	var (
+		start  = time.Now()
+		logged = time.Now()
+	)
+
+	for i, m := range txs {
+		select {
+		case <-cancel:
+			return
+		default:
+		}
+		if err := convert(m); err != nil {
+			// If the conversion fails for any reason, just remove the transaction.
+			// This should be pretty rare and we don't need to retry.
+			log.Error("Failed to convert sidecar", "err", err, "hash", m.hash, "id", m.id)
+			p.lock.Lock()
+			p.store.Delete(m.id)
+			p.lookup.untrack(m)
+			for acc, txs := range p.index {
+				for i, tx := range txs {
+					if tx.hash == m.hash {
+						p.index[acc][i] = nil
+					}
+				}
+			}
+			p.lock.Unlock()
+		}
+		if time.Since(logged) > time.Second*8 {
+			log.Info("Converting blob sidecars", "complete", i, "total", len(txs))
+		}
+	}
+	log.Info("Finished blob sidecar conversion", "elapsed", time.Since(start))
+}
+
 // reorg assembles all the transactors and missing transactions between an old
 // and new head to figure out which account's tx set needs to be rechecked and
 // which transactions need to be requeued.

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"
@@ -43,6 +44,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/internal/testrand"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/holiman/billy"
@@ -2062,6 +2064,94 @@ func TestGetBlobs(t *testing.T) {
 	pool.Close()
 }
 
+func TestSidecarConversion(t *testing.T) {
+	log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true)))
+
+	// Create a temporary folder for the persistent backend
+	storage := t.TempDir()
+
+	os.MkdirAll(filepath.Join(storage, pendingTransactionStore), 0700)
+	// store, _ := billy.Open(billy.Options{Path: filepath.Join(storage, pendingTransactionStore)}, newSlotter(6), nil)
+
+	var (
+		count      = 10
+		keys       = make([]*ecdsa.PrivateKey, count)
+		addrs      = make([]common.Address, count)
+		statedb, _ = state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
+	)
+	for i := 0; i < count; i++ {
+		keys[i], _ = crypto.GenerateKey()
+		addrs[i] = crypto.PubkeyToAddress(keys[i].PublicKey)
+		statedb.AddBalance(addrs[i], uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified)
+	}
+	statedb.Commit(0, true, false)
+
+	config := &params.ChainConfig{
+		ChainID:            big.NewInt(1),
+		LondonBlock:        big.NewInt(0),
+		BerlinBlock:        big.NewInt(0),
+		CancunTime:         newUint64(0),
+		PragueTime:         newUint64(0),
+		OsakaTime:          newUint64(1),
+		BlobScheduleConfig: params.DefaultBlobSchedule,
+	}
+	chain := &testBlockChain{
+		config:  config,
+		basefee: uint256.NewInt(1050),
+		blobfee: uint256.NewInt(105),
+		statedb: statedb,
+	}
+
+	before := chain.CurrentBlock()
+	before.Time = 0
+	after := chain.CurrentBlock()
+	after.Time = 1
+
+	pool := New(Config{Datadir: storage}, chain, nil)
+	if err := pool.Init(1, before, newReserver()); err != nil {
+		t.Fatalf("failed to create blob pool: %v", err)
+	}
+
+	for i, key := range keys {
+		tx := makeMultiBlobTx(0, 1, 1000, 100, 2, 0, key, types.BlobSidecarVersion0)
+		if errs := pool.Add([]*types.Transaction{tx}, true); errs[0] != nil {
+			t.Errorf("failed to insert blob tx from %s: %s", addrs[i], errs[0])
+		}
+	}
+
+	// Should kick off migration.
+	pool.Reset(before, after)
+
+	time.Sleep(time.Second * 2)
+
+	for addr, acc := range pool.index {
+		for _, m := range acc {
+			if m.version != types.BlobSidecarVersion1 {
+				t.Errorf("expected sidecar to have been converted: from %s, hash %s", addr, m.hash)
+			}
+			fmt.Println("loading for test", "id", m.id)
+			tx := pool.Get(m.hash)
+			if tx == nil {
+				t.Errorf("failed to get tx by hash: %s", m.hash)
+			}
+			sc := tx.BlobTxSidecar()
+			fmt.Println(len(sc.Blobs), len(sc.Commitments), len(sc.Proofs))
+			if err := kzg4844.VerifyCellProofs(sc.Blobs, sc.Commitments, sc.Proofs); err != nil {
+				t.Errorf("failed to verify cell proofs for tx %s after conversion: %s", m.hash, err)
+			}
+		}
+	}
+
+	// Verify all the calculated pool internals. Interestingly, this is **not**
+	// a duplication of the above checks, this actually validates the verifier
+	// using the above already hard coded checks.
+	//
+	// Do not remove this, nor alter the above to be generic.
+	verifyPoolInternals(t, pool)
+
+	pool.Close()
+}
+
 // fakeBilly is a billy.Database implementation which just drops data on the floor.
 type fakeBilly struct {
 	billy.Database
@@ -2144,3 +2234,5 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) {
 		}
 	}
 }
+
+func newUint64(val uint64) *uint64 { return &val }