[release-10.11] Cherry pick of #36332 (#36639)

Automatic Merge

server/channels/app/platform/cluster_handlers.go

@@ -66,11 +66,18 @@ func (ps *PlatformService) ClearSessionCacheForUserSkipClusterSend(userID string
 	ps.invalidateWebConnSessionCacheForUserSkipClusterSend(userID)
 }
 
-func (ps *PlatformService) ClearSessionCacheForAllUsersSkipClusterSend() {
+// ClearSessionCacheForAllUsersSkipClusterSend purges the in-memory
+// session cache and invalidates every WebConn on this node. The hub
+// fan-out runs even if the cache purge fails; the purge error is
+// returned so wrappers can propagate it.
+func (ps *PlatformService) ClearSessionCacheForAllUsersSkipClusterSend() error {
 	ps.logger.Info("Purging sessions cache")
-	if err := ps.ClearAllUsersSessionCacheLocal(); err != nil {
+	err := ps.ClearAllUsersSessionCacheLocal()
+	if err != nil {
 		ps.logger.Error("Failed to purge session cache", mlog.Err(err))
 	}
+	ps.invalidateWebConnSessionCacheForAllUsersSkipClusterSend()
+	return err
 }
 
 func (ps *PlatformService) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) {
@@ -78,7 +85,9 @@ func (ps *PlatformService) clusterClearSessionCacheForUserHandler(msg *model.Clu
 }
 
 func (ps *PlatformService) clusterClearSessionCacheForAllUsersHandler(msg *model.ClusterMessage) {
-	ps.ClearSessionCacheForAllUsersSkipClusterSend()
+	if err := ps.ClearSessionCacheForAllUsersSkipClusterSend(); err != nil {
+		ps.logger.Error("Failed to clear session cache for all users from cluster handler", mlog.Err(err))
+	}
 }
 
 func (ps *PlatformService) clusterBusyStateChgHandler(msg *model.ClusterMessage) {
@@ -102,6 +111,17 @@ func (ps *PlatformService) invalidateWebConnSessionCacheForUserSkipClusterSend(u
 	}
 }
 
+// invalidateWebConnSessionCacheForAllUsersSkipClusterSend signals
+// every hub on this node to invalidate the cached session state of
+// all of its WebConns. Companion to ClearAllUsersSessionCacheLocal.
+func (ps *PlatformService) invalidateWebConnSessionCacheForAllUsersSkipClusterSend() {
+	for _, hub := range ps.hubs {
+		if hub != nil {
+			hub.InvalidateAll()
+		}
+	}
+}
+
 func (ps *PlatformService) InvalidateAllCachesSkipSend() *model.AppError {
 	ps.logger.Info("Purging all caches")
 	if err := ps.ClearAllUsersSessionCacheLocal(); err != nil {

server/channels/app/platform/cluster_handlers_revoke_test.go

@@ -0,0 +1,97 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package platform
+
+import (
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+// TestRevokeSessionsFromAllUsersInvalidatesWebConnSession asserts that
+// after the public RevokeSessionsFromAllUsers entry point returns,
+// every live WebConn on this node — across multiple users (hashed to
+// different hubs) and multiple connections per user — has its cached
+// session reset to the authenticated-as-no-one state.
+func TestRevokeSessionsFromAllUsersInvalidatesWebConnSession(t *testing.T) {
+	mainHelper.Parallel(t)
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	s := httptest.NewServer(dummyWebsocketHandler(t))
+	defer s.Close()
+
+	// Spread connections across multiple hubs via GetHubForUserId's
+	// hash(userID) mod len(hubs) sharding, and use multiple conns per
+	// user to cover the multi-device case.
+	type userConns struct {
+		userID string
+		wcs    []*WebConn
+	}
+	users := []*userConns{
+		{userID: th.BasicUser.Id},
+		{userID: th.BasicUser2.Id},
+	}
+	for range 4 {
+		users = append(users, &userConns{userID: model.NewId()})
+	}
+
+	for _, u := range users {
+		preWarmStatusOnline(th, u.userID)
+	}
+
+	const connsPerUser = 2
+	for _, u := range users {
+		for range connsPerUser {
+			session, err := th.Service.CreateSession(th.Context, &model.Session{
+				UserId: u.userID,
+			})
+			require.NoError(t, err)
+
+			session.ExpiresAt = model.GetMillis() + time.Hour.Milliseconds()
+
+			wc := registerDummyWebConn(t, th, s.Listener.Addr(), session)
+			t.Cleanup(func() { wc.Close() })
+			u.wcs = append(u.wcs, wc)
+
+			waitForWebConnRegistered(t, th, session)
+		}
+	}
+
+	for _, u := range users {
+		for _, wc := range u.wcs {
+			require.NotNil(t, wc.GetSession(),
+				"precondition: webconn for user %q must have a cached session before revoke", u.userID)
+			require.Greater(t, wc.GetSessionExpiresAt(), model.GetMillis(),
+				"precondition: cached expiry for user %q must be in the future before revoke", u.userID)
+			require.NotEmpty(t, wc.GetSessionToken(),
+				"precondition: webconn for user %q must have a cached session token before revoke", u.userID)
+		}
+	}
+
+	require.NoError(t, th.Service.RevokeSessionsFromAllUsers(),
+		"RevokeSessionsFromAllUsers should not error")
+
+	require.Eventually(t, func() bool {
+		for _, u := range users {
+			for _, wc := range u.wcs {
+				if wc.GetSession() != nil {
+					return false
+				}
+				if wc.GetSessionExpiresAt() != 0 {
+					return false
+				}
+				if wc.GetSessionToken() != "" {
+					return false
+				}
+			}
+		}
+		return true
+	}, 5*time.Second, 25*time.Millisecond,
+		"RevokeSessionsFromAllUsers did not invalidate every live WebConn across all hubs")
+}

server/channels/app/platform/cluster_handlers_test.go

@@ -0,0 +1,112 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package platform
+
+import (
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+// waitForWebConnRegistered blocks until the hub has processed the
+// WebConn registration. Without it, a test can race past the async
+// register channel and signal invalidation against an empty connIndex.
+func waitForWebConnRegistered(t *testing.T, th *TestHelper, session *model.Session) {
+	t.Helper()
+	require.Eventually(t, func() bool {
+		return th.Service.SessionIsRegistered(*session)
+	}, 2*time.Second, 10*time.Millisecond,
+		"WebConn for session %q (user %q) was not registered with the hub in time",
+		session.Id, session.UserId)
+}
+
+// preWarmStatusOnline marks the user online before any WebConn is
+// created so the async SetStatusOnline goroutine in NewWebConn skips
+// the broadcast path. Otherwise the broadcast can race with the
+// invalidation and re-populate the WebConn's cached session via
+// IsBasicAuthenticated, flipping the post-invalidate assertions.
+func preWarmStatusOnline(th *TestHelper, userID string) {
+	th.Service.AddStatusCacheSkipClusterSend(&model.Status{
+		UserId:         userID,
+		Status:         model.StatusOnline,
+		LastActivityAt: model.GetMillis(),
+	})
+}
+
+// TestClearSessionCacheInvalidatesWebConnSession asserts that after either
+// the per-user or the global session-cache clear runs, every matching
+// active WebSocket connection has its cached session reset to the
+// authenticated-as-no-one state (GetSession() == nil and
+// GetSessionExpiresAt() == 0).
+func TestClearSessionCacheInvalidatesWebConnSession(t *testing.T) {
+	mainHelper.Parallel(t)
+
+	tests := []struct {
+		name   string
+		revoke func(ps *PlatformService, userID string)
+	}{
+		{
+			name: "PerUserRevokeInvalidatesWebConnSession",
+			revoke: func(ps *PlatformService, userID string) {
+				ps.ClearSessionCacheForUserSkipClusterSend(userID)
+			},
+		},
+		{
+			name: "GlobalRevokeInvalidatesWebConnSession",
+			revoke: func(ps *PlatformService, _ string) {
+				_ = ps.ClearSessionCacheForAllUsersSkipClusterSend()
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			th := Setup(t).InitBasic()
+			defer th.TearDown()
+
+			s := httptest.NewServer(dummyWebsocketHandler(t))
+			defer s.Close()
+
+			session, err := th.Service.CreateSession(th.Context, &model.Session{
+				UserId: th.BasicUser.Id,
+			})
+			require.NoError(t, err)
+
+			// Pin a future expiry so IsBasicAuthenticated trusts the
+			// cached session and doesn't re-validate against the store.
+			session.ExpiresAt = model.GetMillis() + time.Hour.Milliseconds()
+
+			preWarmStatusOnline(th, th.BasicUser.Id)
+
+			wc := registerDummyWebConn(t, th, s.Listener.Addr(), session)
+			defer wc.Close()
+
+			waitForWebConnRegistered(t, th, session)
+
+			require.NotNil(t, wc.GetSession(),
+				"precondition: webconn must have a cached session before revoke")
+			require.Greater(t, wc.GetSessionExpiresAt(), model.GetMillis(),
+				"precondition: webconn cached session expiry must be in the future before revoke")
+
+			tt.revoke(th.Service, th.BasicUser.Id)
+
+			// Hub invalidation is async, so poll for the end state.
+			require.Eventually(t, func() bool {
+				return wc.GetSession() == nil && wc.GetSessionExpiresAt() == 0
+			}, 2*time.Second, 25*time.Millisecond,
+				"webconn cached session was not invalidated after %s; "+
+					"expected GetSession()==nil and GetSessionExpiresAt()==0, "+
+					"but got GetSession()!=nil=%t, GetSessionExpiresAt()=%d, GetSessionToken()=%q",
+				tt.name,
+				wc.GetSession() != nil,
+				wc.GetSessionExpiresAt(),
+				wc.GetSessionToken(),
+			)
+		})
+	}
+}

server/channels/app/platform/session.go

@@ -110,9 +110,11 @@ func (ps *PlatformService) ClearUserSessionCache(userID string) {
 }
 
 func (ps *PlatformService) ClearAllUsersSessionCache() error {
-	if err := ps.ClearAllUsersSessionCacheLocal(); err != nil {
-		return err
-	}
+	// Mirrors the per-user shape: the SkipClusterSend helper handles the
+	// local cache purge and the WebConn hub fan-out, then we broadcast to
+	// peer nodes. The broadcast still runs on local-purge failure so peers
+	// can act independently.
+	err := ps.ClearSessionCacheForAllUsersSkipClusterSend()
 
 	if ps.clusterIFace != nil {
 		msg := &model.ClusterMessage{
@@ -121,7 +123,7 @@ func (ps *PlatformService) ClearAllUsersSessionCache() error {
 		}
 		ps.clusterIFace.SendClusterMessage(msg)
 	}
-	return nil
+	return err
 }
 
 func (ps *PlatformService) GetSession(c request.CTX, token string) (*model.Session, error) {

server/channels/app/platform/web_hub.go

@@ -84,6 +84,7 @@ type Hub struct {
 	stop            chan struct{}
 	didStop         chan struct{}
 	invalidateUser  chan string
+	invalidateAll   chan struct{}
 	activity        chan *webConnActivityMessage
 	directMsg       chan *webConnDirectMessage
 	explicitStop    bool
@@ -106,6 +107,7 @@ func newWebHub(ps *PlatformService) *Hub {
 		stop:            make(chan struct{}),
 		didStop:         make(chan struct{}),
 		invalidateUser:  make(chan string),
+		invalidateAll:   make(chan struct{}),
 		activity:        make(chan *webConnActivityMessage),
 		directMsg:       make(chan *webConnDirectMessage),
 		checkRegistered: make(chan *webConnSessionMessage),
@@ -453,6 +455,15 @@ func (h *Hub) InvalidateUser(userID string) {
 	}
 }
 
+// InvalidateAll invalidates the cached session state of every WebConn
+// registered with this hub. Global counterpart of InvalidateUser.
+func (h *Hub) InvalidateAll() {
+	select {
+	case h.invalidateAll <- struct{}{}:
+	case <-h.stop:
+	}
+}
+
 // UpdateActivity sets the LastUserActivityAt field for the connection
 // of the user.
 func (h *Hub) UpdateActivity(userID, sessionToken string, activityAt int64) {
@@ -654,6 +665,18 @@ func (h *Hub) Start() {
 						closeAndRemoveConn(connIndex, webConn)
 					}
 				}
+			case <-h.invalidateAll:
+				// Mirrors the invalidateUser arm across every conn,
+				// also clearing the session token so the next
+				// IsBasicAuthenticated check short-circuits instead
+				// of re-fetching from the cache.
+				for webConn := range connIndex.All() {
+					webConn.InvalidateCache()
+					webConn.SetSessionToken("")
+				}
+				if *h.platform.Config().ServiceSettings.EnableWebHubChannelIteration {
+					connIndex.clearChannels()
+				}
 			case activity := <-h.activity:
 				for webConn := range connIndex.ForUser(activity.userID) {
 					if !webConn.Active.Load() {
@@ -947,6 +970,16 @@ func (i *hubConnectionIndex) ForChannel(channelID string) iter.Seq[*WebConn] {
 	return maps.Keys(i.byChannelID[channelID])
 }
 
+// clearChannels empties the channel-routing index in one shot. Intended
+// for paths that have already invalidated every conn registered with
+// the hub: any broadcast addressed to a channel will be filtered out
+// upstream by ShouldSendEvent, so the routing entries are dead weight
+// until conns either re-handshake or fully reconnect (both of which
+// repopulate the index via Add).
+func (i *hubConnectionIndex) clearChannels() {
+	clear(i.byChannelID)
+}
+
 // ForUserActiveCount returns the number of active connections for a userID
 func (i *hubConnectionIndex) ForUserActiveCount(id string) int {
 	cnt := 0

server/channels/app/session.go

@@ -241,7 +241,9 @@ func (a *App) ClearSessionCacheForUserSkipClusterSend(userID string) {
 }
 
 func (a *App) ClearSessionCacheForAllUsersSkipClusterSend() {
-	a.Srv().Platform().ClearSessionCacheForAllUsersSkipClusterSend()
+	if err := a.Srv().Platform().ClearSessionCacheForAllUsersSkipClusterSend(); err != nil {
+		a.Srv().Platform().Log().Error("Failed to clear session cache for all users", mlog.Err(err))
+	}
 }
 
 func (a *App) RevokeSessionsForDeviceId(c request.CTX, userID string, deviceID string, currentSessionId string) *model.AppError {