dex: redesign p2p network topology

- Let p2p server support direct connection and group connection.
- Introduce node meta table to maintain IP of all nodes in node set,
  in memory and let nodes in the network can sync this table.
- Let peerSet able to manage direct connections to notary set and dkg set.

The mechanism to refresh the network topology when configuration round
change is not done yet.

21 files changed, 2423 insertions, 656 deletions

diff --git a/core/events.go b/core/events.go
index 02a083002..e76aa4784 100644
--- a/core/events.go
+++ b/core/events.go
@@ -46,3 +46,10 @@ type ChainSideEvent struct {
 }
 
 type ChainHeadEvent struct{ Block *types.Block }
+
+type NewNotarySetEvent struct {
+	Round   uint64
+	Pubkeys map[string]struct{} // pubkeys in hex format
+}
+
+type NewCRSEvent struct{ Round uint64 }
diff --git a/dex/handler.go b/dex/handler.go
index bc932fb28..5348b06f2 100644
--- a/dex/handler.go
+++ b/dex/handler.go
@@ -49,6 +49,8 @@ const (
 	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
+
+	metaChanSize = 10240
 )
 
 var (
@@ -59,11 +61,6 @@ var (
 // not compatible (low protocol version restrictions and high requirements).
 var errIncompatibleConfig = errors.New("incompatible configuration")
 
-type newNotarySetEvent struct {
-	Round uint64
-	Set   map[string]struct{}
-}
-
 func errResp(code errCode, format string, v ...interface{}) error {
 	return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...))
 }
@@ -75,32 +72,35 @@ type ProtocolManager struct {
 	acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing)
 
 	txpool      txPool
+	nodeTable   *nodeTable
+	gov         governance
 	blockchain  *core.BlockChain
 	chainconfig *params.ChainConfig
 	maxPeers    int
 
-	downloader       *downloader.Downloader
-	fetcher          *fetcher.Fetcher
-	peers            *peerSet
-	notarySetManager *notarySetManager
+	downloader *downloader.Downloader
+	fetcher    *fetcher.Fetcher
+	peers      *peerSet
 
 	SubProtocols []p2p.Protocol
 
 	eventMux      *event.TypeMux
 	txsCh         chan core.NewTxsEvent
 	txsSub        event.Subscription
+	metasCh       chan newMetasEvent
+	metasSub      event.Subscription
 	minedBlockSub *event.TypeMuxSubscription
 
 	// channels for fetcher, syncer, txsyncLoop
 	newPeerCh   chan *peer
 	txsyncCh    chan *txsync
+	metasyncCh  chan *metasync
 	quitSync    chan struct{}
 	noMorePeers chan struct{}
 
-	// channels for notarySetLoop
-	newNotarySetCh      chan newNotarySetEvent
-	newRoundCh          chan uint64
-	newNotaryNodeInfoCh chan *notaryNodeInfo
+	// channels for peerSetLoop
+	crsCh  chan core.NewCRSEvent
+	crsSub event.Subscription
 
 	srvr p2pServer
 
@@ -111,24 +111,26 @@ type ProtocolManager struct {
 
 // NewProtocolManager returns a new Ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
 // with the Ethereum network.
-func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkID uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
+func NewProtocolManager(
+	config *params.ChainConfig, mode downloader.SyncMode, networkID uint64,
+	mux *event.TypeMux, txpool txPool, engine consensus.Engine,
+	blockchain *core.BlockChain, chaindb ethdb.Database,
+	gov governance) (*ProtocolManager, error) {
+	tab := newNodeTable()
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
-		networkID:           networkID,
-		eventMux:            mux,
-		txpool:              txpool,
-		blockchain:          blockchain,
-		chainconfig:         config,
-		peers:               newPeerSet(),
-		newPeerCh:           make(chan *peer),
-		noMorePeers:         make(chan struct{}),
-		txsyncCh:            make(chan *txsync),
-		quitSync:            make(chan struct{}),
-		newNotarySetCh:      make(chan newNotarySetEvent),
-		newRoundCh:          make(chan uint64),
-		newNotaryNodeInfoCh: make(chan *notaryNodeInfo),
-	}
-	manager.notarySetManager = newNotarySetManager(manager.newNotaryNodeInfoCh)
+		networkID:   networkID,
+		eventMux:    mux,
+		txpool:      txpool,
+		nodeTable:   tab,
+		gov:         gov,
+		blockchain:  blockchain,
+		chainconfig: config,
+		newPeerCh:   make(chan *peer),
+		noMorePeers: make(chan struct{}),
+		txsyncCh:    make(chan *txsync),
+		quitSync:    make(chan struct{}),
+	}
 
 	// Figure out whether to allow fast sync or not
 	if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
@@ -216,22 +218,33 @@ func (pm *ProtocolManager) removePeer(id string) {
 func (pm *ProtocolManager) Start(srvr p2pServer, maxPeers int) {
 	pm.maxPeers = maxPeers
 	pm.srvr = srvr
+	pm.peers = newPeerSet(pm.gov, pm.srvr, pm.nodeTable)
+
+	// if our self in node set build the node info
 
 	// broadcast transactions
 	pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
 	pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)
 	go pm.txBroadcastLoop()
 
+	// broadcast node metas
+	pm.metasCh = make(chan newMetasEvent, metaChanSize)
+	pm.metasSub = pm.nodeTable.SubscribeNewMetasEvent(pm.metasCh)
+	go pm.metaBroadcastLoop()
+
 	// broadcast mined blocks
 	pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
 	go pm.minedBroadcastLoop()
 
-	// run the notary set loop
-	go pm.notarySetLoop()
+	// run the peer set loop
+	pm.crsCh = make(chan core.NewCRSEvent)
+	pm.crsSub = pm.gov.SubscribeNewCRSEvent(pm.crsCh)
+	go pm.peerSetLoop()
 
 	// start sync handlers
 	go pm.syncer()
 	go pm.txsyncLoop()
+	go pm.metasyncLoop()
 }
 
 func (pm *ProtocolManager) Stop() {
@@ -301,6 +314,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	// Propagate existing transactions. new transactions appearing
 	// after this will be sent via broadcasts.
 	pm.syncTransactions(p)
+	pm.syncNodeMetas(p)
 
 	// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork
 	if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {
@@ -699,18 +713,18 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		pm.txpool.AddRemotes(txs)
 
-	case msg.Code == NotaryNodeInfoMsg:
-		var info notaryNodeInfo
-		if err := msg.Decode(&info); err != nil {
+	case msg.Code == MetaMsg:
+		var metas []*NodeMeta
+		if err := msg.Decode(&metas); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-
-		// TODO(sonic): validate signature
-		p.MarkNotaryNodeInfo(info.Hash())
-		// Broadcast to peers
-		pm.BroadcastNotaryNodeInfo(&info)
-
-		pm.notarySetManager.TryAddInfo(&info)
+		for i, meta := range metas {
+			if meta == nil {
+				return errResp(ErrDecode, "node meta %d is nil", i)
+			}
+			p.MarkNodeMeta(meta.Hash())
+		}
+		pm.nodeTable.Add(metas)
 	default:
 		return errResp(ErrInvalidMsgCode, "%v", msg.Code)
 	}
@@ -769,11 +783,20 @@ func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) {
 	}
 }
 
