Refactor dispute manager to new blockchain architecture
This commit is contained in:
parent
956de703e8
commit
5aadbab600
@ -2,8 +2,19 @@ package consensus
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/hex"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
types2 "github.com/Secured-Finance/dione/blockchain/types"
|
||||||
|
|
||||||
|
"github.com/Secured-Finance/dione/types"
|
||||||
|
"github.com/fxamacker/cbor/v2"
|
||||||
|
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
"golang.org/x/crypto/sha3"
|
||||||
|
|
||||||
|
"github.com/Secured-Finance/dione/blockchain"
|
||||||
|
|
||||||
"github.com/Secured-Finance/dione/contracts/dioneDispute"
|
"github.com/Secured-Finance/dione/contracts/dioneDispute"
|
||||||
"github.com/Secured-Finance/dione/contracts/dioneOracle"
|
"github.com/Secured-Finance/dione/contracts/dioneOracle"
|
||||||
"github.com/Secured-Finance/dione/ethclient"
|
"github.com/Secured-Finance/dione/ethclient"
|
||||||
@ -16,9 +27,10 @@ type DisputeManager struct {
|
|||||||
submissionMap map[string]*dioneOracle.DioneOracleSubmittedOracleRequest
|
submissionMap map[string]*dioneOracle.DioneOracleSubmittedOracleRequest
|
||||||
disputeMap map[string]*dioneDispute.DioneDisputeNewDispute
|
disputeMap map[string]*dioneDispute.DioneDisputeNewDispute
|
||||||
voteWindow time.Duration
|
voteWindow time.Duration
|
||||||
|
blockchain *blockchain.BlockChain
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient, pcm *PBFTConsensusManager, voteWindow int) (*DisputeManager, error) {
|
func NewDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient, pcm *PBFTConsensusManager, voteWindow int, bc *blockchain.BlockChain) (*DisputeManager, error) {
|
||||||
newSubmittionsChan, submSubscription, err := ethClient.SubscribeOnNewSubmittions(ctx)
|
newSubmittionsChan, submSubscription, err := ethClient.SubscribeOnNewSubmittions(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@ -36,6 +48,7 @@ func NewDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient,
|
|||||||
submissionMap: map[string]*dioneOracle.DioneOracleSubmittedOracleRequest{},
|
submissionMap: map[string]*dioneOracle.DioneOracleSubmittedOracleRequest{},
|
||||||
disputeMap: map[string]*dioneDispute.DioneDisputeNewDispute{},
|
disputeMap: map[string]*dioneDispute.DioneDisputeNewDispute{},
|
||||||
voteWindow: time.Duration(voteWindow) * time.Second,
|
voteWindow: time.Duration(voteWindow) * time.Second,
|
||||||
|
blockchain: bc,
|
||||||
}
|
}
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
@ -62,95 +75,117 @@ func NewDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient,
|
|||||||
return dm, nil
|
return dm, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (dm *DisputeManager) onNewSubmission(submittion *dioneOracle.DioneOracleSubmittedOracleRequest) {
|
func (dm *DisputeManager) onNewSubmission(submission *dioneOracle.DioneOracleSubmittedOracleRequest) {
|
||||||
//c := dm.pcm.GetConsensusInfo(submittion.ReqID.String())
|
// find a block that contains the dione task with specified request id
|
||||||
//if c == nil {
|
task, block, err := dm.findTaskAndBlockWithRequestID(submission.ReqID.String())
|
||||||
// // todo: warn
|
if err != nil {
|
||||||
// return
|
logrus.Error(err)
|
||||||
//}
|
return
|
||||||
//
|
}
|
||||||
//dm.submissionMap[submittion.ReqID.String()] = submittion
|
|
||||||
//
|
dm.submissionMap[submission.ReqID.String()] = submission
|
||||||
//submHashBytes := sha3.Sum256(submittion.Data)
|
|
||||||
//localHashBytes := sha3.Sum256(c.Task.Payload)
|
submHashBytes := sha3.Sum256(submission.Data)
|
||||||
//submHash := hex.EncodeToString(submHashBytes[:])
|
localHashBytes := sha3.Sum256(task.Payload)
|
||||||
//localHash := hex.EncodeToString(localHashBytes[:])
|
submHash := hex.EncodeToString(submHashBytes[:])
|
||||||
//if submHash != localHash {
|
localHash := hex.EncodeToString(localHashBytes[:])
|
||||||
// logrus.Debugf("submission of request id %s isn't valid - beginning dispute", c.Task.RequestID)
|
if submHash != localHash {
|
||||||
// addr := common.HexToAddress(c.Task.MinerEth)
|
logrus.Debugf("submission of request id %s isn't valid - beginning dispute", submission.ReqID)
|
||||||
// reqID, ok := big.NewInt(0).SetString(c.Task.RequestID, 10)
|
err := dm.ethClient.BeginDispute(block.Header.ProposerEth, submission.ReqID)
|
||||||
// if !ok {
|
if err != nil {
|
||||||
// logrus.Errorf("cannot parse request id: %s", c.Task.RequestID)
|
logrus.Errorf(err.Error())
|
||||||
// return
|
return
|
||||||
// }
|
}
|
||||||
// err := dm.ethClient.BeginDispute(addr, reqID)
|
disputeFinishTimer := time.NewTimer(dm.voteWindow)
|
||||||
// if err != nil {
|
go func() {
|
||||||
// logrus.Errorf(err.Error())
|
for {
|
||||||
// return
|
select {
|
||||||
// }
|
case <-dm.ctx.Done():
|
||||||
// disputeFinishTimer := time.NewTimer(dm.voteWindow)
|
return
|
||||||
// go func() {
|
case <-disputeFinishTimer.C:
|
||||||
// for {
|
{
|
||||||
// select {
|
d, ok := dm.disputeMap[submission.ReqID.String()]
|
||||||
// case <-dm.ctx.Done():
|
if !ok {
|
||||||
// return
|
logrus.Error("cannot finish dispute: it doesn't exist in manager's dispute map!")
|
||||||
// case <-disputeFinishTimer.C:
|
return
|
||||||
// {
|
}
|
||||||
// d, ok := dm.disputeMap[reqID.String()]
|
err := dm.ethClient.FinishDispute(d.Dhash)
|
||||||
// if !ok {
|
if err != nil {
|
||||||
// logrus.Error("cannot finish dispute: it doesn't exist in manager's dispute map!")
|
logrus.Errorf(err.Error())
|
||||||
// return
|
return
|
||||||
// }
|
}
|
||||||
// err := dm.ethClient.FinishDispute(d.Dhash)
|
disputeFinishTimer.Stop()
|
||||||
// if err != nil {
|
return
|
||||||
// logrus.Errorf(err.Error())
|
}
|
||||||
// return
|
}
|
||||||
// }
|
}
|
||||||
// disputeFinishTimer.Stop()
|
}()
|
||||||
// return
|
}
|
||||||
// }
|
}
|
||||||
// }
|
|
||||||
// }
|
func (dm *DisputeManager) findTaskAndBlockWithRequestID(requestID string) (*types.DioneTask, *types2.Block, error) {
|
||||||
// }()
|
height, err := dm.blockchain.GetLatestBlockHeight()
|
||||||
//}
|
if err != nil {
|
||||||
// TODO refactor due to new architecture with blockchain
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
for {
|
||||||
|
block, err := dm.blockchain.FetchBlockByHeight(height)
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, v := range block.Data {
|
||||||
|
var task types.DioneTask
|
||||||
|
err := cbor.Unmarshal(v.Data, &task)
|
||||||
|
if err != nil {
|
||||||
|
logrus.Error(err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if task.RequestID == requestID {
|
||||||
|
return &task, block, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
height--
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (dm *DisputeManager) onNewDispute(dispute *dioneDispute.DioneDisputeNewDispute) {
|
func (dm *DisputeManager) onNewDispute(dispute *dioneDispute.DioneDisputeNewDispute) {
|
||||||
//c := dm.pcm.GetConsensusInfo(dispute.RequestID.String())
|
task, _, err := dm.findTaskAndBlockWithRequestID(dispute.RequestID.String())
|
||||||
//if c == nil {
|
if err != nil {
|
||||||
// // todo: warn
|
logrus.Error(err)
|
||||||
// return
|
return
|
||||||
//}
|
}
|
||||||
//
|
|
||||||
//subm, ok := dm.submissionMap[dispute.RequestID.String()]
|
subm, ok := dm.submissionMap[dispute.RequestID.String()]
|
||||||
//if !ok {
|
if !ok {
|
||||||
// // todo: warn
|
logrus.Warn("desired submission isn't found in map")
|
||||||
// return
|
return
|
||||||
//}
|
}
|
||||||
//
|
|
||||||
//dm.disputeMap[dispute.RequestID.String()] = dispute
|
dm.disputeMap[dispute.RequestID.String()] = dispute
|
||||||
//
|
|
||||||
//if dispute.DisputeInitiator.Hex() == dm.ethClient.GetEthAddress().Hex() {
|
if dispute.DisputeInitiator.Hex() == dm.ethClient.GetEthAddress().Hex() {
|
||||||
// return
|
return
|
||||||
//}
|
}
|
||||||
//
|
|
||||||
//submHashBytes := sha3.Sum256(subm.Data)
|
submHashBytes := sha3.Sum256(subm.Data)
|
||||||
//localHashBytes := sha3.Sum256(c.Task.Payload)
|
localHashBytes := sha3.Sum256(task.Payload)
|
||||||
//submHash := hex.EncodeToString(submHashBytes[:])
|
submHash := hex.EncodeToString(submHashBytes[:])
|
||||||
//localHash := hex.EncodeToString(localHashBytes[:])
|
localHash := hex.EncodeToString(localHashBytes[:])
|
||||||
//if submHash == localHash {
|
if submHash == localHash {
|
||||||
// err := dm.ethClient.VoteDispute(dispute.Dhash, false)
|
err := dm.ethClient.VoteDispute(dispute.Dhash, false)
|
||||||
// if err != nil {
|
if err != nil {
|
||||||
// logrus.Errorf(err.Error())
|
logrus.Errorf(err.Error())
|
||||||
// return
|
return
|
||||||
// }
|
}
|
||||||
//}
|
}
|
||||||
//
|
|
||||||
//err := dm.ethClient.VoteDispute(dispute.Dhash, true)
|
err = dm.ethClient.VoteDispute(dispute.Dhash, true)
|
||||||
//if err != nil {
|
if err != nil {
|
||||||
// logrus.Errorf(err.Error())
|
logrus.Errorf(err.Error())
|
||||||
// return
|
return
|
||||||
//}
|
}
|
||||||
// TODO refactor due to new architecture with blockchain
|
|
||||||
}
|
}
|
||||||
|
13
node/node.go
13
node/node.go
@ -9,6 +9,8 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/multiformats/go-multiaddr"
|
||||||
|
|
||||||
"github.com/asaskevich/EventBus"
|
"github.com/asaskevich/EventBus"
|
||||||
|
|
||||||
"github.com/fxamacker/cbor/v2"
|
"github.com/fxamacker/cbor/v2"
|
||||||
@ -160,7 +162,14 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
|
|||||||
r := provideP2PRPCClient(lhost)
|
r := provideP2PRPCClient(lhost)
|
||||||
|
|
||||||
// initialize sync manager
|
// initialize sync manager
|
||||||
sm, err := provideSyncManager(bus, bc, mp, r, baddrs[0], psb) // FIXME here we just pick up first bootstrap in list
|
|
||||||
|
var baddr multiaddr.Multiaddr
|
||||||
|
if len(baddrs) == 0 {
|
||||||
|
baddr = nil
|
||||||
|
} else {
|
||||||
|
baddr = baddrs[0]
|
||||||
|
}
|
||||||
|
sm, err := provideSyncManager(bus, bc, mp, r, baddr, psb) // FIXME here we just pick up first bootstrap in list
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logrus.Fatal(err)
|
logrus.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -186,7 +195,7 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
|
|||||||
logrus.Info("Random beacon subsystem has been initialized!")
|
logrus.Info("Random beacon subsystem has been initialized!")
|
||||||
|
|
||||||
// initialize dispute subsystem
|
// initialize dispute subsystem
|
||||||
disputeManager, err := provideDisputeManager(context.TODO(), ethClient, consensusManager, config)
|
disputeManager, err := provideDisputeManager(context.TODO(), ethClient, consensusManager, config, bc)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logrus.Fatal(err)
|
logrus.Fatal(err)
|
||||||
}
|
}
|
||||||
|
@ -56,8 +56,8 @@ func provideCache(config *config.Config) cache.Cache {
|
|||||||
return backend
|
return backend
|
||||||
}
|
}
|
||||||
|
|
||||||
func provideDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient, pcm *consensus.PBFTConsensusManager, cfg *config.Config) (*consensus.DisputeManager, error) {
|
func provideDisputeManager(ctx context.Context, ethClient *ethclient.EthereumClient, pcm *consensus.PBFTConsensusManager, cfg *config.Config, bc *blockchain.BlockChain) (*consensus.DisputeManager, error) {
|
||||||
return consensus.NewDisputeManager(ctx, ethClient, pcm, cfg.Ethereum.DisputeVoteWindow)
|
return consensus.NewDisputeManager(ctx, ethClient, pcm, cfg.Ethereum.DisputeVoteWindow, bc)
|
||||||
}
|
}
|
||||||
|
|
||||||
func provideMiner(peerID peer.ID, ethAddress common.Address, ethClient *ethclient.EthereumClient, privateKey crypto.PrivKey, mempool *pool.Mempool) *consensus.Miner {
|
func provideMiner(peerID peer.ID, ethAddress common.Address, ethClient *ethclient.EthereumClient, privateKey crypto.PrivKey, mempool *pool.Mempool) *consensus.Miner {
|
||||||
@ -165,11 +165,16 @@ func provideMemPool() (*pool.Mempool, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func provideSyncManager(bus EventBus.Bus, bp *blockchain.BlockChain, mp *pool.Mempool, r *gorpc.Client, bootstrap multiaddr.Multiaddr, psb *pubsub.PubSubRouter) (sync.SyncManager, error) {
|
func provideSyncManager(bus EventBus.Bus, bp *blockchain.BlockChain, mp *pool.Mempool, r *gorpc.Client, bootstrap multiaddr.Multiaddr, psb *pubsub.PubSubRouter) (sync.SyncManager, error) {
|
||||||
addr, err := peer.AddrInfoFromP2pAddr(bootstrap)
|
bootstrapPeerID := peer.ID("")
|
||||||
if err != nil {
|
if bootstrap != nil {
|
||||||
return nil, err
|
addr, err := peer.AddrInfoFromP2pAddr(bootstrap)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
bootstrapPeerID = addr.ID
|
||||||
}
|
}
|
||||||
return sync.NewSyncManager(bus, bp, mp, r, addr.ID, psb), nil
|
|
||||||
|
return sync.NewSyncManager(bus, bp, mp, r, bootstrapPeerID, psb), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func provideP2PRPCClient(h host.Host) *gorpc.Client {
|
func provideP2PRPCClient(h host.Host) *gorpc.Client {
|
||||||
|
Loading…
Reference in New Issue
Block a user