connmgr: Remain responsive with simul failed conns.

This modifies the connection manager handler for failed connections so
that it remains responsive to requests when there are multiple
simultaneous failed connections in the retry state and adds a test to
ensure proper functionality.

While here it also updates a few formatting nits in the server code.

connmgr/connmanager.go

@@ -250,24 +250,26 @@ func (cm *ConnManager) handleFailedConn(ctx context.Context, c *ConnReq) {
 			d = maxRetryDuration
 		}
 		log.Debugf("Retrying connection to %v in %v", c, d)
-		select {
-		case <-time.After(d):
-			go cm.Connect(ctx, c)
-		case <-cm.quit:
-			return
-		}
+		go func() {
+			select {
+			case <-time.After(d):
+				cm.Connect(ctx, c)
+			case <-cm.quit:
+			}
+		}()
 	} else if cm.cfg.GetNewAddress != nil {
 		cm.failedAttempts++
 		if cm.failedAttempts >= maxFailedAttempts {
 			log.Debugf("Max failed connection attempts reached: [%d] "+
 				"-- retrying connection in: %v", maxFailedAttempts,
 				cm.cfg.RetryDuration)
-			select {
-			case <-time.After(cm.cfg.RetryDuration):
-				go cm.newConnReq(ctx)
-			case <-cm.quit:
-				return
-			}
+			go func() {
+				select {
+				case <-time.After(cm.cfg.RetryDuration):
+					cm.newConnReq(ctx)
+				case <-cm.quit:
+				}
+			}()
 		} else {
 			go cm.newConnReq(ctx)
 		}

connmgr/connmanager_test.go

@@ -8,6 +8,7 @@ package connmgr
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"sync"
@@ -454,6 +455,77 @@ func TestNetworkFailure(t *testing.T) {
 	}
 }
 
+// TestMultipleFailedConns ensures that the connection manager remains
+// responsive when there are multiple simultaneous failed connections for
+// persistent peers in the retry state.
+func TestMultipleFailedConns(t *testing.T) {
+	// Override the max retry duration for this test since it relies on having
+	// multiple connections in the retry state.
+	curMaxRetryDuration := maxRetryDuration
+	maxRetryDuration = 500 * time.Millisecond
+	defer func() {
+		maxRetryDuration = curMaxRetryDuration
+	}()
+
+	const targetFailed = 5
+	var dials uint32
+	var closeOnce sync.Once
+	hitTargetFailed := make(chan struct{})
+	errDialer := func(ctx context.Context, network, addr string) (net.Conn, error) {
+		totalDials := atomic.AddUint32(&dials, 1)
+		if totalDials >= targetFailed {
+			closeOnce.Do(func() { close(hitTargetFailed) })
+		}
+		return nil, errors.New("network down")
+	}
+	cmgr, err := New(&Config{
+		RetryDuration: maxRetryDuration,
+		Dial:          errDialer,
+	})
+	if err != nil {
+		t.Fatalf("New error: %v", err)
+	}
+	ctx, shutdown, wg := runConnMgrAsync(context.Background(), cmgr)
+
+	// Establish several connection requests to localhost IPs.
+	for i := 0; i < targetFailed; i++ {
+		cr := &ConnReq{
+			Addr: &net.TCPAddr{
+				IP:   net.ParseIP(fmt.Sprintf("127.0.0.%d", i+1)),
+				Port: 18555,
+			},
+			Permanent: true,
+		}
+		go cmgr.Connect(ctx, cr)
+	}
+
+	// Wait for the target number of dials and ensure they happen simultaneously
+	// by checking it happens before the retry timeout.
+	select {
+	case <-hitTargetFailed:
+	case <-time.After(20 * time.Millisecond):
+		t.Fatal("did not reach target number of dials before timeout")
+	}
+
+	// Ensure that the connection manager still responds to requests while the
+	// failed connections are still retrying.
+	disconnected := make(chan struct{})
+	go func() {
+		const badID = ^uint64(0)
+		cmgr.Disconnect(badID)
+		close(disconnected)
+	}()
+	select {
+	case <-disconnected:
+	case <-time.After(20 * time.Millisecond):
+		t.Fatal("timeout servicing connmgr requests")
+	}
+
+	// Ensure clean shutdown of connection manager.
+	shutdown()
+	wg.Wait()
+}
+
 // TestShutdownFailedConns tests that failed connections are ignored after
 // connmgr is shutdown.
 //

server.go

@@ -2024,6 +2024,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 				Permanent: msg.permanent,
 			})
 		msg.reply <- nil
+
 	case removeNodeMsg:
 		found := disconnectPeer(state.persistentPeers, msg.cmp, func(sp *serverPeer) {
 			// Keep group counts ok since we remove from
@@ -2045,6 +2046,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 		} else {
 			msg.reply <- errors.New("peer not found")
 		}
+
 	case cancelPendingMsg:
 		netAddr, err := addrStringToNetAddr(msg.addr)
 		if err != nil {
@@ -2052,6 +2054,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 			return
 		}
 		msg.reply <- s.connManager.CancelPending(netAddr)
+
 	case getOutboundGroup:
 		count, ok := state.outboundGroups[msg.key]
 		if ok {
@@ -2059,7 +2062,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 		} else {
 			msg.reply <- 0
 		}
-	// Request a list of the persistent (added) peers.
+
 	case getAddedNodesMsg:
 		// Respond with a slice of the relevant peers.
 		peers := make([]*serverPeer, 0, len(state.persistentPeers))
@@ -2067,6 +2070,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 			peers = append(peers, sp)
 		}
 		msg.reply <- peers
+
 	case disconnectNodeMsg:
 		// Check inbound peers. We pass a nil callback since we don't
 		// require any additional actions on disconnect for inbound peers.