dione/consensus/consensus.go

408 lines
11 KiB
Go
Raw Normal View History

package consensus
import (
"encoding/hex"
"errors"
2020-11-15 13:46:58 +00:00
"math/big"
"sync"
"github.com/libp2p/go-libp2p-core/peer"
2021-07-11 23:23:00 +00:00
"github.com/Secured-Finance/dione/beacon"
"github.com/fxamacker/cbor/v2"
"github.com/Secured-Finance/dione/cache"
"github.com/asaskevich/EventBus"
"github.com/Secured-Finance/dione/blockchain"
types3 "github.com/Secured-Finance/dione/blockchain/types"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/Secured-Finance/dione/blockchain/pool"
"github.com/Secured-Finance/dione/consensus/types"
types2 "github.com/Secured-Finance/dione/types"
2020-11-15 13:46:58 +00:00
"github.com/Secured-Finance/dione/ethclient"
"github.com/sirupsen/logrus"
"github.com/Secured-Finance/dione/pubsub"
)
2021-07-11 00:32:58 +00:00
var (
ErrNoAcceptedBlocks = errors.New("there is no accepted blocks")
)
type StateStatus uint8
const (
StateStatusUnknown = iota
StateStatusPrePrepared
StateStatusPrepared
StateStatusCommited
)
type PBFTConsensusManager struct {
bus EventBus.Bus
psb *pubsub.PubSubRouter
minApprovals int // FIXME
privKey crypto.PrivKey
msgLog *ConsensusMessageLog
validator *ConsensusValidator
ethereumClient *ethclient.EthereumClient
miner *blockchain.Miner
2021-07-11 00:32:58 +00:00
blockPool *pool.BlockPool
2021-07-11 23:23:00 +00:00
mempool *pool.Mempool
blockchain *blockchain.BlockChain
state *State
address peer.ID
}
type State struct {
mutex sync.Mutex
drandRound uint64
randomness []byte
blockHeight uint64
status StateStatus
2021-07-11 23:23:00 +00:00
ready bool
}
2021-07-11 00:32:58 +00:00
func NewPBFTConsensusManager(
bus EventBus.Bus,
psb *pubsub.PubSubRouter,
minApprovals int,
privKey crypto.PrivKey,
ethereumClient *ethclient.EthereumClient,
miner *blockchain.Miner,
2021-07-11 00:32:58 +00:00
bc *blockchain.BlockChain,
bp *pool.BlockPool,
2021-07-11 23:23:00 +00:00
b beacon.BeaconNetwork,
mempool *pool.Mempool,
address peer.ID,
2021-07-11 00:32:58 +00:00
) *PBFTConsensusManager {
pcm := &PBFTConsensusManager{
psb: psb,
miner: miner,
validator: NewConsensusValidator(miner, bc, b),
msgLog: NewConsensusMessageLog(),
minApprovals: minApprovals,
privKey: privKey,
ethereumClient: ethereumClient,
state: &State{
ready: false,
status: StateStatusUnknown,
},
bus: bus,
blockPool: bp,
mempool: mempool,
blockchain: bc,
address: address,
}
2021-07-11 23:23:00 +00:00
pcm.psb.Hook(pubsub.PrePrepareMessageType, pcm.handlePrePrepare)
pcm.psb.Hook(pubsub.PrepareMessageType, pcm.handlePrepare)
pcm.psb.Hook(pubsub.CommitMessageType, pcm.handleCommit)
bus.SubscribeAsync("beacon:newEntry", func(entry types2.BeaconEntry) {
pcm.onNewBeaconEntry(entry)
}, true)
2021-07-11 23:23:00 +00:00
height, _ := pcm.blockchain.GetLatestBlockHeight()
pcm.state.blockHeight = height + 1
return pcm
}
func (pcm *PBFTConsensusManager) propose(blk *types3.Block) error {
prePrepareMsg, err := NewMessage(types.ConsensusMessage{Block: blk}, types.ConsensusMessageTypePrePrepare, pcm.privKey)
2020-11-15 13:46:58 +00:00
if err != nil {
return err
}
pcm.psb.BroadcastToServiceTopic(prePrepareMsg)
pcm.blockPool.AddBlock(blk)
pcm.state.status = StateStatusPrePrepared
2020-11-15 13:46:58 +00:00
return nil
}
2021-07-11 23:23:00 +00:00
func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage) {
//pcm.state.mutex.Lock()
//defer pcm.state.mutex.Unlock()
2021-07-11 23:23:00 +00:00
var prePrepare types.PrePrepareMessage
err := cbor.Unmarshal(message.Payload, &prePrepare)
if err != nil {
logrus.Errorf("failed to convert payload to PrePrepare message: %s", err.Error())
2021-06-02 21:19:52 +00:00
return
}
if *prePrepare.Block.Header.Proposer == pcm.address {
return
}
cmsg := types.ConsensusMessage{
2021-07-11 23:23:00 +00:00
Type: types.ConsensusMessageTypePrePrepare,
From: message.From,
Block: prePrepare.Block,
Blockhash: prePrepare.Block.Header.Hash,
}
2021-06-02 21:19:52 +00:00
if pcm.msgLog.Exists(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Tracef("received existing pre_prepare msg for block %x", cmsg.Block.Header.Hash)
2020-11-15 13:46:58 +00:00
return
}
if !pcm.validator.Valid(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Warnf("received invalid pre_prepare msg for block %x", cmsg.Block.Header.Hash)
2020-11-15 13:46:58 +00:00
return
}
2021-06-02 21:19:52 +00:00
pcm.msgLog.AddMessage(cmsg)
pcm.blockPool.AddBlock(cmsg.Block)
prepareMsg, err := NewMessage(cmsg, types.ConsensusMessageTypePrepare, pcm.privKey)
2020-11-15 13:46:58 +00:00
if err != nil {
logrus.Errorf("failed to create prepare message: %v", err)
return
2020-11-15 13:46:58 +00:00
}
pcm.psb.BroadcastToServiceTopic(prepareMsg)
pcm.state.status = StateStatusPrePrepared
}
2021-07-11 23:23:00 +00:00
func (pcm *PBFTConsensusManager) handlePrepare(message *pubsub.PubSubMessage) {
//pcm.state.mutex.Lock()
//defer pcm.state.mutex.Unlock()
2021-07-11 23:23:00 +00:00
var prepare types.PrepareMessage
err := cbor.Unmarshal(message.Payload, &prepare)
if err != nil {
logrus.Errorf("failed to convert payload to Prepare message: %s", err.Error())
return
}
cmsg := types.ConsensusMessage{
Type: types.ConsensusMessageTypePrepare,
From: message.From,
Blockhash: prepare.Blockhash,
Signature: prepare.Signature,
}
if _, err := pcm.blockPool.GetBlock(cmsg.Blockhash); errors.Is(err, cache.ErrNotFound) {
2021-07-11 23:23:00 +00:00
logrus.Warnf("received unknown block %x", cmsg.Blockhash)
2021-06-02 21:19:52 +00:00
return
}
if pcm.msgLog.Exists(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Tracef("received existing prepare msg for block %x", cmsg.Blockhash)
2020-11-15 13:46:58 +00:00
return
}
if !pcm.validator.Valid(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Warnf("received invalid prepare msg for block %x", cmsg.Blockhash)
return
}
2020-11-15 13:46:58 +00:00
2021-06-02 21:19:52 +00:00
pcm.msgLog.AddMessage(cmsg)
2020-11-15 13:46:58 +00:00
if len(pcm.msgLog.Get(types.ConsensusMessageTypePrepare, cmsg.Blockhash)) >= pcm.minApprovals {
commitMsg, err := NewMessage(cmsg, types.ConsensusMessageTypeCommit, pcm.privKey)
if err != nil {
logrus.Errorf("failed to create commit message: %v", err)
2021-07-11 23:23:00 +00:00
return
}
pcm.psb.BroadcastToServiceTopic(commitMsg)
pcm.state.status = StateStatusPrepared
}
}
2021-07-11 23:23:00 +00:00
func (pcm *PBFTConsensusManager) handleCommit(message *pubsub.PubSubMessage) {
//pcm.state.mutex.Lock()
//defer pcm.state.mutex.Unlock()
2021-07-11 23:23:00 +00:00
var commit types.CommitMessage
err := cbor.Unmarshal(message.Payload, &commit)
if err != nil {
logrus.Errorf("failed to convert payload to Commit message: %s", err.Error())
return
}
cmsg := types.ConsensusMessage{
Type: types.ConsensusMessageTypeCommit,
From: message.From,
Blockhash: commit.Blockhash,
2021-07-11 23:23:00 +00:00
Signature: commit.Signature,
}
if _, err := pcm.blockPool.GetBlock(cmsg.Blockhash); errors.Is(err, cache.ErrNotFound) {
2021-07-11 23:23:00 +00:00
logrus.Warnf("received unknown block %x", cmsg.Blockhash)
2021-06-02 21:19:52 +00:00
return
}
if pcm.msgLog.Exists(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Tracef("received existing commit msg for block %x", cmsg.Blockhash)
return
}
if !pcm.validator.Valid(cmsg) {
2021-07-11 23:23:00 +00:00
logrus.Warnf("received invalid commit msg for block %x", cmsg.Blockhash)
return
}
2021-06-02 21:19:52 +00:00
pcm.msgLog.AddMessage(cmsg)
if len(pcm.msgLog.Get(types.ConsensusMessageTypeCommit, cmsg.Blockhash)) >= pcm.minApprovals {
block, err := pcm.blockPool.GetBlock(cmsg.Blockhash)
if err != nil {
2021-07-11 23:23:00 +00:00
logrus.Error(err)
return
}
pcm.blockPool.AddAcceptedBlock(block)
pcm.state.status = StateStatusCommited
}
}
func (pcm *PBFTConsensusManager) onNewBeaconEntry(entry types2.BeaconEntry) {
block, err := pcm.commitAcceptedBlocks()
if err != nil {
if errors.Is(err, ErrNoAcceptedBlocks) {
logrus.WithFields(logrus.Fields{
"height": pcm.state.blockHeight,
}).Infof("No accepted blocks in the current consensus round")
} else {
logrus.WithFields(logrus.Fields{
"height": pcm.state.blockHeight,
"err": err.Error(),
}).Errorf("Failed to select the block in the current consensus round")
return
}
}
if block != nil {
// broadcast new block
var newBlockMessage pubsub.PubSubMessage
newBlockMessage.Type = pubsub.NewBlockMessageType
blockSerialized, err := cbor.Marshal(block)
if err != nil {
logrus.Errorf("Failed to serialize block %x for broadcasting!", block.Header.Hash)
} else {
newBlockMessage.Payload = blockSerialized
pcm.psb.BroadcastToServiceTopic(&newBlockMessage)
}
// if we are miner of this block
// then post dione tasks to target chains (currently, only Ethereum)
if block.Header.Proposer.String() == pcm.address.String() {
pcm.submitTasksFromBlock(block)
2021-07-11 23:23:00 +00:00
}
pcm.state.blockHeight = pcm.state.blockHeight + 1
}
// get latest block
height, err := pcm.blockchain.GetLatestBlockHeight()
if err != nil {
logrus.Error(err)
return
}
blockHeader, err := pcm.blockchain.FetchBlockHeaderByHeight(height)
if err != nil {
logrus.Error(err)
return
}
2021-07-11 23:23:00 +00:00
pcm.state.drandRound = entry.Round
pcm.state.randomness = entry.Data
minedBlock, err := pcm.miner.MineBlock(entry.Data, entry.Round, blockHeader)
if err != nil {
if errors.Is(err, blockchain.ErrNoTxForBlock) {
logrus.Info("Sealing skipped, no transactions in mempool")
} else {
logrus.Errorf("Failed to mine the block: %s", err.Error())
2021-07-11 23:23:00 +00:00
}
return
}
// if we are round winner
if minedBlock != nil {
logrus.WithField("height", pcm.state.blockHeight).Infof("We have been elected in the current consensus round")
err = pcm.propose(minedBlock)
if err != nil {
logrus.Errorf("Failed to propose the block: %s", err.Error())
return
}
}
}
func (pcm *PBFTConsensusManager) submitTasksFromBlock(block *types3.Block) {
for _, tx := range block.Data {
var task types2.DioneTask
err := cbor.Unmarshal(tx.Data, &task)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to unmarshal transaction payload")
continue // FIXME
}
reqIDNumber, ok := big.NewInt(0).SetString(task.RequestID, 10)
if !ok {
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to parse request id number in Dione task")
continue // FIXME
}
err = pcm.ethereumClient.SubmitRequestAnswer(reqIDNumber, task.Payload)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Error("Failed to submit task to ETH chain")
continue // FIXME
}
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Debug("Dione task has been sucessfully submitted to ETH chain (DioneOracle contract)")
}
}
func (pcm *PBFTConsensusManager) commitAcceptedBlocks() (*types3.Block, error) {
blocks := pcm.blockPool.GetAllAcceptedBlocks()
if blocks == nil {
2021-07-11 00:32:58 +00:00
return nil, ErrNoAcceptedBlocks
}
var maxStake *big.Int
var selectedBlock *types3.Block
for _, v := range blocks {
stake, err := pcm.ethereumClient.GetMinerStake(v.Header.ProposerEth)
if err != nil {
return nil, err
2021-06-02 21:19:52 +00:00
}
if maxStake != nil {
if stake.Cmp(maxStake) == -1 {
continue
}
2021-06-02 21:19:52 +00:00
}
maxStake = stake
selectedBlock = v
2021-06-02 21:19:52 +00:00
}
logrus.WithFields(logrus.Fields{
"hash": hex.EncodeToString(selectedBlock.Header.Hash),
"height": selectedBlock.Header.Height,
"miner": selectedBlock.Header.Proposer.String(),
}).Info("Committed new block")
2021-07-11 23:23:00 +00:00
pcm.blockPool.PruneAcceptedBlocks(selectedBlock)
for _, v := range selectedBlock.Data {
err := pcm.mempool.DeleteTx(v.Hash)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"tx": hex.EncodeToString(v.Hash),
}).Errorf("Failed to delete committed tx from mempool")
continue
}
2021-07-11 23:23:00 +00:00
}
return selectedBlock, pcm.blockchain.StoreBlock(selectedBlock)
2021-06-02 21:19:52 +00:00
}