path: root/core/syncer/consensus.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 syncer

import (
    "context"
    "fmt"
    "sort"
    "sync"
    "time"

    "github.com/dexon-foundation/dexon-consensus/common"
    "github.com/dexon-foundation/dexon-consensus/core"
    "github.com/dexon-foundation/dexon-consensus/core/crypto"
    "github.com/dexon-foundation/dexon-consensus/core/db"
    "github.com/dexon-foundation/dexon-consensus/core/types"
    "github.com/dexon-foundation/dexon-consensus/core/utils"
)

var (
    // ErrAlreadySynced is reported when syncer is synced.
    ErrAlreadySynced = fmt.Errorf("already synced")
    // ErrNotSynced is reported when syncer is not synced yet.
    ErrNotSynced = fmt.Errorf("not synced yet")
    // ErrGenesisBlockReached is reported when genesis block reached.
    ErrGenesisBlockReached = fmt.Errorf("genesis block reached")
    // ErrInvalidBlockOrder is reported when SyncBlocks receives unordered
    // blocks.
    ErrInvalidBlockOrder = fmt.Errorf("invalid block order")
    // ErrMismatchBlockHashSequence means the delivering sequence is not
    // correct, compared to finalized blocks.
    ErrMismatchBlockHashSequence = fmt.Errorf("mismatch block hash sequence")
    // ErrInvalidSyncingFinalizationHeight raised when the blocks to sync is
    // not following the compaction chain tip in database.
    ErrInvalidSyncingFinalizationHeight = fmt.Errorf(
        "invalid syncing finalization height")
)

// Consensus is for syncing consensus module.
type Consensus struct {
    db           db.Database
    gov          core.Governance
    dMoment      time.Time
    logger       common.Logger
    app          core.Application
    prv          crypto.PrivateKey
    network      core.Network
    nodeSetCache *utils.NodeSetCache
    tsigVerifier *core.TSigVerifierCache

    lattice              *core.Lattice
    validatedChains      map[uint32]struct{}
    finalizedBlockHashes common.Hashes
    latticeLastRound     uint64
    randomnessResults    map[common.Hash]*types.BlockRandomnessResult
    blocks               []types.ByPosition
    agreements           []*agreement
    configs              []*types.Config
    roundBeginTimes      []time.Time
    agreementRoundCut    uint64

    // lock for accessing all fields.
    lock               sync.RWMutex
    moduleWaitGroup    sync.WaitGroup
    agreementWaitGroup sync.WaitGroup
    pullChan           chan common.Hash
    receiveChan        chan *types.Block
    ctx                context.Context
    ctxCancel          context.CancelFunc
    syncedLastBlock    *types.Block
    syncedConsensus    *core.Consensus
    dummyCancel        context.CancelFunc
    dummyFinished      <-chan struct{}
    dummyMsgBuffer     []interface{}
}

// NewConsensus creates an instance for Consensus (syncer consensus).
func NewConsensus(
    dMoment time.Time,
    app core.Application,
    gov core.Governance,
    db db.Database,
    network core.Network,
    prv crypto.PrivateKey,
    logger common.Logger) *Consensus {

    con := &Consensus{
        dMoment:         dMoment,
        app:             app,
        gov:             gov,
        db:              db,
        network:         network,
        nodeSetCache:    utils.NewNodeSetCache(gov),
        tsigVerifier:    core.NewTSigVerifierCache(gov, 7),
        prv:             prv,
        logger:          logger,
        validatedChains: make(map[uint32]struct{}),
        configs: []*types.Config{
            utils.GetConfigWithPanic(gov, 0, logger),
        },
        roundBeginTimes:   []time.Time{dMoment},
        receiveChan:       make(chan *types.Block, 1000),
        pullChan:          make(chan common.Hash, 1000),
        randomnessResults: make(map[common.Hash]*types.BlockRandomnessResult),
    }
    con.ctx, con.ctxCancel = context.WithCancel(context.Background())
    return con
}

