dex: some minor improvements (#195)

* dex: improve some msg propagation

* dex: support send a batch of lattice blocks, votes, randomnesses

To reduce msgs number of PullBlocks, PullVotes, PullRandomness

* dex: minor improvement

3 files changed, 117 insertions, 144 deletions

diff --git a/dex/handler.go b/dex/handler.go
index 71962b865..f56c6f5dc 100644
--- a/dex/handler.go
+++ b/dex/handler.go
@@ -40,7 +40,6 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
-	"net"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -794,24 +793,28 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if !pm.isBlockProposer {
 			break
 		}
-		var block coreTypes.Block
-		if err := msg.Decode(&block); err != nil {
+		var blocks []*coreTypes.Block
+		if err := msg.Decode(&blocks); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		pm.cache.addBlock(&block)
-		pm.receiveCh <- &block
+		for _, block := range blocks {
+			pm.cache.addBlock(block)
+			pm.receiveCh <- block
+		}
 	case msg.Code == VoteMsg:
 		if !pm.isBlockProposer {
 			break
 		}
-		var vote coreTypes.Vote
-		if err := msg.Decode(&vote); err != nil {
+		var votes []*coreTypes.Vote
+		if err := msg.Decode(&votes); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		if vote.Type >= coreTypes.VotePreCom {
-			pm.cache.addVote(&vote)
+		for _, vote := range votes {
+			if vote.Type >= coreTypes.VotePreCom {
+				pm.cache.addVote(vote)
+			}
+			pm.receiveCh <- vote
 		}
-		pm.receiveCh <- &vote
 	case msg.Code == AgreementMsg:
 		if !pm.isBlockProposer {
 			break
@@ -821,17 +824,21 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&agreement); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
+		p.MarkAgreement(rlpHash(agreement))
 		pm.receiveCh <- &agreement
 	case msg.Code == RandomnessMsg:
 		if !pm.isBlockProposer {
 			break
 		}
 		// Broadcast this to all peer
-		var randomness coreTypes.BlockRandomnessResult
-		if err := msg.Decode(&randomness); err != nil {
+		var randomnesses []*coreTypes.BlockRandomnessResult
+		if err := msg.Decode(&randomnesses); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		pm.receiveCh <- &randomness
+		for _, randomness := range randomnesses {
+			p.MarkRandomness(rlpHash(randomness))
+			pm.receiveCh <- randomness
+		}
 	case msg.Code == DKGPrivateShareMsg:
 		if !pm.isBlockProposer {
 			break
@@ -841,6 +848,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&ps); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
+		p.MarkDKGPrivateShares(rlpHash(ps))
 		pm.receiveCh <- &ps
 	case msg.Code == DKGPartialSignatureMsg:
 		if !pm.isBlockProposer {
@@ -862,11 +870,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		blocks := pm.cache.blocks(hashes)
 		log.Debug("Push blocks", "blocks", blocks)
-		for _, block := range blocks {
-			if err := p.SendLatticeBlock(block); err != nil {
-				return err
-			}
-		}
+		return p.SendLatticeBlocks(blocks)
 	case msg.Code == PullVotesMsg:
 		if !pm.isBlockProposer {
 			break
@@ -885,11 +889,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		votes := pm.cache.votes(pos)
 		log.Debug("Push votes", "votes", votes)
-		for _, vote := range votes {
-			if err := p.SendVote(vote); err != nil {
-				return err
-			}
-		}
+		return p.SendVotes(votes)
 	case msg.Code == PullRandomnessMsg:
 		if !pm.isBlockProposer {
 			break
@@ -898,13 +898,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&hashes); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		randomness := pm.cache.randomness(hashes)
-		log.Debug("Push randomness", "randomness", randomness)
-		for _, randomness := range randomness {
-			if err := p.SendRandomness(randomness); err != nil {
-				return err
-			}
-		}
+		randomnesses := pm.cache.randomness(hashes)
+		log.Debug("Push randomness", "randomness", randomnesses)
+		return p.SendRandomnesses(randomnesses)
 	case msg.Code == GetGovStateMsg:
 		var hash common.Hash
 		if err := msg.Decode(&hash); err != nil {
@@ -995,12 +991,11 @@ func (pm *ProtocolManager) BroadcastRecords(records []*enr.Record) {
 	}
 }
 
-// TODO(sonic): block size is big, try not to send to all peers
-// to reduce traffic
+// BroadcastLatticeBlock broadcasts the lattice block to all its peers.
 func (pm *ProtocolManager) BroadcastLatticeBlock(block *coreTypes.Block) {
 	pm.cache.addBlock(block)
-	for _, peer := range pm.peers.PeersWithoutLatticeBlock(rlpHash(block)) {
-		peer.AsyncSendLatticeBlock(block)
+	for _, peer := range pm.peers.Peers() {
+		peer.AsyncSendLatticeBlocks([]*coreTypes.Block{block})
 	}
 }
 
@@ -1014,11 +1009,8 @@ func (pm *ProtocolManager) BroadcastVote(vote *coreTypes.Vote) {
 		chainID: vote.Position.ChainID,
 		round:   vote.Position.Round,
 	}
-	h := rlpHash(vote)
 	for _, peer := range pm.peers.PeersWithLabel(label) {
-		if !peer.knownVotes.Contains(h) {
-			peer.AsyncSendVote(vote)
-		}
+		peer.AsyncSendVotes([]*coreTypes.Vote{vote})
 	}
 }
 
@@ -1050,15 +1042,16 @@ func (pm *ProtocolManager) BroadcastRandomnessResult(
 		chainID: randomness.Position.ChainID,
 		round:   randomness.Position.Round,
 	}
+	randomnesses := []*coreTypes.BlockRandomnessResult{randomness}
 	for _, peer := range pm.peers.PeersWithLabel(label) {
 		if !peer.knownRandomnesses.Contains(rlpHash(randomness)) {
-			peer.AsyncSendRandomness(randomness)
+			peer.AsyncSendRandomnesses(randomnesses)
 		}
 	}
 
 	// TODO(sonic): send to some of other nodes (gossip)
 	for _, peer := range pm.peers.PeersWithoutRandomness(rlpHash(randomness)) {
-		peer.AsyncSendRandomness(randomness)
+		peer.AsyncSendRandomnesses(randomnesses)
 	}
 }
 
@@ -1069,12 +1062,13 @@ func (pm *ProtocolManager) SendDKGPrivateShare(
 	if err != nil {
 		panic(err)
 	}
-	n := enode.NewV4(pk, net.IP{}, 0, 0)
 
-	if p := pm.peers.Peer(n.ID().String()); p != nil {
+	id := enode.PubkeyToIDV4(pk)
+
+	if p := pm.peers.Peer(id.String()); p != nil {
 		p.AsyncSendDKGPrivateShare(privateShare)
 	} else {
-		log.Error("Failed to send DKG private share", "publicKey", n.ID().String())
+		log.Error("Failed to send DKG private share", "publicKey", id.String())
 	}
 }
 
@@ -1092,9 +1086,7 @@ func (pm *ProtocolManager) BroadcastDKGPartialSignature(
 	psig *dkgTypes.PartialSignature) {
 	label := peerLabel{set: dkgset, round: psig.Round}
 	for _, peer := range pm.peers.PeersWithLabel(label) {
-		if !peer.knownDKGPartialSignatures.Contains(rlpHash(psig)) {
-			peer.AsyncSendDKGPartialSignature(psig)
-		}
+		peer.AsyncSendDKGPartialSignature(psig)
 	}
 }
 
diff --git a/dex/peer.go b/dex/peer.go
index 2c531ee07..a157709f0 100644
--- a/dex/peer.go
+++ b/dex/peer.go
@@ -66,14 +66,9 @@ const (
 	maxKnownRecords = 32768 // Maximum records hashes to keep in the known list (prevent DOS)
 	maxKnownBlocks  = 1024  // Maximum block hashes to keep in the known list (prevent DOS)
 
-	/*
-		maxKnownLatticeBLocks       = 2048
-		maxKnownVotes               = 2048
-		maxKnownAgreements          = 10240
-		maxKnownRandomnesses        = 10240
-		maxKnownDKGPrivateShare     = 1024 // this related to DKG Size
-		maxKnownDKGPartialSignature = 1024 // this related to DKG Size
-	*/
+	maxKnownAgreements       = 10240
+	maxKnownRandomnesses     = 10240
+	maxKnownDKGPrivateShares = 1024 // this related to DKG Size
 
 	// maxQueuedTxs is the maximum number of transaction lists to queue up before
 	// dropping broadcasts. This is a sensitive number as a transaction list might
@@ -155,20 +150,17 @@ type peer struct {
 	knownTxs                   mapset.Set // Set of transaction hashes known to be known by this peer
 	knownRecords               mapset.Set // Set of node record known to be known by this peer
 	knownBlocks                mapset.Set // Set of block hashes known to be known by this peer
-	knownLatticeBlocks         mapset.Set
-	knownVotes                 mapset.Set
 	knownAgreements            mapset.Set
 	knownRandomnesses          mapset.Set
 	knownDKGPrivateShares      mapset.Set
-	knownDKGPartialSignatures  mapset.Set
 	queuedTxs                  chan []*types.Transaction // Queue of transactions to broadcast to the peer
 	queuedRecords              chan []*enr.Record        // Queue of node records to broadcast to the peer
 	queuedProps                chan *types.Block         // Queue of blocks to broadcast to the peer
 	queuedAnns                 chan *types.Block         // Queue of blocks to announce to the peer
-	queuedLatticeBlocks        chan *coreTypes.Block
-	queuedVotes                chan *coreTypes.Vote
+	queuedLatticeBlocks        chan []*coreTypes.Block
+	queuedVotes                chan []*coreTypes.Vote
 	queuedAgreements           chan *coreTypes.AgreementResult
-	queuedRandomnesses         chan *coreTypes.BlockRandomnessResult
+	queuedRandomnesses         chan []*coreTypes.BlockRandomnessResult
 	queuedDKGPrivateShares     chan *dkgTypes.PrivateShare
 	queuedDKGPartialSignatures chan *dkgTypes.PartialSignature
 	queuedPullBlocks           chan coreCommon.Hashes
@@ -186,20 +178,17 @@ func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
 		knownTxs:                   mapset.NewSet(),
 		knownRecords:               mapset.NewSet(),
 		knownBlocks:                mapset.NewSet(),
-		knownLatticeBlocks:         mapset.NewSet(),
-		knownVotes:                 mapset.NewSet(),
 		knownAgreements:            mapset.NewSet(),
 		knownRandomnesses:          mapset.NewSet(),
 		knownDKGPrivateShares:      mapset.NewSet(),
-		knownDKGPartialSignatures:  mapset.NewSet(),
 		queuedTxs:                  make(chan []*types.Transaction, maxQueuedTxs),
 		queuedRecords:              make(chan []*enr.Record, maxQueuedRecords),
 		queuedProps:                make(chan *types.Block, maxQueuedProps),
 		queuedAnns:                 make(chan *types.Block, maxQueuedAnns),
-		queuedLatticeBlocks:        make(chan *coreTypes.Block, maxQueuedLatticeBlocks),
-		queuedVotes:                make(chan *coreTypes.Vote, maxQueuedVotes),
+		queuedLatticeBlocks:        make(chan []*coreTypes.Block, maxQueuedLatticeBlocks),
+		queuedVotes:                make(chan []*coreTypes.Vote, maxQueuedVotes),
 		queuedAgreements:           make(chan *coreTypes.AgreementResult, maxQueuedAgreements),
-		queuedRandomnesses:         make(chan *coreTypes.BlockRandomnessResult, maxQueuedRandomnesses),
+		queuedRandomnesses:         make(chan []*coreTypes.BlockRandomnessResult, maxQueuedRandomnesses),
 		queuedDKGPrivateShares:     make(chan *dkgTypes.PrivateShare, maxQueuedDKGPrivateShare),
 		queuedDKGPartialSignatures: make(chan *dkgTypes.PartialSignature, maxQueuedDKGParitialSignature),
 		queuedPullBlocks:           make(chan coreCommon.Hashes, maxQueuedPullBlocks),
@@ -232,26 +221,26 @@ func (p *peer) broadcast() {
 				return
 			}
 			p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
-		case block := <-p.queuedLatticeBlocks:
-			if err := p.SendLatticeBlock(block); err != nil {
+		case blocks := <-p.queuedLatticeBlocks:
+			if err := p.SendLatticeBlocks(blocks); err != nil {
 				return
 			}
-			p.Log().Trace("Broadcast lattice block")
-		case vote := <-p.queuedVotes:
-			if err := p.SendVote(vote); err != nil {
+			p.Log().Trace("Broadcast lattice blocks", "count", len(blocks))
+		case votes := <-p.queuedVotes:
+			if err := p.SendVotes(votes); err != nil {
 				return
 			}
-			p.Log().Trace("Broadcast vote", "vote", vote.String(), "hash", rlpHash(vote))
+			p.Log().Trace("Broadcast votes", "count", len(votes))
 		case agreement := <-p.queuedAgreements:
 			if err := p.SendAgreement(agreement); err != nil {
 				return
 			}
 			p.Log().Trace("Broadcast agreement")
-		case randomness := <-p.queuedRandomnesses:
-			if err := p.SendRandomness(randomness); err != nil {
+		case randomnesses := <-p.queuedRandomnesses:
+			if err := p.SendRandomnesses(randomnesses); err != nil {
 				return
 			}
-			p.Log().Trace("Broadcast randomness")
+			p.Log().Trace("Broadcast randomnesses", "count", len(randomnesses))
 		case privateShare := <-p.queuedDKGPrivateShares:
 			if err := p.SendDKGPrivateShare(privateShare); err != nil {
 				return
@@ -276,7 +265,7 @@ func (p *peer) broadcast() {
 			if err := p.SendPullRandomness(hashes); err != nil {
 				return
 			}
-			p.Log().Trace("Pulling Randomness", "hashes", hashes)
+			p.Log().Trace("Pulling Randomnesses", "hashes", hashes)
 		case <-p.term:
 			return
 		case <-time.After(100 * time.Millisecond):
@@ -354,6 +343,27 @@ func (p *peer) MarkNodeRecord(hash common.Hash) {
 	p.knownRecords.Add(hash)
 }
 
+func (p *peer) MarkAgreement(hash common.Hash) {
+	for p.knownAgreements.Cardinality() >= maxKnownAgreements {
+		p.knownAgreements.Pop()
+	}
+	p.knownAgreements.Add(hash)
+}
+
+func (p *peer) MarkRandomness(hash common.Hash) {
+	for p.knownRandomnesses.Cardinality() >= maxKnownRandomnesses {
+		p.knownRandomnesses.Pop()
+	}
+	p.knownRandomnesses.Add(hash)
+}
+
+func (p *peer) MarkDKGPrivateShares(hash common.Hash) {
+	for p.knownDKGPrivateShares.Cardinality() >= maxKnownDKGPrivateShares {
+		p.knownDKGPrivateShares.Pop()
+	}
+	p.knownDKGPrivateShares.Add(hash)
+}
+
 // SendTransactions sends transactions to the peer and includes the hashes
 // in its transaction hash set for future reference.
 func (p *peer) SendTransactions(txs types.Transactions) error {
@@ -442,29 +452,25 @@ func (p *peer) AsyncSendNewBlock(block *types.Block) {
 	}
 }
 
-func (p *peer) SendLatticeBlock(block *coreTypes.Block) error {
-	p.knownLatticeBlocks.Add(rlpHash(block))
-	return p2p.Send(p.rw, LatticeBlockMsg, block)
+func (p *peer) SendLatticeBlocks(blocks []*coreTypes.Block) error {
+	return p2p.Send(p.rw, LatticeBlockMsg, blocks)
 }
 
-func (p *peer) AsyncSendLatticeBlock(block *coreTypes.Block) {
+func (p *peer) AsyncSendLatticeBlocks(blocks []*coreTypes.Block) {
 	select {
-	case p.queuedLatticeBlocks <- block:
-		p.knownLatticeBlocks.Add(rlpHash(block))
+	case p.queuedLatticeBlocks <- blocks:
 	default:
 		p.Log().Debug("Dropping lattice block propagation")
 	}
 }
 
-func (p *peer) SendVote(vote *coreTypes.Vote) error {
-	p.knownVotes.Add(rlpHash(vote))
-	return p2p.Send(p.rw, VoteMsg, vote)
+func (p *peer) SendVotes(votes []*coreTypes.Vote) error {
+	return p2p.Send(p.rw, VoteMsg, votes)
 }
 
-func (p *peer) AsyncSendVote(vote *coreTypes.Vote) {
+func (p *peer) AsyncSendVotes(votes []*coreTypes.Vote) {
 	select {
-	case p.queuedVotes <- vote:
-		p.knownVotes.Add(rlpHash(vote))
+	case p.queuedVotes <- votes:
 	default:
 		p.Log().Debug("Dropping vote propagation")
 	}
@@ -484,15 +490,19 @@ func (p *peer) AsyncSendAgreement(agreement *coreTypes.AgreementResult) {
 	}
 }
 
-func (p *peer) SendRandomness(randomness *coreTypes.BlockRandomnessResult) error {
-	p.knownRandomnesses.Add(rlpHash(randomness))
-	return p2p.Send(p.rw, RandomnessMsg, randomness)
+func (p *peer) SendRandomnesses(randomnesses []*coreTypes.BlockRandomnessResult) error {
+	for _, randomness := range randomnesses {
+		p.knownRandomnesses.Add(rlpHash(randomness))
+	}
+	return p2p.Send(p.rw, RandomnessMsg, randomnesses)
 }
 
-func (p *peer) AsyncSendRandomness(randomness *coreTypes.BlockRandomnessResult) {
+func (p *peer) AsyncSendRandomnesses(randomnesses []*coreTypes.BlockRandomnessResult) {
 	select {
-	case p.queuedRandomnesses <- randomness:
-		p.knownRandomnesses.Add(rlpHash(randomness))
+	case p.queuedRandomnesses <- randomnesses:
+		for _, randomness := range randomnesses {
+			p.knownRandomnesses.Add(rlpHash(randomness))
+		}
 	default:
 		p.Log().Debug("Dropping randomness result")
 	}
@@ -513,14 +523,12 @@ func (p *peer) AsyncSendDKGPrivateShare(privateShare *dkgTypes.PrivateShare) {
 }
 
 func (p *peer) SendDKGPartialSignature(psig *dkgTypes.PartialSignature) error {
-	p.knownDKGPartialSignatures.Add(rlpHash(psig))
 	return p2p.Send(p.rw, DKGPartialSignatureMsg, psig)
 }
 
 func (p *peer) AsyncSendDKGPartialSignature(psig *dkgTypes.PartialSignature) {
 	select {
 	case p.queuedDKGPartialSignatures <- psig:
-		p.knownDKGPartialSignatures.Add(rlpHash(psig))
 	default:
 		p.Log().Debug("Dropping DKG partial signature")
 	}
@@ -854,21 +862,6 @@ func (ps *peerSet) PeersWithLabel(label peerLabel) []*peer {
 	return list
 }
 
-func (ps *peerSet) PeersWithoutVote(hash common.Hash, label peerLabel) []*peer {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-
-	list := make([]*peer, 0, len(ps.label2Nodes[label]))
-	for id := range ps.label2Nodes[label] {
-		if p, ok := ps.peers[id]; ok {
-			if !p.knownVotes.Contains(hash) {
-				list = append(list, p)
-			}
-		}
-	}
-	return list
-}
-
 // PeersWithoutNodeRecord retrieves a list of peers that do not have a
 // given record in their set of known hashes.
 func (ps *peerSet) PeersWithoutNodeRecord(hash common.Hash) []*peer {
@@ -883,18 +876,6 @@ func (ps *peerSet) PeersWithoutNodeRecord(hash common.Hash) []*peer {
 	return list
 }
 
-func (ps *peerSet) PeersWithoutLatticeBlock(hash common.Hash) []*peer {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-	list := make([]*peer, 0, len(ps.peers))
-	for _, p := range ps.peers {
-		if !p.knownLatticeBlocks.Contains(hash) {
-			list = append(list, p)
-		}
-	}
-	return list
-}
-
 func (ps *peerSet) PeersWithoutAgreement(hash common.Hash) []*peer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
@@ -919,12 +900,12 @@ func (ps *peerSet) PeersWithoutRandomness(hash common.Hash) []*peer {
 	return list
 }
 
-func (ps *peerSet) PeersWithoutDKGPartialSignature(hash common.Hash) []*peer {
+func (ps *peerSet) PeersWithoutDKGPrivateShares(hash common.Hash) []*peer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 	list := make([]*peer, 0, len(ps.peers))
 	for _, p := range ps.peers {
-		if !p.knownDKGPartialSignatures.Contains(hash) {
+		if !p.knownDKGPrivateShares.Contains(hash) {
 			list = append(list, p)
 		}
 	}
diff --git a/dex/protocol_test.go b/dex/protocol_test.go
index 74778cdda..aa06d7293 100644
--- a/dex/protocol_test.go
+++ b/dex/protocol_test.go
@@ -316,7 +316,7 @@ func TestSendNodeRecords(t *testing.T) {
 	wg.Wait()
 }
 
-func TestRecvLatticeBlock(t *testing.T) {
+func TestRecvLatticeBlocks(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	p, _ := newTestPeer("peer", dex64, pm, true)
 	defer pm.Stop()
@@ -355,7 +355,7 @@ func TestRecvLatticeBlock(t *testing.T) {
 		},
 	}
 
-	if err := p2p.Send(p.app, LatticeBlockMsg, &block); err != nil {
+	if err := p2p.Send(p.app, LatticeBlockMsg, []*coreTypes.Block{&block}); err != nil {
 		t.Fatalf("send error: %v", err)
 	}
 
@@ -371,7 +371,7 @@ func TestRecvLatticeBlock(t *testing.T) {
 	}
 }
 
-func TestSendLatticeBlock(t *testing.T) {
+func TestSendLatticeBlocks(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	p, _ := newTestPeer("peer", dex64, pm, true)
 	defer pm.Stop()
@@ -419,17 +419,17 @@ func TestSendLatticeBlock(t *testing.T) {
 		t.Errorf("%v: got code %d, want %d", p.Peer, msg.Code, LatticeBlockMsg)
 	}
 
-	var b coreTypes.Block
-	if err := msg.Decode(&b); err != nil {
+	var bs []*coreTypes.Block
+	if err := msg.Decode(&bs); err != nil {
 		t.Errorf("%v: %v", p.Peer, err)
 	}
 
-	if !reflect.DeepEqual(b, block) {
+	if !reflect.DeepEqual(bs, []*coreTypes.Block{&block}) {
 		t.Errorf("block mismatch")
 	}
 }
 
-func TestRecvVote(t *testing.T) {
+func TestRecvVotes(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	p, _ := newTestPeer("peer", dex64, pm, true)
 	defer pm.Stop()
@@ -451,7 +451,7 @@ func TestRecvVote(t *testing.T) {
 		},
 	}
 
-	if err := p2p.Send(p.app, VoteMsg, vote); err != nil {
+	if err := p2p.Send(p.app, VoteMsg, []*coreTypes.Vote{&vote}); err != nil {
 		t.Fatalf("send error: %v", err)
 	}
 
@@ -468,7 +468,7 @@ func TestRecvVote(t *testing.T) {
 	}
 }
 
-func TestSendVote(t *testing.T) {
+func TestSendVotes(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	defer pm.Stop()
 
@@ -509,16 +509,16 @@ func TestSendVote(t *testing.T) {
 			return
 		}
 
-		var v coreTypes.Vote
+		var vs []*coreTypes.Vote
 		if err != nil {
 			t.Errorf("%v: read error: %v", p.Peer, err)
 		} else if msg.Code != VoteMsg {
 			t.Errorf("%v: got code %d, want %d", p.Peer, msg.Code, VoteMsg)
 		}
-		if err := msg.Decode(&v); err != nil {
+		if err := msg.Decode(&vs); err != nil {
 			t.Errorf("%v: %v", p.Peer, err)
 		}
-		if !reflect.DeepEqual(v, vote) {
+		if !reflect.DeepEqual(vs, []*coreTypes.Vote{&vote}) {
 			t.Errorf("vote mismatch")
 		}
 	}
@@ -759,7 +759,7 @@ func TestSendAgreement(t *testing.T) {
 	}
 }
 
-func TestRecvRandomness(t *testing.T) {
+func TestRecvRandomnesses(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	p, _ := newTestPeer("peer", dex64, pm, true)
 	defer pm.Stop()
@@ -776,7 +776,7 @@ func TestRecvRandomness(t *testing.T) {
 		Randomness: []byte{7, 7, 7, 7},
 	}
 
-	if err := p2p.Send(p.app, RandomnessMsg, &randomness); err != nil {
+	if err := p2p.Send(p.app, RandomnessMsg, []*coreTypes.BlockRandomnessResult{&randomness}); err != nil {
 		t.Fatalf("send error: %v", err)
 	}
 
@@ -792,7 +792,7 @@ func TestRecvRandomness(t *testing.T) {
 	}
 }
 
-func TestSendRandomness(t *testing.T) {
+func TestSendRandomnesses(t *testing.T) {
 	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
 	p, _ := newTestPeer("peer", dex64, pm, true)
 	defer pm.Stop()
@@ -818,13 +818,13 @@ func TestSendRandomness(t *testing.T) {
 		t.Errorf("%v: got code %d, want %d", p.Peer, msg.Code, RandomnessMsg)
 	}
 
-	var r coreTypes.BlockRandomnessResult
-	if err := msg.Decode(&r); err != nil {
+	var rs []*coreTypes.BlockRandomnessResult
+	if err := msg.Decode(&rs); err != nil {
 		t.Errorf("%v: %v", p.Peer, err)
 	}
 
-	if !reflect.DeepEqual(r, randomness) {
-		t.Errorf("agreement mismatch")
+	if !reflect.DeepEqual(rs, []*coreTypes.BlockRandomnessResult{&randomness}) {
+		t.Errorf("randomness mismatch")
 	}
 }