Merge pull request #9978 from ziggie1984/fix-peer-connection

multi: fix deadlock in p2p race condition

docs/release-notes/release-notes-0.19.2.md

@@ -29,6 +29,10 @@
 - [Fixed](https://github.com/lightningnetwork/lnd/pull/9962) a case where the
   node may panic if it's running in the remote signer mode.
 
+- [Fixed](https://github.com/lightningnetwork/lnd/pull/9978) a deadlock which
+  can happen when the peer start-up has not yet completed but a another p2p
+  connection attempt tries to disconnect the peer.
+
 # New Features
 
 ## Functional Enhancements

peer/brontide.go

@@ -1591,6 +1591,10 @@ func (p *Brontide) WaitForDisconnect(ready chan struct{}) {
 // Disconnect terminates the connection with the remote peer. Additionally, a
 // signal is sent to the server and htlcSwitch indicating the resources
 // allocated to the peer can now be cleaned up.
+//
+// NOTE: Be aware that this method will block if the peer is still starting up.
+// Therefore consider starting it in a goroutine if you cannot guarantee that
+// the peer has finished starting up before calling this method.
 func (p *Brontide) Disconnect(reason error) {
 	if !atomic.CompareAndSwapInt32(&p.disconnect, 0, 1) {
 		return
@@ -1603,7 +1607,8 @@ func (p *Brontide) Disconnect(reason error) {
 	// started, otherwise we will skip reading it as this chan won't be
 	// closed, hence blocks forever.
 	if atomic.LoadInt32(&p.started) == 1 {
-		p.log.Debugf("Started, waiting on startReady signal")
+		p.log.Debugf("Peer hasn't finished starting up yet, waiting " +
+			"on startReady signal before closing connection")
 
 		select {
 		case <-p.startReady:

rpcserver.go

@@ -1870,11 +1870,13 @@ func (r *rpcServer) DisconnectPeer(ctx context.Context,
 
 	// In order to avoid erroneously disconnecting from a peer that we have
 	// an active channel with, if we have any channels active with this
-	// peer, then we'll disallow disconnecting from them.
+	// peer, then we'll disallow disconnecting from them in certain
+	// situations.
 	if len(nodeChannels) != 0 {
-		// If we are not in a dev environment or the configed dev value
+		// If the configured dev value `unsafedisconnect` is false, we
-		// `unsafedisconnect` is false, we return an error since there
+		// return an error since there are active channels. For
-		// are active channels.
+		// production environments, we allow disconnecting from a peer
+		// even if there are channels active with them.
 		if !r.cfg.Dev.GetUnsafeDisconnect() {
 			return nil, fmt.Errorf("cannot disconnect from "+
 				"peer(%x), still has %d active channels",

server.go

@@ -4091,7 +4091,7 @@ func (s *server) InboundPeerConnected(conn net.Conn) {
 		// Remove the current peer from the server's internal state and
 		// signal that the peer termination watcher does not need to
 		// execute for this peer.
-		s.removePeer(connectedPeer)
+		s.removePeerUnsafe(connectedPeer)
 		s.ignorePeerTermination[connectedPeer] = struct{}{}
 		s.scheduledPeerConnection[pubStr] = func() {
 			s.peerConnected(conn, nil, true)
@@ -4208,7 +4208,7 @@ func (s *server) OutboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn)
 		// Remove the current peer from the server's internal state and
 		// signal that the peer termination watcher does not need to
 		// execute for this peer.
-		s.removePeer(connectedPeer)
+		s.removePeerUnsafe(connectedPeer)
 		s.ignorePeerTermination[connectedPeer] = struct{}{}
 		s.scheduledPeerConnection[pubStr] = func() {
 			s.peerConnected(conn, connReq, false)
@@ -4730,7 +4730,7 @@ func (s *server) peerTerminationWatcher(p *peer.Brontide, ready chan struct{}) {
 
 	// First, cleanup any remaining state the server has regarding the peer
 	// in question.
-	s.removePeer(p)
+	s.removePeerUnsafe(p)
 
 	// Next, check to see if this is a persistent peer or not.
 	if _, ok := s.persistentPeers[pubStr]; !ok {
@@ -4939,29 +4939,24 @@ func (s *server) connectToPersistentPeer(pubKeyStr string) {
 	}()
 }
 
-// removePeer removes the passed peer from the server's state of all active
+// removePeerUnsafe removes the passed peer from the server's state of all
-// peers.
+// active peers.
-func (s *server) removePeer(p *peer.Brontide) {
+//
+// NOTE: Server mutex must be held when calling this function.
+func (s *server) removePeerUnsafe(p *peer.Brontide) {
 	if p == nil {
 		return
 	}
 
-	srvrLog.Debugf("removing peer %v", p)
+	srvrLog.Debugf("Removing peer %v", p)
 
-	// As the peer is now finished, ensure that the TCP connection is
+	// Exit early if we have already been instructed to shutdown, the peers
-	// closed and all of its related goroutines have exited.
+	// will be disconnected in the server shutdown process.
-	p.Disconnect(fmt.Errorf("server: disconnecting peer %v", p))
-
-	// If this peer had an active persistent connection request, remove it.
-	if p.ConnReq() != nil {
-		s.connMgr.Remove(p.ConnReq().ID())
-	}
-
-	// Ignore deleting peers if we're shutting down.
 	if s.Stopped() {
 		return
 	}
 
+	// Capture the peer's public key and string representation.
 	pKey := p.PubKey()
 	pubSer := pKey[:]
 	pubStr := string(pubSer)
@@ -4974,22 +4969,45 @@ func (s *server) removePeer(p *peer.Brontide) {
 		delete(s.outboundPeers, pubStr)
 	}
 
-	// Remove the peer's access permission from the access manager.
+	// When removing the peer we make sure to disconnect it asynchronously
-	peerPubStr := string(p.IdentityKey().SerializeCompressed())
+	// to avoid blocking the main server goroutine because it is holding the
-	s.peerAccessMan.removePeerAccess(peerPubStr)
+	// server's mutex. Disconnecting the peer might block and wait until the
+	// peer has fully started up. This can happen if an inbound and outbound
+	// race condition occurs.
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
 
-	// Copy the peer's error buffer across to the server if it has any items
+		p.Disconnect(fmt.Errorf("server: disconnecting peer %v", p))
-	// in it so that we can restore peer errors across connections.
-	if p.ErrorBuffer().Total() > 0 {
-		s.peerErrors[pubStr] = p.ErrorBuffer()
-	}
 
-	// Inform the peer notifier of a peer offline event so that it can be
+		// If this peer had an active persistent connection request,
-	// reported to clients listening for peer events.
+		// remove it.
-	var pubKey [33]byte
+		if p.ConnReq() != nil {
-	copy(pubKey[:], pubSer)
+			s.connMgr.Remove(p.ConnReq().ID())
+		}
 
-	s.peerNotifier.NotifyPeerOffline(pubKey)
+		// Remove the peer's access permission from the access manager.
+		peerPubStr := string(p.IdentityKey().SerializeCompressed())
+		s.peerAccessMan.removePeerAccess(peerPubStr)
+
+		// Copy the peer's error buffer across to the server if it has
+		// any items in it so that we can restore peer errors across
+		// connections. We need to look up the error after the peer has
+		// been disconnected because we write the error in the
+		// `Disconnect` method.
+		s.mu.Lock()
+		if p.ErrorBuffer().Total() > 0 {
+			s.peerErrors[pubStr] = p.ErrorBuffer()
+		}
+		s.mu.Unlock()
+
+		// Inform the peer notifier of a peer offline event so that it
+		// can be reported to clients listening for peer events.
+		var pubKey [33]byte
+		copy(pubKey[:], pubSer)
+
+		s.peerNotifier.NotifyPeerOffline(pubKey)
+	}()
 }
 
 // ConnectToPeer requests that the server connect to a Lightning Network peer
@@ -5129,8 +5147,11 @@ func (s *server) DisconnectPeer(pubKey *btcec.PublicKey) error {
 	delete(s.persistentPeersBackoff, pubStr)
 
 	// Remove the peer by calling Disconnect. Previously this was done with
-	// removePeer, which bypassed the peerTerminationWatcher.
+	// removePeerUnsafe, which bypassed the peerTerminationWatcher.
-	peer.Disconnect(fmt.Errorf("server: DisconnectPeer called"))
+	//
+	// NOTE: We call it in a goroutine to avoid blocking the main server
+	// goroutine because we might hold the server's mutex.
+	go peer.Disconnect(fmt.Errorf("server: DisconnectPeer called"))
 
 	return nil
 }