func (con *Consensus) initConsensusObj(initBlock *types.Block) {
    func() {
        con.lock.Lock()
        defer con.lock.Unlock()
        con.latticeLastRound = initBlock.Position.Round
        debugApp, _ := con.app.(core.Debug)
        con.lattice = core.NewLattice(
            con.roundBeginTimes[con.latticeLastRound],
            con.latticeLastRound,
            con.configs[con.latticeLastRound],
            utils.NewSigner(con.prv),
            con.app,
            debugApp,
            con.db,
            con.logger,
        )
    }()
    con.startAgreement()
    con.startNetwork()
    con.startCRSMonitor()
}

func (con *Consensus) checkIfValidated() (validated bool) {
    con.lock.RLock()
    defer con.lock.RUnlock()
    var (
        round               = con.blocks[0][0].Position.Round
        numChains           = con.configs[round].NumChains
        validatedChainCount uint32
    )
    // Make sure we validate some block in all chains.
    for chainID := range con.validatedChains {
        if chainID < numChains {
            validatedChainCount++
        }
    }
    validated = validatedChainCount == numChains
    con.logger.Debug("syncer chain-validation status",
        "validated-chain", validatedChainCount,
        "round", round,
        "valid", validated)
    return
}

func (con *Consensus) checkIfSynced(blocks []*types.Block) (synced bool) {
    con.lock.RLock()
    defer con.lock.RUnlock()
    var (
        round          = con.blocks[0][0].Position.Round
        numChains      = con.configs[round].NumChains
        compactionTips = make([]*types.Block, numChains)
        overlapCount   = uint32(0)
    )
    defer func() {
        con.logger.Debug("syncer synced status",
            "overlap-count", overlapCount,
            "num-chain", numChains,
            "last-block", blocks[len(blocks)-1],
            "synced", synced)
    }()
    // Find tips (newset blocks) of each chain in compaction chain.
    b := blocks[len(blocks)-1]
    for tipCount := uint32(0); tipCount < numChains; {
        if compactionTips[b.Position.ChainID] == nil {
            // Check chainID for config change.
            if b.Position.ChainID < numChains {
                compactionTips[b.Position.ChainID] = b
                tipCount++
            }
        }
        if (b.Finalization.ParentHash == common.Hash{}) {
            return
        }
        b1, err := con.db.GetBlock(b.Finalization.ParentHash)
        if err != nil {
            panic(err)
        }
        b = &b1
    }
    // Check if chain tips of compaction chain and current cached confirmed
    // blocks are overlapped on each chain, numChains is decided by the round
    // of last block we seen on compaction chain.
    for chainID, b := range compactionTips {
        if len(con.blocks[chainID]) > 0 {
            if !b.Position.Older(&con.blocks[chainID][0].Position) {
                overlapCount++
            }
        }
    }
    synced = overlapCount == numChains
    return
}

// ensureAgreementOverlapRound ensures the oldest blocks in each chain in
// con.blocks are all in the same round, for avoiding config change while
// syncing.
func (con *Consensus) ensureAgreementOverlapRound() bool {
    con.lock.Lock()
    defer con.lock.Unlock()
    defer func() {
        con.logger.Debug("ensureAgreementOverlapRound returned",
            "round", con.agreementRoundCut)
    }()
    if con.agreementRoundCut > 0 {
        return true
    }
    // Clean empty blocks on tips of chains.
    for idx, bs := range con.blocks {
        for len(bs) > 0 && con.isEmptyBlock(bs[0]) {
            bs = bs[1:]
        }
        con.blocks[idx] = bs
    }
    // Build empty blocks.
    for _, bs := range con.blocks {
        for i := range bs {
            if con.isEmptyBlock(bs[i]) {
                if bs[i-1].Position.Height == bs[i].Position.Height-1 {
                    con.buildEmptyBlock(bs[i], bs[i-1])
                }
            }
        }
    }
    var tipRoundMap map[uint64]uint32
    for {
        tipRoundMap = make(map[uint64]uint32)
        for _, bs := range con.blocks {
            if len(bs) > 0 {
                tipRoundMap[bs[0].Position.Round]++
            }
        }
        if len(tipRoundMap) <= 1 {
            break
        }
        // Make all tips in same round.
        var maxRound uint64
        for r := range tipRoundMap {
            if r > maxRound {
                maxRound = r
            }
        }
        for idx, bs := range con.blocks {
            for len(bs) > 0 && bs[0].Position.Round < maxRound {
                bs = bs[1:]
            }
            con.blocks[idx] = bs
        }
    }
    if len(tipRoundMap) == 1 {
        var r uint64
        for r = range tipRoundMap {
            break
        }
        con.logger.Debug("check agreement round cut",
            "tip-round", r,
            "configs", len(con.configs))
        if tipRoundMap[r] == con.configs[r].NumChains {
            con.agreementRoundCut = r
            return true
        }
    }
    return false
}

