mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 22:26:42 +00:00
upgrade blobpool
This commit is contained in:
maskpp
parent
663fa7b496
commit
22db68b81b
5 changed files with 263 additions and 292 deletions
|
|
@ -89,20 +89,20 @@ const (
|
|||
// bare minimum needed fields to keep the size down (and thus number of entries
|
||||
// larger with the same memory consumption).
|
||||
type blobTxMeta struct {
|
||||
hash common.Hash // Transaction hash to maintain the lookup table
|
||||
vhashes []common.Hash // Blob versioned hashes to maintain the lookup table
|
||||
TxHash common.Hash // Transaction TxHash to maintain the lookup table
|
||||
VHashes []common.Hash // Blob versioned hashes to maintain the lookup table
|
||||
|
||||
id uint64 // Storage ID in the pool's persistent store
|
||||
storageSize uint32 // Byte size in the pool's persistent store
|
||||
size uint64 // RLP-encoded size of transaction including the attached blob
|
||||
Id uint64 // Storage ID in the pool's persistent store
|
||||
StorageSize uint32 // Byte size in the pool's persistent store
|
||||
Size uint64 // RLP-encoded Size of transaction including the attached blob
|
||||
|
||||
nonce uint64 // Needed to prioritize inclusion order within an account
|
||||
costCap *uint256.Int // Needed to validate cumulative balance sufficiency
|
||||
execTipCap *uint256.Int // Needed to prioritize inclusion order across accounts and validate replacement price bump
|
||||
execFeeCap *uint256.Int // Needed to validate replacement price bump
|
||||
blobFeeCap *uint256.Int // Needed to validate replacement price bump
|
||||
execGas uint64 // Needed to check inclusion validity before reading the blob
|
||||
blobGas uint64 // Needed to check inclusion validity before reading the blob
|
||||
Nonce uint64 // Needed to prioritize inclusion order within an account
|
||||
CostCap *uint256.Int // Needed to validate cumulative balance sufficiency
|
||||
ExecTipCap *uint256.Int // Needed to prioritize inclusion order across accounts and validate replacement price bump
|
||||
ExecFeeCap *uint256.Int // Needed to validate replacement price bump
|
||||
BlobFeeCap *uint256.Int // Needed to validate replacement price bump
|
||||
ExecGas uint64 // Needed to check inclusion validity before reading the blob
|
||||
BlobGas uint64 // Needed to check inclusion validity before reading the blob
|
||||
|
||||
basefeeJumps float64 // Absolute number of 1559 fee adjustments needed to reach the tx's fee cap
|
||||
blobfeeJumps float64 // Absolute number of 4844 fee adjustments needed to reach the tx's blob fee cap
|
||||
|
|
@ -116,21 +116,21 @@ type blobTxMeta struct {
|
|||
// and assembles a helper struct to track in memory.
|
||||
func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transaction) *blobTxMeta {
|
||||
meta := &blobTxMeta{
|
||||
hash: tx.Hash(),
|
||||
vhashes: tx.BlobHashes(),
|
||||
id: id,
|
||||
storageSize: storageSize,
|
||||
size: size,
|
||||
nonce: tx.Nonce(),
|
||||
costCap: uint256.MustFromBig(tx.Cost()),
|
||||
execTipCap: uint256.MustFromBig(tx.GasTipCap()),
|
||||
execFeeCap: uint256.MustFromBig(tx.GasFeeCap()),
|
||||
blobFeeCap: uint256.MustFromBig(tx.BlobGasFeeCap()),
|
||||
execGas: tx.Gas(),
|
||||
blobGas: tx.BlobGas(),
|
||||
TxHash: tx.Hash(),
|
||||
VHashes: tx.BlobHashes(),
|
||||
Id: id,
|
||||
StorageSize: storageSize,
|
||||
Size: size,
|
||||
Nonce: tx.Nonce(),
|
||||
CostCap: uint256.MustFromBig(tx.Cost()),
|
||||
ExecTipCap: uint256.MustFromBig(tx.GasTipCap()),
|
||||
ExecFeeCap: uint256.MustFromBig(tx.GasFeeCap()),
|
||||
BlobFeeCap: uint256.MustFromBig(tx.BlobGasFeeCap()),
|
||||
ExecGas: tx.Gas(),
|
||||
BlobGas: tx.BlobGas(),
|
||||
}
|
||||
meta.basefeeJumps = dynamicFeeJumps(meta.execFeeCap)
|
||||
meta.blobfeeJumps = dynamicFeeJumps(meta.blobFeeCap)
|
||||
meta.basefeeJumps = dynamicFeeJumps(meta.ExecFeeCap)
|
||||
meta.blobfeeJumps = dynamicFeeJumps(meta.BlobFeeCap)
|
||||
|
||||
return meta
|
||||
}
|
||||
|
|
@ -490,7 +490,7 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
|
|||
}
|
||||
|
||||
meta := newBlobTxMeta(id, tx.Size(), size, tx)
|
||||
if p.lookup.exists(meta.hash) {
|
||||
if p.lookup.exists(meta.TxHash) {
|
||||
// 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.
|
||||
|
|
@ -513,10 +513,10 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error {
|
|||
p.spent[sender] = new(uint256.Int)
|
||||
}
|
||||
p.index[sender] = append(p.index[sender], meta)
|
||||
p.spent[sender] = new(uint256.Int).Add(p.spent[sender], meta.costCap)
|
||||
p.spent[sender] = new(uint256.Int).Add(p.spent[sender], meta.CostCap)
|
||||
|
||||
p.lookup.track(meta)
|
||||
p.stored += uint64(meta.storageSize)
|
||||
p.stored += uint64(meta.StorageSize)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -529,15 +529,15 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
return
|
||||
}
|
||||
sort.Slice(txs, func(i, j int) bool {
|
||||
return txs[i].nonce < txs[j].nonce
|
||||
return txs[i].Nonce < txs[j].Nonce
|
||||
})
|
||||
// If there is a gap between the chain state and the blob pool, drop
|
||||
// all the transactions as they are non-executable. Similarly, if the
|
||||
// entire tx range was included, drop all.
|
||||
var (
|
||||
next = p.state.GetNonce(addr)
|
||||
gapped = txs[0].nonce > next
|
||||
filled = txs[len(txs)-1].nonce < next
|
||||
gapped = txs[0].Nonce > next
|
||||
filled = txs[len(txs)-1].Nonce < next
|
||||
)
|
||||
if gapped || filled {
|
||||
var (
|
||||
|
|
@ -545,15 +545,15 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
nonces []uint64
|
||||
)
|
||||
for i := 0; i < len(txs); i++ {
|
||||
ids = append(ids, txs[i].id)
|
||||
nonces = append(nonces, txs[i].nonce)
|
||||
ids = append(ids, txs[i].Id)
|
||||
nonces = append(nonces, txs[i].Nonce)
|
||||
|
||||
p.stored -= uint64(txs[i].storageSize)
|
||||
p.stored -= uint64(txs[i].StorageSize)
|
||||
p.lookup.untrack(txs[i])
|
||||
|
||||
// Included transactions blobs need to be moved to the limbo
|
||||
// Included transactions blobs need to be recorded in the limbo
|
||||
if filled && inclusions != nil {
|
||||
p.offload(addr, txs[i].nonce, txs[i].id, inclusions)
|
||||
p.offload(addr, txs[i], inclusions)
|
||||
}
|
||||
}
|
||||
delete(p.index, addr)
|
||||
|
|
@ -570,57 +570,64 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
log.Trace("Dropping filled blob transactions", "from", addr, "filled", nonces, "ids", ids)
|
||||
dropFilledMeter.Mark(int64(len(ids)))
|
||||
}
|
||||
for _, id := range ids {
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
log.Error("Failed to delete blob transaction", "from", addr, "id", id, "err", err)
|
||||
|
||||
// If the txs were recorded in the limbo, we don't delete them.
|
||||
if !(filled && inclusions != nil) {
|
||||
for _, id := range ids {
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
log.Error("Failed to delete blob transaction", "from", addr, "id", id, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
// If there is overlap between the chain state and the blob pool, drop
|
||||
// anything below the current state
|
||||
if txs[0].nonce < next {
|
||||
if txs[0].Nonce < next {
|
||||
var (
|
||||
ids []uint64
|
||||
nonces []uint64
|
||||
)
|
||||
for len(txs) > 0 && txs[0].nonce < next {
|
||||
ids = append(ids, txs[0].id)
|
||||
nonces = append(nonces, txs[0].nonce)
|
||||
for len(txs) > 0 && txs[0].Nonce < next {
|
||||
ids = append(ids, txs[0].Id)
|
||||
nonces = append(nonces, txs[0].Nonce)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[0].costCap)
|
||||
p.stored -= uint64(txs[0].storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[0].CostCap)
|
||||
p.stored -= uint64(txs[0].StorageSize)
|
||||
p.lookup.untrack(txs[0])
|
||||
|
||||
// Included transactions blobs need to be moved to the limbo
|
||||
// Included transactions blobs need to be recorded in the limbo.
|
||||
if inclusions != nil {
|
||||
p.offload(addr, txs[0].nonce, txs[0].id, inclusions)
|
||||
p.offload(addr, txs[0], inclusions)
|
||||
}
|
||||
txs = txs[1:]
|
||||
}
|
||||
log.Trace("Dropping overlapped blob transactions", "from", addr, "overlapped", nonces, "ids", ids, "left", len(txs))
|
||||
dropOverlappedMeter.Mark(int64(len(ids)))
|
||||
|
||||
for _, id := range ids {
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
log.Error("Failed to delete blob transaction", "from", addr, "id", id, "err", err)
|
||||
// If the txs were recorded in the limbo, we don't delete them.
|
||||
if inclusions == nil {
|
||||
for _, id := range ids {
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
log.Error("Failed to delete blob transaction", "from", addr, "id", id, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
p.index[addr] = txs
|
||||
}
|
||||
// Iterate over the transactions to initialize their eviction thresholds
|
||||
// and to detect any nonce gaps
|
||||
txs[0].evictionExecTip = txs[0].execTipCap
|
||||
txs[0].evictionExecTip = txs[0].ExecTipCap
|
||||
txs[0].evictionExecFeeJumps = txs[0].basefeeJumps
|
||||
txs[0].evictionBlobFeeJumps = txs[0].blobfeeJumps
|
||||
|
||||
for i := 1; i < len(txs); i++ {
|
||||
// If there's no nonce gap, initialize the eviction thresholds as the
|
||||
// minimum between the cumulative thresholds and the current tx fees
|
||||
if txs[i].nonce == txs[i-1].nonce+1 {
|
||||
if txs[i].Nonce == txs[i-1].Nonce+1 {
|
||||
txs[i].evictionExecTip = txs[i-1].evictionExecTip
|
||||
if txs[i].evictionExecTip.Cmp(txs[i].execTipCap) > 0 {
|
||||
txs[i].evictionExecTip = txs[i].execTipCap
|
||||
if txs[i].evictionExecTip.Cmp(txs[i].ExecTipCap) > 0 {
|
||||
txs[i].evictionExecTip = txs[i].ExecTipCap
|
||||
}
|
||||
txs[i].evictionExecFeeJumps = txs[i-1].evictionExecFeeJumps
|
||||
if txs[i].evictionExecFeeJumps > txs[i].basefeeJumps {
|
||||
|
|
@ -638,14 +645,14 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
// Also, Billy behind the blobpool does not journal deletes. A process
|
||||
// crash would result in previously deleted entities being resurrected.
|
||||
// That could potentially cause a duplicate nonce to appear.
|
||||
if txs[i].nonce == txs[i-1].nonce {
|
||||
id, _ := p.lookup.storeidOfTx(txs[i].hash)
|
||||
if txs[i].Nonce == txs[i-1].Nonce {
|
||||
id, _ := p.lookup.storeidOfTx(txs[i].TxHash)
|
||||
|
||||
log.Error("Dropping repeat nonce blob transaction", "from", addr, "nonce", txs[i].nonce, "id", id)
|
||||
log.Error("Dropping repeat Nonce blob transaction", "from", addr, "nonce", txs[i].Nonce, "id", id)
|
||||
dropRepeatedMeter.Mark(1)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].costCap)
|
||||
p.stored -= uint64(txs[i].storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].CostCap)
|
||||
p.stored -= uint64(txs[i].StorageSize)
|
||||
p.lookup.untrack(txs[i])
|
||||
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
|
|
@ -663,16 +670,16 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
nonces []uint64
|
||||
)
|
||||
for j := i; j < len(txs); j++ {
|
||||
ids = append(ids, txs[j].id)
|
||||
nonces = append(nonces, txs[j].nonce)
|
||||
ids = append(ids, txs[j].Id)
|
||||
nonces = append(nonces, txs[j].Nonce)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[j].costCap)
|
||||
p.stored -= uint64(txs[j].storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[j].CostCap)
|
||||
p.stored -= uint64(txs[j].StorageSize)
|
||||
p.lookup.untrack(txs[j])
|
||||
}
|
||||
txs = txs[:i]
|
||||
|
||||
log.Error("Dropping gapped blob transactions", "from", addr, "missing", txs[i-1].nonce+1, "drop", nonces, "ids", ids)
|
||||
log.Error("Dropping gapped blob transactions", "from", addr, "missing", txs[i-1].Nonce+1, "drop", nonces, "ids", ids)
|
||||
dropGappedMeter.Mark(int64(len(ids)))
|
||||
|
||||
for _, id := range ids {
|
||||
|
|
@ -701,11 +708,11 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
txs[len(txs)-1] = nil
|
||||
txs = txs[:len(txs)-1]
|
||||
|
||||
ids = append(ids, last.id)
|
||||
nonces = append(nonces, last.nonce)
|
||||
ids = append(ids, last.Id)
|
||||
nonces = append(nonces, last.Nonce)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], last.costCap)
|
||||
p.stored -= uint64(last.storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], last.CostCap)
|
||||
p.stored -= uint64(last.StorageSize)
|
||||
p.lookup.untrack(last)
|
||||
}
|
||||
if len(txs) == 0 {
|
||||
|
|
@ -741,11 +748,11 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
txs[len(txs)-1] = nil
|
||||
txs = txs[:len(txs)-1]
|
||||
|
||||
ids = append(ids, last.id)
|
||||
nonces = append(nonces, last.nonce)
|
||||
ids = append(ids, last.Id)
|
||||
nonces = append(nonces, last.Nonce)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], last.costCap)
|
||||
p.stored -= uint64(last.storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], last.CostCap)
|
||||
p.stored -= uint64(last.StorageSize)
|
||||
p.lookup.untrack(last)
|
||||
}
|
||||
p.index[addr] = txs
|
||||
|
|
@ -773,23 +780,13 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6
|
|||
// any of it since there's no clear error case. Some errors may be due to coding
|
||||
// issues, others caused by signers mining MEV stuff or swapping transactions. In
|
||||
// all cases, the pool needs to continue operating.
|
||||
func (p *BlobPool) offload(addr common.Address, nonce uint64, id uint64, inclusions map[common.Hash]uint64) {
|
||||
data, err := p.store.Get(id)
|
||||
if err != nil {
|
||||
log.Error("Blobs missing for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err)
|
||||
return
|
||||
}
|
||||
var tx types.Transaction
|
||||
if err = rlp.DecodeBytes(data, &tx); err != nil {
|
||||
log.Error("Blobs corrupted for included transaction", "from", addr, "nonce", nonce, "id", id, "err", err)
|
||||
return
|
||||
}
|
||||
block, ok := inclusions[tx.Hash()]
|
||||
func (p *BlobPool) offload(addr common.Address, blobTxMeta *blobTxMeta, inclusions map[common.Hash]uint64) {
|
||||
block, ok := inclusions[blobTxMeta.TxHash]
|
||||
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", blobTxMeta.Nonce, "id", blobTxMeta.Id)
|
||||
return
|
||||
}
|
||||
if err := p.limbo.push(&tx, block); err != nil {
|
||||
if err := p.limbo.push(blobTxMeta, block); err != nil {
|
||||
log.Warn("Failed to offload blob tx into limbo", "err", err)
|
||||
return
|
||||
}
|
||||
|
|
@ -835,8 +832,13 @@ func (p *BlobPool) Reset(oldHead, newHead *types.Header) {
|
|||
}
|
||||
}
|
||||
// Flush out any blobs from limbo that are older than the latest finality
|
||||
// and also delete the txs from the store.
|
||||
if p.chain.Config().IsCancun(p.head.Number, p.head.Time) {
|
||||
p.limbo.finalize(p.chain.CurrentFinalBlock())
|
||||
p.limbo.finalize(p.chain.CurrentFinalBlock(), func(id uint64, txHash common.Hash) {
|
||||
if err := p.store.Delete(id); err != nil {
|
||||
log.Error("Failed to delete blob transaction", "hash", txHash, "id", id, "err", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
// Reset the price heap for the new set of basefee/blobfee pairs
|
||||
var (
|
||||
|
|
@ -990,42 +992,25 @@ func (p *BlobPool) reorg(oldHead, newHead *types.Header) (map[common.Address][]*
|
|||
func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) error {
|
||||
// Retrieve the associated blob from the limbo. Without the blobs, we cannot
|
||||
// add the transaction back into the pool as it is not mineable.
|
||||
tx, err := p.limbo.pull(txhash)
|
||||
meta, err := p.limbo.pull(txhash)
|
||||
if err != nil {
|
||||
log.Error("Blobs unavailable, dropping reorged tx", "err", err)
|
||||
return err
|
||||
}
|
||||
// TODO: seems like an easy optimization here would be getting the serialized tx
|
||||
// from limbo instead of re-serializing it here.
|
||||
|
||||
// Serialize the transaction back into the primary datastore.
|
||||
blob, err := rlp.EncodeToBytes(tx)
|
||||
if err != nil {
|
||||
log.Error("Failed to encode transaction for storage", "hash", tx.Hash(), "err", err)
|
||||
return err
|
||||
}
|
||||
id, err := p.store.Put(blob)
|
||||
if err != nil {
|
||||
log.Error("Failed to write transaction into storage", "hash", tx.Hash(), "err", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Update the indices and metrics
|
||||
meta := newBlobTxMeta(id, tx.Size(), p.store.Size(id), tx)
|
||||
if _, ok := p.index[addr]; !ok {
|
||||
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", meta.TxHash, "from", addr, "err", err)
|
||||
return err
|
||||
}
|
||||
p.index[addr] = []*blobTxMeta{meta}
|
||||
p.spent[addr] = meta.costCap
|
||||
p.spent[addr] = meta.CostCap
|
||||
p.evict.Push(addr)
|
||||
} else {
|
||||
p.index[addr] = append(p.index[addr], meta)
|
||||
p.spent[addr] = new(uint256.Int).Add(p.spent[addr], meta.costCap)
|
||||
p.spent[addr] = new(uint256.Int).Add(p.spent[addr], meta.CostCap)
|
||||
}
|
||||
p.lookup.track(meta)
|
||||
p.stored += uint64(meta.storageSize)
|
||||
p.stored += uint64(meta.StorageSize)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -1043,24 +1028,24 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
|
|||
if old == nil || p.gasTip.Cmp(old) > 0 {
|
||||
for addr, txs := range p.index {
|
||||
for i, tx := range txs {
|
||||
if tx.execTipCap.Cmp(p.gasTip) < 0 {
|
||||
if tx.ExecTipCap.Cmp(p.gasTip) < 0 {
|
||||
// Drop the offending transaction
|
||||
var (
|
||||
ids = []uint64{tx.id}
|
||||
nonces = []uint64{tx.nonce}
|
||||
ids = []uint64{tx.Id}
|
||||
nonces = []uint64{tx.Nonce}
|
||||
)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].costCap)
|
||||
p.stored -= uint64(tx.storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], txs[i].CostCap)
|
||||
p.stored -= uint64(tx.StorageSize)
|
||||
p.lookup.untrack(tx)
|
||||
txs[i] = nil
|
||||
|
||||
// Drop everything afterwards, no gaps allowed
|
||||
for j, tx := range txs[i+1:] {
|
||||
ids = append(ids, tx.id)
|
||||
nonces = append(nonces, tx.nonce)
|
||||
ids = append(ids, tx.Id)
|
||||
nonces = append(nonces, tx.Nonce)
|
||||
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], tx.costCap)
|
||||
p.stored -= uint64(tx.storageSize)
|
||||
p.spent[addr] = new(uint256.Int).Sub(p.spent[addr], tx.CostCap)
|
||||
p.stored -= uint64(tx.StorageSize)
|
||||
p.lookup.untrack(tx)
|
||||
txs[i+1+j] = nil
|
||||
}
|
||||
|
|
@ -1076,7 +1061,7 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
|
|||
p.reserver.Release(addr)
|
||||
}
|
||||
// Clear out the transactions from the data store
|
||||
log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids)
|
||||
log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.Nonce, "tip", tx.ExecTipCap, "want", tip, "drop", nonces, "ids", ids)
|
||||
dropUnderpricedMeter.Mark(int64(len(ids)))
|
||||
|
||||
for _, id := range ids {
|
||||
|
|
@ -1136,7 +1121,7 @@ func (p *BlobPool) checkDelegationLimit(tx *types.Transaction) error {
|
|||
return nil
|
||||
}
|
||||
// If account already has a pending transaction, allow replacement only.
|
||||
if len(pending) == 1 && pending[0].nonce == tx.Nonce() {
|
||||
if len(pending) == 1 && pending[0].Nonce == tx.Nonce() {
|
||||
return nil
|
||||
}
|
||||
return txpool.ErrInflightTxLimitReached
|
||||
|
|
@ -1174,7 +1159,7 @@ func (p *BlobPool) validateTx(tx *types.Transaction) error {
|
|||
ExistingCost: func(addr common.Address, nonce uint64) *big.Int {
|
||||
next := p.state.GetNonce(addr)
|
||||
if uint64(len(p.index[addr])) > nonce-next {
|
||||
return p.index[addr][int(nonce-next)].costCap.ToBig()
|
||||
return p.index[addr][int(nonce-next)].CostCap.ToBig()
|
||||
}
|
||||
return nil
|
||||
},
|
||||
|
|
@ -1194,33 +1179,33 @@ func (p *BlobPool) validateTx(tx *types.Transaction) error {
|
|||
if uint64(len(p.index[from])) > tx.Nonce()-next {
|
||||
prev := p.index[from][int(tx.Nonce()-next)]
|
||||
// Ensure the transaction is different than the one tracked locally
|
||||
if prev.hash == tx.Hash() {
|
||||
if prev.TxHash == tx.Hash() {
|
||||
return txpool.ErrAlreadyKnown
|
||||
}
|
||||
// Account can support the replacement, but the price bump must also be met
|
||||
switch {
|
||||
case tx.GasFeeCapIntCmp(prev.execFeeCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx gas fee cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.GasFeeCap(), prev.execFeeCap)
|
||||
case tx.GasTipCapIntCmp(prev.execTipCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx gas tip cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.GasTipCap(), prev.execTipCap)
|
||||
case tx.BlobGasFeeCapIntCmp(prev.blobFeeCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx blob gas fee cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.BlobGasFeeCap(), prev.blobFeeCap)
|
||||
case tx.GasFeeCapIntCmp(prev.ExecFeeCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx gas fee cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.GasFeeCap(), prev.ExecFeeCap)
|
||||
case tx.GasTipCapIntCmp(prev.ExecTipCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx gas tip cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.GasTipCap(), prev.ExecTipCap)
|
||||
case tx.BlobGasFeeCapIntCmp(prev.BlobFeeCap.ToBig()) <= 0:
|
||||
return fmt.Errorf("%w: new tx blob gas fee cap %v <= %v queued", txpool.ErrReplaceUnderpriced, tx.BlobGasFeeCap(), prev.BlobFeeCap)
|
||||
}
|
||||
var (
|
||||
multiplier = uint256.NewInt(100 + p.config.PriceBump)
|
||||
onehundred = uint256.NewInt(100)
|
||||
|
||||
minGasFeeCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.execFeeCap), onehundred)
|
||||
minGasTipCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.execTipCap), onehundred)
|
||||
minBlobGasFeeCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.blobFeeCap), onehundred)
|
||||
minGasFeeCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.ExecFeeCap), onehundred)
|
||||
minGasTipCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.ExecTipCap), onehundred)
|
||||
minBlobGasFeeCap = new(uint256.Int).Div(new(uint256.Int).Mul(multiplier, prev.BlobFeeCap), onehundred)
|
||||
)
|
||||
switch {
|
||||
case tx.GasFeeCapIntCmp(minGasFeeCap.ToBig()) < 0:
|
||||
return fmt.Errorf("%w: new tx gas fee cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.GasFeeCap(), prev.execFeeCap, p.config.PriceBump)
|
||||
return fmt.Errorf("%w: new tx gas fee cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.GasFeeCap(), prev.ExecFeeCap, p.config.PriceBump)
|
||||
case tx.GasTipCapIntCmp(minGasTipCap.ToBig()) < 0:
|
||||
return fmt.Errorf("%w: new tx gas tip cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.GasTipCap(), prev.execTipCap, p.config.PriceBump)
|
||||
return fmt.Errorf("%w: new tx gas tip cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.GasTipCap(), prev.ExecTipCap, p.config.PriceBump)
|
||||
case tx.BlobGasFeeCapIntCmp(minBlobGasFeeCap.ToBig()) < 0:
|
||||
return fmt.Errorf("%w: new tx blob gas fee cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.BlobGasFeeCap(), prev.blobFeeCap, p.config.PriceBump)
|
||||
return fmt.Errorf("%w: new tx blob gas fee cap %v < %v queued + %d%% replacement penalty", txpool.ErrReplaceUnderpriced, tx.BlobGasFeeCap(), prev.BlobFeeCap, p.config.PriceBump)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
|
@ -1387,7 +1372,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
|
|||
|
||||
// Ensure the transaction is valid from all perspectives
|
||||
if err := p.validateTx(tx); err != nil {
|
||||
log.Trace("Transaction validation failed", "hash", tx.Hash(), "err", err)
|
||||
log.Trace("Transaction validation failed", "TxHash", tx.Hash(), "err", err)
|
||||
switch {
|
||||
case errors.Is(err, txpool.ErrUnderpriced):
|
||||
addUnderpricedMeter.Mark(1)
|
||||
|
|
@ -1456,18 +1441,18 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
|
|||
dropReplacedMeter.Mark(1)
|
||||
|
||||
prev := p.index[from][offset]
|
||||
if err := p.store.Delete(prev.id); err != nil {
|
||||
if err := p.store.Delete(prev.Id); err != nil {
|
||||
// Shitty situation, but try to recover gracefully instead of going boom
|
||||
log.Error("Failed to delete replaced transaction", "id", prev.id, "err", err)
|
||||
log.Error("Failed to delete replaced transaction", "id", prev.Id, "err", err)
|
||||
}
|
||||
// Update the transaction index
|
||||
p.index[from][offset] = meta
|
||||
p.spent[from] = new(uint256.Int).Sub(p.spent[from], prev.costCap)
|
||||
p.spent[from] = new(uint256.Int).Add(p.spent[from], meta.costCap)
|
||||
p.spent[from] = new(uint256.Int).Sub(p.spent[from], prev.CostCap)
|
||||
p.spent[from] = new(uint256.Int).Add(p.spent[from], meta.CostCap)
|
||||
|
||||
p.lookup.untrack(prev)
|
||||
p.lookup.track(meta)
|
||||
p.stored += uint64(meta.storageSize) - uint64(prev.storageSize)
|
||||
p.stored += uint64(meta.StorageSize) - uint64(prev.StorageSize)
|
||||
} else {
|
||||
// Transaction extends previously scheduled ones
|
||||
p.index[from] = append(p.index[from], meta)
|
||||
|
|
@ -1475,9 +1460,9 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
|
|||
p.spent[from] = new(uint256.Int)
|
||||
newacc = true
|
||||
}
|
||||
p.spent[from] = new(uint256.Int).Add(p.spent[from], meta.costCap)
|
||||
p.spent[from] = new(uint256.Int).Add(p.spent[from], meta.CostCap)
|
||||
p.lookup.track(meta)
|
||||
p.stored += uint64(meta.storageSize)
|
||||
p.stored += uint64(meta.StorageSize)
|
||||
}
|
||||
// Recompute the rolling eviction fields. In case of a replacement, this will
|
||||
// recompute all subsequent fields. In case of an append, this will only do
|
||||
|
|
@ -1487,7 +1472,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
|
|||
for i := offset; i < len(txs); i++ {
|
||||
// The first transaction will always use itself
|
||||
if i == 0 {
|
||||
txs[0].evictionExecTip = txs[0].execTipCap
|
||||
txs[0].evictionExecTip = txs[0].ExecTipCap
|
||||
txs[0].evictionExecFeeJumps = txs[0].basefeeJumps
|
||||
txs[0].evictionBlobFeeJumps = txs[0].blobfeeJumps
|
||||
|
||||
|
|
@ -1495,8 +1480,8 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
|
|||
}
|
||||
// Subsequent transactions will use a rolling calculation
|
||||
txs[i].evictionExecTip = txs[i-1].evictionExecTip
|
||||
if txs[i].evictionExecTip.Cmp(txs[i].execTipCap) > 0 {
|
||||
txs[i].evictionExecTip = txs[i].execTipCap
|
||||
if txs[i].evictionExecTip.Cmp(txs[i].ExecTipCap) > 0 {
|
||||
txs[i].evictionExecTip = txs[i].ExecTipCap
|
||||
}
|
||||
txs[i].evictionExecFeeJumps = txs[i-1].evictionExecFeeJumps
|
||||
if txs[i].evictionExecFeeJumps > txs[i].basefeeJumps {
|
||||
|
|
@ -1562,9 +1547,9 @@ func (p *BlobPool) drop() {
|
|||
txs = txs[:len(txs)-1]
|
||||
|
||||
p.index[from] = txs
|
||||
p.spent[from] = new(uint256.Int).Sub(p.spent[from], drop.costCap)
|
||||
p.spent[from] = new(uint256.Int).Sub(p.spent[from], drop.CostCap)
|
||||
}
|
||||
p.stored -= uint64(drop.storageSize)
|
||||
p.stored -= uint64(drop.StorageSize)
|
||||
p.lookup.untrack(drop)
|
||||
|
||||
// Remove the transaction from the pool's eviction heap:
|
||||
|
|
@ -1583,11 +1568,11 @@ func (p *BlobPool) drop() {
|
|||
}
|
||||
}
|
||||
// Remove the transaction from the data store
|
||||
log.Debug("Evicting overflown blob transaction", "from", from, "evicted", drop.nonce, "id", drop.id)
|
||||
log.Debug("Evicting overflown blob transaction", "from", from, "evicted", drop.Nonce, "id", drop.Id)
|
||||
dropOverflownMeter.Mark(1)
|
||||
|
||||
if err := p.store.Delete(drop.id); err != nil {
|
||||
log.Error("Failed to drop evicted transaction", "id", drop.id, "err", err)
|
||||
if err := p.store.Delete(drop.Id); err != nil {
|
||||
log.Error("Failed to drop evicted transaction", "id", drop.Id, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1622,31 +1607,31 @@ func (p *BlobPool) Pending(filter txpool.PendingFilter) map[common.Address][]*tx
|
|||
for _, tx := range txs {
|
||||
// If transaction filtering was requested, discard badly priced ones
|
||||
if filter.MinTip != nil && filter.BaseFee != nil {
|
||||
if tx.execFeeCap.Lt(filter.BaseFee) {
|
||||
if tx.ExecFeeCap.Lt(filter.BaseFee) {
|
||||
break // basefee too low, cannot be included, discard rest of txs from the account
|
||||
}
|
||||
tip := new(uint256.Int).Sub(tx.execFeeCap, filter.BaseFee)
|
||||
if tip.Gt(tx.execTipCap) {
|
||||
tip = tx.execTipCap
|
||||
tip := new(uint256.Int).Sub(tx.ExecFeeCap, filter.BaseFee)
|
||||
if tip.Gt(tx.ExecTipCap) {
|
||||
tip = tx.ExecTipCap
|
||||
}
|
||||
if tip.Lt(filter.MinTip) {
|
||||
break // allowed or remaining tip too low, cannot be included, discard rest of txs from the account
|
||||
}
|
||||
}
|
||||
if filter.BlobFee != nil {
|
||||
if tx.blobFeeCap.Lt(filter.BlobFee) {
|
||||
if tx.BlobFeeCap.Lt(filter.BlobFee) {
|
||||
break // blobfee too low, cannot be included, discard rest of txs from the account
|
||||
}
|
||||
}
|
||||
// Transaction was accepted according to the filter, append to the pending list
|
||||
lazies = append(lazies, &txpool.LazyTransaction{
|
||||
Pool: p,
|
||||
Hash: tx.hash,
|
||||
Hash: tx.TxHash,
|
||||
Time: execStart, // TODO(karalabe): Maybe save these and use that?
|
||||
GasFeeCap: tx.execFeeCap,
|
||||
GasTipCap: tx.execTipCap,
|
||||
Gas: tx.execGas,
|
||||
BlobGas: tx.blobGas,
|
||||
GasFeeCap: tx.ExecFeeCap,
|
||||
GasTipCap: tx.ExecTipCap,
|
||||
Gas: tx.ExecGas,
|
||||
BlobGas: tx.BlobGas,
|
||||
})
|
||||
}
|
||||
if len(lazies) > 0 {
|
||||
|
|
@ -1750,7 +1735,7 @@ func (p *BlobPool) Nonce(addr common.Address) uint64 {
|
|||
defer p.lock.Unlock()
|
||||
|
||||
if txs, ok := p.index[addr]; ok {
|
||||
return txs[len(txs)-1].nonce + 1
|
||||
return txs[len(txs)-1].Nonce + 1
|
||||
}
|
||||
return p.state.GetNonce(addr)
|
||||
}
|
||||
|
|
@ -1769,7 +1754,7 @@ func (p *BlobPool) Stats() (int, int) {
|
|||
}
|
||||
|
||||
// Content retrieves the data content of the transaction pool, returning all the
|
||||
// pending as well as queued transactions, grouped by account and sorted by nonce.
|
||||
// pending as well as queued transactions, grouped by account and sorted by Nonce.
|
||||
//
|
||||
// For the blob pool, this method will return nothing for now.
|
||||
// TODO(karalabe): Abstract out the returned metadata.
|
||||
|
|
|
|||
|
|
@ -283,10 +283,10 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
seen := make(map[common.Hash]struct{})
|
||||
for addr, txs := range pool.index {
|
||||
for _, tx := range txs {
|
||||
if _, ok := seen[tx.hash]; ok {
|
||||
t.Errorf("duplicate hash #%x in transaction index: address %s, nonce %d", tx.hash, addr, tx.nonce)
|
||||
if _, ok := seen[tx.TxHash]; ok {
|
||||
t.Errorf("duplicate hash #%x in transaction index: address %s, nonce %d", tx.TxHash, addr, tx.Nonce)
|
||||
}
|
||||
seen[tx.hash] = struct{}{}
|
||||
seen[tx.TxHash] = struct{}{}
|
||||
}
|
||||
}
|
||||
for hash, id := range pool.lookup.txIndex {
|
||||
|
|
@ -302,11 +302,11 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
blobs := make(map[common.Hash]map[common.Hash]struct{})
|
||||
for _, txs := range pool.index {
|
||||
for _, tx := range txs {
|
||||
for _, vhash := range tx.vhashes {
|
||||
for _, vhash := range tx.VHashes {
|
||||
if blobs[vhash] == nil {
|
||||
blobs[vhash] = make(map[common.Hash]struct{})
|
||||
}
|
||||
blobs[vhash][tx.hash] = struct{}{}
|
||||
blobs[vhash][tx.TxHash] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -328,18 +328,18 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
// and that the first nonce is the next expected one based on the state.
|
||||
for addr, txs := range pool.index {
|
||||
for i := 1; i < len(txs); i++ {
|
||||
if txs[i].nonce != txs[i-1].nonce+1 {
|
||||
t.Errorf("addr %v, tx %d nonce mismatch: have %d, want %d", addr, i, txs[i].nonce, txs[i-1].nonce+1)
|
||||
if txs[i].Nonce != txs[i-1].Nonce+1 {
|
||||
t.Errorf("addr %v, tx %d nonce mismatch: have %d, want %d", addr, i, txs[i].Nonce, txs[i-1].Nonce+1)
|
||||
}
|
||||
}
|
||||
if txs[0].nonce != pool.state.GetNonce(addr) {
|
||||
t.Errorf("addr %v, first tx nonce mismatch: have %d, want %d", addr, txs[0].nonce, pool.state.GetNonce(addr))
|
||||
if txs[0].Nonce != pool.state.GetNonce(addr) {
|
||||
t.Errorf("addr %v, first tx nonce mismatch: have %d, want %d", addr, txs[0].Nonce, pool.state.GetNonce(addr))
|
||||
}
|
||||
}
|
||||
// Verify that calculated evacuation thresholds are correct
|
||||
for addr, txs := range pool.index {
|
||||
if !txs[0].evictionExecTip.Eq(txs[0].execTipCap) {
|
||||
t.Errorf("addr %v, tx %d eviction execution tip mismatch: have %d, want %d", addr, 0, txs[0].evictionExecTip, txs[0].execTipCap)
|
||||
if !txs[0].evictionExecTip.Eq(txs[0].ExecTipCap) {
|
||||
t.Errorf("addr %v, tx %d eviction execution tip mismatch: have %d, want %d", addr, 0, txs[0].evictionExecTip, txs[0].ExecTipCap)
|
||||
}
|
||||
if math.Abs(txs[0].evictionExecFeeJumps-txs[0].basefeeJumps) > 0.001 {
|
||||
t.Errorf("addr %v, tx %d eviction execution fee jumps mismatch: have %f, want %f", addr, 0, txs[0].evictionExecFeeJumps, txs[0].basefeeJumps)
|
||||
|
|
@ -349,8 +349,8 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
}
|
||||
for i := 1; i < len(txs); i++ {
|
||||
wantExecTip := txs[i-1].evictionExecTip
|
||||
if wantExecTip.Gt(txs[i].execTipCap) {
|
||||
wantExecTip = txs[i].execTipCap
|
||||
if wantExecTip.Gt(txs[i].ExecTipCap) {
|
||||
wantExecTip = txs[i].ExecTipCap
|
||||
}
|
||||
if !txs[i].evictionExecTip.Eq(wantExecTip) {
|
||||
t.Errorf("addr %v, tx %d eviction execution tip mismatch: have %d, want %d", addr, i, txs[i].evictionExecTip, wantExecTip)
|
||||
|
|
@ -377,7 +377,7 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
for addr, txs := range pool.index {
|
||||
spent := new(uint256.Int)
|
||||
for _, tx := range txs {
|
||||
spent.Add(spent, tx.costCap)
|
||||
spent.Add(spent, tx.CostCap)
|
||||
}
|
||||
if !pool.spent[addr].Eq(spent) {
|
||||
t.Errorf("addr %v expenditure mismatch: have %d, want %d", addr, pool.spent[addr], spent)
|
||||
|
|
@ -387,7 +387,7 @@ func verifyPoolInternals(t *testing.T, pool *BlobPool) {
|
|||
var stored uint64
|
||||
for _, txs := range pool.index {
|
||||
for _, tx := range txs {
|
||||
stored += uint64(tx.storageSize)
|
||||
stored += uint64(tx.StorageSize)
|
||||
}
|
||||
}
|
||||
if pool.stored != stored {
|
||||
|
|
@ -407,7 +407,7 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) {
|
|||
known := make(map[common.Hash]struct{})
|
||||
for _, txs := range pool.index {
|
||||
for _, tx := range txs {
|
||||
for _, vhash := range tx.vhashes {
|
||||
for _, vhash := range tx.VHashes {
|
||||
known[vhash] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
|
@ -736,36 +736,36 @@ func TestOpenDrops(t *testing.T) {
|
|||
alive := make(map[uint64]struct{})
|
||||
for _, txs := range pool.index {
|
||||
for _, tx := range txs {
|
||||
switch tx.id {
|
||||
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 {
|
||||
t.Errorf("dangling transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := filled[tx.id]; ok {
|
||||
t.Errorf("filled transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := overlapped[tx.id]; ok {
|
||||
t.Errorf("overlapped transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := gapped[tx.id]; ok {
|
||||
t.Errorf("gapped transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := underpaid[tx.id]; ok {
|
||||
t.Errorf("underpaid transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := outpriced[tx.id]; ok {
|
||||
t.Errorf("outpriced transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := exceeded[tx.id]; ok {
|
||||
t.Errorf("fully overdrafted transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := overdrafted[tx.id]; ok {
|
||||
t.Errorf("partially overdrafted transaction remained in storage: %d", tx.id)
|
||||
} else if _, ok := overcapped[tx.id]; 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 {
|
||||
t.Errorf("repeated nonce transaction remained in storage: %d", tx.id)
|
||||
if _, ok := dangling[tx.Id]; ok {
|
||||
t.Errorf("dangling transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := filled[tx.Id]; ok {
|
||||
t.Errorf("filled transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := overlapped[tx.Id]; ok {
|
||||
t.Errorf("overlapped transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := gapped[tx.Id]; ok {
|
||||
t.Errorf("gapped transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := underpaid[tx.Id]; ok {
|
||||
t.Errorf("underpaid transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := outpriced[tx.Id]; ok {
|
||||
t.Errorf("outpriced transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := exceeded[tx.Id]; ok {
|
||||
t.Errorf("fully overdrafted transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := overdrafted[tx.Id]; ok {
|
||||
t.Errorf("partially overdrafted transaction remained in storage: %d", tx.Id)
|
||||
} else if _, ok := overcapped[tx.Id]; 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 {
|
||||
t.Errorf("repeated Nonce transaction remained in storage: %d", tx.Id)
|
||||
} else {
|
||||
alive[tx.id] = struct{}{}
|
||||
alive[tx.Id] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -851,8 +851,8 @@ func TestOpenIndex(t *testing.T) {
|
|||
|
||||
// Verify that the transactions have been sorted by nonce (case 1)
|
||||
for i := 0; i < len(pool.index[addr]); i++ {
|
||||
if pool.index[addr][i].nonce != uint64(i) {
|
||||
t.Errorf("tx %d nonce mismatch: have %d, want %d", i, pool.index[addr][i].nonce, uint64(i))
|
||||
if pool.index[addr][i].Nonce != uint64(i) {
|
||||
t.Errorf("tx %d nonce mismatch: have %d, want %d", i, pool.index[addr][i].Nonce, uint64(i))
|
||||
}
|
||||
}
|
||||
// Verify that the cumulative fee minimums have been correctly calculated (case 2)
|
||||
|
|
|
|||
|
|
@ -145,12 +145,12 @@ func TestPriceHeapSorting(t *testing.T) {
|
|||
blobfeeJumps = dynamicFeeJumps(blobFee)
|
||||
)
|
||||
index[addr] = []*blobTxMeta{{
|
||||
id: uint64(j),
|
||||
storageSize: 128 * 1024,
|
||||
nonce: 0,
|
||||
execTipCap: execTip,
|
||||
execFeeCap: execFee,
|
||||
blobFeeCap: blobFee,
|
||||
Id: uint64(j),
|
||||
StorageSize: 128 * 1024,
|
||||
Nonce: 0,
|
||||
ExecTipCap: execTip,
|
||||
ExecFeeCap: execFee,
|
||||
BlobFeeCap: blobFee,
|
||||
basefeeJumps: basefeeJumps,
|
||||
blobfeeJumps: blobfeeJumps,
|
||||
evictionExecTip: execTip,
|
||||
|
|
@ -204,12 +204,12 @@ func benchmarkPriceHeapReinit(b *testing.B, datacap uint64) {
|
|||
blobfeeJumps = dynamicFeeJumps(blobFee)
|
||||
)
|
||||
index[addr] = []*blobTxMeta{{
|
||||
id: uint64(i),
|
||||
storageSize: 128 * 1024,
|
||||
nonce: 0,
|
||||
execTipCap: execTip,
|
||||
execFeeCap: execFee,
|
||||
blobFeeCap: blobFee,
|
||||
Id: uint64(i),
|
||||
StorageSize: 128 * 1024,
|
||||
Nonce: 0,
|
||||
ExecTipCap: execTip,
|
||||
ExecFeeCap: execFee,
|
||||
BlobFeeCap: blobFee,
|
||||
basefeeJumps: basefeeJumps,
|
||||
blobfeeJumps: blobfeeJumps,
|
||||
evictionExecTip: execTip,
|
||||
|
|
@ -280,12 +280,12 @@ func benchmarkPriceHeapOverflow(b *testing.B, datacap uint64) {
|
|||
blobfeeJumps = dynamicFeeJumps(blobFee)
|
||||
)
|
||||
index[addr] = []*blobTxMeta{{
|
||||
id: uint64(i),
|
||||
storageSize: 128 * 1024,
|
||||
nonce: 0,
|
||||
execTipCap: execTip,
|
||||
execFeeCap: execFee,
|
||||
blobFeeCap: blobFee,
|
||||
Id: uint64(i),
|
||||
StorageSize: 128 * 1024,
|
||||
Nonce: 0,
|
||||
ExecTipCap: execTip,
|
||||
ExecFeeCap: execFee,
|
||||
BlobFeeCap: blobFee,
|
||||
basefeeJumps: basefeeJumps,
|
||||
blobfeeJumps: blobfeeJumps,
|
||||
evictionExecTip: execTip,
|
||||
|
|
@ -311,12 +311,12 @@ func benchmarkPriceHeapOverflow(b *testing.B, datacap uint64) {
|
|||
blobfeeJumps = dynamicFeeJumps(blobFee)
|
||||
)
|
||||
metas[i] = &blobTxMeta{
|
||||
id: uint64(int(blobs) + i),
|
||||
storageSize: 128 * 1024,
|
||||
nonce: 0,
|
||||
execTipCap: execTip,
|
||||
execFeeCap: execFee,
|
||||
blobFeeCap: blobFee,
|
||||
Id: uint64(int(blobs) + i),
|
||||
StorageSize: 128 * 1024,
|
||||
Nonce: 0,
|
||||
ExecTipCap: execTip,
|
||||
ExecFeeCap: execFee,
|
||||
BlobFeeCap: blobFee,
|
||||
basefeeJumps: basefeeJumps,
|
||||
blobfeeJumps: blobfeeJumps,
|
||||
evictionExecTip: execTip,
|
||||
|
|
|
|||
|
|
@ -30,9 +30,8 @@ import (
|
|||
// to which it belongs as well as the block number in which it was included for
|
||||
// finality eviction.
|
||||
type limboBlob struct {
|
||||
TxHash common.Hash // Owner transaction's hash to support resurrecting reorged txs
|
||||
Block uint64 // Block in which the blob transaction was included
|
||||
Tx *types.Transaction
|
||||
Block uint64 // Block in which the blob transaction was included
|
||||
BlobTxMeta *blobTxMeta
|
||||
}
|
||||
|
||||
// limbo is a light, indexed database to temporarily store recently included
|
||||
|
|
@ -43,15 +42,15 @@ type limboBlob struct {
|
|||
type limbo struct {
|
||||
store billy.Database // Persistent data store for limboed blobs
|
||||
|
||||
index map[common.Hash]uint64 // Mappings from tx hashes to datastore ids
|
||||
groups map[uint64]map[uint64]common.Hash // Set of txs included in past blocks
|
||||
index map[common.Hash]uint64 // Mappings from tx hashes to datastore ids
|
||||
limbos map[uint64]*limboBlob // Mappings from datastore ids to limboBlobs
|
||||
}
|
||||
|
||||
// newLimbo opens and indexes a set of limboed blob transactions.
|
||||
func newLimbo(datadir string, maxBlobsPerTransaction int) (*limbo, error) {
|
||||
l := &limbo{
|
||||
index: make(map[common.Hash]uint64),
|
||||
groups: make(map[uint64]map[uint64]common.Hash),
|
||||
limbos: make(map[uint64]*limboBlob),
|
||||
}
|
||||
// Index all limboed blobs on disk and delete anything unprocessable
|
||||
var fails []uint64
|
||||
|
|
@ -94,56 +93,55 @@ func (l *limbo) parseBlob(id uint64, data []byte) error {
|
|||
log.Error("Failed to decode blob limbo entry", "id", id, "err", err)
|
||||
return err
|
||||
}
|
||||
if _, ok := l.index[item.TxHash]; ok {
|
||||
txHash := item.BlobTxMeta.TxHash
|
||||
if _, ok := l.index[txHash]; ok {
|
||||
// This path is impossible, unless due to a programming error a blob gets
|
||||
// inserted into the limbo which was already part of if. Recover gracefully
|
||||
// by ignoring this data entry.
|
||||
log.Error("Dropping duplicate blob limbo entry", "owner", item.TxHash, "id", id)
|
||||
log.Error("Dropping duplicate blob limbo entry", "owner", txHash, "id", id)
|
||||
return errors.New("duplicate blob")
|
||||
}
|
||||
l.index[item.TxHash] = id
|
||||
|
||||
if _, ok := l.groups[item.Block]; !ok {
|
||||
l.groups[item.Block] = make(map[uint64]common.Hash)
|
||||
}
|
||||
l.groups[item.Block][id] = item.TxHash
|
||||
l.index[txHash] = id
|
||||
l.limbos[id] = item
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// finalize evicts all blobs belonging to a recently finalized block or older.
|
||||
func (l *limbo) finalize(final *types.Header) {
|
||||
func (l *limbo) finalize(final *types.Header, fn func(id uint64, txHash common.Hash)) {
|
||||
// Just in case there's no final block yet (network not yet merged, weird
|
||||
// restart, sethead, etc), fail gracefully.
|
||||
if final == nil {
|
||||
log.Error("Nil finalized block cannot evict old blobs")
|
||||
return
|
||||
}
|
||||
for block, ids := range l.groups {
|
||||
if block > final.Number.Uint64() {
|
||||
for id, item := range l.limbos {
|
||||
if item.Block > final.Number.Uint64() {
|
||||
continue
|
||||
}
|
||||
for id, owner := range ids {
|
||||
if err := l.store.Delete(id); err != nil {
|
||||
log.Error("Failed to drop finalized blob", "block", block, "id", id, "err", err)
|
||||
}
|
||||
delete(l.index, owner)
|
||||
// Delete limbo metadata.
|
||||
if err := l.store.Delete(id); err != nil {
|
||||
log.Error("Failed to drop finalized blob", "block", item.Block, "id", id, "err", err)
|
||||
}
|
||||
delete(l.index, item.BlobTxMeta.TxHash)
|
||||
delete(l.limbos, id)
|
||||
if fn != nil { // Delete blob tx if fn is not null.
|
||||
fn(item.BlobTxMeta.Id, item.BlobTxMeta.TxHash)
|
||||
}
|
||||
delete(l.groups, block)
|
||||
}
|
||||
}
|
||||
|
||||
// push stores a new blob transaction into the limbo, waiting until finality for
|
||||
// it to be automatically evicted.
|
||||
func (l *limbo) push(tx *types.Transaction, block uint64) error {
|
||||
func (l *limbo) push(metaData *blobTxMeta, block uint64) error {
|
||||
// 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.
|
||||
if _, ok := l.index[tx.Hash()]; ok {
|
||||
log.Error("Limbo cannot push already tracked blobs", "tx", tx)
|
||||
if _, ok := l.index[metaData.TxHash]; ok {
|
||||
log.Error("Limbo cannot push already tracked blobs", "tx", metaData.TxHash)
|
||||
return errors.New("already tracked blob transaction")
|
||||
}
|
||||
if err := l.setAndIndex(tx, block); err != nil {
|
||||
log.Error("Failed to set and index limboed blobs", "tx", tx, "err", err)
|
||||
if err := l.setAndIndex(metaData, block); err != nil {
|
||||
log.Error("Failed to set and index limboed blobs", "tx", metaData.TxHash, "err", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
|
|
@ -152,7 +150,7 @@ func (l *limbo) push(tx *types.Transaction, block uint64) error {
|
|||
// 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
|
||||
// transaction gets reorged out.
|
||||
func (l *limbo) pull(tx common.Hash) (*types.Transaction, error) {
|
||||
func (l *limbo) pull(tx common.Hash) (*blobTxMeta, error) {
|
||||
// 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
|
||||
// into the network first.
|
||||
|
|
@ -166,7 +164,10 @@ func (l *limbo) pull(tx common.Hash) (*types.Transaction, error) {
|
|||
log.Error("Failed to get and drop limboed blobs", "tx", tx, "id", id, "err", err)
|
||||
return nil, err
|
||||
}
|
||||
return item.Tx, nil
|
||||
meta := item.BlobTxMeta
|
||||
meta.basefeeJumps = dynamicFeeJumps(meta.ExecFeeCap)
|
||||
meta.blobfeeJumps = dynamicFeeJumps(meta.BlobFeeCap)
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// update changes the block number under which a blob transaction is tracked. This
|
||||
|
|
@ -187,7 +188,7 @@ func (l *limbo) update(txhash common.Hash, block uint64) {
|
|||
}
|
||||
// If there was no change in the blob's inclusion block, don't mess around
|
||||
// with heavy database operations.
|
||||
if _, ok := l.groups[block][id]; ok {
|
||||
if item, ok := l.limbos[id]; ok && item.Block == block {
|
||||
log.Trace("Blob transaction unchanged in limbo", "tx", txhash, "block", block)
|
||||
return
|
||||
}
|
||||
|
|
@ -198,7 +199,7 @@ func (l *limbo) update(txhash common.Hash, block uint64) {
|
|||
log.Error("Failed to get and drop limboed blobs", "tx", txhash, "id", id, "err", err)
|
||||
return
|
||||
}
|
||||
if err := l.setAndIndex(item.Tx, block); err != nil {
|
||||
if err := l.setAndIndex(item.BlobTxMeta, block); err != nil {
|
||||
log.Error("Failed to set and index limboed blobs", "tx", txhash, "err", err)
|
||||
return
|
||||
}
|
||||
|
|
@ -208,19 +209,9 @@ func (l *limbo) update(txhash common.Hash, block uint64) {
|
|||
// getAndDrop retrieves a blob item from the limbo store and deletes it both from
|
||||
// the store and indices.
|
||||
func (l *limbo) getAndDrop(id uint64) (*limboBlob, error) {
|
||||
data, err := l.store.Get(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item := new(limboBlob)
|
||||
if err = rlp.DecodeBytes(data, item); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
delete(l.index, item.TxHash)
|
||||
delete(l.groups[item.Block], id)
|
||||
if len(l.groups[item.Block]) == 0 {
|
||||
delete(l.groups, item.Block)
|
||||
}
|
||||
item := l.limbos[id]
|
||||
delete(l.index, item.BlobTxMeta.TxHash)
|
||||
delete(l.limbos, id)
|
||||
if err := l.store.Delete(id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -229,12 +220,10 @@ func (l *limbo) getAndDrop(id uint64) (*limboBlob, error) {
|
|||
|
||||
// setAndIndex assembles a limbo blob database entry and stores it, also updating
|
||||
// the in-memory indices.
|
||||
func (l *limbo) setAndIndex(tx *types.Transaction, block uint64) error {
|
||||
txhash := tx.Hash()
|
||||
func (l *limbo) setAndIndex(metaData *blobTxMeta, block uint64) error {
|
||||
item := &limboBlob{
|
||||
TxHash: txhash,
|
||||
Block: block,
|
||||
Tx: tx,
|
||||
Block: block,
|
||||
BlobTxMeta: metaData,
|
||||
}
|
||||
data, err := rlp.EncodeToBytes(item)
|
||||
if err != nil {
|
||||
|
|
@ -244,10 +233,7 @@ func (l *limbo) setAndIndex(tx *types.Transaction, block uint64) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
l.index[txhash] = id
|
||||
if _, ok := l.groups[block]; !ok {
|
||||
l.groups[block] = make(map[uint64]common.Hash)
|
||||
}
|
||||
l.groups[block][id] = txhash
|
||||
l.index[item.BlobTxMeta.TxHash] = id
|
||||
l.limbos[id] = item
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -83,16 +83,16 @@ func (l *lookup) sizeOfTx(txhash common.Hash) (uint64, bool) {
|
|||
// hashes; and from transaction hashes to datastore storage item ids.
|
||||
func (l *lookup) track(tx *blobTxMeta) {
|
||||
// Map all the blobs to the transaction hash
|
||||
for _, vhash := range tx.vhashes {
|
||||
for _, vhash := range tx.VHashes {
|
||||
if _, ok := l.blobIndex[vhash]; !ok {
|
||||
l.blobIndex[vhash] = make(map[common.Hash]struct{})
|
||||
}
|
||||
l.blobIndex[vhash][tx.hash] = struct{}{} // may be double mapped if a tx contains the same blob twice
|
||||
l.blobIndex[vhash][tx.TxHash] = struct{}{} // may be double mapped if a tx contains the same blob twice
|
||||
}
|
||||
// Map the transaction hash to the datastore id and RLP-encoded transaction size
|
||||
l.txIndex[tx.hash] = &txMetadata{
|
||||
id: tx.id,
|
||||
size: tx.size,
|
||||
l.txIndex[tx.TxHash] = &txMetadata{
|
||||
id: tx.Id,
|
||||
size: tx.Size,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -100,11 +100,11 @@ func (l *lookup) track(tx *blobTxMeta) {
|
|||
// hashes from the blob index.
|
||||
func (l *lookup) untrack(tx *blobTxMeta) {
|
||||
// Unmap the transaction hash from the datastore id
|
||||
delete(l.txIndex, tx.hash)
|
||||
delete(l.txIndex, tx.TxHash)
|
||||
|
||||
// Unmap all the blobs from the transaction hash
|
||||
for _, vhash := range tx.vhashes {
|
||||
delete(l.blobIndex[vhash], tx.hash) // may be double deleted if a tx contains the same blob twice
|
||||
for _, vhash := range tx.VHashes {
|
||||
delete(l.blobIndex[vhash], tx.TxHash) // may be double deleted if a tx contains the same blob twice
|
||||
if len(l.blobIndex[vhash]) == 0 {
|
||||
delete(l.blobIndex, vhash)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue