From 847f42fd4c5957b30a67153147f9d7109031c01a Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Tue, 12 Nov 2024 08:46:18 +0530 Subject: [PATCH] MM-61700: Fix race conditions from web_hub initialization (#29214) The Store variable was written to _after_ the server started which was causing the race. We simply move it to before we start the server. While we are here, we fix yet another race condition which was unrelated, but doing it in one sweep. This was related to the user props access. When a user is updated, there was 3 ws events that get sent out, 2 were deep-copied whereas 1 was not. This led to race condition in postgres binary-param mode where we were trying to set the user props. https://mattermost.atlassian.net/browse/MM-61700 --- server/channels/app/bot.go | 2 +- server/channels/app/platform/helper_test.go | 21 ++++++++++++--------- server/channels/app/shared_channel.go | 2 +- server/channels/app/user.go | 21 ++++++++++++--------- 4 files changed, 26 insertions(+), 20 deletions(-) diff --git a/server/channels/app/bot.go b/server/channels/app/bot.go index 29196f69a1..677efdd325 100644 --- a/server/channels/app/bot.go +++ b/server/channels/app/bot.go @@ -308,7 +308,7 @@ func (a *App) PatchBot(rctx request.CTX, botUserId string, botPatch *model.BotPa a.InvalidateCacheForUser(user.Id) ruser := userUpdate.New - a.sendUpdatedUserEvent(*ruser) + a.sendUpdatedUserEvent(ruser) bot, nErr = a.Srv().Store().Bot().Update(bot) if nErr != nil { diff --git a/server/channels/app/platform/helper_test.go b/server/channels/app/platform/helper_test.go index 8b772e20aa..cc1b7ba22b 100644 --- a/server/channels/app/platform/helper_test.go +++ b/server/channels/app/platform/helper_test.go @@ -106,15 +106,6 @@ func SetupWithStoreMock(tb testing.TB, options ...Option) *TestHelper { mockStore := testlib.GetMockStoreForSetupFunctions() options = append(options, StoreOverride(mockStore)) th := setupTestHelper(mockStore, false, false, tb, options...) - statusMock := mocks.StatusStore{} - statusMock.On("UpdateExpiredDNDStatuses").Return([]*model.Status{}, nil) - statusMock.On("Get", "user1").Return(&model.Status{UserId: "user1", Status: model.StatusOnline}, nil) - statusMock.On("UpdateLastActivityAt", "user1", mock.Anything).Return(nil) - statusMock.On("SaveOrUpdate", mock.AnythingOfType("*model.Status")).Return(nil) - emptyMockStore := mocks.Store{} - emptyMockStore.On("Close").Return(nil) - emptyMockStore.On("Status").Return(&statusMock) - th.Service.Store = &emptyMockStore return th } @@ -170,6 +161,18 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo Suite: &mockSuite{}, } + if _, ok := dbStore.(*mocks.Store); ok { + statusMock := mocks.StatusStore{} + statusMock.On("UpdateExpiredDNDStatuses").Return([]*model.Status{}, nil) + statusMock.On("Get", "user1").Return(&model.Status{UserId: "user1", Status: model.StatusOnline}, nil) + statusMock.On("UpdateLastActivityAt", "user1", mock.Anything).Return(nil) + statusMock.On("SaveOrUpdate", mock.AnythingOfType("*model.Status")).Return(nil) + emptyMockStore := mocks.Store{} + emptyMockStore.On("Close").Return(nil) + emptyMockStore.On("Status").Return(&statusMock) + th.Service.Store = &emptyMockStore + } + // Share same configuration with app.TestHelper th.Service.UpdateConfig(func(cfg *model.Config) { *cfg.TeamSettings.MaxUsersPerTeam = 50 diff --git a/server/channels/app/shared_channel.go b/server/channels/app/shared_channel.go index ea736916a6..71d2252178 100644 --- a/server/channels/app/shared_channel.go +++ b/server/channels/app/shared_channel.go @@ -178,7 +178,7 @@ func (a *App) GetSharedChannelRemotesStatus(channelID string) ([]*model.SharedCh // SharedChannelUsers func (a *App) NotifySharedChannelUserUpdate(user *model.User) { - a.sendUpdatedUserEvent(*user) + a.sendUpdatedUserEvent(user) } // onUserProfileChange is called when a user's profile has changed diff --git a/server/channels/app/user.go b/server/channels/app/user.go index abc7b3c101..335a2e5828 100644 --- a/server/channels/app/user.go +++ b/server/channels/app/user.go @@ -292,7 +292,7 @@ func (a *App) createUserOrGuest(c request.CTX, user *model.User, guest bool) (*m } } - a.sendUpdatedUserEvent(*nUser) + a.sendUpdatedUserEvent(nUser) } recommendedNextStepsPref := model.Preference{UserId: ruser.Id, Category: model.PreferenceCategoryRecommendedNextSteps, Name: model.PreferenceNameRecommendedNextStepsHide, Value: "false"} @@ -1043,7 +1043,7 @@ func (a *App) UpdateActive(c request.CTX, user *model.User, active bool) (*model a.invalidateUserChannelMembersCaches(c, user.Id) a.InvalidateCacheForUser(user.Id) - a.sendUpdatedUserEvent(*ruser) + a.sendUpdatedUserEvent(ruser) if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil && !active && user.DeleteAt != 0 { a.Srv().Go(func() { @@ -1178,11 +1178,14 @@ func (a *App) UpdateUserAuth(c request.CTX, userID string, userAuth *model.UserA return userAuth, nil } -func (a *App) sendUpdatedUserEvent(user model.User) { +func (a *App) sendUpdatedUserEvent(user *model.User) { // exclude event creator user from admin, member user broadcast omitUsers := make(map[string]bool, 1) omitUsers[user.Id] = true + // First, creating a base copy to avoid race conditions + // from setting the binaryParamKey in userstore.Update. + user = user.DeepCopy() // declare admin and unsanitized copy of user adminCopyOfUser := user.DeepCopy() unsanitizedCopyOfUser := user.DeepCopy() @@ -1193,9 +1196,9 @@ func (a *App) sendUpdatedUserEvent(user model.User) { adminMessage.GetBroadcast().ContainsSensitiveData = true a.Publish(adminMessage) - a.SanitizeProfile(&user, false) + a.SanitizeProfile(user, false) message := model.NewWebSocketEvent(model.WebsocketEventUserUpdated, "", "", "", omitUsers, "") - message.Add("user", &user) + message.Add("user", user) message.GetBroadcast().ContainsSanitizedData = true a.Publish(message) @@ -1335,7 +1338,7 @@ func (a *App) UpdateUser(c request.CTX, user *model.User, sendNotifications bool } }) } - a.sendUpdatedUserEvent(*newUser) + a.sendUpdatedUserEvent(newUser) } a.InvalidateCacheForUser(user.Id) @@ -2023,7 +2026,7 @@ func (a *App) VerifyUserEmail(userID, email string) *model.AppError { return err } - a.sendUpdatedUserEvent(*user) + a.sendUpdatedUserEvent(user) return nil } @@ -2406,7 +2409,7 @@ func (a *App) PromoteGuestToUser(c request.CTX, user *model.User, requestorId st if err != nil { c.Logger().Warn("Failed to get user on promote guest to user", mlog.Err(err)) } else { - a.sendUpdatedUserEvent(*promotedUser) + a.sendUpdatedUserEvent(promotedUser) if uErr := a.ch.srv.platform.UpdateSessionsIsGuest(c, promotedUser, promotedUser.IsGuest()); uErr != nil { c.Logger().Warn("Unable to update user sessions", mlog.String("user_id", promotedUser.Id), mlog.Err(uErr)) } @@ -2451,7 +2454,7 @@ func (a *App) DemoteUserToGuest(c request.CTX, user *model.User) *model.AppError return model.NewAppError("DemoteUserToGuest", "app.user.demote_user_to_guest.user_update.app_error", nil, "", http.StatusInternalServerError).Wrap(nErr) } - a.sendUpdatedUserEvent(*demotedUser) + a.sendUpdatedUserEvent(demotedUser) if uErr := a.ch.srv.platform.UpdateSessionsIsGuest(c, demotedUser, demotedUser.IsGuest()); uErr != nil { c.Logger().Warn("Unable to update user sessions", mlog.String("user_id", demotedUser.Id), mlog.Err(uErr)) }