func (con *Consensus) findLatticeSyncBlock(
    blocks []*types.Block) (*types.Block, error) {
    lastBlock := blocks[len(blocks)-1]
    round := lastBlock.Position.Round
    isConfigChanged := func(prev, cur *types.Config) bool {
        return prev.K != cur.K ||
            prev.NumChains != cur.NumChains ||
            prev.PhiRatio != cur.PhiRatio
    }
    for {
        // Find round r which r-1, r, r+1 are all in same total ordering config.
        for {
            sameAsPrevRound := round == 0 || !isConfigChanged(
                con.configs[round-1], con.configs[round])
            sameAsNextRound := !isConfigChanged(
                con.configs[round], con.configs[round+1])
            if sameAsPrevRound && sameAsNextRound {
                break
            }
            if round == 0 {
                // Unable to find a safe round, wait for new rounds.
                return nil, nil
            }
            round--
        }
        // Find the newset block which round is "round".
        for lastBlock.Position.Round != round {
            if (lastBlock.Finalization.ParentHash == common.Hash{}) {
                return nil, ErrGenesisBlockReached
            }
            b, err := con.db.GetBlock(lastBlock.Finalization.ParentHash)
            if err != nil {
                return nil, err
            }
            lastBlock = &b
        }
        // Find the deliver set by hash for two times. Blocks in a deliver set
        // returned by total ordering is sorted by hash. If a block's parent
        // hash is greater than its hash means there is a cut between deliver
        // sets.
        var curBlock, prevBlock *types.Block
        var deliverSetFirstBlock, deliverSetLastBlock *types.Block
        curBlock = lastBlock
        for {
            if (curBlock.Finalization.ParentHash == common.Hash{}) {
                return nil, ErrGenesisBlockReached
            }
            b, err := con.db.GetBlock(curBlock.Finalization.ParentHash)
            if err != nil {
                return nil, err
            }
            prevBlock = &b
            if !prevBlock.Hash.Less(curBlock.Hash) {
                break
            }
            curBlock = prevBlock
        }
        deliverSetLastBlock = prevBlock
        curBlock = prevBlock
        for {
            if (curBlock.Finalization.ParentHash == common.Hash{}) {
                break
            }
            b, err := con.db.GetBlock(curBlock.Finalization.ParentHash)
            if err != nil {
                return nil, err
            }
            prevBlock = &b
            if !prevBlock.Hash.Less(curBlock.Hash) {
                break
            }
            curBlock = prevBlock
        }
        deliverSetFirstBlock = curBlock
        // Check if all blocks from deliverSetFirstBlock to deliverSetLastBlock
        // are in the same round.
        ok := true
        curBlock = deliverSetLastBlock
        for {
            if curBlock.Position.Round != round {
                ok = false
                break
            }
            b, err := con.db.GetBlock(curBlock.Finalization.ParentHash)
            if err != nil {
                return nil, err
            }
            curBlock = &b
            if curBlock.Hash == deliverSetFirstBlock.Hash {
                break
            }
        }
        if ok {
            return deliverSetFirstBlock, nil
        }
        if round == 0 {
            return nil, nil
        }
        round--
    }
}

