feat: perform conversion in Init()

core/txpool/blobpool/blobpool.go

@@ -191,8 +191,9 @@ func (ptx *pooledBlobTx) convert() (*types.Transaction, error) {
 }
 
 // `removeParity` removes parity cells of cellSidecar
-// This is needed to reduce the size of transaction
-// Otherwise, pooledBlobTx will about 2 times bigger than the original tx
+// But in fact, the result of this is just a split of blobs
+// e.g. Blob = [a, b, c] -> Cells = [a], [b], [c]
+// Why do we need to apply EC then?
 func (ptx *pooledBlobTx) removeParity() error {
 	sc := ptx.Sidecar
 	if sc == nil {
@@ -478,9 +479,13 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 	}
 	// Index all transactions on disk and delete anything unprocessable
 	var fails []uint64
+	var convertTxs []*types.Transaction
 	index := func(id uint64, size uint32, blob []byte) {
-		if p.parseTransaction(id, size, blob) != nil {
+		if tx, err := p.parseTransaction(id, size, blob); err != nil {
 			fails = append(fails, id)
+		} else if tx != nil {
+			fails = append(fails, id)
+			convertTxs = append(convertTxs, tx)
 		}
 	}
 	store, err := billy.Open(billy.Options{Path: queuedir, Repair: true}, slotter, index)
@@ -489,6 +494,40 @@ func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reser
 	}
 	p.store = store
 
+	if len(convertTxs) > 0 {
+		log.Trace("Convert plain transaction to pooled transaction", "len", len(convertTxs))
+		for _, tx := range convertTxs {
+			// Note that we skip errors here
+			// Just like parseTransaction failure does not abort the blobpool creation,
+			// conversion process also cannot abort the entire process.
+			pooledTx, err := newPooledBlobTx(tx)
+			if err != nil {
+				continue
+			}
+			if err := pooledTx.removeParity(); err != nil {
+				return err
+			}
+			blob, err := rlp.EncodeToBytes(pooledTx)
+			if err != nil {
+				continue
+			}
+			id, err := p.store.Put(blob)
+			if err != nil {
+				continue
+			}
+			meta := newBlobTxMeta(id, pooledTx.Size, p.store.Size(id), pooledTx)
+
+			sender, err := types.Sender(p.signer, pooledTx.Transaction)
+			if err != nil {
+				fails = append(fails, id)
+				continue
+			}
+			if err := p.trackTransaction(meta, sender); err != nil {
+				fails = append(fails, id)
+				continue
+			}
+		}
+	}
 	if len(fails) > 0 {
 		log.Warn("Dropping invalidated blob transactions", "ids", fails)
 		dropInvalidMeter.Mark(int64(len(fails)))
@@ -561,43 +600,50 @@ func (p *BlobPool) Close() error {
 
 // parseTransaction is a callback method on pool creation that gets called for
 // each transaction on disk to create the in-memory metadata index.
-func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
+// return:
+// - types.Transaction : plain tx if the type of 'blob' is plain tx
+func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) (*types.Transaction, error) {
 	tx := new(types.Transaction)
-
 	pooledTx := new(pooledBlobTx)
+
 	if err := rlp.DecodeBytes(blob, pooledTx); err != nil {
 		// This path is impossible unless the disk data representation changes
 		// across restarts. For that ever improbable case, recover gracefully
 		// by ignoring this data entry.
 		if err := rlp.DecodeBytes(blob, tx); err != nil {
 			log.Error("Failed to decode blob pool entry", "id", id, "err", err)
-			return err
+			return nil, err
 		}
 		if tx.BlobTxSidecar() == nil {
 			log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash())
-			return errors.New("missing blob sidecar")
-		}
-		pooledTx, err = newPooledBlobTx(tx)
-		if err != nil {
-			return err
+			return nil, errors.New("missing blob sidecar")
 		}
+		return tx, nil
+	}
+	if pooledTx.Sidecar == nil {
+		log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash())
+		return nil, errors.New("missing blob sidecar")
 	}
 	meta := newBlobTxMeta(id, pooledTx.Size, size, pooledTx)
 
-	if p.lookup.exists(meta.hash) {
-		// This path is only possible after a crash, where deleted items are not
-		// removed via the normal shutdown-startup procedure and thus may get
-		// partially resurrected.
-		log.Error("Rejecting duplicate blob pool entry", "id", id, "hash", tx.Hash())
-		return errors.New("duplicate blob entry")
-	}
 	sender, err := types.Sender(p.signer, pooledTx.Transaction)
 	if err != nil {
 		// This path is impossible unless the signature validity changes across
 		// restarts. For that ever improbable case, recover gracefully by ignoring
 		// this data entry.
 		log.Error("Failed to recover blob tx sender", "id", id, "hash", tx.Hash(), "err", err)
-		return err
+		return nil, err
+	}
+	return nil, p.trackTransaction(meta, sender)
+}
+
+func (p *BlobPool) trackTransaction(meta *blobTxMeta, sender common.Address) error {
+	if p.lookup.exists(meta.hash) {
+		// This path is only possible after a crash, where deleted items are not
+		// removed via the normal shutdown-startup procedure and thus may get
+		// partially resurrected.
+		log.Error("Rejecting duplicate blob pool entry", "id", meta.id, "hash", meta.hash)
+		return fmt.Errorf("duplicate blob entry %d, %s", meta.id, meta.hash)
 	}
 	if _, ok := p.index[sender]; !ok {
 		if err := p.reserver.Hold(sender); err != nil {
@@ -1356,30 +1402,10 @@ func (p *BlobPool) getRLP(hash common.Hash) []byte {
 	}
 	data, err := p.store.Get(id)
 	if err != nil {
-		log.Error("Failed to get transaction in blobpool", "hash", hash, "id", id, "err", err)
+		log.Error("Tracked blob transaction missing from store", "hash", hash, "id", id, "err", err)
 		return nil
 	}
-	tx := new(types.Transaction)
-	pooledTx := new(pooledBlobTx)
-	if err := rlp.DecodeBytes(data, pooledTx); err != nil {
-		if err := rlp.DecodeBytes(data, tx); err != nil {
-			log.Error("Failed to decode transaction in blobpool", "hash", hash, "id", id, "err", err)
-			return nil
-		}
-		return data
-	}
-	tx, err = pooledTx.convert()
-	if err != nil {
-		log.Error("Failed to convert transaction in blobpool", "hash", hash, "id", id, "err", err)
-		return nil
-	}
-	encoded, err := rlp.EncodeToBytes(tx)
-	if err != nil {
-		log.Error("Failed to encode transaction in blobpool", "hash", hash, "id", id, "err", err)
-		return nil
-	}
-
-	return encoded
+	return data
 }
 
 // Get returns a transaction if it is contained in the pool, or nil otherwise.
@@ -1388,28 +1414,43 @@ func (p *BlobPool) Get(hash common.Hash) *types.Transaction {
 	if len(data) == 0 {
 		return nil
 	}
-	tx := new(types.Transaction)
 	pooledTx := new(pooledBlobTx)
 	if err := rlp.DecodeBytes(data, pooledTx); err != nil {
-		if err := rlp.DecodeBytes(data, tx); err != nil {
-			id, _ := p.lookup.storeidOfTx(hash)
+		id, _ := p.lookup.storeidOfTx(hash)
 
-			log.Error("Blobs corrupted for traced transaction",
-				"hash", hash, "id", id, "err", err)
-			return nil
-		}
+		log.Error("Blobs corrupted for traced transaction",
+			"hash", hash, "id", id, "err", err)
+		return nil
+	}
+	if tx, err := pooledTx.convert(); err != nil {
+		return nil
+	} else {
 		return tx
 	}
-	tx, err := pooledTx.convert()
-	if err == nil {
-		return tx
-	}
-	return nil
 }
 
 // GetRLP returns a RLP-encoded transaction if it is contained in the pool.
 func (p *BlobPool) GetRLP(hash common.Hash) []byte {
-	return p.getRLP(hash)
+	data := p.getRLP(hash)
+	pooledTx := new(pooledBlobTx)
+	if err := rlp.DecodeBytes(data, pooledTx); err != nil {
+		id, _ := p.lookup.storeidOfTx(hash)
+
+		log.Error("Blobs corrupted for traced transaction",
+			"hash", hash, "id", id, "err", err)
+		return nil
+	}
+	tx, err := pooledTx.convert()
+	if err != nil {
+		log.Error("Failed to convert pooledTx", "err", err)
+		return nil
+	}
+	blob, err := rlp.EncodeToBytes(tx)
+	if err != nil {
+		log.Error("Failed to encode", "err", err)
+		return nil
+	}
+	return blob
 }
 
 // GetMetadata returns the transaction type and transaction size with the
@@ -1476,10 +1517,8 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blo
 		tx := new(types.Transaction)
 		pooledTx := new(pooledBlobTx)
 		if err := rlp.DecodeBytes(data, pooledTx); err != nil {
-			if err := rlp.DecodeBytes(data, tx); err != nil {
-				log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err)
-				continue
-			}
+			log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err)
+			continue
 		} else {
 			tx, err = pooledTx.convert()
 			if err != nil {

core/txpool/blobpool/blobpool_test.go

@@ -488,7 +488,7 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) {
 //   - 8. Fully duplicate transactions (matching hash) must be dropped
 //   - 9. Duplicate nonces from the same account must be dropped
 func TestOpenDrops(t *testing.T) {
-	//log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelTrace, true)))
+	// log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelTrace, true)))
 
 	// Create a temporary folder for the persistent backend
 	storage := t.TempDir()
@@ -515,75 +515,76 @@ func TestOpenDrops(t *testing.T) {
 		S:          new(uint256.Int),
 	})
 	blob, _ := rlp.EncodeToBytes(tx)
-	badsig, _ := store.Put(blob)
+	badsig := tx.Hash()
+	store.Put(blob)
 
 	// Insert a sequence of transactions with a nonce gap in between to verify
 	// that anything gapped will get evicted (case 3).
 	var (
 		gapper, _ = crypto.GenerateKey()
 
-		valids = make(map[uint64]struct{})
-		gapped = make(map[uint64]struct{})
+		valids = make(map[common.Hash]struct{})
+		gapped = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 3, 4, 6, 7} { // first gap at #2, another at #5
 		tx := makeTx(nonce, 1, 1, 1, gapper)
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
+		store.Put(blob)
 		if nonce < 2 {
-			valids[id] = struct{}{}
+			valids[tx.Hash()] = struct{}{}
 		} else {
-			gapped[id] = struct{}{}
+			gapped[tx.Hash()] = struct{}{}
 		}
 	}
 	// Insert a sequence of transactions with a gapped starting nonce to verify
 	// that the entire set will get dropped (case 3).
 	var (
 		dangler, _ = crypto.GenerateKey()
-		dangling   = make(map[uint64]struct{})
+		dangling   = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{1, 2, 3} { // first gap at #0, all set dangling
 		tx := makeTx(nonce, 1, 1, 1, dangler)
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
-		dangling[id] = struct{}{}
+		store.Put(blob)
+		dangling[tx.Hash()] = struct{}{}
 	}
 	// Insert a sequence of transactions with already passed nonces to veirfy
 	// that the entire set will get dropped (case 4).
 	var (
 		filler, _ = crypto.GenerateKey()
-		filled    = make(map[uint64]struct{})
+		filled    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2} { // account nonce at 3, all set filled
 		tx := makeTx(nonce, 1, 1, 1, filler)
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
-		filled[id] = struct{}{}
+		store.Put(blob)
+		filled[tx.Hash()] = struct{}{}
 	}
 	// Insert a sequence of transactions with partially passed nonces to verify
 	// that the included part of the set will get dropped (case 4).
 	var (
 		overlapper, _ = crypto.GenerateKey()
-		overlapped    = make(map[uint64]struct{})
+		overlapped    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2, 3} { // account nonce at 2, half filled
 		tx := makeTx(nonce, 1, 1, 1, overlapper)
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
+		store.Put(blob)
 		if nonce >= 2 {
-			valids[id] = struct{}{}
+			valids[tx.Hash()] = struct{}{}
 		} else {
-			overlapped[id] = struct{}{}
+			overlapped[tx.Hash()] = struct{}{}
 		}
 	}
 	// Insert a sequence of transactions with an underpriced first to verify that
 	// the entire set will get dropped (case 5).
 	var (
 		underpayer, _ = crypto.GenerateKey()
-		underpaid     = make(map[uint64]struct{})
+		underpaid     = make(map[common.Hash]struct{})
 	)
 	for i := 0; i < 5; i++ { // make #0 underpriced
 		var tx *types.Transaction
@@ -594,15 +595,15 @@ func TestOpenDrops(t *testing.T) {
 		}
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
-		underpaid[id] = struct{}{}
+		store.Put(blob)
+		underpaid[tx.Hash()] = struct{}{}
 	}
 
 	// Insert a sequence of transactions with an underpriced in between to verify
 	// that it and anything newly gapped will get evicted (case 5).
 	var (
 		outpricer, _ = crypto.GenerateKey()
-		outpriced    = make(map[uint64]struct{})
+		outpriced    = make(map[common.Hash]struct{})
 	)
 	for i := 0; i < 5; i++ { // make #2 underpriced
 		var tx *types.Transaction
@@ -613,18 +614,18 @@ func TestOpenDrops(t *testing.T) {
 		}
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
+		store.Put(blob)
 		if i < 2 {
-			valids[id] = struct{}{}
+			valids[tx.Hash()] = struct{}{}
 		} else {
-			outpriced[id] = struct{}{}
+			outpriced[tx.Hash()] = struct{}{}
 		}
 	}
 	// Insert a sequence of transactions fully overdrafted to verify that the
 	// entire set will get invalidated (case 6).
 	var (
 		exceeder, _ = crypto.GenerateKey()
-		exceeded    = make(map[uint64]struct{})
+		exceeded    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2} { // nonce 0 overdrafts the account
 		var tx *types.Transaction
@@ -635,14 +636,14 @@ func TestOpenDrops(t *testing.T) {
 		}
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
-		exceeded[id] = struct{}{}
+		store.Put(blob)
+		exceeded[tx.Hash()] = struct{}{}
 	}
 	// Insert a sequence of transactions partially overdrafted to verify that part
 	// of the set will get invalidated (case 6).
 	var (
 		overdrafter, _ = crypto.GenerateKey()
-		overdrafted    = make(map[uint64]struct{})
+		overdrafted    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2} { // nonce 1 overdrafts the account
 		var tx *types.Transaction
@@ -653,44 +654,46 @@ func TestOpenDrops(t *testing.T) {
 		}
 		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
+		store.Put(blob)
 		if nonce < 1 {
-			valids[id] = struct{}{}
+			valids[tx.Hash()] = struct{}{}
 		} else {
-			overdrafted[id] = struct{}{}
+			overdrafted[tx.Hash()] = struct{}{}
 		}
 	}
 	// Insert a sequence of transactions overflowing the account cap to verify
 	// that part of the set will get invalidated (case 7).
 	var (
 		overcapper, _ = crypto.GenerateKey()
-		overcapped    = make(map[uint64]struct{})
+		overcapped    = make(map[common.Hash]struct{})
 	)
 	for nonce := uint64(0); nonce < maxTxsPerAccount+3; nonce++ {
-		blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, 1, 1, overcapper))
+		tx := makeTx(nonce, 1, 1, 1, overcapper)
+		blob, _ := rlp.EncodeToBytes(tx)
 
-		id, _ := store.Put(blob)
+		store.Put(blob)
 		if nonce < maxTxsPerAccount {
-			valids[id] = struct{}{}
+			valids[tx.Hash()] = struct{}{}
 		} else {
-			overcapped[id] = struct{}{}
+			overcapped[tx.Hash()] = struct{}{}
 		}
 	}
 	// Insert a batch of duplicated transactions to verify that only one of each
 	// version will remain (case 8).
 	var (
 		duplicater, _ = crypto.GenerateKey()
-		duplicated    = make(map[uint64]struct{})
+		duplicated    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2} {
-		blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, 1, 1, duplicater))
+		tx := makeTx(nonce, 1, 1, 1, duplicater)
+		blob, _ := rlp.EncodeToBytes(tx)
 
 		for i := 0; i < int(nonce)+1; i++ {
-			id, _ := store.Put(blob)
+			store.Put(blob)
 			if i == 0 {
-				valids[id] = struct{}{}
+				valids[tx.Hash()] = struct{}{}
 			} else {
-				duplicated[id] = struct{}{}
+				duplicated[tx.Hash()] = struct{}{}
 			}
 		}
 	}
