woop/consensus/consensus.go

package consensus

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/harmony-one/abool"
	bls_core "github.com/harmony-one/bls/ffi/go/bls"
	"github.com/harmony-one/harmony/consensus/engine"
	"github.com/harmony-one/harmony/consensus/quorum"
	"github.com/harmony-one/harmony/core"
	"github.com/harmony-one/harmony/core/types"
	"github.com/harmony-one/harmony/crypto/bls"
	bls_cosi "github.com/harmony-one/harmony/crypto/bls"
	"github.com/harmony-one/harmony/internal/registry"
	"github.com/harmony-one/harmony/internal/utils"
	"github.com/harmony-one/harmony/multibls"
	"github.com/harmony-one/harmony/p2p"
	"github.com/harmony-one/harmony/shard"
	"github.com/harmony-one/harmony/shard/committee"
	"github.com/harmony-one/harmony/staking/slash"
	"github.com/pkg/errors"
)

const (
	vdFAndProofSize = 516 // size of VDF and Proof
	vdfAndSeedSize  = 548 // size of VDF/Proof and Seed
)

var errLeaderPriKeyNotFound = errors.New("leader private key not found locally")

// ProposalType is to indicate the type of signal for new block proposal
type ProposalType byte

// Constant of the type of new block proposal
const (
	SyncProposal ProposalType = iota
	AsyncProposal
)

type DownloadAsync interface {
	DownloadAsync()
}

// Consensus is the main struct with all states and data related to consensus process.
type Consensus struct {
	decider quorum.Decider
	// FBFTLog stores the pbft messages and blocks during FBFT process
	fBFTLog *FBFTLog
	// phase: different phase of FBFT protocol: pre-prepare, prepare, commit, finish etc
	phase FBFTPhase
	// current indicates what state a node is in
	current State
	// isBackup declarative the node is in backup mode
	isBackup bool
	// 2 types of timeouts: normal and viewchange
	consensusTimeout map[TimeoutType]*utils.Timeout
	// Commits collected from validators.
	aggregatedPrepareSig *bls_core.Sign
	aggregatedCommitSig  *bls_core.Sign
	prepareBitmap        *bls_cosi.Mask
	commitBitmap         *bls_cosi.Mask

	multiSigBitmap *bls_cosi.Mask // Bitmap for parsing multisig bitmap from validators

	// Registry for services.
	registry *registry.Registry
	// Minimal number of peers in the shard
	// If the number of validators is less than minPeers, the consensus won't start
	MinPeers int
	// private/public keys of current node
	priKey multibls.PrivateKeys
	// the publickey of leader
	LeaderPubKey *bls.PublicKeyWrapper
	// blockNum: the next blockNumber that FBFT is going to agree on,
	// should be equal to the blockNumber of next block
	blockNum uint64
	// Blockhash - 32 byte
	blockHash [32]byte
	// Block to run consensus on
	block []byte
	// Shard Id which this node belongs to
	ShardID uint32
	// IgnoreViewIDCheck determines whether to ignore viewID check
	IgnoreViewIDCheck *abool.AtomicBool
	// consensus mutex
	mutex *sync.RWMutex
	// ViewChange struct
	vc *viewChange
	// Signal channel for proposing a new block and start new consensus
	readySignal chan ProposalType
	// Channel to send full commit signatures to finish new block proposal
	commitSigChannel chan []byte
	// The post-consensus job func passed from Node object
	// Called when consensus on a new block is done
	PostConsensusJob func(*types.Block) error
	// verified block to state sync broadcast
	VerifiedNewBlock chan *types.Block
	// Channel for DRG protocol to send pRnd (preimage of randomness resulting from combined vrf
	// randomnesses) to consensus. The first 32 bytes are randomness, the rest is for bitmap.
	PRndChannel chan []byte
	// Channel for DRG protocol to send VDF. The first 516 bytes are the VDF/Proof and the last 32
	// bytes are the seed for deriving VDF
	RndChannel  chan [vdfAndSeedSize]byte
	pendingRnds [][vdfAndSeedSize]byte // A list of pending randomness
	// The p2p host used to send/receive p2p messages
	host p2p.Host
	// MessageSender takes are of sending consensus message and the corresponding retry logic.
	msgSender *MessageSender
	// If true, this consensus will not propose view change.
	disableViewChange bool
	// Have a dedicated reader thread pull from this chan, like in node
	SlashChan chan slash.Record
	// How long in second the leader needs to wait to propose a new block.
	BlockPeriod time.Duration
	// The time due for next block proposal
	NextBlockDue time.Time
	// Temporary flag to control whether aggregate signature signing is enabled
	AggregateSig bool

	// TODO (leo): an new metrics system to keep track of the consensus/viewchange
	// finality of previous consensus in the unit of milliseconds
	finality int64
	// finalityCounter keep tracks of the finality time
	finalityCounter atomic.Value //int64

	dHelper DownloadAsync

	// Both flags only for initialization state.
	start           bool
	isInitialLeader bool
}

