path: root/core/agreement-mgr.go

// Copyright 2018 The dexon-consensus Authors
// This file is part of the dexon-consensus library.
//
// The dexon-consensus library is free software: you can redistribute it
// and/or modify it under the terms of the GNU Lesser General Public License as
// published by the Free Software Foundation, either version 3 of the License,
// or (at your option) any later version.
//
// The dexon-consensus library is distributed in the hope that it will be
// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
// General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the dexon-consensus library. If not, see
// <http://www.gnu.org/licenses/>.

package core

import (
    "context"
    "errors"
    "math"
    "sync"
    "time"

    lru "github.com/hashicorp/golang-lru"

    "github.com/dexon-foundation/dexon-consensus/common"
    "github.com/dexon-foundation/dexon-consensus/core/types"
    typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
    "github.com/dexon-foundation/dexon-consensus/core/utils"
)

// Errors returned from BA modules
var (
    ErrPreviousRoundIsNotFinished = errors.New("previous round is not finished")
    ErrRoundOutOfRange            = errors.New("round out of range")
    ErrInvalidBlock               = errors.New("invalid block")
    ErrNoValidLeader              = errors.New("no valid leader")
    ErrIncorrectCRSSignature      = errors.New("incorrect CRS signature")
    ErrBlockTooOld                = errors.New("block too old")
)

const maxResultCache = 100
const settingLimit = 3

// genValidLeader generate a validLeader function for agreement modules.
func genValidLeader(
    mgr *agreementMgr) validLeaderFn {
    return func(block *types.Block, crs common.Hash) (bool, error) {
        if block.Timestamp.After(time.Now()) {
            return false, nil
        }
        if block.Position.Round >= DKGDelayRound {
            if mgr.recv.npks == nil {
                return false, nil
            }
            if block.Position.Round > mgr.recv.npks.Round {
                return false, nil
            }
            if block.Position.Round < mgr.recv.npks.Round {
                return false, ErrBlockTooOld
            }
        }
        if !utils.VerifyCRSSignature(block, crs, mgr.recv.npks) {
            return false, ErrIncorrectCRSSignature
        }
        if err := mgr.bcModule.sanityCheck(block); err != nil {
            if err == ErrRetrySanityCheckLater {
                return false, nil
            }
            return false, err
        }
        mgr.logger.Debug("Calling Application.VerifyBlock", "block", block)
        switch mgr.app.VerifyBlock(block) {
        case types.VerifyInvalidBlock:
            return false, ErrInvalidBlock
        case types.VerifyRetryLater:
            return false, nil
        default:
        }
        return true, nil
    }
}

type agreementMgrConfig struct {
    utils.RoundBasedConfig

    notarySetSize uint32
    lambdaBA      time.Duration
    crs           common.Hash
}

func (c *agreementMgrConfig) from(
    round uint64, config *types.Config, crs common.Hash) {
    c.notarySetSize = config.NotarySetSize
    c.lambdaBA = config.LambdaBA
    c.crs = crs
    c.SetupRoundBasedFields(round, config)
}

func newAgreementMgrConfig(prev agreementMgrConfig, config *types.Config,
    crs common.Hash) (c agreementMgrConfig) {
    c = agreementMgrConfig{}
    c.from(prev.RoundID()+1, config, crs)
    c.AppendTo(prev.RoundBasedConfig)
    return
}

type baRoundSetting struct {
    round     uint64
    dkgSet    map[types.NodeID]struct{}
    threshold int
    ticker    Ticker
    crs       common.Hash
}

type agreementMgr struct {
    // TODO(mission): unbound Consensus instance from this module.
    con               *Consensus
    ID                types.NodeID
    app               Application
    gov               Governance
    network           Network
    logger            common.Logger
    cache             *utils.NodeSetCache
    signer            *utils.Signer
    bcModule          *blockChain
    ctx               context.Context
    configs           []agreementMgrConfig
    baModule          *agreement
    recv              *consensusBAReceiver
    processedBAResult map[types.Position]struct{}
    voteFilter        *utils.VoteFilter
    settingCache      *lru.Cache
    curRoundSetting   *baRoundSetting
    waitGroup         sync.WaitGroup
    isRunning         bool
    lock              sync.RWMutex
}

