core: add shard (#127)

A shard is basically DEXON v1 components,
except the strongly acked part, including:
- maintaining lattice structure
- total ordering
- generate consensus timestamp

5 files changed, 423 insertions, 29 deletions

diff --git a/core/consensus.go b/core/consensus.go
index 33a3d7f..7700296 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -104,7 +104,7 @@ func (recv *consensusReceiver) ConfirmBlock(hash common.Hash) {
 		log.Println(ErrUnknownBlockConfirmed, hash)
 		return
 	}
-	if err := recv.consensus.ProcessBlock(block); err != nil {
+	if err := recv.consensus.processBlock(block); err != nil {
 		log.Println(err)
 		return
 	}
@@ -273,7 +273,7 @@ BALoop:
 
 // RunLegacy starts running Legacy DEXON Consensus.
 func (con *Consensus) RunLegacy() {
-	go con.processMsg(con.network.ReceiveChan(), con.ProcessBlock)
+	go con.processMsg(con.network.ReceiveChan(), con.processBlock)
 	go con.processWitnessData()
 
 	chainID := uint32(0)
@@ -299,7 +299,7 @@ func (con *Consensus) RunLegacy() {
 	if err := con.PrepareGenesisBlock(genesisBlock, time.Now().UTC()); err != nil {
 		log.Println(err)
 	}
-	if err := con.ProcessBlock(genesisBlock); err != nil {
+	if err := con.processBlock(genesisBlock); err != nil {
 		log.Println(err)
 	}
 	con.network.BroadcastBlock(genesisBlock)
@@ -317,10 +317,10 @@ ProposingBlockLoop:
 				ChainID: chainID,
 			},
 		}
-		if err := con.PrepareBlock(block, time.Now().UTC()); err != nil {
+		if err := con.prepareBlock(block, time.Now().UTC()); err != nil {
 			log.Println(err)
 		}
-		if err := con.ProcessBlock(block); err != nil {
+		if err := con.processBlock(block); err != nil {
 			log.Println(err)
 		}
 		con.network.BroadcastBlock(block)
@@ -368,7 +368,7 @@ func (con *Consensus) proposeBlock(chainID uint32) *types.Block {
 			Height:  con.rbModule.nextHeight(chainID),
 		},
 	}
-	if err := con.PrepareBlock(block, time.Now().UTC()); err != nil {
+	if err := con.prepareBlock(block, time.Now().UTC()); err != nil {
 		log.Println(err)
 		return nil
 	}
@@ -464,8 +464,8 @@ func (con *Consensus) PreProcessBlock(b *types.Block) (err error) {
 	return
 }
 
-// ProcessBlock is the entry point to submit one block to a Consensus instance.
-func (con *Consensus) ProcessBlock(block *types.Block) (err error) {
+// processBlock is the entry point to submit one block to a Consensus instance.
+func (con *Consensus) processBlock(block *types.Block) (err error) {
 	if err := con.sanityCheck(block); err != nil {
 		return err
 	}
@@ -538,7 +538,7 @@ func (con *Consensus) checkPrepareBlock(
 }
 
 // PrepareBlock would setup header fields of block based on its ProposerID.
-func (con *Consensus) PrepareBlock(b *types.Block,
+func (con *Consensus) prepareBlock(b *types.Block,
 	proposeTime time.Time) (err error) {
 	if err = con.checkPrepareBlock(b, proposeTime); err != nil {
 		return
diff --git a/core/consensus_test.go b/core/consensus_test.go
index c3f33fa..701ee00 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -127,7 +127,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 	// to all core.Consensus objects.
 	broadcast := func(b *types.Block) {
 		for _, obj := range objs {
-			req.Nil(obj.con.ProcessBlock(b))
+			req.Nil(obj.con.processBlock(b))
 		}
 	}
 	// Genesis blocks
@@ -152,7 +152,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 	}
 	b11.Hash, err = hashBlock(b11)
 	s.Require().Nil(err)
-	req.Nil(objs[nodes[1]].con.PrepareBlock(b11, time.Now().UTC()))
+	req.Nil(objs[nodes[1]].con.prepareBlock(b11, time.Now().UTC()))
 	req.Len(b11.Acks, 4)
 	req.Contains(b11.Acks, b00.Hash)
 	req.Contains(b11.Acks, b10.Hash)
@@ -168,7 +168,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[0]].con.PrepareBlock(b01, time.Now().UTC()))
+	req.Nil(objs[nodes[0]].con.prepareBlock(b01, time.Now().UTC()))
 	req.Len(b01.Acks, 4)
 	req.Contains(b01.Acks, b11.Hash)
 	// Setup b21.
@@ -180,7 +180,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[2]].con.PrepareBlock(b21, time.Now().UTC()))
+	req.Nil(objs[nodes[2]].con.prepareBlock(b21, time.Now().UTC()))
 	req.Len(b21.Acks, 4)
 	req.Contains(b21.Acks, b11.Hash)
 	// Setup b31.
@@ -192,7 +192,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[3]].con.PrepareBlock(b31, time.Now().UTC()))
+	req.Nil(objs[nodes[3]].con.prepareBlock(b31, time.Now().UTC()))
 	req.Len(b31.Acks, 4)
 	req.Contains(b31.Acks, b11.Hash)
 	// Broadcast other height=1 blocks.
@@ -209,7 +209,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[0]].con.PrepareBlock(b02, time.Now().UTC()))
+	req.Nil(objs[nodes[0]].con.prepareBlock(b02, time.Now().UTC()))
 	req.Len(b02.Acks, 3)
 	req.Contains(b02.Acks, b01.Hash)
 	req.Contains(b02.Acks, b21.Hash)
@@ -223,7 +223,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[1]].con.PrepareBlock(b12, time.Now().UTC()))
+	req.Nil(objs[nodes[1]].con.prepareBlock(b12, time.Now().UTC()))
 	req.Len(b12.Acks, 4)
 	req.Contains(b12.Acks, b01.Hash)
 	req.Contains(b12.Acks, b11.Hash)
