From fdcecdeb337d8f3eef3ed68c8e5359b75d53eba0 Mon Sep 17 00:00:00 2001 From: mattermost-code Date: Fri, 17 Jul 2026 06:59:21 -0400 Subject: [PATCH] [MM-69221] (#37301) (#37468) Automatic Merge --- server/channels/app/file.go | 4 +- server/channels/app/platform/goroutines.go | 6 +- server/channels/app/server.go | 6 +- server/channels/app/upload.go | 2 +- .../jobs/extract_content/scheduler.go | 35 +++ .../channels/jobs/extract_content/worker.go | 178 ++++++++---- .../jobs/extract_content/worker_test.go | 257 ++++++++++++++++++ .../store/sqlstore/file_info_store.go | 4 + .../store/storetest/file_info_store.go | 47 ++++ .../services/docextractor/concurrency.go | 40 +++ .../services/docextractor/docextractor.go | 33 +-- .../docextractor/docextractor_test.go | 85 +++++- server/public/model/file_info.go | 2 + 13 files changed, 616 insertions(+), 83 deletions(-) create mode 100644 server/channels/jobs/extract_content/scheduler.go create mode 100644 server/channels/jobs/extract_content/worker_test.go create mode 100644 server/platform/services/docextractor/concurrency.go diff --git a/server/channels/app/file.go b/server/channels/app/file.go index 448066fc655..96a61516b16 100644 --- a/server/channels/app/file.go +++ b/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)) } } diff --git a/server/channels/app/platform/goroutines.go b/server/channels/app/platform/goroutines.go index 62898996ac4..39d8d0ce2db 100644 --- a/server/channels/app/platform/goroutines.go +++ b/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 { diff --git a/server/channels/app/server.go b/server/channels/app/server.go index e93c20af4f8..f2de199f39a 100644 --- a/server/channels/app/server.go +++ b/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( diff --git a/server/channels/app/upload.go b/server/channels/app/upload.go index 6a86d5b3d54..6290e474a5b 100644 --- a/server/channels/app/upload.go +++ b/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)) } } diff --git a/server/channels/jobs/extract_content/scheduler.go b/server/channels/jobs/extract_content/scheduler.go new file mode 100644 index 00000000000..8c654de0831 --- /dev/null +++ b/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", + }) +} diff --git a/server/channels/jobs/extract_content/worker.go b/server/channels/jobs/extract_content/worker.go index dbdab9e4075..1a085b0122f 100644 --- a/server/channels/jobs/extract_content/worker.go +++ b/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 +} diff --git a/server/channels/jobs/extract_content/worker_test.go b/server/channels/jobs/extract_content/worker_test.go new file mode 100644 index 00000000000..87e49cedae1 --- /dev/null +++ b/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]) +} diff --git a/server/channels/store/sqlstore/file_info_store.go b/server/channels/store/sqlstore/file_info_store.go index 3f8318dbe19..e1ffbcda033 100644 --- a/server/channels/store/sqlstore/file_info_store.go +++ b/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 } diff --git a/server/channels/store/storetest/file_info_store.go b/server/channels/store/storetest/file_info_store.go index e85dc6b9c87..ca856f6f8b0 100644 --- a/server/channels/store/storetest/file_info_store.go +++ b/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() diff --git a/server/platform/services/docextractor/concurrency.go b/server/platform/services/docextractor/concurrency.go new file mode 100644 index 00000000000..ebfe09b8679 --- /dev/null +++ b/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) +} diff --git a/server/platform/services/docextractor/docextractor.go b/server/platform/services/docextractor/docextractor.go index e3e681b6ce4..4e123decb4e 100644 --- a/server/platform/services/docextractor/docextractor.go +++ b/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 diff --git a/server/platform/services/docextractor/docextractor_test.go b/server/platform/services/docextractor/docextractor_test.go index 88052674737..66556ba2b29 100644 --- a/server/platform/services/docextractor/docextractor_test.go +++ b/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") + } + }) +} diff --git a/server/public/model/file_info.go b/server/public/model/file_info.go index a276c6dfedf..286b462bfd8 100644 --- a/server/public/model/file_info.go +++ b/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 {