func newAgreementMgr(con *Consensus) (mgr *agreementMgr, err error) {
    settingCache, _ := lru.New(settingLimit)
    mgr = &agreementMgr{
        con:               con,
        ID:                con.ID,
        app:               con.app,
        gov:               con.gov,
        network:           con.network,
        logger:            con.logger,
        cache:             con.nodeSetCache,
        signer:            con.signer,
        bcModule:          con.bcModule,
        ctx:               con.ctx,
        processedBAResult: make(map[types.Position]struct{}, maxResultCache),
        voteFilter:        utils.NewVoteFilter(),
        settingCache:      settingCache,
    }
    mgr.recv = &consensusBAReceiver{
        consensus:     con,
        restartNotary: make(chan types.Position, 1),
    }
    return mgr, nil
}

func (mgr *agreementMgr) prepare() {
    round := mgr.bcModule.tipRound()
    agr := newAgreement(
        mgr.ID,
        mgr.recv,
        newLeaderSelector(genValidLeader(mgr), mgr.logger),
        mgr.signer,
        mgr.logger)
    setting := mgr.generateSetting(round)
    if setting == nil {
        mgr.logger.Warn("Unable to prepare init setting", "round", round)
        return
    }
    mgr.curRoundSetting = setting
    agr.notarySet = mgr.curRoundSetting.dkgSet
    // Hacky way to make agreement module self contained.
    mgr.recv.agreementModule = agr
    mgr.baModule = agr
    if round >= DKGDelayRound {
        if _, exist := setting.dkgSet[mgr.ID]; exist {
            mgr.logger.Debug("Preparing signer and npks.", "round", round)
            npk, signer, err := mgr.con.cfgModule.getDKGInfo(round, false)
            if err != nil {
                mgr.logger.Error("Failed to prepare signer and npks.",
                    "round", round,
                    "error", err)
            }
            mgr.logger.Debug("Prepared signer and npks.",
                "round", round, "signer", signer != nil, "npks", npk != nil)
        }
    }
    return
}

func (mgr *agreementMgr) run() {
    mgr.lock.Lock()
    defer mgr.lock.Unlock()
    if mgr.isRunning {
        return
    }
    mgr.isRunning = true
    mgr.waitGroup.Add(1)
    go func() {
        defer mgr.waitGroup.Done()
        mgr.runBA(mgr.bcModule.tipRound())
    }()
}

func (mgr *agreementMgr) calcLeader(
    dkgSet map[types.NodeID]struct{},
    crs common.Hash, pos types.Position) (
    types.NodeID, error) {
    nodeSet := types.NewNodeSetFromMap(dkgSet)
    leader := nodeSet.GetSubSet(1, types.NewNodeLeaderTarget(
        crs, pos.Height))
    for nID := range leader {
        return nID, nil
    }
    return types.NodeID{}, ErrNoValidLeader
}

func (mgr *agreementMgr) config(round uint64) *agreementMgrConfig {
    mgr.lock.RLock()
    defer mgr.lock.RUnlock()
    if round < mgr.configs[0].RoundID() {
        panic(ErrRoundOutOfRange)
    }
    roundIndex := round - mgr.configs[0].RoundID()
    if roundIndex >= uint64(len(mgr.configs)) {
        return nil
    }
    return &mgr.configs[roundIndex]
}

func (mgr *agreementMgr) notifyRoundEvents(evts []utils.RoundEventParam) error {
    mgr.lock.Lock()
    defer mgr.lock.Unlock()
    apply := func(e utils.RoundEventParam) error {
        if len(mgr.configs) > 0 {
            lastCfg := mgr.configs[len(mgr.configs)-1]
            if e.BeginHeight != lastCfg.RoundEndHeight() {
                return ErrInvalidBlockHeight
            }
            if lastCfg.RoundID() == e.Round {
                mgr.configs[len(mgr.configs)-1].ExtendLength()
            } else if lastCfg.RoundID()+1 == e.Round {
                mgr.configs = append(mgr.configs, newAgreementMgrConfig(
                    lastCfg, e.Config, e.CRS))
            } else {
                return ErrInvalidRoundID
            }
        } else {
            c := agreementMgrConfig{}
            c.from(e.Round, e.Config, e.CRS)
            c.SetRoundBeginHeight(e.BeginHeight)
            mgr.configs = append(mgr.configs, c)
        }
        return nil
    }
    for _, e := range evts {
        if err := apply(e); err != nil {
            return err
        }
    }
    return nil
}

