Include Mutex for Hubs property (#13917)

Hubs property has been detected by the race condition detector
as a possible one because we could set the Hubs variable
and also we're spawning goroutines that could ask for that property

This condition is not happening right now because we're setting the
Hubs variable in the server start and after that, we're spawning the
goroutines but the tests are failing because of this and in the future
this could be a problem

app/server.go

@@ -69,7 +69,8 @@ type Server struct {
 	EmailBatching    *EmailBatchingJob
 	EmailRateLimiter *throttled.GCRARateLimiter
 
-	Hubs                        []*Hub
+	hubsLock                    sync.RWMutex
+	hubs                        []*Hub
 	HubsStopCheckingForDeadlock chan bool
 
 	PushNotificationsHub PushNotificationsHub
@@ -819,3 +820,42 @@ func (s *Server) shutdownDiagnostics() error {
 
 	return nil
 }
+
+// GetHubs returns the list of hubs. This method is safe
+// for concurrent use by multiple goroutines.
+func (s *Server) GetHubs() []*Hub {
+	s.hubsLock.RLock()
+	defer s.hubsLock.RUnlock()
+	return s.hubs
+}
+
+// getHub gets the element at the given index in the hubs list. This method is safe
+// for concurrent use by multiple goroutines.
+func (s *Server) GetHub(index int) (*Hub, error) {
+	s.hubsLock.RLock()
+	defer s.hubsLock.RUnlock()
+	if index >= len(s.hubs) {
+		return nil, errors.New("Hub element doesn't exist")
+	}
+	return s.hubs[index], nil
+}
+
+// SetHubs sets a new list of hubs. This method is safe
+// for concurrent use by multiple goroutines.
+func (s *Server) SetHubs(hubs []*Hub) {
+	s.hubsLock.Lock()
+	defer s.hubsLock.Unlock()
+	s.hubs = hubs
+}
+
+// SetHub sets the element at the given index in the hubs list. This method is safe
+// for concurrent use by multiple goroutines.
+func (s *Server) SetHub(index int, hub *Hub) error {
+	s.hubsLock.Lock()
+	defer s.hubsLock.Unlock()
+	if index >= len(s.hubs) {
+		return errors.New("Index is greater than the size of the hubs list")
+	}
+	s.hubs[index] = hub
+	return nil
+}

app/web_hub.go

@@ -62,7 +62,7 @@ func (a *App) NewWebHub() *Hub {
 
 func (a *App) TotalWebsocketConnections() int {
 	count := int64(0)
-	for _, hub := range a.Srv().Hubs {
+	for _, hub := range a.Srv().GetHubs() {
 		count = count + atomic.LoadInt64(&hub.connectionCount)
 	}
 
@@ -74,13 +74,18 @@ func (a *App) HubStart() {
 	numberOfHubs := runtime.NumCPU() * 2
 	mlog.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs))
 
-	a.Srv().Hubs = make([]*Hub, numberOfHubs)
+	a.Srv().SetHubs(make([]*Hub, numberOfHubs))
 	a.Srv().HubsStopCheckingForDeadlock = make(chan bool, 1)
 
-	for i := 0; i < len(a.Srv().Hubs); i++ {
-		a.Srv().Hubs[i] = a.NewWebHub()
-		a.Srv().Hubs[i].connectionIndex = i
-		a.Srv().Hubs[i].Start()
+	for i := 0; i < len(a.Srv().GetHubs()); i++ {
+		newHub := a.NewWebHub()
+		newHub.connectionIndex = i
+		err := a.Srv().SetHub(i, newHub)
+		if err != nil {
+			mlog.Warn("Error starting hub", mlog.Err(err), mlog.Int("index", i))
+			continue
+		}
+		newHub.Start()
 	}
 
 	go func() {
@@ -93,7 +98,7 @@ func (a *App) HubStart() {
 		for {
 			select {
 			case <-ticker.C:
-				for _, hub := range a.Srv().Hubs {
+				for _, hub := range a.Srv().GetHubs() {
 					if len(hub.broadcast) >= DEADLOCK_WARN {
 						mlog.Error(
 							"Hub processing might be deadlock with events in the buffer",
@@ -130,22 +135,27 @@ func (a *App) HubStop() {
 		mlog.Warn("We appear to have already sent the stop checking for deadlocks command")
 	}
 
-	for _, hub := range a.Srv().Hubs {
+	for _, hub := range a.Srv().GetHubs() {
 		hub.Stop()
 	}
 
-	a.Srv().Hubs = []*Hub{}
+	a.Srv().SetHubs([]*Hub{})
 }
 
 func (a *App) GetHubForUserId(userId string) *Hub {
-	if len(a.Srv().Hubs) == 0 {
+	if len(a.Srv().GetHubs()) == 0 {
 		return nil
 	}
 
 	hash := fnv.New32a()
 	hash.Write([]byte(userId))
-	index := hash.Sum32() % uint32(len(a.Srv().Hubs))
-	return a.Srv().Hubs[index]
+	index := hash.Sum32() % uint32(len(a.Srv().GetHubs()))
+	hub, err := a.Srv().GetHub(int(index))
+	if err != nil {
+		mlog.Warn("Requested hub doesn't exist", mlog.Int("hub_index", int(index)))
+		return nil
+	}
+	return hub
 }
 
 func (a *App) HubRegister(webConn *WebConn) {
@@ -195,7 +205,7 @@ func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
 			hub.Broadcast(message)
 		}
 	} else {
-		for _, hub := range a.Srv().Hubs {
+		for _, hub := range a.Srv().GetHubs() {
 			hub.Broadcast(message)
 		}
 	}

app/web_hub_test.go

@@ -77,7 +77,7 @@ func TestHubStopRaceCondition(t *testing.T) {
 	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	defer wc1.Close()
 
-	hub := th.App.Srv().Hubs[0]
+	hub := th.App.Srv().GetHubs()[0]
 	th.App.HubStop()
 	time.Sleep(5 * time.Second)