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

@@ -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
+		// Defensive code.
-		inspectedTeamNames := make(map[string]string)
+		if len(notifications) == 0 {
-		for _, notification := range notifications {
+			mlog.Warn("Unexpected result. Got 0 pending notifications for batched email.", mlog.String("user_id", userID))
-			// at most, we'll do one check for each team that notifications were sent for
+			continue
-			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
-				}
-			}
 		}
 
 		// 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
+		batchStartTime := notifications[0].post.CreateAt
-		if len(job.pendingNotifications[userID]) > 0 && now.Sub(time.Unix(batchStartTime/1000, 0)) > time.Duration(interval)*time.Second {
+		// Ignore if it isn't time yet to send.
-			job.service.goFn(func(userID string, notifications []*batchedNotification) func() {
+		if now.Sub(time.UnixMilli(batchStartTime)) <= time.Duration(interval)*time.Second {
-				return func() {
+			continue
-					handler(userID, notifications)
-				}
-			}(userID, job.pendingNotifications[userID]))
-			delete(job.pendingNotifications, userID)
 		}
+
+		// 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)
 	}
 }
 

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")

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 {

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()
+}

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 {

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

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{