func (mgr *agreementMgr) checkProposer(
    round uint64, proposerID types.NodeID) error {
    if round == mgr.curRoundSetting.round {
        if _, exist := mgr.curRoundSetting.dkgSet[proposerID]; !exist {
            return ErrNotInNotarySet
        }
    } else if round == mgr.curRoundSetting.round+1 {
        setting := mgr.generateSetting(round)
        if setting == nil {
            return ErrConfigurationNotReady
        }
        if _, exist := setting.dkgSet[proposerID]; !exist {
            return ErrNotInNotarySet
        }
    }
    return nil
}

func (mgr *agreementMgr) processVote(v *types.Vote) (err error) {
    if !mgr.recv.isNotary {
        return nil
    }
    if mgr.voteFilter.Filter(v) {
        return nil
    }
    if err := mgr.checkProposer(v.Position.Round, v.ProposerID); err != nil {
        return err
    }
    if err = mgr.baModule.processVote(v); err == nil {
        mgr.baModule.updateFilter(mgr.voteFilter)
        mgr.voteFilter.AddVote(v)
    }
    if err == ErrSkipButNoError {
        err = nil
    }
    return
}

func (mgr *agreementMgr) processBlock(b *types.Block) error {
    if err := mgr.checkProposer(b.Position.Round, b.ProposerID); err != nil {
        return err
    }
    return mgr.baModule.processBlock(b)
}

func (mgr *agreementMgr) touchAgreementResult(
    result *types.AgreementResult) (first bool) {
    // DO NOT LOCK THIS FUNCTION!!!!!!!! YOU WILL REGRET IT!!!!!
    if _, exist := mgr.processedBAResult[result.Position]; !exist {
        first = true
        if len(mgr.processedBAResult) > maxResultCache {
            for k := range mgr.processedBAResult {
                // Randomly drop one element.
                delete(mgr.processedBAResult, k)
                break
            }
        }
        mgr.processedBAResult[result.Position] = struct{}{}
    }
    return
}

func (mgr *agreementMgr) untouchAgreementResult(
    result *types.AgreementResult) {
    // DO NOT LOCK THIS FUNCTION!!!!!!!! YOU WILL REGRET IT!!!!!
    delete(mgr.processedBAResult, result.Position)
}

func (mgr *agreementMgr) processAgreementResult(
    result *types.AgreementResult) error {
    aID := mgr.baModule.agreementID()
    if isStop(aID) {
        return nil
    }
    if result.Position == aID && !mgr.baModule.confirmed() {
        mgr.logger.Info("Syncing BA", "position", result.Position)
        if result.Position.Round >= DKGDelayRound {
            return mgr.baModule.processAgreementResult(result)
        }
        for key := range result.Votes {
            if err := mgr.baModule.processVote(&result.Votes[key]); err != nil {
                return err
            }
        }
    } else if result.Position.Newer(aID) {
        mgr.logger.Info("Fast syncing BA", "position", result.Position)
        if result.Position.Round < DKGDelayRound {
            mgr.logger.Debug("Calling Network.PullBlocks for fast syncing BA",
                "hash", result.BlockHash)
            mgr.network.PullBlocks(common.Hashes{result.BlockHash})
            for key := range result.Votes {
                if err := mgr.baModule.processVote(&result.Votes[key]); err != nil {
                    return err
                }
            }
        }
        setting := mgr.generateSetting(result.Position.Round)
        if setting == nil {
            mgr.logger.Warn("unable to get setting", "round",
                result.Position.Round)
            return ErrConfigurationNotReady
        }
        mgr.curRoundSetting = setting
        leader, err := mgr.calcLeader(setting.dkgSet, setting.crs, result.Position)
        if err != nil {
            return err
        }
        mgr.baModule.restart(
            setting.dkgSet, setting.threshold,
            result.Position, leader, setting.crs)
        if result.Position.Round >= DKGDelayRound {
            return mgr.baModule.processAgreementResult(result)
        }
    }
    return nil
}