func (con *Consensus) processFinalizedBlock(block *types.Block) error {
    if con.lattice == nil {
        return nil
    }
    delivered, err := con.lattice.ProcessFinalizedBlock(block)
    if err != nil {
        return err
    }
    con.lock.Lock()
    defer con.lock.Unlock()
    con.finalizedBlockHashes = append(con.finalizedBlockHashes, block.Hash)
    for idx, b := range delivered {
        if con.finalizedBlockHashes[idx] != b.Hash {
            return ErrMismatchBlockHashSequence
        }
        con.validatedChains[b.Position.ChainID] = struct{}{}
    }
    con.finalizedBlockHashes = con.finalizedBlockHashes[len(delivered):]
    return nil
}

// SyncBlocks syncs blocks from compaction chain, latest is true if the caller
// regards the blocks are the latest ones. Notice that latest can be true for
// many times.
// NOTICE: parameter "blocks" should be consecutive in compaction height.
// NOTICE: this method is not expected to be called concurrently.
func (con *Consensus) SyncBlocks(
    blocks []*types.Block, latest bool) (synced bool, err error) {
    defer func() {
        con.logger.Debug("SyncBlocks returned",
            "synced", synced,
            "error", err,
            "last-block", con.syncedLastBlock)
    }()
    if con.syncedLastBlock != nil {
        synced, err = true, ErrAlreadySynced
        return
    }
    if len(blocks) == 0 {
        return
    }
    // Check if blocks are consecutive.
    for i := 1; i < len(blocks); i++ {
        if blocks[i].Finalization.Height != blocks[i-1].Finalization.Height+1 {
            err = ErrInvalidBlockOrder
            return
        }
    }
    // Make sure the first block is the next block of current compaction chain
    // tip in DB.
    _, tipHeight := con.db.GetCompactionChainTipInfo()
    if blocks[0].Finalization.Height != tipHeight+1 {
        con.logger.Error("mismatched finalization height",
            "now", blocks[0].Finalization.Height,
            "expected", tipHeight+1)
        err = ErrInvalidSyncingFinalizationHeight
        return
    }
    con.logger.Trace("syncBlocks",
        "position", &blocks[0].Position,
        "final height", blocks[0].Finalization.Height,
        "len", len(blocks),
        "latest", latest,
    )
    con.setupConfigs(blocks)
    for _, b := range blocks {
        // TODO(haoping) remove this if lattice puts blocks into db.
        if err = con.db.PutBlock(*b); err != nil {
            // A block might be put into db when confirmed by BA, but not
            // finalized yet.
            if err == db.ErrBlockExists {
                err = con.db.UpdateBlock(*b)
            }
            if err != nil {
                return
            }
        }
        if err = con.db.PutCompactionChainTipInfo(
            b.Hash, b.Finalization.Height); err != nil {
            return
        }
        if err = con.processFinalizedBlock(b); err != nil {
            return
        }
    }
    if latest && con.lattice == nil {
        // New Lattice and find the deliver set of total ordering when "latest"
        // is true for first time. Deliver set is found by block hashes.
        var syncBlock *types.Block
        syncBlock, err = con.findLatticeSyncBlock(blocks)
        if err != nil {
            if err == ErrGenesisBlockReached {
                con.logger.Debug("SyncBlocks skip error", "error", err)
                err = nil
            }
            return
        }
        if syncBlock != nil {
            con.logger.Debug("deliver set found", "block", syncBlock)
            // New lattice with the round of syncBlock.
            con.initConsensusObj(syncBlock)
            con.setupConfigs(blocks)
            // Process blocks from syncBlock to blocks' last block.
            b := blocks[len(blocks)-1]
            blocksCount :=
                b.Finalization.Height - syncBlock.Finalization.Height + 1
            blocksToProcess := make([]*types.Block, blocksCount)
            for {
                blocksToProcess[blocksCount-1] = b
                blocksCount--
                if b.Hash == syncBlock.Hash {
                    break
                }
                var b1 types.Block
                b1, err = con.db.GetBlock(b.Finalization.ParentHash)
                if err != nil {
                    return
                }
                b = &b1
            }
            for _, b := range blocksToProcess {
                if err = con.processFinalizedBlock(b); err != nil {
                    return
                }
            }
        }
    }
    if latest && con.ensureAgreementOverlapRound() {
        // Check if compaction and agreements' blocks are overlapped. The
        // overlapping of compaction chain and BA's oldest blocks means the
        // syncing is done.
        if con.checkIfValidated() && con.checkIfSynced(blocks) {
            if err = con.Stop(); err != nil {
                return
            }
            con.dummyCancel, con.dummyFinished = utils.LaunchDummyReceiver(
                context.Background(), con.network.ReceiveChan(),
                func(msg interface{}) {
                    con.dummyMsgBuffer = append(con.dummyMsgBuffer, msg)
                })
            con.syncedLastBlock = blocks[len(blocks)-1]
            synced = true
        }
    }
    return
}

