MM-41260: Revamp ES/Bleve batching logic (#19841)

The older method used to reply completely on timestamps
to take batches of items in a timestamp range and then
just incrementing the timestamp. This led to handling
edge-cases such as more items than the batch count, all
having the same timestamp.

Additionally, relying on timestamp as the page cursor
meant that indexing was not very efficient if you had
several items spread out across large spans of time.

To get away from all of that we use a proper cursor-based
approach consisting of createAt+Id. With this, we move
completely to a constant page size where we can fetch
a given number of objects irrespective of when they
were created. This makes indexing much more faster and
efficient.

https://mattermost.atlassian.net/browse/MM-41260

```release-note
Elasticsearch and Bleve indexing have been revamped to be much
more efficient and faster. The config parameter BulkIndexingTimeWindowSeconds
for both elasticsearch and bleve have been removed.
A new config parameter called BatchSize has been introduced instead.
This parameter controls the number of objects that
can be indexed in a single batch. This makes things
more efficient and maintains a constant workload.
```

db/migrations/mysql/000080_posts_createat_id.down.sql

@@ -0,0 +1,14 @@
+SET @preparedStatement = (SELECT IF(
+    (
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE table_name = 'Posts'
+        AND table_schema = DATABASE()
+        AND index_name = 'idx_posts_create_at_id'
+    ) > 0,
+    'DROP INDEX idx_posts_create_at_id on Posts;',
+    'SELECT 1;'
+));
+
+PREPARE removeIndexIfExists FROM @preparedStatement;
+EXECUTE removeIndexIfExists;
+DEALLOCATE PREPARE removeIndexIfExists;

db/migrations/mysql/000080_posts_createat_id.up.sql

@@ -0,0 +1,14 @@
+SET @preparedStatement = (SELECT IF(
+    (
+        SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE table_name = 'Posts'
+        AND table_schema = DATABASE()
+        AND index_name = 'idx_posts_create_at_id'
+    ) > 0,
+    'SELECT 1;',
+    'CREATE INDEX idx_posts_create_at_id on Posts(CreateAt, Id) LOCK=NONE;'
+));
+
+PREPARE createIndexIfNotExists FROM @preparedStatement;
+EXECUTE createIndexIfNotExists;
+DEALLOCATE PREPARE createIndexIfNotExists;

db/migrations/postgres/000080_posts_createat_id.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF exists idx_posts_create_at_id;

db/migrations/postgres/000080_posts_createat_id.up.sql

@@ -0,0 +1 @@
+CREATE INDEX IF NOT EXISTS idx_posts_create_at_id on posts(createat, id);

i18n/en.json

@@ -8044,8 +8044,8 @@
     "translation": "Invalid RemoteImageProxyURL for atmos/camo. Must be set to your shared key."
   },
   {
-    "id": "model.config.is_valid.bleve_search.bulk_indexing_time_window_seconds.app_error",
+    "id": "model.config.is_valid.bleve_search.bulk_indexing_batch_size.app_error",
-    "translation": "Bleve Bulk Indexing Time Window must be at least 1 second."
+    "translation": "Bleve Bulk Indexing Batch Size must be at least {{.BatchSize}}."
   },
   {
     "id": "model.config.is_valid.bleve_search.enable_autocomplete.app_error",
@@ -8096,8 +8096,8 @@
     "translation": "Elasticsearch AggregatePostsAfterDays setting must be a number greater than or equal to 1."
   },
   {
-    "id": "model.config.is_valid.elastic_search.bulk_indexing_time_window_seconds.app_error",
+    "id": "model.config.is_valid.elastic_search.bulk_indexing_batch_size.app_error",
-    "translation": "Elasticsearch Bulk Indexing Time Window must be at least 1 second."
+    "translation": "Elasticsearch Bulk Indexing Batch Size must be at least {{.BatchSize}}."
   },
   {
     "id": "model.config.is_valid.elastic_search.connection_url.app_error",

model/config.go

@@ -184,24 +184,24 @@ const (
 
 	TeamSettingsDefaultTeamText = "default"
 
-	ElasticsearchSettingsDefaultConnectionURL                 = "http://localhost:9200"
+	ElasticsearchSettingsDefaultConnectionURL               = "http://localhost:9200"
-	ElasticsearchSettingsDefaultUsername                      = "elastic"
+	ElasticsearchSettingsDefaultUsername                    = "elastic"
-	ElasticsearchSettingsDefaultPassword                      = "changeme"
+	ElasticsearchSettingsDefaultPassword                    = "changeme"
-	ElasticsearchSettingsDefaultPostIndexReplicas             = 1
+	ElasticsearchSettingsDefaultPostIndexReplicas           = 1
-	ElasticsearchSettingsDefaultPostIndexShards               = 1
+	ElasticsearchSettingsDefaultPostIndexShards             = 1
-	ElasticsearchSettingsDefaultChannelIndexReplicas          = 1
+	ElasticsearchSettingsDefaultChannelIndexReplicas        = 1
-	ElasticsearchSettingsDefaultChannelIndexShards            = 1
+	ElasticsearchSettingsDefaultChannelIndexShards          = 1
-	ElasticsearchSettingsDefaultUserIndexReplicas             = 1
+	ElasticsearchSettingsDefaultUserIndexReplicas           = 1
-	ElasticsearchSettingsDefaultUserIndexShards               = 1
+	ElasticsearchSettingsDefaultUserIndexShards             = 1
-	ElasticsearchSettingsDefaultAggregatePostsAfterDays       = 365
+	ElasticsearchSettingsDefaultAggregatePostsAfterDays     = 365
-	ElasticsearchSettingsDefaultPostsAggregatorJobStartTime   = "03:00"
+	ElasticsearchSettingsDefaultPostsAggregatorJobStartTime = "03:00"
-	ElasticsearchSettingsDefaultIndexPrefix                   = ""
+	ElasticsearchSettingsDefaultIndexPrefix                 = ""
-	ElasticsearchSettingsDefaultLiveIndexingBatchSize         = 1
+	ElasticsearchSettingsDefaultLiveIndexingBatchSize       = 1
-	ElasticsearchSettingsDefaultBulkIndexingTimeWindowSeconds = 3600
+	ElasticsearchSettingsDefaultRequestTimeoutSeconds       = 30
-	ElasticsearchSettingsDefaultRequestTimeoutSeconds         = 30
+	ElasticsearchSettingsDefaultBatchSize                   = 10000
 
-	BleveSettingsDefaultIndexDir                      = ""
+	BleveSettingsDefaultIndexDir  = ""
-	BleveSettingsDefaultBulkIndexingTimeWindowSeconds = 3600
+	BleveSettingsDefaultBatchSize = 10000
 
 	DataRetentionSettingsDefaultMessageRetentionDays = 365
 	DataRetentionSettingsDefaultFileRetentionDays    = 365
@@ -2480,7 +2480,8 @@ type ElasticsearchSettings struct {
 	PostsAggregatorJobStartTime   *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` // telemetry: none
 	IndexPrefix                   *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
 	LiveIndexingBatchSize         *int    `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
-	BulkIndexingTimeWindowSeconds *int    `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
+	BulkIndexingTimeWindowSeconds *int    `json:",omitempty"` // telemetry: none
+	BatchSize                     *int    `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
 	RequestTimeoutSeconds         *int    `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
 	SkipTLSVerification           *bool   `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
 	Trace                         *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"`
@@ -2555,8 +2556,8 @@ func (s *ElasticsearchSettings) SetDefaults() {
 		s.LiveIndexingBatchSize = NewInt(ElasticsearchSettingsDefaultLiveIndexingBatchSize)
 	}
 
-	if s.BulkIndexingTimeWindowSeconds == nil {
+	if s.BatchSize == nil {
-		s.BulkIndexingTimeWindowSeconds = NewInt(ElasticsearchSettingsDefaultBulkIndexingTimeWindowSeconds)
+		s.BatchSize = NewInt(ElasticsearchSettingsDefaultBatchSize)
 	}
 
 	if s.RequestTimeoutSeconds == nil {
@@ -2577,7 +2578,8 @@ type BleveSettings struct {
 	EnableIndexing                *bool   `access:"experimental_bleve"`
 	EnableSearching               *bool   `access:"experimental_bleve"`
 	EnableAutocomplete            *bool   `access:"experimental_bleve"`
-	BulkIndexingTimeWindowSeconds *int    `access:"experimental_bleve"`
+	BulkIndexingTimeWindowSeconds *int    `json:",omitempty"` // telemetry: none
+	BatchSize                     *int    `access:"experimental_bleve"`
 }
 
 func (bs *BleveSettings) SetDefaults() {
@@ -2597,8 +2599,8 @@ func (bs *BleveSettings) SetDefaults() {
 		bs.EnableAutocomplete = NewBool(false)
 	}
 
-	if bs.BulkIndexingTimeWindowSeconds == nil {
+	if bs.BatchSize == nil {
-		bs.BulkIndexingTimeWindowSeconds = NewInt(BleveSettingsDefaultBulkIndexingTimeWindowSeconds)
+		bs.BatchSize = NewInt(BleveSettingsDefaultBatchSize)
 	}
 }
 
@@ -3632,8 +3634,9 @@ func (s *ElasticsearchSettings) isValid() *AppError {
 		return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.live_indexing_batch_size.app_error", nil, "", http.StatusBadRequest)
 	}
 
-	if *s.BulkIndexingTimeWindowSeconds < 1 {
+	minBatchSize := 1
-		return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.bulk_indexing_time_window_seconds.app_error", nil, "", http.StatusBadRequest)
+	if *s.BatchSize < minBatchSize {
+		return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.bulk_indexing_batch_size.app_error", map[string]interface{}{"BatchSize": minBatchSize}, "", http.StatusBadRequest)
 	}
 
 	if *s.RequestTimeoutSeconds < 1 {
@@ -3656,8 +3659,9 @@ func (bs *BleveSettings) isValid() *AppError {
 			return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.enable_autocomplete.app_error", nil, "", http.StatusBadRequest)
 		}
 	}
-	if *bs.BulkIndexingTimeWindowSeconds < 1 {
+	minBatchSize := 1
-		return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.bulk_indexing_time_window_seconds.app_error", nil, "", http.StatusBadRequest)
+	if *bs.BatchSize < minBatchSize {
+		return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.bulk_indexing_batch_size.app_error", map[string]interface{}{"BatchSize": minBatchSize}, "", http.StatusBadRequest)
 	}
 
 	return nil

model/config_test.go

@@ -31,6 +31,13 @@ func TestConfigDefaults(t *testing.T) {
 		var recursivelyUninitialize func(*Config, string, reflect.Value)
 		recursivelyUninitialize = func(config *Config, name string, v reflect.Value) {
 			if v.Type().Kind() == reflect.Ptr {
+				// Ignoring these 2 settings.
+				// TODO: remove them completely in v8.0.
+				if name == "config.BleveSettings.BulkIndexingTimeWindowSeconds" ||
+					name == "config.ElasticsearchSettings.BulkIndexingTimeWindowSeconds" {
+					return
+				}
+
 				// Set every pointer we find in the tree to nil
 				v.Set(reflect.Zero(v.Type()))
 				require.True(t, v.IsNil())

model/utils_test.go

@@ -778,6 +778,13 @@ func checkNowhereNil(t *testing.T, name string, value interface{}) bool {
 	v := reflect.ValueOf(value)
 	switch v.Type().Kind() {
 	case reflect.Ptr:
+		// Ignoring these 2 settings.
+		// TODO: remove them completely in v8.0.
+		if name == "config.BleveSettings.BulkIndexingTimeWindowSeconds" ||
+			name == "config.ElasticsearchSettings.BulkIndexingTimeWindowSeconds" {
+			return true
+		}
+
 		if v.IsNil() {
 			t.Logf("%s was nil", name)
 			return false

services/searchengine/bleveengine/indexer/indexing_job.go

@@ -17,12 +17,12 @@ import (
 )
 
 const (
-	BatchSize             = 1000
+	timeBetweenBatches = 100 * time.Millisecond
-	TimeBetweenBatches    = 100
+
-	EstimatedPostCount    = 10000000
+	estimatedPostCount    = 10000000
-	EstimatedFilesCount   = 100000
+	estimatedFilesCount   = 100000
-	EstimatedChannelCount = 100000
+	estimatedChannelCount = 100000
-	EstimatedUserCount    = 10000
+	estimatedUserCount    = 10000
 )
 
 type BleveIndexerWorker struct {
@@ -50,22 +50,30 @@ func MakeWorker(jobServer *jobs.JobServer, engine *bleveengine.BleveEngine) mode
 }
 
 type IndexingProgress struct {
-	Now                time.Time
+	Now            time.Time
-	StartAtTime        int64
+	StartAtTime    int64
-	EndAtTime          int64
+	EndAtTime      int64
-	LastEntityTime     int64
+	LastEntityTime int64
-	TotalPostsCount    int64
+
-	DonePostsCount     int64
+	TotalPostsCount int64
-	DonePosts          bool
+	DonePostsCount  int64
-	TotalFilesCount    int64
+	DonePosts       bool
-	DoneFilesCount     int64
+	LastPostID      string
-	DoneFiles          bool
+
+	TotalFilesCount int64
+	DoneFilesCount  int64
+	DoneFiles       bool
+	LastFileID      string
+
 	TotalChannelsCount int64
 	DoneChannelsCount  int64
 	DoneChannels       bool
-	TotalUsersCount    int64
+	LastChannelID      string
-	DoneUsersCount     int64
+
-	DoneUsers          bool
+	TotalUsersCount int64
+	DoneUsersCount  int64
+	DoneUsers       bool
+	LastUserID      string
 }
 
 func (ip *IndexingProgress) CurrentProgress() int64 {
@@ -160,7 +168,6 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 			return
 		}
 		progress.StartAtTime = startInt
-		progress.LastEntityTime = progress.StartAtTime
 	} else {
 		// Set start time to oldest entity in the database.
 		// A user or a channel may be created before any post.
@@ -174,8 +181,8 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 			return
 		}
 		progress.StartAtTime = oldestEntityCreationTime
-		progress.LastEntityTime = progress.StartAtTime
 	}
+	progress.LastEntityTime = progress.StartAtTime
 
 	if endString, ok := job.Data["end_time"]; ok {
 		endInt, err := strconv.ParseInt(endString, 10, 64)
@@ -190,27 +197,43 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 		progress.EndAtTime = endInt
 	}
 
+	if id, ok := job.Data["start_post_id"]; ok {
+		progress.LastPostID = id
+	}
+	if id, ok := job.Data["start_channel_id"]; ok {
+		progress.LastChannelID = id
+	}
+	if id, ok := job.Data["start_user_id"]; ok {
+		progress.LastUserID = id
+	}
+	if id, ok := job.Data["start_file_id"]; ok {
+		progress.LastFileID = id
+	}
+
 	// Counting all posts may fail or timeout when the posts table is large. If this happens, log a warning, but carry
 	// on with the indexing job anyway. The only issue is that the progress % reporting will be inaccurate.
 	if count, err := worker.jobServer.Store.Post().AnalyticsPostCount("", false, false); err != nil {
 		mlog.Warn("Worker: Failed to fetch total post count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
-		progress.TotalPostsCount = EstimatedPostCount
+		progress.TotalPostsCount = estimatedPostCount
 	} else {
 		progress.TotalPostsCount = count
 	}
 
 	// Same possible fail as above can happen when counting channels
-	if count, err := worker.jobServer.Store.Channel().AnalyticsTypeCount("", model.ChannelTypeOpen); err != nil {
+	if count, err := worker.jobServer.Store.Channel().AnalyticsTypeCount("", ""); err != nil {
 		mlog.Warn("Worker: Failed to fetch total channel count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
-		progress.TotalChannelsCount = EstimatedChannelCount
+		progress.TotalChannelsCount = estimatedChannelCount
 	} else {
 		progress.TotalChannelsCount = count
 	}
 
 	// Same possible fail as above can happen when counting users
-	if count, err := worker.jobServer.Store.User().Count(model.UserCountOptions{}); err != nil {
+	if count, err := worker.jobServer.Store.User().Count(model.UserCountOptions{
+		IncludeBotAccounts: true, // This actually doesn't join with the bots table
+		// since ExcludeRegularUsers is set to false
+	}); err != nil {
 		mlog.Warn("Worker: Failed to fetch total user count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
-		progress.TotalUsersCount = EstimatedUserCount
+		progress.TotalUsersCount = estimatedUserCount
 	} else {
 		progress.TotalUsersCount = count
 	}
@@ -219,7 +242,7 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 	// on with the indexing job anyway. The only issue is that the progress % reporting will be inaccurate.
 	if count, err := worker.jobServer.Store.FileInfo().CountAll(); err != nil {
 		mlog.Warn("Worker: Failed to fetch total file info count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
-		progress.TotalFilesCount = EstimatedFilesCount
+		progress.TotalFilesCount = estimatedFilesCount
 	} else {
 		progress.TotalFilesCount = count
 	}
@@ -246,7 +269,7 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 			}
 			return
 
-		case <-time.After(TimeBetweenBatches * time.Millisecond):
+		case <-time.After(timeBetweenBatches):
 			var err *model.AppError
 			if progress, err = worker.IndexBatch(progress); err != nil {
 				mlog.Error("Worker: Failed to index batch for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
@@ -256,6 +279,19 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
 				return
 			}
 
+			// Storing the batch progress in metadata.
+			if job.Data == nil {
+				job.Data = make(model.StringMap)
+			}
+
+			job.Data["start_time"] = strconv.FormatInt(progress.LastEntityTime, 10)
+			job.Data["start_post_id"] = progress.LastPostID
+			job.Data["start_channel_id"] = progress.LastChannelID
+			job.Data["start_user_id"] = progress.LastUserID
+			job.Data["start_file_id"] = progress.LastFileID
+			job.Data["original_start_time"] = strconv.FormatInt(progress.StartAtTime, 10)
+			job.Data["end_time"] = strconv.FormatInt(progress.EndAtTime, 10)
+
 			if err := worker.jobServer.SetJobProgress(job, progress.CurrentProgress()); err != nil {
 				mlog.Error("Worker: Failed to set progress for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
 				if err2 := worker.jobServer.SetJobError(job, err); err2 != nil {
@@ -295,14 +331,12 @@ func (worker *BleveIndexerWorker) IndexBatch(progress IndexingProgress) (Indexin
 }
 
 func (worker *BleveIndexerWorker) IndexPostsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
-	endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
-
 	var posts []*model.PostForIndexing
 
 	tries := 0
 	for posts == nil {
 		var err error
-		posts, err = worker.jobServer.Store.Post().GetPostsBatchForIndexing(progress.LastEntityTime, endTime, BatchSize)
+		posts, err = worker.jobServer.Store.Post().GetPostsBatchForIndexing(progress.LastEntityTime, progress.LastPostID, *worker.jobServer.Config().BleveSettings.BatchSize)
 		if err != nil {
 			if tries >= 10 {
 				return progress, model.NewAppError("IndexPostsBatch", "app.post.get_posts_batch_for_indexing.get.app_error", nil, err.Error(), http.StatusInternalServerError)
@@ -316,40 +350,34 @@ func (worker *BleveIndexerWorker) IndexPostsBatch(progress IndexingProgress) (In
 		tries++
 	}
 
-	newLastMessageTime, err := worker.BulkIndexPosts(posts, progress)
+	// Handle zero messages.
+	if len(posts) == 0 {
+		progress.DonePosts = true
+		progress.LastEntityTime = progress.StartAtTime
+		return progress, nil
+	}
+
+	lastPost, err := worker.BulkIndexPosts(posts, progress)
 	if err != nil {
 		return progress, err
 	}
 
-	// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
+	// Our exit condition is when the last post's createAt reaches the initial endAtTime
-	// case, set the "newLastMessageTime" to the endTime so we don't get stuck running the same query in a loop.
+	// set during job creation.
-	if len(posts) < BatchSize {
+	if progress.EndAtTime <= lastPost.CreateAt {
-		newLastMessageTime = endTime
-	}
-
-	// When to Stop: we index either until we pass a batch of messages where the last
-	// message is created at or after the specified end time when setting up the batch
-	// index, or until two consecutive full batches have the same end time of their final
-	// messages. This second case is safe as long as the assumption that the database
-	// cannot contain more messages with the same CreateAt time than the batch size holds.
-	if progress.EndAtTime <= newLastMessageTime {
-		progress.DonePosts = true
-		progress.LastEntityTime = progress.StartAtTime
-	} else if progress.LastEntityTime == newLastMessageTime && len(posts) == BatchSize {
-		mlog.Warn("More posts with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastMessageTime), mlog.Int("Batch Size", BatchSize))
 		progress.DonePosts = true
 		progress.LastEntityTime = progress.StartAtTime
 	} else {
-		progress.LastEntityTime = newLastMessageTime
+		progress.LastEntityTime = lastPost.CreateAt
 	}
 
+	progress.LastPostID = lastPost.Id
 	progress.DonePostsCount += int64(len(posts))
 
 	return progress, nil
 }
 
-func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, progress IndexingProgress) (int64, *model.AppError) {
+func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, progress IndexingProgress) (*model.Post, *model.AppError) {
-	lastCreateAt := int64(0)
 	batch := worker.engine.PostIndex.NewBatch()
 
 	for _, post := range posts {
@@ -359,28 +387,24 @@ func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing,
 		} else {
 			batch.Delete(post.Id)
 		}
-
-		lastCreateAt = post.CreateAt
 	}
 
 	worker.engine.Mutex.RLock()
 	defer worker.engine.Mutex.RUnlock()
 
 	if err := worker.engine.PostIndex.Batch(batch); err != nil {
-		return 0, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_posts.batch_error", nil, err.Error(), http.StatusInternalServerError)
+		return nil, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_posts.batch_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-	return lastCreateAt, nil
+	return &posts[len(posts)-1].Post, nil
 }
 
 func (worker *BleveIndexerWorker) IndexFilesBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
-	endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
-
 	var files []*model.FileForIndexing
 
 	tries := 0
 	for files == nil {
 		var err error
-		files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, endTime, BatchSize)
+		files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, progress.LastFileID, *worker.jobServer.Config().BleveSettings.BatchSize)
 		if err != nil {
 			if tries >= 10 {
 				return progress, model.NewAppError("IndexFilesBatch", "app.post.get_files_batch_for_indexing.get.app_error", nil, err.Error(), http.StatusInternalServerError)
@@ -394,40 +418,33 @@ func (worker *BleveIndexerWorker) IndexFilesBatch(progress IndexingProgress) (In
 		tries++
 	}
 
-	newLastFileTime, err := worker.BulkIndexFiles(files, progress)
+	if len(files) == 0 {
+		progress.DoneFiles = true
+		progress.LastEntityTime = progress.StartAtTime
+		return progress, nil
+	}
+
+	lastFile, err := worker.BulkIndexFiles(files, progress)
 	if err != nil {
 		return progress, err
 	}
 
-	// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
+	// Our exit condition is when the last file's createAt reaches the initial endAtTime
-	// case, set the "newLastFileTime" to the endTime so we don't get stuck running the same query in a loop.
+	// set during job creation.
-	if len(files) < BatchSize {
+	if progress.EndAtTime <= lastFile.CreateAt {
-		newLastFileTime = endTime
-	}
-
-	// When to Stop: we index either until we pass a batch of messages where the last
-	// message is created at or after the specified end time when setting up the batch
-	// index, or until two consecutive full batches have the same end time of their final
-	// messages. This second case is safe as long as the assumption that the database
-	// cannot contain more messages with the same CreateAt time than the batch size holds.
-	if progress.EndAtTime <= newLastFileTime {
-		progress.DoneFiles = true
-		progress.LastEntityTime = progress.StartAtTime
-	} else if progress.LastEntityTime == newLastFileTime && len(files) == BatchSize {
-		mlog.Warn("More files with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastFileTime), mlog.Int("Batch Size", BatchSize))
 		progress.DoneFiles = true
 		progress.LastEntityTime = progress.StartAtTime
 	} else {
-		progress.LastEntityTime = newLastFileTime
+		progress.LastEntityTime = lastFile.CreateAt
 	}
 
+	progress.LastFileID = lastFile.Id
 	progress.DoneFilesCount += int64(len(files))
 
 	return progress, nil
 }
 
-func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing, progress IndexingProgress) (int64, *model.AppError) {
+func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing, progress IndexingProgress) (*model.FileInfo, *model.AppError) {
-	lastCreateAt := int64(0)
 	batch := worker.engine.FileIndex.NewBatch()
 
 	for _, file := range files {
@@ -437,28 +454,24 @@ func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing,
 		} else {
 			batch.Delete(file.Id)
 		}
-
-		lastCreateAt = file.CreateAt
 	}
 
 	worker.engine.Mutex.RLock()
 	defer worker.engine.Mutex.RUnlock()
 
 	if err := worker.engine.FileIndex.Batch(batch); err != nil {
-		return 0, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_files.batch_error", nil, err.Error(), http.StatusInternalServerError)
+		return nil, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_files.batch_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-	return lastCreateAt, nil
+	return &files[len(files)-1].FileInfo, nil
 }
 
 func (worker *BleveIndexerWorker) IndexChannelsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
-	endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
-
 	var channels []*model.Channel
 
 	tries := 0
 	for channels == nil {
 		var nErr error
-		channels, nErr = worker.jobServer.Store.Channel().GetChannelsBatchForIndexing(progress.LastEntityTime, endTime, BatchSize)
+		channels, nErr = worker.jobServer.Store.Channel().GetChannelsBatchForIndexing(progress.LastEntityTime, progress.LastChannelID, *worker.jobServer.Config().BleveSettings.BatchSize)
 		if nErr != nil {
 			if tries >= 10 {
 				return progress, model.NewAppError("BleveIndexerWorker.IndexChannelsBatch", "app.channel.get_channels_batch_for_indexing.get.app_error", nil, nErr.Error(), http.StatusInternalServerError)
@@ -472,40 +485,33 @@ func (worker *BleveIndexerWorker) IndexChannelsBatch(progress IndexingProgress)
 		tries++
 	}
 
-	newLastChannelTime, err := worker.BulkIndexChannels(channels, progress)
+	if len(channels) == 0 {
+		progress.DoneChannels = true
+		progress.LastEntityTime = progress.StartAtTime
+		return progress, nil
+	}
+
+	lastChannel, err := worker.BulkIndexChannels(channels, progress)
 	if err != nil {
 		return progress, err
 	}
 
-	// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
+	// Our exit condition is when the last channel's createAt reaches the initial endAtTime
-	// case, set the "newLastChannelTime" to the endTime so we don't get stuck running the same query in a loop.
+	// set during job creation.
-	if len(channels) < BatchSize {
+	if progress.EndAtTime <= lastChannel.CreateAt {
-		newLastChannelTime = endTime
-	}
-
-	// When to Stop: we index either until we pass a batch of channels where the last
-	// channel is created at or after the specified end time when setting up the batch
-	// index, or until two consecutive full batches have the same end time of their final
-	// channels. This second case is safe as long as the assumption that the database
-	// cannot contain more channels with the same CreateAt time than the batch size holds.
-	if progress.EndAtTime <= newLastChannelTime {
-		progress.DoneChannels = true
-		progress.LastEntityTime = progress.StartAtTime
-	} else if progress.LastEntityTime == newLastChannelTime && len(channels) == BatchSize {
-		mlog.Warn("More channels with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastChannelTime), mlog.Int("Batch Size", BatchSize))
 		progress.DoneChannels = true
 		progress.LastEntityTime = progress.StartAtTime
 	} else {
-		progress.LastEntityTime = newLastChannelTime
+		progress.LastEntityTime = lastChannel.CreateAt
 	}
 
+	progress.LastChannelID = lastChannel.Id
 	progress.DoneChannelsCount += int64(len(channels))
 
 	return progress, nil
 }
 
-func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, progress IndexingProgress) (int64, *model.AppError) {
+func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, progress IndexingProgress) (*model.Channel, *model.AppError) {
-	lastCreateAt := int64(0)
 	batch := worker.engine.ChannelIndex.NewBatch()
 
 	for _, channel := range channels {
@@ -515,14 +521,14 @@ func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, p
 			if channel.Type == model.ChannelTypePrivate {
 				userIDs, err = worker.jobServer.Store.Channel().GetAllChannelMembersById(channel.Id)
 				if err != nil {
-					return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
+					return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
 				}
 			}
 
 			// Get teamMember ids from channelid
 			teamMemberIDs, err := worker.jobServer.Store.Channel().GetTeamMembersForChannel(channel.Id)
 			if err != nil {
-				return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
+				return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
 			}
 
 			searchChannel := bleveengine.BLVChannelFromChannel(channel, userIDs, teamMemberIDs)
@@ -530,27 +536,23 @@ func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, p
 		} else {
 			batch.Delete(channel.Id)
 		}
-
-		lastCreateAt = channel.CreateAt
 	}
 
 	worker.engine.Mutex.RLock()
 	defer worker.engine.Mutex.RUnlock()
 
 	if err := worker.engine.ChannelIndex.Batch(batch); err != nil {
-		return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
+		return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-	return lastCreateAt, nil
+	return channels[len(channels)-1], nil
 }
 
 func (worker *BleveIndexerWorker) IndexUsersBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
-	endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
-
 	var users []*model.UserForIndexing
 
 	tries := 0
 	for users == nil {
-		if usersBatch, err := worker.jobServer.Store.User().GetUsersBatchForIndexing(progress.LastEntityTime, endTime, BatchSize); err != nil {
+		if usersBatch, err := worker.jobServer.Store.User().GetUsersBatchForIndexing(progress.LastEntityTime, progress.LastUserID, *worker.jobServer.Config().BleveSettings.BatchSize); err != nil {
 			if tries >= 10 {
 				return progress, model.NewAppError("IndexUsersBatch", "app.user.get_users_batch_for_indexing.get_users.app_error", nil, err.Error(), http.StatusInternalServerError)
 			}
@@ -565,40 +567,32 @@ func (worker *BleveIndexerWorker) IndexUsersBatch(progress IndexingProgress) (In
 		tries++
 	}
 
-	newLastUserTime, err := worker.BulkIndexUsers(users, progress)
+	if len(users) == 0 {
+		progress.DoneUsers = true
+		progress.LastEntityTime = progress.StartAtTime
+		return progress, nil
+	}
+
+	lastUser, err := worker.BulkIndexUsers(users, progress)
 	if err != nil {
 		return progress, err
 	}
 
-	// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
+	// Our exit condition is when the last user's createAt reaches the initial endAtTime
-	// case, set the "newLastUserTime" to the endTime so we don't get stuck running the same query in a loop.
+	// set during job creation.
-	if len(users) < BatchSize {
+	if progress.EndAtTime <= lastUser.CreateAt {
-		newLastUserTime = endTime
-	}
-
-	// When to Stop: we index either until we pass a batch of users where the last
-	// user is created at or after the specified end time when setting up the batch
-	// index, or until two consecutive full batches have the same end time of their final
-	// users. This second case is safe as long as the assumption that the database
-	// cannot contain more users with the same CreateAt time than the batch size holds.
-	if progress.EndAtTime <= newLastUserTime {
-		progress.DoneUsers = true
-		progress.LastEntityTime = progress.StartAtTime
-	} else if progress.LastEntityTime == newLastUserTime && len(users) == BatchSize {
-		mlog.Warn("More users with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastUserTime), mlog.Int("Batch Size", BatchSize))
 		progress.DoneUsers = true
 		progress.LastEntityTime = progress.StartAtTime
 	} else {
-		progress.LastEntityTime = newLastUserTime
+		progress.LastEntityTime = lastUser.CreateAt
 	}
-
+	progress.LastUserID = lastUser.Id
 	progress.DoneUsersCount += int64(len(users))
 
 	return progress, nil
 }
 
-func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, progress IndexingProgress) (int64, *model.AppError) {
+func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, progress IndexingProgress) (*model.UserForIndexing, *model.AppError) {
-	lastCreateAt := int64(0)
 	batch := worker.engine.UserIndex.NewBatch()
 
 	for _, user := range users {
@@ -608,15 +602,13 @@ func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing,
 		} else {
 			batch.Delete(user.Id)
 		}
-
-		lastCreateAt = user.CreateAt
 	}
 
 	worker.engine.Mutex.RLock()
 	defer worker.engine.Mutex.RUnlock()
 
 	if err := worker.engine.UserIndex.Batch(batch); err != nil {
-		return 0, model.NewAppError("BleveIndexerWorker.BulkIndexUsers", "bleveengine.indexer.do_job.bulk_index_users.batch_error", nil, err.Error(), http.StatusInternalServerError)
+		return nil, model.NewAppError("BleveIndexerWorker.BulkIndexUsers", "bleveengine.indexer.do_job.bulk_index_users.batch_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-	return lastCreateAt, nil
+	return users[len(users)-1], nil
 }

services/telemetry/telemetry.go

@@ -735,25 +735,25 @@ func (ts *TelemetryService) trackConfig() {
 	})
 
 	ts.SendTelemetry(TrackConfigElasticsearch, map[string]interface{}{
-		"isdefault_connection_url":          isDefault(*cfg.ElasticsearchSettings.ConnectionURL, model.ElasticsearchSettingsDefaultConnectionURL),
+		"isdefault_connection_url": isDefault(*cfg.ElasticsearchSettings.ConnectionURL, model.ElasticsearchSettingsDefaultConnectionURL),
-		"isdefault_username":                isDefault(*cfg.ElasticsearchSettings.Username, model.ElasticsearchSettingsDefaultUsername),
+		"isdefault_username":       isDefault(*cfg.ElasticsearchSettings.Username, model.ElasticsearchSettingsDefaultUsername),
-		"isdefault_password":                isDefault(*cfg.ElasticsearchSettings.Password, model.ElasticsearchSettingsDefaultPassword),
+		"isdefault_password":       isDefault(*cfg.ElasticsearchSettings.Password, model.ElasticsearchSettingsDefaultPassword),
-		"enable_indexing":                   *cfg.ElasticsearchSettings.EnableIndexing,
+		"enable_indexing":          *cfg.ElasticsearchSettings.EnableIndexing,
-		"enable_searching":                  *cfg.ElasticsearchSettings.EnableSearching,
+		"enable_searching":         *cfg.ElasticsearchSettings.EnableSearching,
-		"enable_autocomplete":               *cfg.ElasticsearchSettings.EnableAutocomplete,
+		"enable_autocomplete":      *cfg.ElasticsearchSettings.EnableAutocomplete,
-		"sniff":                             *cfg.ElasticsearchSettings.Sniff,
+		"sniff":                    *cfg.ElasticsearchSettings.Sniff,
-		"post_index_replicas":               *cfg.ElasticsearchSettings.PostIndexReplicas,
+		"post_index_replicas":      *cfg.ElasticsearchSettings.PostIndexReplicas,
-		"post_index_shards":                 *cfg.ElasticsearchSettings.PostIndexShards,
+		"post_index_shards":        *cfg.ElasticsearchSettings.PostIndexShards,
-		"channel_index_replicas":            *cfg.ElasticsearchSettings.ChannelIndexReplicas,
+		"channel_index_replicas":   *cfg.ElasticsearchSettings.ChannelIndexReplicas,
-		"channel_index_shards":              *cfg.ElasticsearchSettings.ChannelIndexShards,
+		"channel_index_shards":     *cfg.ElasticsearchSettings.ChannelIndexShards,
-		"user_index_replicas":               *cfg.ElasticsearchSettings.UserIndexReplicas,
+		"user_index_replicas":      *cfg.ElasticsearchSettings.UserIndexReplicas,
-		"user_index_shards":                 *cfg.ElasticsearchSettings.UserIndexShards,
+		"user_index_shards":        *cfg.ElasticsearchSettings.UserIndexShards,
-		"isdefault_index_prefix":            isDefault(*cfg.ElasticsearchSettings.IndexPrefix, model.ElasticsearchSettingsDefaultIndexPrefix),
+		"isdefault_index_prefix":   isDefault(*cfg.ElasticsearchSettings.IndexPrefix, model.ElasticsearchSettingsDefaultIndexPrefix),
-		"live_indexing_batch_size":          *cfg.ElasticsearchSettings.LiveIndexingBatchSize,
+		"live_indexing_batch_size": *cfg.ElasticsearchSettings.LiveIndexingBatchSize,
-		"bulk_indexing_time_window_seconds": *cfg.ElasticsearchSettings.BulkIndexingTimeWindowSeconds,
+		"bulk_indexing_batch_size": *cfg.ElasticsearchSettings.BatchSize,
-		"request_timeout_seconds":           *cfg.ElasticsearchSettings.RequestTimeoutSeconds,
+		"request_timeout_seconds":  *cfg.ElasticsearchSettings.RequestTimeoutSeconds,
-		"skip_tls_verification":             *cfg.ElasticsearchSettings.SkipTLSVerification,
+		"skip_tls_verification":    *cfg.ElasticsearchSettings.SkipTLSVerification,
-		"trace":                             *cfg.ElasticsearchSettings.Trace,
+		"trace":                    *cfg.ElasticsearchSettings.Trace,
 	})
 
 	ts.trackPluginConfig(cfg, model.PluginSettingsDefaultMarketplaceURL)
@@ -804,10 +804,10 @@ func (ts *TelemetryService) trackConfig() {
 	})
 
 	ts.SendTelemetry(TrackConfigBleve, map[string]interface{}{
-		"enable_indexing":                   *cfg.BleveSettings.EnableIndexing,
+		"enable_indexing":          *cfg.BleveSettings.EnableIndexing,
-		"enable_searching":                  *cfg.BleveSettings.EnableSearching,
+		"enable_searching":         *cfg.BleveSettings.EnableSearching,
-		"enable_autocomplete":               *cfg.BleveSettings.EnableAutocomplete,
+		"enable_autocomplete":      *cfg.BleveSettings.EnableAutocomplete,
-		"bulk_indexing_time_window_seconds": *cfg.BleveSettings.BulkIndexingTimeWindowSeconds,
+		"bulk_indexing_batch_size": *cfg.BleveSettings.BatchSize,
 	})
 
 	ts.SendTelemetry(TrackConfigExport, map[string]interface{}{

store/opentracinglayer/opentracinglayer.go

@@ -1105,7 +1105,7 @@ func (s *OpenTracingLayerChannelStore) GetChannels(teamID string, userID string,
 	return result, err
 }
 
-func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) {
+func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetChannelsBatchForIndexing")
 	s.Root.Store.SetContext(newCtx)
@@ -1114,7 +1114,7 @@ func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int
 	}()
 
 	defer span.Finish()
-	result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit)
+	result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)
@@ -3318,7 +3318,7 @@ func (s *OpenTracingLayerFileInfoStore) GetByPath(path string) (*model.FileInfo,
 	return result, err
 }
 
-func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetFilesBatchForIndexing")
 	s.Root.Store.SetContext(newCtx)
@@ -3327,7 +3327,7 @@ func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64
 	}()
 
 	defer span.Finish()
-	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)
@@ -5700,7 +5700,7 @@ func (s *OpenTracingLayerPostStore) GetPostsAfter(options model.GetPostsOptions)
 	return result, err
 }
 
-func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) {
+func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetPostsBatchForIndexing")
 	s.Root.Store.SetContext(newCtx)
@@ -5709,7 +5709,7 @@ func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, en
 	}()
 
 	defer span.Finish()
-	result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit)
+	result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)
@@ -10610,7 +10610,7 @@ func (s *OpenTracingLayerUserStore) GetUnreadCountForChannel(userID string, chan
 	return result, err
 }
 
-func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) {
+func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetUsersBatchForIndexing")
 	s.Root.Store.SetContext(newCtx)
@@ -10619,7 +10619,7 @@ func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, en
 	}()
 
 	defer span.Finish()
-	result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit)
+	result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)

store/retrylayer/retrylayer.go

@@ -1234,11 +1234,11 @@ func (s *RetryLayerChannelStore) GetChannels(teamID string, userID string, opts
 
 }
 
-func (s *RetryLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) {
+func (s *RetryLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) {
 
 	tries := 0
 	for {
-		result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit)
+		result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit)
 		if err == nil {
 			return result, nil
 		}
@@ -3715,11 +3715,11 @@ func (s *RetryLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error
 
 }
 
-func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
 
 	tries := 0
 	for {
-		result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+		result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
 		if err == nil {
 			return result, nil
 		}
@@ -6448,11 +6448,11 @@ func (s *RetryLayerPostStore) GetPostsAfter(options model.GetPostsOptions) (*mod
 
 }
 
-func (s *RetryLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) {
+func (s *RetryLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) {
 
 	tries := 0
 	for {
-		result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit)
+		result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit)
 		if err == nil {
 			return result, nil
 		}
@@ -12103,11 +12103,11 @@ func (s *RetryLayerUserStore) GetUnreadCountForChannel(userID string, channelID
 
 }
 
-func (s *RetryLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) {
+func (s *RetryLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) {
 
 	tries := 0
 	for {
-		result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit)
+		result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit)
 		if err == nil {
 			return result, nil
 		}

store/sqlstore/channel_store.go

@@ -3949,23 +3949,23 @@ func (s SqlChannelStore) GetAllDirectChannelsForExportAfter(limit int, afterId s
 	return directChannelsForExport, nil
 }
 
-func (s SqlChannelStore) GetChannelsBatchForIndexing(startTime, endTime int64, limit int) ([]*model.Channel, error) {
+func (s SqlChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) {
 	query :=
 		`SELECT
 			 *
 		 FROM
 			 Channels
 		 WHERE
-			 CreateAt >= ?
+			 CreateAt > ?
-		 AND
+		OR
-			 CreateAt < ?
+			(CreateAt = ? AND Id > ?)
 		 ORDER BY
-			 CreateAt
+			 CreateAt ASC, Id ASC
 		 LIMIT
 			 ?`
 
 	channels := []*model.Channel{}
-	err := s.GetSearchReplicaX().Select(&channels, query, startTime, endTime, limit)
+	err := s.GetSearchReplicaX().Select(&channels, query, startTime, startTime, startChannelID, limit)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to find Channels")
 	}

store/sqlstore/file_info_store.go

@@ -688,17 +688,23 @@ func (fs SqlFileInfoStore) CountAll() (int64, error) {
 	return count, nil
 }
 
-func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
 	files := []*model.FileForIndexing{}
 	sql, args, _ := fs.getQueryBuilder().
 		Select(append(fs.queryFields, "Coalesce(p.ChannelId, '') AS ChannelId")...).
 		From("FileInfo").
 		LeftJoin("Posts AS p ON FileInfo.PostId = p.Id").
-		Where(sq.GtOrEq{"FileInfo.CreateAt": startTime}).
+		Where(sq.Or{
-		Where(sq.Lt{"FileInfo.CreateAt": endTime}).
+			sq.Gt{"FileInfo.CreateAt": startTime},
-		OrderBy("FileInfo.CreateAt").
+			sq.And{
+				sq.Eq{"FileInfo.CreateAt": startTime},
+				sq.Gt{"FileInfo.Id": startFileID},
+			},
+		}).
+		OrderBy("FileInfo.CreateAt ASC, FileInfo.Id ASC").
 		Limit(uint64(limit)).
 		ToSql()
+
 	err := fs.GetSearchReplicaX().Select(&files, sql, args...)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to find Files")

store/sqlstore/post_store.go

@@ -2198,22 +2198,26 @@ func (s *SqlPostStore) GetPostsByIds(postIds []string) ([]*model.Post, error) {
 	return posts, nil
 }
 
-func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) {
+func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) {
 	posts := []*model.PostForIndexing{}
-	err := s.GetSearchReplicaX().Select(&posts,
+	table := "Posts"
-		`SELECT
+	// We force this index to avoid any chances of index merge intersection.
-			PostsQuery.*, Channels.TeamId, ParentPosts.CreateAt ParentCreateAt
+	if s.DriverName() == model.DatabaseDriverMysql {
+		table += " USE INDEX(idx_posts_create_at_id)"
+	}
+	query := `SELECT
+			PostsQuery.*, Channels.TeamId
 		FROM (
 			SELECT
 				*
 			FROM
-				Posts
+				` + table + `
 			WHERE
-				Posts.CreateAt >= ?
+				Posts.CreateAt > ?
-			AND
+			OR
-				Posts.CreateAt < ?
+				(Posts.CreateAt = ? AND Posts.Id > ?)
 			ORDER BY
-				CreateAt ASC
+				CreateAt ASC, Id ASC
 			LIMIT
 				?
 			)
@@ -2223,11 +2227,8 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64,
 			Channels
 		ON
 			PostsQuery.ChannelId = Channels.Id
-		LEFT JOIN
+		ORDER BY CreateAt ASC, Id ASC`
-			Posts ParentPosts
+	err := s.GetSearchReplicaX().Select(&posts, query, startTime, startTime, startPostID, limit)
-		ON
-			PostsQuery.RootId = ParentPosts.Id`,
-		startTime, endTime, limit)
 
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to find Posts")

store/sqlstore/user_store.go

@@ -1671,12 +1671,17 @@ func (us SqlUserStore) InferSystemInstallDate() (int64, error) {
 	return createAt, nil
 }
 
-func (us SqlUserStore) GetUsersBatchForIndexing(startTime, endTime int64, limit int) ([]*model.UserForIndexing, error) {
+func (us SqlUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) {
 	users := []*model.User{}
 	usersQuery, args, _ := us.usersQuery.
-		Where(sq.GtOrEq{"u.CreateAt": startTime}).
+		Where(sq.Or{
-		Where(sq.Lt{"u.CreateAt": endTime}).
+			sq.Gt{"u.CreateAt": startTime},
-		OrderBy("u.CreateAt").
+			sq.And{
+				sq.Eq{"u.CreateAt": startTime},
+				sq.Gt{"u.Id": startFileID},
+			},
+		}).
+		OrderBy("u.CreateAt ASC, u.Id ASC").
 		Limit(uint64(limit)).
 		ToSql()
 	err := us.GetSearchReplicaX().Select(&users, usersQuery, args...)

store/store.go

@@ -267,7 +267,7 @@ type ChannelStore interface {
 	GetAllDirectChannelsForExportAfter(limit int, afterID string) ([]*model.DirectChannelForExport, error)
 	GetChannelMembersForExport(userID string, teamID string) ([]*model.ChannelMemberForExport, error)
 	RemoveAllDeactivatedMembers(channelID string) error
-	GetChannelsBatchForIndexing(startTime, endTime int64, limit int) ([]*model.Channel, error)
+	GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error)
 	UserBelongsToChannels(userID string, channelIds []string) (bool, error)
 
 	// UpdateMembersRole sets all of the given team members to admins and all of the other members of the team to
@@ -351,7 +351,7 @@ type PostStore interface {
 	Overwrite(post *model.Post) (*model.Post, error)
 	OverwriteMultiple(posts []*model.Post) ([]*model.Post, int, error)
 	GetPostsByIds(postIds []string) ([]*model.Post, error)
-	GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error)
+	GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error)
 	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
 	DeleteOrphanedRows(limit int) (deleted int64, err error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
@@ -430,7 +430,7 @@ type UserStore interface {
 	ClearAllCustomRoleAssignments() error
 	InferSystemInstallDate() (int64, error)
 	GetAllAfter(limit int, afterID string) ([]*model.User, error)
-	GetUsersBatchForIndexing(startTime, endTime int64, limit int) ([]*model.UserForIndexing, error)
+	GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error)
 	Count(options model.UserCountOptions) (int64, error)
 	GetTeamGroupUsers(teamID string) ([]*model.User, error)
 	GetChannelGroupUsers(channelID string) ([]*model.User, error)
@@ -657,7 +657,7 @@ type FileInfoStore interface {
 	SetContent(fileID, content string) error
 	Search(paramsList []*model.SearchParams, userID, teamID string, page, perPage int) (*model.FileInfoList, error)
 	CountAll() (int64, error)
-	GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error)
+	GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error)
 	ClearCaches()
 }
 

store/storetest/channel_store.go

@@ -7601,7 +7601,6 @@ func testChannelStoreGetChannelsBatchForIndexing(t *testing.T, ss store.Store) {
 	require.NoError(t, nErr)
 
 	time.Sleep(10 * time.Millisecond)
-	startTime := c2.CreateAt
 
 	c3 := &model.Channel{}
 	c3.DisplayName = "Channel3"
@@ -7633,23 +7632,20 @@ func testChannelStoreGetChannelsBatchForIndexing(t *testing.T, ss store.Store) {
 	_, nErr = ss.Channel().Save(c6, -1)
 	require.NoError(t, nErr)
 
-	endTime := c6.CreateAt
-
 	// First and last channel should be outside the range
-	channels, err := ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 1000)
+	channels, err := ss.Channel().GetChannelsBatchForIndexing(c1.CreateAt, "", 4)
 	assert.NoError(t, err)
-	assert.ElementsMatch(t, []*model.Channel{c2, c3, c4, c5}, channels)
+	assert.Len(t, channels, 4)
 
-	// Update the endTime, last channel should be in
+	// From 4th createat+id
-	endTime = model.GetMillis()
+	channels, err = ss.Channel().GetChannelsBatchForIndexing(channels[3].CreateAt, channels[3].Id, 5)
-	channels, err = ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 1000)
 	assert.NoError(t, err)
-	assert.ElementsMatch(t, []*model.Channel{c2, c3, c4, c5, c6}, channels)
+	assert.Len(t, channels, 2)
 
 	// Testing the limit
-	channels, err = ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 2)
+	channels, err = ss.Channel().GetChannelsBatchForIndexing(channels[1].CreateAt, channels[1].Id, 1)
 	assert.NoError(t, err)
-	assert.ElementsMatch(t, []*model.Channel{c2, c3}, channels)
+	assert.Len(t, channels, 0)
 }
 
 func testGroupSyncedChannelCount(t *testing.T, ss store.Store) {

store/storetest/file_info_store.go

@@ -673,42 +673,23 @@ func testFileInfoStoreGetFilesBatchForIndexing(t *testing.T, ss store.Store) {
 		ss.FileInfo().PermanentDelete(f3.Id)
 	}()
 
-	t.Run("get all files", func(t *testing.T) {
+	// Getting all
-		r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt, model.GetMillis()+100000, 100)
+	r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 100)
-		require.NoError(t, err)
+	require.NoError(t, err)
-		require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r))
+	require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r))
-		for _, f := range r {
-			if f.Id == f1.Id {
-				require.Equal(t, f.ChannelId, o1.ChannelId, "Unexpected channel ID")
-				require.Equal(t, f.Path, "file1.txt", "Unexpected filename")
-			} else if f.Id == f2.Id {
-				require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID")
-				require.Equal(t, f.Path, "file2.txt", "Unexpected filename")
-			} else if f.Id == f3.Id {
-				require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID")
-				require.Equal(t, f.Path, "file3.txt", "Unexpected filename")
-			} else {
-				require.Fail(t, "unexpected file returned")
-			}
-		}
-	})
 
-	t.Run("get files after certain date", func(t *testing.T) {
+	// Testing pagination
-		r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt+1, model.GetMillis()+100000, 100)
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 2)
-		require.NoError(t, err)
+	require.NoError(t, err)
-		require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r))
+	require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r))
-		for _, f := range r {
+
-			if f.Id == f2.Id {
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, 2)
-				require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID")
+	require.NoError(t, err)
-				require.Equal(t, f.Path, "file2.txt", "Unexpected filename")
+	require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r))
-			} else if f.Id == f3.Id {
+
-				require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID")
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, 2)
-				require.Equal(t, f.Path, "file3.txt", "Unexpected filename")
+	require.NoError(t, err)
-			} else {
+	require.Len(t, r, 0, "Expected 0 posts in results. Got %v", len(r))
-				require.Fail(t, "unexpected file returned")
-			}
-		}
-	})
 }
 
 func testFileInfoStoreCountAll(t *testing.T, ss store.Store) {

store/storetest/mocks/ChannelStore.go

@@ -697,13 +697,13 @@ func (_m *ChannelStore) GetChannels(teamID string, userID string, opts *model.Ch
 	return r0, r1
 }
 
-// GetChannelsBatchForIndexing provides a mock function with given fields: startTime, endTime, limit
+// GetChannelsBatchForIndexing provides a mock function with given fields: startTime, startChannelID, limit
-func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) {
+func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) {
-	ret := _m.Called(startTime, endTime, limit)
+	ret := _m.Called(startTime, startChannelID, limit)
 
 	var r0 []*model.Channel
-	if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.Channel); ok {
+	if rf, ok := ret.Get(0).(func(int64, string, int) []*model.Channel); ok {
-		r0 = rf(startTime, endTime, limit)
+		r0 = rf(startTime, startChannelID, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.Channel)
@@ -711,8 +711,8 @@ func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok {
+	if rf, ok := ret.Get(1).(func(int64, string, int) error); ok {
-		r1 = rf(startTime, endTime, limit)
+		r1 = rf(startTime, startChannelID, limit)
 	} else {
 		r1 = ret.Error(1)
 	}

store/storetest/mocks/FileInfoStore.go

@@ -144,13 +144,13 @@ func (_m *FileInfoStore) GetByPath(path string) (*model.FileInfo, error) {
 	return r0, r1
 }
 
-// GetFilesBatchForIndexing provides a mock function with given fields: startTime, endTime, limit
+// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit
-func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
-	ret := _m.Called(startTime, endTime, limit)
+	ret := _m.Called(startTime, startFileID, limit)
 
 	var r0 []*model.FileForIndexing
-	if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.FileForIndexing); ok {
+	if rf, ok := ret.Get(0).(func(int64, string, int) []*model.FileForIndexing); ok {
-		r0 = rf(startTime, endTime, limit)
+		r0 = rf(startTime, startFileID, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.FileForIndexing)
@@ -158,8 +158,8 @@ func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok {
+	if rf, ok := ret.Get(1).(func(int64, string, int) error); ok {
-		r1 = rf(startTime, endTime, limit)
+		r1 = rf(startTime, startFileID, limit)
 	} else {
 		r1 = ret.Error(1)
 	}

store/storetest/mocks/PostStore.go

@@ -465,13 +465,13 @@ func (_m *PostStore) GetPostsAfter(options model.GetPostsOptions) (*model.PostLi
 	return r0, r1
 }
 
-// GetPostsBatchForIndexing provides a mock function with given fields: startTime, endTime, limit
+// GetPostsBatchForIndexing provides a mock function with given fields: startTime, startPostID, limit
-func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) {
+func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) {
-	ret := _m.Called(startTime, endTime, limit)
+	ret := _m.Called(startTime, startPostID, limit)
 
 	var r0 []*model.PostForIndexing
-	if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.PostForIndexing); ok {
+	if rf, ok := ret.Get(0).(func(int64, string, int) []*model.PostForIndexing); ok {
-		r0 = rf(startTime, endTime, limit)
+		r0 = rf(startTime, startPostID, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.PostForIndexing)
@@ -479,8 +479,8 @@ func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, li
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok {
+	if rf, ok := ret.Get(1).(func(int64, string, int) error); ok {
-		r1 = rf(startTime, endTime, limit)
+		r1 = rf(startTime, startPostID, limit)
 	} else {
 		r1 = ret.Error(1)
 	}

store/storetest/mocks/UserStore.go

@@ -979,13 +979,13 @@ func (_m *UserStore) GetUnreadCountForChannel(userID string, channelID string) (
 	return r0, r1
 }
 
-// GetUsersBatchForIndexing provides a mock function with given fields: startTime, endTime, limit
+// GetUsersBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit
-func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) {
+func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) {
-	ret := _m.Called(startTime, endTime, limit)
+	ret := _m.Called(startTime, startFileID, limit)
 
 	var r0 []*model.UserForIndexing
-	if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.UserForIndexing); ok {
+	if rf, ok := ret.Get(0).(func(int64, string, int) []*model.UserForIndexing); ok {
-		r0 = rf(startTime, endTime, limit)
+		r0 = rf(startTime, startFileID, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.UserForIndexing)
@@ -993,8 +993,8 @@ func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, li
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok {
+	if rf, ok := ret.Get(1).(func(int64, string, int) error); ok {
-		r1 = rf(startTime, endTime, limit)
+		r1 = rf(startTime, startFileID, limit)
 	} else {
 		r1 = ret.Error(1)
 	}

store/storetest/post_store.go

@@ -2953,7 +2953,7 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) {
 	o2.ChannelId = c2.Id
 	o2.UserId = model.NewId()
 	o2.Message = NewTestId()
-	o2, err = ss.Post().Save(o2)
+	_, err = ss.Post().Save(o2)
 	require.NoError(t, err)
 
 	o3 := &model.Post{}
@@ -2961,26 +2961,30 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) {
 	o3.UserId = model.NewId()
 	o3.RootId = o1.Id
 	o3.Message = NewTestId()
-	o3, err = ss.Post().Save(o3)
+	_, err = ss.Post().Save(o3)
 	require.NoError(t, err)
 
-	r, err := ss.Post().GetPostsBatchForIndexing(o1.CreateAt, model.GetMillis()+100000, 100)
+	// Getting all
+	r, err := ss.Post().GetPostsBatchForIndexing(o1.CreateAt-1, "", 100)
 	require.NoError(t, err)
 	require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r))
-	for _, p := range r {
+
-		if p.Id == o1.Id {
+	// Testing pagination
-			require.Equal(t, p.TeamId, c1.TeamId, "Unexpected team ID")
+	r, err = ss.Post().GetPostsBatchForIndexing(o1.CreateAt-1, "", 1)
-			require.Nil(t, p.ParentCreateAt, "Unexpected parent create at")
+	require.NoError(t, err)
-		} else if p.Id == o2.Id {
+	require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r))
-			require.Equal(t, p.TeamId, c2.TeamId, "Unexpected team ID")
+
-			require.Nil(t, p.ParentCreateAt, "Unexpected parent create at")
+	r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1)
-		} else if p.Id == o3.Id {
+	require.NoError(t, err)
-			require.Equal(t, p.TeamId, c1.TeamId, "Unexpected team ID")
+	require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r))
-			require.Equal(t, *p.ParentCreateAt, o1.CreateAt, "Unexpected parent create at")
+
-		} else {
+	r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1)
-			require.Fail(t, "unexpected post returned")
+	require.NoError(t, err)
-		}
+	require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r))
-	}
+
+	r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1)
+	require.NoError(t, err)
+	require.Len(t, r, 0, "Expected 0 post in results. Got %v", len(r))
 }
 
 func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) {

store/storetest/user_store.go

@@ -4716,7 +4716,6 @@ func testUserStoreGetUsersBatchForIndexing(t *testing.T, ss store.Store) {
 	})
 	require.NoError(t, err)
 
-	startTime := u2.CreateAt
 	time.Sleep(time.Millisecond)
 
 	u3, err := ss.User().Save(&model.User{
@@ -4744,47 +4743,23 @@ func testUserStoreGetUsersBatchForIndexing(t *testing.T, ss store.Store) {
 	})
 	require.NoError(t, err)
 
-	endTime := u3.CreateAt
+	// Getting all users
-
+	res1List, err := ss.User().GetUsersBatchForIndexing(u1.CreateAt-1, "", 100)
-	// First and last user should be outside the range
-	res1List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100)
 	require.NoError(t, err)
+	assert.Len(t, res1List, 3)
 
-	assert.Len(t, res1List, 1)
+	// Testing pagination
-	assert.Equal(t, res1List[0].Username, u2.Username)
+	res2List, err := ss.User().GetUsersBatchForIndexing(u1.CreateAt-1, "", 1)
-	assert.ElementsMatch(t, res1List[0].TeamsIds, []string{t1.Id})
-	assert.ElementsMatch(t, res1List[0].ChannelsIds, []string{cPub1.Id, cPub2.Id})
-
-	// Update startTime to include first user
-	startTime = u1.CreateAt
-	res2List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100)
 	require.NoError(t, err)
+	assert.Len(t, res2List, 1)
 
+	res2List, err = ss.User().GetUsersBatchForIndexing(res2List[0].CreateAt, res2List[0].Id, 2)
+	require.NoError(t, err)
 	assert.Len(t, res2List, 2)
-	assert.Equal(t, res2List[0].Username, u1.Username)
-	assert.Equal(t, res2List[0].ChannelsIds, []string{})
-	assert.Equal(t, res2List[0].TeamsIds, []string{})
-	assert.Equal(t, res2List[1].Username, u2.Username)
 
-	// Update endTime to include last user
+	res2List, err = ss.User().GetUsersBatchForIndexing(res2List[1].CreateAt, res2List[1].Id, 2)
-	endTime = model.GetMillis()
-	res3List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100)
 	require.NoError(t, err)