func (mgr *agreementMgr) processFinalizedBlock(block *types.Block) error {
    aID := mgr.baModule.agreementID()
    if block.Position.Older(aID) {
        return nil
    }
    mgr.baModule.processFinalizedBlock(block)
    return nil
}

func (mgr *agreementMgr) stop() {
    // Stop all running agreement modules.
    func() {
        mgr.lock.Lock()
        defer mgr.lock.Unlock()
        mgr.baModule.stop()
    }()
    // Block until all routines are done.
    mgr.waitGroup.Wait()
}

func (mgr *agreementMgr) generateSetting(round uint64) *baRoundSetting {
    if setting, exist := mgr.settingCache.Get(round); exist {
        return setting.(*baRoundSetting)
    }
    curConfig := mgr.config(round)
    if curConfig == nil {
        return nil
    }
    var dkgSet map[types.NodeID]struct{}
    if round >= DKGDelayRound {
        _, qualidifed, err := typesDKG.CalcQualifyNodes(
            mgr.gov.DKGMasterPublicKeys(round),
            mgr.gov.DKGComplaints(round),
            utils.GetDKGThreshold(mgr.gov.Configuration(round)),
        )
        if err != nil {
            mgr.logger.Error("Failed to get gpk", "round", round, "error", err)
            return nil
        }
        dkgSet = qualidifed
    }
    if len(dkgSet) == 0 {
        var err error
        dkgSet, err = mgr.cache.GetNotarySet(round)
        if err != nil {
            mgr.logger.Error("Failed to get notarySet", "round", round, "error", err)
            return nil
        }
    }
    setting := &baRoundSetting{
        crs:    curConfig.crs,
        dkgSet: dkgSet,
        round:  round,
        threshold: utils.GetBAThreshold(&types.Config{
            NotarySetSize: curConfig.notarySetSize}),
    }
    mgr.settingCache.Add(round, setting)
    return setting
}

func (mgr *agreementMgr) runBA(initRound uint64) {
    // These are round based variables.
    var (
        currentRound uint64
        nextRound    = initRound
        curConfig    = mgr.config(initRound)
        setting      = &baRoundSetting{}
        tickDuration time.Duration
        ticker       Ticker
    )

    // Check if this routine needs to awake in this round and prepare essential
    // variables when yes.
    checkRound := func() (isDKG bool) {
        defer func() {
            currentRound = nextRound
            nextRound++
        }()
        // Wait until the configuartion for next round is ready.
        for {
            if setting = mgr.generateSetting(nextRound); setting != nil {
                break
            } else {
                mgr.logger.Debug("Round is not ready", "round", nextRound)
                time.Sleep(1 * time.Second)
            }
        }
        _, isDKG = setting.dkgSet[mgr.ID]
        if isDKG {
            mgr.logger.Info("Selected as dkg set",
                "ID", mgr.ID,
                "round", nextRound)
        } else {
            mgr.logger.Info("Not selected as dkg set",
                "ID", mgr.ID,
                "round", nextRound)
        }
        // Setup ticker
        if tickDuration != curConfig.lambdaBA {
            if ticker != nil {
                ticker.Stop()
            }
            ticker = newTicker(mgr.gov, nextRound, TickerBA)
            tickDuration = curConfig.lambdaBA
        }
        setting.ticker = ticker
        return
    }
Loop:
    for {
        select {
        case <-mgr.ctx.Done():
            break Loop
        default:
        }
        mgr.recv.isNotary = checkRound()
        mgr.voteFilter = utils.NewVoteFilter()
        mgr.voteFilter.Position.Round = currentRound
        mgr.recv.emptyBlockHashMap = &sync.Map{}
        if currentRound >= DKGDelayRound && mgr.recv.isNotary {
            var err error
            mgr.recv.npks, mgr.recv.psigSigner, err =
                mgr.con.cfgModule.getDKGInfo(currentRound, false)
            if err != nil {
                mgr.logger.Warn("cannot get dkg info",
                    "round", currentRound, "error", err)
            }
        } else {
            mgr.recv.npks = nil
            mgr.recv.psigSigner = nil
        }
        // Run BA for this round.
        mgr.recv.restartNotary <- types.Position{
            Round:  currentRound,
            Height: math.MaxUint64,
        }
        if err := mgr.baRoutineForOneRound(setting); err != nil {
            mgr.logger.Error("BA routine failed",
                "error", err,
                "nodeID", mgr.ID)
            break Loop
        }
    }
}