-// BroadcastNotaryNodeInfo will propagate notary node info to its peers.
-func (pm *ProtocolManager) BroadcastNotaryNodeInfo(info *notaryNodeInfo) {
-	peers := pm.peers.PeersWithoutNotaryNodeInfo(info.Hash())
-	for _, peer := range peers {
-		peer.AsyncSendNotaryNodeInfo(info)
+// BroadcastMetas will propagate node metas to its peers.
+func (pm *ProtocolManager) BroadcastMetas(metas []*NodeMeta) {
+	var metaset = make(map[*peer][]*NodeMeta)
+
+	for _, meta := range metas {
+		peers := pm.peers.PeersWithoutNodeMeta(meta.Hash())
+		for _, peer := range peers {
+			metaset[peer] = append(metaset[peer], meta)
+		}
+		log.Trace("Broadcast meta", "ID", meta.ID, "recipients", len(peers))
+	}
+
+	for peer, metas := range metaset {
+		peer.AsyncSendNodeMetas(metas)
 	}
 }
 
@@ -801,97 +824,32 @@ func (pm *ProtocolManager) txBroadcastLoop() {
 	}
 }
 
-// a loop keep building and maintaining peers in notary set.
-func (pm *ProtocolManager) notarySetLoop() {
-	// TODO:
-	//   * pm need to know current round, and whether we are in notary set.
-	//   * (done) able to handle node info in the near future round.
-	//   * check peer limit.
-	//   * revisit channels (newNotarySetCh, newRoundCh, newNotaryNodeInfoCh) (buffered ?)
-
-	// If we are in notary set and we are synced with network.
-	// events:
-	//   * new notary set is determined, and we are in notary set.
-	//     (TBD: subscribe the event or polling govornance)
-	//     - advance round
-	//     - build new notary set
-	//     - remove old notary set, remove old notary set from static
-
-	//   * current round node info changed.
-
-	self := pm.srvr.Self()
-	var round uint64
-
+func (pm *ProtocolManager) metaBroadcastLoop() {
 	for {
 		select {
-		case event := <-pm.newNotarySetCh:
-			// new notary set is determined.
-			if _, ok := event.Set[self.ID.String()]; ok {
-				// initialize the new notary set and add it to pm.notarySetManager
-				s := newNotarySet(event.Round, event.Set)
-				pm.notarySetManager.Register(event.Round, s)
-
-				// TODO: handle signature
-				pm.BroadcastNotaryNodeInfo(&notaryNodeInfo{
-					ID:        self.ID,
-					IP:        self.IP,
-					UDP:       self.UDP,
-					TCP:       self.TCP,
-					Round:     event.Round,
-					Timestamp: time.Now().Unix(),
-				})
-			}
-
-		case r := <-pm.newRoundCh:
-			// move to new round.
-			round = r
+		case event := <-pm.metasCh:
+			pm.BroadcastMetas(event.Metas)
 
-			// TODO: revisit this to make sure how many previous round's
-			// notary set we need to keep.
-
-			// rmove notary set before current round, they are useless
-			for _, s := range pm.notarySetManager.Before(round) {
-				pm.removeNotarySetFromStatic(s)
-			}
-
-			// prepare notary set connections for new round.
-			if pm.isInNotarySet(round + 1) {
-				// we assume the notary set for round + 1 is added to
-				// pm.notarySetManager before this step.
-				if n, ok := pm.notarySetManager.Round(round + 1); ok {
-					pm.addNotarySetToStatic(n)
-				}
-			}
-
-		case <-pm.newNotaryNodeInfoCh:
-			// some node info in "current round" is updated.
-			// try to connect them by the new info.
-			if pm.isInNotarySet(round) {
-				if n, ok := pm.notarySetManager.Round(round); ok {
-					pm.addNotarySetToStatic(n)
-				}
-			}
-
-		case <-pm.quitSync:
+		// Err() channel will be closed when unsubscribing.
+		case <-pm.metasSub.Err():
 			return
 		}
 	}
 }
 
-func (pm *ProtocolManager) isInNotarySet(r uint64) bool {
-	return false
-}
-
-func (pm *ProtocolManager) addNotarySetToStatic(n *notarySet) {
-	for _, node := range n.NodesToAdd() {
-		pm.srvr.AddNotaryPeer(node)
-		n.MarkAdded(node.ID.String())
-	}
-}
-
-func (pm *ProtocolManager) removeNotarySetFromStatic(n *notarySet) {
-	for _, node := range n.Nodes() {
-		pm.srvr.RemoveNotaryPeer(node)
+// a loop keep building and maintaining peers in notary set.
+// TODO: finish this
+func (pm *ProtocolManager) peerSetLoop() {
+	for {
+		select {
+		case event := <-pm.crsCh:
+			pm.peers.BuildNotaryConn(event.Round)
+			pm.peers.BuildDKGConn(event.Round)
+			pm.peers.ForgetNotaryConn(event.Round - 1)
+			pm.peers.ForgetDKGConn(event.Round - 1)
+		case <-pm.quitSync:
+			return
+		}
 	}
 }
 
diff --git a/dex/handler_test.go b/dex/handler_test.go
new file mode 100644
index 000000000..981362bb5
--- /dev/null
+++ b/dex/handler_test.go
@@ -0,0 +1,445 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package dex
+
+import (
+	"math"
+	"math/big"
+	"math/rand"
+	"testing"
+
+	"github.com/dexon-foundation/dexon/common"
+	"github.com/dexon-foundation/dexon/core"
+	"github.com/dexon-foundation/dexon/core/state"
+	"github.com/dexon-foundation/dexon/core/types"
+	"github.com/dexon-foundation/dexon/crypto"
+	"github.com/dexon-foundation/dexon/eth/downloader"
+	"github.com/dexon-foundation/dexon/ethdb"
+	"github.com/dexon-foundation/dexon/p2p"
+	"github.com/dexon-foundation/dexon/params"
+)
+
+// Tests that protocol versions and modes of operations are matched up properly.
+func TestProtocolCompatibility(t *testing.T) {
+	// Define the compatibility chart
+	tests := []struct {
+		version    uint
+		mode       downloader.SyncMode
+		compatible bool
+	}{
+		{61, downloader.FullSync, true}, {62, downloader.FullSync, true}, {63, downloader.FullSync, true},
+		{61, downloader.FastSync, true}, {62, downloader.FastSync, true}, {63, downloader.FastSync, true},
+	}
+	// Make sure anything we screw up is restored
+	backup := ProtocolVersions
+	defer func() { ProtocolVersions = backup }()
+
+	// Try all available compatibility configs and check for errors
+	for i, tt := range tests {
+		ProtocolVersions = []uint{tt.version}
+
+		pm, _, err := newTestProtocolManager(tt.mode, 0, nil, nil)
+		if pm != nil {
+			defer pm.Stop()
+		}
+		if (err == nil && !tt.compatible) || (err != nil && tt.compatible) {
+			t.Errorf("test %d: compatibility mismatch: have error %v, want compatibility %v", i, err, tt.compatible)
+		}
+	}
+}
+
+// Tests that block headers can be retrieved from a remote chain based on user queries.
+func TestGetBlockHeaders62(t *testing.T) { testGetBlockHeaders(t, 62) }
+func TestGetBlockHeaders63(t *testing.T) { testGetBlockHeaders(t, 63) }
+
+func testGetBlockHeaders(t *testing.T, protocol int) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, downloader.MaxHashFetch+15, nil, nil)
+	peer, _ := newTestPeer("peer", protocol, pm, true)
+	defer peer.close()
+
+	// Create a "random" unknown hash for testing
+	var unknown common.Hash
+	for i := range unknown {
+		unknown[i] = byte(i)
+	}
+	// Create a batch of tests for various scenarios
+	limit := uint64(downloader.MaxHeaderFetch)
+	tests := []struct {
+		query  *getBlockHeadersData // The query to execute for header retrieval
+		expect []common.Hash        // The hashes of the block whose headers are expected
+	}{
+		// A single random block should be retrievable by hash and number too
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Hash: pm.blockchain.GetBlockByNumber(limit / 2).Hash()}, Amount: 1},
+			[]common.Hash{pm.blockchain.GetBlockByNumber(limit / 2).Hash()},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 1},
+			[]common.Hash{pm.blockchain.GetBlockByNumber(limit / 2).Hash()},
+		},
+		// Multiple headers should be retrievable in both directions
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(limit / 2).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 + 1).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 + 2).Hash(),
+			},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(limit / 2).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 - 1).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 - 2).Hash(),
+			},
+		},
+		// Multiple headers with skip lists should be retrievable
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(limit / 2).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 + 4).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 + 8).Hash(),
+			},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(limit / 2).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 - 4).Hash(),
+				pm.blockchain.GetBlockByNumber(limit/2 - 8).Hash(),
+			},
+		},
+		// The chain endpoints should be retrievable
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: 0}, Amount: 1},
+			[]common.Hash{pm.blockchain.GetBlockByNumber(0).Hash()},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: pm.blockchain.CurrentBlock().NumberU64()}, Amount: 1},
+			[]common.Hash{pm.blockchain.CurrentBlock().Hash()},
+		},
+		// Ensure protocol limits are honored
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: pm.blockchain.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true},
+			pm.blockchain.GetBlockHashesFromHash(pm.blockchain.CurrentBlock().Hash(), limit),
+		},
+		// Check that requesting more than available is handled gracefully
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: pm.blockchain.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(pm.blockchain.CurrentBlock().NumberU64() - 4).Hash(),
+				pm.blockchain.GetBlockByNumber(pm.blockchain.CurrentBlock().NumberU64()).Hash(),
+			},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(4).Hash(),
+				pm.blockchain.GetBlockByNumber(0).Hash(),
+			},
+		},
+		// Check that requesting more than available is handled gracefully, even if mid skip
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: pm.blockchain.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(pm.blockchain.CurrentBlock().NumberU64() - 4).Hash(),
+				pm.blockchain.GetBlockByNumber(pm.blockchain.CurrentBlock().NumberU64() - 1).Hash(),
+			},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(4).Hash(),
+				pm.blockchain.GetBlockByNumber(1).Hash(),
+			},
+		},
+		// Check a corner case where requesting more can iterate past the endpoints
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Number: 2}, Amount: 5, Reverse: true},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(2).Hash(),
+				pm.blockchain.GetBlockByNumber(1).Hash(),
+				pm.blockchain.GetBlockByNumber(0).Hash(),
+			},
+		},
+		// Check a corner case where skipping overflow loops back into the chain start
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Hash: pm.blockchain.GetBlockByNumber(3).Hash()}, Amount: 2, Reverse: false, Skip: math.MaxUint64 - 1},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(3).Hash(),
+			},
+		},
+		// Check a corner case where skipping overflow loops back to the same header
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Hash: pm.blockchain.GetBlockByNumber(1).Hash()}, Amount: 2, Reverse: false, Skip: math.MaxUint64},
+			[]common.Hash{
+				pm.blockchain.GetBlockByNumber(1).Hash(),
+			},
+		},
+		// Check that non existing headers aren't returned
+		{
+			&getBlockHeadersData{Origin: hashOrNumber{Hash: unknown}, Amount: 1},
+			[]common.Hash{},
+		}, {
+			&getBlockHeadersData{Origin: hashOrNumber{Number: pm.blockchain.CurrentBlock().NumberU64() + 1}, Amount: 1},
+			[]common.Hash{},
+		},
+	}
+	// Run each of the tests and verify the results against the chain
+	for i, tt := range tests {
+		// Collect the headers to expect in the response
+		headers := []*types.Header{}
+		for _, hash := range tt.expect {
+			headers = append(headers, pm.blockchain.GetBlockByHash(hash).Header())
+		}
+		// Send the hash request and verify the response
+		p2p.Send(peer.app, 0x03, tt.query)
+		if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
+			t.Errorf("test %d: headers mismatch: %v", i, err)
+		}
+		// If the test used number origins, repeat with hashes as the too
+		if tt.query.Origin.Hash == (common.Hash{}) {
+			if origin := pm.blockchain.GetBlockByNumber(tt.query.Origin.Number); origin != nil {
+				tt.query.Origin.Hash, tt.query.Origin.Number = origin.Hash(), 0
+
+				p2p.Send(peer.app, 0x03, tt.query)
+				if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
+					t.Errorf("test %d: headers mismatch: %v", i, err)
+				}
+			}
+		}
+	}
+}
+
+// Tests that block contents can be retrieved from a remote chain based on their hashes.
+func TestGetBlockBodies62(t *testing.T) { testGetBlockBodies(t, 62) }
+func TestGetBlockBodies63(t *testing.T) { testGetBlockBodies(t, 63) }
+
+func testGetBlockBodies(t *testing.T, protocol int) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, downloader.MaxBlockFetch+15, nil, nil)
+	peer, _ := newTestPeer("peer", protocol, pm, true)
+	defer peer.close()
+
+	// Create a batch of tests for various scenarios
+	limit := downloader.MaxBlockFetch
+	tests := []struct {
+		random    int           // Number of blocks to fetch randomly from the chain
+		explicit  []common.Hash // Explicitly requested blocks
+		available []bool        // Availability of explicitly requested blocks
+		expected  int           // Total number of existing blocks to expect
+	}{
+		{1, nil, nil, 1},                                                         // A single random block should be retrievable
+		{10, nil, nil, 10},                                                       // Multiple random blocks should be retrievable
+		{limit, nil, nil, limit},                                                 // The maximum possible blocks should be retrievable
+		{limit + 1, nil, nil, limit},                                             // No more than the possible block count should be returned
+		{0, []common.Hash{pm.blockchain.Genesis().Hash()}, []bool{true}, 1},      // The genesis block should be retrievable
+		{0, []common.Hash{pm.blockchain.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable
+		{0, []common.Hash{{}}, []bool{false}, 0},                                 // A non existent block should not be returned
+
+		// Existing and non-existing blocks interleaved should not cause problems
+		{0, []common.Hash{
+			{},
+			pm.blockchain.GetBlockByNumber(1).Hash(),
+			{},
+			pm.blockchain.GetBlockByNumber(10).Hash(),
+			{},
+			pm.blockchain.GetBlockByNumber(100).Hash(),
+			{},
+		}, []bool{false, true, false, true, false, true, false}, 3},
+	}
+	// Run each of the tests and verify the results against the chain
+	for i, tt := range tests {
+		// Collect the hashes to request, and the response to expect
+		hashes, seen := []common.Hash{}, make(map[int64]bool)
+		bodies := []*blockBody{}
+
+		for j := 0; j < tt.random; j++ {
+			for {
+				num := rand.Int63n(int64(pm.blockchain.CurrentBlock().NumberU64()))
+				if !seen[num] {
+					seen[num] = true
+
+					block := pm.blockchain.GetBlockByNumber(uint64(num))
+					hashes = append(hashes, block.Hash())
+					if len(bodies) < tt.expected {
+						bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()})
+					}
+					break
+				}
+			}
+		}
+		for j, hash := range tt.explicit {
+			hashes = append(hashes, hash)
+			if tt.available[j] && len(bodies) < tt.expected {
+				block := pm.blockchain.GetBlockByHash(hash)
+				bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()})
+			}
+		}
+		// Send the hash request and verify the response
+		p2p.Send(peer.app, 0x05, hashes)
+		if err := p2p.ExpectMsg(peer.app, 0x06, bodies); err != nil {
+			t.Errorf("test %d: bodies mismatch: %v", i, err)
+		}
+	}
+}
+
+// Tests that the node state database can be retrieved based on hashes.
+func TestGetNodeData63(t *testing.T) { testGetNodeData(t, 63) }
+
+func testGetNodeData(t *testing.T, protocol int) {
+	// Define three accounts to simulate transactions with
+	acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
+	acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
+	acc1Addr := crypto.PubkeyToAddress(acc1Key.PublicKey)
+	acc2Addr := crypto.PubkeyToAddress(acc2Key.PublicKey)
+
+	signer := types.HomesteadSigner{}
+	// Create a chain generator with some simple transactions (blatantly stolen from @fjl/chain_markets_test)
+	generator := func(i int, block *core.BlockGen) {
+		switch i {
+		case 0:
+			// In block 1, the test bank sends account #1 some ether.
+			tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, testBankKey)
+			block.AddTx(tx)
+		case 1:
+			// In block 2, the test bank sends some more ether to account #1.
+			// acc1Addr passes it on to account #2.
+			tx1, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey)
+			tx2, _ := types.SignTx(types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, acc1Key)
+			block.AddTx(tx1)
+			block.AddTx(tx2)
+		case 2:
+			// Block 3 is empty but was mined by account #2.
+			block.SetCoinbase(acc2Addr)
+			block.SetExtra([]byte("yeehaw"))
+		case 3:
+			// Block 4 includes blocks 2 and 3 as uncle headers (with modified extra data).
+			b2 := block.PrevBlock(1).Header()
+			b2.Extra = []byte("foo")
+			block.AddUncle(b2)
+			b3 := block.PrevBlock(2).Header()
+			b3.Extra = []byte("foo")
+			block.AddUncle(b3)
+		}
+	}
+	// Assemble the test environment
+	pm, db := newTestProtocolManagerMust(t, downloader.FullSync, 4, generator, nil)
+	peer, _ := newTestPeer("peer", protocol, pm, true)
+	defer peer.close()
+
+	// Fetch for now the entire chain db
+	hashes := []common.Hash{}
+	for _, key := range db.Keys() {
+		if len(key) == len(common.Hash{}) {
+			hashes = append(hashes, common.BytesToHash(key))
+		}
+	}
+	p2p.Send(peer.app, 0x0d, hashes)
+	msg, err := peer.app.ReadMsg()
+	if err != nil {
+		t.Fatalf("failed to read node data response: %v", err)
+	}
+	if msg.Code != 0x0e {
+		t.Fatalf("response packet code mismatch: have %x, want %x", msg.Code, 0x0c)
+	}
+	var data [][]byte
+	if err := msg.Decode(&data); err != nil {
+		t.Fatalf("failed to decode response node data: %v", err)
+	}
+	// Verify that all hashes correspond to the requested data, and reconstruct a state tree
+	for i, want := range hashes {
+		if hash := crypto.Keccak256Hash(data[i]); hash != want {
+			t.Errorf("data hash mismatch: have %x, want %x", hash, want)
+		}
+	}
+	statedb := ethdb.NewMemDatabase()
+	for i := 0; i < len(data); i++ {
+		statedb.Put(hashes[i].Bytes(), data[i])
+	}
+	accounts := []common.Address{testBank, acc1Addr, acc2Addr}
+	for i := uint64(0); i <= pm.blockchain.CurrentBlock().NumberU64(); i++ {
+		trie, _ := state.New(pm.blockchain.GetBlockByNumber(i).Root(), state.NewDatabase(statedb))
+
+		for j, acc := range accounts {
+			state, _ := pm.blockchain.State()
+			bw := state.GetBalance(acc)
+			bh := trie.GetBalance(acc)
+
+			if (bw != nil && bh == nil) || (bw == nil && bh != nil) {
+				t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw)
+			}
+			if bw != nil && bh != nil && bw.Cmp(bw) != 0 {
+				t.Errorf("test %d, account %d: balance mismatch: have %v, want %v", i, j, bh, bw)
+			}
+		}
+	}
+}
+
+// Tests that the transaction receipts can be retrieved based on hashes.
+func TestGetReceipt63(t *testing.T) { testGetReceipt(t, 63) }
+
+func testGetReceipt(t *testing.T, protocol int) {
+	// Define three accounts to simulate transactions with
+	acc1Key, _ := crypto.HexToECDSA("8a1f9a8f95be41cd7ccb6168179afb4504aefe388d1e14474d32c45c72ce7b7a")
+	acc2Key, _ := crypto.HexToECDSA("49a7b37aa6f6645917e7b807e9d1c00d4fa71f18343b0d4122a4d2df64dd6fee")
+	acc1Addr := crypto.PubkeyToAddress(acc1Key.PublicKey)
+	acc2Addr := crypto.PubkeyToAddress(acc2Key.PublicKey)
+
+	signer := types.HomesteadSigner{}
+	// Create a chain generator with some simple transactions (blatantly stolen from @fjl/chain_markets_test)
+	generator := func(i int, block *core.BlockGen) {
+		switch i {
+		case 0:
+			// In block 1, the test bank sends account #1 some ether.
+			tx, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), signer, testBankKey)
+			block.AddTx(tx)
+		case 1:
+			// In block 2, the test bank sends some more ether to account #1.
+			// acc1Addr passes it on to account #2.
+			tx1, _ := types.SignTx(types.NewTransaction(block.TxNonce(testBank), acc1Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, testBankKey)
+			tx2, _ := types.SignTx(types.NewTransaction(block.TxNonce(acc1Addr), acc2Addr, big.NewInt(1000), params.TxGas, nil, nil), signer, acc1Key)
+			block.AddTx(tx1)
+			block.AddTx(tx2)
+		case 2:
+			// Block 3 is empty but was mined by account #2.
+			block.SetCoinbase(acc2Addr)
+			block.SetExtra([]byte("yeehaw"))
+		case 3:
+			// Block 4 includes blocks 2 and 3 as uncle headers (with modified extra data).
+			b2 := block.PrevBlock(1).Header()
+			b2.Extra = []byte("foo")
+			block.AddUncle(b2)
+			b3 := block.PrevBlock(2).Header()
+			b3.Extra = []byte("foo")
+			block.AddUncle(b3)
+		}
+	}
+	// Assemble the test environment
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 4, generator, nil)
+	peer, _ := newTestPeer("peer", protocol, pm, true)
+	defer peer.close()
+
+	// Collect the hashes to request, and the response to expect
+	hashes, receipts := []common.Hash{}, []types.Receipts{}
+	for i := uint64(0); i <= pm.blockchain.CurrentBlock().NumberU64(); i++ {
+		block := pm.blockchain.GetBlockByNumber(i)
+
+		hashes = append(hashes, block.Hash())
+		receipts = append(receipts, pm.blockchain.GetReceiptsByHash(block.Hash()))
+	}
+	// Send the hash request and verify the response
+	p2p.Send(peer.app, 0x0f, hashes)
+	if err := p2p.ExpectMsg(peer.app, 0x10, receipts); err != nil {
+		t.Errorf("receipts mismatch: %v", err)
+	}
+}
diff --git a/dex/helper_test.go b/dex/helper_test.go
index 7e3479958..dcda6f4d2 100644
--- a/dex/helper_test.go
+++ b/dex/helper_test.go
@@ -37,7 +37,7 @@ import (
 	"github.com/dexon-foundation/dexon/ethdb"
 	"github.com/dexon-foundation/dexon/event"
 	"github.com/dexon-foundation/dexon/p2p"
-	"github.com/dexon-foundation/dexon/p2p/discover"
+	"github.com/dexon-foundation/dexon/p2p/enode"
 	"github.com/dexon-foundation/dexon/params"
 )
 
@@ -48,24 +48,47 @@ var (
 
 // testP2PServer is a fake, helper p2p server for testing purposes.
 type testP2PServer struct {
-	added   chan *discover.Node
-	removed chan *discover.Node
+	mu     sync.Mutex
+	self   *enode.Node
+	direct map[enode.NodeID]*enode.Node
+	group  map[string][]*enode.Node
 }
 
-func (s *testP2PServer) Self() *discover.Node {
-	return &discover.Node{}
+func newTestP2PServer(self *enode.Node) *testP2PServer {
+	return &testP2PServer{
+		self:   self,
+		direct: make(map[enode.NodeID]*enode.Node),
+		group:  make(map[string][]*enode.Node),
+	}
 }
 
-func (s *testP2PServer) AddNotaryPeer(node *discover.Node) {
-	if s.added != nil {
-		s.added <- node
-	}
+func (s *testP2PServer) Self() *enode.Node {
+	return s.self
 }
 
-func (s *testP2PServer) RemoveNotaryPeer(node *discover.Node) {
-	if s.removed != nil {
-		s.removed <- node
-	}
+func (s *testP2PServer) AddDirectPeer(node *enode.Node) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.direct[node.ID] = node
+}
+
+func (s *testP2PServer) RemoveDirectPeer(node *enode.Node) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.direct, node.ID)
+}
+
+func (s *testP2PServer) AddGroup(
+	name string, nodes []*enode.Node, num uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.group[name] = nodes
+}
+
+func (s *testP2PServer) RemoveGroup(name string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.group, name)
 }
 
 // newTestProtocolManager creates a new protocol manager for testing purposes,
@@ -88,11 +111,11 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func
 		panic(err)
 	}
 
-	pm, err := NewProtocolManager(gspec.Config, mode, DefaultConfig.NetworkId, evmux, &testTxPool{added: newtx}, engine, blockchain, db)
+	pm, err := NewProtocolManager(gspec.Config, mode, DefaultConfig.NetworkId, evmux, &testTxPool{added: newtx}, engine, blockchain, db, &testGovernance{})
 	if err != nil {
 		return nil, nil, err
 	}
-	pm.Start(&testP2PServer{}, 1000)
+	pm.Start(newTestP2PServer(&enode.Node{}), 1000)
 	return pm, db, nil
 }
 
