ChronosXYZ/dione
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rewrite consensus part

Browse Source
This commit is contained in:
ChronosX88 2020-11-15 17:46:58 +04:00
parent 78e4ec34a6
commit 15190a5b8a
Signed by: ChronosXYZ
GPG Key ID: 085A69A82C8C511A
13 changed files with 529 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/config.go
View File

@ -16,7 +16,7 @@ type Config struct {
Filecoin FilecoinConfig `mapstructure:"filecoin"`
PubSub PubSubConfig `mapstructure:"pubSub"`
Store StoreConfig `mapstructure:"store"`
ConsensusMaxFaultNodes int `mapstructure:"consensus_max_fault_nodes"`
ConsensusMinApprovals int `mapstructure:"consensus_min_approvals"`
}
type EthereumConfig struct {

77
consensus/commit_pool.go Normal file
View File

@ -0,0 +1,77 @@
package consensus
import (
"encoding/hex"
"github.com/Secured-Finance/dione/models"
"github.com/Secured-Finance/dione/sigs"
"github.com/Secured-Finance/dione/types"
)
type CommitPool struct {
commitMsgs map[string][]*models.Message
}
func NewCommitPool() *CommitPool {
return &CommitPool{
commitMsgs: map[string][]*models.Message{},
}
}
func (cp *CommitPool) CreateCommit(prepareMsg *models.Message, privateKey []byte) (*models.Message, error) {
var message models.Message
message.Type = models.MessageTypeCommit
var consensusMsg models.ConsensusMessage
prepareCMessage := prepareMsg.Payload
consensusMsg.ConsensusID = prepareCMessage.ConsensusID
consensusMsg.RequestID = prepareMsg.Payload.RequestID
consensusMsg.CallbackAddress = prepareMsg.Payload.CallbackAddress
consensusMsg.Data = prepareCMessage.Data
signature, err := sigs.Sign(types.SigTypeEd25519, privateKey, []byte(prepareCMessage.Data))
if err != nil {
return nil, err
}
consensusMsg.Signature = hex.EncodeToString(signature.Data)
message.Payload = consensusMsg
return &message, nil
}
func (cp *CommitPool) IsExistingCommit(commitMsg *models.Message) bool {
consensusMessage := commitMsg.Payload
var exists bool
for _, v := range cp.commitMsgs[consensusMessage.ConsensusID] {
if v.From == commitMsg.From {
exists = true
}
}
return exists
}
func (cp *CommitPool) IsValidCommit(commit *models.Message) bool {
consensusMsg := commit.Payload
buf, err := hex.DecodeString(consensusMsg.Signature)
if err != nil {
return false
}
err = sigs.Verify(&types.Signature{Type: types.SigTypeEd25519, Data: buf}, commit.From, []byte(consensusMsg.Data))
if err != nil {
return false
}
return true
}
func (cp *CommitPool) AddCommit(commit *models.Message) {
consensusID := commit.Payload.ConsensusID
if _, ok := cp.commitMsgs[consensusID]; !ok {
cp.commitMsgs[consensusID] = []*models.Message{}
}
cp.commitMsgs[consensusID] = append(cp.commitMsgs[consensusID], commit)
}
func (cp *CommitPool) CommitSize(consensusID string) int {
if v, ok := cp.commitMsgs[consensusID]; ok {
return len(v)
}
return 0
}

335
consensus/consensus.go
View File

@ -1,133 +1,266 @@
package consensus
import (
"sync"
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/Secured-Finance/dione/ethclient"
"github.com/sirupsen/logrus"
"github.com/Secured-Finance/dione/models"
"github.com/Secured-Finance/dione/pb"
"github.com/sirupsen/logrus"
)
type ConsensusState int
const (
ConsensusPrePrepared ConsensusState = 0x0
ConsensusPrepared ConsensusState = 0x1
ConsensusCommitted ConsensusState = 0x2
testValidData = "test"
)
type PBFTConsensusManager struct {
psb *pb.PubSubRouter
Consensuses map[string]*ConsensusData
maxFaultNodes int
miner *Miner
//Consensuses map[string]*ConsensusData
//maxFaultNodes int
minApprovals int
privKey []byte
prePreparePool *PrePreparePool
preparePool *PreparePool
commitPool *CommitPool
consensusLeaders map[string]bool
ethereumClient *ethclient.EthereumClient
}
type ConsensusData struct {
preparedCount int
commitCount int
State ConsensusState
mutex sync.Mutex
result string
test bool
onConsensusFinishCallback func(finalData string)
}
//type ConsensusData struct {
// preparedCount int
// commitCount int
// mutex sync.Mutex
// result string
// test bool
// onConsensusFinishCallback func(finalData string)
//}
func NewPBFTConsensusManager(psb *pb.PubSubRouter, maxFaultNodes int) *PBFTConsensusManager {
func NewPBFTConsensusManager(psb *pb.PubSubRouter, minApprovals int, privKey []byte, ethereumClient *ethclient.EthereumClient) *PBFTConsensusManager {
pcm := &PBFTConsensusManager{}
pcm.Consensuses = make(map[string]*ConsensusData)
pcm.psb = psb
pcm.psb.Hook("prepared", pcm.handlePreparedMessage)
pcm.psb.Hook("commit", pcm.handleCommitMessage)
pcm.prePreparePool = NewPrePreparePool()
pcm.preparePool = NewPreparePool()
pcm.commitPool = NewCommitPool()
pcm.minApprovals = minApprovals
pcm.privKey = privKey
pcm.ethereumClient = ethereumClient
pcm.consensusLeaders = map[string]bool{}
pcm.psb.Hook(models.MessageTypePrePrepare, pcm.handlePrePrepare)
pcm.psb.Hook(models.MessageTypePrepare, pcm.handlePrepare)
pcm.psb.Hook(models.MessageTypeCommit, pcm.handleCommit)
return pcm
}
func (pcm *PBFTConsensusManager) NewTestConsensus(data string, consensusID string, onConsensusFinishCallback func(finalData string)) {
//consensusID := uuid.New().String()
cData := &ConsensusData{}
cData.test = true
cData.onConsensusFinishCallback = onConsensusFinishCallback
pcm.Consensuses[consensusID] = cData
//func (pcm *PBFTConsensusManager) NewTestConsensus(data string, consensusID string, onConsensusFinishCallback func(finalData string)) {
// //consensusID := uuid.New().String()
// cData := &ConsensusData{}
// cData.test = true
// cData.onConsensusFinishCallback = onConsensusFinishCallback
// pcm.Consensuses[consensusID] = cData
//
// // here we will create DioneTask
//
// msg := models.Message{}
// msg.Type = "prepared"
// msg.Payload = make(map[string]interface{})
// msg.Payload["consensusID"] = consensusID
// msg.Payload["data"] = data
// sign, err := sigs.Sign(types.SigTypeEd25519, pcm.privKey, []byte(data))
// if err != nil {
// logrus.Warnf("failed to sign data: %w", err)
// return
// }
// msg.Payload["signature"] = string(sign.Data)
// pcm.psb.BroadcastToServiceTopic(&msg)
//
// cData.State = ConsensusPrePrepared
// logrus.Debug("started new consensus: " + consensusID)
//}
//
//func (pcm *PBFTConsensusManager) handlePrePrepareMessage(sender peer.ID, message *models.Message) {
// consensusID := message.Payload["consensusID"].(string)
// if _, ok := pcm.Consensuses[consensusID]; !ok {
// logrus.Warn("Unknown consensus ID,: " + consensusID + ", creating consensusInfo")
// pcm.Consensuses[consensusID] = &ConsensusData{
// State: ConsensusPrePrepared,
// onConsensusFinishCallback: func(finalData string) {},
// }
// }
// data := pcm.Consensuses[consensusID]
// logrus.Debug("received pre_prepare msg")
//}
//
//func (pcm *PBFTConsensusManager) handlePreparedMessage(sender peer.ID, message *models.Message) {
// // TODO add check on view of the message
// consensusID := message.Payload["consensusID"].(string)
// if _, ok := pcm.Consensuses[consensusID]; !ok {
// logrus.Warn("Unknown consensus ID,: " + consensusID + ", creating consensusInfo")
// pcm.Consensuses[consensusID] = &ConsensusData{
// State: ConsensusPrePrepared,
// onConsensusFinishCallback: func(finalData string) {},
// }
// }
// data := pcm.Consensuses[consensusID]
// logrus.Debug("received prepared msg")
// //data := pcm.Consensuses[consensusID]
//
// // TODO
// // here we can validate miner which produced this task, is he winner, and so on
// // validation steps:
// // 1. validate sender eligibility to mine (check if it has minimal stake)
// // 2. validate sender wincount
// // 3. validate randomness
// // 4. validate vrf
// // 5. validate payload signature
// // 6. validate transaction (get from rpc client and compare with received)
//
// signStr := message.Payload["signature"].(string)
// signRaw := []byte(signStr)
// err := sigs.Verify(&types.Signature{Data: signRaw, Type: types.SigTypeEd25519}, sender, message.Payload["data"].([]byte))
// if err != nil {
// logrus.Warn("failed to verify data signature")
// return
// }
//
// data.mutex.Lock()
// defer data.mutex.Unlock()
// data.preparedCount++
//
// if data.preparedCount > 2*pcm.maxFaultNodes+1 {
// msg := models.Message{}
// msg.Payload = make(map[string]interface{})
// msg.Type = "commit"
// msg.Payload["consensusID"] = consensusID
// msg.Payload["data"] = message.Payload["data"]
// sign, err := sigs.Sign(types.SigTypeEd25519, pcm.privKey, message.Payload["data"].([]byte))
// if err != nil {
// logrus.Warnf("failed to sign data: %w", err)
// return
// }
// msg.Payload["signature"] = string(sign.Data)
// err = pcm.psb.BroadcastToServiceTopic(&msg)
// if err != nil {
// logrus.Warn("Unable to send COMMIT message: " + err.Error())
// return
// }
// data.State = ConsensusPrepared
// }
//}
//
//func (pcm *PBFTConsensusManager) handleCommitMessage(sender peer.ID, message *models.Message) {
// // TODO add check on view of the message
// // TODO add validation of data by hash to this stage
// consensusID := message.Payload["consensusID"].(string)
// if _, ok := pcm.Consensuses[consensusID]; !ok {
// logrus.Warn("Unknown consensus ID: " + consensusID)
// return
// }
// data := pcm.Consensuses[consensusID]
//
// data.mutex.Lock()
// defer data.mutex.Unlock()
// if data.State == ConsensusCommitted {
// logrus.Debug("consensus already finished, dropping COMMIT message")
// return
// }
//
// logrus.Debug("received commit msg")
//
// signStr := message.Payload["signature"].(string)
// signRaw := []byte(signStr)
// err := sigs.Verify(&types.Signature{Data: signRaw, Type: types.SigTypeEd25519}, sender, message.Payload["data"].([]byte))
// if err != nil {
// logrus.Warn("failed to verify data signature")
// return
// }
//
// data.commitCount++
//
// if data.commitCount > 2*pcm.maxFaultNodes+1 {
// data.State = ConsensusCommitted
// data.result = message.Payload["data"].(string)
// logrus.Debug("consensus successfully finished with result: " + data.result)
// data.onConsensusFinishCallback(data.result)
// }
//}
// here we will create DioneTask
msg := models.Message{}
msg.Type = "prepared"
msg.Payload = make(map[string]interface{})
msg.Payload["consensusID"] = consensusID
msg.Payload["data"] = data
pcm.psb.BroadcastToServiceTopic(&msg)
cData.State = ConsensusPrePrepared
logrus.Debug("started new consensus: " + consensusID)
}
func (pcm *PBFTConsensusManager) handlePreparedMessage(message *models.Message) {
// TODO add check on view of the message
consensusID := message.Payload["consensusID"].(string)
if _, ok := pcm.Consensuses[consensusID]; !ok {
logrus.Warn("Unknown consensus ID: " + consensusID)
return
}
logrus.Debug("received prepared msg")
data := pcm.Consensuses[consensusID]
// TODO
// here we can validate miner which produced this task, is he winner, and so on
// validation steps:
// 1. validate sender eligibility to mine (check if it has minimal stake)
// 2. validate sender wincount
// 3. validate randomness
// 4. validate vrf
// 5. validate payload signature
// 6. validate transaction (get from rpc client and compare with received)
data.mutex.Lock()
data.preparedCount++
data.mutex.Unlock()
if data.preparedCount > 2*pcm.maxFaultNodes+1 {
msg := models.Message{}
msg.Payload = make(map[string]interface{})
msg.Type = "commit"
msg.Payload["consensusID"] = consensusID
msg.Payload["data"] = message.Payload["data"]
err := pcm.psb.BroadcastToServiceTopic(&msg)
func (pcm *PBFTConsensusManager) Propose(consensusID, data string, requestID *big.Int, callbackAddress common.Address) error {
pcm.consensusLeaders[consensusID] = true
reqIDRaw := requestID.String()
callbackAddressHex := callbackAddress.Hex()
prePrepareMsg, err := pcm.prePreparePool.CreatePrePrepare(consensusID, data, reqIDRaw, callbackAddressHex, pcm.privKey)
if err != nil {
logrus.Warn("Unable to send COMMIT message: " + err.Error())
return
}
data.State = ConsensusPrepared
return err
}
pcm.psb.BroadcastToServiceTopic(prePrepareMsg)
return nil
}
func (pcm *PBFTConsensusManager) handleCommitMessage(message *models.Message) {
// TODO add check on view of the message
// TODO add validation of data by hash to this stage
consensusID := message.Payload["consensusID"].(string)
if _, ok := pcm.Consensuses[consensusID]; !ok {
logrus.Warn("Unknown consensus ID: " + consensusID)
func (pcm *PBFTConsensusManager) handlePrePrepare(message *models.Message) {
if pcm.prePreparePool.IsExistingPrePrepare(message) {
logrus.Debug("received existing pre_prepare msg, dropping...")
return
}
data := pcm.Consensuses[consensusID]
data.mutex.Lock()
defer data.mutex.Unlock()
if data.State == ConsensusCommitted {
logrus.Debug("consensus already finished, dropping COMMIT message")
if !pcm.prePreparePool.IsValidPrePrepare(message) {
logrus.Debug("received invalid pre_prepare msg, dropping...")
return
}
logrus.Debug("received commit msg")
pcm.psb.BroadcastToServiceTopic(message)
prepareMsg, err := pcm.preparePool.CreatePrepare(message, pcm.privKey)
if err != nil {
logrus.Errorf("failed to create prepare message: %w", err)
}
pcm.psb.BroadcastToServiceTopic(prepareMsg)
}
data.commitCount++
func (pcm *PBFTConsensusManager) handlePrepare(message *models.Message) {
if pcm.preparePool.IsExistingPrepare(message) {
logrus.Debug("received existing prepare msg, dropping...")
return
}
if !pcm.preparePool.IsValidPrepare(message) {
logrus.Debug("received invalid prepare msg, dropping...")
return
}
if data.commitCount > 2*pcm.maxFaultNodes+1 {
data.State = ConsensusCommitted
data.result = message.Payload["data"].(string)
logrus.Debug("consensus successfully finished with result: " + data.result)
data.onConsensusFinishCallback(data.result)
pcm.preparePool.AddPrepare(message)
pcm.psb.BroadcastToServiceTopic(message)
if pcm.preparePool.PrepareSize(message.Payload.ConsensusID) >= pcm.minApprovals {
commitMsg, err := pcm.commitPool.CreateCommit(message, pcm.privKey)
if err != nil {
logrus.Errorf("failed to create commit message: %w", err)
}
pcm.psb.BroadcastToServiceTopic(commitMsg)
}
}
func (pcm *PBFTConsensusManager) handleCommit(message *models.Message) {
if pcm.commitPool.IsExistingCommit(message) {
logrus.Debug("received existing commit msg, dropping...")
return
}
if !pcm.commitPool.IsValidCommit(message) {
logrus.Debug("received invalid commit msg, dropping...")
return
}
pcm.commitPool.AddCommit(message)
pcm.psb.BroadcastToServiceTopic(message)
consensusMsg := message.Payload
if pcm.commitPool.CommitSize(consensusMsg.ConsensusID) >= pcm.minApprovals {
if pcm.consensusLeaders[consensusMsg.ConsensusID] {
logrus.Infof("Submitting on-chain result for consensus ID: %s", consensusMsg.ConsensusID)
reqID, ok := new(big.Int).SetString(consensusMsg.RequestID, 10)
if !ok {
logrus.Errorf("Failed to parse big int: %v", consensusMsg.RequestID)
}
callbackAddress := common.HexToAddress(consensusMsg.CallbackAddress)
err := pcm.ethereumClient.SubmitRequestAnswer(reqID, consensusMsg.Data, callbackAddress)
if err != nil {
logrus.Errorf("Failed to submit on-chain result: %w", err)
}
}
}
}

70
consensus/pre_prepare_pool.go Normal file
View File

@ -0,0 +1,70 @@
package consensus
import (
"encoding/hex"
"github.com/Secured-Finance/dione/models"
"github.com/Secured-Finance/dione/sigs"
"github.com/Secured-Finance/dione/types"
)
type PrePreparePool struct {
prePrepareMsgs map[string][]*models.Message
}
func NewPrePreparePool() *PrePreparePool {
return &PrePreparePool{
prePrepareMsgs: map[string][]*models.Message{},
}
}
func (pp *PrePreparePool) CreatePrePrepare(consensusID, data string, requestID string, callbackAddress string, privateKey []byte) (*models.Message, error) {
var message models.Message
message.Type = models.MessageTypePrePrepare
var consensusMsg models.ConsensusMessage
consensusMsg.ConsensusID = consensusID
consensusMsg.RequestID = requestID
consensusMsg.CallbackAddress = callbackAddress
consensusMsg.Data = data
signature, err := sigs.Sign(types.SigTypeEd25519, privateKey, []byte(data))
if err != nil {
return nil, err
}
consensusMsg.Signature = hex.EncodeToString(signature.Data)
message.Payload = consensusMsg
return &message, nil
}
func (ppp *PrePreparePool) IsExistingPrePrepare(prepareMsg *models.Message) bool {
consensusMessage := prepareMsg.Payload
var exists bool
for _, v := range ppp.prePrepareMsgs[consensusMessage.ConsensusID] {
if v.From == prepareMsg.From {
exists = true
}
}
return exists
}
func (ppp *PrePreparePool) IsValidPrePrepare(prePrepare *models.Message) bool {
// TODO here we need to do validation of tx itself
consensusMsg := prePrepare.Payload
buf, err := hex.DecodeString(consensusMsg.Signature)
if err != nil {
return false
}
err = sigs.Verify(&types.Signature{Type: types.SigTypeEd25519, Data: buf}, prePrepare.From, []byte(consensusMsg.Data))
if err != nil {
return false
}
return true
}
func (ppp *PrePreparePool) AddPrePrepare(prePrepare *models.Message) {
consensusID := prePrepare.Payload.ConsensusID
if _, ok := ppp.prePrepareMsgs[consensusID]; !ok {
ppp.prePrepareMsgs[consensusID] = []*models.Message{}
}
ppp.prePrepareMsgs[consensusID] = append(ppp.prePrepareMsgs[consensusID], prePrepare)
}

78
consensus/prepare_pool.go Normal file
View File

@ -0,0 +1,78 @@
package consensus
import (
"encoding/hex"
"github.com/Secured-Finance/dione/models"
"github.com/Secured-Finance/dione/sigs"
"github.com/Secured-Finance/dione/types"
)
type PreparePool struct {
prepareMsgs map[string][]*models.Message
privateKey []byte
}
func NewPreparePool() *PreparePool {
return &PreparePool{
prepareMsgs: map[string][]*models.Message{},
}
}
func (pp *PreparePool) CreatePrepare(prePrepareMsg *models.Message, privateKey []byte) (*models.Message, error) {
var message models.Message
message.Type = models.MessageTypePrepare
var consensusMsg models.ConsensusMessage
prepareCMessage := prePrepareMsg.Payload
consensusMsg.ConsensusID = prepareCMessage.ConsensusID
consensusMsg.RequestID = prePrepareMsg.Payload.RequestID
consensusMsg.CallbackAddress = prePrepareMsg.Payload.CallbackAddress
consensusMsg.Data = prepareCMessage.Data
signature, err := sigs.Sign(types.SigTypeEd25519, privateKey, []byte(prepareCMessage.Data))
if err != nil {
return nil, err
}
consensusMsg.Signature = hex.EncodeToString(signature.Data)
message.Payload = consensusMsg
return &message, nil
}
func (pp *PreparePool) IsExistingPrepare(prepareMsg *models.Message) bool {
consensusMessage := prepareMsg.Payload
var exists bool
for _, v := range pp.prepareMsgs[consensusMessage.ConsensusID] {
if v.From == prepareMsg.From {
exists = true
}
}
return exists
}
func (pp *PreparePool) IsValidPrepare(prepare *models.Message) bool {
consensusMsg := prepare.Payload
buf, err := hex.DecodeString(consensusMsg.Signature)
if err != nil {
return false
}
err = sigs.Verify(&types.Signature{Type: types.SigTypeEd25519, Data: buf}, prepare.From, []byte(consensusMsg.Data))
if err != nil {
return false
}
return true
}
func (pp *PreparePool) AddPrepare(prepare *models.Message) {
consensusID := prepare.Payload.ConsensusID
if _, ok := pp.prepareMsgs[consensusID]; !ok {
pp.prepareMsgs[consensusID] = []*models.Message{}
}
pp.prepareMsgs[consensusID] = append(pp.prepareMsgs[consensusID], prepare)
}
func (pp *PreparePool) PrepareSize(consensusID string) int {
if v, ok := pp.prepareMsgs[consensusID]; ok {
return len(v)
}
return 0
}

2
ethclient/ethereum.go
View File

@ -136,7 +136,7 @@ func (c *EthereumClient) SubscribeOnOracleEvents(ctx context.Context) (chan *ora
return resChan, subscription, err
}
func (c *EthereumClient) SubmitRequestAnswer(reqID *big.Int, data string, callbackAddress common.Address, callbackMethodID [4]byte) error {
func (c *EthereumClient) SubmitRequestAnswer(reqID *big.Int, data string, callbackAddress common.Address) error {
// privateKey, err := crypto.HexToECDSA(private_key)
// if err != nil {
// c.Logger.Fatal("Failed to generate private key", err)

1
go.mod
View File

@ -41,6 +41,7 @@ require (
github.com/mattn/go-colorable v0.1.6 // indirect
github.com/mattn/go-sqlite3 v1.9.0
github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
github.com/mitchellh/mapstructure v1.3.3
github.com/multiformats/go-multiaddr v0.3.1
github.com/olekukonko/tablewriter v0.0.4 // indirect
github.com/onsi/ginkgo v1.14.0 // indirect

2
go.sum
View File

@ -743,6 +743,8 @@ github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0Qu
github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/mitchellh/mapstructure v1.3.3 h1:SzB1nHZ2Xi+17FP0zVQBHIZqvwRN9408fJO8h+eeNA8=
github.com/mitchellh/mapstructure v1.3.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=

30
models/message.go
View File

@ -1,7 +1,29 @@
package models
type Message struct {
Type string `json:"type"`
Payload map[string]interface{} `json:"payload"`
From string `json:"-"`
import (
"github.com/libp2p/go-libp2p-core/peer"
)
type MessageType uint8
const (
MessageTypeUnknown = MessageType(iota)
MessageTypePrePrepare
MessageTypePrepare
MessageTypeCommit
)
type ConsensusMessage struct {
ConsensusID string
Signature string
RequestID string
CallbackAddress string
Data string
}
type Message struct {
Type MessageType `json:"type"`
Payload ConsensusMessage `json:"payload"`
From peer.ID `json:"-"`
}

29
node/ethereum.go
View File

@ -27,19 +27,24 @@ func (n *Node) subscribeOnEthContracts(ctx context.Context) {
}
logrus.Info("BlockHash for Solana transaction: ", task.BlockHash)
logrus.Info("Started new consensus round with ID: ", task.BlockHash)
n.ConsensusManager.NewTestConsensus(string(task.BlockHash), task.BlockHash, func(finalData string) {
if finalData != string(task.BlockHash) {
logrus.Warnf("Expected final data to be %s, not %s", task.BlockHash, finalData)
return
//n.ConsensusManager.NewTestConsensus(string(task.BlockHash), task.BlockHash, func(finalData string) {
// if finalData != string(task.BlockHash) {
// logrus.Warnf("Expected final data to be %s, not %s", task.BlockHash, finalData)
// return
// }
// logrus.Info("Consensus ID: ", task.BlockHash, " was successfull")
// logrus.Info("Submitting on-chain result: ", task.BlockHash, "for consensus ID: ", task.BlockHash)
// if task.Miner == n.Host.ID() {
// if err := n.Ethereum.SubmitRequestAnswer(event.RequestID, task.BlockHash, event.CallbackAddress, event.CallbackMethodID); err != nil {
// logrus.Warn("Can't submit request to ethereum chain: ", err)
// }
// }
//})
err = n.ConsensusManager.Propose(task.BlockHash, task.BlockHash, event.RequestID, event.CallbackAddress)
if err != nil {
logrus.Errorf("Failed to propose task: %w", err)
}
logrus.Info("Consensus ID: ", task.BlockHash, " was successfull")
logrus.Info("Submitting on-chain result: ", task.BlockHash, "for consensus ID: ", task.BlockHash)
if task.Miner == n.Host.ID() {
if err := n.Ethereum.SubmitRequestAnswer(event.RequestID, task.BlockHash, event.CallbackAddress, event.CallbackMethodID); err != nil {
logrus.Warn("Can't submit request to ethereum chain: ", err)
}
}
})
}
case <-ctx.Done():
break EventLoop

14
node/node.go
View File

@ -74,7 +74,10 @@ func (n *Node) setupNode(ctx context.Context, prvKey crypto.PrivKey, pexDiscover
}
n.setupSolanaClient()
n.setupPubsub()
n.setupConsensusManager(n.Config.ConsensusMaxFaultNodes)
err = n.setupConsensusManager(prvKey, n.Config.ConsensusMinApprovals)
if err != nil {
logrus.Fatalf("Failed to setup consensus manager: %w", err)
}
err = n.setupBeacon()
if err != nil {
logrus.Fatal(err)
@ -153,8 +156,13 @@ func (n *Node) setupPubsub() {
//time.Sleep(3 * time.Second)
}
func (n *Node) setupConsensusManager(maxFaultNodes int) {
n.ConsensusManager = consensus.NewPBFTConsensusManager(n.PubSubRouter, maxFaultNodes)
func (n *Node) setupConsensusManager(privateKey crypto.PrivKey, minApprovals int) error {
pkeyRaw, err := privateKey.Raw()
if err != nil {
return err
}
n.ConsensusManager = consensus.NewPBFTConsensusManager(n.PubSubRouter, minApprovals, pkeyRaw, n.Ethereum)
return nil
}
func (n *Node) setupLibp2pHost(ctx context.Context, privateKey crypto.PrivKey, pexDiscoveryUpdateTime time.Duration) {

2
node/node_test.go
View File

@ -27,7 +27,7 @@ func TestConsensus(t *testing.T) {
PubSub: config.PubSubConfig{
ProtocolID: "/dione/1.0",
},
ConsensusMaxFaultNodes: 3,
ConsensusMinApprovals: 3,
}
var nodes []*Node

10
pb/pubsub_router.go
View File

@ -16,7 +16,7 @@ type PubSubRouter struct {
Pubsub *pubsub.PubSub
context context.Context
contextCancel context.CancelFunc
handlers map[string][]Handler
handlers map[models.MessageType][]Handler
oracleTopic string
}
@ -27,7 +27,7 @@ func NewPubSubRouter(h host.Host, oracleTopic string) *PubSubRouter {
node: h,
context: ctx,
contextCancel: ctxCancel,
handlers: make(map[string][]Handler),
handlers: make(map[models.MessageType][]Handler),
}
pb, err := pubsub.NewFloodSub(
@ -83,10 +83,10 @@ func (psr *PubSubRouter) handleMessage(p *pubsub.Message) {
logrus.Warn("Unable to decode message data! " + err.Error())
return
}
message.From = senderPeerID.String()
message.From = senderPeerID
handlers, ok := psr.handlers[message.Type]
if !ok {
logrus.Warn("Dropping message " + message.Type + " because we don't have any handlers!")
logrus.Warn("Dropping message " + string(message.Type) + " because we don't have any handlers!")
return
}
for _, v := range handlers {
@ -94,7 +94,7 @@ func (psr *PubSubRouter) handleMessage(p *pubsub.Message) {
}
}
func (psr *PubSubRouter) Hook(messageType string, handler Handler) {
func (psr *PubSubRouter) Hook(messageType models.MessageType, handler Handler) {
handlers, ok := psr.handlers[messageType]
if !ok {
emptyArray := []Handler{}