// GetSyncedConsensus returns the core.Consensus instance after synced.
func (con *Consensus) GetSyncedConsensus() (*core.Consensus, error) {
    con.lock.Lock()
    defer con.lock.Unlock()
    if con.syncedConsensus != nil {
        return con.syncedConsensus, nil
    }
    if con.syncedLastBlock == nil {
        return nil, ErrNotSynced
    }
    // flush all blocks in con.blocks into core.Consensus, and build
    // core.Consensus from syncer.
    confirmedBlocks := make([][]*types.Block, len(con.blocks))
    for i, bs := range con.blocks {
        confirmedBlocks[i] = []*types.Block(bs)
    }
    randomnessResults := []*types.BlockRandomnessResult{}
    for _, r := range con.randomnessResults {
        randomnessResults = append(randomnessResults, r)
    }
    con.dummyCancel()
    <-con.dummyFinished
    var err error
    con.syncedConsensus, err = core.NewConsensusFromSyncer(
        con.syncedLastBlock,
        con.roundBeginTimes[con.syncedLastBlock.Position.Round],
        con.app,
        con.gov,
        con.db,
        con.network,
        con.prv,
        con.lattice,
        confirmedBlocks,
        randomnessResults,
        con.dummyMsgBuffer,
        con.logger)
    return con.syncedConsensus, err
}

// Stop the syncer.
//
// This method is mainly for caller to stop the syncer before synced, the syncer
// would call this method automatically after being synced.
func (con *Consensus) Stop() error {
    con.logger.Trace("syncer is about to stop")
    // Stop network and CRS routines, wait until they are all stoped.
    con.ctxCancel()
    con.logger.Trace("stop syncer modules")
    con.moduleWaitGroup.Wait()
    // Stop agreements.
    con.logger.Trace("stop syncer agreement modules")
    con.stopAgreement()
    con.logger.Trace("syncer stopped")
    return nil
}

// isEmptyBlock checks if a block is an empty block by both its hash and parent
// hash are empty.
func (con *Consensus) isEmptyBlock(b *types.Block) bool {
    return b.Hash == common.Hash{} && b.ParentHash == common.Hash{}
}

// buildEmptyBlock builds an empty block in agreement.
func (con *Consensus) buildEmptyBlock(b *types.Block, parent *types.Block) {
    cfg := con.configs[b.Position.Round]
    b.Timestamp = parent.Timestamp.Add(cfg.MinBlockInterval)
    b.Witness.Height = parent.Witness.Height
    b.Witness.Data = make([]byte, len(parent.Witness.Data))
    copy(b.Witness.Data, parent.Witness.Data)
    b.Acks = common.NewSortedHashes(common.Hashes{parent.Hash})
}