@@ -238,7 +238,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[2]].con.PrepareBlock(b22, time.Now().UTC()))
+	req.Nil(objs[nodes[2]].con.prepareBlock(b22, time.Now().UTC()))
 	req.Len(b22.Acks, 3)
 	req.Contains(b22.Acks, b01.Hash)
 	req.Contains(b22.Acks, b21.Hash)
@@ -252,7 +252,7 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		},
 		Hash: common.NewRandomHash(),
 	}
-	req.Nil(objs[nodes[3]].con.PrepareBlock(b32, time.Now().UTC()))
+	req.Nil(objs[nodes[3]].con.prepareBlock(b32, time.Now().UTC()))
 	req.Len(b32.Acks, 3)
 	req.Contains(b32.Acks, b01.Hash)
 	req.Contains(b32.Acks, b21.Hash)
@@ -318,10 +318,10 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 func (s *ConsensusTestSuite) TestPrepareBlock() {
 	// This test case would test these steps:
 	//  - Add all genesis blocks into lattice.
-	//  - Make sure Consensus.PrepareBlock would attempt to ack
+	//  - Make sure Consensus.prepareBlock would attempt to ack
 	//    all genesis blocks.
 	//  - Add the prepared block into lattice.
-	//  - Make sure Consensus.PrepareBlock would only attempt to
+	//  - Make sure Consensus.prepareBlock would only attempt to
 	//    ack the prepared block.
 	var (
 		gov, err = test.NewGovernance(4, time.Second)
@@ -350,17 +350,17 @@ func (s *ConsensusTestSuite) TestPrepareBlock() {
 	b30 := s.prepareGenesisBlock(nodes[3], 3, objs[nodes[3]].con)
 	for _, obj := range objs {
 		con := obj.con
-		req.Nil(con.ProcessBlock(b00))
-		req.Nil(con.ProcessBlock(b10))
-		req.Nil(con.ProcessBlock(b20))
-		req.Nil(con.ProcessBlock(b30))
+		req.Nil(con.processBlock(b00))
+		req.Nil(con.processBlock(b10))
+		req.Nil(con.processBlock(b20))
+		req.Nil(con.processBlock(b30))
 	}
 	b11 := &types.Block{
 		ProposerID: nodes[1],
 	}
 	// Sleep to make sure 'now' is slower than b10's timestamp.
 	time.Sleep(100 * time.Millisecond)
-	req.Nil(objs[nodes[1]].con.PrepareBlock(b11, time.Now().UTC()))
+	req.Nil(objs[nodes[1]].con.prepareBlock(b11, time.Now().UTC()))
 	// Make sure we would assign 'now' to the timestamp belongs to
 	// the proposer.
 	req.True(
@@ -368,12 +368,12 @@ func (s *ConsensusTestSuite) TestPrepareBlock() {
 			b10.Timestamp) > 100*time.Millisecond)
 	for _, obj := range objs {
 		con := obj.con
-		req.Nil(con.ProcessBlock(b11))
+		req.Nil(con.processBlock(b11))
 	}
 	b12 := &types.Block{
 		ProposerID: nodes[1],
 	}
-	req.Nil(objs[nodes[1]].con.PrepareBlock(b12, time.Now().UTC()))
+	req.Nil(objs[nodes[1]].con.prepareBlock(b12, time.Now().UTC()))
 	req.Len(b12.Acks, 1)
 	req.Contains(b12.Acks, b11.Hash)
 }
diff --git a/core/shard.go b/core/shard.go
new file mode 100644
index 0000000..b6a17e1
--- /dev/null
+++ b/core/shard.go
@@ -0,0 +1,197 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core 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-core 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-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"sync"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+	"github.com/dexon-foundation/dexon-consensus-core/core/blockdb"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+	"github.com/dexon-foundation/dexon-consensus-core/crypto"
+)
+
+// Shard represents a unit to produce a global ordering from multiple chains.
+type Shard struct {
+	lock     sync.RWMutex
+	ID       uint32
+	nodeID   types.NodeID
+	prvKey   crypto.PrivateKey
+	sigToPub SigToPubFn
+	chainNum uint32
+	app      Application
+	db       blockdb.BlockDatabase
+	pool     blockPool
+	lattice  *blockLattice
+	toModule *totalOrdering
+	ctModule *consensusTimestamp
+}
+
+// NewShard constructs an Shard instance.
+func NewShard(
+	ID uint32,
+	cfg *types.Config,
+	prvKey crypto.PrivateKey,
+	sigToPub SigToPubFn,
+	app Application,
+	db blockdb.BlockDatabase) (s *Shard) {
+
+	s = &Shard{
+		ID:       ID,
+		nodeID:   types.NewNodeID(prvKey.PublicKey()),
+		prvKey:   prvKey,
+		sigToPub: sigToPub,
+		chainNum: cfg.NumChains,
+		app:      app,
+		db:       db,
+		pool:     newBlockPool(cfg.NumChains),
+		lattice:  newBlockLattice(ID, cfg.NumChains),
+		toModule: newTotalOrdering(
+			uint64(cfg.K),
+			uint64(float32(cfg.NumChains-1)*cfg.PhiRatio+1),
+			cfg.NumChains),
+		ctModule: newConsensusTimestamp(),
+	}
+	return
+}
+
+// PrepareBlock setup block's field based on current lattice status.
+func (s *Shard) PrepareBlock(
+	b *types.Block, proposeTime time.Time) (err error) {
+
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	s.lattice.prepareBlock(b)
+	// TODO(mission): the proposeTime might be earlier than tip block of
+	//                that chain. We should let blockLattice suggest the time.
+	b.ProposerID = s.nodeID
+	b.Timestamp = proposeTime
+	b.Payload = s.app.PreparePayload(b.Position)
+	if b.Hash, err = hashBlock(b); err != nil {
+		return
+	}
+	if b.Signature, err = s.prvKey.Sign(b.Hash); err != nil {
+		return
+	}
+	return
+}
+
+// SanityCheck check if a block is valid based on current lattice status.
+//
+// If some acking blocks don't exists, Shard would help to cache this block
+// and retry when lattice updated in Shard.ProcessBlock.
+func (s *Shard) SanityCheck(b *types.Block) (err error) {
+	// Check block.Position.
+	if b.Position.ShardID != s.ID {
+		err = ErrIncorrectBlockPosition
+		return
+	}
+	// Check the hash of block.
+	hash, err := hashBlock(b)
+	if err != nil || hash != b.Hash {
+		err = ErrIncorrectHash
+		return
+	}
+	// Check the signer.
+	pubKey, err := s.sigToPub(b.Hash, b.Signature)
+	if err != nil {
+		return
+	}
+	if !b.ProposerID.Equal(crypto.Keccak256Hash(pubKey.Bytes())) {
+		err = ErrIncorrectSignature
+		return
+	}
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	if err = s.lattice.sanityCheck(b); err != nil {
+		// Add to block pool, once the lattice updated,
+		// would be checked again.
+		if err == ErrAckingBlockNotExists {
+			s.pool.addBlock(b)
+		}
+		return
+	}
+	return
+}
+
+// ProcessBlock adds a block into lattice, and deliver ordered blocks.
+// If any block pass sanity check after this block add into lattice, they
+// would be returned, too.
+//
+// NOTE: assume the block passed sanity check.
+func (s *Shard) ProcessBlock(
+	input *types.Block) (verified, delivered []*types.Block, err error) {
+
+	var (
+		tip, b         *types.Block
+		toDelivered    []*types.Block
+		inLattice      []*types.Block
+		earlyDelivered bool
+	)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	if inLattice, err = s.lattice.addBlock(input); err != nil {
+		return
+	}
+	if err = s.db.Put(*input); err != nil {
+		return
+	}
+	// TODO(mission): remove this hack, BA related stuffs should not
+	//                be done here.
+	s.app.StronglyAcked(input.Hash)
+	s.app.BlockConfirmed(input)
+	// Purge blocks in pool with the same chainID and lower height.
+	s.pool.purgeBlocks(input.Position.ChainID, input.Position.Height)
+	// Replay tips in pool to check their validity.
+	for i := uint32(0); i < s.chainNum; i++ {
+		if tip = s.pool.tip(i); tip == nil {
+			continue
+		}
+		err = s.lattice.sanityCheck(tip)
+		if err == nil {
+			verified = append(verified, tip)
+		}
+		if err == ErrAckingBlockNotExists {
+			continue
+		}
+		s.pool.removeTip(i)
+	}
+	// Perform total ordering for each block added to lattice.
+	for _, b = range inLattice {
+		toDelivered, earlyDelivered, err = s.toModule.processBlock(b)
+		if err != nil {
+			return
+		}
+		if len(toDelivered) == 0 {
+			continue
+		}
+		hashes := make(common.Hashes, len(toDelivered))
+		for idx := range toDelivered {
+			hashes[idx] = toDelivered[idx].Hash
+		}
+		s.app.TotalOrderingDeliver(hashes, earlyDelivered)
+		// Perform timestamp generation.
+		if err = s.ctModule.processBlocks(toDelivered); err != nil {
+			return
+		}
+		delivered = append(delivered, toDelivered...)
+	}
+	return
+}
diff --git a/core/shard_test.go b/core/shard_test.go
new file mode 100644
index 0000000..db341a3
--- /dev/null
+++ b/core/shard_test.go
@@ -0,0 +1,192 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core 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-core 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-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus-core/core/blockdb"
+	"github.com/dexon-foundation/dexon-consensus-core/core/test"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+	"github.com/dexon-foundation/dexon-consensus-core/crypto/eth"
+	"github.com/stretchr/testify/suite"
+)
+
+// testShardMgr wraps compaction chain and shard.
+type testShardMgr struct {
+	shard    *Shard
+	ccModule *compactionChain
+	app      *test.App
+	db       blockdb.BlockDatabase
+}
+
+func (mgr *testShardMgr) prepareBlock(
+	chainID uint32) (b *types.Block, err error) {
+
+	b = &types.Block{
+		Position: types.Position{
+			ChainID: chainID,
+		}}
+	err = mgr.shard.PrepareBlock(b, time.Now().UTC())
+	return
+}
+
+// Process describes the usage of Shard.ProcessBlock.
+func (mgr *testShardMgr) processBlock(b *types.Block) (err error) {
+	var (
+		delivered []*types.Block
+		verified  []*types.Block
+		pendings  = []*types.Block{b}
+	)
+	if err = mgr.shard.SanityCheck(b); err != nil {
+		if err == ErrAckingBlockNotExists {
+			err = nil
+		}
+		return
+	}
+	for {
+		if len(pendings) == 0 {
+			break
+		}
+		b, pendings = pendings[0], pendings[1:]
+		if verified, delivered, err = mgr.shard.ProcessBlock(b); err != nil {
+			return
+		}
+		// Deliver blocks.
+		for _, b = range delivered {
+			if err = mgr.ccModule.processBlock(b); err != nil {
+				return
+			}
+			if err = mgr.db.Update(*b); err != nil {
+				return
+			}
+			mgr.app.DeliverBlock(b.Hash, b.Witness.Timestamp)
+		}
+		// Update pending blocks for verified block (pass sanity check).
+		pendings = append(pendings, verified...)
+	}
+	return
+}
+
+type ShardTestSuite struct {
+	suite.Suite
+}
+
+func (s *ShardTestSuite) newTestShardMgr(cfg *types.Config) *testShardMgr {
+	var req = s.Require()
+	// Setup private key.
+	prvKey, err := eth.NewPrivateKey()
+	req.Nil(err)
+	// Setup blockdb.
+	db, err := blockdb.NewMemBackedBlockDB()
+	req.Nil(err)
+	// Setup application.
+	app := test.NewApp()
+	// Setup shard.
+	return &testShardMgr{
+		ccModule: newCompactionChain(db, eth.SigToPub),
+		app:      app,
+		db:       db,
+		shard: NewShard(
+			uint32(0),
+			cfg,
+			prvKey,
+			eth.SigToPub,
+			app,
+			db)}
+}
+
+func (s *ShardTestSuite) TestBasicUsage() {
+	// One shard prepare blocks on chains randomly selected each time
+	// and process it. Those generated blocks and kept into a buffer, and
+	// process by other shard instances with random order.
+	var (
+		blockNum      = 100
+		chainNum      = uint32(19)
+		otherShardNum = 20
+		req           = s.Require()
+		err           error
+		cfg           = types.Config{
+			NumChains: chainNum,
+			PhiRatio:  float32(2) / float32(3),
+			K:         0,
+		}
+		master    = s.newTestShardMgr(&cfg)
+		apps      = []*test.App{master.app}
+		revealSeq = map[string]struct{}{}
+	)
+	// Master-shard generates blocks.
+	for i := uint32(0); i < chainNum; i++ {
+		// Produce genesis blocks should be delivered before all other blocks,
+		// or the consensus time would be wrong.
+		b, err := master.prepareBlock(i)
+		req.NotNil(b)
+		req.Nil(err)
+		// We've ignored the error for "acking blocks don't exist".
+		req.Nil(master.processBlock(b))
+	}
+	for i := 0; i < (blockNum - int(chainNum)); i++ {
+		b, err := master.prepareBlock(uint32(rand.Intn(int(chainNum))))
+		req.NotNil(b)
+		req.Nil(err)
+		// We've ignored the error for "acking blocks don't exist".
+		req.Nil(master.processBlock(b))
+	}
+	// Now we have some blocks, replay them on different shards.
+	iter, err := master.db.GetAll()
+	req.Nil(err)
+	revealer, err := test.NewRandomRevealer(iter)
+	req.Nil(err)
+	for i := 0; i < otherShardNum; i++ {
+		revealer.Reset()
+		revealed := ""
+		other := s.newTestShardMgr(&cfg)
+		for {
+			b, err := revealer.Next()
+			if err != nil {
+				if err == blockdb.ErrIterationFinished {
+					err = nil
+					break
+				}
+			}
+			req.Nil(err)
+			req.Nil(other.processBlock(&b))
+			revealed += b.Hash.String() + ","
+			revealSeq[revealed] = struct{}{}
+		}
+		apps = append(apps, other.app)
+	}
+	// Make sure not only one revealing sequence.
+	req.True(len(revealSeq) > 1)
+	// Make sure nothing goes wrong.
+	for i, app := range apps {
+		req.Nil(app.Verify())
+		for j, otherApp := range apps {
+			if i >= j {
+				continue
+			}
+			req.Nil(app.Compare(otherApp))
+		}
+	}
+}
+
+func TestShard(t *testing.T) {
+	suite.Run(t, new(ShardTestSuite))
+}
diff --git a/core/test/governance.go b/core/test/governance.go
index a2e6f69..63462d0 100644
--- a/core/test/governance.go
+++ b/core/test/governance.go
@@ -65,8 +65,13 @@ func NewGovernance(nodeCount int, lambda time.Duration) (
 
 // GetNotarySet implements Governance interface to return current
 // notary set.
-func (g *Governance) GetNotarySet() map[types.NodeID]struct{} {
-	return g.notarySet
+func (g *Governance) GetNotarySet() (ret map[types.NodeID]struct{}) {
+	// Return a cloned map.
+	ret = make(map[types.NodeID]struct{})
+	for k := range g.notarySet {
+		ret[k] = struct{}{}
+	}
+	return
 }
 
 // GetConfiguration returns the configuration at a given block height.