[MM-69221] (#37301) (#37468)

Automatic Merge

server/channels/app/file.go

@@ -873,7 +873,7 @@ func (a *App) UploadFileX(rctx request.CTX, channelID, name string, input io.Rea
 				rctx.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
 		}) {
-			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until the scheduled content extraction catch-up job runs or an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
 		}
 	}
 
@@ -1142,7 +1142,7 @@ func (a *App) DoUploadFileExpectModification(rctx request.CTX, now time.Time, ra
 				rctx.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
 		}) {
-			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until the scheduled content extraction catch-up job runs or an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
 		}
 	}
 

server/channels/app/platform/goroutines.go

@@ -88,9 +88,9 @@ drain:
 
 // GoExtraction submits f to the bounded document extraction worker pool. It
 // never blocks the caller: if every worker is busy and the queue is full it
-// returns false without running f. Skipped files stay unextracted until an
-// admin runs a content extraction job (e.g. mmctl extract); there is no
-// scheduler that picks them up automatically. This keeps expensive extractions
+// returns false without running f. Skipped files stay unextracted until the
+// scheduled ExtractContent catch-up job or an admin runs a content extraction job
+// (e.g. mmctl extract). This keeps expensive extractions from stalling the
 // from stalling the request goroutines that dispatch them.
 func (ps *PlatformService) GoExtraction(f func()) bool {
 	select {

server/channels/app/server.go

@@ -877,8 +877,8 @@ func (s *Server) GoBuffered(f func()) {
 
 // GoExtraction submits f to the bounded document extraction worker pool without
 // blocking the caller. It returns false if the pool is saturated and f was not
-// run; skipped files stay unextracted until an admin runs a content extraction
-// job (e.g. mmctl extract).
+// run; skipped files stay unextracted until the scheduled ExtractContent catch-up
+// job or an admin runs a content extraction job (e.g. mmctl extract).
 func (s *Server) GoExtraction(f func()) bool {
 	return s.platform.GoExtraction(f)
 }
@@ -1680,7 +1680,7 @@ func (s *Server) initJobs() {
 	s.Jobs.RegisterJobType(
 		model.JobTypeExtractContent,
 		extract_content.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store()),
-		nil,
+		extract_content.MakeScheduler(s.Jobs),
 	)
 
 	s.Jobs.RegisterJobType(

server/channels/app/upload.go

@@ -354,7 +354,7 @@ func (a *App) UploadData(rctx request.CTX, us *model.UploadSession, rd io.Reader
 				rctx.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
 		}) {
-			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+			rctx.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until the scheduled content extraction catch-up job runs or an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
 		}
 	}
 

server/channels/jobs/extract_content/scheduler.go

@@ -0,0 +1,35 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package extract_content
+
+import (
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+)
+
+const schedFreq = 1 * time.Hour
+
+type Scheduler struct {
+	*jobs.PeriodicScheduler
+	jobServer *jobs.JobServer
+}
+
+func MakeScheduler(jobServer *jobs.JobServer) *Scheduler {
+	isEnabled := func(cfg *model.Config) bool {
+		return *cfg.FileSettings.ExtractContent
+	}
+	return &Scheduler{
+		PeriodicScheduler: jobs.NewPeriodicScheduler(jobServer, model.JobTypeExtractContent, schedFreq, isEnabled),
+		jobServer:         jobServer,
+	}
+}
+
+func (scheduler *Scheduler) ScheduleJob(rctx request.CTX, cfg *model.Config, pendingJobs bool, lastSuccessfulJob *model.Job) (*model.Job, *model.AppError) {
+	return scheduler.jobServer.CreateJob(rctx, model.JobTypeExtractContent, map[string]string{
+		catchupJobDataKey: "true",
+	})
+}

server/channels/jobs/extract_content/worker.go

@@ -5,6 +5,7 @@ package extract_content
 
 import (
 	"strconv"
+	"time"
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
@@ -13,6 +14,12 @@ import (
 	"github.com/mattermost/mattermost/server/v8/channels/store"
 )
 
+const (
+	catchupJobDataKey = "catchup"
+	catchupLookback   = 24 * time.Hour
+	catchupBatchSize  = 1000
+)
+
 var ignoredFiles = map[string]bool{
 	"png": true, "jpg": true, "jpeg": true, "gif": true, "wmv": true,
 	"mpg": true, "mpeg": true, "mp3": true, "mp4": true, "ogg": true,
@@ -33,64 +40,123 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface, store store.Store) *job
 	execute := func(logger mlog.LoggerIFace, job *model.Job) error {
 		jobServer.HandleJobPanic(logger, job)
 
-		var err error
-		var fromTS int64
-		var toTS int64 = model.GetMillis()
-		if fromStr, ok := job.Data["from"]; ok {
-			if fromTS, err = strconv.ParseInt(fromStr, 10, 64); err != nil {
-				return err
-			}
-			fromTS *= 1000
+		if job.Data[catchupJobDataKey] == "true" {
+			return runCatchupExtraction(logger, job, jobServer, app, store)
 		}
-		if toStr, ok := job.Data["to"]; ok {
-			if toTS, err = strconv.ParseInt(toStr, 10, 64); err != nil {
-				return err
-			}
-			toTS *= 1000
-		}
-
-		var nFiles int
-		var nErrs int
-		for {
-			opts := model.GetFileInfosOptions{
-				Since:          fromTS,
-				SortBy:         model.FileinfoSortByCreated,
-				IncludeDeleted: false,
-			}
-			fileInfos, err := store.FileInfo().GetWithOptions(0, 1000, &opts)
-			if err != nil {
-				return err
-			}
-			if len(fileInfos) == 0 {
-				break
-			}
-			for _, fileInfo := range fileInfos {
-				if !ignoredFiles[fileInfo.Extension] {
-					logger.Debug("Extracting file", mlog.String("filename", fileInfo.Name), mlog.String("filepath", fileInfo.Path))
-
-					err = app.ExtractContentFromFileInfo(request.EmptyContext(logger), fileInfo)
-					if err != nil {
-						logger.Warn("Failed to extract file content", mlog.Err(err), mlog.String("file_info_id", fileInfo.Id))
-						nErrs++
-					}
-					nFiles++
-				}
-			}
-			lastFileInfo := fileInfos[len(fileInfos)-1]
-			if lastFileInfo.CreateAt > toTS {
-				break
-			}
-			fromTS = lastFileInfo.CreateAt + 1
-		}
-
-		job.Data["errors"] = strconv.Itoa(nErrs)
-		job.Data["processed"] = strconv.Itoa(nFiles)
-
-		if err := jobServer.UpdateInProgressJobData(job); err != nil {
-			logger.Error("Worker: Failed to update job data", mlog.Err(err))
-		}
-		return nil
+		return runRangeExtraction(logger, job, jobServer, app, store)
 	}
 	worker := jobs.NewSimpleWorker(workerName, jobServer, execute, isEnabled)
 	return worker
 }
+
+func runCatchupExtraction(logger mlog.LoggerIFace, job *model.Job, jobServer *jobs.JobServer, app AppIface, store store.Store) error {
+	cursor := model.GetMillis() - catchupLookback.Milliseconds()
+
+	var nFiles int
+	var nErrs int
+	for {
+		opts := model.GetFileInfosOptions{
+			Since:            cursor,
+			SortBy:           model.FileinfoSortByCreated,
+			IncludeDeleted:   false,
+			OnlyEmptyContent: true,
+		}
+		fileInfos, err := store.FileInfo().GetWithOptions(0, catchupBatchSize, &opts)
+		if err != nil {
+			return err
+		}
+		if len(fileInfos) == 0 {
+			break
+		}
+
+		for _, fileInfo := range fileInfos {
+			if ignoredFiles[fileInfo.Extension] {
+				continue
+			}
+
+			logger.Debug("Extracting file", mlog.String("filename", fileInfo.Name), mlog.String("filepath", fileInfo.Path))
+
+			err = app.ExtractContentFromFileInfo(request.EmptyContext(logger), fileInfo)
+			if err != nil {
+				logger.Warn("Failed to extract file content", mlog.Err(err), mlog.String("file_info_id", fileInfo.Id))
+				nErrs++
+			}
+			nFiles++
+		}
+
+		lastFileInfo := fileInfos[len(fileInfos)-1]
+		cursor = lastFileInfo.CreateAt + 1
+
+		if len(fileInfos) < catchupBatchSize {
+			break
+		}
+	}
+
+	job.Data["errors"] = strconv.Itoa(nErrs)
+	job.Data["processed"] = strconv.Itoa(nFiles)
+
+	if err := jobServer.UpdateInProgressJobData(job); err != nil {
+		logger.Error("Worker: Failed to update job data", mlog.Err(err))
+	}
+	return nil
+}
+
+func runRangeExtraction(logger mlog.LoggerIFace, job *model.Job, jobServer *jobs.JobServer, app AppIface, store store.Store) error {
+	var err error
+	var fromTS int64
+	var toTS int64 = model.GetMillis()
+	if fromStr, ok := job.Data["from"]; ok {
+		if fromTS, err = strconv.ParseInt(fromStr, 10, 64); err != nil {
+			return err
+		}
+		fromTS *= 1000
+	}
+	if toStr, ok := job.Data["to"]; ok {
+		if toTS, err = strconv.ParseInt(toStr, 10, 64); err != nil {
+			return err
+		}
+		toTS *= 1000
+	}
+
+	var nFiles int
+	var nErrs int
+	for {
+		opts := model.GetFileInfosOptions{
+			Since:          fromTS,
+			SortBy:         model.FileinfoSortByCreated,
+			IncludeDeleted: false,
+		}
+		fileInfos, err := store.FileInfo().GetWithOptions(0, catchupBatchSize, &opts)
+		if err != nil {
+			return err
+		}
+		if len(fileInfos) == 0 {
+			break
+		}
+		for _, fileInfo := range fileInfos {
+			if !ignoredFiles[fileInfo.Extension] {
+				logger.Debug("Extracting file", mlog.String("filename", fileInfo.Name), mlog.String("filepath", fileInfo.Path))
+
+				err = app.ExtractContentFromFileInfo(request.EmptyContext(logger), fileInfo)
+				if err != nil {
+					logger.Warn("Failed to extract file content", mlog.Err(err), mlog.String("file_info_id", fileInfo.Id))
+					nErrs++
+				}
+				nFiles++
+			}
+		}
+		lastFileInfo := fileInfos[len(fileInfos)-1]
+		if lastFileInfo.CreateAt > toTS {
+			break
+		}
+		fromTS = lastFileInfo.CreateAt + 1
+	}
+
+	job.Data["errors"] = strconv.Itoa(nErrs)
+	job.Data["processed"] = strconv.Itoa(nFiles)
+
+	if err := jobServer.UpdateInProgressJobData(job); err != nil {
+		logger.Error("Worker: Failed to update job data", mlog.Err(err))
+	}
+	return nil
+}

server/channels/jobs/extract_content/worker_test.go

@@ -0,0 +1,257 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package extract_content
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+	"github.com/mattermost/mattermost/server/v8/channels/store/storetest"
+	"github.com/mattermost/mattermost/server/v8/channels/utils/testutils"
+)
+
+type trackingApp struct {
+	errOn map[string]error
+	calls []string
+}
+
+func (a *trackingApp) ExtractContentFromFileInfo(_ request.CTX, fileInfo *model.FileInfo) error {
+	a.calls = append(a.calls, fileInfo.Id)
+	if err, ok := a.errOn[fileInfo.Id]; ok {
+		return err
+	}
+	return nil
+}
+
+func makeTestJobServer(t *testing.T) (*jobs.JobServer, *storetest.Store) {
+	t.Helper()
+
+	mockStore := &storetest.Store{}
+	t.Cleanup(func() {
+		mockStore.AssertExpectations(t)
+	})
+
+	jobServer := jobs.NewJobServer(
+		&testutils.StaticConfigService{},
+		mockStore,
+		nil,
+		mlog.CreateConsoleTestLogger(t),
+	)
+
+	return jobServer, mockStore
+}
+
+func expectJobDataUpdate(mockStore *storetest.Store) {
+	mockStore.JobStore.On("UpdateOptimistically", mock.AnythingOfType("*model.Job"), model.JobStatusInProgress).Return(true, nil)
+}
+
+func expectWorkerJobCompletion(mockStore *storetest.Store, job *model.Job) {
+	claimed := *job
+	claimed.Status = model.JobStatusInProgress
+	mockStore.JobStore.On("UpdateStatusOptimistically", job.Id, model.JobStatusPending, model.JobStatusInProgress).Return(&claimed, nil)
+	mockStore.JobStore.On("UpdateStatus", job.Id, model.JobStatusSuccess).Return(&claimed, nil)
+}
+
+func makeFileInfo(id string, createAt int64, ext string) *model.FileInfo {
+	return &model.FileInfo{
+		Id:        id,
+		CreateAt:  createAt,
+		Extension: ext,
+		Name:      "file." + ext,
+		Path:      "path/" + id,
+	}
+}
+
+func makeFileInfoBatch(n int, startCreateAt int64) []*model.FileInfo {
+	batch := make([]*model.FileInfo, n)
+	for i := range n {
+		batch[i] = makeFileInfo(model.NewId(), startCreateAt+int64(i), "pdf")
+	}
+	return batch
+}
+
+func TestRunCatchupExtraction(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+
+	t.Run("extracts non-ignored files and passes OnlyEmptyContent filter", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		pdf := makeFileInfo("pdf1", 1000, "pdf")
+		png := makeFileInfo("png1", 1001, "png")
+		docx := makeFileInfo("docx1", 1002, "docx")
+
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.MatchedBy(func(opt *model.GetFileInfosOptions) bool {
+			return opt.OnlyEmptyContent && !opt.IncludeDeleted && opt.SortBy == model.FileinfoSortByCreated
+		})).Return([]*model.FileInfo{pdf, png, docx}, nil).Once()
+
+		app := &trackingApp{}
+		job := &model.Job{Data: map[string]string{}}
+
+		err := runCatchupExtraction(logger, job, jobServer, app, mockStore)
+		require.NoError(t, err)
+
+		require.Equal(t, []string{"pdf1", "docx1"}, app.calls)
+		require.Equal(t, "2", job.Data["processed"])
+		require.Equal(t, "0", job.Data["errors"])
+	})
+
+	t.Run("empty result is a no-op", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.Anything).Return(nil, nil).Once()
+
+		job := &model.Job{Data: map[string]string{}}
+		err := runCatchupExtraction(logger, job, jobServer, &trackingApp{}, mockStore)
+		require.NoError(t, err)
+
+		require.Equal(t, "0", job.Data["processed"])
+		require.Equal(t, "0", job.Data["errors"])
+	})
+
+	t.Run("full batch advances cursor and fetches the next page", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		first := makeFileInfoBatch(catchupBatchSize, 1000)
+		second := []*model.FileInfo{makeFileInfo("tail1", 2000, "pdf"), makeFileInfo("tail2", 2001, "pdf")}
+
+		var firstPageOpts *model.GetFileInfosOptions
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.MatchedBy(func(opt *model.GetFileInfosOptions) bool {
+			return opt.OnlyEmptyContent
+		})).Run(func(args mock.Arguments) {
+			opt := args.Get(2).(*model.GetFileInfosOptions)
+			copied := *opt
+			firstPageOpts = &copied
+		}).Return(first, nil).Once()
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.MatchedBy(func(opt *model.GetFileInfosOptions) bool {
+			return opt.Since == 1999+1
+		})).Return(second, nil).Once()
+
+		sinceLower := model.GetMillis() - catchupLookback.Milliseconds()
+		app := &trackingApp{}
+		job := &model.Job{Data: map[string]string{}}
+
+		err := runCatchupExtraction(logger, job, jobServer, app, mockStore)
+		sinceUpper := model.GetMillis() - catchupLookback.Milliseconds()
+		require.NoError(t, err)
+
+		require.NotNil(t, firstPageOpts)
+		require.GreaterOrEqual(t, firstPageOpts.Since, sinceLower)
+		require.LessOrEqual(t, firstPageOpts.Since, sinceUpper)
+		require.Len(t, app.calls, catchupBatchSize+2)
+		require.Equal(t, "1002", job.Data["processed"])
+	})
+
+	t.Run("partial batch terminates after one fetch", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		batch := []*model.FileInfo{
+			makeFileInfo("f1", 1000, "pdf"),
+			makeFileInfo("f2", 1001, "pdf"),
+			makeFileInfo("f3", 1002, "pdf"),
+		}
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.Anything).Return(batch, nil).Once()
+
+		job := &model.Job{Data: map[string]string{}}
+		err := runCatchupExtraction(logger, job, jobServer, &trackingApp{}, mockStore)
+		require.NoError(t, err)
+
+		require.Equal(t, "3", job.Data["processed"])
+	})
+
+	t.Run("accumulates extraction errors in job data", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		okFile := makeFileInfo("ok", 1000, "pdf")
+		failFile := makeFileInfo("fail", 1001, "pdf")
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.Anything).
+			Return([]*model.FileInfo{okFile, failFile}, nil).Once()
+
+		app := &trackingApp{errOn: map[string]error{"fail": errors.New("extract failed")}}
+		job := &model.Job{Data: map[string]string{}}
+
+		err := runCatchupExtraction(logger, job, jobServer, app, mockStore)
+		require.NoError(t, err)
+
+		require.Equal(t, "2", job.Data["processed"])
+		require.Equal(t, "1", job.Data["errors"])
+	})
+
+	t.Run("store error propagates", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+
+		wantErr := errors.New("store failed")
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.Anything).Return(nil, wantErr).Once()
+
+		job := &model.Job{Data: map[string]string{}}
+		err := runCatchupExtraction(logger, job, jobServer, &trackingApp{}, mockStore)
+		require.ErrorIs(t, err, wantErr)
+	})
+}
+
+func TestWorkerDispatch(t *testing.T) {
+	t.Run("catchup job uses OnlyEmptyContent filter", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.MatchedBy(func(opt *model.GetFileInfosOptions) bool {
+			return opt.OnlyEmptyContent
+		})).Return(nil, nil).Once()
+
+		worker := MakeWorker(jobServer, &trackingApp{}, mockStore)
+		job := &model.Job{Id: model.NewId(), Data: map[string]string{catchupJobDataKey: "true"}}
+		expectWorkerJobCompletion(mockStore, job)
+
+		worker.DoJob(job)
+	})
+
+	t.Run("range job does not use OnlyEmptyContent filter", func(t *testing.T) {
+		jobServer, mockStore := makeTestJobServer(t)
+		expectJobDataUpdate(mockStore)
+
+		mockStore.FileInfoStore.On("GetWithOptions", 0, catchupBatchSize, mock.MatchedBy(func(opt *model.GetFileInfosOptions) bool {
+			return !opt.OnlyEmptyContent
+		})).Return(nil, nil).Once()
+
+		worker := MakeWorker(jobServer, &trackingApp{}, mockStore)
+		job := &model.Job{Id: model.NewId(), Data: map[string]string{}}
+		expectWorkerJobCompletion(mockStore, job)
+
+		worker.DoJob(job)
+	})
+}
+
+func TestSchedulerScheduleJob(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	jobServer, mockStore := makeTestJobServer(t)
+
+	app := &trackingApp{}
+	jobServer.RegisterJobType(model.JobTypeExtractContent, MakeWorker(jobServer, app, mockStore), nil)
+
+	savedJob := &model.Job{
+		Id:   model.NewId(),
+		Type: model.JobTypeExtractContent,
+		Data: map[string]string{catchupJobDataKey: "true"},
+	}
+	mockStore.JobStore.On("Save", mock.MatchedBy(func(job *model.Job) bool {
+		return job.Type == model.JobTypeExtractContent && job.Data[catchupJobDataKey] == "true"
+	})).Return(savedJob, nil)
+
+	scheduler := MakeScheduler(jobServer)
+	job, appErr := scheduler.ScheduleJob(request.EmptyContext(logger), &model.Config{}, false, nil)
+	require.Nil(t, appErr)
+	require.NotNil(t, job)
+	require.Equal(t, "true", job.Data[catchupJobDataKey])
+}