func (mgr *agreementMgr) baRoutineForOneRound(
    setting *baRoundSetting) (err error) {
    agr := mgr.baModule
    recv := mgr.recv
    oldPos := agr.agreementID()
    restart := func(restartPos types.Position) (breakLoop bool, err error) {
        if !isStop(restartPos) {
            if restartPos.Height+1 >= mgr.config(setting.round).RoundEndHeight() {
                for {
                    select {
                    case <-mgr.ctx.Done():
                        break
                    default:
                    }
                    tipRound := mgr.bcModule.tipRound()
                    if tipRound > setting.round {
                        break
                    } else {
                        mgr.logger.Debug("Waiting blockChain to change round...",
                            "curRound", setting.round,
                            "tipRound", tipRound)
                    }
                    time.Sleep(100 * time.Millisecond)
                }
                // This round is finished.
                breakLoop = true
                return
            }
            if restartPos.Older(oldPos) {
                // The restartNotary event is triggered by 'BlockConfirmed'
                // of some older block.
                return
            }
        }
        var nextHeight uint64
        var nextTime time.Time
        for {
            // Make sure we are stoppable.
            select {
            case <-mgr.ctx.Done():
                breakLoop = true
                return
            default:
            }
            nextHeight, nextTime = mgr.bcModule.nextBlock()
            if nextHeight != notReadyHeight {
                if isStop(restartPos) {
                    break
                }
                if nextHeight > restartPos.Height {
                    break
                }
            }
            mgr.logger.Debug("BlockChain not ready!!!",
                "old", oldPos, "restart", restartPos, "next", nextHeight)
            time.Sleep(100 * time.Millisecond)
        }
        nextPos := types.Position{
            Round:  setting.round,
            Height: nextHeight,
        }
        oldPos = nextPos
        var leader types.NodeID
        leader, err = mgr.calcLeader(setting.dkgSet, setting.crs, nextPos)
        if err != nil {
            return
        }
        time.Sleep(nextTime.Sub(time.Now()))
        setting.ticker.Restart()
        agr.restart(setting.dkgSet, setting.threshold, nextPos, leader, setting.crs)
        return
    }
Loop:
    for {
        select {
        case <-mgr.ctx.Done():
            break Loop
        default:
        }
        if agr.confirmed() {
            // Block until receive restartPos
            select {
            case restartPos := <-recv.restartNotary:
                breakLoop, err := restart(restartPos)
                if err != nil {
                    return err
                }
                if breakLoop {
                    break Loop
                }
            case <-mgr.ctx.Done():
                break Loop
            }
        }
        select {
        case restartPos := <-recv.restartNotary:
            breakLoop, err := restart(restartPos)
            if err != nil {
                return err
            }
            if breakLoop {
                break Loop
            }
        default:
        }
        if !mgr.recv.isNotary {
            select {
            case <-setting.ticker.Tick():
                continue Loop
            case <-mgr.ctx.Done():
                break Loop
            }
        }
        if err = agr.nextState(); err != nil {
            mgr.logger.Error("Failed to proceed to next state",
                "nodeID", mgr.ID.String(),
                "error", err)
            break Loop
        }
        if agr.pullVotes() {
            pos := agr.agreementID()
            mgr.logger.Debug("Calling Network.PullVotes for syncing votes",
                "position", pos)
            mgr.network.PullVotes(pos)
        }
        for i := 0; i < agr.clocks(); i++ {
            // Priority select for agreement.done().
            select {
            case <-agr.done():
                continue Loop
            default:
            }
            select {
            case <-agr.done():
                continue Loop
            case <-setting.ticker.Tick():
            }
        }
    }
    return nil
}