path: root/core/lattice-data.go

// Copyright 2018 The dexon-consensus-core Authors
// This file is part of the dexon-consensus-core library.
//
// The dexon-consensus-core 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-core 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-core library. If not, see
// <http://www.gnu.org/licenses/>.

package core

import (
    "errors"
    "fmt"
    "time"

    "github.com/dexon-foundation/dexon-consensus-core/common"
    "github.com/dexon-foundation/dexon-consensus-core/core/blockdb"
    "github.com/dexon-foundation/dexon-consensus-core/core/types"
)

// Errors for sanity check error.
var (
    ErrAckingBlockNotExists    = fmt.Errorf("acking block not exists")
    ErrInvalidParentChain      = fmt.Errorf("invalid parent chain")
    ErrDuplicatedAckOnOneChain = fmt.Errorf("duplicated ack on one chain")
    ErrChainStatusCorrupt      = fmt.Errorf("chain status corrupt")
    ErrInvalidChainID          = fmt.Errorf("invalid chain id")
    ErrInvalidProposerID       = fmt.Errorf("invalid proposer id")
    ErrInvalidTimestamp        = fmt.Errorf("invalid timestamp")
    ErrInvalidWitness          = fmt.Errorf("invalid witness data")
    ErrInvalidBlock            = fmt.Errorf("invalid block")
    ErrForkBlock               = fmt.Errorf("fork block")
    ErrNotAckParent            = fmt.Errorf("not ack parent")
    ErrDoubleAck               = fmt.Errorf("double ack")
    ErrAcksNotSorted           = fmt.Errorf("acks not sorted")
    ErrInvalidBlockHeight      = fmt.Errorf("invalid block height")
    ErrAlreadyInLattice        = fmt.Errorf("block already in lattice")
    ErrIncorrectBlockTime      = fmt.Errorf("block timestamp is incorrect")
    ErrUnknownRoundID          = fmt.Errorf("unknown round id")
)

// Errors for method usage
var (
    ErrRoundNotIncreasing  = errors.New("round not increasing")
    ErrPurgedBlockNotFound = errors.New("purged block not found")
)

// latticeDataConfig is the configuration for latticeData for each round.
type latticeDataConfig struct {
    // Number of chains between runs
    prevNumChains uint32
    numChains     uint32
    // Block interval specifies reasonable time difference between
    // parent/child blocks.
    minBlockTimeInterval time.Duration
    maxBlockTimeInterval time.Duration
    // the range of clean cut at the beginning of this round.
    cleanCutRange int
}

// newLatticeDataConfig constructs a latticeDataConfig instance.
func newLatticeDataConfig(prev, cur *types.Config) *latticeDataConfig {
    config := &latticeDataConfig{
        numChains:            cur.NumChains,
        minBlockTimeInterval: cur.MinBlockInterval,
        maxBlockTimeInterval: cur.MaxBlockInterval,
    }
    if prev != nil {
        config.prevNumChains = prev.NumChains
        if prev.K != cur.K ||
            prev.PhiRatio != cur.PhiRatio ||
            prev.NumChains != cur.NumChains {
            // K plus 2 is a magic from GOD.
            config.cleanCutRange = prev.K + 2
        }
    }
    return config
}

// latticeData is a module for storing lattice.
type latticeData struct {
    // we need blockdb to read blocks purged from cache in memory.
    db blockdb.Reader
    // chains stores chains' blocks and other info.
    chains []*chainStatus
    // blockByHash stores blocks, indexed by block hash.
    blockByHash map[common.Hash]*types.Block
    // This stores configuration for each round.
    configs  []*latticeDataConfig
    minRound uint64
}

// newLatticeData creates a new latticeData struct.
func newLatticeData(db blockdb.Reader,
    round uint64, config *latticeDataConfig) (data *latticeData) {
    data = &latticeData{
        db:          db,
        chains:      make([]*chainStatus, config.numChains),
        blockByHash: make(map[common.Hash]*types.Block),
        configs:     []*latticeDataConfig{config},
        minRound:    round,
    }
    for i := range data.chains {
        data.chains[i] = &chainStatus{
            ID:      uint32(i),
            blocks:  []*types.Block{},
            nextAck: make([]uint64, config.numChains),
        }
    }
    return
}

func (data *latticeData) sanityCheck(b *types.Block) error {
    config := data.getConfig(b.Position.Round)
    if config == nil {
        return ErrUnknownRoundID
    }
    // Check if the chain id is valid.
    if b.Position.ChainID >= config.numChains {
        return ErrInvalidChainID
    }

    // TODO(mission): Check if its proposer is in validator set somewhere,
    //                lattice doesn't have to know about node set.

    // Check if it forks
    if bInLattice := data.chains[b.Position.ChainID].getBlockByHeight(
        b.Position.Height); bInLattice != nil {

        if b.Hash != bInLattice.Hash {
            return ErrForkBlock
        }
        return ErrAlreadyInLattice
    }
    // TODO(mission): check if fork by loading blocks from DB if the block
    //                doesn't exists because forking is serious.

    // Check if it acks older blocks.
    acksByChainID := make(map[uint32]struct{}, len(data.chains))
    for _, hash := range b.Acks {
        bAck, err := data.findBlock(hash)
        if err != nil {
            if err == blockdb.ErrBlockDoesNotExist {
                return ErrAckingBlockNotExists
            }
            return err
        }
        if bAck.Position.Height <
            data.chains[bAck.Position.ChainID].nextAck[b.Position.ChainID] {
            return ErrDoubleAck
        }
        // Check if ack two blocks on the same chain. This would need
        // to check after we replace map with slice for acks.
        if _, acked := acksByChainID[bAck.Position.ChainID]; acked {
            return ErrDuplicatedAckOnOneChain
        }
        acksByChainID[bAck.Position.ChainID] = struct{}{}
    }

    // Check non-genesis blocks if it acks its parent.
    if b.Position.Height > 0 {
        if !b.IsAcking(b.ParentHash) {
            return ErrNotAckParent
        }
        bParent, err := data.findBlock(b.ParentHash)
        if err != nil {
            // This error should never happened, a parent block is always
            // an acked block.
            panic(err)
        }
        if bParent.Position.ChainID != b.Position.ChainID {
            return ErrInvalidParentChain
        }
        if bParent.Position.Height != b.Position.Height-1 {
            return ErrInvalidBlockHeight
        }
        if bParent.Witness.Height > b.Witness.Height {
            return ErrInvalidWitness
        }
        // Check if its timestamp is valid.
        if !b.Timestamp.After(bParent.Timestamp) {
            return ErrInvalidTimestamp
        }
        // Check if its timestamp is in expected range.
        if b.Timestamp.Before(
            bParent.Timestamp.Add(config.minBlockTimeInterval)) ||
            b.Timestamp.After(
                bParent.Timestamp.Add(config.maxBlockTimeInterval)) {

            return ErrIncorrectBlockTime
        }
    }
    return nil
}

// addBlock processes block, it does sanity check, inserts block into
// lattice and deletes blocks which will not be used.
func (data *latticeData) addBlock(
    block *types.Block) (deliverable []*types.Block, err error) {

    var (
        bAck    *types.Block
        updated bool
    )
    // TODO(mission): sanity check twice, might hurt performance.
    // If a block does not pass sanity check, report error.
    if err = data.sanityCheck(block); err != nil {
        return
    }
    if err = data.chains[block.Position.ChainID].addBlock(block); err != nil {
        return
    }
    data.blockByHash[block.Hash] = block
    // Update nextAcks.
    for _, ack := range block.Acks {
        if bAck, err = data.findBlock(ack); err != nil {
            return
        }
        data.chains[bAck.Position.ChainID].nextAck[block.Position.ChainID] =
            bAck.Position.Height + 1
    }
    // Extract blocks that deliverable to total ordering.
    // A block is deliverable to total ordering iff:
    //  - All its acking blocks are delivered to total ordering.
    for {
        updated = false
        for _, status := range data.chains {
            tip := status.getBlockByHeight(status.nextOutput)
            if tip == nil {
                continue
            }
            allAckingBlockDelivered := true
            for _, ack := range tip.Acks {
                if bAck, err = data.findBlock(ack); err != nil {
                    return
                }
                if data.chains[bAck.Position.ChainID].nextOutput >
                    bAck.Position.Height {

                    continue
                }
                // This acked block exists and not delivered yet.
                allAckingBlockDelivered = false
            }
            if allAckingBlockDelivered {
                deliverable = append(deliverable, tip)
                status.nextOutput++
                updated = true
            }
        }
        if !updated {
            break
        }
    }

    // Delete old blocks in "chains" and "blocks" to release memory space.
    //
    // A block is safe to be deleted iff:
    //  - It's delivered to total ordering
    //  - All chains (including its proposing chain) acks some block with
    //    higher height in its proposing chain.
    //
    // This works because blocks of height below this minimum are not going to be
    // acked anymore, the ackings of these blocks are illegal.
    for _, status := range data.chains {
        status.purge()
    }
    return
}

// prepareBlock helps to setup fields of block based on its ProposerID,
// including:
//  - Set 'Acks' and 'Timestamps' for the highest block of each validator not
//    acked by this proposer before.
//  - Set 'ParentHash' and 'Height' from parent block, if we can't find a
//    parent, these fields would be setup like a genesis block.
func (data *latticeData) prepareBlock(block *types.Block) {
    config := data.getConfig(block.Position.Round)
    // Reset fields to make sure we got these information from parent block.
    block.Position.Height = 0
    block.ParentHash = common.Hash{}
    acks := common.Hashes{}
    for chainID := range data.chains {
        // find height of the latest block for that validator.
        var (
            curBlock   *types.Block
            nextHeight = data.chains[chainID].nextAck[block.Position.ChainID]
        )
        for {
            tmpBlock := data.chains[chainID].getBlockByHeight(nextHeight)
            if tmpBlock == nil {
                break
            }
            curBlock = tmpBlock
            nextHeight++
        }
        if curBlock == nil {
            continue
        }
        acks = append(acks, curBlock.Hash)
        if uint32(chainID) == block.Position.ChainID {
            block.ParentHash = curBlock.Hash
            block.Position.Height = curBlock.Position.Height + 1
            block.Witness.Height = curBlock.Witness.Height
            minTimestamp := curBlock.Timestamp.Add(config.minBlockTimeInterval)
            maxTimestamp := curBlock.Timestamp.Add(config.maxBlockTimeInterval)
            if block.Timestamp.Before(minTimestamp) {
                block.Timestamp = minTimestamp
            } else if block.Timestamp.After(maxTimestamp) {
                block.Timestamp = maxTimestamp
            }
        }
    }
    block.Acks = common.NewSortedHashes(acks)
    return
}

