diff --git a/server/channels/app/platform/web_conn.go b/server/channels/app/platform/web_conn.go index 55837d27e5..3e04d1c342 100644 --- a/server/channels/app/platform/web_conn.go +++ b/server/channels/app/platform/web_conn.go @@ -98,9 +98,7 @@ type WebConn struct { deadQueuePointer int // active indicates whether there is an open websocket connection attached // to this webConn or not. - // It is not used as an atomic, because there is no need to. - // So do not use this outside the web hub. - active bool + active atomic.Bool // reuseCount indicates how many times this connection has been reused. // This is used to differentiate between a fresh connection and // a reused connection. @@ -220,7 +218,6 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runn UserId: cfg.Session.UserId, T: cfg.TFunc, Locale: cfg.Locale, - active: cfg.Active, reuseCount: cfg.ReuseCount, endWritePump: make(chan struct{}), pumpFinished: make(chan struct{}), @@ -228,6 +225,7 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runn lastLogTimeSlow: time.Now(), lastLogTimeFull: time.Now(), } + wc.active.Store(cfg.Active) wc.SetSession(&cfg.Session) wc.SetSessionToken(cfg.Session.Token) @@ -295,7 +293,7 @@ func (wc *WebConn) GetConnectionID() string { // are inactive or not. func areAllInactive(conns []*WebConn) bool { for _, conn := range conns { - if conn.active { + if conn.active.Load() { return false } } @@ -474,7 +472,7 @@ func (wc *WebConn) writePump() { continue } - if len(wc.send) >= sendFullWarn && time.Since(wc.lastLogTimeFull) > websocketSuppressWarnThreshold { + if wc.active.Load() && len(wc.send) >= sendFullWarn && time.Since(wc.lastLogTimeFull) > websocketSuppressWarnThreshold { logData := []mlog.Field{ mlog.String("user_id", wc.UserId), mlog.String("conn_id", wc.GetConnectionID()), @@ -727,7 +725,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { case model.WebsocketEventTyping, model.WebsocketEventStatusChange, model.WebsocketEventMultipleChannelsViewed: - if time.Since(wc.lastLogTimeSlow) > websocketSuppressWarnThreshold { + if wc.active.Load() && time.Since(wc.lastLogTimeSlow) > websocketSuppressWarnThreshold { mlog.Warn( "websocket.slow: dropping message", mlog.String("user_id", wc.UserId), diff --git a/server/channels/app/platform/web_hub.go b/server/channels/app/platform/web_hub.go index 5717b042c0..8630cb5bb4 100644 --- a/server/channels/app/platform/web_hub.go +++ b/server/channels/app/platform/web_hub.go @@ -383,7 +383,7 @@ func (h *Hub) Start() { conns := connIndex.ForUser(webSessionMessage.userID) var isRegistered bool for _, conn := range conns { - if !conn.active { + if !conn.active.Load() { continue } if conn.GetSessionToken() == webSessionMessage.sessionToken { @@ -411,7 +411,7 @@ func (h *Hub) Start() { // Mark the current one as active. // There is no need to check if it was inactive or not, // we will anyways need to make it active. - webConn.active = true + webConn.active.Store(true) connIndex.Add(webConn) atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive())) @@ -426,7 +426,7 @@ func (h *Hub) Start() { case webConn := <-h.unregister: // If already removed (via queue full), then removing again becomes a noop. // But if not removed, mark inactive. - webConn.active = false + webConn.active.Store(false) atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive())) @@ -444,7 +444,7 @@ func (h *Hub) Start() { } var latestActivity int64 for _, conn := range conns { - if !conn.active { + if !conn.active.Load() { continue } if conn.lastUserActivityAt > latestActivity { @@ -464,7 +464,7 @@ func (h *Hub) Start() { } case activity := <-h.activity: for _, webConn := range connIndex.ForUser(activity.userID) { - if !webConn.active { + if !webConn.active.Load() { continue } if webConn.GetSessionToken() == activity.sessionToken { @@ -478,9 +478,12 @@ func (h *Hub) Start() { 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), - mlog.String("conn_id", directMsg.conn.GetConnectionID())) + // Don't log the warning if it's an inactive connection. + if directMsg.conn.active.Load() { + mlog.Error("webhub.broadcast: cannot send, closing websocket for user", + mlog.String("user_id", directMsg.conn.UserId), + mlog.String("conn_id", directMsg.conn.GetConnectionID())) + } close(directMsg.conn.send) connIndex.Remove(directMsg.conn) } @@ -497,9 +500,12 @@ func (h *Hub) Start() { select { case webConn.send <- msg: default: - mlog.Error("webhub.broadcast: cannot send, closing websocket for user", - mlog.String("user_id", webConn.UserId), - mlog.String("conn_id", webConn.GetConnectionID())) + // Don't log the warning if it's an inactive connection. + if webConn.active.Load() { + mlog.Error("webhub.broadcast: cannot send, closing websocket for user", + mlog.String("user_id", webConn.UserId), + mlog.String("conn_id", webConn.GetConnectionID())) + } close(webConn.send) connIndex.Remove(webConn) } @@ -639,7 +645,7 @@ func (i *hubConnectionIndex) RemoveInactiveByConnectionID(userID, connectionID s return nil } for _, conn := range i.ForUser(userID) { - if conn.GetConnectionID() == connectionID && !conn.active { + if conn.GetConnectionID() == connectionID && !conn.active.Load() { i.Remove(conn) return conn } @@ -652,7 +658,7 @@ func (i *hubConnectionIndex) RemoveInactiveByConnectionID(userID, connectionID s func (i *hubConnectionIndex) RemoveInactiveConnections() { now := model.GetMillis() for conn := range i.byConnection { - if !conn.active && now-conn.lastUserActivityAt > i.staleThreshold.Milliseconds() { + if !conn.active.Load() && now-conn.lastUserActivityAt > i.staleThreshold.Milliseconds() { i.Remove(conn) } } @@ -664,7 +670,7 @@ func (i *hubConnectionIndex) RemoveInactiveConnections() { func (i *hubConnectionIndex) AllActive() int { cnt := 0 for conn := range i.byConnection { - if conn.active { + if conn.active.Load() { cnt++ } } diff --git a/server/channels/app/platform/web_hub_test.go b/server/channels/app/platform/web_hub_test.go index 7277380d2e..79128ae89c 100644 --- a/server/channels/app/platform/web_hub_test.go +++ b/server/channels/app/platform/web_hub_test.go @@ -379,8 +379,8 @@ func TestHubConnIndexInactive(t *testing.T) { wc1 := &WebConn{ Platform: th.Service, UserId: model.NewId(), - active: true, } + wc1.active.Store(true) wc1.SetConnectionID("conn1") wc1.SetSession(&model.Session{}) @@ -388,16 +388,16 @@ func TestHubConnIndexInactive(t *testing.T) { wc2 := &WebConn{ Platform: th.Service, UserId: model.NewId(), - active: true, } + wc2.active.Store(true) wc2.SetConnectionID("conn2") wc2.SetSession(&model.Session{}) wc3 := &WebConn{ Platform: th.Service, UserId: wc2.UserId, - active: false, } + wc3.active.Store(false) wc3.SetConnectionID("conn3") wc3.SetSession(&model.Session{})