// Blockchain returns the blockchain.
func (consensus *Consensus) Blockchain() core.BlockChain {
	return consensus.registry.GetBlockchain()
}

func (consensus *Consensus) FBFTLog() FBFT {
	return threadsafeFBFTLog{
		log: consensus.fBFTLog,
		mu:  consensus.mutex,
	}
}

// ChainReader returns the chain reader.
// This is mostly the same as Blockchain, but it returns only read methods, so we assume it's safe for concurrent use.
func (consensus *Consensus) ChainReader() engine.ChainReader {
	return consensus.Blockchain()
}

func (consensus *Consensus) ReadySignal(p ProposalType) {
	consensus.readySignal <- p
}

func (consensus *Consensus) GetReadySignal() chan ProposalType {
	return consensus.readySignal
}

func (consensus *Consensus) GetCommitSigChannel() chan []byte {
	return consensus.commitSigChannel
}

// Beaconchain returns the beaconchain.
func (consensus *Consensus) Beaconchain() core.BlockChain {
	return consensus.registry.GetBeaconchain()
}

// verifyBlock is a function used to verify the block and keep trace of verified blocks.
func (consensus *Consensus) verifyBlock(block *types.Block) error {
	if !consensus.fBFTLog.IsBlockVerified(block.Hash()) {
		if err := consensus.BlockVerifier(block); err != nil {
			return errors.Errorf("Block verification failed: %s", err)
		}
		consensus.fBFTLog.MarkBlockVerified(block)
	}
	return nil
}

// BlocksSynchronized lets the main loop know that block synchronization finished
// thus the blockchain is likely to be up to date.
func (consensus *Consensus) BlocksSynchronized() {
	err := consensus.AddConsensusLastMile()
	if err != nil {
		consensus.GetLogger().Error().Err(err).Msg("add last mile failed")
	}
	consensus.mutex.Lock()
	defer consensus.mutex.Unlock()
	consensus.syncReadyChan()
}

// BlocksNotSynchronized lets the main loop know that block is not synchronized
func (consensus *Consensus) BlocksNotSynchronized(reason string) {
	consensus.mutex.Lock()
	defer consensus.mutex.Unlock()
	consensus.syncNotReadyChan(reason)
}

// VdfSeedSize returns the number of VRFs for VDF computation
func (consensus *Consensus) VdfSeedSize() int {
	consensus.mutex.RLock()
	defer consensus.mutex.RUnlock()
	return int(consensus.decider.ParticipantsCount()) * 2 / 3
}

// GetPublicKeys returns the public keys
func (consensus *Consensus) GetPublicKeys() multibls.PublicKeys {
	return consensus.getPublicKeys()
}

func (consensus *Consensus) getPublicKeys() multibls.PublicKeys {
	return consensus.priKey.GetPublicKeys()
}

func (consensus *Consensus) GetLeaderPubKey() *bls_cosi.PublicKeyWrapper {
	consensus.mutex.RLock()
	defer consensus.mutex.RUnlock()
	return consensus.getLeaderPubKey()
}

func (consensus *Consensus) getLeaderPubKey() *bls_cosi.PublicKeyWrapper {
	return consensus.LeaderPubKey
}

func (consensus *Consensus) SetLeaderPubKey(pub *bls_cosi.PublicKeyWrapper) {
	consensus.mutex.Lock()
	defer consensus.mutex.Unlock()
	consensus.setLeaderPubKey(pub)
}

func (consensus *Consensus) setLeaderPubKey(pub *bls_cosi.PublicKeyWrapper) {
	consensus.LeaderPubKey = pub
}

func (consensus *Consensus) GetPrivateKeys() multibls.PrivateKeys {
	return consensus.priKey
}

// GetLeaderPrivateKey returns leader private key if node is the leader
func (consensus *Consensus) getLeaderPrivateKey(leaderKey *bls_core.PublicKey) (*bls.PrivateKeyWrapper, error) {
	for i, key := range consensus.priKey {
		if key.Pub.Object.IsEqual(leaderKey) {
			return &consensus.priKey[i], nil
		}
	}
	return nil, errors.Wrapf(errLeaderPriKeyNotFound, leaderKey.SerializeToHexStr())
}

// getConsensusLeaderPrivateKey returns consensus leader private key if node is the leader
func (consensus *Consensus) getConsensusLeaderPrivateKey() (*bls.PrivateKeyWrapper, error) {
	return consensus.getLeaderPrivateKey(consensus.LeaderPubKey.Object)
}

func (consensus *Consensus) IsBackup() bool {
	return consensus.isBackup
}

func (consensus *Consensus) BlockNum() uint64 {
	return atomic.LoadUint64(&consensus.blockNum)
}