// TODO(mission): make more abstraction for this method.
// nextHeight returns the next height for the chain.
func (data *latticeData) nextPosition(chainID uint32) types.Position {
    return data.chains[chainID].nextPosition()
}

// findBlock seeks blocks in memory or db.
func (data *latticeData) findBlock(h common.Hash) (b *types.Block, err error) {
    if b = data.blockByHash[h]; b != nil {
        return
    }
    var tmpB types.Block
    if tmpB, err = data.db.Get(h); err != nil {
        return
    }
    b = &tmpB
    return
}

// purgeBlocks purges blocks from cache.
func (data *latticeData) purgeBlocks(blocks []*types.Block) error {
    for _, b := range blocks {
        if _, exists := data.blockByHash[b.Hash]; !exists {
            return ErrPurgedBlockNotFound
        }
        delete(data.blockByHash, b.Hash)
    }
    return nil
}

// getConfig get configuration for lattice-data by round ID.
func (data *latticeData) getConfig(round uint64) (config *latticeDataConfig) {
    if round < data.minRound {
        return
    }
    diff := round - data.minRound
    if diff >= uint64(len(data.configs)) {
        return
    }
    return data.configs[diff]
}

// appendConfig appends a configuration for upcoming round. When you append
// a config for round R, next time you can only append the config for round R+1.
func (data *latticeData) appendConfig(
    round uint64, config *latticeDataConfig) (err error) {

    if round != data.minRound+uint64(len(data.configs)) {
        return ErrRoundNotIncreasing
    }
    data.configs = append(data.configs, config)
    return nil
}

type chainStatus struct {
    // ID keeps the chainID of this chain status.
    ID uint32

    // blocks stores blocks proposed for this chain, sorted by height.
    blocks []*types.Block

    // minHeight keeps minimum height in blocks.
    minHeight uint64

    // nextAck stores the height of next height that should be acked, i.e. last
    // acked height + 1. Initialized to 0.
    // being acked. For example, rb.chains[vid1].nextAck[vid2] - 1 is the last
    // acked height by vid2 acking vid1.
    nextAck []uint64

    // nextOutput is the next output height of block, default to 0.
    nextOutput uint64
}

func (s *chainStatus) getBlockByHeight(height uint64) (b *types.Block) {
    if height < s.minHeight {
        return
    }
    idx := int(height - s.minHeight)
    if idx >= len(s.blocks) {
        return
    }
    b = s.blocks[idx]
    return
}

func (s *chainStatus) addBlock(b *types.Block) error {
    if len(s.blocks) > 0 {
        // Make sure the height of incoming block should be
        // plus one to current latest blocks if exists.
        if s.blocks[len(s.blocks)-1].Position.Height != b.Position.Height-1 {
            return ErrChainStatusCorrupt
        }
    } else {
        if b.Position.Height != 0 {
            return ErrChainStatusCorrupt
        }
    }
    s.blocks = append(s.blocks, b)
    return nil
}

func (s *chainStatus) calcPurgeHeight() (safe uint64, ok bool) {
    // blocks with height less than min(nextOutput, nextAck...)
    // are safe to be purged.
    safe = s.nextOutput
    for _, ackedHeight := range s.nextAck {
        if safe > ackedHeight {
            safe = ackedHeight
        }
    }
    // Both 'nextOutput' and 'nextAck' represents some block to be
    // outputed/acked. To find a block already outputed/acked, the height
    // needs to be minus 1.
    if safe == 0 {
        // Avoid underflow.
        return
    }
    safe--
    if safe < s.minHeight {
        return
    }
    ok = true
    return
}

// purge blocks if they are safe to be deleted from working set.
func (s *chainStatus) purge() {
    safe, ok := s.calcPurgeHeight()
    if !ok {
        return
    }
    newMinIndex := safe - s.minHeight + 1
    s.blocks = s.blocks[newMinIndex:]
    s.minHeight = safe + 1
    return
}

// nextPosition returns a valid position for new block in this chain.
func (s *chainStatus) nextPosition() types.Position {
    return types.Position{
        ChainID: s.ID,
        Height:  s.minHeight + uint64(len(s.blocks)),
    }
}