diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 4a0296dab9..699bb9fb65 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -358,6 +358,35 @@ func (p *BlobPool) Filter(tx *types.Transaction) bool { return tx.Type() == types.BlobTxType } +// migrateBilly migrates the blob data from one billy instance to another one +// with a different slotter. +func migrateBilly(path string, maxBlobs int) error { + oldSlotter := newSlotter(maxBlobs) + newSlotter := newSlotterEIP7594(maxBlobs) + // Open the new billy instance + newStore, err := billy.Open(billy.Options{Path: path, Repair: true}, newSlotter, nil) + if err != nil { + return err + } + // Iterate over the old billy instance and copy the data to the new instance + var indexingErr error + index := func(key uint64, size uint32, data []byte) { + _, indexingErr = newStore.Put(data) + } + oldStore, err := billy.Open(billy.Options{Path: path, Repair: true}, oldSlotter, index) + if err != nil { + return err + } + if indexingErr != nil { + return indexingErr + } + // TODO (MariusVanDerWijden) + // remove the old shelves here + oldStore.Close() + newStore.Close() + return nil +} + // Init sets the gas price needed to keep a transaction in the pool and the chain // head to allow balance / nonce checks. The transaction journal will be loaded // from disk and filtered based on the provided starting settings. @@ -390,6 +419,19 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser } p.head, p.state = head, state + // Check if we need to migrate the blob database to a new format + slotter := newSlotter(eip4844.LatestMaxBlobsPerBlock(p.chain.Config())) + if p.chain.Config().IsOsaka(head.Number, head.Time) { + // Check if we need to migrate our blob db to the new slotter + oldSlotSize := 135168 + if os.Stat(filepath.Join(queuedir, fmt.Sprintf("bkt_%08d.bag", oldSlotSize))); err == nil { + if err := migrateBilly(queuedir, eip4844.LatestMaxBlobsPerBlock(p.chain.Config())); err != nil { + return err + } + } + slotter = newSlotterEIP7594(eip4844.LatestMaxBlobsPerBlock(p.chain.Config())) + } + // Index all transactions on disk and delete anything unprocessable var fails []uint64 index := func(id uint64, size uint32, blob []byte) { @@ -397,7 +439,6 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser fails = append(fails, id) } } - slotter := newSlotter(eip4844.LatestMaxBlobsPerBlock(p.chain.Config())) store, err := billy.Open(billy.Options{Path: queuedir, Repair: true}, slotter, index) if err != nil { return err diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 66c3ee2f45..d3c99d4b0e 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -1165,6 +1165,114 @@ func TestChangingSlotterSize(t *testing.T) { } } +// TestBillyMigration tests the billy migration from the default slotter to +// the PeerDAS slotter. This tests both the migration of the slotter +// as well as increasing the slotter size of the new slotter. +func TestBillyMigration(t *testing.T) { + //log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelTrace, true))) + + // Create a temporary folder for the persistent backend + storage := t.TempDir() + + os.MkdirAll(filepath.Join(storage, pendingTransactionStore), 0700) + // Create the billy with the old slotter + oldSlotter := newSlotter(6) + store, _ := billy.Open(billy.Options{Path: filepath.Join(storage, pendingTransactionStore)}, oldSlotter, nil) + + // Create transactions from a few accounts. + var ( + key1, _ = crypto.GenerateKey() + key2, _ = crypto.GenerateKey() + key3, _ = crypto.GenerateKey() + + addr1 = crypto.PubkeyToAddress(key1.PublicKey) + addr2 = crypto.PubkeyToAddress(key2.PublicKey) + addr3 = crypto.PubkeyToAddress(key3.PublicKey) + + tx1 = makeMultiBlobTx(0, 1, 1000, 100, 6, key1) + tx2 = makeMultiBlobTx(0, 1, 800, 70, 6, key2) + tx3 = makeMultiBlobTx(0, 1, 800, 110, 24, key3) + + blob1, _ = rlp.EncodeToBytes(tx1) + blob2, _ = rlp.EncodeToBytes(tx2) + ) + + // Write the two safely sized txs to store. note: although the store is + // configured for a blob count of 6, it can also support around ~1mb of call + // data - all this to say that we aren't using the the absolute largest shelf + // available. + store.Put(blob1) + store.Put(blob2) + store.Close() + + // Mimic a blobpool with max blob count of 6 upgrading to a max blob count of 24. + for _, maxBlobs := range []int{6, 24} { + statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) + statedb.AddBalance(addr1, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified) + statedb.AddBalance(addr2, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified) + statedb.AddBalance(addr3, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified) + statedb.Commit(0, true, false) + + // Make custom chain config where the max blob count changes based on the loop variable. + zero := uint64(0) + config := ¶ms.ChainConfig{ + ChainID: big.NewInt(1), + LondonBlock: big.NewInt(0), + BerlinBlock: big.NewInt(0), + CancunTime: &zero, + OsakaTime: &zero, + BlobScheduleConfig: ¶ms.BlobScheduleConfig{ + Cancun: ¶ms.BlobConfig{ + Target: maxBlobs / 2, + Max: maxBlobs, + UpdateFraction: params.DefaultCancunBlobConfig.UpdateFraction, + }, + Osaka: ¶ms.BlobConfig{ + Target: maxBlobs / 2, + Max: maxBlobs, + UpdateFraction: params.DefaultCancunBlobConfig.UpdateFraction, + }, + }, + } + chain := &testBlockChain{ + config: config, + basefee: uint256.NewInt(1050), + blobfee: uint256.NewInt(105), + 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) + } + + // Try to add the big blob tx. In the initial iteration it should overflow + // the pool. On the subsequent iteration it should be accepted. + errs := pool.Add([]*types.Transaction{tx3}, true) + if _, ok := pool.index[addr3]; ok && maxBlobs == 6 { + t.Errorf("expected insert of oversized blob tx to fail: blobs=24, maxBlobs=%d, err=%v", maxBlobs, errs[0]) + } else if !ok && maxBlobs == 10 { + t.Errorf("expected insert of oversized blob tx to succeed: blobs=24, maxBlobs=%d, err=%v", maxBlobs, errs[0]) + } + + // Verify the regular two txs are always available. + if got := pool.Get(tx1.Hash()); got == nil { + t.Errorf("expected tx %s from %s in pool", tx1.Hash(), addr1) + } + if got := pool.Get(tx2.Hash()); got == nil { + t.Errorf("expected tx %s from %s in pool", tx2.Hash(), addr2) + } + + // 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() + } +} + // TestBlobCountLimit tests the blobpool enforced limits on the max blob count. func TestBlobCountLimit(t *testing.T) { var (