forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 02:12:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core: rework reorg to use headers only; emit reverted logs in reverse

Browse source
This commit is contained in:
Péter Szilágyi 2024-10-16 18:30:20 +03:00
parent b0799c24e7
commit a6c4273a59
1 changed files with 92 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
core/blockchain.go
View file

@ -23,6 +23,7 @@ import (
"io" "io"
"math/big" "math/big"
"runtime" "runtime"
"slices"
"strings" "strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -1435,7 +1436,7 @@ func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (e
func (bc *BlockChain) writeKnownBlock(block *types.Block) error { func (bc *BlockChain) writeKnownBlock(block *types.Block) error {
current := bc.CurrentBlock() current := bc.CurrentBlock()
if block.ParentHash() != current.Hash() { if block.ParentHash() != current.Hash() {
if err := bc.reorg(current, block); err != nil { if err := bc.reorg(current, block.Header()); err != nil {
return err return err
} }
} }
@ -1541,7 +1542,7 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
// Reorganise the chain if the parent is not the head block // Reorganise the chain if the parent is not the head block
if block.ParentHash() != currentBlock.Hash() { if block.ParentHash() != currentBlock.Hash() {
if err := bc.reorg(currentBlock, block); err != nil { if err := bc.reorg(currentBlock, block.Header()); err != nil {
return NonStatTy, err return NonStatTy, err
} }
} }
@ -2154,8 +2155,8 @@ func (bc *BlockChain) recoverAncestors(block *types.Block, makeWitness bool) (co
return block.Hash(), nil return block.Hash(), nil
} }
// collectLogs collects the logs that were generated or removed during // collectLogs collects the logs that were generated or removed during the
// the processing of a block. These logs are later announced as deleted or reborn. // processing of a block. These logs are later announced as deleted or reborn.
func (bc *BlockChain) collectLogs(b *types.Block, removed bool) []*types.Log { func (bc *BlockChain) collectLogs(b *types.Block, removed bool) []*types.Log {
var blobGasPrice *big.Int var blobGasPrice *big.Int
excessBlobGas := b.ExcessBlobGas() excessBlobGas := b.ExcessBlobGas()
@ -2181,70 +2182,60 @@ func (bc *BlockChain) collectLogs(b *types.Block, removed bool) []*types.Log {
// reorg takes two blocks, an old chain and a new chain and will reconstruct the // reorg takes two blocks, an old chain and a new chain and will reconstruct the
// blocks and inserts them to be part of the new canonical chain and accumulates // blocks and inserts them to be part of the new canonical chain and accumulates
// potential missing transactions and post an event about them. // potential missing transactions and post an event about them.
//
// Note the new head block won't be processed here, callers need to handle it // Note the new head block won't be processed here, callers need to handle it
// externally. // externally.
func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error { func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error {
var ( var (
newChain []*types.Header newChain []*types.Header
oldChain []*types.Header oldChain []*types.Header
commonBlock *types.Block commonBlock *types.Header
deletedTxs []common.Hash
addedTxs []common.Hash
) )
oldBlock := bc.GetBlock(oldHead.Hash(), oldHead.Number.Uint64()) // Skip handling the newHead as it's handled outside
if oldBlock == nil { newHead = bc.GetHeader(newHead.ParentHash, newHead.Number.Uint64()-1)
return errors.New("current head block missing") if newHead == nil {
return errInvalidNewChain
} }
newBlock := newHead
// Reduce the longer chain to the same number as the shorter one // Reduce the longer chain to the same number as the shorter one
if oldBlock.NumberU64() > newBlock.NumberU64() { if oldHead.Number.Uint64() > newHead.Number.Uint64() {
// Old chain is longer, gather all transactions and logs as deleted ones // Old chain is longer, gather all transactions and logs as deleted ones
for ; oldBlock != nil && oldBlock.NumberU64() != newBlock.NumberU64(); oldBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1) { for ; oldHead != nil && oldHead.Number.Uint64() != newHead.Number.Uint64(); oldHead = bc.GetHeader(oldHead.ParentHash, oldHead.Number.Uint64()-1) {
oldChain = append(oldChain, oldBlock.Header()) oldChain = append(oldChain, oldHead)
for _, tx := range oldBlock.Transactions() {
deletedTxs = append(deletedTxs, tx.Hash())
}
} }
} else { } else {
// New chain is longer, stash all blocks away for subsequent insertion // New chain is longer, stash all blocks away for subsequent insertion
for ; newBlock != nil && newBlock.NumberU64() != oldBlock.NumberU64(); newBlock = bc.GetBlock(newBlock.ParentHash(), newBlock.NumberU64()-1) { for ; newHead != nil && newHead.Number.Uint64() != oldHead.Number.Uint64(); newHead = bc.GetHeader(newHead.ParentHash, newHead.Number.Uint64()-1) {
newChain = append(newChain, newBlock.Header()) newChain = append(newChain, newHead)
} }
} }
if oldBlock == nil { if oldHead == nil {
return errInvalidOldChain return errInvalidOldChain
} }
if newBlock == nil { if newHead == nil {
return errInvalidNewChain return errInvalidNewChain
} }
// Both sides of the reorg are at the same number, reduce both until the common // Both sides of the reorg are at the same number, reduce both until the common
// ancestor is found // ancestor is found
for { for {
// If the common ancestor was found, bail out // If the common ancestor was found, bail out
if oldBlock.Hash() == newBlock.Hash() { if oldHead.Hash() == newHead.Hash() {
commonBlock = oldBlock commonBlock = oldHead
break break
} }
// Remove an old block as well as stash away a new block // Remove an old block as well as stash away a new block
oldChain = append(oldChain, oldBlock.Header()) oldChain = append(oldChain, oldHead)
for _, tx := range oldBlock.Transactions() { newChain = append(newChain, newHead)
deletedTxs = append(deletedTxs, tx.Hash())
}
newChain = append(newChain, newBlock.Header())
// Step back with both chains // Step back with both chains
oldBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1) oldHead = bc.GetHeader(oldHead.ParentHash, oldHead.Number.Uint64()-1)
if oldBlock == nil { if oldHead == nil {
return errInvalidOldChain return errInvalidOldChain
} }
newBlock = bc.GetBlock(newBlock.ParentHash(), newBlock.NumberU64()-1) newHead = bc.GetHeader(newHead.ParentHash, newHead.Number.Uint64()-1)
if newBlock == nil { if newHead == nil {
return errInvalidNewChain return errInvalidNewChain
} }
} }
// Ensure the user sees large reorgs // Ensure the user sees large reorgs
if len(oldChain) > 0 && len(newChain) > 0 { if len(oldChain) > 0 && len(newChain) > 0 {
logFn := log.Info logFn := log.Info
@ -2253,7 +2244,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error {
msg = "Large chain reorg detected" msg = "Large chain reorg detected"
logFn = log.Warn logFn = log.Warn
} }
logFn(msg, "number", commonBlock.Number(), "hash", commonBlock.Hash(), logFn(msg, "number", commonBlock.Number, "hash", commonBlock.Hash(),
"drop", len(oldChain), "dropfrom", oldChain[0].Hash(), "add", len(newChain), "addfrom", newChain[0].Hash()) "drop", len(oldChain), "dropfrom", oldChain[0].Hash(), "add", len(newChain), "addfrom", newChain[0].Hash())
blockReorgAddMeter.Mark(int64(len(newChain))) blockReorgAddMeter.Mark(int64(len(newChain)))
blockReorgDropMeter.Mark(int64(len(oldChain))) blockReorgDropMeter.Mark(int64(len(oldChain)))
@ -2266,68 +2257,35 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error {
} else { } else {
// len(newChain) == 0 && len(oldChain) > 0 // len(newChain) == 0 && len(oldChain) > 0
// rewind the canonical chain to a lower point. // rewind the canonical chain to a lower point.
log.Error("Impossible reorg, please file an issue", "oldnum", oldBlock.Number(), "oldhash", oldBlock.Hash(), "oldblocks", len(oldChain), "newnum", newBlock.Number(), "newhash", newBlock.Hash(), "newblocks", len(newChain)) log.Error("Impossible reorg, please file an issue", "oldnum", oldHead.Number, "oldhash", oldHead.Hash(), "oldblocks", len(oldChain), "newnum", newHead.Number, "newhash", newHead.Hash(), "newblocks", len(newChain))
} }
// Acquire the tx-lookup lock before mutation. This step is essential // Acquire the tx-lookup lock before mutation. This step is essential
// as the txlookups should be changed atomically, and all subsequent // as the txlookups should be changed atomically, and all subsequent
// reads should be blocked until the mutation is complete. // reads should be blocked until the mutation is complete.
bc.txLookupLock.Lock() bc.txLookupLock.Lock()
// Insert the new chain segment in incremental order, from the old // Reorg can be executed, start reducing the chain's old blocks and appending
// to the new. The new chain head (newChain[0]) is not inserted here, // the new blocks
// as it will be handled separately outside of this function
for i := len(newChain) - 1; i >= 1; i-- {
// Insert the block in the canonical way, re-writing history
newBlock = bc.GetBlock(newChain[i].Hash(), newChain[i].Number.Uint64())
bc.writeHeadBlock(newBlock)
// Collect the new added transactions.
for _, tx := range newBlock.Transactions() {
addedTxs = append(addedTxs, tx.Hash())
}
}
// Delete useless indexes right now which includes the non-canonical
// transaction indexes, canonical chain indexes which above the head.
var ( var (
indexesBatch = bc.db.NewBatch() deletedTxs []common.Hash
diffs = types.HashDifference(deletedTxs, addedTxs) rebirthTxs []common.Hash
deletedLogs []*types.Log
rebirthLogs []*types.Log
) )
for _, tx := range diffs { // Undo old blocks in reverse order
rawdb.DeleteTxLookupEntry(indexesBatch, tx) for i := 0; i < len(oldChain); i++ {
} // Collect all the deleted transactions
// Delete all hash markers that are not part of the new canonical chain. block := bc.GetBlock(oldChain[i].Hash(), oldChain[i].Number.Uint64())
// Because the reorg function does not handle new chain head, all hash if block == nil {
// markers greater than or equal to new chain head should be deleted. return errInvalidOldChain // Corrupt database, mostly here to avoid weird panics
number := commonBlock.NumberU64()
if len(newChain) > 1 {
number = newChain[1].Number.Uint64()
}
for i := number + 1; ; i++ {
hash := rawdb.ReadCanonicalHash(bc.db, i)
if hash == (common.Hash{}) {
break
} }
rawdb.DeleteCanonicalHash(indexesBatch, i) for _, tx := range block.Transactions() {
} deletedTxs = append(deletedTxs, tx.Hash())
if err := indexesBatch.Write(); err != nil { }
log.Crit("Failed to delete useless indexes", "err", err) // Collect deleted logs and emit them
} if logs := bc.collectLogs(block, true); len(logs) > 0 {
// Reset the tx lookup cache to clear stale txlookup cache. slices.Reverse(logs) // Emit revertals latest first, older then
bc.txLookupCache.Purge()
// Release the tx-lookup lock after mutation.
bc.txLookupLock.Unlock()
// Send out events for logs from the old canon chain, and 'reborn'
// logs from the new canon chain. The number of logs can be very
// high, so the events are sent in batches of size around 512.
// Deleted logs + blocks:
var deletedLogs []*types.Log
for i := len(oldChain) - 1; i >= 0; i-- {
oldBlock = bc.GetBlock(oldChain[i].Hash(), oldChain[i].Number.Uint64())
// Collect deleted logs for notification
if logs := bc.collectLogs(oldBlock, true); len(logs) > 0 {
deletedLogs = append(deletedLogs, logs...) deletedLogs = append(deletedLogs, logs...)
} }
if len(deletedLogs) > 512 { if len(deletedLogs) > 512 {
@ -2338,21 +2296,59 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error {
if len(deletedLogs) > 0 { if len(deletedLogs) > 0 {
bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs}) bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs})
} }
// New logs: // Apply new blocks in forward order
var rebirthLogs []*types.Log for i := len(newChain) - 1; i > 0; i-- {
for i := len(newChain) - 1; i >= 1; i-- { // Collect all the included transactions
newBlock = bc.GetBlock(newChain[i].Hash(), newChain[i].Number.Uint64()) block := bc.GetBlock(newChain[i].Hash(), newChain[i].Number.Uint64())
if logs := bc.collectLogs(newBlock, false); len(logs) > 0 { if block == nil {
return errInvalidNewChain // Corrupt database, mostly here to avoid weird panics
}
for _, tx := range block.Transactions() {
rebirthTxs = append(rebirthTxs, tx.Hash())
}
// Collect inserted logs and emit them
if logs := bc.collectLogs(block, false); len(logs) > 0 {
rebirthLogs = append(rebirthLogs, logs...) rebirthLogs = append(rebirthLogs, logs...)
} }
if len(rebirthLogs) > 512 { if len(rebirthLogs) > 512 {
bc.logsFeed.Send(rebirthLogs) bc.logsFeed.Send(rebirthLogs)
rebirthLogs = nil rebirthLogs = nil
} }
// Update the head block
bc.writeHeadBlock(block)
} }
if len(rebirthLogs) > 0 { if len(rebirthLogs) > 0 {
bc.logsFeed.Send(rebirthLogs) bc.logsFeed.Send(rebirthLogs)
} }
// Delete useless indexes right now which includes the non-canonical
// transaction indexes, canonical chain indexes which above the head.
batch := bc.db.NewBatch()
for _, tx := range types.HashDifference(deletedTxs, rebirthTxs) {
rawdb.DeleteTxLookupEntry(batch, tx)
}
// Delete all hash markers that are not part of the new canonical chain.
// Because the reorg function does not handle new chain head, all hash
// markers greater than or equal to new chain head should be deleted.
number := commonBlock.Number
if len(newChain) > 0 {
number = newChain[0].Number
}
for i := number.Uint64() + 1; ; i++ {
hash := rawdb.ReadCanonicalHash(bc.db, i)
if hash == (common.Hash{}) {
break
}
rawdb.DeleteCanonicalHash(batch, i)
}
if err := batch.Write(); err != nil {
log.Crit("Failed to delete useless indexes", "err", err)
}
// Reset the tx lookup cache to clear stale txlookup cache.
bc.txLookupCache.Purge()
// Release the tx-lookup lock after mutation.
bc.txLookupLock.Unlock()
return nil return nil
} }
@ -2390,7 +2386,7 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) {
// Run the reorg if necessary and set the given block as new head. // Run the reorg if necessary and set the given block as new head.
start := time.Now() start := time.Now()
if head.ParentHash() != bc.CurrentBlock().Hash() { if head.ParentHash() != bc.CurrentBlock().Hash() {
if err := bc.reorg(bc.CurrentBlock(), head); err != nil { if err := bc.reorg(bc.CurrentBlock(), head.Header()); err != nil {
return common.Hash{}, err return common.Hash{}, err
} }
} }