func (con *Consensus) setupConfigsUntilRound(round uint64) {
    curMaxNumChains := uint32(0)
    func() {
        con.lock.Lock()
        defer con.lock.Unlock()
        con.logger.Debug("syncer setupConfigs",
            "until-round", round,
            "length", len(con.configs),
            "lattice", con.latticeLastRound)
        for r := uint64(len(con.configs)); r <= round; r++ {
            cfg := utils.GetConfigWithPanic(con.gov, r, con.logger)
            con.configs = append(con.configs, cfg)
            con.roundBeginTimes = append(
                con.roundBeginTimes,
                con.roundBeginTimes[r-1].Add(con.configs[r-1].RoundInterval))
            if cfg.NumChains >= curMaxNumChains {
                curMaxNumChains = cfg.NumChains
            }
        }
        // Notify core.Lattice for new configs.
        if con.lattice != nil {
            for con.latticeLastRound+1 <= round {
                con.latticeLastRound++
                if err := con.lattice.AppendConfig(
                    con.latticeLastRound,
                    con.configs[con.latticeLastRound]); err != nil {
                    panic(err)
                }
            }
        }
    }()
    con.resizeByNumChains(curMaxNumChains)
    con.logger.Trace("setupConfgis finished", "round", round)
}

// setupConfigs is called by SyncBlocks with blocks from compaction chain. In
// the first time, setupConfigs setups from round 0.
func (con *Consensus) setupConfigs(blocks []*types.Block) {
    // Find max round in blocks.
    var maxRound uint64
    for _, b := range blocks {
        if b.Position.Round > maxRound {
            maxRound = b.Position.Round
        }
    }
    // Get configs from governance.
    //
    // In fullnode, the notification of new round is yet another TX, which
    // needs to be executed after corresponding block delivered. Thus, the
    // configuration for 'maxRound + core.ConfigRoundShift' won't be ready when
    // seeing this block.
    con.setupConfigsUntilRound(maxRound + core.ConfigRoundShift - 1)
}

// resizeByNumChains resizes fake lattice and agreement if numChains increases.
// Notice the decreasing case is neglected.
func (con *Consensus) resizeByNumChains(numChains uint32) {
    con.lock.Lock()
    defer con.lock.Unlock()
    if numChains > uint32(len(con.blocks)) {
        for i := uint32(len(con.blocks)); i < numChains; i++ {
            // Resize the pool of blocks.
            con.blocks = append(con.blocks, types.ByPosition{})
            // Resize agreement modules.
            a := newAgreement(
                con.receiveChan, con.pullChan, con.nodeSetCache, con.logger)
            con.agreements = append(con.agreements, a)
            con.agreementWaitGroup.Add(1)
            go func() {
                defer con.agreementWaitGroup.Done()
                a.run()
            }()
        }
    }
}

// startAgreement starts agreements for receiving votes and agreements.
func (con *Consensus) startAgreement() {
    // Start a routine for listening receive channel and pull block channel.
    go func() {
        for {
            select {
            case b, ok := <-con.receiveChan:
                if !ok {
                    return
                }
                chainID := b.Position.ChainID
                func() {
                    con.lock.Lock()
                    defer con.lock.Unlock()
                    // If round is cut in agreements, do not add blocks with
                    // round less then cut round.
                    if b.Position.Round < con.agreementRoundCut {
                        return
                    }
                    con.blocks[chainID] = append(con.blocks[chainID], b)
                    sort.Sort(con.blocks[chainID])
                }()
            case h, ok := <-con.pullChan:
                if !ok {
                    return
                }
                con.network.PullBlocks(common.Hashes{h})
            }
        }
    }()
}

