Merge pull request #257 from MainframeHQ/peer-discovery-fix

swarm/network: Bugfix swarm peer discovery

swarm/network/discovery_test.go

@@ -17,10 +17,8 @@
 package network
 
 import (
-	"fmt"
 	"testing"
 
-	"github.com/ethereum/go-ethereum/log"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
 )
 
@@ -30,28 +28,30 @@ import (
  *
  */
 func TestDiscovery(t *testing.T) {
-	addr := RandomAddr()
+	params := NewHiveParams()
-	to := NewKademlia(addr.OAddr, NewKadParams())
+	s, pp := newHiveTester(t, params)
 
-	run := func(p *BzzPeer) error {
+	id := s.IDs[0]
-		dp := newDiscovery(p, to)
+	raddr := NewAddrFromNodeID(id)
-		to.On(p)
+	pp.Register([]OverlayAddr{OverlayAddr(raddr)})
-		defer to.Off(p)
-		log.Trace(fmt.Sprintf("kademlia on %v", p))
-		return p.Run(dp.HandleMsg)
-	}
 
-	s := newBzzBaseTester(t, 1, addr, DiscoverySpec, run)
+	// start the hive and wait for the connection
-	defer s.Stop()
+	pp.Start(s.Server)
+	defer pp.Stop()
 
-	s.TestExchanges(p2ptest.Exchange{
+	// send subPeersMsg to the peer
-		Label: "outgoing SubPeersMsg",
+	err := s.TestExchanges(p2ptest.Exchange{
+		Label: "outgoing subPeersMsg",
 		Expects: []p2ptest.Expect{
 			{
-				Code: 3,
+				Code: 1,
 				Msg:  &subPeersMsg{Depth: 0},
-				Peer: s.ProtocolTester.IDs[0],
+				Peer: id,
 			},
 		},
 	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
 }

swarm/network/hive.go

@@ -162,9 +162,15 @@ func (h *Hive) connect() {
 func (h *Hive) Run(p *BzzPeer) error {
 	dp := newDiscovery(p, h)
 	depth, changed := h.On(dp)
-	// if we want discovery, advertise changed depth of depth
+	// if we want discovery, advertise change of depth
-	if h.Discovery && changed {
+	if h.Discovery {
+		if changed {
+			// if depth changed, send to all peers
 			NotifyDepth(depth, h)
+		} else {
+			// otherwise just send depth to new peer
+			dp.NotifyDepth(depth)
+		}
 	}
 	NotifyPeer(p.Off(), h)
 	defer h.Off(dp)

swarm/network/protocol.go

@@ -124,10 +124,10 @@ func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
 
 // UpdateLocalAddr updates underlayaddress of the running node
 func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr {
-	b.localAddr.Update(&BzzAddr{
+	b.localAddr = b.localAddr.Update(&BzzAddr{
 		UAddr: byteaddr,
 		OAddr: b.localAddr.OAddr,
-	})
+	}).(*BzzAddr)
 	return b.localAddr
 }
 

swarm/network/protocol_test.go

@@ -103,7 +103,7 @@ func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec,
 		return run(p)
 	}
 
-	protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+	protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 		return srv(&BzzPeer{
 			Peer:      protocols.NewPeer(p, rw, spec),
 			localAddr: addr,
@@ -111,7 +111,7 @@ func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec,
 		})
 	}
 
-	s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocall)
+	s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocol)
 
 	for _, id := range s.IDs {
 		cs[id.String()] = make(chan bool)
@@ -130,21 +130,25 @@ type bzzTester struct {
 	cs   map[string]chan bool
 }
 
-func newBzzTester(t *testing.T, n int, addr *BzzAddr, pp *p2ptest.TestPeerPool, spec *protocols.Spec, services func(Peer) error) *bzzTester {
+func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr) *bzzTester {
+	config := &BzzConfig{
+		OverlayAddr:  addr.Over(),
+		UnderlayAddr: addr.Under(),
+		HiveParams:   NewHiveParams(),
+	}
+	kad := NewKademlia(addr.OAddr, NewKadParams())
+	bzz := NewBzz(config, kad, nil)
 
-	extraservices := func(p *BzzPeer) error {
+	s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, bzz.runBzz)
-		pp.Add(p)
+
-		defer pp.Remove(p)
+	return &bzzTester{
-		if services == nil {
+		addr:           addr,
-			return nil
+		ProtocolTester: s,
 	}
-		return services(p)
-	}
-	return newBzzBaseTester(t, n, addr, spec, extraservices)
 }
 
 // should test handshakes in one exchange? parallelisation
-func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) {
+func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) error {
 	var peers []discover.NodeID
 	id := NewNodeIDFromAddr(rhs.Addr)
 	if len(disconnects) > 0 {
@@ -155,66 +159,82 @@ func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptes
 		peers = []discover.NodeID{id}
 	}
 
-	s.TestExchanges(HandshakeMsgExchange(lhs, rhs, id)...)
+	if err := s.TestExchanges(HandshakeMsgExchange(lhs, rhs, id)...); err != nil {
-	s.TestDisconnected(disconnects...)
+		return err
 	}
 
-func (s *bzzTester) runHandshakes(ids ...discover.NodeID) {
+	if len(disconnects) > 0 {
-	if len(ids) == 0 {
+		return s.TestDisconnected(disconnects...)
-		ids = s.IDs
-	}
-	for _, id := range ids {
-		s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewAddrFromNodeID(id)))
-		<-s.cs[id.String()]
 	}
 
+	// If we don't expect disconnect, ensure peers remain connected
+	err := s.TestDisconnected(&p2ptest.Disconnect{
+		Peer:  s.IDs[0],
+		Error: nil,
+	})
+
+	if err == nil {
+		return fmt.Errorf("Unexpected peer disconnect")
+	}
+
+	if err.Error() != "timed out waiting for peers to disconnect" {
+		return err
+	}
+
+	return nil
 }
 
 func correctBzzHandshake(addr *BzzAddr) *HandshakeMsg {
 	return &HandshakeMsg{
-		Version:   0,
+		Version:   1,
 		NetworkID: 322,
 		Addr:      addr,
 	}
 }
 
 func TestBzzHandshakeNetworkIDMismatch(t *testing.T) {
-	pp := p2ptest.NewTestPeerPool()
 	addr := RandomAddr()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
+	s := newBzzHandshakeTester(t, 1, addr)
-	defer s.Stop()
-
 	id := s.IDs[0]
-	s.testHandshake(
+
+	err := s.testHandshake(
 		correctBzzHandshake(addr),
-		&HandshakeMsg{Version: 0, NetworkID: 321, Addr: NewAddrFromNodeID(id)},
+		&HandshakeMsg{Version: 1, NetworkID: 321, Addr: NewAddrFromNodeID(id)},
-		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")},
+		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): network id mismatch 321 (!= 322)")},
 	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func TestBzzHandshakeVersionMismatch(t *testing.T) {
-	pp := p2ptest.NewTestPeerPool()
 	addr := RandomAddr()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
+	s := newBzzHandshakeTester(t, 1, addr)
-	defer s.Stop()
-
 	id := s.IDs[0]
-	s.testHandshake(
+
+	err := s.testHandshake(
 		correctBzzHandshake(addr),
-		&HandshakeMsg{Version: 1, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
+		&HandshakeMsg{Version: 0, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
-		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("version mismatch 1 (!= 0)")},
+		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("Handshake error: Message handler error: (msg code 0): version mismatch 0 (!= 1)")},
 	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func TestBzzHandshakeSuccess(t *testing.T) {
-	pp := p2ptest.NewTestPeerPool()
 	addr := RandomAddr()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
+	s := newBzzHandshakeTester(t, 1, addr)
-	defer s.Stop()
-
 	id := s.IDs[0]
-	s.testHandshake(
+
+	err := s.testHandshake(
 		correctBzzHandshake(addr),
-		&HandshakeMsg{Version: 0, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
+		&HandshakeMsg{Version: 1, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
 	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
 }

swarm/swarm.go

@@ -209,7 +209,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 
 	// update uaddr to correct enode
 	newaddr := self.bzz.UpdateLocalAddr([]byte(srv.Self().String()))
-	log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%x", newaddr.UAddr))
+	log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%s", newaddr.UAddr))
 
 	// set chequebook
 	if self.config.SwapEnabled {