From 346fab1620db4e9f57a2abea10ac0150647d3b0b Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Wed, 8 Mar 2023 08:42:21 +0530 Subject: [PATCH] MM-50393: Revamp email batching performance (#22358) This ticket does a number of improvements to the email batching code in general. - Fix unbounded goroutines: We now send mails using a single goroutine only. Since anyways the processing is asynchronous, performance should not matter. - Move the exit condition earlier: We have moved up the check for the per-user email interval preference. This prevents unnecessary DB calls from happening. - Break from the loop properly: The objective was to not send notifications if a user has viewed any channel since a notification was queued. But there were 2 nested for loops and we were breaking from just one and not the other. That is correctly fixed now to prevent unnecessary DB calls. - Gracefully shutdown the job: The batching task wasn't properly shut down and could have notifications in-flight which would have not got sent. We fix that now by cancelling the task and waiting for it to finish. https://mattermost.atlassian.net/browse/MM-50393 --- app/email/email_batching.go | 112 +++++++++++++++++----------- app/email/email_batching_test.go | 2 +- app/email/helper_test.go | 1 - app/email/mocks/ServiceInterface.go | 5 ++ app/email/service.go | 14 +++- app/server.go | 5 +- app/team_test.go | 6 ++ 7 files changed, 95 insertions(+), 50 deletions(-) diff --git a/app/email/email_batching.go b/app/email/email_batching.go index dd3a17d5e5..e24b34a34f 100644 --- a/app/email/email_batching.go +++ b/app/email/email_batching.go @@ -54,7 +54,7 @@ func (es *Service) AddNotificationEmailToBatch(user *model.User, post *model.Pos } if !es.EmailBatching.Add(user, post, team) { - mlog.Error("Email batching job's receiving channel was full. Please increase the EmailBatchingBufferSize.") + mlog.Error("Email batching job's receiving buffer was full. Please increase the EmailBatchingBufferSize. Falling back to sending immediate mail.") return model.NewAppError("AddNotificationEmailToBatch", "api.email_batching.add_notification_email_to_batch.channel_full.app_error", nil, "", http.StatusInternalServerError) } @@ -100,6 +100,17 @@ func (job *EmailBatchingJob) Start() { } } +// Stop will cancel the task properly, flushing out any pending notifications. +// Although this still won't send those notifications which are yet to be sent +// due to a user's PreferenceNameEmailInterval. +func (job *EmailBatchingJob) Stop() { + job.taskMutex.Lock() + if task := job.task; task != nil { + task.Cancel() + } + job.taskMutex.Unlock() +} + func (job *EmailBatchingJob) Add(user *model.User, post *model.Post, team *model.Team) bool { notification := &batchedNotification{ userID: user.Id, @@ -123,7 +134,7 @@ func (job *EmailBatchingJob) CheckPendingEmails() { // without actually sending emails job.checkPendingNotifications(time.Now(), job.service.sendBatchedEmailNotification) - mlog.Debug("Email batching job ran. Some users still have notifications pending.", mlog.Int("number_of_users", len(job.pendingNotifications))) + mlog.Debug("Email batching job ran. Notifications might be still pending.", mlog.Int("number_of_users", len(job.pendingNotifications))) } func (job *EmailBatchingJob) handleNewNotifications() { @@ -148,39 +159,10 @@ func (job *EmailBatchingJob) handleNewNotifications() { func (job *EmailBatchingJob) checkPendingNotifications(now time.Time, handler func(string, []*batchedNotification)) { for userID, notifications := range job.pendingNotifications { - batchStartTime := notifications[0].post.CreateAt - inspectedTeamNames := make(map[string]string) - for _, notification := range notifications { - // at most, we'll do one check for each team that notifications were sent for - if inspectedTeamNames[notification.teamName] != "" { - continue - } - - team, nErr := job.service.store.Team().GetByName(notifications[0].teamName) - if nErr != nil { - mlog.Error("Unable to find Team id for notification", mlog.Err(nErr)) - continue - } - - if team != nil { - inspectedTeamNames[notification.teamName] = team.Id - } - - // if the user has viewed any channels in this team since the notification was queued, delete - // all queued notifications - channelMembers, err := job.service.store.Channel().GetMembersForUser(inspectedTeamNames[notification.teamName], userID) - if err != nil { - mlog.Error("Unable to find ChannelMembers for user", mlog.Err(err)) - continue - } - - for _, channelMember := range channelMembers { - if channelMember.LastViewedAt >= batchStartTime { - mlog.Debug("Deleted notifications for user", mlog.String("user_id", userID)) - delete(job.pendingNotifications, userID) - break - } - } + // Defensive code. + if len(notifications) == 0 { + mlog.Warn("Unexpected result. Got 0 pending notifications for batched email.", mlog.String("user_id", userID)) + continue } // get how long we need to wait to send notifications to the user @@ -198,15 +180,59 @@ func (job *EmailBatchingJob) checkPendingNotifications(now time.Time, handler fu } } - // send the email notification if there are notifications to send AND it's been long enough - if len(job.pendingNotifications[userID]) > 0 && now.Sub(time.Unix(batchStartTime/1000, 0)) > time.Duration(interval)*time.Second { - job.service.goFn(func(userID string, notifications []*batchedNotification) func() { - return func() { - handler(userID, notifications) - } - }(userID, job.pendingNotifications[userID])) - delete(job.pendingNotifications, userID) + batchStartTime := notifications[0].post.CreateAt + // Ignore if it isn't time yet to send. + if now.Sub(time.UnixMilli(batchStartTime)) <= time.Duration(interval)*time.Second { + continue } + + // If the user has viewed any channels in this team since the notification was queued, delete + // all queued notifications + inspectedTeamNames := make(map[string]string) + for _, notification := range notifications { + // at most, we'll do one check for each team that notifications were sent for + if inspectedTeamNames[notification.teamName] != "" { + continue + } + + team, nErr := job.service.store.Team().GetByName(notifications[0].teamName) + if nErr != nil { + mlog.Error("Unable to find Team id for notification", mlog.Err(nErr)) + continue + } + + if team != nil { + inspectedTeamNames[notification.teamName] = team.Id + } + + channelMembers, err := job.service.store.Channel().GetMembersForUser(inspectedTeamNames[notification.teamName], userID) + if err != nil { + mlog.Error("Unable to find ChannelMembers for user", mlog.Err(err)) + continue + } + + deleted := false + for _, channelMember := range channelMembers { + if channelMember.LastViewedAt >= batchStartTime { + mlog.Debug("Deleted notifications for user", mlog.String("user_id", userID)) + delete(job.pendingNotifications, userID) + deleted = true + break + } + } + if deleted { + break + } + } + + // The notifications might have been cleared from the above step. + // We need to check again. + if len(job.pendingNotifications[userID]) == 0 { + continue + } + + handler(userID, job.pendingNotifications[userID]) + delete(job.pendingNotifications, userID) } } diff --git a/app/email/email_batching_test.go b/app/email/email_batching_test.go index 68b7b9451b..4444bc253d 100644 --- a/app/email/email_batching_test.go +++ b/app/email/email_batching_test.go @@ -138,7 +138,7 @@ func TestCheckPendingNotifications(t *testing.T) { // We do a check outside the email handler, because otherwise, failing from // inside the handler doesn't let the .Go() function exit cleanly, and it gets // stuck during server shutdown, trying to wait for the goroutine to exit - require.Equal(t, atomic.LoadInt32(&wasCalled), int32(0), "email handler should not have been called") + require.Equal(t, int32(0), atomic.LoadInt32(&wasCalled), "email handler should not have been called") require.Nil(t, job.pendingNotifications[th.BasicUser.Id]) require.Empty(t, job.pendingNotifications[th.BasicUser.Id], "should've remove queued post since user acted") diff --git a/app/email/helper_test.go b/app/email/helper_test.go index 75fa7605b1..b14d32197a 100644 --- a/app/email/helper_test.go +++ b/app/email/helper_test.go @@ -122,7 +122,6 @@ func setupTestHelper(s store.Store, tb testing.TB) *TestHelper { license: licenseFn, config: configStore.Get, templatesContainer: htmlTemplateWatcher, - goFn: func(f func()) { go f() }, } if err := service.setUpRateLimiters(); err != nil { diff --git a/app/email/mocks/ServiceInterface.go b/app/email/mocks/ServiceInterface.go index 16e6c05195..b63fec8678 100644 --- a/app/email/mocks/ServiceInterface.go +++ b/app/email/mocks/ServiceInterface.go @@ -553,3 +553,8 @@ func (_m *ServiceInterface) SendWelcomeEmail(userID string, _a1 string, verified return r0 } + +// Stop provides a mock function with given fields: +func (_m *ServiceInterface) Stop() { + _m.Called() +} diff --git a/app/email/service.go b/app/email/service.go index d75e81dab0..daa7d1a351 100644 --- a/app/email/service.go +++ b/app/email/service.go @@ -15,6 +15,7 @@ import ( "github.com/mattermost/mattermost-server/v6/app/users" "github.com/mattermost/mattermost-server/v6/model" "github.com/mattermost/mattermost-server/v6/shared/i18n" + "github.com/mattermost/mattermost-server/v6/shared/mlog" "github.com/mattermost/mattermost-server/v6/shared/templates" "github.com/mattermost/mattermost-server/v6/store" ) @@ -42,7 +43,6 @@ func condenseSiteURL(siteURL string) string { type Service struct { config func() *model.Config - goFn func(f func()) license func() *model.License userService *users.UserService @@ -57,7 +57,6 @@ type Service struct { type ServiceConfig struct { ConfigFn func() *model.Config LicenseFn func() *model.License - GoFn func(f func()) TemplatesContainer *templates.Container UserService *users.UserService @@ -72,7 +71,6 @@ func NewService(config ServiceConfig) (*Service, error) { config: config.ConfigFn, templatesContainer: config.TemplatesContainer, license: config.LicenseFn, - goFn: config.GoFn, store: config.Store, userService: config.UserService, } @@ -83,8 +81,15 @@ func NewService(config ServiceConfig) (*Service, error) { return service, nil } +func (es *Service) Stop() { + mlog.Info("Shutting down Email batching service...") + if es.EmailBatching != nil { + es.EmailBatching.Stop() + } +} + func (c *ServiceConfig) validate() error { - if c.ConfigFn == nil || c.GoFn == nil || c.Store == nil || c.LicenseFn == nil || c.TemplatesContainer == nil { + if c.ConfigFn == nil || c.Store == nil || c.LicenseFn == nil || c.TemplatesContainer == nil { return errors.New("invalid service config") } return nil @@ -159,6 +164,7 @@ type ServiceInterface interface { SendChangeUsernameEmail(newUsername, email, locale, siteURL string) error CreateVerifyEmailToken(userID string, newEmail string) (*model.Token, error) SendLicenseInactivityEmail(email, name, locale, siteURL string) error + Stop() } func (es *Service) GetPerDayEmailRateLimiter() *throttled.GCRARateLimiter { diff --git a/app/server.go b/app/server.go index d74af9907d..5f7e12d982 100644 --- a/app/server.go +++ b/app/server.go @@ -366,7 +366,6 @@ func NewServer(options ...Option) (*Server, error) { emailService, err := email.NewService(email.ServiceConfig{ ConfigFn: s.platform.Config, LicenseFn: s.License, - GoFn: s.Go, TemplatesContainer: s.TemplatesContainer(), UserService: s.userService, Store: s.GetStore(), @@ -719,6 +718,10 @@ func (s *Server) Shutdown() { s.Log().Warn("Failed to stop metrics server", mlog.Err(err)) } + // Stopping email service after HTTP server has stopped to prevent + // any stray notifications from being queued. + s.EmailService.Stop() + // This must be done after the cluster is stopped. if s.Jobs != nil { // For simplicity we don't check if workers and schedulers are active diff --git a/app/team_test.go b/app/team_test.go index 8601d51b53..83cfc3cd71 100644 --- a/app/team_test.go +++ b/app/team_test.go @@ -1502,6 +1502,7 @@ func TestInviteNewUsersToTeamGracefully(t *testing.T) { false, false, ).Once().Return(nil) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteNewUsersToTeamGracefully(memberInvite, th.BasicTeam.Id, th.BasicUser.Id, "") @@ -1526,6 +1527,7 @@ func TestInviteNewUsersToTeamGracefully(t *testing.T) { false, false, ).Once().Return(email.SendMailError) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteNewUsersToTeamGracefully(memberInvite, th.BasicTeam.Id, th.BasicUser.Id, "") @@ -1554,6 +1556,7 @@ func TestInviteNewUsersToTeamGracefully(t *testing.T) { false, false, ).Once().Return([]*model.EmailInviteWithError{}, nil) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteNewUsersToTeamGracefully(memberInvite, th.BasicTeam.Id, th.BasicUser.Id, "") @@ -1578,6 +1581,7 @@ func TestInviteNewUsersToTeamGracefully(t *testing.T) { false, false, ).Once().Return(nil) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteNewUsersToTeamGracefully(memberInvite, th.BasicTeam.Id, th.BasicUser.Id, "") @@ -1610,6 +1614,7 @@ func TestInviteGuestsToChannelsGracefully(t *testing.T) { false, false, ).Once().Return(nil) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteGuestsToChannelsGracefully(th.BasicTeam.Id, &model.GuestsInvite{ @@ -1636,6 +1641,7 @@ func TestInviteGuestsToChannelsGracefully(t *testing.T) { false, false, ).Once().Return(email.SendMailError) + emailServiceMock.On("Stop").Once().Return() th.App.Srv().EmailService = &emailServiceMock res, err := th.App.InviteGuestsToChannelsGracefully(th.BasicTeam.Id, &model.GuestsInvite{