Fix reorg when justification chain is split

consensus/consensus.go

@@ -142,5 +142,6 @@ type PoS interface {
 	GetJustifiedNumberAndHash(chain ChainHeaderReader, headers []*types.Header) (uint64, common.Hash, error)
 	GetFinalizedHeader(chain ChainHeaderReader, header *types.Header) *types.Header
 	VerifyVote(chain ChainHeaderReader, vote *types.VoteEnvelope) error
+	DecodeVoteAttestation(header *types.Header) *types.VoteAttestation
 	IsActiveValidatorAt(chain ChainHeaderReader, header *types.Header, checkVoteKeyFn func(bLSPublicKey *types.BLSPublicKey) bool) bool
 }

consensus/oasys/oasys.go

@@ -413,7 +413,7 @@ func (o *Oasys) getParent(chain consensus.ChainHeaderReader, header *types.Heade
 
 // verifyVoteAttestation checks whether the vote attestation in the header is valid.
 func (o *Oasys) verifyVoteAttestation(chain consensus.ChainHeaderReader, header *types.Header, parents []*types.Header, env *params.EnvironmentValue) error {
-	attestation, err := getVoteAttestationFromHeader(header, o.chainConfig, o.config, env)
+	attestation, err := getVoteAttestationFromHeader(header, o.chainConfig, o.config, env.IsEpoch(header.Number.Uint64()))
 	if err != nil {
 		return err
 	}
@@ -574,7 +574,7 @@ func getEnvironmentFromHeader(header *types.Header) (*params.EnvironmentValue, e
 }
 
 // getVoteAttestationFromHeader returns the vote attestation extracted from the header's extra field if exists.
-func getVoteAttestationFromHeader(header *types.Header, chainConfig *params.ChainConfig, oasysConfig *params.OasysConfig, env *params.EnvironmentValue) (*types.VoteAttestation, error) {
+func getVoteAttestationFromHeader(header *types.Header, chainConfig *params.ChainConfig, oasysConfig *params.OasysConfig, isEpoch bool) (*types.VoteAttestation, error) {
 	if len(header.Extra) <= extraVanity+extraSeal {
 		return nil, nil
 	}
@@ -584,8 +584,13 @@ func getVoteAttestationFromHeader(header *types.Header, chainConfig *params.Chai
 	}
 
 	var attestationBytes []byte
-	if env.IsEpoch(header.Number.Uint64()) {
-		num := int(header.Extra[extraVanity+envValuesLen])
+	if isEpoch {
+		// Strictly check the length because it might be called from the
+		// `DecodeVoteAttestation(...)` even though it is not actually an epoch block.
+		var num int
+		if len(header.Extra) >= extraVanity+envValuesLen {
+			num = int(header.Extra[extraVanity+envValuesLen])
+		}
 		if len(header.Extra) <= extraVanity+extraSeal+validatorNumberSize+num*validatorInfoBytesLen {
 			return nil, nil
 		}
@@ -608,6 +613,18 @@ func getVoteAttestationFromHeader(header *types.Header, chainConfig *params.Chai
 	return &attestation, nil
 }
 
+// Decode vote atestation from the block header. It is a wrapper method that allows
+// calls from outside the consensus engine. The provided block header may depend on
+// an unknown ancestor, so it must not access the Environment or Snapshot.
+func (o *Oasys) DecodeVoteAttestation(header *types.Header) *types.VoteAttestation {
+	attestation, _ := getVoteAttestationFromHeader(header, o.chainConfig, o.config, false)
+	if attestation == nil {
+		// Possible epoch block.
+		attestation, _ = getVoteAttestationFromHeader(header, o.chainConfig, o.config, true)
+	}
+	return attestation
+}
+
 // snapshot retrieves the authorization snapshot at a given point in time.
 // !!! be careful
 // the block with `number` and `hash` is just the last element of `parents`,

consensus/oasys/snapshot.go

@@ -155,7 +155,7 @@ func (s *Snapshot) updateAttestation(header *types.Header, chainConfig *params.C
 	}
 
 	// The attestation should have been checked in verify header, update directly
-	attestation, _ := getVoteAttestationFromHeader(header, chainConfig, oasysConfig, s.Environment)
+	attestation, _ := getVoteAttestationFromHeader(header, chainConfig, oasysConfig, s.Environment.IsEpoch(header.Number.Uint64()))
 	if attestation == nil {
 		return
 	}

eth/backend.go

@@ -45,6 +45,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
+	"github.com/ethereum/go-ethereum/eth/protocols/bsc"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -73,11 +74,12 @@ type Ethereum struct {
 	// Handlers
 	txPool *txpool.TxPool
 
-	blockchain         *core.BlockChain
-	handler            *handler
-	ethDialCandidates  enode.Iterator
-	snapDialCandidates enode.Iterator
-	merger             *consensus.Merger
+	blockchain          *core.BlockChain
+	handler             *handler
+	ethDialCandidates   enode.Iterator
+	snapDialCandidates  enode.Iterator
+	emptyDialCandidates enode.Iterator
+	merger              *consensus.Merger
 
 	// DB interfaces
 	chainDb ethdb.Database // Block chain database
@@ -314,6 +316,10 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 	if err != nil {
 		return nil, err
 	}
+	eth.emptyDialCandidates, err = dnsclient.NewIterator()
+	if err != nil {
+		return nil, err
+	}
 
 	// Start the RPC service
 	eth.netRPCService = ethapi.NewNetAPI(eth.p2pServer, networkID)
@@ -560,6 +566,7 @@ func (s *Ethereum) Protocols() []p2p.Protocol {
 	if s.config.SnapshotCache > 0 {
 		protos = append(protos, snap.MakeProtocols((*snapHandler)(s.handler), s.snapDialCandidates)...)
 	}
+	protos = append(protos, bsc.MakeProtocols((*bscHandler)(s.handler), s.emptyDialCandidates)...)
 	return protos
 }
 

eth/handler_eth.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
@@ -138,5 +139,13 @@ func (h *ethHandler) handleBlockBroadcast(peer *eth.Peer, block *types.Block, td
 		peer.SetHead(trueHead, trueTD)
 		h.chainSync.handlePeerEvent()
 	}
+	// Update the peer's justfied head if better than the previous
+	if pos, ok := h.chain.Engine().(consensus.PoS); ok {
+		if attestation := pos.DecodeVoteAttestation(block.Header()); attestation != nil {
+			if cur := peer.JustifiedBlock(); cur == nil || attestation.Data.SourceNumber > *cur {
+				peer.SetJustifiedBlock(attestation.Data.SourceNumber)
+			}
+		}
+	}
 	return nil
 }

eth/peerset.go

@@ -399,6 +399,26 @@ func (ps *peerSet) peerWithHighestTD() *eth.Peer {
 	return bestPeer
 }
 
+// peerWithHighestJustifiedBlockAndTD() returns the peer with the
+// highest justified block height. If multiple peers have the same
+// height, the one with the highest total difficulty is returned.
+func (ps *peerSet) peerWithHighestJustifiedBlockAndTD() (bestPeer *eth.Peer, bestJustified uint64, bestTd *big.Int) {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	for _, p := range ps.peers {
+		justified := p.JustifiedBlock()
+		if justified == nil {
+			continue
+		}
+		_, td := p.Head()
+		if *justified > bestJustified || (*justified == bestJustified && td.Cmp(bestTd) > 0) {
+			bestPeer, bestJustified, bestTd = p.Peer, *justified, td
+		}
+	}
+	return
+}
+
 // close disconnects all peers.
 func (ps *peerSet) close() {
 	ps.lock.Lock()

eth/protocols/eth/peer.go

@@ -75,6 +75,8 @@ type Peer struct {
 	head common.Hash // Latest advertised head block hash
 	td   *big.Int    // Latest advertised head block total difficulty
 
+	justifiedBlock *uint64 // Latest advertised justified block
+
 	knownBlocks     *knownCache            // Set of block hashes known to be known by this peer
 	queuedBlocks    chan *blockPropagation // Queue of blocks to broadcast to the peer
 	queuedBlockAnns chan *types.Block      // Queue of blocks to announce to the peer
@@ -156,6 +158,22 @@ func (p *Peer) SetHead(hash common.Hash, td *big.Int) {
 	p.td.Set(td)
 }
 
+// JustifiedBlock retrieves the latest advertised justified block of the peer.
+func (p *Peer) JustifiedBlock() *uint64 {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	return p.justifiedBlock
+}
+
+// SetJustifiedBlock updates the latest advertised justified block of the peer.
+func (p *Peer) SetJustifiedBlock(num uint64) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	p.justifiedBlock = &num
+}
+
 // KnownBlock returns whether peer is known to already have a block.
 func (p *Peer) KnownBlock(hash common.Hash) bool {
 	return p.knownBlocks.Contains(hash)

eth/sync.go

@@ -22,7 +22,9 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/log"
@@ -171,14 +173,31 @@ func (cs *chainSyncer) nextSyncOp() *chainSyncOp {
 	if cs.handler.peers.len() < minPeers {
 		return nil
 	}
+
+	// Find the peer with the highest justified block height.
+	mode, ourTD := cs.modeAndLocalHead()
+	if peer, peerJustified, peerTD := cs.handler.peers.peerWithHighestJustifiedBlockAndTD(); peer != nil {
+		if ourJustified := cs.getJustifiedBlockNumber(); ourJustified != nil {
+			// Ignored because justified block height is lower.
+			if peerJustified < *ourJustified {
+				return nil
+			}
+			// Ignored because justified block height is the same but TD is lower.
+			if peerJustified == *ourJustified && peerTD.Cmp(ourTD) < 0 {
+				return nil
+			}
+		}
+		return peerToSyncOp(mode, peer)
+	}
+
 	// We have enough peers, pick the one with the highest TD, but avoid going
 	// over the terminal total difficulty. Above that we expect the consensus
 	// clients to direct the chain head to sync to.
 	peer := cs.handler.peers.peerWithHighestTD()
 	if peer == nil {
 		return nil
 	}
-	mode, ourTD := cs.modeAndLocalHead()
+
 	op := peerToSyncOp(mode, peer)
 	if op.td.Cmp(ourTD) <= 0 {
 		// We seem to be in sync according to the legacy rules. In the merge
@@ -193,6 +212,25 @@ func (cs *chainSyncer) nextSyncOp() *chainSyncOp {
 	return op
 }
 
+// Retrieve the justified block number of the canonical chain from the consensus engine.
+func (cs *chainSyncer) getJustifiedBlockNumber() *uint64 {
+	head := cs.handler.chain.CurrentHeader()
+	if !cs.handler.chain.Config().IsFastFinalityEnabled(head.Number) {
+		return nil
+	}
+	pos, ok := cs.handler.chain.Engine().(consensus.PoS)
+	if !ok {
+		return nil
+	}
+	// Note: Do not use methods like `(*core.BlockChain).CurrentSafeBlock()`, as they may
+	// fall back to the latest block if the justified block cannot be retrieved.
+	justified, _, err := pos.GetJustifiedNumberAndHash(cs.handler.chain, []*types.Header{head})
+	if err != nil {
+		return nil
+	}
+	return &justified
+}
+
 func peerToSyncOp(mode downloader.SyncMode, p *eth.Peer) *chainSyncOp {
 	peerHead, peerTD := p.Head()
 	return &chainSyncOp{mode: mode, peer: p, td: peerTD, head: peerHead}