multi+refactor: persistent peer manager

peer/persistentpeer_mgr.go

@@ -52,6 +52,11 @@ type PersistentPeerMgrConfig struct {
 	// AddrTypeIsSupported returns true if we can connect to this type of
 	// address.
 	AddrTypeIsSupported func(addr net.Addr) bool
+
+	// FetchNodeAdvertisedAddrs can be used to fetch the advertised
+	// addresses of a node that we have persisted. This should only ge used
+	// to fetch an initial set of addresses for the peer.
+	FetchNodeAdvertisedAddrs func(route.Vertex) ([]net.Addr, error)
 }
 
 // PersistentPeerManager manages persistent peers.
@@ -193,11 +198,18 @@ func (m *PersistentPeerManager) Stop() {
 func (m *PersistentPeerManager) AddPeer(pubKey *btcec.PublicKey, perm bool,
 	addrs ...*lnwire.NetAddress) {
 
+	peerKey := route.NewVertex(pubKey)
+
+	// Fetch any stored addresses we may have for this peer.
+	advertisedAddrs, err := m.cfg.FetchNodeAdvertisedAddrs(peerKey)
+	if err != nil {
+		peerLog.Errorf("Unable to retrieve advertised address for "+
+			"node %s: %v", peerKey, err)
+	}
+
 	m.Lock()
 	defer m.Unlock()
 
-	peerKey := route.NewVertex(pubKey)
-
 	backoff := m.cfg.MinBackoff
 	if peer, ok := m.conns[peerKey]; ok {
 		backoff = peer.backoff
@@ -209,7 +221,7 @@ func (m *PersistentPeerManager) AddPeer(pubKey *btcec.PublicKey, perm bool,
 		backoff: backoff,
 	}
 
-	m.setPeerAddrsUnsafe(peerKey, nil, addrs)
+	m.setPeerAddrsUnsafe(peerKey, advertisedAddrs, addrs)
 }
 
 // IsPersistentPeer returns true if the given peer is a peer that the
@@ -263,29 +275,6 @@ func (m *PersistentPeerManager) PersistentPeers() []*btcec.PublicKey {
 	return peers
 }
 
-// AddPeerAddresses is used to add addresses to a peers existing list of
-// addresses.
-func (m *PersistentPeerManager) AddPeerAddresses(pubKey *btcec.PublicKey,
-	addrs ...*lnwire.NetAddress) {
-
-	m.Lock()
-	defer m.Unlock()
-
-	peerKey := route.NewVertex(pubKey)
-
-	peer, ok := m.conns[peerKey]
-	if !ok {
-		return
-	}
-
-	peerAddrs := make([]*lnwire.NetAddress, 0, len(peer.addrs))
-	for _, addr := range peer.addrs {
-		peerAddrs = append(peerAddrs, addr)
-	}
-
-	m.setPeerAddrsUnsafe(peerKey, nil, append(peerAddrs, addrs...))
-}
-
 // NumPeerConnReqs returns the number of connection requests for the given peer.
 func (m *PersistentPeerManager) NumPeerConnReqs(pubKey *btcec.PublicKey) int {
 	m.RLock()

peer/persistentpeer_mgr_test.go

@@ -44,6 +44,11 @@ func TestPersistentPeerManagerBasics(t *testing.T) {
 		AddrTypeIsSupported: func(addr net.Addr) bool {
 			return true
 		},
+		FetchNodeAdvertisedAddrs: func(vertex route.Vertex) ([]net.Addr,
+			error) {
+
+			return nil, nil
+		},
 	})
 	defer m.Stop()
 
@@ -82,31 +87,6 @@ func TestPersistentPeerManagerBasics(t *testing.T) {
 	require.Len(t, peers, 1)
 	require.True(t, peers[0].IsEqual(bobPubKey))
 
-	// Add an address for Bob.
-	m.AddPeerAddresses(bobPubKey, &lnwire.NetAddress{
-		IdentityKey: bobPubKey,
-		Address:     testAddr1,
-	})
-
-	// Add another address for Bob.
-	m.AddPeerAddresses(bobPubKey, &lnwire.NetAddress{
-		IdentityKey: bobPubKey,
-		Address:     testAddr2,
-	})
-
-	// Both addresses should appear in Bob's address list.
-	var addrs []*lnwire.NetAddress
-	for _, addr := range m.conns[route.NewVertex(bobPubKey)].addrs {
-		addrs = append(addrs, addr)
-	}
-	require.Len(t, addrs, 2)
-	if addrs[0].Address.String() == testAddr1.String() {
-		require.Equal(t, addrs[1].Address.String(), testAddr2.String())
-	} else {
-		require.Equal(t, addrs[0].Address.String(), testAddr2.String())
-		require.Equal(t, addrs[1].Address.String(), testAddr1.String())
-	}
-
 	// Delete Bob.
 	m.DelPeer(bobPubKey)
 	peers = m.PersistentPeers()
@@ -119,6 +99,11 @@ func TestRetryCanceller(t *testing.T) {
 	m := NewPersistentPeerManager(&PersistentPeerMgrConfig{
 		MinBackoff: time.Millisecond * 10,
 		MaxBackoff: time.Millisecond * 100,
+		FetchNodeAdvertisedAddrs: func(_ route.Vertex) ([]net.Addr,
+			error) {
+
+			return nil, nil
+		},
 	})
 	defer m.Stop()
 
@@ -196,6 +181,8 @@ func TestConnectionLogic(t *testing.T) {
 	cm := newMockConnMgr(t)
 	defer cm.stop()
 
+	_, alicePubKey := btcec.PrivKeyFromBytes(channels.AlicesPrivKey)
+
 	// Create a new PersistentPeerManager.
 	m := NewPersistentPeerManager(&PersistentPeerMgrConfig{
 		ConnMgr:           cm,
@@ -206,37 +193,37 @@ func TestConnectionLogic(t *testing.T) {
 		AddrTypeIsSupported: func(addr net.Addr) bool {
 			return true
 		},
+		FetchNodeAdvertisedAddrs: func(vertex route.Vertex) ([]net.Addr,
+			error) {
+
+			switch vertex {
+			case route.NewVertex(alicePubKey):
+				return []net.Addr{testAddr1}, nil
+			default:
+				return nil, fmt.Errorf("unknown node")
+			}
+		},
 	})
 	require.NoError(t, m.Start())
 	defer m.Stop()
 
-	_, alicePubKey := btcec.PrivKeyFromBytes(channels.AlicesPrivKey)
-
 	// Add Alice as a persistent peer.
 	m.AddPeer(alicePubKey, false)
 
-	// There are currently no addresses stored for Alice, so calling
-	// ConnectPeer should not result in any connection requests.
+	// Since an address from FetchNodeAdvertisedAddrs would have been found
+	// for Alice, calling ConnectPeer should result in a connection request
+	// for that address.
 	m.ConnectPeer(alicePubKey)
-	require.Equal(t, cm.totalNumConnReqs(), 0)
+	assertOneConnReqPerAddress(t, cm, testAddr1)
 
-	// Now we add an address for Alice and attempt to connect again. This
-	// should result in 1 connection request for the given address.
-	m.AddPeerAddresses(alicePubKey, &lnwire.NetAddress{
-		IdentityKey: alicePubKey,
-		Address:     testAddr1,
-	})
-	m.ConnectPeer(alicePubKey)
+	// Advertise the same address for Alice in a new NodeAnnouncement.
+	// There should still only be one connection request for this address.
+	addUpdate(alicePubKey, testAddr1)
 	assertOneConnReqPerAddress(t, cm, testAddr1)
 
-	// If we now add a second address for Alice, calling ConnectPeer again
-	// should result in one more connection request for the new address.
-	// The connection for the first address should remain intact.
-	m.AddPeerAddresses(alicePubKey, &lnwire.NetAddress{
-		IdentityKey: alicePubKey,
-		Address:     testAddr2,
-	})
-	m.ConnectPeer(alicePubKey)
+	// Advertise new addresses for Alice, one being the same as what she
+	// had before and the other being a new one.
+	addUpdate(alicePubKey, testAddr1, testAddr2)
 	assertOneConnReqPerAddress(t, cm, testAddr1, testAddr2)
 
 	// If addresses come through from NodeAnnouncement updates, they should
@@ -296,15 +283,6 @@ func (m *mockConnMgr) stop() {
 	m.cm.Stop()
 }
 
-// totalNumConnReqs returns the number of connection requests that the
-// mockConnMgr is keeping track of.
-func (m *mockConnMgr) totalNumConnReqs() int {
-	m.Lock()
-	defer m.Unlock()
-
-	return len(m.reqs)
-}
-
 // numConnReqs returns the number of active connection requests to the given
 // address.
 func (m *mockConnMgr) numConnReqs(addr net.Addr) int {

server.go

@@ -1407,6 +1407,16 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 
 				return false
 			},
+			FetchNodeAdvertisedAddrs: func(
+				peer route.Vertex) ([]net.Addr, error) {
+
+				node, err := s.graphDB.FetchLightningNode(peer)
+				if err != nil {
+					return nil, err
+				}
+
+				return node.Addresses, nil
+			},
 		},
 	)
 
@@ -3660,63 +3670,6 @@ func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) {
 		return
 	}
 
-	// Get the last address that we used to connect to the peer.
-	addrs := []net.Addr{
-		p.NetAddress().Address,
-	}
-
-	// We'll ensure that we locate all the peers advertised addresses for
-	// reconnection purposes.
-	advertisedAddrs, err := s.fetchNodeAdvertisedAddrs(pubKey)
-	switch {
-	// We found advertised addresses, so use them.
-	case err == nil:
-		addrs = advertisedAddrs
-
-	// The peer doesn't have an advertised address.
-	case err == errNoAdvertisedAddr:
-		// If it is an outbound peer then we fall back to the existing
-		// peer address.
-		if !p.Inbound() {
-			break
-		}
-
-		// Fall back to the existing peer address if
-		// we're not accepting connections over Tor.
-		if s.torController == nil {
-			break
-		}
-
-		// If we are, the peer's address won't be known
-		// to us (we'll see a private address, which is
-		// the address used by our onion service to dial
-		// to lnd), so we don't have enough information
-		// to attempt a reconnect.
-		srvrLog.Debugf("Ignoring reconnection attempt "+
-			"to inbound peer %v without "+
-			"advertised address", p)
-		return
-
-	// We came across an error retrieving an advertised
-	// address, log it, and fall back to the existing peer
-	// address.
-	default:
-		srvrLog.Errorf("Unable to retrieve advertised "+
-			"address for node %x: %v", p.PubKey(),
-			err)
-	}
-
-	// Add any missing addresses for this peer to persistentPeerAddr.
-	for _, addr := range addrs {
-		s.persistentPeerMgr.AddPeerAddresses(
-			p.IdentityKey(), &lnwire.NetAddress{
-				IdentityKey: p.IdentityKey(),
-				Address:     addr,
-				ChainNet:    p.NetAddress().ChainNet,
-			},
-		)
-	}
-
 	// Record the computed backoff in the backoff map.
 	backoff := s.persistentPeerMgr.NextPeerBackoff(pubKey, p.StartTime())
 
@@ -4000,29 +3953,6 @@ func (s *server) Peers() []*peer.Brontide {
 	return peers
 }
 
-// errNoAdvertisedAddr is an error returned when we attempt to retrieve the
-// advertised address of a node, but they don't have one.
-var errNoAdvertisedAddr = errors.New("no advertised address found")
-
-// fetchNodeAdvertisedAddrs attempts to fetch the advertised addresses of a node.
-func (s *server) fetchNodeAdvertisedAddrs(pub *btcec.PublicKey) ([]net.Addr, error) {
-	vertex, err := route.NewVertexFromBytes(pub.SerializeCompressed())
-	if err != nil {
-		return nil, err
-	}
-
-	node, err := s.graphDB.FetchLightningNode(vertex)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(node.Addresses) == 0 {
-		return nil, errNoAdvertisedAddr
-	}
-
-	return node.Addresses, nil
-}
-
 // fetchLastChanUpdate returns a function which is able to retrieve our latest
 // channel update for a target channel.
 func (s *server) fetchLastChanUpdate() func(lnwire.ShortChannelID) (