Merge branch 'master' of github.com:harmony-one/harmony into staking_tx

api/service/networkinfo/service.go

@@ -41,8 +41,8 @@ type Service struct {
 
 // ConnectionRetry set the number of retry of connection to bootnode in case the initial connection is failed
 var (
-	// retry for 10 minutes and give up then
-	ConnectionRetry = 300
+	// retry for 30s and give up then
+	ConnectionRetry = 15
 
 	// context
 	ctx context.Context

consensus/consensus.go

@@ -79,9 +79,6 @@ type Consensus struct {
 	// If the number of validators is less than minPeers, the consensus won't start
 	MinPeers int
 
-	// Leader's address
-	leader p2p.Peer
-
 	CommitteePublicKeys map[string]bool
 
 	pubKeyLock sync.Mutex

consensus/consensus_service.go

@@ -85,17 +85,6 @@ func (consensus *Consensus) signAndMarshalConsensusMessage(message *msg_pb.Messa
 	return marshaledMessage, nil
 }
 
-// SetLeaderPubKey deserialize the public key of consensus leader
-func (consensus *Consensus) SetLeaderPubKey(k []byte) error {
-	consensus.leader.ConsensusPubKey = &bls.PublicKey{}
-	return consensus.leader.ConsensusPubKey.Deserialize(k)
-}
-
-// GetLeaderPubKey returns the public key of consensus leader
-func (consensus *Consensus) GetLeaderPubKey() *bls.PublicKey {
-	return consensus.leader.ConsensusPubKey
-}
-
 // GetNodeIDs returns Node IDs of all nodes in the same shard
 func (consensus *Consensus) GetNodeIDs() []libp2p_peer.ID {
 	nodes := make([]libp2p_peer.ID, 0)
@@ -131,8 +120,7 @@ func (consensus *Consensus) UpdatePublicKeys(pubKeys []*bls.PublicKey) int64 {
 		utils.Logger().Info().Int("index", i).Str("BlsPubKey", pubKey).Msg("Member")
 		consensus.CommitteePublicKeys[pubKey] = true
 	}
-	// TODO: use pubkey to identify leader rather than p2p.Peer.
-	consensus.leader = p2p.Peer{ConsensusPubKey: pubKeys[0]}
+
 	consensus.LeaderPubKey = pubKeys[0]
 	utils.Logger().Info().
 		Str("info", consensus.LeaderPubKey.SerializeToHexStr()).Msg("My Leader")
@@ -498,6 +486,7 @@ func (consensus *Consensus) UpdateConsensusInformation() Mode {
 	}
 
 	// update public keys committee
+	oldLeader := consensus.LeaderPubKey
 	consensus.getLogger().Info().
 		Int("numPubKeys", len(pubKeys)).
 		Msg("[UpdateConsensusInformation] Successfully updated public keys")
@@ -525,6 +514,18 @@ func (consensus *Consensus) UpdateConsensusInformation() Mode {
 			if hasError {
 				return Syncing
 			}
+
+			// If the leader changed and I myself become the leader
+			if !consensus.LeaderPubKey.IsEqual(oldLeader) && consensus.LeaderPubKey.IsEqual(consensus.PubKey) {
+				go func() {
+					utils.Logger().Debug().
+						Str("myKey", consensus.PubKey.SerializeToHexStr()).
+						Uint64("viewID", consensus.viewID).
+						Uint64("block", consensus.blockNum).
+						Msg("[onEpochChange] I am the New Leader")
+					consensus.ReadySignal <- struct{}{}
+				}()
+			}
 			return Normal
 		}
 	}

consensus/consensus_v2.go

@@ -1026,6 +1026,7 @@ func (consensus *Consensus) tryCatchup() {
 		}
 		utils.Logger().Info().Msg("[TryCatchup] prepared message found to commit")
 
+		// TODO(Chao): Explain the reasoning for these code
 		consensus.blockHash = [32]byte{}
 		consensus.blockNum = consensus.blockNum + 1
 		consensus.viewID = msgs[0].ViewID + 1
@@ -1129,6 +1130,11 @@ func (consensus *Consensus) Start(blockChannel chan *types.Block, stopChan chan
 				utils.Logger().Info().Msg("Node is out of sync")
 
 			case newBlock := <-blockChannel:
+				// Debug code to trigger leader change.
+				//if consensus.ShardID == 0 && newBlock.NumberU64() == 2 && strings.Contains(consensus.PubKey.SerializeToHexStr(), "65f55eb") {
+				//	continue
+				//}
+
 				utils.Logger().Info().
 					Uint64("MsgBlockNum", newBlock.NumberU64()).
 					Msg("[ConsensusMainLoop] Received Proposed New Block!")
@@ -1223,6 +1229,7 @@ func (consensus *Consensus) Start(blockChannel chan *types.Block, stopChan chan
 				consensus.handleMessageUpdate(msg)
 
 			case viewID := <-consensus.commitFinishChan:
+				// Only Leader execute this condition
 				func() {
 					consensus.mutex.Lock()
 					defer consensus.mutex.Unlock()

consensus/view_change.go

@@ -355,6 +355,7 @@ func (consensus *Consensus) onViewChange(msg *msg_pb.Message) {
 		consensus.LeaderPubKey = consensus.PubKey
 		consensus.ResetState()
 		if len(consensus.m1Payload) == 0 {
+			// TODO(Chao): explain why ReadySignal is sent only in this case but not the other case.
 			go func() {
 				consensus.ReadySignal <- struct{}{}
 			}()

node/node_handler.go

@@ -315,8 +315,6 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block, commitSigAndBit
 			Err(err).
 			Msg("Error when adding new block")
 		return
-	} else if core.IsEpochLastBlock(newBlock) {
-		node.Consensus.UpdateConsensusInformation()
 	}
 
 	// Update last consensus time for metrics
@@ -349,6 +347,11 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block, commitSigAndBit
 	// Broadcast client requested missing cross shard receipts if there is any
 	node.BroadcastMissingCXReceipts()
 
+	// Update consensus keys at last so the change of leader status doesn't mess up normal flow
+	if core.IsEpochLastBlock(newBlock) {
+		node.Consensus.UpdateConsensusInformation()
+	}
+
 	// TODO chao: uncomment this after beacon syncing is stable
 	// node.Blockchain().UpdateCXReceiptsCheckpointsByBlock(newBlock)