func (consensus *Consensus) getBlockNum() uint64 {
	return atomic.LoadUint64(&consensus.blockNum)
}

// New create a new Consensus record
func New(
	host p2p.Host, shard uint32, multiBLSPriKey multibls.PrivateKeys,
	registry *registry.Registry,
	Decider quorum.Decider, minPeers int, aggregateSig bool,
) (*Consensus, error) {
	consensus := Consensus{
		mutex:        &sync.RWMutex{},
		ShardID:      shard,
		fBFTLog:      NewFBFTLog(),
		phase:        FBFTAnnounce,
		current:      State{mode: Normal},
		decider:      Decider,
		registry:     registry,
		MinPeers:     minPeers,
		AggregateSig: aggregateSig,
		host:         host,
		msgSender:    NewMessageSender(host),
		// FBFT timeout
		consensusTimeout: createTimeout(),
		dHelper:          downloadAsync{},
	}

	if multiBLSPriKey != nil {
		consensus.priKey = multiBLSPriKey
		utils.Logger().Info().
			Str("publicKey", consensus.GetPublicKeys().SerializeToHexStr()).Msg("My Public Key")
	} else {
		utils.Logger().Error().Msg("the bls key is nil")
		return nil, fmt.Errorf("nil bls key, aborting")
	}

	// viewID has to be initialized as the height of
	// the blockchain during initialization as it was
	// displayed on explorer as Height right now
	consensus.current.setCurBlockViewID(0)
	consensus.SlashChan = make(chan slash.Record)
	consensus.readySignal = make(chan ProposalType)
	consensus.commitSigChannel = make(chan []byte)
	// channel for receiving newly generated VDF
	consensus.RndChannel = make(chan [vdfAndSeedSize]byte)
	consensus.IgnoreViewIDCheck = abool.NewBool(false)
	// Make Sure Verifier is not null
	consensus.vc = newViewChange()
	// init prometheus metrics
	initMetrics()
	consensus.AddPubkeyMetrics()

	err := consensus.InitConsensusWithValidators()
	if err != nil {
		return nil, errors.WithMessage(err, "failed to init consensus with validators")
	}

	return &consensus, nil
}

func (consensus *Consensus) GetHost() p2p.Host {
	return consensus.host
}

func (consensus *Consensus) Registry() *registry.Registry {
	return consensus.registry
}

func (consensus *Consensus) Decider() quorum.Decider {
	return quorum.NewThreadSafeDecider(consensus.decider, consensus.mutex)
}

// InitConsensusWithValidators initialize shard state
// from latest epoch and update committee pub
// keys for consensus
func (consensus *Consensus) InitConsensusWithValidators() (err error) {
	shardID := consensus.ShardID
	currentBlock := consensus.Blockchain().CurrentBlock()
	blockNum := currentBlock.NumberU64()
	consensus.SetMode(Listening)
	epoch := currentBlock.Epoch()
	utils.Logger().Info().
		Uint64("blockNum", blockNum).
		Uint32("shardID", shardID).
		Uint64("epoch", epoch.Uint64()).
		Msg("[InitConsensusWithValidators] Try To Get PublicKeys")
	shardState, err := committee.WithStakingEnabled.Compute(
		epoch, consensus.Blockchain(),
	)
	if err != nil {
		utils.Logger().Err(err).
			Uint64("blockNum", blockNum).
			Uint32("shardID", shardID).
			Uint64("epoch", epoch.Uint64()).
			Msg("[InitConsensusWithValidators] Failed getting shard state")
		return err
	}
	subComm, err := shardState.FindCommitteeByID(shardID)
	if err != nil {
		utils.Logger().Err(err).
			Interface("shardState", shardState).
			Msg("[InitConsensusWithValidators] Find CommitteeByID")
		return err
	}
	pubKeys, err := subComm.BLSPublicKeys()
	if err != nil {
		utils.Logger().Error().
			Uint32("shardID", shardID).
			Uint64("blockNum", blockNum).
			Msg("[InitConsensusWithValidators] PublicKeys is Empty, Cannot update public keys")
		return errors.Wrapf(
			err,
			"[InitConsensusWithValidators] PublicKeys is Empty, Cannot update public keys",
		)
	}

	for _, key := range pubKeys {
		if consensus.GetPublicKeys().Contains(key.Object) {
			utils.Logger().Info().
				Uint64("blockNum", blockNum).
				Int("numPubKeys", len(pubKeys)).
				Str("mode", consensus.Mode().String()).
				Msg("[InitConsensusWithValidators] Successfully updated public keys")
			consensus.UpdatePublicKeys(pubKeys, shard.Schedule.InstanceForEpoch(epoch).ExternalAllowlist())
			consensus.SetMode(Normal)
			return nil
		}
	}
	return nil
}

type downloadAsync struct {
}

func (a downloadAsync) DownloadAsync() {
}