@@ -157,6 +180,29 @@ func newTestTransaction(from *ecdsa.PrivateKey, nonce uint64, datasize int) *typ
 	return tx
 }
 
+// testGovernance is a fake, helper governance for testing purposes
+type testGovernance struct {
+	getChainNumFunc  func(uint64) uint32
+	getNotarySetFunc func(uint32, uint64) map[string]struct{}
+	getDKGSetFunc    func(uint64) map[string]struct{}
+}
+
+func (g *testGovernance) GetChainNum(round uint64) uint32 {
+	return g.getChainNumFunc(round)
+}
+
+func (g *testGovernance) GetNotarySet(chainID uint32, round uint64) map[string]struct{} {
+	return g.getNotarySetFunc(chainID, round)
+}
+
+func (g *testGovernance) GetDKGSet(round uint64) map[string]struct{} {
+	return g.getDKGSetFunc(round)
+}
+
+func (g *testGovernance) SubscribeNewCRSEvent(ch chan core.NewCRSEvent) event.Subscription {
+	return nil
+}
+
 // testPeer is a simulated peer to allow testing direct network calls.
 type testPeer struct {
 	net p2p.MsgReadWriter // Network layer reader/writer to simulate remote messaging
@@ -170,7 +216,7 @@ func newTestPeer(name string, version int, pm *ProtocolManager, shake bool) (*te
 	app, net := p2p.MsgPipe()
 
 	// Generate a random id and create the peer
-	var id discover.NodeID
+	var id enode.NodeID
 	rand.Read(id[:])
 
 	peer := pm.newPeer(version, p2p.NewPeer(id, name, nil), net)
diff --git a/dex/network.go b/dex/network.go
index c3a0ef792..b19b46147 100644
--- a/dex/network.go
+++ b/dex/network.go
@@ -23,14 +23,6 @@ func (n *DexconNetwork) BroadcastVote(vote *types.Vote) {
 func (n *DexconNetwork) BroadcastBlock(block *types.Block) {
 }
 
-// BroadcastRandomnessRequest broadcasts rand request to DKG set.
-func (n *DexconNetwork) BroadcastRandomnessRequest(randRequest *types.AgreementResult) {
-}
-
-// BroadcastRandomnessResult broadcasts rand request to Notary set.
-func (n *DexconNetwork) BroadcastRandomnessResult(randResult *types.BlockRandomnessResult) {
-}
-
 // SendDKGPrivateShare sends PrivateShare to a DKG participant.
 func (n *DexconNetwork) SendDKGPrivateShare(
 	pub crypto.PublicKey, prvShare *types.DKGPrivateShare) {
@@ -47,6 +39,15 @@ func (n *DexconNetwork) BroadcastDKGPartialSignature(
 	psig *types.DKGPartialSignature) {
 }
 
+// BroadcastRandomnessRequest broadcasts rand request to DKG set.
+func (n *DexconNetwork) BroadcastRandomnessRequest(randRequest *types.AgreementResult) {
+
+}
+
+// BroadcastRandomnessResult broadcasts rand request to Notary set.
+func (n *DexconNetwork) BroadcastRandomnessResult(randResult *types.BlockRandomnessResult) {
+}
+
 // ReceiveChan returns a channel to receive messages from DEXON network.
 func (n *DexconNetwork) ReceiveChan() <-chan interface{} {
 	return n.receiveChan
diff --git a/dex/nodetable.go b/dex/nodetable.go
new file mode 100644
index 000000000..929b168a8
--- /dev/null
+++ b/dex/nodetable.go
@@ -0,0 +1,79 @@
+package dex
+
+import (
+	"net"
+	"sync"
+
+	"github.com/dexon-foundation/dexon/common"
+	"github.com/dexon-foundation/dexon/crypto/sha3"
+	"github.com/dexon-foundation/dexon/event"
+	"github.com/dexon-foundation/dexon/p2p/enode"
+	"github.com/dexon-foundation/dexon/rlp"
+)
+
+type NodeMeta struct {
+	ID        enode.ID
+	IP        net.IP
+	UDP       int
+	TCP       int
+	Timestamp uint64
+	Sig       []byte
+}
+
+func (n *NodeMeta) Hash() (h common.Hash) {
+	hw := sha3.NewKeccak256()
+	rlp.Encode(hw, n)
+	hw.Sum(h[:0])
+	return h
+}
+
+type newMetasEvent struct{ Metas []*NodeMeta }
+
+type nodeTable struct {
+	mu    sync.RWMutex
+	entry map[enode.ID]*NodeMeta
+	feed  event.Feed
+}
+
+func newNodeTable() *nodeTable {
+	return &nodeTable{
+		entry: make(map[enode.ID]*NodeMeta),
+	}
+}
+
+func (t *nodeTable) Get(id enode.ID) *NodeMeta {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.entry[id]
+}
+
+func (t *nodeTable) Add(metas []*NodeMeta) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	var newMetas []*NodeMeta
+	for _, meta := range metas {
+		// TODO: validate the meta
+		if e, ok := t.entry[meta.ID]; ok && e.Timestamp > meta.Timestamp {
+			continue
+		}
+		t.entry[meta.ID] = meta
+		newMetas = append(newMetas, meta)
+	}
+	t.feed.Send(newMetasEvent{newMetas})
+}
+
+func (t *nodeTable) Metas() []*NodeMeta {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	metas := make([]*NodeMeta, 0, len(t.entry))
+	for _, meta := range t.entry {
+		metas = append(metas, meta)
+	}
+	return metas
+}
+
+func (t *nodeTable) SubscribeNewMetasEvent(
+	ch chan<- newMetasEvent) event.Subscription {
+	return t.feed.Subscribe(ch)
+}
diff --git a/dex/nodetable_test.go b/dex/nodetable_test.go
new file mode 100644
index 000000000..5b3f7de57
--- /dev/null
+++ b/dex/nodetable_test.go
@@ -0,0 +1,93 @@
+package dex
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/dexon-foundation/dexon/common"
+	"github.com/dexon-foundation/dexon/p2p/enode"
+)
+
+func TestNodeTable(t *testing.T) {
+	table := newNodeTable()
+	ch := make(chan newMetasEvent)
+	table.SubscribeNewMetasEvent(ch)
+
+	metas1 := []*NodeMeta{
+		&NodeMeta{ID: randomID()},
+		&NodeMeta{ID: randomID()},
+	}
+
+	metas2 := []*NodeMeta{
+		&NodeMeta{ID: randomID()},
+		&NodeMeta{ID: randomID()},
+	}
+
+	go table.Add(metas1)
+
+	select {
+	case newMetas := <-ch:
+		m := map[common.Hash]struct{}{}
+		for _, meta := range newMetas.Metas {
+			m[meta.Hash()] = struct{}{}
+		}
+
+		if len(m) != len(metas1) {
+			t.Errorf("len mismatch: got %d, want: %d",
+				len(m), len(metas1))
+		}
+
+		for _, meta := range metas1 {
+			if _, ok := m[meta.Hash()]; !ok {
+				t.Errorf("expected meta (%s) not exists", meta.Hash())
+			}
+		}
+	case <-time.After(1 * time.Second):
+		t.Error("did not receive new metas event within one second")
+	}
+
+	go table.Add(metas2)
+	select {
+	case newMetas := <-ch:
+		m := map[common.Hash]struct{}{}
+		for _, meta := range newMetas.Metas {
+			m[meta.Hash()] = struct{}{}
+		}
+
+		if len(m) != len(metas1) {
+			t.Errorf("len mismatch: got %d, want: %d",
+				len(m), len(metas2))
+		}
+
+		for _, meta := range metas2 {
+			if _, ok := m[meta.Hash()]; !ok {
+				t.Errorf("expected meta (%s) not exists", meta.Hash())
+			}
+		}
+	case <-time.After(1 * time.Second):
+		t.Error("did not receive new metas event within one second")
+	}
+
+	var metas []*NodeMeta
+	metas = append(metas, metas1...)
+	metas = append(metas, metas2...)
+	allMetas := table.Metas()
+	if len(allMetas) != len(metas) {
+		t.Errorf("all metas num mismatch: got %d, want %d",
+			len(metas), len(allMetas))
+	}
+
+	for _, m := range metas {
+		if m.Hash() != table.Get(m.ID).Hash() {
+			t.Errorf("meta (%s) mismatch", m.ID.String())
+		}
+	}
+}
+
+func randomID() (id enode.ID) {
+	for i := range id {
+		id[i] = byte(rand.Intn(255))
+	}
+	return id
+}
diff --git a/dex/notaryset.go b/dex/notaryset.go
deleted file mode 100644
index 74d259314..000000000
--- a/dex/notaryset.go
+++ /dev/null
@@ -1,203 +0,0 @@
-package dex
-
-import (
-	"fmt"
-	"sync"
-
-	"github.com/dexon-foundation/dexon/p2p/discover"
-)
-
-type nodeInfo struct {
-	info  *notaryNodeInfo
-	added bool
-}
-
-func (n *nodeInfo) NewNode() *discover.Node {
-	return discover.NewNode(n.info.ID, n.info.IP, n.info.UDP, n.info.TCP)
-}
-
-type notarySet struct {
-	round uint64
-	m     map[string]*nodeInfo
-	lock  sync.RWMutex
-}
-
-func newNotarySet(round uint64, s map[string]struct{}) *notarySet {
-	m := make(map[string]*nodeInfo)
-	for nodeID := range s {
-		m[nodeID] = &nodeInfo{}
-	}
-
-	return &notarySet{
-		round: round,
-		m:     m,
-	}
-}
-
-// Call this function when the notaryNodeInfoMsg is received.
-func (n *notarySet) AddInfo(info *notaryNodeInfo) error {
-	n.lock.Lock()
-	defer n.lock.Unlock()
-
-	// check round
-	if info.Round != n.round {
-		return fmt.Errorf("invalid round")
-	}
-
-	nInfo, ok := n.m[info.ID.String()]
-	if !ok {
-		return fmt.Errorf("not in notary set")
-	}
-
-	// if the info exists check timstamp
-	if nInfo.info != nil {
-		if nInfo.info.Timestamp > info.Timestamp {
-			return fmt.Errorf("old msg")
-		}
-	}
-
-	nInfo.info = info
-	return nil
-}
-
-// MarkAdded mark the notary node as added
-// to prevent duplcate addition in the future.
-func (n *notarySet) MarkAdded(nodeID string) {
-	if info, ok := n.m[nodeID]; ok {
-		info.added = true
-	}
-}
-
-// Return all nodes
-func (n *notarySet) Nodes() []*discover.Node {
-	n.lock.RLock()
-	defer n.lock.RUnlock()
-
-	list := make([]*discover.Node, 0, len(n.m))
-	for _, info := range n.m {
-		list = append(list, info.NewNode())
-	}
-	return list
-}
-
-// Return nodes that need to be added to p2p server as notary node.
-func (n *notarySet) NodesToAdd() []*discover.Node {
-	n.lock.RLock()
-	defer n.lock.RUnlock()
-
-	var list []*discover.Node
-	for _, info := range n.m {
-		// craete a new discover.Node
-		if !info.added {
-			list = append(list, info.NewNode())
-		}
-	}
-	return list
-}
-
-type notarySetManager struct {
-	m                   map[uint64]*notarySet
-	lock                sync.RWMutex
-	queued              map[uint64]map[string]*notaryNodeInfo
-	round               uint64 // biggest round of managed notary sets
-	newNotaryNodeInfoCh chan *notaryNodeInfo
-}
-
-func newNotarySetManager(
-	newNotaryNodeInfoCh chan *notaryNodeInfo) *notarySetManager {
-	return &notarySetManager{
-		m:                   make(map[uint64]*notarySet),
-		queued:              make(map[uint64]map[string]*notaryNodeInfo),
-		newNotaryNodeInfoCh: newNotaryNodeInfoCh,
-	}
-}
-
-// Register injects a new notary set into the manager and
-// processes the queued info.
-func (n *notarySetManager) Register(r uint64, s *notarySet) {
-	n.lock.Lock()
-	defer n.lock.Unlock()
-	if r > n.round {
-		n.round = r
-	}
-	n.m[r] = s
-	n.processQueuedInfo()
-}
-
-// Unregister removes the notary set of the given round.
-func (n *notarySetManager) Unregister(r uint64) {
-	n.lock.Lock()
-	defer n.lock.Unlock()
-	delete(n.m, r)
-}
-
-// Round returns the notary set of the given round.
-func (n *notarySetManager) Round(r uint64) (*notarySet, bool) {
-	n.lock.RLock()
-	defer n.lock.RUnlock()
-	s, ok := n.m[r]
-	return s, ok
-}
-
-// Before returns all the notary sets that before the given round.
-func (n *notarySetManager) Before(r uint64) []*notarySet {
-	n.lock.RLock()
-	defer n.lock.RUnlock()
-	var list []*notarySet
-	for round, s := range n.m {
-		if round < r {
-			list = append(list, s)
-		}
-	}
-	return list
-}
-
-// TryAddInfo associates the given info to the notary set if the notary set is
-// managed by the manager.
-// If the notary node info is belong to future notary set, queue the info.
-func (n *notarySetManager) TryAddInfo(info *notaryNodeInfo) {
-	n.lock.Lock()
-	defer n.lock.Unlock()
-	n.tryAddInfo(info)
-}
-
-// This function is extract for calling without lock.
-// Make sure the caller already accquired the lock.
-func (n *notarySetManager) tryAddInfo(info *notaryNodeInfo) {
-	if info.Round > n.round {
-		if q, ok := n.queued[info.Round]; ok {
-			q[info.Hash().String()] = info
-			return
-		}
-		n.queued[info.Round] = map[string]*notaryNodeInfo{
-			info.Hash().String(): info,
-		}
-		return
-	}
-
-	s, ok := n.Round(info.Round)
-	if !ok {
-		return
-	}
-	s.AddInfo(info)
-
-	// TODO(sonic): handle timeout
-	n.newNotaryNodeInfoCh <- info
-}
-
-func (n *notarySetManager) processQueuedInfo() {
-	n.lock.Lock()
-	defer n.lock.Unlock()
-	if q, ok := n.queued[n.round]; ok {
-		for _, info := range q {
-			n.tryAddInfo(info)
-		}
-	}
-
-	// Clear queue infos before current round.
-	for round := range n.queued {
-		if round <= n.round {
-			delete(n.queued, round)
-		}
-	}
-}
diff --git a/dex/peer.go b/dex/peer.go
index f1a4335d1..8c218f1d9 100644
--- a/dex/peer.go
+++ b/dex/peer.go
@@ -27,6 +27,7 @@ import (
 	"github.com/dexon-foundation/dexon/common"
 	"github.com/dexon-foundation/dexon/core/types"
 	"github.com/dexon-foundation/dexon/p2p"
+	"github.com/dexon-foundation/dexon/p2p/discover"
 	"github.com/dexon-foundation/dexon/rlp"
 )
 
@@ -34,19 +35,20 @@ var (
 	errClosed            = errors.New("peer set is closed")
 	errAlreadyRegistered = errors.New("peer is already registered")
 	errNotRegistered     = errors.New("peer is not registered")
-	errInvalidRound      = errors.New("invalid round")
 )
 
 const (
 	maxKnownTxs    = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS)
+	maxKnownMetas  = 32768 // Maximum metas hashes to keep in the known list (prevent DOS)
 	maxKnownBlocks = 1024  // Maximum block hashes to keep in the known list (prevent DOS)
-	maxKnownInfos  = 1024
 
 	// maxQueuedTxs is the maximum number of transaction lists to queue up before
 	// dropping broadcasts. This is a sensitive number as a transaction list might
 	// contain a single transaction, or thousands.
 	maxQueuedTxs = 128
 
+	maxQueuedMetas = 512
+
 	// maxQueuedProps is the maximum number of block propagations to queue up before
 	// dropping broadcasts. There's not much point in queueing stale blocks, so a few
 	// that might cover uncles should be enough.
@@ -57,9 +59,9 @@ const (
 	// above some healthy uncle limit, so use that.
 	maxQueuedAnns = 4
 
-	maxQueuedInfos = 1024
-
 	handshakeTimeout = 5 * time.Second
+
+	groupNodeNum = 3
 )
 
 // PeerInfo represents a short summary of the Ethereum sub-protocol metadata known
@@ -76,13 +78,27 @@ type propEvent struct {
 	td    *big.Int
 }
 
+type setType uint32
+
+const (
+	dkgset = iota
+	notaryset
+)
+
+type peerLabel struct {
+	set     setType
+	chainID uint32
+	round   uint64
+}
+
 type peer struct {
 	id string
 
 	*p2p.Peer
 	rw p2p.MsgReadWriter
 
-	version  int         // Protocol version negotiated
+	version  int // Protocol version negotiated
+	labels   mapset.Set
 	forkDrop *time.Timer // Timed connection dropper if forks aren't validated in time
 
 	head common.Hash
@@ -90,12 +106,12 @@ type peer struct {
 	lock sync.RWMutex
 
 	knownTxs    mapset.Set                // Set of transaction hashes known to be known by this peer
+	knownMetas  mapset.Set                // Set of node metas known to be known by this peer
 	knownBlocks mapset.Set                // Set of block hashes known to be known by this peer
-	knownInfos  mapset.Set                // Set of infos known to be known by this peer
 	queuedTxs   chan []*types.Transaction // Queue of transactions to broadcast to the peer
+	queuedMetas chan []*NodeMeta          // Queue of node metas to broadcast to the peer
 	queuedProps chan *propEvent           // Queue of blocks to broadcast to the peer
 	queuedAnns  chan *types.Block         // Queue of blocks to announce to the peer
-	queuedInfos chan *notaryNodeInfo      // Queue of infos to broadcast to the peer
 	term        chan struct{}             // Termination channel to stop the broadcaster
 }
 
@@ -104,20 +120,21 @@ func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
 		Peer:        p,
 		rw:          rw,
 		version:     version,
-		id:          fmt.Sprintf("%x", p.ID().Bytes()[:8]),
+		labels:      mapset.NewSet(),
+		id:          p.ID().String(),
 		knownTxs:    mapset.NewSet(),
+		knownMetas:  mapset.NewSet(),
 		knownBlocks: mapset.NewSet(),
-		knownInfos:  mapset.NewSet(),
 		queuedTxs:   make(chan []*types.Transaction, maxQueuedTxs),
+		queuedMetas: make(chan []*NodeMeta, maxQueuedMetas),
 		queuedProps: make(chan *propEvent, maxQueuedProps),
 		queuedAnns:  make(chan *types.Block, maxQueuedAnns),
-		queuedInfos: make(chan *notaryNodeInfo, maxQueuedInfos),
 		term:        make(chan struct{}),
 	}
 }
 
 // broadcast is a write loop that multiplexes block propagations, announcements,
-// transaction and notary node infos broadcasts into the remote peer.
+// transaction and notary node metas broadcasts into the remote peer.
 // The goal is to have an async writer that does not lock up node internals.
 func (p *peer) broadcast() {
 	for {
@@ -128,6 +145,12 @@ func (p *peer) broadcast() {
 			}
 			p.Log().Trace("Broadcast transactions", "count", len(txs))
 
+		case metas := <-p.queuedMetas:
+			if err := p.SendNodeMetas(metas); err != nil {
+				return
+			}
+			p.Log().Trace("Broadcast node metas", "count", len(metas))
+
 		case prop := <-p.queuedProps:
 			if err := p.SendNewBlock(prop.block, prop.td); err != nil {
 				return
@@ -140,12 +163,6 @@ func (p *peer) broadcast() {
 			}
 			p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
 
-		case info := <-p.queuedInfos:
-			if err := p.SendNotaryNodeInfo(info); err != nil {
-				return
-			}
-			p.Log().Trace("Broadcast notary node info")
-
 		case <-p.term:
 			return
 		}
@@ -157,6 +174,14 @@ func (p *peer) close() {
 	close(p.term)
 }
 
+func (p *peer) addLabel(label peerLabel) {
+	p.labels.Add(label)
+}
+
+func (p *peer) removeLabel(label peerLabel) {
+	p.labels.Remove(label)
+}
+
 // Info gathers and returns a collection of metadata known about a peer.
 func (p *peer) Info() *PeerInfo {
 	hash, td := p.Head()
@@ -207,11 +232,11 @@ func (p *peer) MarkTransaction(hash common.Hash) {
 	p.knownTxs.Add(hash)
 }
 
-func (p *peer) MarkNotaryNodeInfo(hash common.Hash) {
-	for p.knownInfos.Cardinality() >= maxKnownInfos {
-		p.knownInfos.Pop()
+func (p *peer) MarkNodeMeta(hash common.Hash) {
+	for p.knownMetas.Cardinality() >= maxKnownMetas {
+		p.knownMetas.Pop()
 	}
-	p.knownInfos.Add(hash)
+	p.knownMetas.Add(hash)
 }
 
 // SendTransactions sends transactions to the peer and includes the hashes
@@ -236,21 +261,26 @@ func (p *peer) AsyncSendTransactions(txs []*types.Transaction) {
 	}
 }
 
-// SendNotaryNodeInfo sends the info to the peer and includes the hashes
-// in its info hash set for future reference.
-func (p *peer) SendNotaryNodeInfo(info *notaryNodeInfo) error {
-	return p2p.Send(p.rw, NotaryNodeInfoMsg, info)
+// SendNodeMetas sends the metas to the peer and includes the hashes
+// in its metas hash set for future reference.
+func (p *peer) SendNodeMetas(metas []*NodeMeta) error {
+	for _, meta := range metas {
+		p.knownMetas.Add(meta.Hash())
+	}
+	return p2p.Send(p.rw, MetaMsg, metas)
 }
 
-// AsyncSendNotaryNodeInfo queues list of notary node info propagation to a
+// AsyncSendNodeMeta queues list of notary node meta propagation to a
 // remote peer. If the peer's broadcast queue is full, the event is silently
 // dropped.
-func (p *peer) AsyncSendNotaryNodeInfo(info *notaryNodeInfo) {
+func (p *peer) AsyncSendNodeMetas(metas []*NodeMeta) {
 	select {
-	case p.queuedInfos <- info:
-		p.knownInfos.Add(info.Hash())
+	case p.queuedMetas <- metas:
+		for _, meta := range metas {
+			p.knownMetas.Add(meta.Hash())
+		}
 	default:
-		p.Log().Debug("Dropping notary node info propagation")
+		p.Log().Debug("Dropping node meta propagation", "count", len(metas))
 	}
 }
 
@@ -431,7 +461,7 @@ func (p *peer) readStatus(network uint64, status *statusData, genesis common.Has
 // String implements fmt.Stringer.
 func (p *peer) String() string {
 	return fmt.Sprintf("Peer %s [%s]", p.id,
-		fmt.Sprintf("eth/%2d", p.version),
+		fmt.Sprintf("dex/%2d", p.version),
 	)
 }
 
@@ -441,18 +471,25 @@ type peerSet struct {
 	peers  map[string]*peer
 	lock   sync.RWMutex
 	closed bool
+	tab    *nodeTable
 
-	// TODO(sonic): revist this map after dexon core SDK is finalized.
-	// use types.ValidatorID? or implement Stringer for types.ValidatorID
-	notaryPeers map[uint64]map[string]*peer
-	round       uint64
+	srvr          p2pServer
+	gov           governance
+	peerLabels    map[string]map[peerLabel]struct{}
+	notaryHistory map[uint64]struct{}
+	dkgHistory    map[uint64]struct{}
 }
 
 // newPeerSet creates a new peer set to track the active participants.
-func newPeerSet() *peerSet {
+func newPeerSet(gov governance, srvr p2pServer, tab *nodeTable) *peerSet {
 	return &peerSet{
-		peers:       make(map[string]*peer),
-		notaryPeers: make(map[uint64]map[string]*peer),
+		peers:         make(map[string]*peer),
+		gov:           gov,
+		srvr:          srvr,
+		tab:           tab,
+		peerLabels:    make(map[string]map[peerLabel]struct{}),
+		notaryHistory: make(map[uint64]struct{}),
+		dkgHistory:    make(map[uint64]struct{}),
 	}
 }
 
@@ -537,14 +574,14 @@ func (ps *peerSet) PeersWithoutTx(hash common.Hash) []*peer {
 	return list
 }
 
-// PeersWithoutNotaryNodeInfo retrieves a list of peers that do not have a
-// given info in their set of known hashes.
-func (ps *peerSet) PeersWithoutNotaryNodeInfo(hash common.Hash) []*peer {
+// PeersWithoutNodeMeta retrieves a list of peers that do not have a
+// given meta in their set of known hashes.
+func (ps *peerSet) PeersWithoutNodeMeta(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.knownInfos.Contains(hash) {
+		if !p.knownMetas.Contains(hash) {
 			list = append(list, p)
 		}
 	}
@@ -580,44 +617,171 @@ func (ps *peerSet) Close() {
 	ps.closed = true
 }
 
-// AddNotaryPeer adds a peer into notary peer of the given round.
-// Caller of this function need to make sure that the peer is acutally in
-// notary set.
-func (ps *peerSet) AddNotaryPeer(round uint64, p *peer) error {
+func (ps *peerSet) BuildNotaryConn(round uint64) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	// TODO(sonic): revisit this round check after dexon core SDK is finalized.
-	if round < ps.round || round > ps.round+2 {
-		return errInvalidRound
+	if _, ok := ps.notaryHistory[round]; ok {
+		return
 	}
 
-	if _, ok := ps.peers[p.id]; !ok {
-		return errNotRegistered
+	ps.notaryHistory[round] = struct{}{}
+
+	selfID := ps.srvr.Self().ID.String()
+	for chainID := uint32(0); chainID < ps.gov.GetChainNum(round); chainID++ {
+		s := ps.gov.GetNotarySet(chainID, round)
+
+		// not in notary set, add group
+		if _, ok := s[selfID]; !ok {
+			var nodes []*discover.Node
+			for id := range s {
+				nodes = append(nodes, ps.newNode(id))
+			}
+			ps.srvr.AddGroup(notarySetName(chainID, round), nodes, groupNodeNum)
+			continue
+		}
+
+		label := peerLabel{
+			set:     notaryset,
+			chainID: chainID,
+			round:   round,
+		}
+		delete(s, selfID)
+		for id := range s {
+			ps.addDirectPeer(id, label)
+		}
 	}
+}
 
-	ps.notaryPeers[round][p.id] = p
-	return nil
+func (ps *peerSet) ForgetNotaryConn(round uint64) {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	// forget all the rounds before the given round
+	for r := range ps.notaryHistory {
+		if r <= round {
+			ps.forgetNotaryConn(r)
+			delete(ps.notaryHistory, r)
+		}
+	}
 }
 
-// NotaryPeers return peers in notary set of the given round.
-func (ps *peerSet) NotaryPeers(round uint64) []*peer {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
+func (ps *peerSet) forgetNotaryConn(round uint64) {
+	selfID := ps.srvr.Self().ID.String()
+	for chainID := uint32(0); chainID < ps.gov.GetChainNum(round); chainID++ {
+		s := ps.gov.GetNotarySet(chainID, round)
+		if _, ok := s[selfID]; !ok {
+			ps.srvr.RemoveGroup(notarySetName(chainID, round))
+			continue
+		}
 
-	list := make([]*peer, 0, len(ps.notaryPeers[round]))
-	for _, p := range ps.notaryPeers[round] {
-		if _, ok := ps.peers[p.id]; ok {
-			list = append(list, p)
+		label := peerLabel{
+			set:     notaryset,
+			chainID: chainID,
+			round:   round,
+		}
+		delete(s, selfID)
+		for id := range s {
+			ps.removeDirectPeer(id, label)
 		}
 	}
-	return list
 }
 
-// NextRound moves notary peer set to next round.
-func (ps *peerSet) NextRound() {
+func notarySetName(chainID uint32, round uint64) string {
+	return fmt.Sprintf("%d-%d-notaryset", chainID, round)
+}
+
+func (ps *peerSet) BuildDKGConn(round uint64) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
-	delete(ps.notaryPeers, ps.round)
-	ps.round = ps.round + 1
+	selfID := ps.srvr.Self().ID.String()
+	s := ps.gov.GetDKGSet(round)
+	if _, ok := s[selfID]; !ok {
+		return
+	}
+	ps.dkgHistory[round] = struct{}{}
+
+	delete(s, selfID)
+	for id := range s {
+		ps.addDirectPeer(id, peerLabel{
+			set:   dkgset,
+			round: round,
+		})
+	}
+}
+
+func (ps *peerSet) ForgetDKGConn(round uint64) {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	// forget all the rounds before the given round
+	for r := range ps.dkgHistory {
+		if r <= round {
+			ps.forgetDKGConn(r)
+			delete(ps.dkgHistory, r)
+		}
+	}
+}
+
+func (ps *peerSet) forgetDKGConn(round uint64) {
+	selfID := ps.srvr.Self().ID.String()
+	s := ps.gov.GetDKGSet(round)
+	if _, ok := s[selfID]; !ok {
+		return
+	}
+
+	delete(s, selfID)
+	label := peerLabel{
+		set:   dkgset,
+		round: round,
+	}
+	for id := range s {
+		ps.removeDirectPeer(id, label)
+	}
+}
+
+// make sure the ps.lock is hold
+func (ps *peerSet) addDirectPeer(id string, label peerLabel) {
+	// if the peer exists add the label
+	if p, ok := ps.peers[id]; ok {
+		p.addLabel(label)
+	}
+
+	if _, ok := ps.peerLabels[id]; !ok {
+		ps.peerLabels[id] = make(map[peerLabel]struct{})
+	}
+
+	ps.peerLabels[id][label] = struct{}{}
+	ps.srvr.AddDirectPeer(ps.newNode(id))
+}
+
+// make sure the ps.lock is hold
+func (ps *peerSet) removeDirectPeer(id string, label peerLabel) {
+	if p, ok := ps.peers[id]; ok {
+		p.removeLabel(label)
+	}
+
+	delete(ps.peerLabels[id], label)
+
+	if len(ps.peerLabels[id]) == 0 {
+		ps.srvr.RemoveDirectPeer(ps.newNode(id))
+		delete(ps.peerLabels, id)
+	}
+}
+
+func (ps *peerSet) newNode(id string) *enode.Node {
+	nodeID := enode.HexID(id)
+	meta := ps.tab.Get(enode.HexID(id))
+
+	var r enr.Record
+	r.Set(enr.ID(nodeID.String()))
+	r.Set(enr.IP(meta.IP))
+	r.Set(enr.TCP(meta.TCP))
+	r.Set(enr.UDP(meta.UDP))
+
+	n, err := enode.New(enode.ValidSchemes, &r)
+	if err != nil {
+		panic(err)
+	}
+	return n
 }
diff --git a/dex/peer_test.go b/dex/peer_test.go
new file mode 100644
index 000000000..bac6ed5ec
--- /dev/null
+++ b/dex/peer_test.go
@@ -0,0 +1,628 @@
+package dex
+
+import (
+	"fmt"
+	"math/big"
+	"testing"
+
+	mapset "github.com/deckarep/golang-set"
+	"github.com/dexon-foundation/dexon/p2p/discover"
+	"github.com/dexon-foundation/dexon/p2p/enode"
+)
+
+func TestPeerSetBuildAndForgetNotaryConn(t *testing.T) {
+	self := discover.Node{ID: nodeID(0)}
+	server := newTestP2PServer(&self)
+	table := newNodeTable()
+
+	gov := &testGovernance{
+		getChainNumFunc: func(uint64) uint32 {
+			return 3
+		},
+	}
+
+	round10 := [][]enode.ID{
+		[]enode.ID{nodeID(0), nodeID(1), nodeID(2)},
+		[]enode.ID{nodeID(1), nodeID(3)},
+		[]enode.ID{nodeID(2), nodeID(4)},
+	}
+	round11 := [][]enode.ID{
+		[]enode.ID{nodeID(0), nodeID(1), nodeID(5)},
+		[]enode.ID{nodeID(5), nodeID(6)},
+		[]enode.ID{nodeID(0), nodeID(2), nodeID(4)},
+	}
+	round12 := [][]enode.ID{
+		[]enode.ID{nodeID(0), nodeID(3), nodeID(5)},
+		[]enode.ID{nodeID(0), nodeID(7), nodeID(8)},
+		[]enode.ID{nodeID(0), nodeID(2), nodeID(6)},
+	}
+
+	gov.getNotarySetFunc = func(cid uint32, round uint64) map[string]struct{} {
+		m := map[uint64][][]enode.ID{
+			10: round10,
+			11: round11,
+			12: round12,
+		}
+		return newTestNodeSet(m[round][cid])
+	}
+
+	ps := newPeerSet(gov, server, table)
+	peer1 := newDummyPeer(nodeID(1))
+	peer2 := newDummyPeer(nodeID(2))
+	var err error
+	err = ps.Register(peer1)
+	if err != nil {
+		t.Error(err)
+	}
+	err = ps.Register(peer2)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 10
+	ps.BuildNotaryConn(10)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10}, notaryset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkGroup(server, []string{
+		notarySetName(1, 10),
+		notarySetName(2, 10),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 11
+	ps.BuildNotaryConn(11)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+		peerLabel{notaryset, 0, 11},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+		peerLabel{notaryset, 2, 11},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+			peerLabel{notaryset, 0, 11},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+			peerLabel{notaryset, 2, 11},
+		},
+		nodeID(4).String(): []peerLabel{
+			peerLabel{notaryset, 2, 11},
+		},
+		nodeID(5).String(): []peerLabel{
+			peerLabel{notaryset, 0, 11},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10, 11}, notaryset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2), nodeID(4), nodeID(5),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkGroup(server, []string{
+		notarySetName(1, 10),
+		notarySetName(2, 10),
+		notarySetName(1, 11),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 12
+	ps.BuildNotaryConn(12)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+		peerLabel{notaryset, 0, 11},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{notaryset, 0, 10},
+		peerLabel{notaryset, 2, 11},
+		peerLabel{notaryset, 2, 12},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+			peerLabel{notaryset, 0, 11},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{notaryset, 0, 10},
+			peerLabel{notaryset, 2, 11},
+			peerLabel{notaryset, 2, 12},
+		},
+		nodeID(3).String(): []peerLabel{
+			peerLabel{notaryset, 0, 12},
+		},
+		nodeID(4).String(): []peerLabel{
+			peerLabel{notaryset, 2, 11},
+		},
+		nodeID(5).String(): []peerLabel{
+			peerLabel{notaryset, 0, 11},
+			peerLabel{notaryset, 0, 12},
+		},
+		nodeID(6).String(): []peerLabel{
+			peerLabel{notaryset, 2, 12},
+		},
+		nodeID(7).String(): []peerLabel{
+			peerLabel{notaryset, 1, 12},
+		},
+		nodeID(8).String(): []peerLabel{
+			peerLabel{notaryset, 1, 12},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10, 11, 12}, notaryset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2), nodeID(3), nodeID(4),
+		nodeID(5), nodeID(6), nodeID(7), nodeID(8),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkGroup(server, []string{
+		notarySetName(1, 10),
+		notarySetName(2, 10),
+		notarySetName(1, 11),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// forget round 11
+	ps.ForgetNotaryConn(11)
+
+	err = checkLabels(peer1, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{notaryset, 2, 12},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(2).String(): []peerLabel{
+			peerLabel{notaryset, 2, 12},
+		},
+		nodeID(3).String(): []peerLabel{
+			peerLabel{notaryset, 0, 12},
+		},
+		nodeID(5).String(): []peerLabel{
+			peerLabel{notaryset, 0, 12},
+		},
+		nodeID(6).String(): []peerLabel{
+			peerLabel{notaryset, 2, 12},
+		},
+		nodeID(7).String(): []peerLabel{
+			peerLabel{notaryset, 1, 12},
+		},
+		nodeID(8).String(): []peerLabel{
+			peerLabel{notaryset, 1, 12},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{12}, notaryset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(2), nodeID(3),
+		nodeID(5), nodeID(6), nodeID(7), nodeID(8),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkGroup(server, []string{})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// forget round 12
+	ps.ForgetNotaryConn(12)
+	err = checkLabels(peer1, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{}, notaryset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkGroup(server, []string{})
+	if err != nil {
+		t.Error(err)
+	}
+
+}
+
+func TestPeerSetBuildDKGConn(t *testing.T) {
+	self := discover.Node{ID: nodeID(0)}
+	server := newTestP2PServer(&self)
+	table := newNodeTable()
+
+	gov := &testGovernance{}
+
+	gov.getDKGSetFunc = func(round uint64) map[string]struct{} {
+		m := map[uint64][]enode.ID{
+			10: []enode.ID{nodeID(0), nodeID(1), nodeID(2)},
+			11: []enode.ID{nodeID(1), nodeID(2), nodeID(5)},
+			12: []enode.ID{nodeID(0), nodeID(3), nodeID(5)},
+		}
+		return newTestNodeSet(m[round])
+	}
+
+	ps := newPeerSet(gov, server, table)
+	peer1 := newDummyPeer(nodeID(1))
+	peer2 := newDummyPeer(nodeID(2))
+	var err error
+	err = ps.Register(peer1)
+	if err != nil {
+		t.Error(err)
+	}
+	err = ps.Register(peer2)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 10
+	ps.BuildDKGConn(10)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10}, dkgset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 11
+	ps.BuildDKGConn(11)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10}, dkgset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// build round 12
+	ps.BuildDKGConn(12)
+
+	err = checkLabels(peer1, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{
+		peerLabel{dkgset, 0, 10},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(1).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+		nodeID(2).String(): []peerLabel{
+			peerLabel{dkgset, 0, 10},
+		},
+		nodeID(3).String(): []peerLabel{
+			peerLabel{dkgset, 0, 12},
+		},
+		nodeID(5).String(): []peerLabel{
+			peerLabel{dkgset, 0, 12},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{10, 12}, dkgset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(1), nodeID(2), nodeID(3), nodeID(5),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// forget round 11
+	ps.ForgetDKGConn(11)
+
+	err = checkLabels(peer1, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{
+		nodeID(3).String(): []peerLabel{
+			peerLabel{dkgset, 0, 12},
+		},
+		nodeID(5).String(): []peerLabel{
+			peerLabel{dkgset, 0, 12},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{12}, dkgset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{
+		nodeID(3), nodeID(5),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	// forget round 12
+	ps.ForgetDKGConn(12)
+	err = checkLabels(peer1, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkLabels(peer2, []peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerLabels(ps, map[string][]peerLabel{})
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkPeerSetHistory(ps, []uint64{}, dkgset)
+	if err != nil {
+		t.Error(err)
+	}
+	err = checkDirectPeer(server, []enode.ID{})
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func checkLabels(p *peer, want []peerLabel) error {
+	if p.labels.Cardinality() != len(want) {
+		return fmt.Errorf("num of labels mismatch: got %d, want %d",
+			p.labels.Cardinality(), len(want))
+	}
+
+	for _, label := range want {
+		if !p.labels.Contains(label) {
+			return fmt.Errorf("label %+v not exist", label)
+		}
+	}
+	return nil
+}
+
+func checkPeerLabels(ps *peerSet, want map[string][]peerLabel) error {
+	if len(ps.peerLabels) != len(want) {
+		return fmt.Errorf("peer num mismatch: got %d, want %d",
+			len(ps.peerLabels), len(want))
+	}
+
+	for peerID, gotLabels := range ps.peerLabels {
+		wantLabels, ok := want[peerID]
+		if !ok {
+			return fmt.Errorf("peer id %s not exists", peerID)
+		}
+
+		if len(gotLabels) != len(wantLabels) {
+			return fmt.Errorf(
+				"num of labels of peer id %s mismatch: got %d, want %d",
+				peerID, len(gotLabels), len(wantLabels))
+		}
+
+		for _, label := range wantLabels {
+			if _, ok := gotLabels[label]; !ok {
+				fmt.Errorf("label: %+v not exists", label)
+			}
+		}
+	}
+	return nil
+}
+
+func checkPeerSetHistory(ps *peerSet, want []uint64, set setType) error {
+	var history map[uint64]struct{}
+	switch set {
+	case notaryset:
+		history = ps.notaryHistory
+	case dkgset:
+		history = ps.dkgHistory
+	default:
+		return fmt.Errorf("invalid set: %d", set)
+	}
+
+	if len(history) != len(want) {
+		return fmt.Errorf("num of history mismatch: got %d, want %d",
+			len(history), len(want))
+	}
+
+	for _, r := range want {
+		if _, ok := history[r]; !ok {
+			return fmt.Errorf("round %d not exists", r)
+		}
+	}
+	return nil
+}
+
+func checkDirectPeer(srvr *testP2PServer, want []enode.ID) error {
+	if len(srvr.direct) != len(want) {
+		return fmt.Errorf("num of direct peer mismatch: got %d, want %d",
+			len(srvr.direct), len(want))
+	}
+
+	for _, id := range want {
+		if _, ok := srvr.direct[id]; !ok {
+			return fmt.Errorf("direct peer %s not exists", id.String())
+		}
+	}
+	return nil
+}
+func checkGroup(srvr *testP2PServer, want []string) error {
+	if len(srvr.group) != len(want) {
+		return fmt.Errorf("num of group mismatch: got %d, want %d",
+			len(srvr.group), len(want))
+	}
+
+	for _, name := range want {
+		if _, ok := srvr.group[name]; !ok {
+			return fmt.Errorf("group %s not exists", name)
+		}
+	}
+	return nil
+}
+
+func nodeID(n int64) enode.ID {
+	b := big.NewInt(n).Bytes()
+	var id enode.ID
+	copy(id[len(id)-len(b):], b)
+	return id
+}
+
+func newTestNodeSet(nodes []enode.ID) map[string]struct{} {
+	m := make(map[string]struct{})
+	for _, node := range nodes {
+		m[node.String()] = struct{}{}
+	}
+	return m
+}
+
+func newDummyPeer(id enode.ID) *peer {
+	return &peer{
+		labels: mapset.NewSet(),
+		id:     id.String(),
+	}
+}
diff --git a/dex/protocol.go b/dex/protocol.go
index 0111edf18..7b01217ff 100644
--- a/dex/protocol.go
+++ b/dex/protocol.go
@@ -20,14 +20,11 @@ import (
 	"fmt"
 	"io"
 	"math/big"
-	"net"
 
 	"github.com/dexon-foundation/dexon/common"
 	"github.com/dexon-foundation/dexon/core"
 	"github.com/dexon-foundation/dexon/core/types"
-	"github.com/dexon-foundation/dexon/crypto/sha3"
 	"github.com/dexon-foundation/dexon/event"
-	"github.com/dexon-foundation/dexon/p2p/discover"
 	"github.com/dexon-foundation/dexon/p2p/enode"
 	"github.com/dexon-foundation/dexon/rlp"
 )
@@ -67,7 +64,7 @@ const (
 	ReceiptsMsg    = 0x10
 
 	// Protocol messages belonging to dex/64
-	NotaryNodeInfoMsg = 0x11
+	MetaMsg = 0x11
 )
 
 type errCode int
@@ -114,12 +111,26 @@ type txPool interface {
 	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
 }
 
+type governance interface {
+	GetChainNum(uint64) uint32
+
+	GetNotarySet(uint32, uint64) map[string]struct{}
+
+	GetDKGSet(uint64) map[string]struct{}
+
+	SubscribeNewCRSEvent(ch chan core.NewCRSEvent) event.Subscription
+}
+
 type p2pServer interface {
 	Self() *enode.Node
 
-	AddNotaryPeer(*discover.Node)
+	AddDirectPeer(*enode.Node)
+
+	RemoveDirectPeer(*enode.Node)
 
-	RemoveNotaryPeer(*discover.Node)
+	AddGroup(string, []*enode.Node, uint64)
+
+	RemoveGroup(string)
 }
 
 // statusData is the network packet for the status message.
@@ -195,22 +206,3 @@ type blockBody struct {
 
 // blockBodiesData is the network packet for block content distribution.
 type blockBodiesData []*blockBody
-
-// TODO(sonic): revisit this msg when dexon core SDK is finalized.
-// notartyNodeInfo is the network packet for notary node ip info.
-type notaryNodeInfo struct {
-	ID        discover.NodeID
-	IP        net.IP
-	UDP       uint16
-	TCP       uint16
-	Round     uint64
-	Sig       []byte
-	Timestamp int64
-}
-
-func (n *notaryNodeInfo) Hash() (h common.Hash) {
-	hw := sha3.NewKeccak256()
-	rlp.Encode(hw, n)
-	hw.Sum(h[:0])
-	return h
-}
diff --git a/dex/protocol_test.go b/dex/protocol_test.go
index c1b6efcfc..8c7638b2b 100644
--- a/dex/protocol_test.go
+++ b/dex/protocol_test.go
@@ -221,3 +221,85 @@ func TestGetBlockHeadersDataEncodeDecode(t *testing.T) {
 		}
 	}
 }
+
+func TestRecvNodeMetas(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	p, _ := newTestPeer("peer", dex64, pm, true)
+	defer pm.Stop()
+	defer p.close()
+
+	meta := NodeMeta{
+		ID: nodeID(1),
+	}
+
+	ch := make(chan newMetasEvent)
+	pm.nodeTable.SubscribeNewMetasEvent(ch)
+
+	if err := p2p.Send(p.app, MetaMsg, []interface{}{meta}); err != nil {
+		t.Fatalf("send error: %v", err)
+	}
+
+	select {
+	case event := <-ch:
+		metas := event.Metas
+		if len(metas) != 1 {
+			t.Errorf("wrong number of new metas: got %d, want 1", len(metas))
+		} else if metas[0].Hash() != meta.Hash() {
+			t.Errorf("added wrong meta hash: got %v, want %v", metas[0].Hash(), meta.Hash())
+		}
+	case <-time.After(3 * time.Second):
+		t.Errorf("no newMetasEvent received within 3 seconds")
+	}
+}
+
+func TestSendNodeMetas(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	defer pm.Stop()
+
+	allmetas := make([]*NodeMeta, 100)
+	for nonce := range allmetas {
+		allmetas[nonce] = &NodeMeta{ID: nodeID(int64(nonce))}
+	}
+
+	// Connect several peers. They should all receive the pending transactions.
+	var wg sync.WaitGroup
+	checkmetas := func(p *testPeer) {
+		defer wg.Done()
+		defer p.close()
+		seen := make(map[common.Hash]bool)
+		for _, meta := range allmetas {
+			seen[meta.Hash()] = false
+		}
+		for n := 0; n < len(allmetas) && !t.Failed(); {
+			var metas []*NodeMeta
+			msg, err := p.app.ReadMsg()
+			if err != nil {
+				t.Errorf("%v: read error: %v", p.Peer, err)
+			} else if msg.Code != MetaMsg {
+				t.Errorf("%v: got code %d, want MetaMsg", p.Peer, msg.Code)
+			}
+			if err := msg.Decode(&metas); err != nil {
+				t.Errorf("%v: %v", p.Peer, err)
+			}
+			for _, meta := range metas {
+				hash := meta.Hash()
+				seenmeta, want := seen[hash]
+				if seenmeta {
+					t.Errorf("%v: got meta more than once: %x", p.Peer, hash)
+				}
+				if !want {
+					t.Errorf("%v: got unexpected meta: %x", p.Peer, hash)
+				}
+				seen[hash] = true
+				n++
+			}
+		}
+	}
+	for i := 0; i < 3; i++ {
+		p, _ := newTestPeer(fmt.Sprintf("peer #%d", i), dex64, pm, true)
+		wg.Add(1)
+		go checkmetas(p)
+	}
+	pm.nodeTable.Add(allmetas)
+	wg.Wait()
+}
diff --git a/dex/sync.go b/dex/sync.go
index d7fe748bc..5af6076bc 100644
--- a/dex/sync.go
+++ b/dex/sync.go
@@ -25,7 +25,7 @@ import (
 	"github.com/dexon-foundation/dexon/core/types"
 	"github.com/dexon-foundation/dexon/eth/downloader"
 	"github.com/dexon-foundation/dexon/log"
-	"github.com/dexon-foundation/dexon/p2p/discover"
+	"github.com/dexon-foundation/dexon/p2p/enode"
 )
 
 const (
@@ -35,6 +35,9 @@ const (
 	// This is the target size for the packs of transactions sent by txsyncLoop.
 	// A pack can get larger than this if a single transactions exceeds this size.
 	txsyncPackSize = 100 * 1024
+
+	// This is the target number for the packs of metas sent by metasyncLoop.
+	metasyncPackNum = 1024
 )
 
 type txsync struct {
@@ -64,7 +67,7 @@ func (pm *ProtocolManager) syncTransactions(p *peer) {
 // the transactions in small packs to one peer at a time.
 func (pm *ProtocolManager) txsyncLoop() {
 	var (
-		pending = make(map[discover.NodeID]*txsync)
+		pending = make(map[enode.ID]*txsync)
 		sending = false               // whether a send is active
 		pack    = new(txsync)         // the pack that is being sent
 		done    = make(chan error, 1) // result of the send
@@ -129,6 +132,94 @@ func (pm *ProtocolManager) txsyncLoop() {
 	}
 }
 
+type metasync struct {
+	p     *peer
+	metas []*NodeMeta
+}
+
+// syncNodeMetas starts sending all node metas to the given peer.
+func (pm *ProtocolManager) syncNodeMetas(p *peer) {
+	metas := pm.nodeTable.Metas()
+	if len(metas) == 0 {
+		return
+	}
+	select {
+	case pm.metasyncCh <- &metasync{p, metas}:
+	case <-pm.quitSync:
+	}
+}
+
+// metasyncLoop takes care of the initial node meta sync for each new
+// connection. When a new peer appears, we relay all currently node metas.
+// In order to minimise egress bandwidth usage, we send
+// the metas in small packs to one peer at a time.
+func (pm *ProtocolManager) metasyncLoop() {
+	var (
+		pending = make(map[enode.ID]*metasync)
+		sending = false               // whether a send is active
+		pack    = new(metasync)       // the pack that is being sent
+		done    = make(chan error, 1) // result of the send
+	)
+
+	// send starts a sending a pack of transactions from the sync.
+	send := func(s *metasync) {
+		// Fill pack with node metas up to the target num.
+		var num int
+		pack.p = s.p
+		pack.metas = pack.metas[:0]
+		for i := 0; i < len(s.metas) && num < metasyncPackNum; i++ {
+			pack.metas = append(pack.metas, s.metas[i])
+			num += 1
+		}
+		// Remove the metas that will be sent.
+		s.metas = s.metas[:copy(s.metas, s.metas[len(pack.metas):])]
+		if len(s.metas) == 0 {
+			delete(pending, s.p.ID())
+		}
+		// Send the pack in the background.
+		s.p.Log().Trace("Sending batch of transactions", "count", len(pack.metas), "bytes", num)
+		sending = true
+		go func() { done <- pack.p.SendNodeMetas(pack.metas) }()
+	}
+
+	// pick chooses the next pending sync.
+	pick := func() *metasync {
+		if len(pending) == 0 {
+			return nil
+		}
+		n := rand.Intn(len(pending)) + 1
+		for _, s := range pending {
+			if n--; n == 0 {
+				return s
+			}
+		}
+		return nil
+	}
+
+	for {
+		select {
+		case s := <-pm.metasyncCh:
+			pending[s.p.ID()] = s
+			if !sending {
+				send(s)
+			}
+		case err := <-done:
+			sending = false
+			// Stop tracking peers that cause send failures.
+			if err != nil {
+				pack.p.Log().Debug("NodeMeta send failed", "err", err)
+				delete(pending, pack.p.ID())
+			}
+			// Schedule the next send.
+			if s := pick(); s != nil {
+				send(s)
+			}
+		case <-pm.quitSync:
+			return
+		}
+	}
+}
+
 // syncer is responsible for periodically synchronising with the network, both
 // downloading hashes and blocks as well as handling the announcement handler.
 func (pm *ProtocolManager) syncer() {
diff --git a/p2p/dial.go b/p2p/dial.go
index 9b24ed96a..909bed863 100644
--- a/p2p/dial.go
+++ b/p2p/dial.go
@@ -64,6 +64,12 @@ func (t TCPDialer) Dial(dest *enode.Node) (net.Conn, error) {
 	return t.Dialer.Dial("tcp", addr.String())
 }
 
+type dialGroup struct {
+	name  string
+	nodes map[enode.ID]*enode.Node
+	num   uint64
+}
+
 // dialstate schedules dials and discovery lookups.
 // it get's a chance to compute new tasks on every iteration
 // of the main loop in Server.run.
@@ -78,6 +84,8 @@ type dialstate struct {
 	lookupBuf     []*enode.Node // current discovery lookup results
 	randomNodes   []*enode.Node // filled from Table
 	static        map[enode.ID]*dialTask
+	direct        map[enode.ID]*dialTask
+	group         map[string]*dialGroup
 	hist          *dialHistory
 
 	start     time.Time     // time when the dialer was first used
@@ -85,6 +93,7 @@ type dialstate struct {
 }
 
 type discoverTable interface {
+	Self() *enode.Node
 	Close()
 	Resolve(*enode.Node) *enode.Node
 	LookupRandom() []*enode.Node
@@ -133,6 +142,8 @@ func newDialState(self enode.ID, static []*enode.Node, bootnodes []*enode.Node,
 		self:        self,
 		netrestrict: netrestrict,
 		static:      make(map[enode.ID]*dialTask),
+		direct:      make(map[enode.ID]*dialTask),
+		group:       make(map[string]*dialGroup),
 		dialing:     make(map[enode.ID]connFlag),
 		bootnodes:   make([]*enode.Node, len(bootnodes)),
 		randomNodes: make([]*enode.Node, maxdyn/2),
@@ -159,6 +170,23 @@ func (s *dialstate) removeStatic(n *enode.Node) {
 	s.hist.remove(n.ID())
 }
 
+func (s *dialstate) addDirect(n *enode.Node) {
+	s.direct[n.ID()] = &dialTask{flags: directDialedConn, dest: n}
+}
+
+func (s *dialstate) removeDirect(n *enode.Node) {
+	delete(s.direct, n.ID())
+	s.hist.remove(n.ID())
+}
+
+func (s *dialstate) addGroup(g *dialGroup) {
+	s.group[g.name] = g
+}
+
+func (s *dialstate) removeGroup(g *dialGroup) {
+	delete(s.group, g.name)
+}
+
 func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Time) []task {
 	if s.start.IsZero() {
 		s.start = now
@@ -196,13 +224,69 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
 		err := s.checkDial(t.dest, peers)
 		switch err {
 		case errNotWhitelisted, errSelf:
-			log.Warn("Removing static dial candidate", "id", t.dest.ID, "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()}, "err", err)
+			log.Warn("Removing static dial candidate", "id", t.dest.ID(), "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()}, "err", err)
 			delete(s.static, t.dest.ID())
 		case nil:
 			s.dialing[id] = t.flags
 			newtasks = append(newtasks, t)
 		}
 	}
+
+	for id, t := range s.direct {
+		err := s.checkDial(t.dest, peers)
+		switch err {
+		case errNotWhitelisted, errSelf:
+			log.Warn("Removing direct dial candidate", "id", t.dest.ID(), "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()}, "err", err)
+			delete(s.direct, t.dest.ID())
+		case nil:
+			s.dialing[id] = t.flags
+			newtasks = append(newtasks, t)
+		}
+	}
+
+	// compute connected
+	connected := map[string]map[enode.ID]struct{}{}
+	for _, g := range s.group {
+		connected[g.name] = map[enode.ID]struct{}{}
+	}
+
+	for id := range peers {
+		for _, g := range s.group {
+			if _, ok := g.nodes[id]; ok {
+				connected[g.name][id] = struct{}{}
+			}
+		}
+	}
+
+	for id := range s.dialing {
+		for _, g := range s.group {
+			if _, ok := g.nodes[id]; ok {
+				connected[g.name][id] = struct{}{}
+			}
+		}
+	}
+
+	groupNodes := map[enode.ID]*enode.Node{}
+	for _, g := range s.group {
+		for _, n := range g.nodes {
+			if uint64(len(connected[g.name])) >= g.num {
+				break
+			}
+			err := s.checkDial(n, peers)
+			switch err {
+			case errNotWhitelisted, errSelf:
+				log.Warn("Removing group dial candidate", "id", n.ID(), "addr", &net.TCPAddr{IP: n.IP(), Port: n.TCP()}, "err", err)
+				delete(g.nodes, n.ID())
+			case nil:
+				groupNodes[n.ID()] = n
+				connected[g.name][n.ID()] = struct{}{}
+			}
+		}
+	}
+	for _, n := range groupNodes {
+		addDial(groupDialedConn, n)
+	}
+
 	// If we don't have any peers whatsoever, try to dial a random bootnode. This
 	// scenario is useful for the testnet (and private networks) where the discovery
 	// table might be full of mostly bad peers, making it hard to find good ones.
diff --git a/p2p/dial_test.go b/p2p/dial_test.go
index 3805ff690..f9122de6f 100644
--- a/p2p/dial_test.go
+++ b/p2p/dial_test.go
@@ -525,6 +525,223 @@ func TestDialStateStaticDial(t *testing.T) {
 	})
 }
 
+func TestDialStateDirectDial(t *testing.T) {
+	wantDirect := []*enode.Node{
+		newNode(uintID(1), nil),
+		newNode(uintID(2), nil),
+		newNode(uintID(3), nil),
+		newNode(uintID(4), nil),
+		newNode(uintID(5), nil),
+	}
+	init := newDialState(enode.ID{}, nil, nil, fakeTable{}, 0, nil)
+	for _, node := range wantDirect {
+		init.addDirect(node)
+	}
+
+	runDialTest(t, dialtest{
+		init: init,
+		rounds: []round{
+			// Direct dials are launched for the nodes that
+			// aren't yet connected.
+			{
+				peers: []*Peer{
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
+				},
+				new: []task{
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(3), nil)},
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(4), nil)},
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(5), nil)},
+				},
+			},
+			// No new tasks are launched in this round because all direct
+			// nodes are either connected or still being dialed.
+			{
+				peers: []*Peer{
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
+					{rw: &conn{flags: staticDialedConn, node: newNode(uintID(3), nil)}},
+				},
+				done: []task{
+					&dialTask{flags: staticDialedConn, dest: newNode(uintID(3), nil)},
+				},
+			},
+			// No new dial tasks are launched because all direct
+			// nodes are now connected.
+			{
+				peers: []*Peer{
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(3), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(4), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(5), nil)}},
+				},
+				done: []task{
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(4), nil)},
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(5), nil)},
+				},
+				new: []task{
+					&waitExpireTask{Duration: 14 * time.Second},
+				},
+			},
+			// Wait a round for dial history to expire, no new tasks should spawn.
+			{
+				peers: []*Peer{
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(3), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(4), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(5), nil)}},
+				},
+			},
+			// If a direct node is dropped, it should be immediately redialed,
+			// irrespective whether it was originally static or dynamic.
+			{
+				peers: []*Peer{
+					{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(3), nil)}},
+					{rw: &conn{flags: directDialedConn, node: newNode(uintID(5), nil)}},
+				},
+				new: []task{
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(2), nil)},
+					&dialTask{flags: directDialedConn, dest: newNode(uintID(4), nil)},
+				},
+			},
+		},
+	})
+}
+
+func TestDialStateGroupDial(t *testing.T) {
+	groups := []*dialGroup{
+		&dialGroup{
+			name: "g1",
+			nodes: map[enode.ID]*enode.Node{
+				uintID(1): newNode(uintID(1), nil),
+				uintID(2): newNode(uintID(2), nil),
+			},
+			num: 2,
+		},
+		&dialGroup{
+			name: "g2",
+			nodes: map[enode.ID]*enode.Node{
+				uintID(2): newNode(uintID(2), nil),
+				uintID(3): newNode(uintID(3), nil),
+				uintID(4): newNode(uintID(4), nil),
+				uintID(5): newNode(uintID(5), nil),
+				uintID(6): newNode(uintID(6), nil),
+			},
+			num: 2,
+		},
+	}
+
+	type groupTest struct {
+		peers   []*Peer
+		dialing map[enode.ID]connFlag
+		ceiling map[string]uint64
+	}
+
+	tests := []groupTest{
+		{
+			peers:   nil,
+			dialing: map[enode.ID]connFlag{},
+			ceiling: map[string]uint64{"g1": 2, "g2": 4},
+		},
+		{
+			peers: []*Peer{
+				{rw: &conn{flags: staticDialedConn, node: newNode(uintID(2), nil)}},
+			},
+			dialing: map[enode.ID]connFlag{
+				uintID(1): staticDialedConn,
+			},
+			ceiling: map[string]uint64{"g1": 2, "g2": 2},
+		},
+		{
+			peers: []*Peer{
+				{rw: &conn{flags: staticDialedConn, node: newNode(uintID(1), nil)}},
+				{rw: &conn{flags: staticDialedConn, node: newNode(uintID(3), nil)}},
+				{rw: &conn{flags: staticDialedConn, node: newNode(uintID(4), nil)}},
+				{rw: &conn{flags: staticDialedConn, node: newNode(uintID(5), nil)}},
+			},
+			dialing: map[enode.ID]connFlag{
+				uintID(2): staticDialedConn,
+			},
+			ceiling: map[string]uint64{"g1": 2, "g2": 4},
+		},
+		{
+			peers: nil,
+			dialing: map[enode.ID]connFlag{
+				uintID(1): staticDialedConn,
+				uintID(2): staticDialedConn,
+				uintID(3): staticDialedConn,
+			},
+			ceiling: map[string]uint64{"g1": 2, "g2": 4},
+		},
+	}
+
+	pm := func(ps []*Peer) map[enode.ID]*Peer {
+		m := make(map[enode.ID]*Peer)
+		for _, p := range ps {
+			m[p.rw.node.ID()] = p
+		}
+		return m
+	}
+
+	run := func(i int, tt groupTest) {
+		d := newDialState(enode.ID{}, nil, nil, fakeTable{}, 0, nil)
+		d.dialing = make(map[enode.ID]connFlag)
+		for k, v := range tt.dialing {
+			d.dialing[k] = v
+		}
+
+		for _, g := range groups {
+			d.addGroup(g)
+		}
+		peermap := pm(tt.peers)
+		new := d.newTasks(len(tt.dialing), peermap, time.Now())
+
+		cnt := map[string]uint64{}
+		for id := range peermap {
+			for _, g := range groups {
+				if _, ok := g.nodes[id]; ok {
+					cnt[g.name]++
+				}
+			}
+		}
+
+		for id := range tt.dialing {
+			for _, g := range groups {
+				if _, ok := g.nodes[id]; ok {
+					cnt[g.name]++
+				}
+			}
+		}
+
+		for _, task := range new {
+			id := task.(*dialTask).dest.ID()
+			for _, g := range groups {
+				if _, ok := g.nodes[id]; ok {
+					cnt[g.name]++
+				}
+			}
+		}
+
+		for _, g := range groups {
+			if cnt[g.name] < g.num {
+				t.Errorf("test %d) group %s peers + dialing + new < num (%d < %d)",
+					i, g.name, cnt[g.name], g.num)
+			}
+			if cnt[g.name] > tt.ceiling[g.name] {
+				t.Errorf("test %d) group %s peers + dialing + new > ceiling (%d > %d)",
+					i, g.name, cnt[g.name], tt.ceiling[g.name])
+			}
+		}
+	}
+
+	for i, tt := range tests {
+		run(i, tt)
+	}
+}
+
 // This test checks that static peers will be redialed immediately if they were re-added to a static list.
 func TestDialStaticAfterReset(t *testing.T) {
 	wantStatic := []*enode.Node{
diff --git a/p2p/discover/table.go b/p2p/discover/table.go
index 25ea7b0b2..e8a219871 100644
--- a/p2p/discover/table.go
+++ b/p2p/discover/table.go
@@ -127,7 +127,7 @@ func newTable(t transport, db *enode.DB, bootnodes []*enode.Node) (*Table, error
 	return tab, nil
 }
 
-func (tab *Table) self() *enode.Node {
+func (tab *Table) Self() *enode.Node {
 	return tab.net.self()
 }
 
@@ -258,7 +258,7 @@ func (tab *Table) lookup(targetKey encPubkey, refreshIfEmpty bool) []*node {
 	)
 	// don't query further if we hit ourself.
 	// unlikely to happen often in practice.
-	asked[tab.self().ID()] = true
+	asked[tab.Self().ID()] = true
 
 	for {
 		tab.mutex.Lock()
@@ -419,7 +419,7 @@ func (tab *Table) doRefresh(done chan struct{}) {
 	// Run self lookup to discover new neighbor nodes.
 	// We can only do this if we have a secp256k1 identity.
 	var key ecdsa.PublicKey
-	if err := tab.self().Load((*enode.Secp256k1)(&key)); err == nil {
+	if err := tab.Self().Load((*enode.Secp256k1)(&key)); err == nil {
 		tab.lookup(encodePubkey(&key), false)
 	}
 
@@ -544,7 +544,7 @@ func (tab *Table) len() (n int) {
 
 // bucket returns the bucket for the given node ID hash.
 func (tab *Table) bucket(id enode.ID) *bucket {
-	d := enode.LogDist(tab.self().ID(), id)
+	d := enode.LogDist(tab.Self().ID(), id)
 	if d <= bucketMinDistance {
 		return tab.buckets[0]
 	}
@@ -557,7 +557,7 @@ func (tab *Table) bucket(id enode.ID) *bucket {
 //
 // The caller must not hold tab.mutex.
 func (tab *Table) addSeenNode(n *node) {
-	if n.ID() == tab.self().ID() {
+	if n.ID() == tab.Self().ID() {
 		return
 	}
 
@@ -599,7 +599,7 @@ func (tab *Table) addVerifiedNode(n *node) {
 	if !tab.isInitDone() {
 		return
 	}
-	if n.ID() == tab.self().ID() {
+	if n.ID() == tab.Self().ID() {
 		return
 	}
 
diff --git a/p2p/discover/table_test.go b/p2p/discover/table_test.go
index 2321e743f..8947a074e 100644
--- a/p2p/discover/table_test.go
+++ b/p2p/discover/table_test.go
@@ -147,7 +147,7 @@ func TestTable_IPLimit(t *testing.T) {
 	defer tab.Close()
 
 	for i := 0; i < tableIPLimit+1; i++ {
-		n := nodeAtDistance(tab.self().ID(), i, net.IP{172, 0, 1, byte(i)})
+		n := nodeAtDistance(tab.Self().ID(), i, net.IP{172, 0, 1, byte(i)})
 		tab.addSeenNode(n)
 	}
 	if tab.len() > tableIPLimit {
@@ -165,7 +165,7 @@ func TestTable_BucketIPLimit(t *testing.T) {
 
 	d := 3
 	for i := 0; i < bucketIPLimit+1; i++ {
-		n := nodeAtDistance(tab.self().ID(), d, net.IP{172, 0, 1, byte(i)})
+		n := nodeAtDistance(tab.Self().ID(), d, net.IP{172, 0, 1, byte(i)})
 		tab.addSeenNode(n)
 	}
 	if tab.len() > bucketIPLimit {
@@ -264,7 +264,7 @@ func TestTable_ReadRandomNodesGetAll(t *testing.T) {
 
 		for i := 0; i < len(buf); i++ {
 			ld := cfg.Rand.Intn(len(tab.buckets))
-			fillTable(tab, []*node{nodeAtDistance(tab.self().ID(), ld, intIP(ld))})
+			fillTable(tab, []*node{nodeAtDistance(tab.Self().ID(), ld, intIP(ld))})
 		}
 		gotN := tab.ReadRandomNodes(buf)
 		if gotN != tab.len() {
@@ -312,8 +312,8 @@ func TestTable_addVerifiedNode(t *testing.T) {
 	defer tab.Close()
 
 	// Insert two nodes.
-	n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
-	n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
+	n1 := nodeAtDistance(tab.Self().ID(), 256, net.IP{88, 77, 66, 1})
+	n2 := nodeAtDistance(tab.Self().ID(), 256, net.IP{88, 77, 66, 2})
 	tab.addSeenNode(n1)
 	tab.addSeenNode(n2)
 
@@ -344,8 +344,8 @@ func TestTable_addSeenNode(t *testing.T) {
 	defer tab.Close()
 
 	// Insert two nodes.
-	n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
-	n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
+	n1 := nodeAtDistance(tab.Self().ID(), 256, net.IP{88, 77, 66, 1})
+	n2 := nodeAtDistance(tab.Self().ID(), 256, net.IP{88, 77, 66, 2})
 	tab.addSeenNode(n1)
 	tab.addSeenNode(n2)
 
diff --git a/p2p/discover/table_util_test.go b/p2p/discover/table_util_test.go
index 20238aabc..7e149b22c 100644
--- a/p2p/discover/table_util_test.go
+++ b/p2p/discover/table_util_test.go
@@ -75,10 +75,10 @@ func intIP(i int) net.IP {
 
 // fillBucket inserts nodes into the given bucket until it is full.
 func fillBucket(tab *Table, n *node) (last *node) {
-	ld := enode.LogDist(tab.self().ID(), n.ID())
+	ld := enode.LogDist(tab.Self().ID(), n.ID())
 	b := tab.bucket(n.ID())
 	for len(b.entries) < bucketSize {
-		b.entries = append(b.entries, nodeAtDistance(tab.self().ID(), ld, intIP(ld)))
+		b.entries = append(b.entries, nodeAtDistance(tab.Self().ID(), ld, intIP(ld)))
 	}
 	return b.entries[bucketSize-1]
 }
diff --git a/p2p/peer.go b/p2p/peer.go
index 3c75d7dd5..2c357fdc9 100644
--- a/p2p/peer.go
+++ b/p2p/peer.go
@@ -434,7 +434,6 @@ type PeerInfo struct {
 		RemoteAddress string `json:"remoteAddress"` // Remote endpoint of the TCP data connection
 		Inbound       bool   `json:"inbound"`
 		Trusted       bool   `json:"trusted"`
-		Notary        bool   `json:"notary"`
 		Static        bool   `json:"static"`
 	} `json:"network"`
 	Protocols map[string]interface{} `json:"protocols"` // Sub-protocol specific metadata fields
@@ -459,7 +458,6 @@ func (p *Peer) Info() *PeerInfo {
 	info.Network.RemoteAddress = p.RemoteAddr().String()
 	info.Network.Inbound = p.rw.is(inboundConn)
 	info.Network.Trusted = p.rw.is(trustedConn)
-	info.Network.Notary = p.rw.is(notaryConn)
 	info.Network.Static = p.rw.is(staticDialedConn)
 
 	// Gather all the running protocol infos
diff --git a/p2p/server.go b/p2p/server.go
index c3b3a00d4..a58673342 100644
--- a/p2p/server.go
+++ b/p2p/server.go
@@ -33,13 +33,13 @@ import (
 	"github.com/dexon-foundation/dexon/crypto"
 	"github.com/dexon-foundation/dexon/event"
 	"github.com/dexon-foundation/dexon/log"
-	"github.com/dexon-foundation/dexon/p2p/discover"
 	"github.com/dexon-foundation/dexon/p2p/discv5"
 	"github.com/dexon-foundation/dexon/p2p/enode"
 	"github.com/dexon-foundation/dexon/p2p/enr"
 	"github.com/dexon-foundation/dexon/p2p/nat"
 	"github.com/dexon-foundation/dexon/p2p/netutil"
 	"github.com/dexon-foundation/dexon/rlp"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
 const (
@@ -178,10 +178,12 @@ type Server struct {
 	quit          chan struct{}
 	addstatic     chan *enode.Node
 	removestatic  chan *enode.Node
+	adddirect     chan *enode.Node
+	removedirect  chan *enode.Node
 	addtrusted    chan *enode.Node
 	removetrusted chan *enode.Node
-	addnotary     chan *enode.Node
-	removenotary  chan *enode.Node
+	addgroup      chan *dialGroup
+	removegroup   chan *dialGroup
 	posthandshake chan *conn
 	addpeer       chan *conn
 	delpeer       chan peerDrop
@@ -203,9 +205,10 @@ type connFlag int32
 const (
 	dynDialedConn connFlag = 1 << iota
 	staticDialedConn
+	directDialedConn
+	groupDialedConn
 	inboundConn
 	trustedConn
-	notaryConn
 )
 
 // conn wraps a network connection with information gathered
@@ -254,12 +257,15 @@ func (f connFlag) String() string {
 	if f&staticDialedConn != 0 {
 		s += "-staticdial"
 	}
+	if f&directDialedConn != 0 {
+		s += "-directdial"
+	}
+	if f&groupDialedConn != 0 {
+		s += "-groupdial"
+	}
 	if f&inboundConn != 0 {
 		s += "-inbound"
 	}
-	if f&notaryConn != 0 {
-		s += "-notary"
-	}
 	if s != "" {
 		s = s[1:]
 	}
@@ -333,40 +339,57 @@ func (srv *Server) RemovePeer(node *enode.Node) {
 	}
 }
 
-// AddTrustedPeer adds the given node to a reserved whitelist which allows the
-// node to always connect, even if the slot are full.
-func (srv *Server) AddTrustedPeer(node *enode.Node) {
+// AddDirectPeer connects to the given node and maintains the connection until the
+// server is shut down. If the connection fails for any reason, the server will
+// attempt to reconnect the peer.
+func (srv *Server) AddDirectPeer(node *enode.Node) {
 	select {
-	case srv.addtrusted <- node:
+	case srv.adddirect <- node:
 	case <-srv.quit:
 	}
 }
 
-// RemoveTrustedPeer removes the given node from the trusted peer set.
-func (srv *Server) RemoveTrustedPeer(node *enode.Node) {
+// RemoveDirectPeer disconnects from the given node
+func (srv *Server) RemoveDirectPeer(node *enode.Node) {
 	select {
-	case srv.removetrusted <- node:
+	case srv.removedirect <- node:
 	case <-srv.quit:
 	}
 }
 
-// AddNotaryPeer connects to the given node and maintains the connection until the
-// server is shut down. If the connection fails for any reason, the server will
-// attempt to reconnect the peer.
-// AddNotaryPeer also adds the given node to a reserved whitelist which allows the
+func (srv *Server) AddGroup(name string, nodes []*enode.Node, num uint64) {
+	m := map[enode.ID]*enode.Node{}
+	for _, node := range nodes {
+		m[node.ID()] = node
+	}
+	g := &dialGroup{name: name, nodes: m, num: num}
+	select {
+	case srv.addgroup <- g:
+	case <-srv.quit:
+	}
+}
+
+func (srv *Server) RemoveGroup(name string) {
+	g := &dialGroup{name: name}
+	select {
+	case srv.removegroup <- g:
+	case <-srv.quit:
+	}
+}
+
+// AddTrustedPeer adds the given node to a reserved whitelist which allows the
 // node to always connect, even if the slot are full.
-func (srv *Server) AddNotaryPeer(node *discover.Node) {
+func (srv *Server) AddTrustedPeer(node *enode.Node) {
 	select {
-	case srv.addnotary <- node:
+	case srv.addtrusted <- node:
 	case <-srv.quit:
 	}
 }
 
-// RemoveNotaryPeer disconnects from the given node.
-// RemoveNotaryPeer also removes the given node from the notary peer set.
-func (srv *Server) RemoveNotaryPeer(node *discover.Node) {
+// RemoveTrustedPeer removes the given node from the trusted peer set.
+func (srv *Server) RemoveTrustedPeer(node *enode.Node) {
 	select {
-	case srv.removenotary <- node:
+	case srv.removetrusted <- node:
 	case <-srv.quit:
 	}
 }
@@ -388,6 +411,36 @@ func (srv *Server) Self() *enode.Node {
 	return ln.Node()
 }
 
+func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *enode.Node {
+	// If the node is running but discovery is off, manually assemble the node infos.
+	if ntab == nil {
+		addr := srv.tcpAddr(listener)
+		return enode.NewV4(&srv.PrivateKey.PublicKey, addr.IP, addr.Port, 0)
+	}
+	// Otherwise return the discovery node.
+	return ntab.Self()
+}
+
+func (srv *Server) tcpAddr(listener net.Listener) net.TCPAddr {
+	addr := net.TCPAddr{IP: net.IP{0, 0, 0, 0}}
+	if listener == nil {
+		return addr // Inbound connections disabled, use zero address.
+	}
+	// Otherwise inject the listener address too.
+	if a, ok := listener.Addr().(*net.TCPAddr); ok {
+		addr = *a
+	}
+	if srv.NAT != nil {
+		if ip, err := srv.NAT.ExternalIP(); err == nil {
+			addr.IP = ip
+		}
+	}
+	if addr.IP.IsUnspecified() {
+		addr.IP = net.IP{127, 0, 0, 1}
+	}
+	return addr
+}
+
 // Stop terminates the server and all active peer connections.
 // It blocks until all active connections have been closed.
 func (srv *Server) Stop() {
@@ -465,10 +518,12 @@ func (srv *Server) Start() (err error) {
 	srv.posthandshake = make(chan *conn)
 	srv.addstatic = make(chan *enode.Node)
 	srv.removestatic = make(chan *enode.Node)
+	srv.adddirect = make(chan *enode.Node)
+	srv.removedirect = make(chan *enode.Node)
+	srv.addgroup = make(chan *dialGroup)
+	srv.removegroup = make(chan *dialGroup)
 	srv.addtrusted = make(chan *enode.Node)
 	srv.removetrusted = make(chan *enode.Node)
-	srv.addnotary = make(chan *enode.Node)
-	srv.removenotary = make(chan *enode.Node)
 	srv.peerOp = make(chan peerOpFunc)
 	srv.peerOpDone = make(chan struct{})
 
@@ -628,6 +683,10 @@ type dialer interface {
 	taskDone(task, time.Time)
 	addStatic(*enode.Node)
 	removeStatic(*enode.Node)
+	addDirect(*enode.Node)
+	removeDirect(*enode.Node)
+	addGroup(*dialGroup)
+	removeGroup(*dialGroup)
 }
 
 func (srv *Server) run(dialstate dialer) {
@@ -636,13 +695,15 @@ func (srv *Server) run(dialstate dialer) {
 	defer srv.nodedb.Close()
 
 	var (
-		peers        = make(map[enode.ID]*Peer)
-		inboundCount = 0
-		trusted      = make(map[enode.ID]bool, len(srv.TrustedNodes))
-		notary       = make(map[enode.ID]bool)
-		taskdone     = make(chan task, maxActiveDialTasks)
-		runningTasks []task
-		queuedTasks  []task // tasks that can't run yet
+		peers         = make(map[enode.ID]*Peer)
+		inboundCount  = 0
+		trusted       = make(map[enode.ID]bool, len(srv.TrustedNodes))
+		peerflags     = make(map[enode.ID]connFlag)
+		groupRefCount = make(map[enode.ID]int32)
+		groups        = make(map[string]*dialGroup)
+		taskdone      = make(chan task, maxActiveDialTasks)
+		runningTasks  []task
+		queuedTasks   []task // tasks that can't run yet
 	)
 	// Put trusted nodes into a map to speed up checks.
 	// Trusted peers are loaded on startup or added via AddTrustedPeer RPC.
@@ -680,6 +741,60 @@ func (srv *Server) run(dialstate dialer) {
 		}
 	}
 
+	// remember and maintain the connection flags locally
+	setConnFlags := func(id enode.ID, f connFlag, val bool) {
+		if p, ok := peers[id]; ok {
+			p.rw.set(f, val)
+		}
+		if val {
+			peerflags[id] |= f
+		} else {
+			peerflags[id] &= ^f
+		}
+		if peerflags[id] == 0 {
+			delete(peerflags, id)
+		}
+	}
+
+	// Put trusted nodes into a map to speed up checks.
+	// Trusted peers are loaded on startup or added via AddTrustedPeer RPC.
+	for _, n := range srv.TrustedNodes {
+		setConnFlags(n.ID(), trustedConn, true)
+	}
+
+	canDisconnect := func(p *Peer) bool {
+		f, ok := peerflags[p.ID()]
+		if ok && f != 0 {
+			return false
+		}
+		return !p.rw.is(dynDialedConn | inboundConn)
+	}
+
+	removeGroup := func(g *dialGroup) {
+		if gg, ok := groups[g.name]; ok {
+			for id := range gg.nodes {
+				groupRefCount[id]--
+				if groupRefCount[id] == 0 {
+					setConnFlags(id, groupDialedConn, false)
+					delete(groupRefCount, id)
+				}
+			}
+		}
+	}
+
+	addGroup := func(g *dialGroup) {
+		if _, ok := groups[g.name]; ok {
+			removeGroup(groups[g.name])
+		}
+		for id := range g.nodes {
+			groupRefCount[id]++
+			if groupRefCount[id] > 0 {
+				setConnFlags(id, groupDialedConn, true)
+			}
+		}
+		groups[g.name] = g
+	}
+
 running:
 	for {
 		scheduleTasks()
@@ -693,63 +808,59 @@ running:
 			// ephemeral static peer list. Add it to the dialer,
 			// it will keep the node connected.
 			srv.log.Trace("Adding static node", "node", n)
+			setConnFlags(n.ID(), staticDialedConn, true)
 			dialstate.addStatic(n)
 		case n := <-srv.removestatic:
 			// This channel is used by RemovePeer to send a
 			// disconnect request to a peer and begin the
 			// stop keeping the node connected.
 			srv.log.Trace("Removing static node", "node", n)
+			setConnFlags(n.ID(), staticDialedConn, false)
 			dialstate.removeStatic(n)
-			if p, ok := peers[n.ID()]; ok {
+			if p, ok := peers[n.ID()]; ok && canDisconnect(p) {
 				p.Disconnect(DiscRequested)
 			}
+		case n := <-srv.adddirect:
+			// This channel is used by AddDirectPeer to add to the
+			// ephemeral direct peer list. Add it to the dialer,
+			// it will keep the node connected.
+			srv.log.Trace("Adding direct node", "node", n)
+			setConnFlags(n.ID(), directDialedConn, true)
+			if p, ok := peers[n.ID()]; ok {
+				p.rw.set(directDialedConn, true)
+			}
+			dialstate.addDirect(n)
+		case n := <-srv.removedirect:
+			// This channel is used by RemoveDirectPeer to send a
+			// disconnect request to a peer and begin the
+			// stop keeping the node connected.
+			srv.log.Trace("Removing direct node", "node", n)
+			setConnFlags(n.ID(), directDialedConn, false)
+			if p, ok := peers[n.ID()]; ok {
+				p.rw.set(directDialedConn, false)
+				if !p.rw.is(trustedConn | groupDialedConn) {
+					p.Disconnect(DiscRequested)
+				}
+			}
+			dialstate.removeDirect(n)
+		case g := <-srv.addgroup:
+			srv.log.Trace("Adding group", "group", g)
+			addGroup(g)
+			dialstate.addGroup(g)
+		case g := <-srv.removegroup:
+			srv.log.Trace("Removing group", "group", g)
+			removeGroup(g)
+			dialstate.removeGroup(g)
 		case n := <-srv.addtrusted:
 			// This channel is used by AddTrustedPeer to add an enode
 			// to the trusted node set.
 			srv.log.Trace("Adding trusted node", "node", n)
-			trusted[n.ID()] = true
-			// Mark any already-connected peer as trusted
-			if p, ok := peers[n.ID()]; ok {
-				p.rw.set(trustedConn, true)
-			}
+			setConnFlags(n.ID(), trustedConn, true)
 		case n := <-srv.removetrusted:
 			// This channel is used by RemoveTrustedPeer to remove an enode
 			// from the trusted node set.
 			srv.log.Trace("Removing trusted node", "node", n)
-			if _, ok := trusted[n.ID()]; ok {
-				delete(trusted, n.ID())
-			}
-			// Unmark any already-connected peer as trusted
-			if p, ok := peers[n.ID()]; ok {
-				p.rw.set(trustedConn, false)
-			}
-		case n := <-srv.addnotary:
-			// This channel is used by AddNotaryPeer to add to the
-			// ephemeral notary peer list. Add it to the dialer,
-			// it will keep the node connected.
-			srv.log.Trace("Adding notary node", "node", n)
-			notary[n.ID] = true
-			if p, ok := peers[n.ID]; ok {
-				p.rw.set(notaryConn, true)
-			}
-			dialstate.addStatic(n)
-		case n := <-srv.removenotary:
-			// This channel is used by RemoveNotaryPeer to send a
-			// disconnect request to a peer and begin the
-			// stop keeping the node connected.
-			srv.log.Trace("Removing notary node", "node", n)
-			if _, ok := notary[n.ID]; ok {
-				delete(notary, n.ID)
-			}
-
-			if p, ok := peers[n.ID]; ok {
-				p.rw.set(notaryConn, false)
-			}
-
-			dialstate.removeStatic(n)
-			if p, ok := peers[n.ID]; ok {
-				p.Disconnect(DiscRequested)
-			}
+			setConnFlags(n.ID(), trustedConn, false)
 		case op := <-srv.peerOp:
 			// This channel is used by Peers and PeerCount.
 			op(peers)
@@ -764,15 +875,9 @@ running:
 		case c := <-srv.posthandshake:
 			// A connection has passed the encryption handshake so
 			// the remote identity is known (but hasn't been verified yet).
-			if trusted[c.node.ID()] {
-				// Ensure that the trusted flag is set before checking against MaxPeers.
-				c.flags |= trustedConn
+			if f, ok := peerflags[c.node.ID()]; ok {
+				c.flags |= f
 			}
-
-			if notary[c.id] {
-				c.flags |= notaryConn
-			}
-
 			// TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
 			select {
 			case c.cont <- srv.encHandshakeChecks(peers, inboundCount, c):
@@ -853,9 +958,9 @@ func (srv *Server) protoHandshakeChecks(peers map[enode.ID]*Peer, inboundCount i
 
 func (srv *Server) encHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error {
 	switch {
-	case !c.is(trustedConn|notaryConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
+	case !c.is(trustedConn|staticDialedConn|directDialedConn|groupDialedConn) && len(peers) >= srv.MaxPeers:
 		return DiscTooManyPeers
-	case !c.is(trustedConn|notaryConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns():
+	case !c.is(trustedConn|directDialedConn|groupDialedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns():
 		return DiscTooManyPeers
 	case peers[c.node.ID()] != nil:
 		return DiscAlreadyConnected
diff --git a/p2p/server_test.go b/p2p/server_test.go
index b46240722..8bd113791 100644
--- a/p2p/server_test.go
+++ b/p2p/server_test.go
@@ -27,7 +27,6 @@ import (
 
 	"github.com/dexon-foundation/dexon/crypto"
 	"github.com/dexon-foundation/dexon/log"
-	"github.com/dexon-foundation/dexon/p2p/discover"
 	"github.com/dexon-foundation/dexon/p2p/enode"
 	"github.com/dexon-foundation/dexon/p2p/enr"
 	"golang.org/x/crypto/sha3"
@@ -173,14 +172,10 @@ func TestServerDial(t *testing.T) {
 			}
 
 			// Test AddTrustedPeer/RemoveTrustedPeer and changing Trusted flags
-			// Test AddNotaryPeer/RemoveTrustedPeer and changing Notary flags.
 			// Particularly for race conditions on changing the flag state.
 			if peer := srv.Peers()[0]; peer.Info().Network.Trusted {
 				t.Errorf("peer is trusted prematurely: %v", peer)
 			}
-			if peer := srv.Peers()[0]; peer.Info().Network.Notary {
-				t.Errorf("peer is notary prematurely: %v", peer)
-			}
 			done := make(chan bool)
 			go func() {
 				srv.AddTrustedPeer(node)
@@ -192,14 +187,6 @@ func TestServerDial(t *testing.T) {
 					t.Errorf("peer is trusted after RemoveTrustedPeer: %v", peer)
 				}
 
-				srv.AddNotaryPeer(node)
-				if peer := srv.Peers()[0]; !peer.Info().Network.Notary {
-					t.Errorf("peer is not notary after AddNotaryPeer: %v", peer)
-				}
-				srv.RemoveNotaryPeer(node)
-				if peer := srv.Peers()[0]; peer.Info().Network.Notary {
-					t.Errorf("peer is notary after RemoveNotaryPeer: %v", peer)
-				}
 				done <- true
 			}()
 			// Trigger potential race conditions
@@ -216,70 +203,89 @@ func TestServerDial(t *testing.T) {
 	}
 }
 
-// TestNotaryPeer checks that the node is added to and remove from static when
-// AddNotaryPeer and RemoveNotaryPeer is called.
-func TestNotaryPeer(t *testing.T) {
-	var (
-		returned    = make(chan struct{})
-		add, remove = make(chan *discover.Node), make(chan *discover.Node)
-		tg          = taskgen{
-			newFunc: func(running int, peers map[discover.NodeID]*Peer) []task {
-				return []task{}
-			},
-			doneFunc: func(t task) {},
-			addFunc: func(n *discover.Node) {
-				add <- n
-			},
-			removeFunc: func(n *discover.Node) {
-				remove <- n
-			},
-		}
-	)
-
+func TestServerPeerConnFlag(t *testing.T) {
 	srv := &Server{
-		Config:       Config{MaxPeers: 10},
-		quit:         make(chan struct{}),
-		ntab:         fakeTable{},
-		addnotary:    make(chan *discover.Node),
-		removenotary: make(chan *discover.Node),
-		running:      true,
-		log:          log.New(),
+		Config: Config{
+			PrivateKey: newkey(),
+			MaxPeers:   10,
+			NoDial:     true,
+		},
 	}
-	srv.loopWG.Add(1)
-	go func() {
-		srv.run(tg)
-		close(returned)
-	}()
+	if err := srv.Start(); err != nil {
+		t.Fatalf("could not start: %v", err)
+	}
+	defer srv.Stop()
 
-	notaryID := randomID()
-	go srv.AddNotaryPeer(&discover.Node{ID: notaryID})
+	// inject a peer
+	key := newkey()
+	id := enode.PubkeyToIDV4(&key.PublicKey)
+	node := newNode(id, nil)
+	fd, _ := net.Pipe()
+	c := &conn{
+		node:      node,
+		fd:        fd,
+		transport: newTestTransport(&key.PublicKey, fd),
+		flags:     inboundConn,
+		cont:      make(chan error),
+	}
+	if err := srv.checkpoint(c, srv.addpeer); err != nil {
+		t.Fatalf("could not add conn: %v", err)
+	}
 
-	select {
-	case n := <-add:
-		if n.ID != notaryID {
-			t.Errorf("node ID mismatched: got %s, want %s",
-				n.ID.String(), notaryID.String())
-		}
-	case <-time.After(1 * time.Second):
-		t.Error("add static is not called within one second")
+	srv.AddTrustedPeer(node)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | trustedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | trustedConn))
 	}
 
-	go srv.RemoveNotaryPeer(&discover.Node{ID: notaryID})
-	select {
-	case n := <-remove:
-		if n.ID != notaryID {
-			t.Errorf("node ID mismatched: got %s, want %s",
-				n.ID.String(), notaryID.String())
-		}
-	case <-time.After(1 * time.Second):
-		t.Error("remove static is not called within one second")
+	srv.AddDirectPeer(node)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | trustedConn | directDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | trustedConn | directDialedConn))
 	}
 
-	srv.Stop()
-	select {
-	case <-returned:
-	case <-time.After(500 * time.Millisecond):
-		t.Error("Server.run did not return within 500ms")
+	srv.AddGroup("g1", []*enode.Node{node}, 1)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | trustedConn | directDialedConn | groupDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | trustedConn | directDialedConn | groupDialedConn))
+	}
+
+	srv.AddGroup("g2", []*enode.Node{node}, 1)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | trustedConn | directDialedConn | groupDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | trustedConn | directDialedConn | groupDialedConn))
+	}
+
+	srv.RemoveTrustedPeer(node)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | directDialedConn | groupDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | directDialedConn | directDialedConn))
+	}
+
+	srv.RemoveDirectPeer(node)
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | groupDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | directDialedConn))
+	}
+
+	srv.RemoveGroup("g1")
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != (inboundConn | groupDialedConn) {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, (inboundConn | directDialedConn))
+	}
+
+	srv.RemoveGroup("g2")
+	srv.Peers() // leverage this function to ensure trusted peer is added
+	if c.flags != inboundConn {
+		t.Errorf("flags mismatch: got %d, want %d",
+			c.flags, inboundConn)
 	}
 }
 
@@ -407,9 +413,6 @@ func TestServerManyTasks(t *testing.T) {
 type taskgen struct {
 	newFunc  func(running int, peers map[enode.ID]*Peer) []task
 	doneFunc func(task)
-
-	addFunc    func(*discover.Node)
-	removeFunc func(*discover.Node)
 }
 
 func (tg taskgen) newTasks(running int, peers map[enode.ID]*Peer, now time.Time) []task {
@@ -418,11 +421,17 @@ func (tg taskgen) newTasks(running int, peers map[enode.ID]*Peer, now time.Time)
 func (tg taskgen) taskDone(t task, now time.Time) {
 	tg.doneFunc(t)
 }
-func (tg taskgen) addStatic(n *enode.Node) {
-	tg.addFunc(n)
+func (tg taskgen) addStatic(*enode.Node) {
+}
+func (tg taskgen) removeStatic(*enode.Node) {
 }
-func (tg taskgen) removeStatic(n *enode.Node) {
-	tg.removeFunc(n)
+func (tg taskgen) addDirect(*enode.Node) {
+}
+func (tg taskgen) removeDirect(*enode.Node) {
+}
+func (tg taskgen) addGroup(*dialGroup) {
+}
+func (tg taskgen) removeGroup(*dialGroup) {
 }
 
 type testTask struct {
@@ -436,11 +445,10 @@ func (t *testTask) Do(srv *Server) {
 
 // This test checks that connections are disconnected
 // just after the encryption handshake when the server is
-// at capacity. Trusted and Notary connections should still be accepted.
+// at capacity. Trusted connections should still be accepted.
 func TestServerAtCap(t *testing.T) {
 	trustedNode := newkey()
 	trustedID := enode.PubkeyToIDV4(&trustedNode.PublicKey)
-	notaryID := randomID()
 	srv := &Server{
 		Config: Config{
 			PrivateKey:   newkey(),
@@ -453,7 +461,6 @@ func TestServerAtCap(t *testing.T) {
 		t.Fatalf("could not start: %v", err)
 	}
 	defer srv.Stop()
-	srv.AddNotaryPeer(&discover.Node{ID: notaryID})
 
 	newconn := func(id enode.ID) *conn {
 		fd, _ := net.Pipe()
@@ -484,15 +491,6 @@ func TestServerAtCap(t *testing.T) {
 		t.Error("Server did not set trusted flag")
 	}
 
-	// Try inserting a notary connection.
-	c = newconn(notaryID)
-	if err := srv.checkpoint(c, srv.posthandshake); err != nil {
-		t.Error("unexpected error for notary conn @posthandshake:", err)
-	}
-	if !c.is(notaryConn) {
-		t.Error("Server did not set notary flag")
-	}
-
 	// Remove from trusted set and try again
 	srv.RemoveTrustedPeer(newNode(trustedID, nil))
 	c = newconn(trustedID)
@@ -509,24 +507,6 @@ func TestServerAtCap(t *testing.T) {
 	if !c.is(trustedConn) {
 		t.Error("Server did not set trusted flag")
 	}
-
-	// Remove from notary set and try again
-	srv.RemoveNotaryPeer(&discover.Node{ID: notaryID})
-	c = newconn(notaryID)
-	if err := srv.checkpoint(c, srv.posthandshake); err != DiscTooManyPeers {
-		t.Error("wrong error for insert:", err)
-	}
-
-	// Add anotherID to notary set and try again
-	anotherNotaryID := randomID()
-	srv.AddNotaryPeer(&discover.Node{ID: anotherNotaryID})
-	c = newconn(anotherNotaryID)
-	if err := srv.checkpoint(c, srv.posthandshake); err != nil {
-		t.Error("unexpected error for notary conn @posthandshake:", err)
-	}
-	if !c.is(notaryConn) {
-		t.Error("Server did not set notary flag")
-	}
 }
 
 func TestServerPeerLimits(t *testing.T) {