mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-18 22:14:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

blockchain/v2: fix "panic: duplicate block enqueued by processor" (#5499)

Browse Source 
When a peer is stopped due to some network issue, the Reactor calls scheduler#handleRemovePeer, which removes the peer from the scheduler. BUT the peer stays in the processor, which sometimes could lead to "duplicate block enqueued by processor" panic WHEN the same block is requested by the scheduler again from a different peer. The solution is to return scPeerError, which will be propagated to the processor. The processor will clean up the blocks associated with the peer in purgePeer.

Closes #5513, #5517
This commit is contained in:
Anton Kaliaev
2020-10-21 13:26:20 +04:00
parent 29ca7de63c
commit 79d535dd67
6 changed files with 124 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blockchain/v2/processor.go
+23 -8
View File
@@ -17,6 +17,11 @@ type pcBlockVerificationFailure struct {
secondPeerID p2p.ID
}
func (e pcBlockVerificationFailure) String() string {
return fmt.Sprintf("pcBlockVerificationFailure{%d 1st peer: %v, 2nd peer: %v}",
e.height, e.firstPeerID, e.secondPeerID)
}
// successful block execution
type pcBlockProcessed struct {
priorityNormal
@@ -24,6 +29,10 @@ type pcBlockProcessed struct {
peerID p2p.ID
}
func (e pcBlockProcessed) String() string {
return fmt.Sprintf("pcBlockProcessed{%d peer: %v}", e.height, e.peerID)
}
// processor has finished
type pcFinished struct {
priorityNormal
@@ -87,9 +96,12 @@ func (state *pcState) synced() bool {
}
func (state *pcState) enqueue(peerID p2p.ID, block *types.Block, height int64) {
if _, ok := state.queue[height]; ok {
panic("duplicate block enqueued by processor")
if item, ok := state.queue[height]; ok {
panic(fmt.Sprintf(
"duplicate block %d (%X) enqueued by processor (sent by %v; existing block %X from %v)",
height, block.Hash(), peerID, item.block.Hash(), item.peerID))
}
state.queue[height] = queueItem{block: block, peerID: peerID}
}
@@ -145,16 +157,20 @@ func (state *pcState) handle(event Event) (Event, error) {
}
return noOp, nil
}
first, second := firstItem.block, secondItem.block
firstParts := first.MakePartSet(types.BlockPartSizeBytes)
firstPartSetHeader := firstParts.Header()
firstID := types.BlockID{Hash: first.Hash(), PartSetHeader: firstPartSetHeader}
var (
first, second = firstItem.block, secondItem.block
firstParts = first.MakePartSet(types.BlockPartSizeBytes)
firstID = types.BlockID{Hash: first.Hash(), PartSetHeader: firstParts.Header()}
)
// verify if +second+ last commit "confirms" +first+ block
err = state.context.verifyCommit(tmState.ChainID, firstID, first.Height, second.LastCommit)
if err != nil {
state.purgePeer(firstItem.peerID)
state.purgePeer(secondItem.peerID)
if firstItem.peerID != secondItem.peerID {
state.purgePeer(secondItem.peerID)
}
return pcBlockVerificationFailure{
height: first.Height, firstPeerID: firstItem.peerID, secondPeerID: secondItem.peerID},
nil
@@ -170,7 +186,6 @@ func (state *pcState) handle(event Event) (Event, error) {
state.blocksSynced++
return pcBlockProcessed{height: first.Height, peerID: firstItem.peerID}, nil
}
return noOp, nil