func (con *Consensus) cacheRandomnessResult(r *types.BlockRandomnessResult) {
    // There is no block randomness at round-0.
    if r.Position.Round == 0 {
        return
    }
    // We only have to cache randomness result after cutting round.
    if r.Position.Round < func() uint64 {
        con.lock.RLock()
        defer con.lock.RUnlock()
        return con.agreementRoundCut
    }() {
        return
    }
    if func() (exists bool) {
        con.lock.RLock()
        defer con.lock.RUnlock()
        _, exists = con.randomnessResults[r.BlockHash]
        return
    }() {
        return
    }
    v, ok, err := con.tsigVerifier.UpdateAndGet(r.Position.Round)
    if err != nil {
        con.logger.Error("Unable to get tsig verifier",
            "hash", r.BlockHash.String()[:6],
            "position", &r.Position,
            "error", err)
        return
    }
    if !ok {
        con.logger.Error("Tsig is not ready", "position", &r.Position)
        return
    }
    if !v.VerifySignature(r.BlockHash, crypto.Signature{
        Type:      "bls",
        Signature: r.Randomness}) {
        con.logger.Info("Block randomness is not valid",
            "position", &r.Position,
            "hash", r.BlockHash.String()[:6])
        return
    }
    con.lock.Lock()
    defer con.lock.Unlock()
    con.randomnessResults[r.BlockHash] = r
}

// startNetwork starts network for receiving blocks and agreement results.
func (con *Consensus) startNetwork() {
    con.moduleWaitGroup.Add(1)
    go func() {
        defer con.moduleWaitGroup.Done()
    Loop:
        for {
            select {
            case val := <-con.network.ReceiveChan():
                var pos types.Position
                switch v := val.(type) {
                case *types.Block:
                    pos = v.Position
                case *types.AgreementResult:
                    pos = v.Position
                case *types.BlockRandomnessResult:
                    con.cacheRandomnessResult(v)
                    continue Loop
                default:
                    continue Loop
                }
                if func() bool {
                    con.lock.RLock()
                    defer con.lock.RUnlock()
                    if pos.ChainID >= uint32(len(con.agreements)) {
                        // This error might be easily encountered when the
                        // "latest" parameter of SyncBlocks is turned on too
                        // early.
                        con.logger.Error(
                            "Unknown chainID message received (syncer)",
                            "position", &pos)
                        return false
                    }
                    return true
                }() {
                    con.agreements[pos.ChainID].inputChan <- val
                }
            case <-con.ctx.Done():
                return
            }
        }
    }()
}

// startCRSMonitor is the dummiest way to verify if the CRS for one round
// is ready or not.
func (con *Consensus) startCRSMonitor() {
    var lastNotifiedRound uint64
    // Notify all agreements for new CRS.
    notifyNewCRS := func(round uint64) {
        con.setupConfigsUntilRound(round)
        if round == lastNotifiedRound {
            return
        }
        con.logger.Debug("CRS is ready", "round", round)
        lastNotifiedRound = round
        con.lock.Lock()
        defer con.lock.Unlock()
        for idx, a := range con.agreements {
        loop:
            for {
                select {
                case <-con.ctx.Done():
                    break loop
                case a.inputChan <- round:
                    break loop
                case <-time.After(500 * time.Millisecond):
                    con.logger.Debug(
                        "agreement input channel is full when putting CRS",
                        "chainID", idx,
                        "round", round)
                }
            }
        }
    }
    con.moduleWaitGroup.Add(1)
    go func() {
        defer con.moduleWaitGroup.Done()
        for {
            select {
            case <-con.ctx.Done():
                return
            case <-time.After(500 * time.Millisecond):
            }
            // Notify agreement modules for the latest round that CRS is
            // available if the round is not notified yet.
            checked := lastNotifiedRound + 1
            for (con.gov.CRS(checked) != common.Hash{}) {
                checked++
            }
            checked--
            if checked > lastNotifiedRound {
                notifyNewCRS(checked)
            }
        }
    }()
}

func (con *Consensus) stopAgreement() {
    func() {
        con.lock.Lock()
        defer con.lock.Unlock()
        for _, a := range con.agreements {
            if a.inputChan != nil {
                close(a.inputChan)
                a.inputChan = nil
            }
        }
    }()
    con.agreementWaitGroup.Wait()
    close(con.receiveChan)
    close(con.pullChan)
}