Prevent zombie connections due to simultaneous connect

cmd/syncthing/main.go

@@ -250,20 +250,12 @@ func main() {
 		"clusterHash":   clusterHash(cfg.Repositories[0].Nodes),
 	}
 
-	// Routine to listen for incoming connections
-	if verbose {
-		infoln("Listening for incoming connections")
-	}
-	for _, addr := range cfg.Options.ListenAddress {
-		go listen(myID, addr, m, tlsCfg, connOpts)
-	}
-
 	// Routine to connect out to configured nodes
 	if verbose {
 		infoln("Attempting to connect to other nodes")
 	}
 	disc := discovery()
-	go connect(myID, disc, m, tlsCfg, connOpts)
+	go listenConnect(myID, disc, m, tlsCfg, connOpts)
 
 	// Routine to pull blocks from other nodes to synchronize the local
 	// repository. Does not run when we are in read only (publish only) mode.
@@ -401,14 +393,19 @@ func printStatsLoop(m *Model) {
 	}
 }
 
-func listen(myID string, addr string, m *Model, tlsCfg *tls.Config, connOpts map[string]string) {
+func listenConnect(myID string, disc *discover.Discoverer, m *Model, tlsCfg *tls.Config, connOpts map[string]string) {
+	var conns = make(chan *tls.Conn)
+
+	// Listen
+	for _, addr := range cfg.Options.ListenAddress {
+		addr := addr
+		go func() {
 			if debugNet {
 				dlog.Println("listening on", addr)
 			}
 			l, err := tls.Listen("tcp", addr, tlsCfg)
 			fatalErr(err)
 
-listen:
 			for {
 				conn, err := l.Accept()
 				if err != nil {
@@ -428,52 +425,13 @@ listen:
 					continue
 				}
 
-		remoteID := certID(tc.ConnectionState().PeerCertificates[0].Raw)
+				conns <- tc
-
+			}
-		if remoteID == myID {
+		}()
-			warnf("Connect from myself (%s) - should not happen", remoteID)
-			conn.Close()
-			continue
 	}
 
-		if m.ConnectedTo(remoteID) {
+	// Connect
-			warnf("Connect from connected node (%s)", remoteID)
+	go func() {
-		}
-
-		for _, nodeCfg := range cfg.Repositories[0].Nodes {
-			if nodeCfg.NodeID == remoteID {
-				protoConn := protocol.NewConnection(remoteID, conn, conn, m, connOpts)
-				m.AddConnection(conn, protoConn)
-				continue listen
-			}
-		}
-		conn.Close()
-	}
-}
-
-func discovery() *discover.Discoverer {
-	if !cfg.Options.LocalAnnEnabled {
-		return nil
-	}
-
-	infoln("Sending local discovery announcements")
-
-	if !cfg.Options.GlobalAnnEnabled {
-		cfg.Options.GlobalAnnServer = ""
-	} else if verbose {
-		infoln("Sending external discovery announcements")
-	}
-
-	disc, err := discover.NewDiscoverer(myID, cfg.Options.ListenAddress, cfg.Options.GlobalAnnServer)
-
-	if err != nil {
-		warnf("No discovery possible (%v)", err)
-	}
-
-	return disc
-}
-
-func connect(myID string, disc *discover.Discoverer, m *Model, tlsCfg *tls.Config, connOpts map[string]string) {
 		for {
 		nextNode:
 			for _, nodeCfg := range cfg.Repositories[0].Nodes {
@@ -505,21 +463,62 @@ func connect(myID string, disc *discover.Discoverer, m *Model, tlsCfg *tls.Confi
 						continue
 					}
 
-				remoteID := certID(conn.ConnectionState().PeerCertificates[0].Raw)
+					conns <- conn
-				if remoteID != nodeCfg.NodeID {
-					warnln("Unexpected nodeID", remoteID, "!=", nodeCfg.NodeID)
-					conn.Close()
-					continue
-				}
-
-				protoConn := protocol.NewConnection(remoteID, conn, conn, m, connOpts)
-				m.AddConnection(conn, protoConn)
 					continue nextNode
 				}
 			}
 
 			time.Sleep(time.Duration(cfg.Options.ReconnectIntervalS) * time.Second)
 		}
+	}()
+
+next:
+	for conn := range conns {
+		remoteID := certID(conn.ConnectionState().PeerCertificates[0].Raw)
+
+		if remoteID == myID {
+			warnf("Connected to myself (%s) - should not happen", remoteID)
+			conn.Close()
+			continue
+		}
+
+		if m.ConnectedTo(remoteID) {
+			warnf("Connected to already connected node (%s)", remoteID)
+			conn.Close()
+			continue
+		}
+
+		for _, nodeCfg := range cfg.Repositories[0].Nodes {
+			if nodeCfg.NodeID == remoteID {
+				protoConn := protocol.NewConnection(remoteID, conn, conn, m, connOpts)
+				m.AddConnection(conn, protoConn)
+				continue next
+			}
+		}
+		conn.Close()
+	}
+}
+
+func discovery() *discover.Discoverer {
+	if !cfg.Options.LocalAnnEnabled {
+		return nil
+	}
+
+	infoln("Sending local discovery announcements")
+
+	if !cfg.Options.GlobalAnnEnabled {
+		cfg.Options.GlobalAnnServer = ""
+	} else if verbose {
+		infoln("Sending external discovery announcements")
+	}
+
+	disc, err := discover.NewDiscoverer(myID, cfg.Options.ListenAddress, cfg.Options.GlobalAnnServer)
+
+	if err != nil {
+		warnf("No discovery possible (%v)", err)
+	}
+
+	return disc
 }
 
 func updateLocalModel(m *Model, w *scanner.Walker) {

cmd/syncthing/model.go

@@ -507,7 +507,13 @@ func (m *Model) RepoID() string {
 func (m *Model) AddConnection(rawConn io.Closer, protoConn Connection) {
 	nodeID := protoConn.ID()
 	m.pmut.Lock()
+	if _, ok := m.protoConn[nodeID]; ok {
+		panic("add existing node")
+	}
 	m.protoConn[nodeID] = protoConn
+	if _, ok := m.rawConn[nodeID]; ok {
+		panic("add existing node")
+	}
 	m.rawConn[nodeID] = rawConn
 	m.pmut.Unlock()