swarm/network: fix hive bug not sending shallow peers

-  hive bug: needed shallow peers were not sent to nodes beyond connection's proximity order
- add extensive protocol exchange tests for initial subPeersMsg-peersMsg exchange
- modify bzzProtocolTester to allow pregenerated overlay addresses

swarm/network/discovery.go

@@ -156,19 +156,30 @@ func (msg subPeersMsg) String() string {
 	return fmt.Sprintf("%T: request peers > PO%02d. ", msg, msg.Depth)
 }
 
+// handleSubPeersMsg handles incoming subPeersMsg
+// this message represents the saturation depth of the remote peer
+// saturation depth is the radius within which the peer subscribes to peers
+// the first time this is received we send peer info on all
+// our connected peers that fall within peers saturation depth
+// otherwise this depth is just recorded on the peer, so that
+// subsequent new connections are sent iff they fall within the radius
 func (d *Peer) handleSubPeersMsg(msg *subPeersMsg) error {
+	// only do this once
+	d.setDepth(msg.Depth)
 	if !d.sentPeers {
-		d.setDepth(msg.Depth)
 		var peers []*BzzAddr
+		// iterate connection in ascending order of disctance from the remote address
 		d.kad.EachConn(d.Over(), 255, func(p *Peer, po int) bool {
-			if pob, _ := Pof(d, d.kad.BaseAddr(), 0); pob > po {
+			// terminate if we are beyond the radius
+			if uint8(po) < msg.Depth {
 				return false
 			}
-			if !d.seen(p.BzzAddr) {
+			if !d.seen(p.BzzAddr) { // here just records the peer sent
 				peers = append(peers, p.BzzAddr)
 			}
 			return true
 		})
+		// if useful  peers are found, send them over
 		if len(peers) > 0 {
 			go d.Send(context.TODO(), &peersMsg{Peers: peers})
 		}
@@ -177,7 +188,7 @@ func (d *Peer) handleSubPeersMsg(msg *subPeersMsg) error {
 	return nil
 }
 
-// seen takes an peer address and checks if it was sent to a peer already
+// seen takes a peer address and checks if it was sent to a peer already
 // if not, marks the peer as sent
 func (d *Peer) seen(p *BzzAddr) bool {
 	d.mtx.Lock()

swarm/network/discovery_test.go

@@ -17,9 +17,19 @@
 package network
 
 import (
+	"crypto/ecdsa"
+	"crypto/rand"
+	"fmt"
+	"net"
 	"testing"
+	"time"
 
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/swarm/pot"
 )
 
 /***
@@ -58,3 +68,164 @@ func TestDiscovery(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+const (
+	maxPO     = 10
+	maxPeerPO = 8
+)
+
+// TestInitialPeersMsg tests if peersMsg response to incoming subPeersMsg is correct
+func TestSubPeersMsg(t *testing.T) {
+	for po := 0; po < maxPO; po++ {
+		for depth := 0; depth < maxPO; depth++ {
+			t.Run(fmt.Sprintf("PO=%d,advetised depth=%d", po, depth), func(t *testing.T) {
+				testSubPeersMsg(t, po, depth)
+			})
+		}
+	}
+}
+
+// testSubPeersMsg tests that the correct set of peer info is sent
+// to another peer after receiving their subPeersMsg request
+func testSubPeersMsg(t *testing.T, peerPO, peerDepth int) {
+	// generate random pivot address
+	prvkey, err := crypto.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+	pivotAddr := pot.NewAddressFromBytes(PrivateKeyToBzzKey(prvkey))
+	// generate control peers address at peerPO wrt pivot
+	peerAddr := pot.RandomAddressAt(pivotAddr, peerPO)
+	// construct kademlia and hive
+	to := NewKademlia(pivotAddr[:], NewKadParams())
+	hive := NewHive(NewHiveParams(), to, nil)
+
+	// expected addrs in peersMsg response
+	var expBzzAddrs []*BzzAddr
+	addrAt := func(a pot.Address, po int) []byte {
+		b := pot.RandomAddressAt(a, po)
+		return b[:]
+	}
+	connect := func(base pot.Address, po int) *BzzAddr {
+		on := addrAt(base, po)
+		peer := newDiscPeer(on)
+		hive.On(peer)
+		return peer.BzzAddr
+	}
+	register := func(base pot.Address, po int) {
+		hive.Register(&BzzAddr{OAddr: addrAt(base, po)})
+	}
+
+	for po := maxPeerPO; po >= 0; po-- {
+		// create a fake connected peer at po from peerAddr
+		on := connect(peerAddr, po)
+		// create a fake registered address at po from peerAddr
+		register(peerAddr, po)
+		// we collect expected peer addresses only up till peerPO
+		if po < peerDepth {
+			continue
+		}
+		expBzzAddrs = append(expBzzAddrs, on)
+	}
+
+	// create a special bzzBaseTester in which we can associate `enode.ID` to the `bzzAddr` we created above
+	s, _, err := newBzzBaseTesterWithAddrs(t, prvkey, [][]byte{peerAddr[:]}, DiscoverySpec, hive.Run)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// peerID to use in the protocol tester testExchange expect/trigger
+	peerID := s.Nodes[0].ID()
+
+	// now we need to wait until the tester's control peer appears in the hive
+	// so the protocol started
+	ticker := time.NewTicker(10 * time.Millisecond)
+	attempts := 100
+	for range ticker.C {
+		if _, found := hive.peers[peerID]; found {
+			break
+		}
+		attempts--
+		if attempts == 0 {
+			t.Fatal("timeout waiting for control peer to be in kademlia")
+		}
+	}
+
+	// pivotDepth is the advertised depth of the pivot node we expect in the outgoing subPeersMsg
+	pivotDepth := hive.saturation()
+	// the test exchange is as follows:
+	// 1. pivot sends to the control peer a `subPeersMsg` advertising its depth (ignored)
+	// 2. peer sends to pivot a `subPeersMsg` advertising its own depth (arbitrarily chosen)
+	// 3. pivot responds with `peersMsg` with the set of expected peers
+	err = s.TestExchanges(
+		p2ptest.Exchange{
+			Label: "outgoing subPeersMsg",
+			Expects: []p2ptest.Expect{
+				{
+					Code: 1,
+					Msg:  &subPeersMsg{Depth: uint8(pivotDepth)},
+					Peer: peerID,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Label: "trigger subPeersMsg and expect peersMsg",
+			Triggers: []p2ptest.Trigger{
+				{
+					Code: 1,
+					Msg:  &subPeersMsg{Depth: uint8(peerDepth)},
+					Peer: peerID,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				{
+					Code:    0,
+					Msg:     &peersMsg{Peers: expBzzAddrs},
+					Peer:    peerID,
+					Timeout: 100 * time.Millisecond,
+				},
+			},
+		})
+
+	// for values MaxPeerPO < peerPO < MaxPO the pivot has no peers to offer to the control peer
+	// in this case, no peersMsg will be sent out, and we would run into a time out
+	if err != nil {
+		if len(expBzzAddrs) > 0 {
+			t.Fatal(err)
+		} else if err.Error() != "exchange #1 \"trigger subPeersMsg and expect peersMsg\": timed out" {
+			t.Fatalf("expected timeout, got %v", err)
+		}
+	} else {
+		if len(expBzzAddrs) == 0 {
+			t.Fatalf("expected timeout, got no error")
+		}
+	}
+}
+
+// as we are not creating a real node via the protocol,
+// we need to create the discovery peer objects for the additional kademlia
+// nodes manually
+func newDiscPeer(addr []byte) *Peer {
+	pKey, err := ecdsa.GenerateKey(crypto.S256(), rand.Reader)
+	if err != nil {
+		panic(err.Error())
+	}
+	pubKey := pKey.PublicKey
+	nod := enode.NewV4(&pubKey, net.IPv4(127, 0, 0, 1), 0, 0)
+	bzzAddr := &BzzAddr{OAddr: addr, UAddr: []byte(nod.String())}
+	id := nod.ID()
+	p2pPeer := p2p.NewPeer(id, id.String(), nil)
+	return NewPeer(&BzzPeer{
+		Peer:    protocols.NewPeer(p2pPeer, &dummyMsgRW{}, DiscoverySpec),
+		BzzAddr: bzzAddr,
+	}, nil)
+}
+
+type dummyMsgRW struct{}
+
+func (d *dummyMsgRW) ReadMsg() (p2p.Msg, error) {
+	return p2p.Msg{}, nil
+}
+func (d *dummyMsgRW) WriteMsg(msg p2p.Msg) error {
+	return nil
+}

swarm/network/protocol_test.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -31,6 +32,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/swarm/pot"
 )
 
 const (
@@ -71,19 +73,36 @@ func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id enode.ID) []p2ptest.Exchang
 }
 
 func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *protocols.Spec, run func(*BzzPeer) error) (*bzzTester, error) {
-	cs := make(map[string]chan bool)
+	var addrs [][]byte
+	for i := 0; i < n; i++ {
+		addr := pot.RandomAddress()
+		addrs = append(addrs, addr[:])
+	}
+	pt, _, err := newBzzBaseTesterWithAddrs(t, prvkey, addrs, spec, run)
+	return pt, err
+}
+
+func newBzzBaseTesterWithAddrs(t *testing.T, prvkey *ecdsa.PrivateKey, addrs [][]byte, spec *protocols.Spec, run func(*BzzPeer) error) (*bzzTester, [][]byte, error) {
+	n := len(addrs)
+	cs := make(map[enode.ID]chan bool)
 
 	srv := func(p *BzzPeer) error {
 		defer func() {
-			if cs[p.ID().String()] != nil {
-				close(cs[p.ID().String()])
+			if cs[p.ID()] != nil {
+				close(cs[p.ID()])
 			}
 		}()
 		return run(p)
 	}
-
+	mu := &sync.Mutex{}
+	nodeToAddr := make(map[enode.ID][]byte)
 	protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-		return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: NewAddr(p.Node())})
+		mu.Lock()
+		defer mu.Unlock()
+		nodeToAddr[p.ID()] = addrs[0]
+		bzzAddr := &BzzAddr{addrs[0], []byte(p.Node().String())}
+		addrs = addrs[1:]
+		return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: bzzAddr})
 	}
 
 	s := p2ptest.NewProtocolTester(prvkey, n, protocol)
@@ -92,30 +111,36 @@ func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *proto
 	record.Set(NewENRAddrEntry(bzzKey))
 	err := enode.SignV4(&record, prvkey)
 	if err != nil {
-		return nil, fmt.Errorf("unable to generate ENR: %v", err)
+		return nil, nil, fmt.Errorf("unable to generate ENR: %v", err)
 	}
 	nod, err := enode.New(enode.V4ID{}, &record)
 	if err != nil {
-		return nil, fmt.Errorf("unable to create enode: %v", err)
+		return nil, nil, fmt.Errorf("unable to create enode: %v", err)
 	}
 	addr := getENRBzzAddr(nod)
 
 	for _, node := range s.Nodes {
 		log.Warn("node", "node", node)
-		cs[node.ID().String()] = make(chan bool)
+		cs[node.ID()] = make(chan bool)
 	}
 
-	return &bzzTester{
+	var nodeAddrs [][]byte
+	pt := &bzzTester{
 		addr:           addr,
 		ProtocolTester: s,
 		cs:             cs,
-	}, nil
+	}
+	for _, n := range pt.Nodes {
+		nodeAddrs = append(nodeAddrs, nodeToAddr[n.ID()])
+	}
+
+	return pt, nodeAddrs, nil
 }
 
 type bzzTester struct {
 	*p2ptest.ProtocolTester
 	addr *BzzAddr
-	cs   map[string]chan bool
+	cs   map[enode.ID]chan bool
 	bzz  *Bzz
 }