[MM-58500] Turn off PostedAck when the connection is no longer registered (#27212)

* [MM-58500] Turn off PostedAck when the connection is no longer registered

* Expose active and just check for active instead
Этот коммит содержится в:
Devin Binnie
2024-06-03 09:44:53 -04:00
коммит произвёл GitHub
родитель 4163db4e5e
Коммит 4ec50a7ddd
5 изменённых файлов: 42 добавлений и 22 удалений

Просмотреть файл

@@ -107,7 +107,7 @@ type WebConn struct {
deadQueuePointer int deadQueuePointer int
// active indicates whether there is an open websocket connection attached // active indicates whether there is an open websocket connection attached
// to this webConn or not. // to this webConn or not.
active atomic.Bool Active atomic.Bool
// reuseCount indicates how many times this connection has been reused. // reuseCount indicates how many times this connection has been reused.
// This is used to differentiate between a fresh connection and // This is used to differentiate between a fresh connection and
// a reused connection. // a reused connection.
@@ -245,7 +245,7 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runn
lastLogTimeFull: time.Now(), lastLogTimeFull: time.Now(),
originClient: cfg.OriginClient, originClient: cfg.OriginClient,
} }
wc.active.Store(cfg.Active) wc.Active.Store(cfg.Active)
wc.SetSession(&cfg.Session) wc.SetSession(&cfg.Session)
wc.SetSessionToken(cfg.Session.Token) wc.SetSessionToken(cfg.Session.Token)
@@ -555,7 +555,7 @@ func (wc *WebConn) writePump() {
continue continue
} }
if wc.active.Load() && 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{ logData := []mlog.Field{
mlog.String("user_id", wc.UserId), mlog.String("user_id", wc.UserId),
mlog.String("conn_id", wc.GetConnectionID()), mlog.String("conn_id", wc.GetConnectionID()),
@@ -812,7 +812,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
case model.WebsocketEventTyping, case model.WebsocketEventTyping,
model.WebsocketEventStatusChange, model.WebsocketEventStatusChange,
model.WebsocketEventMultipleChannelsViewed: model.WebsocketEventMultipleChannelsViewed:
if wc.active.Load() && time.Since(wc.lastLogTimeSlow) > websocketSuppressWarnThreshold { if wc.Active.Load() && time.Since(wc.lastLogTimeSlow) > websocketSuppressWarnThreshold {
mlog.Warn( mlog.Warn(
"websocket.slow: dropping message", "websocket.slow: dropping message",
mlog.String("user_id", wc.UserId), mlog.String("user_id", wc.UserId),

Просмотреть файл

@@ -389,7 +389,7 @@ func (h *Hub) Start() {
conns := connIndex.ForUser(webSessionMessage.userID) conns := connIndex.ForUser(webSessionMessage.userID)
var isRegistered bool var isRegistered bool
for _, conn := range conns { for _, conn := range conns {
if !conn.active.Load() { if !conn.Active.Load() {
continue continue
} }
if conn.GetSessionToken() == webSessionMessage.sessionToken { if conn.GetSessionToken() == webSessionMessage.sessionToken {
@@ -419,7 +419,7 @@ func (h *Hub) Start() {
// Mark the current one as active. // Mark the current one as active.
// There is no need to check if it was inactive or not, // There is no need to check if it was inactive or not,
// we will anyways need to make it active. // we will anyways need to make it active.
webConn.active.Store(true) webConn.Active.Store(true)
connIndex.Add(webConn) connIndex.Add(webConn)
atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive())) atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive()))
@@ -434,7 +434,7 @@ func (h *Hub) Start() {
case webConn := <-h.unregister: case webConn := <-h.unregister:
// If already removed (via queue full), then removing again becomes a noop. // If already removed (via queue full), then removing again becomes a noop.
// But if not removed, mark inactive. // But if not removed, mark inactive.
webConn.active.Store(false) webConn.Active.Store(false)
atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive())) atomic.StoreInt64(&h.connectionCount, int64(connIndex.AllActive()))
@@ -471,7 +471,7 @@ func (h *Hub) Start() {
} }
var latestActivity int64 var latestActivity int64
for _, conn := range conns { for _, conn := range conns {
if !conn.active.Load() { if !conn.Active.Load() {
continue continue
} }
if conn.lastUserActivityAt > latestActivity { if conn.lastUserActivityAt > latestActivity {
@@ -491,7 +491,7 @@ func (h *Hub) Start() {
} }
case activity := <-h.activity: case activity := <-h.activity:
for _, webConn := range connIndex.ForUser(activity.userID) { for _, webConn := range connIndex.ForUser(activity.userID) {
if !webConn.active.Load() { if !webConn.Active.Load() {
continue continue
} }
if webConn.GetSessionToken() == activity.sessionToken { if webConn.GetSessionToken() == activity.sessionToken {
@@ -506,7 +506,7 @@ func (h *Hub) Start() {
case directMsg.conn.send <- directMsg.msg: case directMsg.conn.send <- directMsg.msg:
default: default:
// Don't log the warning if it's an inactive connection. // Don't log the warning if it's an inactive connection.
if directMsg.conn.active.Load() { if directMsg.conn.Active.Load() {
mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.Error("webhub.broadcast: cannot send, closing websocket for user",
mlog.String("user_id", directMsg.conn.UserId), mlog.String("user_id", directMsg.conn.UserId),
mlog.String("conn_id", directMsg.conn.GetConnectionID())) mlog.String("conn_id", directMsg.conn.GetConnectionID()))
@@ -533,7 +533,7 @@ func (h *Hub) Start() {
case webConn.send <- h.runBroadcastHooks(msg, webConn, broadcastHooks, broadcastHookArgs): case webConn.send <- h.runBroadcastHooks(msg, webConn, broadcastHooks, broadcastHookArgs):
default: default:
// Don't log the warning if it's an inactive connection. // Don't log the warning if it's an inactive connection.
if webConn.active.Load() { if webConn.Active.Load() {
mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.Error("webhub.broadcast: cannot send, closing websocket for user",
mlog.String("user_id", webConn.UserId), mlog.String("user_id", webConn.UserId),
mlog.String("conn_id", webConn.GetConnectionID())) mlog.String("conn_id", webConn.GetConnectionID()))
@@ -601,7 +601,7 @@ func (h *Hub) Start() {
// are inactive or not. // are inactive or not.
func areAllInactive(conns []*WebConn) bool { func areAllInactive(conns []*WebConn) bool {
for _, conn := range conns { for _, conn := range conns {
if conn.active.Load() { if conn.Active.Load() {
return false return false
} }
} }
@@ -689,7 +689,7 @@ func (i *hubConnectionIndex) ForUser(id string) []*WebConn {
func (i *hubConnectionIndex) ForUserActiveCount(id string) int { func (i *hubConnectionIndex) ForUserActiveCount(id string) int {
cnt := 0 cnt := 0
for _, conn := range i.ForUser(id) { for _, conn := range i.ForUser(id) {
if conn.active.Load() { if conn.Active.Load() {
cnt++ cnt++
} }
} }
@@ -714,7 +714,7 @@ func (i *hubConnectionIndex) RemoveInactiveByConnectionID(userID, connectionID s
return nil return nil
} }
for _, conn := range i.ForUser(userID) { for _, conn := range i.ForUser(userID) {
if conn.GetConnectionID() == connectionID && !conn.active.Load() { if conn.GetConnectionID() == connectionID && !conn.Active.Load() {
i.Remove(conn) i.Remove(conn)
return conn return conn
} }
@@ -727,7 +727,7 @@ func (i *hubConnectionIndex) RemoveInactiveByConnectionID(userID, connectionID s
func (i *hubConnectionIndex) RemoveInactiveConnections() { func (i *hubConnectionIndex) RemoveInactiveConnections() {
now := model.GetMillis() now := model.GetMillis()
for conn := range i.byConnection { for conn := range i.byConnection {
if !conn.active.Load() && now-conn.lastUserActivityAt > i.staleThreshold.Milliseconds() { if !conn.Active.Load() && now-conn.lastUserActivityAt > i.staleThreshold.Milliseconds() {
i.Remove(conn) i.Remove(conn)
} }
} }
@@ -739,7 +739,7 @@ func (i *hubConnectionIndex) RemoveInactiveConnections() {
func (i *hubConnectionIndex) AllActive() int { func (i *hubConnectionIndex) AllActive() int {
cnt := 0 cnt := 0
for conn := range i.byConnection { for conn := range i.byConnection {
if conn.active.Load() { if conn.Active.Load() {
cnt++ cnt++
} }
} }

Просмотреть файл

@@ -429,7 +429,7 @@ func TestHubConnIndexInactive(t *testing.T) {
Platform: th.Service, Platform: th.Service,
UserId: model.NewId(), UserId: model.NewId(),
} }
wc1.active.Store(true) wc1.Active.Store(true)
wc1.SetConnectionID("conn1") wc1.SetConnectionID("conn1")
wc1.SetSession(&model.Session{}) wc1.SetSession(&model.Session{})
@@ -438,7 +438,7 @@ func TestHubConnIndexInactive(t *testing.T) {
Platform: th.Service, Platform: th.Service,
UserId: model.NewId(), UserId: model.NewId(),
} }
wc2.active.Store(true) wc2.Active.Store(true)
wc2.SetConnectionID("conn2") wc2.SetConnectionID("conn2")
wc2.SetSession(&model.Session{}) wc2.SetSession(&model.Session{})
@@ -446,7 +446,7 @@ func TestHubConnIndexInactive(t *testing.T) {
Platform: th.Service, Platform: th.Service,
UserId: wc2.UserId, UserId: wc2.UserId,
} }
wc3.active.Store(false) wc3.Active.Store(false)
wc3.SetConnectionID("conn3") wc3.SetConnectionID("conn3")
wc3.SetSession(&model.Session{}) wc3.SetSession(&model.Session{})

Просмотреть файл

@@ -83,7 +83,7 @@ func usePostedAckHook(message *model.WebSocketEvent, postedUserId string, channe
func (h *postedAckBroadcastHook) Process(msg *platform.HookedWebSocketEvent, webConn *platform.WebConn, args map[string]any) error { func (h *postedAckBroadcastHook) Process(msg *platform.HookedWebSocketEvent, webConn *platform.WebConn, args map[string]any) error {
// Don't ACK unless we say to explicitly // Don't ACK unless we say to explicitly
if !webConn.PostedAck { if !(webConn.PostedAck && webConn.Active.Load()) {
return nil return nil
} }

Просмотреть файл

@@ -91,6 +91,7 @@ func TestPostedAckHook_Process(t *testing.T) {
Platform: &platform.PlatformService{}, Platform: &platform.PlatformService{},
PostedAck: true, PostedAck: true,
} }
webConn.Active.Store(true)
webConn.SetSession(&model.Session{}) webConn.SetSession(&model.Session{})
t.Run("should ack if user is in the list of users to notify", func(t *testing.T) { t.Run("should ack if user is in the list of users to notify", func(t *testing.T) {
@@ -142,14 +143,33 @@ func TestPostedAckHook_Process(t *testing.T) {
}) })
t.Run("should not ack if posted ack is false", func(t *testing.T) { t.Run("should not ack if posted ack is false", func(t *testing.T) {
mobileWebConn := &platform.WebConn{ noAckWebConn := &platform.WebConn{
UserId: userID, UserId: userID,
Platform: &platform.PlatformService{}, Platform: &platform.PlatformService{},
PostedAck: false, PostedAck: false,
} }
noAckWebConn.Active.Store(true)
msg := platform.MakeHookedWebSocketEvent(model.NewWebSocketEvent(model.WebsocketEventPosted, "", "", "", nil, "")) msg := platform.MakeHookedWebSocketEvent(model.NewWebSocketEvent(model.WebsocketEventPosted, "", "", "", nil, ""))
hook.Process(msg, mobileWebConn, map[string]any{ hook.Process(msg, noAckWebConn, map[string]any{
"posted_user_id": model.NewId(),
"channel_type": model.ChannelTypeDirect,
"users": []string{},
})
assert.Nil(t, msg.Event().GetData()["should_ack"])
})
t.Run("should not ack if connection is not active", func(t *testing.T) {
inactiveWebConn := &platform.WebConn{
UserId: userID,
Platform: &platform.PlatformService{},
PostedAck: false,
}
inactiveWebConn.Active.Store(true)
msg := platform.MakeHookedWebSocketEvent(model.NewWebSocketEvent(model.WebsocketEventPosted, "", "", "", nil, ""))
hook.Process(msg, inactiveWebConn, map[string]any{
"posted_user_id": model.NewId(), "posted_user_id": model.NewId(),
"channel_type": model.ChannelTypeDirect, "channel_type": model.ChannelTypeDirect,
"users": []string{}, "users": []string{},