From 1e3b6b56a6cea6b1023c76aa060574aae170ff69 Mon Sep 17 00:00:00 2001 From: John Tzikas Date: Wed, 10 Feb 2021 15:51:04 +0200 Subject: [PATCH] Optimise creation of DM (#16819) * Optimise creation of dm * Handle direct channels with the same user * Cover GetMany with specs and add it on tha cache layer as well * Fix specs by handling user dming themselves * Apply PR suggestions * Apply PR suggestions * Use require.NoError instead of require.Nil on userstore test * Improve readability of GetOrCreateDirectChannel * Apply PR suggestions * Update layers Co-authored-by: Mattermod --- app/app_iface.go | 2 +- app/channel.go | 129 ++++++++++++--------- app/opentracing/opentracing_layer.go | 4 +- store/localcachelayer/main_test.go | 10 ++ store/localcachelayer/user_layer.go | 65 +++++++++++ store/localcachelayer/user_layer_test.go | 51 ++++++++ store/opentracinglayer/opentracinglayer.go | 18 +++ store/retrylayer/retrylayer.go | 20 ++++ store/sqlstore/user_store.go | 16 +++ store/store.go | 1 + store/storetest/mocks/UserStore.go | 23 ++++ store/timerlayer/timerlayer.go | 16 +++ 12 files changed, 297 insertions(+), 58 deletions(-) diff --git a/app/app_iface.go b/app/app_iface.go index 4ab47dc4f8..4e96506893 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -622,7 +622,7 @@ type AppIface interface { GetOAuthSignupEndpoint(w http.ResponseWriter, r *http.Request, service, teamID string) (string, *model.AppError) GetOAuthStateToken(token string) (*model.Token, *model.AppError) GetOpenGraphMetadata(requestURL string) *opengraph.OpenGraph - GetOrCreateDirectChannel(userID, otherUserId string) (*model.Channel, *model.AppError) + GetOrCreateDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError) GetOutgoingWebhook(hookID string) (*model.OutgoingWebhook, *model.AppError) GetOutgoingWebhooksForChannelPageByUser(channelId string, userID string, page, perPage int) ([]*model.OutgoingWebhook, *model.AppError) GetOutgoingWebhooksForTeamPage(teamID string, page, perPage int) ([]*model.OutgoingWebhook, *model.AppError) diff --git a/app/channel.go b/app/channel.go index 109e9a800d..0d39c7cfd0 100644 --- a/app/channel.go +++ b/app/channel.go @@ -320,71 +320,76 @@ func (a *App) CreateChannel(channel *model.Channel, addMember bool) (*model.Chan return sc, nil } -func (a *App) GetOrCreateDirectChannel(userID, otherUserId string) (*model.Channel, *model.AppError) { - channel, nErr := a.Srv().Store.Channel().GetByName("", model.GetDMNameFromIds(userID, otherUserId), true) +func (a *App) GetOrCreateDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError) { + channel, nErr := a.getDirectChannel(userID, otherUserID) if nErr != nil { - var nfErr *store.ErrNotFound - if errors.As(nErr, &nfErr) { - var err *model.AppError - channel, err = a.createDirectChannel(userID, otherUserId) - if err != nil { - if err.Id == store.ChannelExistsError { - return channel, nil - } - return nil, err - } + return nil, nErr + } - a.WaitForChannelMembership(channel.Id, userID) - - a.InvalidateCacheForUser(userID) - a.InvalidateCacheForUser(otherUserId) - - if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil { - a.Srv().Go(func() { - pluginContext := a.PluginContext() - pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool { - hooks.ChannelHasBeenCreated(pluginContext, channel) - return true - }, plugin.ChannelHasBeenCreatedId) - }) - } - - message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_DIRECT_ADDED, "", channel.Id, "", nil) - message.Add("teammate_id", otherUserId) - a.Publish(message) + if channel != nil { + return channel, nil + } + channel, err := a.createDirectChannel(userID, otherUserID) + if err != nil { + if err.Id == store.ChannelExistsError { return channel, nil } - return nil, model.NewAppError("GetOrCreateDirectChannel", "web.incoming_webhook.channel.app_error", nil, nErr.Error(), http.StatusInternalServerError) + return nil, err } + + a.WaitForChannelMembership(channel.Id, userID) + + a.InvalidateCacheForUser(userID) + a.InvalidateCacheForUser(otherUserID) + + if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil { + a.Srv().Go(func() { + pluginContext := a.PluginContext() + pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool { + hooks.ChannelHasBeenCreated(pluginContext, channel) + return true + }, plugin.ChannelHasBeenCreatedId) + }) + } + + message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_DIRECT_ADDED, "", channel.Id, "", nil) + message.Add("teammate_id", otherUserID) + a.Publish(message) + return channel, nil } -func (a *App) createDirectChannel(userID string, otherUserId string) (*model.Channel, *model.AppError) { - uc1 := make(chan store.StoreResult, 1) - uc2 := make(chan store.StoreResult, 1) - go func() { - user, err := a.Srv().Store.User().Get(userID) - uc1 <- store.StoreResult{Data: user, NErr: err} - close(uc1) - }() - go func() { - user, err := a.Srv().Store.User().Get(otherUserId) - uc2 <- store.StoreResult{Data: user, NErr: err} - close(uc2) - }() - - result := <-uc1 - if result.NErr != nil { - return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, userID, http.StatusBadRequest) +func (a *App) createDirectChannel(userID string, otherUserID string) (*model.Channel, *model.AppError) { + users, err := a.Srv().Store.User().GetMany([]string{userID, otherUserID}) + if err != nil { + return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, err.Error(), http.StatusBadRequest) } - user := result.Data.(*model.User) - result = <-uc2 - if result.NErr != nil { - return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, otherUserId, http.StatusBadRequest) + if len(users) == 0 { + return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, fmt.Sprintf("No users found for ids: %s. %s", userID, otherUserID), http.StatusBadRequest) + } + + // We are doing this because we allow a user to create a direct channel with themselves + if userID == otherUserID { + users = append(users, users[0]) + } + + // After we counted for direct channels with the same user, if we do not have two users then we failed to find one + if len(users) != 2 { + return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, fmt.Sprintf("No users found for ids: %s. %s", userID, otherUserID), http.StatusBadRequest) + } + + // The potential swap dance bellow is necessary in order to guarantee determinism when creating a direct channel. + // When we query the database for some given user ids, the database result is not deterministic, meaning we can get + // the same results but in different order. In order to conform the contract of Channel.CreateDirectChannel method + // bellow we need to identify which user is who. + user := users[0] + otherUser := users[1] + if user.Id != userID { + user = users[1] + otherUser = users[0] } - otherUser := result.Data.(*model.User) channel, nErr := a.Srv().Store.Channel().CreateDirectChannel(user, otherUser) if nErr != nil { @@ -421,8 +426,8 @@ func (a *App) createDirectChannel(userID string, otherUserId string) (*model.Cha if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(userID, channel.Id, model.GetMillis()); err != nil { return nil, model.NewAppError("CreateDirectChannel", "app.channel_member_history.log_join_event.internal_error", nil, err.Error(), http.StatusInternalServerError) } - if userID != otherUserId { - if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(otherUserId, channel.Id, model.GetMillis()); err != nil { + if userID != otherUserID { + if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(otherUserID, channel.Id, model.GetMillis()); err != nil { return nil, model.NewAppError("CreateDirectChannel", "app.channel_member_history.log_join_event.internal_error", nil, err.Error(), http.StatusInternalServerError) } } @@ -2960,3 +2965,17 @@ func (a *App) GetMemberCountsByGroup(channelID string, includeTimezones bool) ([ return channelMemberCounts, nil } + +func (a *App) getDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError) { + channel, nErr := a.Srv().Store.Channel().GetByName("", model.GetDMNameFromIds(userID, otherUserID), true) + if nErr != nil { + var nfErr *store.ErrNotFound + if errors.As(nErr, &nfErr) { + return nil, nil + } + + return nil, model.NewAppError("GetOrCreateDirectChannel", "web.incoming_webhook.channel.app_error", nil, nErr.Error(), http.StatusInternalServerError) + } + + return channel, nil +} diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 8455aee48c..27df8c0573 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -6726,7 +6726,7 @@ func (a *OpenTracingAppLayer) GetOpenGraphMetadata(requestURL string) *opengraph return resultVar0 } -func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserId string) (*model.Channel, *model.AppError) { +func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserID string) (*model.Channel, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetOrCreateDirectChannel") @@ -6738,7 +6738,7 @@ func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserI }() defer span.Finish() - resultVar0, resultVar1 := a.app.GetOrCreateDirectChannel(userID, otherUserId) + resultVar0, resultVar1 := a.app.GetOrCreateDirectChannel(userID, otherUserID) if resultVar1 != nil { span.LogFields(spanlog.Error(resultVar1)) diff --git a/store/localcachelayer/main_test.go b/store/localcachelayer/main_test.go index 096b6a29c0..25169d0eb3 100644 --- a/store/localcachelayer/main_test.go +++ b/store/localcachelayer/main_test.go @@ -142,6 +142,16 @@ func getMockStore() *mocks.Store { mockUserStore.On("GetAllProfilesInChannel", "123", false).Return(fakeProfilesInChannelMap, nil) mockUserStore.On("Get", "123").Return(fakeUser[0], nil) + users := []*model.User{ + fakeUser[0], + { + Id: "456", + AuthData: model.NewString("authData"), + AuthService: "authService", + }, + } + mockUserStore.On("GetMany", []string{"123", "456"}).Return(users, nil) + mockUserStore.On("GetMany", []string{"123"}).Return(users[0:1], nil) mockStore.On("User").Return(&mockUserStore) fakeUserTeamIds := []string{"1", "2", "3"} diff --git a/store/localcachelayer/user_layer.go b/store/localcachelayer/user_layer.go index ea427ca15f..edda5598d4 100644 --- a/store/localcachelayer/user_layer.go +++ b/store/localcachelayer/user_layer.go @@ -4,6 +4,8 @@ package localcachelayer import ( + "sort" + "github.com/mattermost/mattermost-server/v5/model" "github.com/mattermost/mattermost-server/v5/store" ) @@ -155,3 +157,66 @@ func (s LocalCacheUserStore) Get(id string) (*model.User, error) { s.rootStore.doStandardAddToCache(s.rootStore.userProfileByIdsCache, id, user) return user, nil } + +// GetMany is a cache wrapper around the SqlStore method to get a user profiles by ids. +// It checks if the user entries are present in the cache, returning the entries from cache +// if it is present. Otherwise, it fetches the entries from the store and stores it in the +// cache. +func (s LocalCacheUserStore) GetMany(ids []string) ([]*model.User, error) { + // we are doing a loop instead of caching the full set in the cache because the number of permutations that we can have + // in this func is making caching of the total set not beneficial. + var cachedUsers []*model.User + var notCachedUserIds []string + uniqIDs := dedup(ids) + + for _, id := range uniqIDs { + var cachedUser *model.User + if err := s.rootStore.doStandardReadCache(s.rootStore.userProfileByIdsCache, id, &cachedUser); err == nil { + if s.rootStore.metrics != nil { + s.rootStore.metrics.AddMemCacheHitCounter("Profile By Id", float64(1)) + } + cachedUsers = append(cachedUsers, cachedUser) + } else { + if s.rootStore.metrics != nil { + s.rootStore.metrics.AddMemCacheMissCounter("Profile By Id", float64(1)) + } + + notCachedUserIds = append(notCachedUserIds, id) + } + } + + if len(notCachedUserIds) > 0 { + dbUsers, err := s.UserStore.GetMany(notCachedUserIds) + if err != nil { + return nil, err + } + for _, user := range dbUsers { + s.rootStore.doStandardAddToCache(s.rootStore.userProfileByIdsCache, user.Id, user) + cachedUsers = append(cachedUsers, user) + } + } + + return cachedUsers, nil +} + +func dedup(elements []string) []string { + if len(elements) == 0 { + return elements + } + + sort.Strings(elements) + + j := 0 + for i := 1; i < len(elements); i++ { + if elements[j] == elements[i] { + continue + } + j++ + // preserve the original data + // in[i], in[j] = in[j], in[i] + // only set what is required + elements[j] = elements[i] + } + + return elements[:j+1] +} diff --git a/store/localcachelayer/user_layer_test.go b/store/localcachelayer/user_layer_test.go index e46fa171cf..157a1cb703 100644 --- a/store/localcachelayer/user_layer_test.go +++ b/store/localcachelayer/user_layer_test.go @@ -258,3 +258,54 @@ func TestUserStoreGetCache(t *testing.T) { storedUser.NotifyProps = originalProps }) } + +func TestUserStoreGetManyCache(t *testing.T) { + fakeUser := &model.User{ + Id: "123", + AuthData: model.NewString("authData"), + AuthService: "authService", + } + otherFakeUser := &model.User{ + Id: "456", + AuthData: model.NewString("authData"), + AuthService: "authService", + } + t.Run("first call not cached, second cached and returning same data", func(t *testing.T) { + mockStore := getMockStore() + mockCacheProvider := getMockCacheProvider() + cachedStore, err := NewLocalCacheLayer(mockStore, nil, nil, mockCacheProvider) + require.NoError(t, err) + + gotUsers, err := cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id}) + require.Nil(t, err) + assert.Len(t, gotUsers, 2) + assert.Contains(t, gotUsers, fakeUser) + assert.Contains(t, gotUsers, otherFakeUser) + + gotUsers, err = cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id}) + require.Nil(t, err) + assert.Len(t, gotUsers, 2) + mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetMany", 1) + }) + + t.Run("first call not cached, invalidate one user, and then check that one is cached and one is fetched from db", func(t *testing.T) { + mockStore := getMockStore() + mockCacheProvider := getMockCacheProvider() + cachedStore, err := NewLocalCacheLayer(mockStore, nil, nil, mockCacheProvider) + require.NoError(t, err) + + gotUsers, err := cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id}) + require.Nil(t, err) + assert.Len(t, gotUsers, 2) + assert.Contains(t, gotUsers, fakeUser) + assert.Contains(t, gotUsers, otherFakeUser) + + cachedStore.User().InvalidateProfileCacheForUser("123") + + gotUsers, err = cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id}) + require.NoError(t, err) + assert.Len(t, gotUsers, 2) + mockStore.User().(*mocks.UserStore).AssertCalled(t, "GetMany", []string{"123"}) + mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetMany", 2) + }) +} diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index e2e772a76c..5e3bd9eb3d 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -8703,6 +8703,24 @@ func (s *OpenTracingLayerUserStore) GetKnownUsers(userID string) ([]string, erro return result, err } +func (s *OpenTracingLayerUserStore) GetMany(ids []string) ([]*model.User, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetMany") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.UserStore.GetMany(ids) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + func (s *OpenTracingLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetNewUsersForTeam") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 5cc55fbefd..79041bbd28 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -9428,6 +9428,26 @@ func (s *RetryLayerUserStore) GetKnownUsers(userID string) ([]string, error) { } +func (s *RetryLayerUserStore) GetMany(ids []string) ([]*model.User, error) { + + tries := 0 + for { + result, err := s.UserStore.GetMany(ids) + if err == nil { + return result, nil + } + if !isRepeatableError(err) { + return result, err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return result, err + } + } + +} + func (s *RetryLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) { tries := 0 diff --git a/store/sqlstore/user_store.go b/store/sqlstore/user_store.go index 753dd0cbe4..655911630c 100644 --- a/store/sqlstore/user_store.go +++ b/store/sqlstore/user_store.go @@ -325,6 +325,22 @@ func (us SqlUserStore) UpdateMfaActive(userId string, active bool) error { return nil } +// GetMany returns a list of users for the provided list of ids +func (us SqlUserStore) GetMany(ids []string) ([]*model.User, error) { + query := us.usersQuery.Where(sq.Eq{"Id": ids}) + queryString, args, err := query.ToSql() + if err != nil { + return nil, errors.Wrap(err, "users_get_many_tosql") + } + + var users []*model.User + if _, err := us.GetReplica().Select(&users, queryString, args...); err != nil { + return nil, errors.Wrap(err, "users_get_many_select") + } + + return users, nil +} + func (us SqlUserStore) Get(id string) (*model.User, error) { query := us.usersQuery.Where("Id = ?", id) queryString, args, err := query.ToSql() diff --git a/store/store.go b/store/store.go index 7611c9abea..09a5755c21 100644 --- a/store/store.go +++ b/store/store.go @@ -323,6 +323,7 @@ type UserStore interface { UpdateMfaSecret(userId, secret string) error UpdateMfaActive(userId string, active bool) error Get(id string) (*model.User, error) + GetMany(ids []string) ([]*model.User, error) GetAll() ([]*model.User, error) ClearCaches() InvalidateProfilesInChannelCacheByUser(userId string) diff --git a/store/storetest/mocks/UserStore.go b/store/storetest/mocks/UserStore.go index fd11cfdb6e..af4ffdd380 100644 --- a/store/storetest/mocks/UserStore.go +++ b/store/storetest/mocks/UserStore.go @@ -603,6 +603,29 @@ func (_m *UserStore) GetKnownUsers(userID string) ([]string, error) { return r0, r1 } +// GetMany provides a mock function with given fields: ids +func (_m *UserStore) GetMany(ids []string) ([]*model.User, error) { + ret := _m.Called(ids) + + var r0 []*model.User + if rf, ok := ret.Get(0).(func([]string) []*model.User); ok { + r0 = rf(ids) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*model.User) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func([]string) error); ok { + r1 = rf(ids) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // GetNewUsersForTeam provides a mock function with given fields: teamId, offset, limit, viewRestrictions func (_m *UserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) { ret := _m.Called(teamId, offset, limit, viewRestrictions) diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index ad16dcce93..eef57fc638 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -7862,6 +7862,22 @@ func (s *TimerLayerUserStore) GetKnownUsers(userID string) ([]string, error) { return result, err } +func (s *TimerLayerUserStore) GetMany(ids []string) ([]*model.User, error) { + start := timemodule.Now() + + result, err := s.UserStore.GetMany(ids) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("UserStore.GetMany", success, elapsed) + } + return result, err +} + func (s *TimerLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) { start := timemodule.Now()