forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-17 18:30:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core: refactor function reorg (#23761, #24616, #24996, #30600)

Browse source
This commit is contained in:
Daniel Liu 2024-12-25 14:10:44 +08:00
parent adba311c5c
commit cbdf0e69f9
2 changed files with 179 additions and 90 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

251
core/blockchain.go
View file

@ -84,6 +84,8 @@ var (
errInsertionInterrupted = errors.New("insertion is interrupted") errInsertionInterrupted = errors.New("insertion is interrupted")
errChainStopped = errors.New("blockchain is stopped") errChainStopped = errors.New("blockchain is stopped")
errInvalidOldChain = errors.New("invalid old chain")
errInvalidNewChain = errors.New("invalid new chain")
CheckpointCh = make(chan int) CheckpointCh = make(chan int)
) )
@ -1483,7 +1485,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
if reorg { if reorg {
// 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.Header(), block.Header()); err != nil {
return NonStatTy, err return NonStatTy, err
} }
} }
@ -1498,9 +1500,8 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
bc.writeHeadBlock(block, false) bc.writeHeadBlock(block, false)
// prepare set of masternodes for the next epoch // prepare set of masternodes for the next epoch
if bc.chainConfig.XDPoS != nil && ((block.NumberU64() % bc.chainConfig.XDPoS.Epoch) == (bc.chainConfig.XDPoS.Epoch - bc.chainConfig.XDPoS.Gap)) { if bc.chainConfig.XDPoS != nil && ((block.NumberU64() % bc.chainConfig.XDPoS.Epoch) == (bc.chainConfig.XDPoS.Epoch - bc.chainConfig.XDPoS.Gap)) {
err := bc.UpdateM1() if err := bc.UpdateM1(); err != nil {
if err != nil { log.Crit("Fail to update masternodes during writeBlockWithState", "number", block.Number, "hash", block.Hash().Hex(), "err", err)
log.Crit("Error when update masternodes set. Stopping node", "err", err, "blockNum", block.NumberU64())
} }
} }
} }
@ -2282,59 +2283,54 @@ 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.
func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error { func (bc *BlockChain) reorg(oldHead, newHead *types.Header) error {
var ( log.Warn("Reorg", "OldHash", oldHead.Hash().Hex(), "OldNum", oldHead.Number, "NewHash", newHead.Hash().Hex(), "NewNum", newHead.Number)
newChain types.Blocks
oldChain types.Blocks
commonBlock *types.Block
deletedTxs types.Transactions
addedTxs types.Transactions
deletedLogs []*types.Log
)
log.Warn("Reorg", "oldBlock hash", oldBlock.Hash().Hex(), "number", oldBlock.NumberU64(), "newBlock hash", newBlock.Hash().Hex(), "number", newBlock.NumberU64())
// first reduce whoever is higher bound var (
if oldBlock.NumberU64() > newBlock.NumberU64() { newChain []*types.Header
// reduce old chain oldChain []*types.Header
for ; oldBlock != nil && oldBlock.NumberU64() != newBlock.NumberU64(); oldBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1) { commonBlock *types.Header
oldChain = append(oldChain, oldBlock) )
deletedTxs = append(deletedTxs, oldBlock.Transactions()...)
if logs := bc.collectLogs(oldBlock, true); len(logs) > 0 { // Reduce the longer chain to the same number as the shorter one
deletedLogs = append(deletedLogs, logs...) if oldHead.Number.Uint64() > newHead.Number.Uint64() {
} // Old chain is longer, gather all transactions and logs as deleted ones
for ; oldHead != nil && oldHead.Number.Uint64() != newHead.Number.Uint64(); oldHead = bc.GetHeader(oldHead.ParentHash, oldHead.Number.Uint64()-1) {
oldChain = append(oldChain, oldHead)
} }
} else { } else {
// reduce new chain and append new chain blocks for inserting later on // 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) newChain = append(newChain, newHead)
} }
} }
if oldBlock == nil { if oldHead == nil {
return errors.New("invalid old chain") return errInvalidOldChain
} }
if newBlock == nil { if newHead == nil {
return errors.New("invalid new chain") return errInvalidNewChain
} }
// Both sides of the reorg are at the same number, reduce both until the common
// ancestor is found
for { for {
if oldBlock.Hash() == newBlock.Hash() { // If the common ancestor was found, bail out
commonBlock = oldBlock if oldHead.Hash() == newHead.Hash() {
commonBlock = oldHead
break break
} }
// Remove an old block as well as stash away a new block
oldChain = append(oldChain, oldHead)
newChain = append(newChain, newHead)
oldChain = append(oldChain, oldBlock) // Step back with both chains
newChain = append(newChain, newBlock) oldHead = bc.GetHeader(oldHead.ParentHash, oldHead.Number.Uint64()-1)
deletedTxs = append(deletedTxs, oldBlock.Transactions()...) if oldHead == nil {
if logs := bc.collectLogs(oldBlock, true); len(logs) > 0 { return errInvalidOldChain
deletedLogs = append(deletedLogs, logs...)
} }
newHead = bc.GetHeader(newHead.ParentHash, newHead.Number.Uint64()-1)
oldBlock, newBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1), bc.GetBlock(newBlock.ParentHash(), newBlock.NumberU64()-1) if newHead == nil {
if oldBlock == nil { return errInvalidNewChain
return errors.New("invalid old chain")
}
if newBlock == nil {
return errors.New("invalid new chain")
} }
} }
@ -2342,23 +2338,21 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
if xdpos, ok := bc.Engine().(*XDPoS.XDPoS); ok { if xdpos, ok := bc.Engine().(*XDPoS.XDPoS); ok {
latestCommittedBlock := xdpos.EngineV2.GetLatestCommittedBlockInfo() latestCommittedBlock := xdpos.EngineV2.GetLatestCommittedBlockInfo()
if latestCommittedBlock != nil { if latestCommittedBlock != nil {
currentBlock := bc.CurrentBlock() cmp := commonBlock.Number.Cmp(latestCommittedBlock.Number)
currentBlock.Number().Cmp(latestCommittedBlock.Number)
cmp := commonBlock.Number().Cmp(latestCommittedBlock.Number)
if cmp < 0 { if cmp < 0 {
for _, oldBlock := range oldChain { for _, oldBlock := range oldChain {
if oldBlock.Number().Cmp(latestCommittedBlock.Number) == 0 { if oldBlock.Number.Cmp(latestCommittedBlock.Number) == 0 {
if oldBlock.Hash() != latestCommittedBlock.Hash { if oldBlock.Hash() != latestCommittedBlock.Hash {
log.Error("Impossible reorg, please file an issue", "oldnum", oldBlock.Number(), "oldhash", oldBlock.Hash(), "committed hash", latestCommittedBlock.Hash) log.Error("Impossible reorg, please file an issue", "OldNum", oldBlock.Number, "OldHash", oldBlock.Hash().Hex(), "LatestCommittedHash", latestCommittedBlock.Hash.Hex())
} else { } else {
log.Warn("Stop reorg, blockchain is under forking attack", "old committed num", oldBlock.Number(), "old committed hash", oldBlock.Hash()) log.Warn("Stop reorg, blockchain is under forking attack", "OldCommittedNum", oldBlock.Number, "OldCommittedHash", oldBlock.Hash().Hex())
return fmt.Errorf("stop reorg, blockchain is under forking attack. old committed num %d, hash %x", oldBlock.Number(), oldBlock.Hash()) return fmt.Errorf("stop reorg, blockchain is under forking attack. OldCommitted num %d, hash %s", oldBlock.Number, oldBlock.Hash().Hex())
} }
} }
} }
} else if cmp == 0 { } else if cmp == 0 {
if commonBlock.Hash() != latestCommittedBlock.Hash { if commonBlock.Hash() != latestCommittedBlock.Hash {
log.Error("Impossible reorg, please file an issue", "oldnum", commonBlock.Number(), "oldhash", commonBlock.Hash(), "committed hash", latestCommittedBlock.Hash) log.Error("Impossible reorg, please file an issue", "OldNum", commonBlock.Number.Uint64(), "OldHash", commonBlock.Hash().Hex(), "LatestCommittedHash", latestCommittedBlock.Hash.Hex())
} }
} }
} }
@ -2366,69 +2360,146 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
// 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.Warn logFn := log.Info
msg := "Chain reorg detected"
if len(oldChain) > 63 { if len(oldChain) > 63 {
msg = "Large chain reorg detected"
logFn = log.Warn logFn = log.Warn
} }
logFn("Chain split detected", "number", commonBlock.Number(), "hash", commonBlock.Hash(), logFn(msg, "number", commonBlock.Number, "hash", commonBlock.Hash().Hex(),
"drop", len(oldChain), "dropfrom", oldChain[0].Hash(), "add", len(newChain), "addfrom", newChain[0].Hash()) "drop", len(oldChain), "dropfrom", oldChain[0].Hash().Hex(), "add", len(newChain), "addfrom", newChain[0].Hash().Hex())
blockReorgAddMeter.Mark(int64(len(newChain))) blockReorgAddMeter.Mark(int64(len(newChain)))
blockReorgDropMeter.Mark(int64(len(oldChain))) blockReorgDropMeter.Mark(int64(len(oldChain)))
blockReorgMeter.Mark(1) blockReorgMeter.Mark(1)
} else if len(newChain) > 0 {
// Special case happens in the post merge stage that current head is
// the ancestor of new head while these two blocks are not consecutive
log.Info("Extend chain", "add", len(newChain), "number", newChain[0].Number, "hash", newChain[0].Hash())
blockReorgAddMeter.Mark(int64(len(newChain)))
} else { } else {
log.Error("Impossible reorg, please file an issue", "oldnum", oldBlock.Number(), "oldhash", oldBlock.Hash(), "newnum", newBlock.Number(), "newhash", newBlock.Hash()) // len(newChain) == 0 && len(oldChain) > 0
// rewind the canonical chain to a lower point.
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))
} }
// Insert the new chain(except the head block(reverse order)), // Acquire the tx-lookup lock before mutation. This step is essential
// taking care of the proper incremental order. // as the txlookups should be changed atomically, and all subsequent
// reads should be blocked until the mutation is complete.
// bc.txLookupLock.Lock()
// Reorg can be executed, start reducing the chain's old blocks and appending
// the new blocks
var (
deletedTxs []common.Hash
rebirthTxs []common.Hash
deletedLogs []*types.Log
rebirthLogs []*types.Log
)
// Deleted log emission on the API uses forward order, which is borked, but
// we'll leave it in for legacy reasons.
//
// TODO(karalabe): This should be nuked out, no idea how, deprecate some APIs?
{
for i := len(oldChain) - 1; i >= 0; i-- {
block := bc.GetBlock(oldChain[i].Hash(), oldChain[i].Number.Uint64())
if block == nil {
return errInvalidOldChain // Corrupt database, mostly here to avoid weird panics
}
if logs := bc.collectLogs(block, true); len(logs) > 0 {
deletedLogs = append(deletedLogs, logs...)
}
if len(deletedLogs) > 512 {
go bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs})
deletedLogs = nil
}
// TODO(daniel): remove chainSideFeed, reference PR #30601
// Also send event for blocks removed from the canon chain.
// bc.chainSideFeed.Send(ChainSideEvent{Block: block})
}
if len(deletedLogs) > 0 {
go bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs})
}
}
// Undo old blocks in reverse order
for i := 0; i < len(oldChain); i++ {
// Collect all the deleted transactions
block := bc.GetBlock(oldChain[i].Hash(), oldChain[i].Number.Uint64())
if block == nil {
return errInvalidOldChain // Corrupt database, mostly here to avoid weird panics
}
for _, tx := range block.Transactions() {
deletedTxs = append(deletedTxs, tx.Hash())
}
// Collect deleted logs and emit them for new integrations
// if logs := bc.collectLogs(block, true); len(logs) > 0 {
// slices.Reverse(logs) // Emit revertals latest first, older then
// }
}
// Apply new blocks in forward order
for i := len(newChain) - 1; i >= 0; i-- { for i := len(newChain) - 1; i >= 0; i-- {
// insert the block in the canonical way, re-writing history // Collect all the included transactions
bc.writeHeadBlock(newChain[i], true) block := bc.GetBlock(newChain[i].Hash(), newChain[i].Number.Uint64())
if block == nil {
// Collect the new added transactions. return errInvalidNewChain // Corrupt database, mostly here to avoid weird panics
addedTxs = append(addedTxs, newChain[i].Transactions()...) }
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...)
}
if len(rebirthLogs) > 512 {
bc.logsFeed.Send(rebirthLogs)
rebirthLogs = nil
}
// Update the head block
bc.writeHeadBlock(block, true)
// prepare set of masternodes for the next epoch // prepare set of masternodes for the next epoch
if bc.chainConfig.XDPoS != nil && ((newChain[i].NumberU64() % bc.chainConfig.XDPoS.Epoch) == (bc.chainConfig.XDPoS.Epoch - bc.chainConfig.XDPoS.Gap)) { if bc.chainConfig.XDPoS != nil && ((block.NumberU64() % bc.chainConfig.XDPoS.Epoch) == (bc.chainConfig.XDPoS.Epoch - bc.chainConfig.XDPoS.Gap)) {
err := bc.UpdateM1() if err := bc.UpdateM1(); err != nil {
if err != nil { log.Crit("Fail to update masternodes during reorg", "number", block.Number, "hash", block.Hash().Hex(), "err", err)
log.Crit("Error when update masternodes set. Stopping node", "err", err, "blockNumber", newChain[i].NumberU64())
} }
} }
} }
if len(rebirthLogs) > 0 {
bc.logsFeed.Send(rebirthLogs)
}
// Delete useless indexes right now which includes the non-canonical // Delete useless indexes right now which includes the non-canonical
// transaction indexes, canonical chain indexes which above the head. // transaction indexes, canonical chain indexes which above the head.
indexesBatch := bc.db.NewBatch() batch := bc.db.NewBatch()
for _, tx := range types.TxDifference(deletedTxs, addedTxs) { for _, tx := range types.HashDifference(deletedTxs, rebirthTxs) {
rawdb.DeleteTxLookupEntry(indexesBatch, tx.Hash()) rawdb.DeleteTxLookupEntry(batch, tx)
} }
// Delete any canonical number assignments above the new head // Delete all hash markers that are not part of the new canonical chain.
number := bc.CurrentBlock().NumberU64() // Because the reorg function handles new chain head, all hash
for i := number + 1; ; i++ { // markers greater than 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) hash := rawdb.ReadCanonicalHash(bc.db, i)
if hash == (common.Hash{}) { if hash == (common.Hash{}) {
break break
} }
rawdb.DeleteCanonicalHash(indexesBatch, i) rawdb.DeleteCanonicalHash(batch, i)
} }
if err := indexesBatch.Write(); err != nil { if err := batch.Write(); err != nil {
log.Crit("Failed to delete useless indexes", "err", err) log.Crit("Failed to delete useless indexes", "err", err)
} }
// If any logs need to be fired, do it now. In theory we could avoid creating
// this goroutine if there are no events to fire, but realistcally that only // Reset the tx lookup cache to clear stale txlookup cache.
// ever happens if we're reorging empty blocks, which will only happen on idle // bc.txLookupCache.Purge()
// networks where performance is not an issue either way.
if len(deletedLogs) > 0 { // Release the tx-lookup lock after mutation.
go bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs}) // bc.txLookupLock.Unlock()
}
if len(oldChain) > 0 {
go func() {
for i := len(oldChain) - 1; i >= 0; i-- {
bc.chainSideFeed.Send(ChainSideEvent{Block: oldChain[i]})
}
}()
}
return nil return nil
} }

18
core/types/transaction.go
View file

@ -677,6 +677,24 @@ func TxDifference(a, b Transactions) (keep Transactions) {
return keep return keep
} }
// HashDifference returns a new set of hashes that are present in a but not in b.
func HashDifference(a, b []common.Hash) []common.Hash {
keep := make([]common.Hash, 0, len(a))
remove := make(map[common.Hash]struct{})
for _, hash := range b {
remove[hash] = struct{}{}
}
for _, hash := range a {
if _, ok := remove[hash]; !ok {
keep = append(keep, hash)
}
}
return keep
}
// TxByNonce implements the sort interface to allow sorting a list of transactions // TxByNonce implements the sort interface to allow sorting a list of transactions
// by their nonces. This is usually only useful for sorting transactions from a // by their nonces. This is usually only useful for sorting transactions from a
// single account, otherwise a nonce comparison doesn't make much sense. // single account, otherwise a nonce comparison doesn't make much sense.