@@ -698,17 +701,18 @@ func TestOpenDrops(t *testing.T) {
 	// remain (case 9).
 	var (
 		repeater, _ = crypto.GenerateKey()
-		repeated    = make(map[uint64]struct{})
+		repeated    = make(map[common.Hash]struct{})
 	)
 	for _, nonce := range []uint64{0, 1, 2} {
 		for i := 0; i < int(nonce)+1; i++ {
-			blob, _ := rlp.EncodeToBytes(makeTx(nonce, 1, uint64(i)+1 /* unique hashes */, 1, repeater))
+			tx := makeTx(nonce, 1, uint64(i)+1 /* unique hashes */, 1, repeater)
+			blob, _ := rlp.EncodeToBytes(tx)
 
-			id, _ := store.Put(blob)
+			store.Put(blob)
 			if i == 0 {
-				valids[id] = struct{}{}
+				valids[tx.Hash()] = struct{}{}
 			} else {
-				repeated[id] = struct{}{}
+				repeated[tx.Hash()] = struct{}{}
 			}
 		}
 	}
@@ -745,39 +749,41 @@ func TestOpenDrops(t *testing.T) {
 
 	// Verify that the malformed (case 1), badly signed (case 2) and gapped (case
 	// 3) txs have been deleted from the pool
-	alive := make(map[uint64]struct{})
+	alive := make(map[common.Hash]struct{})
 	for _, txs := range pool.index {
 		for _, tx := range txs {
 			switch tx.id {
 			case malformed:
 				t.Errorf("malformed RLP transaction remained in storage")
-			case badsig:
-				t.Errorf("invalidly signed transaction remained in storage")
 			default:
-				if _, ok := dangling[tx.id]; ok {
+				if badsig == tx.hash {
+					t.Errorf("invalidly signed transaction remained in storage")
+				}
+				if _, ok := dangling[tx.hash]; ok {
 					t.Errorf("dangling transaction remained in storage: %d", tx.id)
-				} else if _, ok := filled[tx.id]; ok {
+				} else if _, ok := filled[tx.hash]; ok {
 					t.Errorf("filled transaction remained in storage: %d", tx.id)
-				} else if _, ok := overlapped[tx.id]; ok {
+				} else if _, ok := overlapped[tx.hash]; ok {
 					t.Errorf("overlapped transaction remained in storage: %d", tx.id)
-				} else if _, ok := gapped[tx.id]; ok {
+				} else if _, ok := gapped[tx.hash]; ok {
 					t.Errorf("gapped transaction remained in storage: %d", tx.id)
-				} else if _, ok := underpaid[tx.id]; ok {
+				} else if _, ok := underpaid[tx.hash]; ok {
 					t.Errorf("underpaid transaction remained in storage: %d", tx.id)
-				} else if _, ok := outpriced[tx.id]; ok {
+				} else if _, ok := outpriced[tx.hash]; ok {
 					t.Errorf("outpriced transaction remained in storage: %d", tx.id)
-				} else if _, ok := exceeded[tx.id]; ok {
+				} else if _, ok := exceeded[tx.hash]; ok {
 					t.Errorf("fully overdrafted transaction remained in storage: %d", tx.id)
-				} else if _, ok := overdrafted[tx.id]; ok {
+				} else if _, ok := overdrafted[tx.hash]; ok {
 					t.Errorf("partially overdrafted transaction remained in storage: %d", tx.id)
-				} else if _, ok := overcapped[tx.id]; ok {
+				} else if _, ok := overcapped[tx.hash]; ok {
 					t.Errorf("overcapped transaction remained in storage: %d", tx.id)
-				} else if _, ok := duplicated[tx.id]; ok {
-					t.Errorf("duplicated transaction remained in storage: %d", tx.id)
-				} else if _, ok := repeated[tx.id]; ok {
+				} else if _, ok := repeated[tx.hash]; ok {
 					t.Errorf("repeated nonce transaction remained in storage: %d", tx.id)
 				} else {
-					alive[tx.id] = struct{}{}
+					if _, ok := alive[tx.hash]; ok {
+						t.Errorf("duplicated transaction remained in storage: %d", tx.id)
+					}
+					alive[tx.hash] = struct{}{}
 				}
 			}
 		}
@@ -786,14 +792,14 @@ func TestOpenDrops(t *testing.T) {
 	if len(alive) != len(valids) {
 		t.Errorf("valid transaction count mismatch: have %d, want %d", len(alive), len(valids))
 	}
-	for id := range alive {
-		if _, ok := valids[id]; !ok {
-			t.Errorf("extra transaction %d", id)
+	for hash := range alive {
+		if _, ok := valids[hash]; !ok {
+			t.Errorf("extra transaction %s", hash)
 		}
 	}
-	for id := range valids {
-		if _, ok := alive[id]; !ok {
-			t.Errorf("missing transaction %d", id)
+	for hash := range valids {
+		if _, ok := alive[hash]; !ok {
+			t.Errorf("missing transaction %s", hash)
 		}
 	}
 	// Verify all the calculated pool internals. Interestingly, this is **not**