From 768fe43d3af1bc52080ee25067bead598892be4e Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 3 Mar 2022 08:50:19 +0530 Subject: [PATCH] MM-41752: Batch update mention increment (#19596) Previously, we were incrementing mentions one-by-one all concurrently in an unbounded fashion. This would cause a big spike in memory usage if there were an `@all` mention in a large channel. We fix this by changing the SQL query to take all userIDs at once. https://mattermost.atlassian.net/browse/MM-41752 ```release-note NONE ``` Co-authored-by: Mattermod --- app/notification.go | 33 ++++++---------------- i18n/en.json | 4 --- store/opentracinglayer/opentracinglayer.go | 4 +-- store/retrylayer/retrylayer.go | 4 +-- store/sqlstore/channel_store.go | 32 +++++++++++++-------- store/store.go | 2 +- store/storetest/channel_store.go | 8 +++--- store/storetest/mocks/ChannelStore.go | 10 +++---- store/storetest/user_store.go | 29 ++++++++++++++++--- store/timerlayer/timerlayer.go | 4 +-- 10 files changed, 69 insertions(+), 61 deletions(-) diff --git a/app/notification.go b/app/notification.go index 1ccb9256de..4e4b79c52a 100644 --- a/app/notification.go +++ b/app/notification.go @@ -196,7 +196,6 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod } mentionedUsersList := make(model.StringArray, 0, len(mentions.Mentions)) - updateMentionChans := []chan *model.AppError{} mentionAutofollowChans := []chan *model.AppError{} threadParticipants := map[string]bool{post.UserId: true} participantMemberships := map[string]*model.ThreadMembership{} @@ -290,32 +289,16 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod } for id := range mentions.Mentions { mentionedUsersList = append(mentionedUsersList, id) - - umc := make(chan *model.AppError, 1) - go func(userID string) { - defer close(umc) - nErr := a.Srv().Store.Channel().IncrementMentionCount(post.ChannelId, userID, post.RootId == "") - if nErr != nil { - umc <- model.NewAppError("SendNotifications", "app.channel.increment_mention_count.app_error", nil, nErr.Error(), http.StatusInternalServerError) - return - } - umc <- nil - }(id) - updateMentionChans = append(updateMentionChans, umc) } - // Make sure all mention updates are complete to prevent race conditions. - // Probably better to batch these DB updates in the future - // MUST be completed before push notifications send - for _, umc := range updateMentionChans { - if err := <-umc; err != nil { - mlog.Warn( - "Failed to update mention count", - mlog.String("post_id", post.Id), - mlog.String("channel_id", post.ChannelId), - mlog.Err(err), - ) - } + nErr := a.Srv().Store.Channel().IncrementMentionCount(post.ChannelId, mentionedUsersList, post.RootId == "") + if nErr != nil { + mlog.Warn( + "Failed to update mention count", + mlog.String("post_id", post.Id), + mlog.String("channel_id", post.ChannelId), + mlog.Err(nErr), + ) } // Log the problems that might have occurred while auto following the thread diff --git a/i18n/en.json b/i18n/en.json index 00b84ccc82..def55168b9 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -4635,10 +4635,6 @@ "id": "app.channel.get_unread.app_error", "translation": "Unable to get the channel unread messages." }, - { - "id": "app.channel.increment_mention_count.app_error", - "translation": "Unable to increment the mention count." - }, { "id": "app.channel.migrate_channel_members.select.app_error", "translation": "Failed to select the batch of channel members." diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 41f4f326e1..123ae01c56 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -1748,7 +1748,7 @@ func (s *OpenTracingLayerChannelStore) GroupSyncedChannelCount() (int64, error) return result, err } -func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error { +func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, userIDs []string, isRoot bool) error { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.IncrementMentionCount") s.Root.Store.SetContext(newCtx) @@ -1757,7 +1757,7 @@ func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, u }() defer span.Finish() - err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot) + err := s.ChannelStore.IncrementMentionCount(channelID, userIDs, isRoot) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 4202dadbcd..174f70c61d 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -1975,11 +1975,11 @@ func (s *RetryLayerChannelStore) GroupSyncedChannelCount() (int64, error) { } -func (s *RetryLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error { +func (s *RetryLayerChannelStore) IncrementMentionCount(channelID string, userIDs []string, isRoot bool) error { tries := 0 for { - err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot) + err := s.ChannelStore.IncrementMentionCount(channelID, userIDs, isRoot) if err == nil { return nil } diff --git a/store/sqlstore/channel_store.go b/store/sqlstore/channel_store.go index 5108c5e67a..9f1c16622a 100644 --- a/store/sqlstore/channel_store.go +++ b/store/sqlstore/channel_store.go @@ -2549,24 +2549,32 @@ func (s SqlChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID s return result, nil } -func (s SqlChannelStore) IncrementMentionCount(channelId string, userId string, isRoot bool) error { +func (s SqlChannelStore) IncrementMentionCount(channelId string, userIDs []string, isRoot bool) error { now := model.GetMillis() + rootInc := 0 if isRoot { rootInc = 1 } - _, err := s.GetMasterX().Exec( - `UPDATE - ChannelMembers - SET - MentionCount = MentionCount + 1, - MentionCountRoot = MentionCountRoot + ?, - LastUpdateAt = ? - WHERE - UserId = ? - AND ChannelId = ?`, rootInc, now, userId, channelId) + + sql, args, err := s.getQueryBuilder(). + Update("ChannelMembers"). + Set("MentionCount", sq.Expr("MentionCount + 1")). + Set("MentionCountRoot", sq.Expr("MentionCountRoot + ?", rootInc)). + Set("LastUpdateAt", now). + Where(sq.Eq{ + "UserId": userIDs, + "ChannelId": channelId, + }). + ToSql() + if err != nil { - return errors.Wrapf(err, "failed to Update ChannelMembers with channelId=%s and userId=%s", channelId, userId) + return errors.Wrap(err, "IncrementMentionCount_Tosql") + } + + _, err = s.GetMasterX().Exec(sql, args...) + if err != nil { + return errors.Wrapf(err, "failed to Update ChannelMembers with channelId=%s and userId=%v", channelId, userIDs) } return nil } diff --git a/store/store.go b/store/store.go index 358f7ecae2..dc5cc33caf 100644 --- a/store/store.go +++ b/store/store.go @@ -226,7 +226,7 @@ type ChannelStore interface { UpdateLastViewedAt(channelIds []string, userID string, updateThreads bool) (map[string]int64, error) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) CountPostsAfter(channelID string, timestamp int64, userID string) (int, int, error) - IncrementMentionCount(channelID string, userID string, isRoot bool) error + IncrementMentionCount(channelID string, userIDs []string, isRoot bool) error AnalyticsTypeCount(teamID string, channelType model.ChannelType) (int64, error) GetMembersForUser(teamID string, userID string) (model.ChannelMembers, error) GetTeamMembersForChannel(channelID string) ([]string, error) diff --git a/store/storetest/channel_store.go b/store/storetest/channel_store.go index 74ffaa6c19..1e79647a68 100644 --- a/store/storetest/channel_store.go +++ b/store/storetest/channel_store.go @@ -4790,16 +4790,16 @@ func testChannelStoreIncrementMentionCount(t *testing.T, ss store.Store) { _, err := ss.Channel().SaveMember(&m1) require.NoError(t, err) - err = ss.Channel().IncrementMentionCount(m1.ChannelId, m1.UserId, false) + err = ss.Channel().IncrementMentionCount(m1.ChannelId, []string{m1.UserId}, false) require.NoError(t, err, "failed to update") - err = ss.Channel().IncrementMentionCount(m1.ChannelId, "missing id", false) + err = ss.Channel().IncrementMentionCount(m1.ChannelId, []string{"missing id"}, false) require.NoError(t, err, "failed to update") - err = ss.Channel().IncrementMentionCount("missing id", m1.UserId, false) + err = ss.Channel().IncrementMentionCount("missing id", []string{m1.UserId}, false) require.NoError(t, err, "failed to update") - err = ss.Channel().IncrementMentionCount("missing id", "missing id", false) + err = ss.Channel().IncrementMentionCount("missing id", []string{"missing id"}, false) require.NoError(t, err, "failed to update") } diff --git a/store/storetest/mocks/ChannelStore.go b/store/storetest/mocks/ChannelStore.go index a1b4b4471a..4671ec9a11 100644 --- a/store/storetest/mocks/ChannelStore.go +++ b/store/storetest/mocks/ChannelStore.go @@ -1508,13 +1508,13 @@ func (_m *ChannelStore) GroupSyncedChannelCount() (int64, error) { return r0, r1 } -// IncrementMentionCount provides a mock function with given fields: channelID, userID, isRoot -func (_m *ChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error { - ret := _m.Called(channelID, userID, isRoot) +// IncrementMentionCount provides a mock function with given fields: channelID, userIDs, isRoot +func (_m *ChannelStore) IncrementMentionCount(channelID string, userIDs []string, isRoot bool) error { + ret := _m.Called(channelID, userIDs, isRoot) var r0 error - if rf, ok := ret.Get(0).(func(string, string, bool) error); ok { - r0 = rf(channelID, userID, isRoot) + if rf, ok := ret.Get(0).(func(string, []string, bool) error); ok { + r0 = rf(channelID, userIDs, isRoot) } else { r0 = ret.Error(0) } diff --git a/store/storetest/user_store.go b/store/storetest/user_store.go index f9079bbe28..962d63dfe4 100644 --- a/store/storetest/user_store.go +++ b/store/storetest/user_store.go @@ -2285,6 +2285,15 @@ func testUserUnreadCount(t *testing.T, ss store.Store) { _, nErr = ss.Team().SaveMember(&model.TeamMember{TeamId: teamId, UserId: u2.Id}, -1) require.NoError(t, nErr) + u3 := &model.User{} + u3.Email = MakeEmail() + u3.Username = "user3" + model.NewId() + _, err = ss.User().Save(u3) + require.NoError(t, err) + defer func() { require.NoError(t, ss.User().PermanentDelete(u3.Id)) }() + _, nErr = ss.Team().SaveMember(&model.TeamMember{TeamId: teamId, UserId: u3.Id}, -1) + require.NoError(t, nErr) + _, nErr = ss.Channel().Save(&c1, -1) require.NoError(t, nErr, "couldn't save item") @@ -2301,6 +2310,14 @@ func testUserUnreadCount(t *testing.T, ss store.Store) { _, nErr = ss.Channel().SaveMember(&m2) require.NoError(t, nErr) + m3 := model.ChannelMember{} + m3.ChannelId = c1.Id + m3.UserId = u3.Id + m3.NotifyProps = model.GetDefaultChannelNotifyProps() + + _, nErr = ss.Channel().SaveMember(&m3) + require.NoError(t, nErr) + m1.ChannelId = c2.Id m2.ChannelId = c2.Id @@ -2310,12 +2327,12 @@ func testUserUnreadCount(t *testing.T, ss store.Store) { p1 := model.Post{} p1.ChannelId = c1.Id p1.UserId = u1.Id - p1.Message = "this is a message for @" + u2.Username + p1.Message = "this is a message for @" + u2.Username + " and " + "@" + u3.Username // Post one message with mention to open channel _, nErr = ss.Post().Save(&p1) require.NoError(t, nErr) - nErr = ss.Channel().IncrementMentionCount(c1.Id, u2.Id, false) + nErr = ss.Channel().IncrementMentionCount(c1.Id, []string{u2.Id, u3.Id}, false) require.NoError(t, nErr) // Post 2 messages without mention to direct channel @@ -2326,7 +2343,7 @@ func testUserUnreadCount(t *testing.T, ss store.Store) { _, nErr = ss.Post().Save(&p2) require.NoError(t, nErr) - nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false) + nErr = ss.Channel().IncrementMentionCount(c2.Id, []string{u2.Id}, false) require.NoError(t, nErr) p3 := model.Post{} @@ -2336,13 +2353,17 @@ func testUserUnreadCount(t *testing.T, ss store.Store) { _, nErr = ss.Post().Save(&p3) require.NoError(t, nErr) - nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false) + nErr = ss.Channel().IncrementMentionCount(c2.Id, []string{u2.Id}, false) require.NoError(t, nErr) badge, unreadCountErr := ss.User().GetUnreadCount(u2.Id) require.NoError(t, unreadCountErr) require.Equal(t, int64(3), badge, "should have 3 unread messages") + badge, unreadCountErr = ss.User().GetUnreadCount(u3.Id) + require.NoError(t, unreadCountErr) + require.Equal(t, int64(1), badge, "should have 1 unread message") + badge, unreadCountErr = ss.User().GetUnreadCountForChannel(u2.Id, c1.Id) require.NoError(t, unreadCountErr) require.Equal(t, int64(1), badge, "should have 1 unread messages for that channel") diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index c16d4d03c0..66ce872576 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -1605,10 +1605,10 @@ func (s *TimerLayerChannelStore) GroupSyncedChannelCount() (int64, error) { return result, err } -func (s *TimerLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error { +func (s *TimerLayerChannelStore) IncrementMentionCount(channelID string, userIDs []string, isRoot bool) error { start := timemodule.Now() - err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot) + err := s.ChannelStore.IncrementMentionCount(channelID, userIDs, isRoot) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil {