diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index 4c503047a5..e96a0323cd 100644 --- a/swarm/network/discovery.go +++ b/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() diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index ea0d776e61..3ab2297f39 100644 --- a/swarm/network/discovery_test.go +++ b/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 +} diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 1e7bb04aaa..b655f8743a 100644 --- a/swarm/network/protocol_test.go +++ b/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 }