core: run first DKG at startup. (#129)

9 files changed, 239 insertions, 40 deletions

diff --git a/core/consensus.go b/core/consensus.go
index 7700296..dc5bbba 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -111,26 +111,87 @@ func (recv *consensusReceiver) ConfirmBlock(hash common.Hash) {
 	recv.restart <- struct{}{}
 }
 
+// consensusDKGReceiver implements dkgReceiver.
+type consensusDKGReceiver struct {
+	ID      types.NodeID
+	gov     Governance
+	prvKey  crypto.PrivateKey
+	network Network
+}
+
+// ProposeDKGComplaint proposes a DKGComplaint.
+func (recv *consensusDKGReceiver) ProposeDKGComplaint(
+	complaint *types.DKGComplaint) {
+	var err error
+	complaint.Signature, err = recv.prvKey.Sign(hashDKGComplaint(complaint))
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	recv.gov.AddDKGComplaint(complaint)
+}
+
+// ProposeDKGMasterPublicKey propose a DKGMasterPublicKey.
+func (recv *consensusDKGReceiver) ProposeDKGMasterPublicKey(
+	mpk *types.DKGMasterPublicKey) {
+	var err error
+	mpk.Signature, err = recv.prvKey.Sign(hashDKGMasterPublicKey(mpk))
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	recv.gov.AddDKGMasterPublicKey(mpk)
+}
+
+// ProposeDKGPrivateShare propose a DKGPrivateShare.
+func (recv *consensusDKGReceiver) ProposeDKGPrivateShare(
+	prv *types.DKGPrivateShare) {
+	var err error
+	prv.Signature, err = recv.prvKey.Sign(hashDKGPrivateShare(prv))
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	recv.network.SendDKGPrivateShare(prv.ReceiverID, prv)
+}
+
+// ProposeDKGAntiNackComplaint propose a DKGPrivateShare as an anti complaint.
+func (recv *consensusDKGReceiver) ProposeDKGAntiNackComplaint(
+	prv *types.DKGPrivateShare) {
+	if prv.ProposerID == recv.ID {
+		var err error
+		prv.Signature, err = recv.prvKey.Sign(hashDKGPrivateShare(prv))
+		if err != nil {
+			log.Println(err)
+			return
+		}
+	}
+	recv.network.BroadcastDKGPrivateShare(prv)
+}
+
 // Consensus implements DEXON Consensus algorithm.
 type Consensus struct {
-	ID        types.NodeID
-	app       Application
-	gov       Governance
-	config    *types.Config
-	baModules []*agreement
-	receivers []*consensusReceiver
-	rbModule  *reliableBroadcast
-	toModule  *totalOrdering
-	ctModule  *consensusTimestamp
-	ccModule  *compactionChain
-	db        blockdb.BlockDatabase
-	network   Network
-	tickerObj Ticker
-	prvKey    crypto.PrivateKey
-	sigToPub  SigToPubFn
-	lock      sync.RWMutex
-	ctx       context.Context
-	ctxCancel context.CancelFunc
+	ID         types.NodeID
+	app        Application
+	gov        Governance
+	config     *types.Config
+	baModules  []*agreement
+	receivers  []*consensusReceiver
+	rbModule   *reliableBroadcast
+	toModule   *totalOrdering
+	ctModule   *consensusTimestamp
+	ccModule   *compactionChain
+	db         blockdb.BlockDatabase
+	network    Network
+	tickerObj  Ticker
+	prvKey     crypto.PrivateKey
+	dkgRunning int32
+	dkgReady   *sync.Cond
+	dkgModule  *dkgProtocol
+	sigToPub   SigToPubFn
+	lock       sync.RWMutex
+	ctx        context.Context
+	ctxCancel  context.CancelFunc
 }
 
 // NewConsensus construct an Consensus instance.
@@ -144,6 +205,7 @@ func NewConsensus(
 
 	config := gov.GetConfiguration(0)
 	notarySet := gov.GetNotarySet()
+	ID := types.NewNodeID(prv.PublicKey())
 
 	// Setup acking by information returned from Governace.
 	rb := newReliableBroadcast()
@@ -164,8 +226,21 @@ func NewConsensus(
 		uint64(float32(len(notarySet)-1)*config.PhiRatio+1),
 		config.NumChains)
 
+	// Setup DKG Protocol.
+	dkgModule := newDKGProtocol(
+		ID,
+		&consensusDKGReceiver{
+			ID:      ID,
+			gov:     gov,
+			prvKey:  prv,
+			network: network,
+		},
+		0,
+		len(gov.GetNotarySet())/3,
+		sigToPub)
+
 	con := &Consensus{
-		ID:        types.NewNodeID(prv.PublicKey()),
+		ID:        ID,
 		rbModule:  rb,
 		toModule:  to,
 		ctModule:  newConsensusTimestamp(),
@@ -175,8 +250,10 @@ func NewConsensus(
 		config:    config,
 		db:        db,
 		network:   network,
-		tickerObj: newTicker(gov),
+		tickerObj: newTicker(gov, TickerBA),
 		prvKey:    prv,
+		dkgReady:  sync.NewCond(&sync.Mutex{}),
+		dkgModule: dkgModule,
 		sigToPub:  sigToPub,
 		ctx:       ctx,
 		ctxCancel: ctxCancel,
@@ -210,13 +287,19 @@ func NewConsensus(
 
 // Run starts running DEXON Consensus.
 func (con *Consensus) Run() {
+	go con.processMsg(con.network.ReceiveChan(), con.PreProcessBlock)
+	con.runDKG()
+	con.dkgReady.L.Lock()
+	defer con.dkgReady.L.Unlock()
+	for con.dkgRunning != 2 {
+		con.dkgReady.Wait()
+	}
 	ticks := make([]chan struct{}, 0, con.config.NumChains)
 	for i := uint32(0); i < con.config.NumChains; i++ {
 		tick := make(chan struct{})
 		ticks = append(ticks, tick)
 		go con.runBA(i, tick)
 	}
-	go con.processMsg(con.network.ReceiveChan(), con.PreProcessBlock)
 	go con.processWitnessData()
 
 	// Reset ticker.
@@ -271,6 +354,58 @@ BALoop:
 	}
 }
 
+// runDKG starts running DKG protocol.
+func (con *Consensus) runDKG() {
+	con.dkgReady.L.Lock()
+	defer con.dkgReady.L.Unlock()
+	if con.dkgRunning != 0 {
+		return
+	}
+	con.dkgRunning = 1
+	go func() {
+		defer func() {
+			con.dkgReady.L.Lock()
+			defer con.dkgReady.L.Unlock()
+			con.dkgReady.Broadcast()
+			con.dkgRunning = 2
+		}()
+		ticker := newTicker(con.gov, TickerDKG)
+		round := con.dkgModule.round
+		<-ticker.Tick()
+		// Phase 2(T = 0): Exchange DKG secret key share.
+		con.dkgModule.processMasterPublicKeys(con.gov.DKGMasterPublicKeys(round))
+		// Phase 3(T = 0~λ): Propose complaint.
+		// Propose complaint is done in `processMasterPublicKeys`.
+		<-ticker.Tick()
+		// Phase 4(T = λ): Propose nack complaints.
+		con.dkgModule.proposeNackComplaints()
+		<-ticker.Tick()
+		// Phase 5(T = 2λ): Propose Anti nack complaint.
+		con.dkgModule.processNackComplaints(con.gov.DKGComplaints(round))
+		<-ticker.Tick()
+		// Phase 6(T = 3λ): Rebroadcast anti nack complaint.
+		// Rebroadcast is done in `processPrivateShare`.
+		<-ticker.Tick()
+		// Phase 7(T = 4λ): Enforce complaints and nack complaints.
+		con.dkgModule.enforceNackComplaints(con.gov.DKGComplaints(round))
+		// Enforce complaint is done in `processPrivateShare`.
+		// Phase 8(T = 5λ): DKG is ready.
+		gpk, err := newDKGGroupPublicKey(round,
+			con.gov.DKGMasterPublicKeys(round),
+			con.gov.DKGComplaints(round),
+			con.dkgModule.threshold, con.sigToPub)
+		if err != nil {
+			panic(err)
+		}
+		qualifies := ""
+		for _, nID := range gpk.qualifyNodeIDs {
+			qualifies += fmt.Sprintf("%s ", nID.String()[:6])
+		}
+		log.Printf("[%s] Qualify Nodes(%d): %s\n",
+			con.ID, len(gpk.qualifyIDs), qualifies)
+	}()
+}
+
 // RunLegacy starts running Legacy DEXON Consensus.
 func (con *Consensus) RunLegacy() {
 	go con.processMsg(con.network.ReceiveChan(), con.processBlock)
@@ -356,6 +491,13 @@ func (con *Consensus) processMsg(
 			if err := con.ProcessVote(val); err != nil {
 				log.Println(err)
 			}
+		case *types.DKGPrivateShare:
+			if con.dkgRunning == 0 {
+				break
+			}
+			if err := con.dkgModule.processPrivateShare(val); err != nil {
+				log.Println(err)
+			}
 		}
 	}
 }
diff --git a/core/consensus_test.go b/core/consensus_test.go
index 701ee00..c5ef452 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -50,6 +50,11 @@ func (n *network) SendDKGPrivateShare(
 	recv types.NodeID, prvShare *types.DKGPrivateShare) {
 }
 
+// BroadcastDKGPrivateShare broadcasts PrivateShare to all DKG participants.
+func (n *network) BroadcastDKGPrivateShare(
+	prvShare *types.DKGPrivateShare) {
+}
+
 // ReceiveChan returns a channel to receive messages from DEXON network.
 func (n *network) ReceiveChan() <-chan interface{} {
 	return make(chan interface{})
diff --git a/core/dkg-tsig-protocol.go b/core/dkg-tsig-protocol.go
index dc4e630..ccd2439 100644
--- a/core/dkg-tsig-protocol.go
+++ b/core/dkg-tsig-protocol.go
@@ -83,6 +83,7 @@ type dkgShareSecret struct {
 type dkgGroupPublicKey struct {
 	round          uint64
 	qualifyIDs     dkg.IDs
+	qualifyNodeIDs types.NodeIDs
 	idMap          map[types.NodeID]dkg.ID
 	publicKeys     map[types.NodeID]*dkg.PublicKey
 	groupPublicKey *dkg.PublicKey
@@ -282,6 +283,7 @@ func (d *dkgProtocol) processPrivateShare(
 		if _, exist := d.antiComplaintReceived[prvShare.ReceiverID]; !exist {
 			d.antiComplaintReceived[prvShare.ReceiverID] =
 				make(map[types.NodeID]struct{})
+			d.recv.ProposeDKGAntiNackComplaint(prvShare)
 		}
 		d.antiComplaintReceived[prvShare.ReceiverID][prvShare.ProposerID] =
 			struct{}{}
@@ -330,6 +332,7 @@ func newDKGGroupPublicKey(
 		}
 	}
 	qualifyIDs := make(dkg.IDs, 0, len(mpks)-len(disqualifyIDs))
+	qualifyNodeIDs := make(types.NodeIDs, 0, cap(qualifyIDs))
 	mpkMap := make(map[dkg.ID]*types.DKGMasterPublicKey, cap(qualifyIDs))
 	idMap := make(map[types.NodeID]dkg.ID)
 	for _, mpk := range mpks {
@@ -339,6 +342,7 @@ func newDKGGroupPublicKey(
 		mpkMap[mpk.DKGID] = mpk
 		idMap[mpk.ProposerID] = mpk.DKGID
 		qualifyIDs = append(qualifyIDs, mpk.DKGID)
+		qualifyNodeIDs = append(qualifyNodeIDs, mpk.ProposerID)
 	}
 	// Recover qualify members' public key.
 	pubKeys := make(map[types.NodeID]*dkg.PublicKey, len(qualifyIDs))
@@ -368,6 +372,7 @@ func newDKGGroupPublicKey(
 	return &dkgGroupPublicKey{
 		round:          round,
 		qualifyIDs:     qualifyIDs,
+		qualifyNodeIDs: qualifyNodeIDs,
 		idMap:          idMap,
 		publicKeys:     pubKeys,
 		threshold:      threshold,
diff --git a/core/dkg-tsig-protocol_test.go b/core/dkg-tsig-protocol_test.go
index 2bcbe9e..63b47ec 100644
--- a/core/dkg-tsig-protocol_test.go
+++ b/core/dkg-tsig-protocol_test.go
@@ -193,6 +193,13 @@ func (s *DKGTSIGProtocolTestSuite) TestDKGTSIGProtocol() {
 		qualifyIDs[id] = struct{}{}
 	}
 
+	for _, nID := range gpk.qualifyNodeIDs {
+		id, exist := gpk.idMap[nID]
+		s.Require().True(exist)
+		_, exist = qualifyIDs[id]
+		s.Require().True(exist)
+	}
+
 	shareSecrets := make(
 		map[types.NodeID]*dkgShareSecret, len(qualifyIDs))
 
@@ -299,21 +306,26 @@ func (s *DKGTSIGProtocolTestSuite) TestComplaint() {
 		ReceiverID: targetID,
 		Round:      round,
 	})
-	s.Error(ErrNotDKGParticipant, err)
-	err = protocol.processPrivateShare(&types.DKGPrivateShare{
+	s.Equal(ErrNotDKGParticipant, err)
+	receivers[byzantineID].ProposeDKGPrivateShare(&types.DKGPrivateShare{
 		ProposerID: byzantineID,
 		ReceiverID: targetID,
 		Round:      round,
 	})
-	s.Error(ErrIncorrectPrivateShareSignature, err)
+	invalidShare := receivers[byzantineID].prvShare[targetID]
+	invalidShare.Signature[1]++
+	err = protocol.processPrivateShare(invalidShare)
+	s.Equal(ErrIncorrectPrivateShareSignature, err)
+	delete(receivers[byzantineID].prvShare, targetID)
 
 	// Byzantine node is sending incorrect private share.
 	receivers[byzantineID].ProposeDKGPrivateShare(&types.DKGPrivateShare{
-		ProposerID: byzantineID,
-		ReceiverID: targetID,
-		Round:      round,
+		ProposerID:   byzantineID,
+		ReceiverID:   targetID,
+		Round:        round,
+		PrivateShare: *dkg.NewPrivateKey(),
 	})
-	invalidShare := receivers[byzantineID].prvShare[targetID]
+	invalidShare = receivers[byzantineID].prvShare[targetID]
 	s.Require().NoError(protocol.processPrivateShare(invalidShare))
 	s.Require().Len(receiver.complaints, 1)
 	complaint, exist := receiver.complaints[byzantineID]
@@ -575,16 +587,17 @@ func (s *DKGTSIGProtocolTestSuite) TestPartialSignature() {
 			PartialSignature: shareSecret.sign(msgHash),
 		}
 		if nID == byzantineID2 {
-			psig.PartialSignature[0]++
+			psig.PartialSignature = shareSecret.sign(
+				crypto.Keccak256Hash([]byte("💣")))
 		}
 		var err error
 		psig.Signature, err = s.prvKeys[nID].Sign(hashDKGPartialSignature(psig))
 		s.Require().NoError(err)
 		err = tsig.processPartialSignature(msgHash, psig)
 		if nID == byzantineID {
-			s.Require().Error(ErrNotQualifyDKGParticipant, err)
+			s.Require().Equal(ErrNotQualifyDKGParticipant, err)
 		} else if nID == byzantineID2 {
-			s.Require().Error(ErrIncorrectPartialSignature, err)
+			s.Require().Equal(ErrIncorrectPartialSignature, err)
 		} else {
 			s.Require().NoError(err)
 		}
diff --git a/core/interfaces.go b/core/interfaces.go
index 8ecfb3c..36b0160 100644
--- a/core/interfaces.go
+++ b/core/interfaces.go
@@ -69,6 +69,9 @@ type Network interface {
 	// SendDKGPrivateShare sends PrivateShare to a DKG participant.
 	SendDKGPrivateShare(recv types.NodeID, prvShare *types.DKGPrivateShare)
 
+	// BroadcastDKGPrivateShare broadcasts PrivateShare to all DKG participants.
+	BroadcastDKGPrivateShare(prvShare *types.DKGPrivateShare)
+
 	// ReceiveChan returns a channel to receive messages from DEXON network.
 	ReceiveChan() <-chan interface{}
 }
diff --git a/core/test/governance.go b/core/test/governance.go
index 63462d0..e486afe 100644
--- a/core/test/governance.go
+++ b/core/test/governance.go
@@ -34,7 +34,8 @@ var (
 
 // Governance is an implementation of Goverance for testing purpose.
 type Governance struct {
-	lambda             time.Duration
+	lambdaBA           time.Duration
+	lambdaDKG          time.Duration
 	notarySet          map[types.NodeID]struct{}
 	privateKeys        map[types.NodeID]crypto.PrivateKey
 	DKGComplaint       map[uint64][]*types.DKGComplaint
@@ -45,7 +46,8 @@ type Governance struct {
 func NewGovernance(nodeCount int, lambda time.Duration) (
 	g *Governance, err error) {
 	g = &Governance{
-		lambda:             lambda,
+		lambdaBA:           lambda,
+		lambdaDKG:          lambda * 10,
 		notarySet:          make(map[types.NodeID]struct{}),
 		privateKeys:        make(map[types.NodeID]crypto.PrivateKey),
 		DKGComplaint:       make(map[uint64][]*types.DKGComplaint),
@@ -80,7 +82,8 @@ func (g *Governance) GetConfiguration(blockHeight uint64) *types.Config {
 		NumShards:  1,
 		NumChains:  uint32(len(g.notarySet)),
 		GenesisCRS: "__ DEXON",
-		Lambda:     g.lambda,
+		LambdaBA:   g.lambdaBA,
+		LambdaDKG:  g.lambdaDKG,
 		K:          0,
 		PhiRatio:   0.667,
 	}
diff --git a/core/ticker.go b/core/ticker.go
index bb5afb4..5dbbc2a 100644
--- a/core/ticker.go
+++ b/core/ticker.go
@@ -19,6 +19,15 @@ package core
 
 import "time"
 
+// TickerType is the type of ticker.
+type TickerType int
+
+// TickerType enum.
+const (
+	TickerBA TickerType = iota
+	TickerDKG
+)
+
 // defaultTicker is a wrapper to implement ticker interface based on
 // time.Ticker.
 type defaultTicker struct {
@@ -43,16 +52,23 @@ func (t *defaultTicker) Stop() {
 // newTicker is a helper to setup a ticker by giving an Governance. If
 // the governace object implements a ticker generator, a ticker from that
 // generator would be returned, else constructs a default one.
-func newTicker(gov Governance) (t Ticker) {
+func newTicker(gov Governance, tickerType TickerType) (t Ticker) {
 	type tickerGenerator interface {
-		NewTicker() Ticker
+		NewTicker(TickerType) Ticker
 	}
 
 	if gen, ok := gov.(tickerGenerator); ok {
-		t = gen.NewTicker()
+		t = gen.NewTicker(tickerType)
 	}
 	if t == nil {
-		t = newDefaultTicker(gov.GetConfiguration(0).Lambda)
+		var duration time.Duration
+		switch tickerType {
+		case TickerBA:
+			duration = gov.GetConfiguration(0).LambdaBA
+		case TickerDKG:
+			duration = gov.GetConfiguration(0).LambdaDKG
+		}
+		t = newDefaultTicker(duration)
 	}
 	return
 }
diff --git a/core/types/config.go b/core/types/config.go
index 1392ec5..4f5bd6e 100644
--- a/core/types/config.go
+++ b/core/types/config.go
@@ -26,8 +26,9 @@ type Config struct {
 	NumChains  uint32
 	GenesisCRS string
 
-	// Byzantine agreement related.
-	Lambda time.Duration
+	// Lambda related.
+	LambdaBA  time.Duration
+	LambdaDKG time.Duration
 
 	// Total ordering related.
 	K        int
diff --git a/core/types/dkg.go b/core/types/dkg.go
index d70bf98..ff0aa75 100644
--- a/core/types/dkg.go
+++ b/core/types/dkg.go
@@ -18,6 +18,8 @@
 package types
 
 import (
+	"fmt"
+
 	"github.com/dexon-foundation/dexon-consensus-core/crypto"
 	"github.com/dexon-foundation/dexon-consensus-core/crypto/dkg"
 )
@@ -31,6 +33,15 @@ type DKGPrivateShare struct {
 	Signature    crypto.Signature `json:"signature"`
 }
 
+func (p *DKGPrivateShare) String() string {
+	return fmt.Sprintf("prvShare(%d:%s->%s:%s:%s)",
+		p.Round,
+		p.ProposerID.String()[:6],
+		p.ReceiverID.String()[:6],
+		p.PrivateShare.String(),
+		p.Signature.String()[:6])
+}
+
 // DKGMasterPublicKey decrtibe a master public key in DKG protocol.
 type DKGMasterPublicKey struct {
 	ProposerID      NodeID              `json:"proposer_id"`