-
+	assert.Len(t, res2List, 0)
-	assert.Len(t, res3List, 3)
-	assert.Equal(t, res3List[0].Username, u1.Username)
-	assert.Equal(t, res3List[1].Username, u2.Username)
-	assert.Equal(t, res3List[2].Username, u3.Username)
-	assert.ElementsMatch(t, res3List[2].TeamsIds, []string{})
-	assert.ElementsMatch(t, res3List[2].ChannelsIds, []string{cPub2.Id})
-
-	// Testing the limit
-	res4List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 2)
-	require.NoError(t, err)
-
-	assert.Len(t, res4List, 2)
-	assert.Equal(t, res4List[0].Username, u1.Username)
-	assert.Equal(t, res4List[1].Username, u2.Username)
 }
 
 func testUserStoreGetTeamGroupUsers(t *testing.T, ss store.Store) {

store/timerlayer/timerlayer.go

@@ -1029,10 +1029,10 @@ func (s *TimerLayerChannelStore) GetChannels(teamID string, userID string, opts
 	return result, err
 }
 
-func (s *TimerLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) {
+func (s *TimerLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) {
 	start := timemodule.Now()
 
-	result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit)
+	result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {
@@ -3036,10 +3036,10 @@ func (s *TimerLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error
 	return result, err
 }
 
-func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
 	start := timemodule.Now()
 
-	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {
@@ -5162,10 +5162,10 @@ func (s *TimerLayerPostStore) GetPostsAfter(options model.GetPostsOptions) (*mod
 	return result, err
 }
 
-func (s *TimerLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) {
+func (s *TimerLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) {
 	start := timemodule.Now()
 
-	result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit)
+	result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {
@@ -9557,10 +9557,10 @@ func (s *TimerLayerUserStore) GetUnreadCountForChannel(userID string, channelID
 	return result, err
 }
 
-func (s *TimerLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) {
+func (s *TimerLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) {
 	start := timemodule.Now()
 
-	result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit)
+	result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {

tests/test-config.json

@@ -340,7 +340,7 @@
         "PostsAggregatorJobStartTime": "03:00",
         "IndexPrefix": "",
         "LiveIndexingBatchSize": 1,
-        "BulkIndexingTimeWindowSeconds": 3600,
+        "BatchSize": 10000,
         "RequestTimeoutSeconds": 30
     },
     "DataRetentionSettings": {