server/channels/store/sqlstore/file_info_store.go

@@ -287,6 +287,10 @@ func (fs SqlFileInfoStore) GetWithOptions(page, perPage int, opt *model.GetFileI
 		query = query.Where("FileInfo.DeleteAt = 0")
 	}
 
+	if opt.OnlyEmptyContent {
+		query = query.Where("(FileInfo.Content IS NULL OR FileInfo.Content = '')")
+	}
+
 	if opt.SortBy == "" {
 		opt.SortBy = model.FileinfoSortByCreated
 	}

server/channels/store/storetest/file_info_store.go

@@ -26,6 +26,7 @@ func TestFileInfoStore(t *testing.T, rctx request.CTX, ss store.Store, s SqlStor
 	t.Run("FileInfoGetForPost", func(t *testing.T) { testFileInfoGetForPost(t, rctx, ss) })
 	t.Run("FileInfoGetForUser", func(t *testing.T) { testFileInfoGetForUser(t, rctx, ss) })
 	t.Run("FileInfoGetWithOptions", func(t *testing.T) { testFileInfoGetWithOptions(t, rctx, ss) })
+	t.Run("FileInfoGetWithOptionsOnlyEmptyContent", func(t *testing.T) { testFileInfoGetWithOptionsOnlyEmptyContent(t, rctx, ss) })
 	t.Run("FileInfoAttachToPost", func(t *testing.T) { testFileInfoAttachToPost(t, rctx, ss) })
 	t.Run("FileInfoDeleteForPost", func(t *testing.T) { testFileInfoDeleteForPost(t, rctx, ss) })
 	t.Run("FileInfoPermanentDelete", func(t *testing.T) { testFileInfoPermanentDelete(t, rctx, ss) })
@@ -421,6 +422,52 @@ func testFileInfoGetWithOptions(t *testing.T, rctx request.CTX, ss store.Store)
 	}
 }
 
