diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 55d24c7a93..cd9b1b896c 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/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. diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index e46529a241..da0f1f3f7f 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" @@ -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 := ¶ms.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 }