MM-23805: Refactor web_hub (#14277)

* MM-23800: remove goroutineID and stack printing Each hub has a goroutineID which is calculated with a known hack. The FAQ clearly explains why goroutines don't have an id: https://golang.org/doc/faq#no_goroutine_id. We only added that because sometimes the hub would be deadlocked and having the goroutineID would be useful when getting the stack trace. This is also problematic in stress tests because the hubs would frequently get overloaded and the logs would unnecessarily have stack traces. But that was in the past, and we have done extensive testing with load tests and fuzz testing to smooth any rough edges remaining. Including adding additional metrics for hub buffer size. Monitoring the metrics is a better way to approach this problem. Therefore, we remove these kludges from the code. * Also remove deadlock checking code There is no need for that anymore since we are getting rid of the stack printing anyways. Let's do a wholesale refactor and clean up the codebase. * MM-23805: Refactor web_hub This is a beginning of the refactoring of the websocket code. To start off with, we unexport some methods and constants which did not need to be exported. There are more remaining but some are out of scope for this PR. The main chunk of refactor is to unexport the webconn send channel which was the main cause of panics. Since we were directly sending to the connection from various parts of the codebase, it would be possible that the send channel would be closed and we could still send a message. This would crash the server. To fix this, we refactor the code to centralize all sending from the main hub goroutine. This means we can leverage the connections map to check if the connection exists or not, and only then send the message. We also move the cluster calls to cluster.go. * bring back cluster code inside hub * Incorporate review comments * Address review comments * rename index * MM-23807: Refactor web_conn - Unexport some struct fields and constants which are not necessary to be accessed from outside the package. This will help us moving the entire websocket handling code to a separate package later. - Change some empty string checks to check for empty string rather than doing a len check which is more idiomatic. Both of them compile to the same code. So it doesn't make a difference performance-wise. - Remove redundant ToJson calls to get the length. - Incorporate review comments - Unexport some more methods * Fix field name * Run make app-layers * Add note on hub check
2020-04-23 13:16:18 +05:30
--- a/app/app_iface.go
+++ b/app/app_iface.go
@@ -154,6 +154,8 @@ type AppIface interface {
 	GetEmojiStaticUrl(emojiName string) (string, *model.AppError)
 	// GetEnvironmentConfig returns a map of configuration keys whose values have been overridden by an environment variable.
 	GetEnvironmentConfig() map[string]interface{}
+	// GetHubForUserId returns the hub for a given user id.
+	GetHubForUserId(userId string) *Hub
 	// GetLdapGroup retrieves a single LDAP group by the given LDAP group id.
 	GetLdapGroup(ldapGroupID string) (*model.Group, *model.AppError)
 	// GetMarketplacePlugins returns a list of plugins from the marketplace-server,
@@ -183,6 +185,14 @@ type AppIface interface {
 	GetTeamSchemeChannelRoles(teamId string) (guestRoleName string, userRoleName string, adminRoleName string, err *model.AppError)
 	// GetTotalUsersStats is used for the DM list total
 	GetTotalUsersStats(viewRestrictions *model.ViewUsersRestrictions) (*model.UsersStats, *model.AppError)
+	// HubRegister registers a connection to a hub.
+	HubRegister(webConn *WebConn)
+	// HubStart starts all the hubs.
+	HubStart()
+	// HubStop stops all the hubs.
+	HubStop()
+	// HubUnregister unregisters a connection from a hub.
+	HubUnregister(webConn *WebConn)
 	// InstallMarketplacePlugin installs a plugin listed in the marketplace server. It will get the plugin bundle
 	// from the prepackaged folder, if available, or remotely if EnableRemoteMarketplace is true.
 	InstallMarketplacePlugin(request *model.InstallMarketplacePluginRequest) (*model.Manifest, *model.AppError)
@@ -204,6 +214,10 @@ type AppIface interface {
 	MakeAuditRecord(event string, initialStatus string) *audit.Record
 	// MarkChanelAsUnreadFromPost will take a post and set the channel as unread from that one.
 	MarkChannelAsUnreadFromPost(postID string, userID string) (*model.ChannelUnreadAt, *model.AppError)
+	// NewWebConn returns a new WebConn instance.
+	NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn
+	// NewWebHub creates a new Hub.
+	NewWebHub() *Hub
 	// OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon,
 	// so that it points to the URL (relative) of the emoji - static if emoji is default, /api if custom.
 	OverrideIconURLIfEmoji(post *model.Post)
@@ -281,6 +295,8 @@ type AppIface interface {
 	UpdateChannel(channel *model.Channel) (*model.Channel, *model.AppError)
 	// UpdateChannelScheme saves the new SchemeId of the channel passed.
 	UpdateChannelScheme(channel *model.Channel) (*model.Channel, *model.AppError)
+	// UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session.
+	UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	// UploadFile uploads a single file in form of a completely constructed byte array for a channel.
 	UploadFile(data []byte, channelId string, filename string) (*model.FileInfo, *model.AppError)
 	// UploadFileX uploads a single file as specified in t. It applies the upload
@@ -523,7 +539,6 @@ type AppIface interface {
 	GetGroupsBySource(groupSource model.GroupSource) ([]*model.Group, *model.AppError)
 	GetGroupsByTeam(teamId string, opts model.GroupSearchOpts) ([]*model.GroupWithSchemeAdmin, int, *model.AppError)
 	GetGroupsByUserId(userId string) ([]*model.Group, *model.AppError)
-	GetHubForUserId(userId string) *Hub
 	GetIncomingWebhook(hookId string) (*model.IncomingWebhook, *model.AppError)
 	GetIncomingWebhooksForTeamPage(teamId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError)
 	GetIncomingWebhooksForTeamPageByUser(teamId string, userId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError)
@@ -674,10 +689,6 @@ type AppIface interface {
 	HasPermissionToChannelByPost(askingUserId string, postId string, permission *model.Permission) bool
 	HasPermissionToTeam(askingUserId string, teamId string, permission *model.Permission) bool
 	HasPermissionToUser(askingUserId string, userId string) bool
-	HubRegister(webConn *WebConn)
-	HubStart()
-	HubStop()
-	HubUnregister(webConn *WebConn)
 	ImageProxy() *imageproxy.ImageProxy
 	ImageProxyAdder() func(string) string
 	ImageProxyRemover() (f func(string) string)
@@ -724,8 +735,6 @@ type AppIface interface {
 	MoveFile(oldPath, newPath string) *model.AppError
 	NewClusterDiscoveryService() *ClusterDiscoveryService
 	NewPluginAPI(manifest *model.Manifest) plugin.API
-	NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn
-	NewWebHub() *Hub
 	Notification() einterfaces.NotificationInterface
 	NotificationsLog() *mlog.Logger
 	OpenInteractiveDialog(request model.OpenDialogRequest) *model.AppError
@@ -949,7 +958,6 @@ type AppIface interface {
 	UpdateUserAuth(userId string, userAuth *model.UserAuth) (*model.UserAuth, *model.AppError)
 	UpdateUserNotifyProps(userId string, props map[string]string) (*model.User, *model.AppError)
 	UpdateUserRoles(userId string, newRoles string, sendWebSocketEvent bool) (*model.User, *model.AppError)
-	UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	UploadEmojiImage(id string, imageData *multipart.FileHeader) *model.AppError
 	UploadMultipartFiles(teamId string, channelId string, userId string, fileHeaders []*multipart.FileHeader, clientIds []string, now time.Time) (*model.FileUploadResponse, *model.AppError)
 	UpsertGroupMember(groupID string, userID string) (*model.GroupMember, *model.AppError)
--- a/app/cluster_handlers.go
+++ b/app/cluster_handlers.go
@@ -59,7 +59,7 @@ func (a *App) clusterInvalidateCacheForUserHandler(msg *model.ClusterMessage) {
 }

 func (a *App) clusterInvalidateCacheForUserTeamsHandler(msg *model.ClusterMessage) {
-	a.invalidateCacheForUserTeamsSkipClusterSend(msg.Data)
+	a.InvalidateWebConnSessionCacheForUser(msg.Data)
 }

 func (a *App) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) {
--- a/app/web_conn.go
+++ b/app/web_conn.go
@@ -9,44 +9,49 @@ import (
 	"sync/atomic"
 	"time"

-	"github.com/gorilla/websocket"
-	goi18n "github.com/mattermost/go-i18n/i18n"
 	"github.com/mattermost/mattermost-server/v5/mlog"
 	"github.com/mattermost/mattermost-server/v5/model"
+
+	"github.com/gorilla/websocket"
+	goi18n "github.com/mattermost/go-i18n/i18n"
 )

 const (
-	SEND_QUEUE_SIZE           = 256
-	SEND_SLOW_WARN            = (SEND_QUEUE_SIZE * 50) / 100
-	SEND_DEADLOCK_WARN        = (SEND_QUEUE_SIZE * 95) / 100
-	WRITE_WAIT                = 30 * time.Second
-	PONG_WAIT                 = 100 * time.Second
-	PING_PERIOD               = (PONG_WAIT * 6) / 10
-	AUTH_TIMEOUT              = 5 * time.Second
-	WEBCONN_MEMBER_CACHE_TIME = 1000 * 60 * 30 // 30 minutes
+	sendQueueSize          = 256
+	sendSlowWarn           = (sendQueueSize * 50) / 100
+	sendFullWarn           = (sendQueueSize * 95) / 100
+	writeWaitTime          = 30 * time.Second
+	pongWaitTime           = 100 * time.Second
+	pingInterval           = (pongWaitTime * 6) / 10
+	authCheckInterval      = 5 * time.Second
+	webConnMemberCacheTime = 1000 * 60 * 30 // 30 minutes
 )

+// WebConn represents a single websocket connection to a user.
+// It contains all the necesarry state to manage sending/receiving data to/from
+// a websocket.
 type WebConn struct {
-	sessionExpiresAt          int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
-	App                       *App
-	WebSocket                 *websocket.Conn
-	Send                      chan model.WebSocketMessage
+	sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
+	App              *App
+	WebSocket        *websocket.Conn
+	T                goi18n.TranslateFunc
+	Locale           string
+	Sequence         int64
+	UserId           string
+
+	allChannelMembers         map[string]string
+	lastAllChannelMembersTime int64
+	lastUserActivityAt        int64
+	send                      chan model.WebSocketMessage
 	sessionToken              atomic.Value
 	session                   atomic.Value
-	LastUserActivityAt        int64
-	UserId                    string
-	T                         goi18n.TranslateFunc
-	Locale                    string
-	AllChannelMembers         map[string]string
-	LastAllChannelMembersTime int64
-	Sequence                  int64
-	closeOnce                 sync.Once
 	endWritePump              chan struct{}
 	pumpFinished              chan struct{}
 }

+// NewWebConn returns a new WebConn instance.
 func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn {
-	if len(session.UserId) > 0 {
+	if session.UserId != "" {
 		a.Srv().Go(func() {
 			a.SetStatusOnline(session.UserId, false)
 			a.UpdateLastActivityAtIfNeeded(session)
@@ -55,9 +60,9 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra

 	wc := &WebConn{
 		App:                a,
-		Send:               make(chan model.WebSocketMessage, SEND_QUEUE_SIZE),
+		send:               make(chan model.WebSocketMessage, sendQueueSize),
 		WebSocket:          ws,
-		LastUserActivityAt: model.GetMillis(),
+		lastUserActivityAt: model.GetMillis(),
 		UserId:             session.UserId,
 		T:                  t,
 		Locale:             locale,
@@ -72,34 +77,38 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra
 	return wc
 }

+// Close closes the WebConn.
 func (wc *WebConn) Close() {
 	wc.WebSocket.Close()
-	wc.closeOnce.Do(func() {
-		close(wc.endWritePump)
-	})
 	<-wc.pumpFinished
 }

+// GetSessionExpiresAt returns the time at which the session expires.
 func (wc *WebConn) GetSessionExpiresAt() int64 {
 	return atomic.LoadInt64(&wc.sessionExpiresAt)
 }

+// SetSessionExpiresAt sets the time at which the session expires.
 func (wc *WebConn) SetSessionExpiresAt(v int64) {
 	atomic.StoreInt64(&wc.sessionExpiresAt, v)
 }

+// GetSessionToken returns the session token of the connection.
 func (wc *WebConn) GetSessionToken() string {
 	return wc.sessionToken.Load().(string)
 }

+// SetSessionToken sets the session token of the connection.
 func (wc *WebConn) SetSessionToken(v string) {
 	wc.sessionToken.Store(v)
 }

+// GetSession returns the session of the connection.
 func (wc *WebConn) GetSession() *model.Session {
 	return wc.session.Load().(*model.Session)
 }

+// SetSession sets the session of the connection.
 func (wc *WebConn) SetSession(v *model.Session) {
 	if v != nil {
 		v = v.DeepCopy()
@@ -108,17 +117,18 @@ func (wc *WebConn) SetSession(v *model.Session) {
 	wc.session.Store(v)
 }

+// Pump starts the WebConn instance. After this, the websocket
+// is ready to send/receive messages.
 func (wc *WebConn) Pump() {
-	ch := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		wc.writePump()
-		close(ch)
 	}()
 	wc.readPump()
-	wc.closeOnce.Do(func() {
-		close(wc.endWritePump)
-	})
-	<-ch
+	close(wc.endWritePump)
+	wg.Wait()
 	wc.App.HubUnregister(wc)
 	close(wc.pumpFinished)
 }
@@ -128,9 +138,9 @@ func (wc *WebConn) readPump() {
 		wc.WebSocket.Close()
 	}()
 	wc.WebSocket.SetReadLimit(model.SOCKET_MAX_MESSAGE_SIZE_KB)
-	wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT))
+	wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
 	wc.WebSocket.SetPongHandler(func(string) error {
-		wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT))
+		wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
 		if wc.IsAuthenticated() {
 			wc.App.Srv().Go(func() {
 				wc.App.SetStatusAwayIfNeeded(wc.UserId, false)
@@ -142,12 +152,7 @@ func (wc *WebConn) readPump() {
 	for {
 		var req model.WebSocketRequest
 		if err := wc.WebSocket.ReadJSON(&req); err != nil {
-			// browsers will appear as CloseNoStatusReceived
-			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-				mlog.Debug("websocket.read: client side closed socket", mlog.String("user_id", wc.UserId))
-			} else {
-				mlog.Debug("websocket.read: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-			}
+			wc.logSocketErr("websocket.read", err)
 			return
 		}
 		wc.App.Srv().WebSocketRouter.ServeWebSocket(wc, &req)
@@ -155,8 +160,8 @@ func (wc *WebConn) readPump() {
 }

 func (wc *WebConn) writePump() {
-	ticker := time.NewTicker(PING_PERIOD)
-	authTicker := time.NewTicker(AUTH_TIMEOUT)
+	ticker := time.NewTicker(pingInterval)
+	authTicker := time.NewTicker(authCheckInterval)

 	defer func() {
 		ticker.Stop()
@@ -166,9 +171,9 @@ func (wc *WebConn) writePump() {

 	for {
 		select {
-		case msg, ok := <-wc.Send:
+		case msg, ok := <-wc.send:
 			if !ok {
-				wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+				wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 				wc.WebSocket.WriteMessage(websocket.CloseMessage, []byte{})
 				return
 			}
@@ -176,11 +181,12 @@ func (wc *WebConn) writePump() {
 			evt, evtOk := msg.(*model.WebSocketEvent)

 			skipSend := false
-			if len(wc.Send) >= SEND_SLOW_WARN {
+			if len(wc.send) >= sendSlowWarn {
 				// When the pump starts to get slow we'll drop non-critical messages
-				if msg.EventType() == model.WEBSOCKET_EVENT_TYPING ||
-					msg.EventType() == model.WEBSOCKET_EVENT_STATUS_CHANGE ||
-					msg.EventType() == model.WEBSOCKET_EVENT_CHANNEL_VIEWED {
+				switch msg.EventType() {
+				case model.WEBSOCKET_EVENT_TYPING,
+					model.WEBSOCKET_EVENT_STATUS_CHANGE,
+					model.WEBSOCKET_EVENT_CHANNEL_VIEWED:
 					mlog.Info(
 						"websocket.slow: dropping message",
 						mlog.String("user_id", wc.UserId),
@@ -201,33 +207,22 @@ func (wc *WebConn) writePump() {
 					msgBytes = []byte(msg.ToJson())
 				}

-				if len(wc.Send) >= SEND_DEADLOCK_WARN {
-					if evtOk {
-						mlog.Warn(
-							"websocket.full",
-							mlog.String("user_id", wc.UserId),
-							mlog.String("type", msg.EventType()),
-							mlog.String("channel_id", evt.GetBroadcast().ChannelId),
-							mlog.Int("size", len(msg.ToJson())),
-						)
-					} else {
-						mlog.Warn(
-							"websocket.full",
-							mlog.String("user_id", wc.UserId),
-							mlog.String("type", msg.EventType()),
-							mlog.Int("size", len(msg.ToJson())),
-						)
+				if len(wc.send) >= sendFullWarn {
+					logData := []mlog.Field{
+						mlog.String("user_id", wc.UserId),
+						mlog.String("type", msg.EventType()),
+						mlog.Int("size", len(msgBytes)),
 					}
+					if evtOk {
+						logData = append(logData, mlog.String("channel_id", evt.GetBroadcast().ChannelId))
+					}
+
+					mlog.Warn("websocket.full", logData...)
 				}

-				wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+				wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 				if err := wc.WebSocket.WriteMessage(websocket.TextMessage, msgBytes); err != nil {
-					// browsers will appear as CloseNoStatusReceived
-					if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-						mlog.Debug("websocket.send: client side closed socket", mlog.String("user_id", wc.UserId))
-					} else {
-						mlog.Debug("websocket.send: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-					}
+					wc.logSocketErr("websocket.send", err)
 					return
 				}

@@ -237,14 +232,9 @@ func (wc *WebConn) writePump() {
 			}

 		case <-ticker.C:
-			wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT))
+			wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
 			if err := wc.WebSocket.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
-				// browsers will appear as CloseNoStatusReceived
-				if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
-					mlog.Debug("websocket.ticker: client side closed socket", mlog.String("user_id", wc.UserId))
-				} else {
-					mlog.Debug("websocket.ticker: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
-				}
+				wc.logSocketErr("websocket.ticker", err)
 				return
 			}

@@ -261,13 +251,15 @@ func (wc *WebConn) writePump() {
 	}
 }

+// InvalidateCache resets all internal data of the WebConn.
 func (wc *WebConn) InvalidateCache() {
-	wc.AllChannelMembers = nil
-	wc.LastAllChannelMembersTime = 0
+	wc.allChannelMembers = nil
+	wc.lastAllChannelMembersTime = 0
 	wc.SetSession(nil)
 	wc.SetSessionExpiresAt(0)
 }

+// IsAuthenticated returns whether the given WebConn is authenticated or not.
 func (wc *WebConn) IsAuthenticated() bool {
 	// Check the expiry to see if we need to check for a new session
 	if wc.GetSessionExpiresAt() < model.GetMillis() {
@@ -324,7 +316,8 @@ func (wc *WebConn) shouldSendEventToGuest(msg *model.WebSocketEvent) bool {
 	return canSee
 }

-func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
+// shouldSendEvent returns whether the message should be sent or not.
+func (wc *WebConn) shouldSendEvent(msg *model.WebSocketEvent) bool {
 	// IMPORTANT: Do not send event if WebConn does not have a session
 	if !wc.IsAuthenticated() {
 		return false
@@ -353,7 +346,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}

 	// If the event is destined to a specific user
-	if len(msg.GetBroadcast().UserId) > 0 {
+	if msg.GetBroadcast().UserId != "" {
 		return wc.UserId == msg.GetBroadcast().UserId
 	}

@@ -365,31 +358,31 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}

 	// Only report events to users who are in the channel for the event
-	if len(msg.GetBroadcast().ChannelId) > 0 {
-		if model.GetMillis()-wc.LastAllChannelMembersTime > WEBCONN_MEMBER_CACHE_TIME {
-			wc.AllChannelMembers = nil
-			wc.LastAllChannelMembersTime = 0
+	if msg.GetBroadcast().ChannelId != "" {
+		if model.GetMillis()-wc.lastAllChannelMembersTime > webConnMemberCacheTime {
+			wc.allChannelMembers = nil
+			wc.lastAllChannelMembersTime = 0
 		}

-		if wc.AllChannelMembers == nil {
+		if wc.allChannelMembers == nil {
 			result, err := wc.App.Srv().Store.Channel().GetAllChannelMembersForUser(wc.UserId, true, false)
 			if err != nil {
 				mlog.Error("webhub.shouldSendEvent.", mlog.Err(err))
 				return false
 			}
-			wc.AllChannelMembers = result
-			wc.LastAllChannelMembersTime = model.GetMillis()
+			wc.allChannelMembers = result
+			wc.lastAllChannelMembersTime = model.GetMillis()
 		}

-		if _, ok := wc.AllChannelMembers[msg.GetBroadcast().ChannelId]; ok {
+		if _, ok := wc.allChannelMembers[msg.GetBroadcast().ChannelId]; ok {
 			return true
 		}
 		return false
 	}

 	// Only report events to users who are in the team for the event
-	if len(msg.GetBroadcast().TeamId) > 0 {
-		return wc.IsMemberOfTeam(msg.GetBroadcast().TeamId)
+	if msg.GetBroadcast().TeamId != "" {
+		return wc.isMemberOfTeam(msg.GetBroadcast().TeamId)
 	}

 	if wc.GetSession().Props[model.SESSION_PROP_IS_GUEST] == "true" {
@@ -399,10 +392,12 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	return true
 }

-func (wc *WebConn) IsMemberOfTeam(teamId string) bool {
+// IsMemberOfTeam returns whether the user of the WebConn
+// is a member of the given teamId or not.
+func (wc *WebConn) isMemberOfTeam(teamId string) bool {
 	currentSession := wc.GetSession()

-	if currentSession == nil || len(currentSession.Token) == 0 {
+	if currentSession == nil || currentSession.Token == "" {
 		session, err := wc.App.GetSession(wc.GetSessionToken())
 		if err != nil {
 			mlog.Error("Invalid session.", mlog.Err(err))
@@ -414,3 +409,12 @@ func (wc *WebConn) IsMemberOfTeam(teamId string) bool {

 	return currentSession.GetTeamByTeamId(teamId) != nil
 }
+
+func (wc *WebConn) logSocketErr(source string, err error) {
+	// browsers will appear as CloseNoStatusReceived
+	if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
+		mlog.Debug(source+": client side closed socket", mlog.String("user_id", wc.UserId))
+	} else {
+		mlog.Debug(source+": closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
+	}
+}
--- a/app/web_conn_test.go
+++ b/app/web_conn_test.go
@@ -86,15 +86,15 @@ func TestWebConnShouldSendEvent(t *testing.T) {
 	event := model.NewWebSocketEvent("some_event", "", "", "", nil)
 	for _, c := range cases {
 		event = event.SetBroadcast(c.Broadcast)
-		assert.Equal(t, c.User1Expected, basicUserWc.ShouldSendEvent(event), c.Description)
-		assert.Equal(t, c.User2Expected, basicUser2Wc.ShouldSendEvent(event), c.Description)
-		assert.Equal(t, c.AdminExpected, adminUserWc.ShouldSendEvent(event), c.Description)
+		assert.Equal(t, c.User1Expected, basicUserWc.shouldSendEvent(event), c.Description)
+		assert.Equal(t, c.User2Expected, basicUser2Wc.shouldSendEvent(event), c.Description)
+		assert.Equal(t, c.AdminExpected, adminUserWc.shouldSendEvent(event), c.Description)
 	}

 	event2 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, th.BasicTeam.Id, "", "", nil)
-	assert.True(t, basicUserWc.ShouldSendEvent(event2))
-	assert.True(t, basicUser2Wc.ShouldSendEvent(event2))
+	assert.True(t, basicUserWc.shouldSendEvent(event2))
+	assert.True(t, basicUser2Wc.shouldSendEvent(event2))

 	event3 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, "wrongId", "", "", nil)
-	assert.False(t, basicUserWc.ShouldSendEvent(event3))
+	assert.False(t, basicUserWc.shouldSendEvent(event3))
 }
--- a/app/web_hub.go
+++ b/app/web_hub.go
@@ -15,15 +15,23 @@ import (
 )

 const (
-	BROADCAST_QUEUE_SIZE = 4096
+	broadcastQueueSize = 4096
 )

-type WebConnActivityMessage struct {
-	UserId       string
-	SessionToken string
-	ActivityAt   int64
+type webConnActivityMessage struct {
+	userId       string
+	sessionToken string
+	activityAt   int64
 }

+type webConnDirectMessage struct {
+	conn *WebConn
+	msg  model.WebSocketMessage
+}
+
+// Hub is the central place to manage all websocket connections in the server.
+// It handles different websocket events and sending messages to individual
+// user connections.
 type Hub struct {
 	// connectionCount should be kept first.
 	// See https://github.com/mattermost/mattermost-server/pull/7281
@@ -36,21 +44,23 @@ type Hub struct {
 	stop            chan struct{}
 	didStop         chan struct{}
 	invalidateUser  chan string
-	activity        chan *WebConnActivityMessage
-	ExplicitStop    bool
+	activity        chan *webConnActivityMessage
+	directMsg       chan *webConnDirectMessage
+	explicitStop    bool
 }

+// NewWebHub creates a new Hub.
 func (a *App) NewWebHub() *Hub {
 	return &Hub{
 		app:            a,
 		register:       make(chan *WebConn, 1),
 		unregister:     make(chan *WebConn, 1),
-		broadcast:      make(chan *model.WebSocketEvent, BROADCAST_QUEUE_SIZE),
+		broadcast:      make(chan *model.WebSocketEvent, broadcastQueueSize),
 		stop:           make(chan struct{}),
 		didStop:        make(chan struct{}),
 		invalidateUser: make(chan string),
-		activity:       make(chan *WebConnActivityMessage),
-		ExplicitStop:   false,
+		activity:       make(chan *webConnActivityMessage),
+		directMsg:      make(chan *webConnDirectMessage),
 	}
 }

@@ -58,6 +68,7 @@ func (a *App) TotalWebsocketConnections() int {
 	return a.Srv().TotalWebsocketConnections()
 }

+// HubStart starts all the hubs.
 func (a *App) HubStart() {
 	// Total number of hubs is twice the number of CPUs.
 	numberOfHubs := runtime.NumCPU() * 2
@@ -77,6 +88,36 @@ func (a *App) HubStart() {
 	}
 }

+func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
+	if message.GetBroadcast().UserId != "" {
+		hub := a.GetHubForUserId(message.GetBroadcast().UserId)
+		if hub != nil {
+			hub.Broadcast(message)
+		}
+		return
+	}
+	for _, hub := range a.Srv().GetHubs() {
+		hub.Broadcast(message)
+	}
+}
+
+func (a *App) invalidateCacheForUserSkipClusterSend(userId string) {
+	a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId)
+	a.InvalidateWebConnSessionCacheForUser(userId)
+}
+
+func (a *App) invalidateCacheForWebhook(webhookId string) {
+	a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId)
+}
+
+func (a *App) InvalidateWebConnSessionCacheForUser(userId string) {
+	hub := a.GetHubForUserId(userId)
+	if hub != nil {
+		hub.InvalidateUser(userId)
+	}
+}
+
+// HubStop stops all the hubs.
 func (a *App) HubStop() {
 	mlog.Info("stopping websocket hub connections")

@@ -87,6 +128,7 @@ func (a *App) HubStop() {
 	a.Srv().SetHubs([]*Hub{})
 }

+// GetHubForUserId returns the hub for a given user id.
 func (a *App) GetHubForUserId(userId string) *Hub {
 	if len(a.Srv().GetHubs()) == 0 {
 		return nil
@@ -103,6 +145,7 @@ func (a *App) GetHubForUserId(userId string) *Hub {
 	return hub
 }

+// HubRegister registers a connection to a hub.
 func (a *App) HubRegister(webConn *WebConn) {
 	hub := a.GetHubForUserId(webConn.UserId)
 	if hub != nil {
@@ -113,6 +156,7 @@ func (a *App) HubRegister(webConn *WebConn) {
 	}
 }

+// HubUnregister unregisters a connection from a hub.
 func (a *App) HubUnregister(webConn *WebConn) {
 	hub := a.GetHubForUserId(webConn.UserId)
 	if hub != nil {
@@ -149,19 +193,6 @@ func (a *App) Publish(message *model.WebSocketEvent) {
 	}
 }

-func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
-	if message.GetBroadcast().UserId != "" {
-		hub := a.GetHubForUserId(message.GetBroadcast().UserId)
-		if hub != nil {
-			hub.Broadcast(message)
-		}
-	} else {
-		for _, hub := range a.Srv().GetHubs() {
-			hub.Broadcast(message)
-		}
-	}
-}
-
 func (a *App) invalidateCacheForChannel(channel *model.Channel) {
 	a.Srv().Store.Channel().InvalidateChannel(channel.Id)
 	a.invalidateCacheForChannelByNameSkipClusterSend(channel.TeamId, channel.Name)
@@ -237,7 +268,7 @@ func (a *App) InvalidateCacheForUser(userId string) {
 }

 func (a *App) invalidateCacheForUserTeams(userId string) {
-	a.invalidateCacheForUserTeamsSkipClusterSend(userId)
+	a.InvalidateWebConnSessionCacheForUser(userId)
 	a.Srv().Store.Team().InvalidateAllTeamIdsForUser(userId)

 	if a.Cluster() != nil {
@@ -250,33 +281,7 @@ func (a *App) invalidateCacheForUserTeams(userId string) {
 	}
 }

-func (a *App) invalidateCacheForUserSkipClusterSend(userId string) {
-	a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId)
-
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
-func (a *App) invalidateCacheForUserTeamsSkipClusterSend(userId string) {
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
-func (a *App) invalidateCacheForWebhook(webhookId string) {
-	a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId)
-}
-
-func (a *App) InvalidateWebConnSessionCacheForUser(userId string) {
-	hub := a.GetHubForUserId(userId)
-	if hub != nil {
-		hub.InvalidateUser(userId)
-	}
-}
-
+// UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session.
 func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64) {
 	hub := a.GetHubForUserId(session.UserId)
 	if hub != nil {
@@ -284,13 +289,15 @@ func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64)
 	}
 }

+// Register registers a connection to the hub.
 func (h *Hub) Register(webConn *WebConn) {
 	select {
 	case h.register <- webConn:
-	case <-h.didStop:
+	case <-h.stop:
 	}
 }

+// Unregister unregisters a connection from the hub.
 func (h *Hub) Unregister(webConn *WebConn) {
 	select {
 	case h.unregister <- webConn:
@@ -298,37 +305,64 @@ func (h *Hub) Unregister(webConn *WebConn) {
 	}
 }

+// Broadcast broadcasts the message to all connections in the hub.
 func (h *Hub) Broadcast(message *model.WebSocketEvent) {
-	if h != nil && h.broadcast != nil && message != nil {
+	// XXX: The hub nil check is because of the way we setup our tests. We call `app.NewServer()`
+	// which returns a server, but only after that, we call `wsapi.Init()` through our FakeApp adapter
+	// to initialize the hub. But in the `NewServer` call itself, we call `RunOldAppInitialization`
+	// which directly proceeds to broadcast some messages happily.
+	// This needs to be fixed once the FakeApp adapter goes away. And possibly, we can look into
+	// doing hub initialization inside NewServer itself.
+	if h != nil && message != nil {
 		if metrics := h.app.Metrics(); metrics != nil {
 			metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 		}
 		select {
 		case h.broadcast <- message:
-		case <-h.didStop:
+		case <-h.stop:
 		}
 	}
 }

+// InvalidateUser invalidates the cache for the given user.
 func (h *Hub) InvalidateUser(userId string) {
 	select {
 	case h.invalidateUser <- userId:
-	case <-h.didStop:
+	case <-h.stop:
 	}
 }

+// UpdateActivity sets the LastUserActivityAt field for the connection
+// of the user.
 func (h *Hub) UpdateActivity(userId, sessionToken string, activityAt int64) {
 	select {
-	case h.activity <- &WebConnActivityMessage{UserId: userId, SessionToken: sessionToken, ActivityAt: activityAt}:
-	case <-h.didStop:
+	case h.activity <- &webConnActivityMessage{
+		userId:       userId,
+		sessionToken: sessionToken,
+		activityAt:   activityAt,
+	}:
+	case <-h.stop:
 	}
 }

+// SendMessage sends the given message to the given connection.
+func (h *Hub) SendMessage(conn *WebConn, msg model.WebSocketMessage) {
+	select {
+	case h.directMsg <- &webConnDirectMessage{
+		conn: conn,
+		msg:  msg,
+	}:
+	case <-h.stop:
+	}
+}
+
+// Stop stops the hub.
 func (h *Hub) Stop() {
 	close(h.stop)
 	<-h.didStop
 }

+// Start starts the hub.
 func (h *Hub) Start() {
 	var doStart func()
 	var doRecoverableStart func()
@@ -337,85 +371,94 @@ func (h *Hub) Start() {
 	doStart = func() {
 		mlog.Debug("Hub is starting", mlog.Int("index", h.connectionIndex))

-		connections := newHubConnectionIndex()
+		connIndex := newHubConnectionIndex()

 		for {
 			select {
-			case webCon := <-h.register:
-				connections.Add(webCon)
-				atomic.StoreInt64(&h.connectionCount, int64(len(connections.All())))
-				if webCon.IsAuthenticated() {
-					webCon.Send <- webCon.createHelloMessage()
+			case webConn := <-h.register:
+				connIndex.Add(webConn)
+				atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All())))
+				if webConn.IsAuthenticated() {
+					webConn.send <- webConn.createHelloMessage()
 				}
-			case webCon := <-h.unregister:
-				connections.Remove(webCon)
-				atomic.StoreInt64(&h.connectionCount, int64(len(connections.All())))
+			case webConn := <-h.unregister:
+				connIndex.Remove(webConn)
+				atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All())))

-				if len(webCon.UserId) == 0 {
+				if len(webConn.UserId) == 0 {
 					continue
 				}

-				conns := connections.ForUser(webCon.UserId)
+				conns := connIndex.ForUser(webConn.UserId)
 				if len(conns) == 0 {
 					h.app.Srv().Go(func() {
-						h.app.SetStatusOffline(webCon.UserId, false)
+						h.app.SetStatusOffline(webConn.UserId, false)
 					})
-				} else {
-					var latestActivity int64 = 0
-					for _, conn := range conns {
-						if conn.LastUserActivityAt > latestActivity {
-							latestActivity = conn.LastUserActivityAt
-						}
-					}
-					if h.app.IsUserAway(latestActivity) {
-						h.app.Srv().Go(func() {
-							h.app.SetStatusLastActivityAt(webCon.UserId, latestActivity)
-						})
+					continue
+				}
+				var latestActivity int64 = 0
+				for _, conn := range conns {
+					if conn.lastUserActivityAt > latestActivity {
+						latestActivity = conn.lastUserActivityAt
 					}
 				}
+
+				if h.app.IsUserAway(latestActivity) {
+					h.app.Srv().Go(func() {
+						h.app.SetStatusLastActivityAt(webConn.UserId, latestActivity)
+					})
+				}
 			case userId := <-h.invalidateUser:
-				for _, webCon := range connections.ForUser(userId) {
-					webCon.InvalidateCache()
+				for _, webConn := range connIndex.ForUser(userId) {
+					webConn.InvalidateCache()
 				}
 			case activity := <-h.activity:
-				for _, webCon := range connections.ForUser(activity.UserId) {
-					if webCon.GetSessionToken() == activity.SessionToken {
-						webCon.LastUserActivityAt = activity.ActivityAt
+				for _, webConn := range connIndex.ForUser(activity.userId) {
+					if webConn.GetSessionToken() == activity.sessionToken {
+						webConn.lastUserActivityAt = activity.activityAt
 					}
 				}
+			case directMsg := <-h.directMsg:
+				if !connIndex.Has(directMsg.conn) {
+					continue
+				}
+				select {
+				case directMsg.conn.send <- directMsg.msg:
+				default:
+					mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", directMsg.conn.UserId))
+					close(directMsg.conn.send)
+					connIndex.Remove(directMsg.conn)
+				}
 			case msg := <-h.broadcast:
 				if metrics := h.app.Metrics(); metrics != nil {
 					metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 				}
-				candidates := connections.All()
+				candidates := connIndex.All()
 				if msg.GetBroadcast().UserId != "" {
-					candidates = connections.ForUser(msg.GetBroadcast().UserId)
+					candidates = connIndex.ForUser(msg.GetBroadcast().UserId)
 				}
 				msg = msg.PrecomputeJSON()
-				for _, webCon := range candidates {
-					if webCon.ShouldSendEvent(msg) {
+				for _, webConn := range candidates {
+					if !connIndex.Has(webConn) {
+						continue
+					}
+					if webConn.shouldSendEvent(msg) {
 						select {
-						case webCon.Send <- msg:
+						case webConn.send <- msg:
 						default:
-							mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webCon.UserId))
-							close(webCon.Send)
-							connections.Remove(webCon)
+							mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webConn.UserId))
+							close(webConn.send)
+							connIndex.Remove(webConn)
 						}
 					}
 				}
 			case <-h.stop:
-				userIds := make(map[string]bool)
-
-				for _, webCon := range connections.All() {
-					userIds[webCon.UserId] = true
-					webCon.Close()
+				for _, webConn := range connIndex.All() {
+					webConn.Close()
+					h.app.SetStatusOffline(webConn.UserId, false)
 				}

-				for userId := range userIds {
-					h.app.SetStatusOffline(userId, false)
-				}
-
-				h.ExplicitStop = true
+				h.explicitStop = true
 				close(h.didStop)

 				return
@@ -429,7 +472,7 @@ func (h *Hub) Start() {
 	}

 	doRecover = func() {
-		if !h.ExplicitStop {
+		if !h.explicitStop {
 			if r := recover(); r != nil {
 				mlog.Error("Recovering from Hub panic.", mlog.Any("panic", r))
 			} else {
@@ -494,6 +537,11 @@ func (i *hubConnectionIndex) Remove(wc *WebConn) {
 	delete(i.connectionIndexes, wc)
 }

+func (i *hubConnectionIndex) Has(wc *WebConn) bool {
+	_, ok := i.connectionIndexes[wc]
+	return ok
+}
+
 func (i *hubConnectionIndex) ForUser(id string) []*WebConn {
 	return i.connectionsByUserId[id]
 }
--- a/app/web_hub_test.go
+++ b/app/web_hub_test.go
@@ -90,7 +90,7 @@ func TestHubStopRaceCondition(t *testing.T) {

 		hub.UpdateActivity("userId", "sessionToken", 0)

-		for i := 0; i <= BROADCAST_QUEUE_SIZE; i++ {
+		for i := 0; i <= broadcastQueueSize; i++ {
 			hub.Broadcast(model.NewWebSocketEvent("", "", "", "", nil))
 		}

--- a/app/websocket_router.go
+++ b/app/websocket_router.go
@@ -27,13 +27,13 @@ func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) {
 func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketRequest) {
 	if r.Action == "" {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.no_action.app_error", nil, "", http.StatusBadRequest)
-		ReturnWebSocketError(conn, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}

 	if r.Seq <= 0 {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_seq.app_error", nil, "", http.StatusBadRequest)
-		ReturnWebSocketError(conn, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}

@@ -66,28 +66,32 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque
 		wr.app.HubRegister(conn)

 		resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, nil)
-		conn.Send <- resp
+		hub := wr.app.GetHubForUserId(conn.UserId)
+		if hub == nil {
+			return
+		}
+		hub.SendMessage(conn, resp)

 		return
 	}

 	if !conn.IsAuthenticated() {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.not_authenticated.app_error", nil, "", http.StatusUnauthorized)
-		ReturnWebSocketError(conn, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}

 	handler, ok := wr.handlers[r.Action]
 	if !ok {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_action.app_error", nil, "", http.StatusInternalServerError)
-		ReturnWebSocketError(conn, r, err)
+		returnWebSocketError(wr.app, conn, r, err)
 		return
 	}

 	handler.ServeWebSocket(conn, r)
 }

-func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.AppError) {
+func returnWebSocketError(app *App, conn *WebConn, r *model.WebSocketRequest, err *model.AppError) {
 	mlog.Error(
 		"websocket routing error.",
 		mlog.Int64("seq", r.Seq),
@@ -96,8 +100,12 @@ func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.A
 		mlog.Err(err),
 	)

+	hub := app.GetHubForUserId(conn.UserId)
+	if hub == nil {
+		return
+	}
+
 	err.DetailedError = ""
 	errorResp := model.NewWebSocketError(r.Seq, err)
-
-	conn.Send <- errorResp
+	hub.SendMessage(conn, errorResp)
 }
--- a/wsapi/websocket_handler.go
+++ b/wsapi/websocket_handler.go
@@ -24,6 +24,10 @@ type webSocketHandler struct {
 func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketRequest) {
 	mlog.Debug("Websocket request", mlog.String("action", r.Action))

+	hub := wh.app.GetHubForUserId(conn.UserId)
+	if hub == nil {
+		return
+	}
 	session, sessionErr := wh.app.GetSession(conn.GetSessionToken())
 	if sessionErr != nil {
 		mlog.Error(
@@ -36,8 +40,7 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR
 		)
 		sessionErr.DetailedError = ""
 		errResp := model.NewWebSocketError(r.Seq, sessionErr)
-
-		conn.Send <- errResp
+		hub.SendMessage(conn, errResp)
 		return
 	}

@@ -59,14 +62,12 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR
 		)
 		err.DetailedError = ""
 		errResp := model.NewWebSocketError(r.Seq, err)
-
-		conn.Send <- errResp
+		hub.SendMessage(conn, errResp)
 		return
 	}

 	resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, data)
-
-	conn.Send <- resp
+	hub.SendMessage(conn, resp)
 }

 func NewInvalidWebSocketParamError(action string, name string) *model.AppError {