// Copyright 2018 The dexon-consensus Authors
// This file is part of the dexon-consensus library.
//
// The dexon-consensus 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 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 library. If not, see
// <http://www.gnu.org/licenses/>.
package core
import (
"context"
"encoding/json"
"sync"
"testing"
"time"
"github.com/stretchr/testify/suite"
"github.com/dexon-foundation/dexon-consensus/common"
"github.com/dexon-foundation/dexon-consensus/core/crypto"
"github.com/dexon-foundation/dexon-consensus/core/db"
"github.com/dexon-foundation/dexon-consensus/core/test"
"github.com/dexon-foundation/dexon-consensus/core/types"
typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
"github.com/dexon-foundation/dexon-consensus/core/utils"
)
// network implements core.Network.
type network struct {
nID types.NodeID
conn *networkConnection
}
// PullBlocks tries to pull blocks from the DEXON network.
func (n *network) PullBlocks(common.Hashes) {
}
// PullVotes tries to pull votes from the DEXON network.
func (n *network) PullVotes(types.Position) {
}
// PullRandomness tries to pull randomness from the DEXON network.
func (n *network) PullRandomness(common.Hashes) {
}
// BroadcastVote broadcasts vote to all nodes in DEXON network.
func (n *network) BroadcastVote(vote *types.Vote) {
n.conn.broadcast(n.nID, vote)
}
// BroadcastBlock broadcasts block to all nodes in DEXON network.
func (n *network) BroadcastBlock(block *types.Block) {
n.conn.broadcast(n.nID, block)
}
// BroadcastAgreementResult broadcasts agreement result to DKG set.
func (n *network) BroadcastAgreementResult(
randRequest *types.AgreementResult) {
n.conn.broadcast(n.nID, randRequest)
}
// SendDKGPrivateShare sends PrivateShare to a DKG participant.
func (n *network) SendDKGPrivateShare(
recv crypto.PublicKey, prvShare *typesDKG.PrivateShare) {
n.conn.send(n.nID, types.NewNodeID(recv), prvShare)
}
// BroadcastDKGPrivateShare broadcasts PrivateShare to all DKG participants.
func (n *network) BroadcastDKGPrivateShare(
prvShare *typesDKG.PrivateShare) {
n.conn.broadcast(n.nID, prvShare)
}
// BroadcastDKGPartialSignature broadcasts partialSignature to all
// DKG participants.
func (n *network) BroadcastDKGPartialSignature(
psig *typesDKG.PartialSignature) {
n.conn.broadcast(n.nID, psig)
}
// ReceiveChan returns a channel to receive messages from DEXON network.
func (n *network) ReceiveChan() <-chan types.Msg {
return make(chan types.Msg)
}
// ReportBadPeer reports that a peer is sending bad message.
func (n *network) ReportBadPeerChan() chan<- interface{} {
return n.conn.s.sink
}
func (nc *networkConnection) broadcast(from types.NodeID, msg interface{}) {
for nID := range nc.cons {
if nID == from {
continue
}
nc.send(from, nID, msg)
}
}
func (nc *networkConnection) send(from, to types.NodeID, msg interface{}) {
ch, exist := nc.cons[to]
if !exist {
return
}
msgCopy := msg
// Clone msg if necessary.
switch val := msg.(type) {
case *types.Block:
msgCopy = val.Clone()
case *typesDKG.PrivateShare:
// Use Marshal/Unmarshal to do deep copy.
data, err := json.Marshal(val)
if err != nil {
panic(err)
}
valCopy := &typesDKG.PrivateShare{}
if err := json.Unmarshal(data, valCopy); err != nil {
panic(err)
}
msgCopy = valCopy
}
ch <- types.Msg{
PeerID: from,
Payload: msgCopy,
}
}
type networkConnection struct {
s *ConsensusTestSuite
cons map[types.NodeID]chan types.Msg
}
func (nc *networkConnection) newNetwork(nID types.NodeID) *network {
return &network{
nID: nID,
conn: nc,
}
}
func (nc *networkConnection) setCon(nID types.NodeID, con *Consensus) {
ch := make(chan types.Msg, 1000)
nc.s.wg.Add(1)
go func() {
defer nc.s.wg.Done()
for {
var msg types.Msg
select {
case msg = <-ch:
case <-nc.s.ctx.Done():
return
}
var err error
// Testify package does not support concurrent call.
// Use panic() to detact error.
switch val := msg.Payload.(type) {
case *types.Block:
err = con.preProcessBlock(val)
case *types.Vote:
err = con.ProcessVote(val)
case *types.AgreementResult:
err = con.ProcessAgreementResult(val)
case *typesDKG.PrivateShare:
err = con.cfgModule.processPrivateShare(val)
case *typesDKG.PartialSignature:
err = con.cfgModule.processPartialSignature(val)
}
if err != nil {
panic(err)
}
}
}()
nc.cons[nID] = ch
}
type ConsensusTestSuite struct {
suite.Suite
ctx context.Context
ctxCancel context.CancelFunc
conn *networkConnection
sink chan interface{}
wg sync.WaitGroup
}
func (s *ConsensusTestSuite) SetupTest() {
s.ctx, s.ctxCancel = context.WithCancel(context.Background())
s.sink = make(chan interface{}, 1000)
s.wg.Add(1)
go func() {
defer s.wg.Done()
select {
case <-s.sink:
case <-s.ctx.Done():
return
}
}()
}
func (s *ConsensusTestSuite) TearDownTest() {
s.ctxCancel()
s.wg.Wait()
}
func (s *ConsensusTestSuite) newNetworkConnection() *networkConnection {
return &networkConnection{
s: s,
cons: make(map[types.NodeID]chan types.Msg),
}
}
func (s *ConsensusTestSuite) prepareConsensus(
dMoment time.Time,
gov *test.Governance,
prvKey crypto.PrivateKey,
conn *networkConnection) (
*test.App, *Consensus) {
app := test.NewApp(0, nil, nil)
dbInst, err := db.NewMemBackedDB()
s.Require().NoError(err)
nID := types.NewNodeID(prvKey.PublicKey())
network := conn.newNetwork(nID)
con := NewConsensus(
dMoment, app, gov, dbInst, network, prvKey, &common.NullLogger{})
conn.setCon(nID, con)
return app, con
}
func (s *ConsensusTestSuite) prepareConsensusWithDB(
dMoment time.Time,
gov *test.Governance,
prvKey crypto.PrivateKey,
conn *networkConnection,
dbInst db.Database) (
*test.App, *Consensus) {
app := test.NewApp(0, nil, nil)
nID := types.NewNodeID(prvKey.PublicKey())
network := conn.newNetwork(nID)
con := NewConsensus(
dMoment, app, gov, dbInst, network, prvKey, &common.NullLogger{})
conn.setCon(nID, con)
return app, con
}
func (s *ConsensusTestSuite) TestRegisteredDKGRecover() {
conn := s.newNetworkConnection()
prvKeys, pubKeys, err := test.NewKeys(1)
s.Require().NoError(err)
gov, err := test.NewGovernance(test.NewState(DKGDelayRound,
pubKeys, time.Second, &common.NullLogger{}, true), ConfigRoundShift)
s.Require().NoError(err)
gov.State().RequestChange(test.StateChangeRoundLength, uint64(200))
dMoment := time.Now().UTC()
dbInst, err := db.NewMemBackedDB()
s.Require().NoError(err)
_, con := s.prepareConsensusWithDB(dMoment, gov, prvKeys[0], conn, dbInst)
s.Require().Nil(con.cfgModule.dkg)
con.cfgModule.registerDKG(con.ctx, 0, 0, 10)
con.cfgModule.dkgLock.Lock()
defer con.cfgModule.dkgLock.Unlock()
_, newCon := s.prepareConsensusWithDB(dMoment, gov, prvKeys[0], conn, dbInst)
newCon.cfgModule.registerDKG(newCon.ctx, 0, 0, 10)
newCon.cfgModule.dkgLock.Lock()
defer newCon.cfgModule.dkgLock.Unlock()
s.Require().NotNil(newCon.cfgModule.dkg)
s.Require().True(newCon.cfgModule.dkg.prvShares.Equal(con.cfgModule.dkg.prvShares))
}
func (s *ConsensusTestSuite) TestDKGCRS() {
n := 21
lambda := 200 * time.Millisecond
if testing.Short() {
n = 7
lambda = 100 * time.Millisecond
}
if isTravisCI() {
lambda *= 5
}
conn := s.newNetworkConnection()
prvKeys, pubKeys, err := test.NewKeys(n)
s.Require().NoError(err)
gov, err := test.NewGovernance(test.NewState(DKGDelayRound,
pubKeys, lambda, &common.NullLogger{}, true), ConfigRoundShift)
s.Require().NoError(err)
gov.State().RequestChange(test.StateChangeRoundLength, uint64(200))
cons := map[types.NodeID]*Consensus{}
dMoment := time.Now().UTC()
for _, key := range prvKeys {
_, con := s.prepareConsensus(dMoment, gov, key, conn)
nID := types.NewNodeID(key.PublicKey())
cons[nID] = con
}
time.Sleep(gov.Configuration(0).MinBlockInterval * 4)
for _, con := range cons {
go con.runDKG(0, 0, 0, 0)
}
crsFinish := make(chan struct{}, len(cons))
for _, con := range cons {
go func(con *Consensus) {
height := uint64(0)
Loop:
for {
select {
case <-crsFinish:
break Loop
case <-time.After(lambda):
}
con.event.NotifyHeight(height)
height++
}
}(con)
}
for _, con := range cons {
func() {
con.dkgReady.L.Lock()
defer con.dkgReady.L.Unlock()
for con.dkgRunning != 2 {
con.dkgReady.Wait()
}
}()
}
for _, con := range cons {
go func(con *Consensus) {
con.runCRS(0, gov.CRS(0), false)
crsFinish <- struct{}{}
}(con)
}
s.NotNil(gov.CRS(1))
}
func (s *ConsensusTestSuite) TestSyncBA() {
lambdaBA := time.Second
conn := s.newNetworkConnection()
prvKeys, pubKeys, err := test.NewKeys(4)
s.Require().NoError(err)
gov, err := test.NewGovernance(test.NewState(DKGDelayRound,
pubKeys, lambdaBA, &common.NullLogger{}, true), ConfigRoundShift)
s.Require().NoError(err)
prvKey := prvKeys[0]
_, con := s.prepareConsensus(time.Now().UTC(), gov, prvKey, conn)
go con.Run()
defer con.Stop()
hash := common.NewRandomHash()
signers := make([]*utils.Signer, 0, len(prvKeys))
for _, prvKey := range prvKeys {
signers = append(signers, utils.NewSigner(prvKey))
}
pos := types.Position{
Round: 0,
Height: 20,
}
baResult := &types.AgreementResult{
BlockHash: hash,
Position: pos,
}
for _, signer := range signers {
vote := types.NewVote(types.VoteCom, hash, 0)
vote.Position = pos
s.Require().NoError(signer.SignVote(vote))
baResult.Votes = append(baResult.Votes, *vote)
}
// Make sure each agreement module is running. ProcessAgreementResult only
// works properly when agreement module is running:
// - the bias for round begin time would be 4 * lambda.
// - the ticker is 1 lambdaa.
time.Sleep(5 * lambdaBA)
s.Require().NoError(con.ProcessAgreementResult(baResult))
aID := con.baMgr.baModule.agreementID()
s.Equal(pos, aID)
// Negative cases are moved to TestVerifyAgreementResult in utils_test.go.
}
func (s *ConsensusTestSuite) TestInitialHeightEventTriggered() {
// Initial block is the last block of corresponding round, in this case,
// we should make sure all height event handlers could be triggered after
// returned from Consensus.prepare().
prvKeys, pubKeys, err := test.NewKeys(4)
s.Require().NoError(err)
// Prepare a governance instance, whose DKG-reset-count for round 2 is 1.
gov, err := test.NewGovernance(test.NewState(DKGDelayRound,
pubKeys, time.Second, &common.NullLogger{}, true), ConfigRoundShift)
gov.State().RequestChange(test.StateChangeRoundLength, uint64(100))
s.Require().NoError(err)
gov.NotifyRound(2, 201)
gov.NotifyRound(3, 301)
hash := common.NewRandomHash()
gov.ProposeCRS(2, hash[:])
hash = common.NewRandomHash()
gov.ResetDKG(hash[:])
s.Require().Equal(gov.DKGResetCount(2), uint64(1))
prvKey := prvKeys[0]
initBlock := &types.Block{
Hash: common.NewRandomHash(),
Position: types.Position{Round: 1, Height: 200},
}
dbInst, err := db.NewMemBackedDB()
s.Require().NoError(err)
nID := types.NewNodeID(prvKey.PublicKey())
conn := s.newNetworkConnection()
network := conn.newNetwork(nID)
con, err := NewConsensusFromSyncer(
initBlock,
false,
time.Now().UTC(),
test.NewApp(0, nil, nil),
gov,
dbInst,
network,
prvKey,
[]*types.Block(nil),
[]types.Msg{},
&common.NullLogger{},
)
s.Require().NoError(err)
// Here is the tricky part, check if block chain module can handle the
// block with height == 200.
s.Require().Equal(con.bcModule.configs[0].RoundID(), uint64(1))
s.Require().Equal(con.bcModule.configs[0].RoundEndHeight(), uint64(301))
}
func TestConsensus(t *testing.T) {
suite.Run(t, new(ConsensusTestSuite))
}
