Improve session cleanup (#18123)

- We move logging statements to the upper layer.
Store functions are low-level methods and should return error
upwards rather than logging.
- Used IN instead of any (array ()) which is equivalent.
- Made the delay to be of type time.Duration and un-exported it.
- Unexported the batch size constant.

```release-note
NONE
```

app/server.go

@@ -1511,11 +1511,15 @@ func doCommandWebhookCleanup(s *Server) {
 }
 
 const (
-	SessionsCleanupBatchSize = 1000
+	sessionsCleanupBatchSize = 1000
 )
 
 func doSessionCleanup(s *Server) {
-	s.Store.Session().Cleanup(model.GetMillis(), SessionsCleanupBatchSize)
+	mlog.Debug("Cleaning up session store.")
+	err := s.Store.Session().Cleanup(model.GetMillis(), sessionsCleanupBatchSize)
+	if err != nil {
+		mlog.Warn("Error while cleaning up sessions", mlog.Err(err))
+	}
 }
 
 func doCheckAdminSupportStatus(a *App, c *request.Context) {

store/opentracinglayer/opentracinglayer.go

@@ -6955,7 +6955,7 @@ func (s *OpenTracingLayerSessionStore) AnalyticsSessionCount() (int64, error) {
 	return result, err
 }
 
-func (s *OpenTracingLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) {
+func (s *OpenTracingLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) error {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "SessionStore.Cleanup")
 	s.Root.Store.SetContext(newCtx)
@@ -6964,8 +6964,13 @@ func (s *OpenTracingLayerSessionStore) Cleanup(expiryTime int64, batchSize int64
 	}()
 
 	defer span.Finish()
-	s.SessionStore.Cleanup(expiryTime, batchSize)
+	err := s.SessionStore.Cleanup(expiryTime, batchSize)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
 
+	return err
 }
 
 func (s *OpenTracingLayerSessionStore) Get(ctx context.Context, sessionIDOrToken string) (*model.Session, error) {

store/retrylayer/retrylayer.go

@@ -7550,9 +7550,23 @@ func (s *RetryLayerSessionStore) AnalyticsSessionCount() (int64, error) {
 
 }
 
-func (s *RetryLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) {
+func (s *RetryLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) error {
 
-	s.SessionStore.Cleanup(expiryTime, batchSize)
+	tries := 0
+	for {
+		err := s.SessionStore.Cleanup(expiryTime, batchSize)
+		if err == nil {
+			return nil
+		}
+		if !isRepeatableError(err) {
+			return err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return err
+		}
+	}
 
 }
 

store/sqlstore/session_store.go

@@ -12,12 +12,11 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/mattermost/mattermost-server/v6/model"
-	"github.com/mattermost/mattermost-server/v6/shared/mlog"
 	"github.com/mattermost/mattermost-server/v6/store"
 )
 
 const (
-	SessionsCleanupDelayMilliseconds = 100
+	sessionsCleanupDelay = 100 * time.Millisecond
 )
 
 type SqlSessionStore struct {
@@ -275,12 +274,10 @@ func (me SqlSessionStore) AnalyticsSessionCount() (int64, error) {
 	return count, nil
 }
 
-func (me SqlSessionStore) Cleanup(expiryTime int64, batchSize int64) {
+func (me SqlSessionStore) Cleanup(expiryTime int64, batchSize int64) error {
-	mlog.Debug("Cleaning up session store.")
-
 	var query string
 	if me.DriverName() == model.DatabaseDriverPostgres {
-		query = "DELETE FROM Sessions WHERE Id = any (array (SELECT Id FROM Sessions WHERE ExpiresAt != 0 AND :ExpiresAt > ExpiresAt LIMIT :Limit))"
+		query = "DELETE FROM Sessions WHERE Id IN (SELECT Id FROM Sessions WHERE ExpiresAt != 0 AND :ExpiresAt > ExpiresAt LIMIT :Limit)"
 	} else {
 		query = "DELETE FROM Sessions WHERE ExpiresAt != 0 AND :ExpiresAt > ExpiresAt LIMIT :Limit"
 	}
@@ -290,16 +287,16 @@ func (me SqlSessionStore) Cleanup(expiryTime int64, batchSize int64) {
 	for rowsAffected > 0 {
 		sqlResult, err := me.GetMaster().Exec(query, map[string]interface{}{"ExpiresAt": expiryTime, "Limit": batchSize})
 		if err != nil {
-			mlog.Error("Unable to cleanup session store.", mlog.Err(err))
+			return errors.Wrap(err, "unable to delete sessions")
-			return
 		}
 		var rowErr error
 		rowsAffected, rowErr = sqlResult.RowsAffected()
 		if rowErr != nil {
-			mlog.Error("Unable to cleanup session store.", mlog.Err(err))
+			return errors.Wrap(err, "unable to delete sessions")
-			return
 		}
 
-		time.Sleep(SessionsCleanupDelayMilliseconds * time.Millisecond)
+		time.Sleep(sessionsCleanupDelay)
 	}
+
+	return nil
 }

store/store.go

@@ -451,7 +451,7 @@ type SessionStore interface {
 	UpdateDeviceId(id string, deviceID string, expiresAt int64) (string, error)
 	UpdateProps(session *model.Session) error
 	AnalyticsSessionCount() (int64, error)
-	Cleanup(expiryTime int64, batchSize int64)
+	Cleanup(expiryTime int64, batchSize int64) error
 }
 
 type AuditStore interface {

store/storetest/mocks/SessionStore.go

@@ -38,8 +38,17 @@ func (_m *SessionStore) AnalyticsSessionCount() (int64, error) {
 }
 
 // Cleanup provides a mock function with given fields: expiryTime, batchSize
-func (_m *SessionStore) Cleanup(expiryTime int64, batchSize int64) {
+func (_m *SessionStore) Cleanup(expiryTime int64, batchSize int64) error {
-	_m.Called(expiryTime, batchSize)
+	ret := _m.Called(expiryTime, batchSize)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(int64, int64) error); ok {
+		r0 = rf(expiryTime, batchSize)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
 }
 
 // Get provides a mock function with given fields: ctx, sessionIDOrToken

store/storetest/session_store.go

@@ -303,7 +303,8 @@ func testSessionCleanup(t *testing.T, ss store.Store) {
 	s4, err = ss.Session().Save(s4)
 	require.NoError(t, err)
 
-	ss.Session().Cleanup(now, 1)
+	err = ss.Session().Cleanup(now, 1)
+	require.NoError(t, err)
 
 	_, err = ss.Session().Get(context.Background(), s1.Id)
 	assert.NoError(t, err)

store/timerlayer/timerlayer.go

@@ -6281,19 +6281,20 @@ func (s *TimerLayerSessionStore) AnalyticsSessionCount() (int64, error) {
 	return result, err
 }
 
-func (s *TimerLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) {
+func (s *TimerLayerSessionStore) Cleanup(expiryTime int64, batchSize int64) error {
 	start := timemodule.Now()
 
-	s.SessionStore.Cleanup(expiryTime, batchSize)
+	err := s.SessionStore.Cleanup(expiryTime, batchSize)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {
 		success := "false"
-		if true {
+		if err == nil {
 			success = "true"
 		}
 		s.Root.Metrics.ObserveStoreMethodDuration("SessionStore.Cleanup", success, elapsed)
 	}
+	return err
 }
 
 func (s *TimerLayerSessionStore) Get(ctx context.Context, sessionIDOrToken string) (*model.Session, error) {