+func testFileInfoGetWithOptionsOnlyEmptyContent(t *testing.T, rctx request.CTX, ss store.Store) {
+	post := &model.Post{
+		ChannelId: model.NewId(),
+		UserId:    model.NewId(),
+	}
+	_, err := ss.Post().Save(rctx, post)
+	require.NoError(t, err)
+
+	createAt := model.GetMillis()
+	emptyFile := &model.FileInfo{
+		Id:        model.NewId(),
+		CreatorId: post.UserId,
+		PostId:    post.Id,
+		ChannelId: post.ChannelId,
+		Path:      "empty.txt",
+		Name:      "empty.txt",
+		Extension: "txt",
+		CreateAt:  createAt,
+	}
+	_, err = ss.FileInfo().Save(rctx, emptyFile)
+	require.NoError(t, err)
+
+	indexedFile := &model.FileInfo{
+		Id:        model.NewId(),
+		CreatorId: post.UserId,
+		PostId:    post.Id,
+		ChannelId: post.ChannelId,
+		Path:      "indexed.txt",
+		Name:      "indexed.txt",
+		Extension: "txt",
+		CreateAt:  createAt + 1,
+	}
+	_, err = ss.FileInfo().Save(rctx, indexedFile)
+	require.NoError(t, err)
+	err = ss.FileInfo().SetContent(rctx, indexedFile.Id, "already indexed")
+	require.NoError(t, err)
+
+	fileInfos, err := ss.FileInfo().GetWithOptions(0, 10, &model.GetFileInfosOptions{
+		Since:            createAt - 1,
+		OnlyEmptyContent: true,
+	})
+	require.NoError(t, err)
+	require.Len(t, fileInfos, 1)
+	assert.Equal(t, emptyFile.Id, fileInfos[0].Id)
+}
+
 func testFileInfoAttachToPost(t *testing.T, rctx request.CTX, ss store.Store) {
 	t.Run("should attach files", func(t *testing.T) {
 		userID := model.NewId()

server/platform/services/docextractor/concurrency.go

@@ -0,0 +1,40 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package docextractor
+
+import (
+	"runtime"
+	"sync"
+)
+
+// maxConcurrentExtractions caps how many docconv extractions may run at once,
+// including detached goroutines that continue after a per-extraction timeout.
+var maxConcurrentExtractions = runtime.NumCPU()
+
+var (
+	extractionSlotsMu sync.Mutex
+	extractionSlots   = make(chan struct{}, maxConcurrentExtractions)
+)
+
+func tryAcquireExtractionSlot() bool {
+	select {
+	case extractionSlots <- struct{}{}:
+		return true
+	default:
+		return false
+	}
+}
+
+func releaseExtractionSlot() {
+	<-extractionSlots
+}
+
+// resetExtractionConcurrencyForTest reinitializes the extraction slot semaphore.
+// It is only intended for use from tests in this package.
+func resetExtractionConcurrencyForTest(limit int) {
+	extractionSlotsMu.Lock()
+	defer extractionSlotsMu.Unlock()
+	maxConcurrentExtractions = limit
+	extractionSlots = make(chan struct{}, limit)
+}

server/platform/services/docextractor/docextractor.go

@@ -18,14 +18,12 @@ type ExtractSettings struct {
 	MMPreviewURL     string
 	MMPreviewSecret  string
 	// Timeout bounds how long a caller waits for a single extraction. A value
-	// <= 0 disables it. NOTE: this bounds wall-clock wait time (and thus how
-	// long an extraction occupies its caller's worker slot), NOT CPU work.
+	// <= 0 disables it. NOTE: this bounds wall-clock wait time, not CPU work.
 	// The docconv converters are not context-aware, so on timeout the
-	// converter keeps running to completion on a detached goroutine and keeps
-	// consuming CPU until it finishes on its own. Under sustained load,
-	// detached extractions can therefore accumulate and run concurrently. The
-	// primary bound on the work of any single extraction is MaxFileSize, which
-	// limits how much input the converter reads.
+	// converter keeps running to completion on a detached goroutine. A global
+	// cap on concurrent extractions (including those detached goroutines)
+	// prevents sustained uploads from accumulating unbounded docconv work.
+	// The per-extraction input bound is MaxFileSize.
 	Timeout time.Duration
 	// ReaderCloser, when set, transfers ownership of closing the input reader
 	// to this package. It is closed only after extraction has actually
@@ -79,17 +77,19 @@ func ExtractWithExtraExtractors(logger mlog.LoggerIFace, filename string, r io.R
 // settings.Timeout elapses. Because the underlying docconv converters are not
 // context-aware, the extraction runs on a detached goroutine: on timeout we
 // stop waiting and return an error, releasing the caller (and its worker slot)
-// even though the converter keeps running.
-//
-// This decouples extraction from the caller, but it does NOT cap CPU: the
-// detached converter continues to completion in the background, so a sustained
-// stream of expensive documents can leave several detached extractions running
-// at once. The per-extraction work is bounded instead by MaxFileSize (input
-// size). Load-shedding on the number of in-flight detached extractions is a
-// possible future improvement; it is intentionally not done here so it does
-// not also throttle the backfill job that re-extracts skipped content.
+// even though the converter keeps running. A global concurrency cap applies to
+// every in-flight extraction, including detached goroutines, so timed-out work
+// cannot accumulate beyond the limit.
 func extractWithTimeout(e Extractor, filename string, r io.ReadSeeker, settings ExtractSettings) (string, error) {
+	if !tryAcquireExtractionSlot() {
+		if settings.ReaderCloser != nil {
+			settings.ReaderCloser.Close()
+		}
+		return "", fmt.Errorf("document text extraction capacity exhausted (%d concurrent extractions)", maxConcurrentExtractions)
+	}
+
 	if settings.Timeout <= 0 {
+		defer releaseExtractionSlot()
 		if settings.ReaderCloser != nil {
 			defer settings.ReaderCloser.Close()
 		}
@@ -102,6 +102,7 @@ func extractWithTimeout(e Extractor, filename string, r io.ReadSeeker, settings
 	}
 	resultCh := make(chan extractResult, 1)
 	go func() {
+		defer releaseExtractionSlot()
 		// This goroutine owns the reader for the lifetime of the extraction.
 		// After the timeout fires the caller returns, but the converter may
 		// still be reading r here, so the reader is closed only once this

server/platform/services/docextractor/docextractor_test.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"runtime"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -217,6 +218,7 @@ func TestExtractWithExtraExtractors(t *testing.T) {
 
 type slowExtractor struct {
 	delay time.Duration
+	done  chan struct{}
 }
 
 func (se *slowExtractor) Name() string { return "slowExtractor" }
@@ -224,6 +226,11 @@ func (se *slowExtractor) Name() string { return "slowExtractor" }
 func (se *slowExtractor) Match(filename string) bool { return true }
 
 func (se *slowExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (string, error) {
+	defer func() {
+		if se.done != nil {
+			close(se.done)
+		}
+	}()
 	time.Sleep(se.delay)
 	return "done", nil
 }
@@ -233,14 +240,21 @@ func TestExtractTimeout(t *testing.T) {
 	data := []byte("hello world")
 
 	t.Run("aborts a slow extraction once the timeout elapses", func(t *testing.T) {
+		extractDone := make(chan struct{})
 		start := time.Now()
-		text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: 50 * time.Millisecond}, []Extractor{&slowExtractor{delay: 10 * time.Second}})
+		text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: 50 * time.Millisecond}, []Extractor{&slowExtractor{delay: 500 * time.Millisecond, done: extractDone}})
 		elapsed := time.Since(start)
 
 		require.Error(t, err)
 		require.Empty(t, text)
 		assert.Contains(t, err.Error(), "timed out")
-		assert.Less(t, elapsed, 5*time.Second, "should return shortly after the timeout, not wait for the extraction")
+		assert.Less(t, elapsed, 200*time.Millisecond, "should return shortly after the timeout, not wait for the extraction")
+
+		select {
+		case <-extractDone:
+		case <-time.After(2 * time.Second):
+			require.FailNow(t, "detached extraction did not finish within the deadline")
+		}
 	})
 
 	t.Run("returns the result when extraction finishes within the timeout", func(t *testing.T) {
@@ -287,6 +301,7 @@ func (c *recordingCloser) Close() error {
 type blockingExtractor struct {
 	started chan struct{}
 	release chan struct{}
+	done    chan struct{}
 }
 
 func (be *blockingExtractor) Name() string { return "blockingExtractor" }
@@ -296,6 +311,9 @@ func (be *blockingExtractor) Match(filename string) bool { return true }
 func (be *blockingExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (string, error) {
 	close(be.started)
 	<-be.release
+	if be.done != nil {
+		close(be.done)
+	}
 	return "done", nil
 }
 
@@ -421,3 +439,66 @@ func TestArchiveMaxFileSize(t *testing.T) {
 		})
 	}
 }
+
+func TestExtractConcurrency(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	resetExtractionConcurrencyForTest(1)
+	t.Cleanup(func() { resetExtractionConcurrencyForTest(runtime.NumCPU()) })
+
+	data := []byte("hello world")
+
+	t.Run("rejects new extractions while the concurrency limit is reached", func(t *testing.T) {
+		be := &blockingExtractor{started: make(chan struct{}), release: make(chan struct{}), done: make(chan struct{})}
+		settings := ExtractSettings{Timeout: 50 * time.Millisecond, ReaderCloser: &recordingCloser{}}
+
+		go func() {
+			_, _ = ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), settings, []Extractor{be})
+		}()
+
+		select {
+		case <-be.started:
+		case <-time.After(2 * time.Second):
+			require.FailNow(t, "first extraction did not start within the deadline")
+		}
+
+		text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: time.Second}, []Extractor{&slowExtractor{delay: 0}})
+		require.Error(t, err)
+		require.Empty(t, text)
+		require.Contains(t, err.Error(), "capacity exhausted")
+
+		close(be.release)
+		select {
+		case <-be.done:
+		case <-time.After(2 * time.Second):
+			require.FailNow(t, "detached extraction did not finish within the deadline")
+		}
+	})
+
+	t.Run("a timed-out extraction keeps its slot until the detached goroutine finishes", func(t *testing.T) {
+		resetExtractionConcurrencyForTest(1)
+
+		be := &blockingExtractor{started: make(chan struct{}), release: make(chan struct{}), done: make(chan struct{})}
+		settings := ExtractSettings{Timeout: 50 * time.Millisecond, ReaderCloser: &recordingCloser{}}
+
+		_, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), settings, []Extractor{be})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "timed out")
+
+		select {
+		case <-be.started:
+		case <-time.After(2 * time.Second):
+			require.FailNow(t, "detached extraction did not start within the deadline")
+		}
+
+		_, err = ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: time.Second}, []Extractor{&slowExtractor{delay: 0}})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "capacity exhausted")
+
+		close(be.release)
+		select {
+		case <-be.done:
+		case <-time.After(2 * time.Second):
+			require.FailNow(t, "detached extraction did not finish within the deadline")
+		}
+	})
+}

server/public/model/file_info.go

@@ -51,6 +51,8 @@ type GetFileInfosOptions struct {
 	SortBy string `json:"sort_by"`
 	// SortDescending changes the sort direction to descending order when true.
 	SortDescending bool `json:"sort_descending"`
+	// OnlyEmptyContent limits results to files that have no extracted content stored.
+	OnlyEmptyContent bool `json:"only_empty_content"`
 }
 
 type FileInfo struct {