add: implement pooledTx type

This commit is contained in:
healthykim 2025-09-08 22:04:03 +09:00
parent 86df16d6e6
commit d35d52b0ca
5 changed files with 122 additions and 50 deletions

View file

@ -137,8 +137,8 @@ func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transac
vhashes: tx.BlobHashes(), vhashes: tx.BlobHashes(),
version: tx.BlobTxSidecar().Version, version: tx.BlobTxSidecar().Version,
id: id, id: id,
storageSize: storageSize, storageSize: storageSize, // size of tx including cells
size: size, size: size, // size of tx only
nonce: tx.Nonce(), nonce: tx.Nonce(),
costCap: uint256.MustFromBig(tx.Cost()), costCap: uint256.MustFromBig(tx.Cost()),
execTipCap: uint256.MustFromBig(tx.GasTipCap()), execTipCap: uint256.MustFromBig(tx.GasTipCap()),
@ -153,6 +153,26 @@ func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transac
return meta return meta
} }
type PooledBlobTx struct {
Transaction *types.Transaction
Sidecar *types.BlobTxCellSidecar
}
func NewPooledBlobTx(tx *types.Transaction, sidecar *types.BlobTxCellSidecar) *PooledBlobTx {
return &PooledBlobTx{
Transaction: tx,
Sidecar: sidecar,
}
}
func (ptx *PooledBlobTx) Hash() common.Hash {
return ptx.Transaction.Hash()
}
func (ptx *PooledBlobTx) Convert() *types.Transaction {
return ptx.Transaction.WithBlobTxSidecar(ptx.Sidecar.ToBlobTxSidecar())
}
// BlobPool is the transaction pool dedicated to EIP-4844 blob transactions. // BlobPool is the transaction pool dedicated to EIP-4844 blob transactions.
// //
// Blob transactions are special snowflakes that are designed for a very specific // Blob transactions are special snowflakes that are designed for a very specific
@ -497,19 +517,31 @@ func (p *BlobPool) Close() error {
// each transaction on disk to create the in-memory metadata index. // each transaction on disk to create the in-memory metadata index.
func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error { func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
tx := new(types.Transaction) tx := new(types.Transaction)
if err := rlp.DecodeBytes(blob, tx); err != nil {
pooledTx := new(PooledBlobTx)
if err := rlp.DecodeBytes(blob, pooledTx); err != nil {
// 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.
if err := rlp.DecodeBytes(blob, tx); err != nil {
log.Error("Failed to decode blob pool entry", "id", id, "err", err) log.Error("Failed to decode blob pool entry", "id", id, "err", err)
return err return errors.New("unknown tx type")
} }
if tx.BlobTxSidecar() == nil { if tx.BlobTxSidecar() == nil {
log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash()) log.Error("Missing sidecar in blob pool entry", "id", id, "hash", tx.Hash())
return errors.New("missing blob sidecar") return errors.New("missing sidecar")
}
} else {
sidecar := pooledTx.Sidecar.ToBlobTxSidecar()
if sidecar == nil {
log.Error("Missing sidecar in blob pool entry", "id", id, "hash", pooledTx.Transaction.Hash())
return errors.New("missing sidecar")
}
tx = pooledTx.Convert()
} }
meta := newBlobTxMeta(id, tx.Size(), size, tx) meta := newBlobTxMeta(id, tx.Size(), size, tx)
if p.lookup.exists(meta.hash) { if p.lookup.exists(meta.hash) {
// This path is only possible after a crash, where deleted items are not // This path is only possible after a crash, where deleted items are not
// removed via the normal shutdown-startup procedure and thus may get // removed via the normal shutdown-startup procedure and thus may get
@ -799,17 +831,17 @@ func (p *BlobPool) offload(addr common.Address, nonce uint64, id uint64, inclusi
log.Error("Blobs missing for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err) log.Error("Blobs missing for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err)
return return
} }
var tx types.Transaction var pooledTx PooledBlobTx
if err = rlp.DecodeBytes(data, &tx); err != nil { if err = rlp.DecodeBytes(data, &pooledTx); err != nil {
log.Error("Blobs corrupted for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err) log.Error("Blobs corrupted for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err)
return return
} }
block, ok := inclusions[tx.Hash()] block, ok := inclusions[pooledTx.Transaction.Hash()]
if !ok { if !ok {
log.Warn("Blob transaction swapped out by signer", "from", addr, "nonce", nonce, "id", id) log.Warn("Blob transaction swapped out by signer", "from", addr, "nonce", nonce, "id", id)
return return
} }
if err := p.limbo.push(&tx, block); err != nil { if err := p.limbo.push(&pooledTx, block); err != nil {
log.Warn("Failed to offload blob tx into limbo", "err", err) log.Warn("Failed to offload blob tx into limbo", "err", err)
return return
} }
@ -1021,20 +1053,20 @@ func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) error {
// Serialize the transaction back into the primary datastore. // Serialize the transaction back into the primary datastore.
blob, err := rlp.EncodeToBytes(tx) blob, err := rlp.EncodeToBytes(tx)
if err != nil { if err != nil {
log.Error("Failed to encode transaction for storage", "hash", tx.Hash(), "err", err) log.Error("Failed to encode transaction for storage", "hash", tx.Transaction.Hash(), "err", err)
return err return err
} }
id, err := p.store.Put(blob) id, err := p.store.Put(blob)
if err != nil { if err != nil {
log.Error("Failed to write transaction into storage", "hash", tx.Hash(), "err", err) log.Error("Failed to write transaction into storage", "hash", tx.Transaction.Hash(), "err", err)
return err return err
} }
// Update the indices and metrics // Update the indices and metrics
meta := newBlobTxMeta(id, tx.Size(), p.store.Size(id), tx) meta := newBlobTxMeta(id, tx.Transaction.Size(), p.store.Size(id), tx.Transaction)
if _, ok := p.index[addr]; !ok { if _, ok := p.index[addr]; !ok {
if err := p.reserver.Hold(addr); err != nil { if err := p.reserver.Hold(addr); err != nil {
log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err) log.Warn("Failed to reserve account for blob pool", "tx", tx.Transaction.Hash(), "from", addr, "err", err)
return err return err
} }
p.index[addr] = []*blobTxMeta{meta} p.index[addr] = []*blobTxMeta{meta}
@ -1282,10 +1314,26 @@ func (p *BlobPool) getRLP(hash common.Hash) []byte {
} }
data, err := p.store.Get(id) data, err := p.store.Get(id)
if err != nil { if err != nil {
log.Error("Tracked blob transaction missing from store", "hash", hash, "id", id, "err", err) log.Error("Failed to get transaction in blobpool", "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 nil
} }
return data return data
}
encoded, err := rlp.EncodeToBytes(pooledTx.Convert())
if err != nil {
log.Error("Failed to encode transaction in blobpool", "hash", hash, "id", id, "err", err)
return nil
}
return encoded
} }
// Get returns a transaction if it is contained in the pool, or nil otherwise. // Get returns a transaction if it is contained in the pool, or nil otherwise.
@ -1294,15 +1342,19 @@ func (p *BlobPool) Get(hash common.Hash) *types.Transaction {
if len(data) == 0 { if len(data) == 0 {
return nil return nil
} }
item := new(types.Transaction) tx := new(types.Transaction)
if err := rlp.DecodeBytes(data, item); err != nil { 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", log.Error("Blobs corrupted for traced transaction",
"hash", hash, "id", id, "err", err) "hash", hash, "id", id, "err", err)
return nil return nil
} }
return item return tx
}
return pooledTx.Convert()
} }
// GetRLP returns a RLP-encoded transaction if it is contained in the pool. // GetRLP returns a RLP-encoded transaction if it is contained in the pool.
@ -1314,7 +1366,7 @@ func (p *BlobPool) GetRLP(hash common.Hash) []byte {
// given transaction hash. // given transaction hash.
// //
// The size refers the length of the 'rlp encoding' of a blob transaction // The size refers the length of the 'rlp encoding' of a blob transaction
// including the attached blobs. // excluding the attached blobs.
func (p *BlobPool) GetMetadata(hash common.Hash) *txpool.TxMetadata { func (p *BlobPool) GetMetadata(hash common.Hash) *txpool.TxMetadata {
p.lock.RLock() p.lock.RLock()
defer p.lock.RUnlock() defer p.lock.RUnlock()
@ -1372,11 +1424,18 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blo
// Decode the blob transaction // Decode the blob transaction
tx := new(types.Transaction) tx := new(types.Transaction)
var sidecar *types.BlobTxSidecar
pooledTx := new(PooledBlobTx)
if err := rlp.DecodeBytes(data, pooledTx); err != nil {
if err := rlp.DecodeBytes(data, tx); err != nil { if err := rlp.DecodeBytes(data, tx); err != nil {
log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err) log.Error("Blobs corrupted for traced transaction", "id", txID, "err", err)
continue continue
} }
sidecar := tx.BlobTxSidecar() sidecar = tx.BlobTxSidecar()
} else {
tx = pooledTx.Transaction
sidecar = pooledTx.Sidecar.ToBlobTxSidecar()
}
if sidecar == nil { if sidecar == nil {
log.Error("Blob tx without sidecar", "hash", tx.Hash(), "id", txID) log.Error("Blob tx without sidecar", "hash", tx.Hash(), "id", txID)
continue continue
@ -1454,7 +1513,11 @@ func (p *BlobPool) Add(txs []*types.Transaction, sync bool) []error {
if errs[i] != nil { if errs[i] != nil {
continue continue
} }
cellSidecar := tx.BlobTxSidecar().ToBlobTxCellSidecar() cellSidecar, err := tx.BlobTxSidecar().ToBlobTxCellSidecar()
if err != nil {
errs[i] = err
continue
}
errs[i] = p.add(tx.WithoutBlobTxSidecar(), cellSidecar) errs[i] = p.add(tx.WithoutBlobTxSidecar(), cellSidecar)
if errs[i] == nil { if errs[i] == nil {
adds = append(adds, tx.WithoutBlobTxSidecar()) adds = append(adds, tx.WithoutBlobTxSidecar())
@ -1526,12 +1589,11 @@ func (p *BlobPool) add(tx *types.Transaction, cellSidecar *types.BlobTxCellSidec
}() }()
} }
//todo(healthykim) remove this and seperate database or introduce list
blobSidecar := cellSidecar.ToBlobTxSidecar()
tx = tx.WithBlobTxSidecar(blobSidecar)
// Transaction permitted into the pool from a nonce and cost perspective, // Transaction permitted into the pool from a nonce and cost perspective,
// insert it into the database and update the indices // insert it into the database and update the indices
blob, err := rlp.EncodeToBytes(tx) pooledTx := NewPooledBlobTx(tx, cellSidecar)
blob, err := rlp.EncodeToBytes(pooledTx)
if err != nil { if err != nil {
log.Error("Failed to encode transaction for storage", "hash", tx.Hash(), "err", err) log.Error("Failed to encode transaction for storage", "hash", tx.Hash(), "err", err)
return err return err

View file

@ -452,8 +452,12 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) {
} }
// Item retrieved, make sure it matches the expectation // Item retrieved, make sure it matches the expectation
index := testBlobIndices[hash] index := testBlobIndices[hash]
if *blobs1[i] != *testBlobs[index] || proofs1[i][0] != testBlobProofs[index] { if *blobs1[i] != *testBlobs[index] {
t.Errorf("retrieved blob or proof mismatch: item %d, hash %x", i, hash) t.Errorf("retrieved blob mismatch: item %d, hash %x", i, hash)
continue
}
if proofs1[i][0] != testBlobProofs[index] {
t.Errorf("retrieved proof mismatch: item %d, hash %x", i, hash)
continue continue
} }
if *blobs2[i] != *testBlobs[index] || !slices.Equal(proofs2[i], testBlobCellProofs[index]) { if *blobs2[i] != *testBlobs[index] || !slices.Equal(proofs2[i], testBlobCellProofs[index]) {
@ -1752,7 +1756,9 @@ func TestAdd(t *testing.T) {
// Add each transaction one by one, verifying the pool internals in between // Add each transaction one by one, verifying the pool internals in between
for j, add := range tt.adds { for j, add := range tt.adds {
signed, _ := types.SignNewTx(keys[add.from], types.LatestSigner(params.MainnetChainConfig), add.tx) signed, _ := types.SignNewTx(keys[add.from], types.LatestSigner(params.MainnetChainConfig), add.tx)
if err := pool.add(signed.WithoutBlobTxSidecar(), signed.BlobTxSidecar().ToBlobTxCellSidecar()); !errors.Is(err, add.err) { sidecar, _ := signed.BlobTxSidecar().ToBlobTxCellSidecar()
if err := pool.add(signed.WithoutBlobTxSidecar(), sidecar); !errors.Is(err, add.err) {
t.Errorf("test %d, tx %d: adding transaction error mismatch: have %v, want %v", i, j, err, add.err) t.Errorf("test %d, tx %d: adding transaction error mismatch: have %v, want %v", i, j, err, add.err)
} }
if add.err == nil { if add.err == nil {
@ -1760,7 +1766,7 @@ func TestAdd(t *testing.T) {
if !exist { if !exist {
t.Errorf("test %d, tx %d: failed to lookup transaction's size", i, j) t.Errorf("test %d, tx %d: failed to lookup transaction's size", i, j)
} }
if size != signed.Size() { if size != signed.WithoutBlobTxSidecar().Size() {
t.Errorf("test %d, tx %d: transaction's size mismatches: have %v, want %v", t.Errorf("test %d, tx %d: transaction's size mismatches: have %v, want %v",
i, j, size, signed.Size()) i, j, size, signed.Size())
} }
@ -2124,7 +2130,8 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) {
b.Fatal(err) b.Fatal(err)
} }
statedb.AddBalance(addr, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified) statedb.AddBalance(addr, uint256.NewInt(1_000_000_000), tracing.BalanceChangeUnspecified)
pool.add(tx.WithoutBlobTxSidecar(), tx.BlobTxSidecar().ToBlobTxCellSidecar()) sidecar, _ := tx.BlobTxSidecar().ToBlobTxCellSidecar()
pool.add(tx.WithoutBlobTxSidecar(), sidecar)
} }
statedb.Commit(0, true, false) statedb.Commit(0, true, false)
defer pool.Close() defer pool.Close()

View file

@ -34,7 +34,7 @@ import (
type limboBlob struct { type limboBlob struct {
TxHash common.Hash // Owner transaction's hash to support resurrecting reorged txs TxHash common.Hash // Owner transaction's hash to support resurrecting reorged txs
Block uint64 // Block in which the blob transaction was included Block uint64 // Block in which the blob transaction was included
Tx *types.Transaction Tx *PooledBlobTx
} }
// limbo is a light, indexed database to temporarily store recently included // limbo is a light, indexed database to temporarily store recently included
@ -147,7 +147,7 @@ func (l *limbo) finalize(final *types.Header) {
// push stores a new blob transaction into the limbo, waiting until finality for // push stores a new blob transaction into the limbo, waiting until finality for
// it to be automatically evicted. // it to be automatically evicted.
func (l *limbo) push(tx *types.Transaction, block uint64) error { func (l *limbo) push(tx *PooledBlobTx, block uint64) error {
// If the blobs are already tracked by the limbo, consider it a programming // If the blobs are already tracked by the limbo, consider it a programming
// error. There's not much to do against it, but be loud. // error. There's not much to do against it, but be loud.
if _, ok := l.index[tx.Hash()]; ok { if _, ok := l.index[tx.Hash()]; ok {
@ -164,7 +164,7 @@ func (l *limbo) push(tx *types.Transaction, block uint64) error {
// pull retrieves a previously pushed set of blobs back from the limbo, removing // pull retrieves a previously pushed set of blobs back from the limbo, removing
// it at the same time. This method should be used when a previously included blob // it at the same time. This method should be used when a previously included blob
// transaction gets reorged out. // transaction gets reorged out.
func (l *limbo) pull(tx common.Hash) (*types.Transaction, error) { func (l *limbo) pull(tx common.Hash) (*PooledBlobTx, error) {
// If the blobs are not tracked by the limbo, there's not much to do. This // If the blobs are not tracked by the limbo, there's not much to do. This
// can happen for example if a blob transaction is mined without pushing it // can happen for example if a blob transaction is mined without pushing it
// into the network first. // into the network first.
@ -241,7 +241,7 @@ func (l *limbo) getAndDrop(id uint64) (*limboBlob, error) {
// setAndIndex assembles a limbo blob database entry and stores it, also updating // setAndIndex assembles a limbo blob database entry and stores it, also updating
// the in-memory indices. // the in-memory indices.
func (l *limbo) setAndIndex(tx *types.Transaction, block uint64) error { func (l *limbo) setAndIndex(tx *PooledBlobTx, block uint64) error {
txhash := tx.Hash() txhash := tx.Hash()
item := &limboBlob{ item := &limboBlob{
TxHash: txhash, TxHash: txhash,

View file

@ -22,7 +22,7 @@ import (
type txMetadata struct { type txMetadata struct {
id uint64 // the billy id of transction id uint64 // the billy id of transction
size uint64 // the RLP encoded size of transaction (blobs are included) size uint64 // the RLP encoded size of transaction (blobs are excluded)
} }
// lookup maps blob versioned hashes to transaction hashes that include them, // lookup maps blob versioned hashes to transaction hashes that include them,

View file

@ -176,15 +176,18 @@ func (sc *BlobTxSidecar) Copy() *BlobTxSidecar {
} }
} }
func (sc *BlobTxSidecar) ToBlobTxCellSidecar() *BlobTxCellSidecar { func (sc *BlobTxSidecar) ToBlobTxCellSidecar() (*BlobTxCellSidecar, error) {
cells, _ := kzg4844.ComputeCells(sc.Blobs) cells, err := kzg4844.ComputeCells(sc.Blobs)
if err != nil {
return nil, err
}
return &BlobTxCellSidecar{ return &BlobTxCellSidecar{
Version: sc.Version, Version: sc.Version,
Cells: cells, Cells: cells,
Commitments: sc.Commitments, Commitments: sc.Commitments,
Proofs: sc.Proofs, Proofs: sc.Proofs,
CellIndices: CustodyBitmap{}.SetAll(), CellIndices: CustodyBitmap{}.SetAll(),
} }, nil
} }
type BlobTxCellSidecar struct { type BlobTxCellSidecar struct {