Feature: 1 second finality

consensus/consensus_service.go

@@ -365,6 +365,9 @@ func (consensus *Consensus) updateConsensusInformation(reason string) Mode {
 	if consensus.Blockchain().Config().IsTwoSeconds(nextEpoch) {
 		consensus.BlockPeriod = 2 * time.Second
 	}
+	if consensus.Blockchain().Config().IsOneSecond(nextEpoch) {
+		consensus.BlockPeriod = 1 * time.Second
+	}
 
 	isFirstTimeStaking := consensus.Blockchain().Config().IsStaking(nextEpoch) &&
 		curHeader.IsLastBlockInEpoch() && !consensus.Blockchain().Config().IsStaking(curEpoch)

consensus/consensus_v2.go

@@ -138,7 +138,7 @@ func (consensus *Consensus) HandleMessageUpdate(ctx context.Context, peer libp2p
 	return nil
 }
 
-func (consensus *Consensus) finalCommit(waitTime time.Duration, viewID uint64, isLeader bool) {
+func (consensus *Consensus) finalCommit(waitTime time.Duration, viewID uint64) {
 	consensus.getLogger().Info().Str("waitTime", waitTime.String()).
 		Msg("[OnCommit] Starting Grace Period")
 	time.Sleep(waitTime)
@@ -307,6 +307,157 @@ func (consensus *Consensus) _finalCommit(isLeader bool) {
 	}
 }
 
+// finalCommit uses locks, not suited to be called internally
+func (consensus *Consensus) finalCommit1s(isLeader bool, viewID uint64, nextBlockDue time.Time, network *NetworkMessage) {
+	waitTime := 0 * time.Millisecond
+	maxWaitTime := time.Until(nextBlockDue) - 200*time.Millisecond
+	if maxWaitTime > waitTime {
+		waitTime = maxWaitTime
+	}
+	consensus.GetLogger().Info().Str("waitTime", waitTime.String()).
+		Msg("[OnCommit] Starting Grace Period")
+	time.Sleep(waitTime)
+
+	consensus.mutex.Lock()
+	defer consensus.mutex.Unlock()
+	if viewID != consensus.getCurBlockViewID() {
+		return
+	}
+	numCommits := consensus.decider.SignersCount(quorum.Commit)
+
+	consensus.getLogger().Info().
+		Int64("NumCommits", numCommits).
+		Msg("[finalCommit] Finalizing Consensus")
+	beforeCatchupNum := consensus.getBlockNum()
+
+	//leaderPriKey, err := consensus.getConsensusLeaderPrivateKey()
+	//if err != nil {
+	//	consensus.getLogger().Error().Err(err).Msg("[finalCommit] leader not found")
+	//	return
+	//}
+	// Construct committed message
+	//network, err := consensus.construct(msg_pb.MessageType_COMMITTED, nil, []*bls.PrivateKeyWrapper{leaderPriKey})
+	//if err != nil {
+	//	consensus.getLogger().Warn().Err(err).
+	//		Msg("[finalCommit] Unable to construct Committed message")
+	//	return
+	//}
+	var (
+		msgToSend          = network.Bytes
+		FBFTMsg            = network.FBFTMsg
+		commitSigAndBitmap = FBFTMsg.Payload
+	)
+	consensus.fBFTLog.AddVerifiedMessage(FBFTMsg)
+	// find correct block content
+	curBlockHash := consensus.blockHash
+	block := consensus.fBFTLog.GetBlockByHash(curBlockHash)
+	if block == nil {
+		consensus.getLogger().Warn().
+			Str("blockHash", hex.EncodeToString(curBlockHash[:])).
+			Msg("[finalCommit] Cannot find block by hash")
+		return
+	}
+
+	if err := consensus.verifyLastCommitSig(commitSigAndBitmap, block); err != nil {
+		consensus.getLogger().Warn().Err(err).Msg("[finalCommit] failed verifying last commit sig")
+		return
+	}
+	consensus.getLogger().Info().Hex("new", commitSigAndBitmap).Msg("[finalCommit] Overriding commit signatures!!")
+
+	//if err := consensus.Blockchain().WriteCommitSig(block.NumberU64(), commitSigAndBitmap); err != nil {
+	//	consensus.getLogger().Warn().Err(err).Msg("[finalCommit] failed writting commit sig")
+	//}
+
+	// Send committed message before block insertion.
+	// if leader successfully finalizes the block, send committed message to validators
+	// Note: leader already sent 67% commit in preCommit. The 100% commit won't be sent immediately
+	// to save network traffic. It will only be sent in retry if consensus doesn't move forward.
+	// Or if the leader is changed for next block, the 100% committed sig will be sent to the next leader immediately.
+	if !isLeader || block.IsLastBlockInEpoch() {
+		// send immediately
+		if err := consensus.msgSender.SendWithRetry(
+			block.NumberU64(),
+			msg_pb.MessageType_COMMITTED, []nodeconfig.GroupID{
+				nodeconfig.NewGroupIDByShardID(nodeconfig.ShardID(consensus.ShardID)),
+			},
+			p2p.ConstructMessage(msgToSend)); err != nil {
+			consensus.getLogger().Warn().Err(err).Msg("[finalCommit] Cannot send committed message")
+		} else {
+			consensus.getLogger().Info().
+				Hex("blockHash", curBlockHash[:]).
+				Uint64("blockNum", consensus.BlockNum()).
+				Msg("[finalCommit] Sent Committed Message")
+		}
+		consensus.getLogger().Info().Msg("[finalCommit] Start consensus timer")
+		consensus.consensusTimeout[timeoutConsensus].Start()
+	} else {
+		// delayed send
+		consensus.msgSender.DelayedSendWithRetry(
+			block.NumberU64(),
+			msg_pb.MessageType_COMMITTED, []nodeconfig.GroupID{
+				nodeconfig.NewGroupIDByShardID(nodeconfig.ShardID(consensus.ShardID)),
+			},
+			p2p.ConstructMessage(msgToSend))
+		consensus.getLogger().Info().
+			Hex("blockHash", curBlockHash[:]).
+			Uint64("blockNum", consensus.BlockNum()).
+			Hex("lastCommitSig", commitSigAndBitmap).
+			Msg("[finalCommit] Queued Committed Message")
+	}
+
+	block.SetCurrentCommitSig(commitSigAndBitmap)
+	err := consensus.commitBlock(block, FBFTMsg)
+
+	if err != nil || consensus.BlockNum()-beforeCatchupNum != 1 {
+		consensus.getLogger().Err(err).
+			Uint64("beforeCatchupBlockNum", beforeCatchupNum).
+			Msg("[finalCommit] Leader failed to commit the confirmed block")
+	}
+
+	// Dump new block into level db
+	// In current code, we add signatures in block in tryCatchup, the block dump to explorer does not contains signatures
+	// but since explorer doesn't need signatures, it should be fine
+	// in future, we will move signatures to next block
+	//explorer.GetStorageInstance(consensus.leader.IP, consensus.leader.Port, true).Dump(block, beforeCatchupNum)
+
+	if consensus.consensusTimeout[timeoutBootstrap].IsActive() {
+		consensus.consensusTimeout[timeoutBootstrap].Stop()
+		consensus.getLogger().Info().Msg("[finalCommit] stop bootstrap timer only once")
+	}
+
+	consensus.getLogger().Info().
+		Uint64("blockNum", block.NumberU64()).
+		Uint64("epochNum", block.Epoch().Uint64()).
+		Uint64("ViewId", block.Header().ViewID().Uint64()).
+		Str("blockHash", block.Hash().String()).
+		Int("numTxns", len(block.Transactions())).
+		Int("numStakingTxns", len(block.StakingTransactions())).
+		Msg("HOORAY!!!!!!! CONSENSUS REACHED!!!!!!!")
+
+	consensus.UpdateLeaderMetrics(float64(numCommits), float64(block.NumberU64()))
+
+	// If still the leader, send commit sig/bitmap to finish the new block proposal,
+	// else, the block proposal will timeout by itself.
+	if isLeader {
+		if block.IsLastBlockInEpoch() {
+			// No pipelining
+			go func() {
+				consensus.getLogger().Info().Msg("[finalCommit] sending block proposal signal")
+				consensus.ReadySignal(NewProposal(SyncProposal))
+			}()
+		} else {
+			// pipelining
+			go func() {
+				select {
+				case consensus.GetCommitSigChannel() <- commitSigAndBitmap:
+				case <-time.After(CommitSigSenderTimeout):
+					utils.Logger().Error().Err(err).Msg("[finalCommit] channel not received after 6s for commitSigAndBitmap")
+				}
+			}()
+		}
+	}
+}
+
 // BlockCommitSigs returns the byte array of aggregated
 // commit signature and bitmap signed on the block
 func (consensus *Consensus) BlockCommitSigs(blockNum uint64) ([]byte, error) {
@@ -721,6 +872,9 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int, defaultKey *bls.PublicK
 		curBlock  = bc.CurrentBlock()
 		curNumber = curBlock.NumberU64()
 		curEpoch  = curBlock.Epoch().Uint64()
+		wasFound  = false
+		next      *bls.PublicKeyWrapper
+		offset    = 1
 	)
 	if epoch.Uint64() != curEpoch {
 		return defaultKey
@@ -741,6 +895,21 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int, defaultKey *bls.PublicK
 		utils.Logger().Error().Err(err).Msg("Failed to find committee")
 		return defaultKey
 	}
+
+	if bc.Config().IsRotationEachBlock(epoch) {
+		if bc.Config().IsLeaderRotationExternalValidatorsAllowed(epoch) {
+			wasFound, next = consensus.decider.NthNextValidator(committee.Slots, leader, offset)
+		} else {
+			wasFound, next = consensus.decider.NthNextHmy(shard.Schedule.InstanceForEpoch(epoch), leader, offset)
+		}
+		if !wasFound {
+			utils.Logger().Error().Msg("Failed to get next leader")
+			// Seems like nothing we can do here.
+			return defaultKey
+		}
+		return next
+	}
+
 	slotsCount := len(committee.Slots)
 	blocksPerEpoch := shard.Schedule.InstanceForEpoch(epoch).BlocksPerEpoch()
 	if blocksPerEpoch == 0 {
@@ -769,12 +938,6 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int, defaultKey *bls.PublicK
 	// Passed all checks, we can change leader.
 	// NthNext will move the leader to the next leader in the committee.
 	// It does not know anything about external or internal validators.
-	var (
-		wasFound bool
-		next     *bls.PublicKeyWrapper
-		offset   = 1
-	)
-
 	for i := 0; i < len(committee.Slots); i++ {
 		if bc.Config().IsLeaderRotationV2Epoch(epoch) {
 			wasFound, next = consensus.decider.NthNextValidatorV2(committee.Slots, leader, offset)

consensus/leader.go

@@ -223,6 +223,7 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 	if !consensus.isRightBlockNumAndViewID(recvMsg) {
 		return
 	}
+	currentHeader := consensus.Blockchain().CurrentHeader()
 	// proceed only when the message is not received before
 	for _, signer := range recvMsg.SenderPubkeys {
 		signed := consensus.decider.ReadBallot(quorum.Commit, signer.Bytes)
@@ -328,7 +329,7 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 
 		if !blockObj.IsLastBlockInEpoch() {
 			// only do early commit if it's not epoch block to avoid problems
-			consensus.preCommitAndPropose(blockObj)
+			consensus.preCommitAndPropose1s(blockObj)
 		}
 		consensus.transitions.finalCommit = true
 		waitTime := 1000 * time.Millisecond

internal/chain/reward.go

@@ -220,9 +220,14 @@ func getDefaultStakingReward(bc engine.ChainReader, epoch *big.Int, blockNum uin
 		if bc.Config().IsTwoSeconds(epoch) {
 			defaultReward = stakingReward.TwoSecStakedBlocks
 		}
+		if bc.Config().IsOneSecond(epoch) {
+			defaultReward = stakingReward.OneSecStakedBlock
+		}
 	} else {
 		// Mainnet (other nets):
-		if bc.Config().IsHIP30(epoch) {
+		if bc.Config().IsOneSecond(epoch) {
+			defaultReward = stakingReward.OneSecStakedBlock
+		} else if bc.Config().IsHIP30(epoch) {
 			defaultReward = stakingReward.HIP30StakedBlocks
 		} else if bc.Config().IsTwoSeconds(epoch) {
 			defaultReward = stakingReward.TwoSecStakedBlocks

internal/configs/sharding/localnet.go

@@ -51,6 +51,8 @@ const (
 
 func (ls localnetSchedule) InstanceForEpoch(epoch *big.Int) Instance {
 	switch {
+	case params.LocalnetChainConfig.IsOneSecond(epoch):
+		return localnetV4
 	case params.LocalnetChainConfig.IsHIP30(epoch):
 		return localnetV4
 	case params.LocalnetChainConfig.IsFeeCollectEpoch(epoch):

internal/params/config.go

@@ -323,6 +323,8 @@ var (
 		MaxRateEpoch:                          EpochTBD,
 		DevnetExternalEpoch:                   EpochTBD,
 		TestnetExternalEpoch:                  EpochTBD,
+		IsOneSecondEpoch:                      big.NewInt(6),
+		IsRotationEachBlockEpoch:              big.NewInt(6),
 	}
 
 	// AllProtocolChanges ...
@@ -374,6 +376,8 @@ var (
 		big.NewInt(0),
 		big.NewInt(0),
 		big.NewInt(0),
+		big.NewInt(0),
+		big.NewInt(0),
 	}
 
 	// TestChainConfig ...
@@ -425,6 +429,8 @@ var (
 		big.NewInt(0),        // MaxRateEpoch
 		big.NewInt(0),
 		big.NewInt(0),
+		big.NewInt(0),
+		big.NewInt(0),
 	}
 
 	// TestRules ...
@@ -606,6 +612,10 @@ type ChainConfig struct {
 	// vote power feature  https://github.com/harmony-one/harmony/pull/4683
 	// if crosslink are not sent for an entire epoch signed and toSign will be 0 and 0. when that happen, next epoch there will no shard 1 validator elected in the committee.
 	HIP32Epoch *big.Int `json:"hip32-epoch,omitempty"`
+
+	IsOneSecondEpoch *big.Int `json:"is-one-second-epoch,omitempty"`
+
+	IsRotationEachBlockEpoch *big.Int `json:"is-rotation-each-block-epoch"`
 }
 
 // String implements the fmt.Stringer interface.
@@ -731,6 +741,10 @@ func (c *ChainConfig) IsTwoSeconds(epoch *big.Int) bool {
 	return isForked(c.TwoSecondsEpoch, epoch)
 }
 
+func (c *ChainConfig) IsOneSecond(epoch *big.Int) bool {
+	return isForked(c.IsOneSecondEpoch, epoch)
+}
+
 // IsSixtyPercent determines whether it is the epoch to reduce internal voting power to 60%
 func (c *ChainConfig) IsSixtyPercent(epoch *big.Int) bool {
 	return isForked(c.SixtyPercentEpoch, epoch)
@@ -895,6 +909,10 @@ func (c *ChainConfig) IsOneEpochBeforeHIP30(epoch *big.Int) bool {
 	return new(big.Int).Sub(c.HIP30Epoch, epoch).Cmp(common.Big1) == 0
 }
 
+func (c *ChainConfig) IsRotationEachBlock(epoch *big.Int) bool {
+	return isForked(c.IsRotationEachBlockEpoch, epoch)
+}
+
 // UpdateEthChainIDByShard update the ethChainID based on shard ID.
 func UpdateEthChainIDByShard(shardID uint32) {
 	once.Do(func() {

staking/reward/values.go

@@ -39,6 +39,11 @@ var (
 		big.NewInt(14*denominations.Nano), big.NewInt(denominations.Nano),
 	))
 
+	// OneSecStakedBlock is half of HIP30
+	OneSecStakedBlock = numeric.NewDecFromBigInt(new(big.Int).Mul(
+		big.NewInt(7*denominations.Nano), big.NewInt(denominations.Nano),
+	))
+
 	// TotalInitialTokens is the total amount of tokens (in ONE) at block 0 of the network.
 	// This should be set/change on the node's init according to the core.GenesisSpec.
 	TotalInitialTokens = numeric.Dec{Int: big.NewInt(0)}