core: Remove agreement result (#514)

* core: remove agreement result for round with randomness

* remove agr test in syncer

* fixup

* remove randomness field from agreement result

* modify test

2 files changed, 168 insertions, 7 deletions

diff --git a/core/test/network.go b/core/test/network.go
index 6034fa6..c903c57 100644
--- a/core/test/network.go
+++ b/core/test/network.go
@@ -130,13 +130,52 @@ func (req *PullRequest) UnmarshalJSON(data []byte) (err error) {
 	return
 }
 
+// NetworkCensor is a interface to determine if a message should be censored.
+type NetworkCensor interface {
+	Censor(interface{}) bool
+}
+
+type censorClient struct {
+	TransportClient
+
+	censor NetworkCensor
+	lock   sync.RWMutex
+}
+
+func (cc *censorClient) Send(ID types.NodeID, msg interface{}) error {
+	if func() bool {
+		cc.lock.RLock()
+		defer cc.lock.RUnlock()
+		return cc.censor.Censor(msg)
+	}() {
+		return nil
+	}
+	return cc.TransportClient.Send(ID, msg)
+}
+
+func (cc *censorClient) Broadcast(
+	IDs map[types.NodeID]struct{}, latency LatencyModel, msg interface{}) error {
+	if func() bool {
+		cc.lock.RLock()
+		defer cc.lock.RUnlock()
+		return cc.censor.Censor(msg)
+	}() {
+		return nil
+	}
+	return cc.TransportClient.Broadcast(IDs, latency, msg)
+}
+
+type dummyCensor struct{}
+
+func (dc *dummyCensor) Censor(interface{}) bool { return false }
+
 // Network implements core.Network interface based on TransportClient.
 type Network struct {
 	ID                   types.NodeID
 	config               NetworkConfig
 	ctx                  context.Context
 	ctxCancel            context.CancelFunc
-	trans                TransportClient
+	trans                *censorClient
 	dMoment              time.Time
 	fromTransport        <-chan *TransportEnvelope
 	toConsensus          chan interface{}
@@ -156,6 +195,8 @@ type Network struct {
 	cache                *utils.NodeSetCache
 	notarySetCachesLock  sync.Mutex
 	notarySetCaches      map[uint64]map[types.NodeID]struct{}
+	censor               NetworkCensor
+	censorLock           sync.RWMutex
 }
 
 // NewNetwork setup network stuffs for nodes, which provides an
@@ -175,22 +216,48 @@ func NewNetwork(pubKey crypto.PublicKey, config NetworkConfig) (
 		notarySetCaches:  make(map[uint64]map[types.NodeID]struct{}),
 		voteCache: make(
 			map[types.Position]map[types.VoteHeader]*types.Vote),
+		censor: &dummyCensor{},
 	}
 	n.ctx, n.ctxCancel = context.WithCancel(context.Background())
 	// Construct transport layer.
+	var trans TransportClient
 	switch config.Type {
 	case NetworkTypeTCPLocal:
-		n.trans = NewTCPTransportClient(pubKey, config.Marshaller, true)
+		trans = NewTCPTransportClient(pubKey, config.Marshaller, true)
 	case NetworkTypeTCP:
-		n.trans = NewTCPTransportClient(pubKey, config.Marshaller, false)
+		trans = NewTCPTransportClient(pubKey, config.Marshaller, false)
 	case NetworkTypeFake:
-		n.trans = NewFakeTransportClient(pubKey)
+		trans = NewFakeTransportClient(pubKey)
 	default:
 		panic(fmt.Errorf("unknown network type: %v", config.Type))
 	}
+	n.trans = &censorClient{
+		TransportClient: trans,
+		censor:          &dummyCensor{},
+	}
 	return
 }
 
+// SetCensor to this network module.
+func (n *Network) SetCensor(censorIn, censorOut NetworkCensor) {
+	if censorIn == nil {
+		censorIn = &dummyCensor{}
+	}
+	if censorOut == nil {
+		censorOut = &dummyCensor{}
+	}
+	func() {
+		n.censorLock.Lock()
+		defer n.censorLock.Unlock()
+		n.censor = censorIn
+	}()
+	func() {
+		n.trans.lock.Lock()
+		defer n.trans.lock.Unlock()
+		n.trans.censor = censorOut
+	}()
+}
+
 // PullBlocks implements core.Network interface.
 func (n *Network) PullBlocks(hashes common.Hashes) {
 	go n.pullBlocksAsync(hashes)
@@ -224,6 +291,12 @@ func (n *Network) BroadcastBlock(block *types.Block) {
 		panic(err)
 	}
 	n.addBlockToCache(block)
+	if block.IsFinalized() {
+		n.addBlockFinalizationToCache(
+			block.Hash,
+			block.Finalization.Height,
+			block.Finalization.Randomness)
+	}
 }
 
 // BroadcastAgreementResult implements core.Network interface.
@@ -232,9 +305,13 @@ func (n *Network) BroadcastAgreementResult(
 	if !n.markAgreementResultAsSent(result.BlockHash) {
 		return
 	}
-	n.addBlockRandomnessToCache(result.BlockHash, result.Randomness)
+	n.addBlockFinalizationToCache(
+		result.BlockHash,
+		result.FinalizationHeight,
+		nil,
+	)
 	notarySet := n.getNotarySet(result.Position.Round)
-	count := len(notarySet) * gossipAgreementResultPercent / 100
+	count := len(notarySet)*gossipAgreementResultPercent/100 + 1
 	for nID := range notarySet {
 		if count--; count < 0 {
 			break
@@ -303,6 +380,13 @@ func (n *Network) Setup(serverEndpoint interface{}) (err error) {
 }
 
 func (n *Network) dispatchMsg(e *TransportEnvelope) {
+	if func() bool {
+		n.censorLock.RLock()
+		defer n.censorLock.RUnlock()
+		return n.censor.Censor(e.Msg)
+	}() {
+		return
+	}
 	msg := n.cloneForFake(e.Msg)
 	switch v := msg.(type) {
 	case *types.Block:
@@ -539,13 +623,15 @@ func (n *Network) addBlockToCache(b *types.Block) {
 	n.blockCache[b.Hash] = b.Clone()
 }
 
-func (n *Network) addBlockRandomnessToCache(hash common.Hash, rand []byte) {
+func (n *Network) addBlockFinalizationToCache(
+	hash common.Hash, height uint64, rand []byte) {
 	n.blockCacheLock.Lock()
 	defer n.blockCacheLock.Unlock()
 	block, exist := n.blockCache[hash]
 	if !exist {
 		return
 	}
+	block.Finalization.Height = height
 	block.Finalization.Randomness = rand
 }
 
diff --git a/core/test/network_test.go b/core/test/network_test.go
index 993ae70..d0e9fb2 100644
--- a/core/test/network_test.go
+++ b/core/test/network_test.go
@@ -291,6 +291,81 @@ func (s *NetworkTestSuite) TestBroadcastToSet() {
 	req.IsType(&types.Block{}, <-notaryNode.ReceiveChan())
 }
 
+type testVoteCensor struct{}
+
+func (vc *testVoteCensor) Censor(msg interface{}) bool {
+	if _, ok := msg.(*types.Vote); ok {
+		return true
+	}
+	return false
+}
+
+func (s *NetworkTestSuite) TestCensor() {
+	var (
+		req       = s.Require()
+		peerCount = 5
+	)
+	_, pubKeys, err := NewKeys(peerCount)
+	req.NoError(err)
+	networks := s.setupNetworks(pubKeys)
+	receiveChans := make(map[types.NodeID]<-chan interface{}, peerCount)
+	for nID, node := range networks {
+		receiveChans[nID] = node.ReceiveChan()
+	}
+
+	censor := &testVoteCensor{}
+	vote := &types.Vote{}
+	censorNodeID := types.NewNodeID(pubKeys[0])
+	otherNodeID := types.NewNodeID(pubKeys[1])
+	censorNode := networks[censorNodeID]
+	otherNode := networks[otherNodeID]
+
+	// Censor incomming votes.
+	censorNode.SetCensor(censor, nil)
+	otherNode.BroadcastVote(vote)
+	time.Sleep(50 * time.Millisecond)
+	for nID, receiveChan := range receiveChans {
+		if nID == otherNodeID || nID == censorNodeID {
+			req.Equal(0, len(receiveChan))
+		} else {
+			req.Equal(1, len(receiveChan))
+			req.IsType(&types.Vote{}, <-receiveChan)
+		}
+	}
+
+	// Censor outgoing votes.
+	censorNode.SetCensor(nil, censor)
+	censorNode.BroadcastVote(vote)
+	time.Sleep(50 * time.Millisecond)
+	for _, receiveChan := range receiveChans {
+		req.Equal(0, len(receiveChan))
+	}
+
+	// No censorship.
+	censorNode.SetCensor(nil, nil)
+	otherNode.BroadcastVote(vote)
+	time.Sleep(50 * time.Millisecond)
+	for nID, receiveChan := range receiveChans {
+		if nID == otherNodeID {
+			req.Equal(0, len(receiveChan))
+		} else {
+			req.Equal(1, len(receiveChan))
+			req.IsType(&types.Vote{}, <-receiveChan)
+		}
+	}
+	censorNode.BroadcastVote(vote)
+	time.Sleep(50 * time.Millisecond)
+	for nID, receiveChan := range receiveChans {
+		if nID == censorNodeID {
+			req.Equal(0, len(receiveChan))
+		} else {
+			req.Equal(1, len(receiveChan))
+			req.IsType(&types.Vote{}, <-receiveChan)
+		}
+	}
+
+}
+
 func TestNetwork(t *testing.T) {
 	suite.Run(t, new(NetworkTestSuite))
 }