From acc19baca0b0e22e3aa5b4591d93341f14e7487b Mon Sep 17 00:00:00 2001 From: Julien Tant <785518+JulienTant@users.noreply.github.com> Date: Sun, 14 Jun 2026 23:35:26 -0700 Subject: [PATCH] Bound document content extraction time and decouple it from uploads (MM-69098) (#36856) (#37043) Automatic Merge --- .../support/api/on_prem_default_config.json | 1 + .../lib/src/server/default_config.ts | 1 + server/channels/app/file.go | 19 ++- server/channels/app/platform/goroutines.go | 57 ++++++- .../channels/app/platform/goroutines_test.go | 79 ++++++++++ server/channels/app/platform/service.go | 15 ++ server/channels/app/server.go | 8 + server/channels/app/upload.go | 6 +- server/i18n/en.json | 4 + .../services/docextractor/docextractor.go | 85 ++++++++++- .../docextractor/docextractor_test.go | 140 ++++++++++++++++++ .../services/docextractor/documents.go | 13 +- server/platform/services/docextractor/pdf.go | 13 +- .../services/docextractor/pdf_test.go | 27 ++++ server/public/model/config.go | 9 ++ server/public/model/config_test.go | 32 ++++ .../admin_console/admin_definition.tsx | 12 ++ webapp/channels/src/i18n/en.json | 4 + webapp/platform/types/src/config.ts | 1 + 19 files changed, 513 insertions(+), 13 deletions(-) create mode 100644 server/channels/app/platform/goroutines_test.go diff --git a/e2e-tests/cypress/tests/support/api/on_prem_default_config.json b/e2e-tests/cypress/tests/support/api/on_prem_default_config.json index a8ba0aa840..0ddd88e07c 100644 --- a/e2e-tests/cypress/tests/support/api/on_prem_default_config.json +++ b/e2e-tests/cypress/tests/support/api/on_prem_default_config.json @@ -215,6 +215,7 @@ "Directory": "./data/", "EnablePublicLink": false, "ExtractContent": true, + "ExtractContentTimeout": 10, "ArchiveRecursion": false, "PublicLinkSalt": "", "InitialFont": "nunito-bold.ttf", diff --git a/e2e-tests/playwright/lib/src/server/default_config.ts b/e2e-tests/playwright/lib/src/server/default_config.ts index 8887dc41dc..7f0c7d8253 100644 --- a/e2e-tests/playwright/lib/src/server/default_config.ts +++ b/e2e-tests/playwright/lib/src/server/default_config.ts @@ -310,6 +310,7 @@ const defaultServerConfig: AdminConfig = { Directory: './data/', EnablePublicLink: false, ExtractContent: true, + ExtractContentTimeout: 10, ArchiveRecursion: false, PublicLinkSalt: '', InitialFont: 'nunito-bold.ttf', diff --git a/server/channels/app/file.go b/server/channels/app/file.go index 254b4e7d04..a2fb7c1481 100644 --- a/server/channels/app/file.go +++ b/server/channels/app/file.go @@ -858,12 +858,14 @@ func (a *App) UploadFileX(c request.CTX, channelID, name string, input io.Reader if *a.Config().FileSettings.ExtractContent && t.ExtractContent { infoCopy := *t.fileinfo - a.Srv().GoBuffered(func() { + if !a.Srv().GoExtraction(func() { err := a.ExtractContentFromFileInfo(c, &infoCopy) if err != nil { c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id)) } - }) + }) { + c.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)) + } } return t.fileinfo, nil @@ -1125,12 +1127,14 @@ func (a *App) DoUploadFileExpectModification(c request.CTX, now time.Time, rawTe // and something we can do without. if *a.Config().FileSettings.ExtractContent && extractContent { infoCopy := *info - a.Srv().GoBuffered(func() { + if !a.Srv().GoExtraction(func() { err := a.ExtractContentFromFileInfo(c, &infoCopy) if err != nil { c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id)) } - }) + }) { + c.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)) + } } return info, data, nil @@ -1584,10 +1588,15 @@ func (a *App) ExtractContentFromFileInfo(rctx request.CTX, fileInfo *model.FileI if aerr != nil { return errors.Wrap(aerr, "failed to open file for extract file content") } - defer file.Close() + // Ownership of closing the file is handed to docextractor.Extract via + // ReaderCloser: with a timeout configured, extraction may continue on a + // detached goroutine after Extract returns, so closing the file here would + // race with that goroutine still reading it. text, err := docextractor.Extract(rctx.Logger(), fileInfo.Name, file, docextractor.ExtractSettings{ ArchiveRecursion: *a.Config().FileSettings.ArchiveRecursion, MaxFileSize: *a.Config().FileSettings.MaxFileSize, + Timeout: time.Duration(*a.Config().FileSettings.ExtractContentTimeout) * time.Second, + ReaderCloser: file, }) if err != nil { return errors.Wrap(err, "failed to extract file content") diff --git a/server/channels/app/platform/goroutines.go b/server/channels/app/platform/goroutines.go index fa8407da75..62898996ac 100644 --- a/server/channels/app/platform/goroutines.go +++ b/server/channels/app/platform/goroutines.go @@ -3,7 +3,10 @@ package platform -import "sync/atomic" +import ( + "runtime" + "sync/atomic" +) // Go creates a goroutine, but maintains a record of it to ensure that execution completes before // the server is shutdown. @@ -45,3 +48,55 @@ func (ps *PlatformService) GoBuffered(f func()) { <-ps.goroutineBuffered }() } + +// startExtractionWorkers launches the fixed-size pool of workers that run +// document extraction tasks submitted through GoExtraction. +func (ps *PlatformService) startExtractionWorkers() { + numWorkers := runtime.NumCPU() + for range numWorkers { + ps.extractionWG.Go(func() { + for { + select { + case <-ps.extractionStop: + return + case f := <-ps.extractionQueue: + f() + } + } + }) + } +} + +// stopExtractionWorkers signals the extraction workers to exit and waits for +// any in-flight extraction to finish. Queued-but-not-started tasks are drained +// and discarded so a worker cannot dequeue and run them after shutdown has been +// signaled. +func (ps *PlatformService) stopExtractionWorkers() { + close(ps.extractionStop) + +drain: + for { + select { + case <-ps.extractionQueue: + default: + break drain + } + } + + ps.extractionWG.Wait() +} + +// 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 +// from stalling the request goroutines that dispatch them. +func (ps *PlatformService) GoExtraction(f func()) bool { + select { + case ps.extractionQueue <- f: + return true + default: + return false + } +} diff --git a/server/channels/app/platform/goroutines_test.go b/server/channels/app/platform/goroutines_test.go new file mode 100644 index 0000000000..df9c796413 --- /dev/null +++ b/server/channels/app/platform/goroutines_test.go @@ -0,0 +1,79 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package platform + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestGoExtraction(t *testing.T) { + mainHelper.Parallel(t) + + t.Run("runs submitted work on the pool", func(t *testing.T) { + const tasks = 5 + ps := &PlatformService{ + extractionQueue: make(chan func(), tasks), + extractionStop: make(chan struct{}), + } + ps.startExtractionWorkers() + defer ps.stopExtractionWorkers() + + var wg sync.WaitGroup + wg.Add(tasks) + for range tasks { + require.True(t, ps.GoExtraction(func() { + wg.Done() + })) + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + require.Fail(t, "submitted extraction tasks did not run") + } + }) + + t.Run("never blocks and skips work once the queue is saturated", func(t *testing.T) { + // No workers are started, so nothing drains the queue. + ps := &PlatformService{ + extractionQueue: make(chan func(), 2), + extractionStop: make(chan struct{}), + } + + require.True(t, ps.GoExtraction(func() {})) + require.True(t, ps.GoExtraction(func() {})) + // The queue is now full; further submissions must be rejected rather + // than block the caller. + require.False(t, ps.GoExtraction(func() {})) + }) + + t.Run("stop waits for in-flight extraction to finish", func(t *testing.T) { + ps := &PlatformService{ + extractionQueue: make(chan func(), 1), + extractionStop: make(chan struct{}), + } + ps.startExtractionWorkers() + + var finished bool + started := make(chan struct{}) + require.True(t, ps.GoExtraction(func() { + close(started) + time.Sleep(100 * time.Millisecond) + finished = true + })) + + <-started + ps.stopExtractionWorkers() + require.True(t, finished, "stopExtractionWorkers should wait for the running task to complete") + }) +} diff --git a/server/channels/app/platform/service.go b/server/channels/app/platform/service.go index 3910070ac4..01484ca852 100644 --- a/server/channels/app/platform/service.go +++ b/server/channels/app/platform/service.go @@ -107,6 +107,13 @@ type PlatformService struct { goroutineExitSignal chan struct{} goroutineBuffered chan struct{} + // Document content extraction runs on a dedicated, bounded worker pool so + // that expensive extractions cannot saturate the generic worker pool and + // block the request goroutines that dispatch them. + extractionQueue chan func() + extractionStop chan struct{} + extractionWG sync.WaitGroup + additionalClusterHandlers map[model.ClusterEvent]einterfaces.ClusterMessageHandler shareChannelServiceMux sync.RWMutex @@ -136,6 +143,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { hashSeed: maphash.MakeSeed(), goroutineExitSignal: make(chan struct{}, 1), goroutineBuffered: make(chan struct{}, runtime.NumCPU()), + extractionQueue: make(chan func(), runtime.NumCPU()), + extractionStop: make(chan struct{}), WebSocketRouter: &WebSocketRouter{ handlers: make(map[string]webSocketHandler), }, @@ -401,6 +410,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { ps.searchConfigListenerId = searchConfigListenerId ps.searchLicenseListenerId = searchLicenseListenerId + ps.startExtractionWorkers() + return ps, nil } @@ -517,6 +528,10 @@ func (ps *PlatformService) Shutdown() error { ps.RemoveLicenseListener(ps.licenseListenerId) + // Stop the document extraction workers and wait for any in-flight + // extraction to finish before closing the store it depends on. + ps.stopExtractionWorkers() + // we need to wait the goroutines to finish before closing the store // and this needs to be called after hub stop because hub generates goroutines // when it is active. If we wait first we have no mechanism to prevent adding diff --git a/server/channels/app/server.go b/server/channels/app/server.go index c65c8f0133..48a08eaf2b 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -813,6 +813,14 @@ func (s *Server) GoBuffered(f func()) { s.platform.GoBuffered(f) } +// 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). +func (s *Server) GoExtraction(f func()) bool { + return s.platform.GoExtraction(f) +} + var corsAllowedMethods = []string{ "POST", "GET", diff --git a/server/channels/app/upload.go b/server/channels/app/upload.go index cc16ede7d7..34c7072124 100644 --- a/server/channels/app/upload.go +++ b/server/channels/app/upload.go @@ -339,12 +339,14 @@ func (a *App) UploadData(c request.CTX, us *model.UploadSession, rd io.Reader) ( if *a.Config().FileSettings.ExtractContent { infoCopy := *info - a.Srv().Go(func() { + if !a.Srv().GoExtraction(func() { err := a.ExtractContentFromFileInfo(c, &infoCopy) if err != nil { c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id)) } - }) + }) { + c.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)) + } } // delete upload session diff --git a/server/i18n/en.json b/server/i18n/en.json index a8e4dbe565..ac8765eed2 100644 --- a/server/i18n/en.json +++ b/server/i18n/en.json @@ -9452,6 +9452,10 @@ "id": "model.config.is_valid.export.retention_days_too_low.app_error", "translation": "Invalid value for RetentionDays. Value should be greater than 0" }, + { + "id": "model.config.is_valid.extract_content_timeout.app_error", + "translation": "Invalid content extraction timeout for file settings. Must be a whole number of seconds greater than or equal to zero." + }, { "id": "model.config.is_valid.file_driver.app_error", "translation": "Invalid driver name for file settings. Must be 'local' or 'amazons3'." diff --git a/server/platform/services/docextractor/docextractor.go b/server/platform/services/docextractor/docextractor.go index a4de967eb3..e3e681b6ce 100644 --- a/server/platform/services/docextractor/docextractor.go +++ b/server/platform/services/docextractor/docextractor.go @@ -4,7 +4,9 @@ package docextractor import ( + "fmt" "io" + "time" "github.com/mattermost/mattermost/server/public/shared/mlog" ) @@ -15,6 +17,23 @@ type ExtractSettings struct { MaxFileSize int64 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. + // 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. + Timeout time.Duration + // ReaderCloser, when set, transfers ownership of closing the input reader + // to this package. It is closed only after extraction has actually + // finished reading. This matters with Timeout set: on timeout the caller + // returns while the converter may still be reading on a detached + // goroutine, so the caller must NOT close the reader itself or it would + // race with (and close the file out from under) that goroutine. + ReaderCloser io.Closer } // Extract extract the text from a document using the system default extractors @@ -45,7 +64,71 @@ func ExtractWithExtraExtractors(logger mlog.LoggerIFace, filename string, r io.R enabledExtractors.Add(&plainExtractor{}) if enabledExtractors.Match(filename) { - return enabledExtractors.Extract(filename, r, settings.MaxFileSize) + return extractWithTimeout(enabledExtractors, filename, r, settings) + } + + // No extractor matched, so nothing will read r; close it here since + // extractWithTimeout (which otherwise owns the close) is never reached. + if settings.ReaderCloser != nil { + settings.ReaderCloser.Close() } return "", nil } + +// extractWithTimeout runs the extraction and stops waiting for it once +// 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. +func extractWithTimeout(e Extractor, filename string, r io.ReadSeeker, settings ExtractSettings) (string, error) { + if settings.Timeout <= 0 { + if settings.ReaderCloser != nil { + defer settings.ReaderCloser.Close() + } + return e.Extract(filename, r, settings.MaxFileSize) + } + + type extractResult struct { + text string + err error + } + resultCh := make(chan extractResult, 1) + go func() { + // 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 + // goroutine is done with it - never by the caller. + if settings.ReaderCloser != nil { + defer settings.ReaderCloser.Close() + } + // This goroutine is detached, so an unrecovered panic in an extractor + // would crash the whole server. Convert it into an error instead. + // resultCh is buffered (cap 1), so this send never blocks even if the + // caller already timed out and stopped receiving. + defer func() { + if rec := recover(); rec != nil { + resultCh <- extractResult{err: fmt.Errorf("panic during document text extraction: %v", rec)} + } + }() + text, err := e.Extract(filename, r, settings.MaxFileSize) + resultCh <- extractResult{text: text, err: err} + }() + + timer := time.NewTimer(settings.Timeout) + defer timer.Stop() + + select { + case res := <-resultCh: + return res.text, res.err + case <-timer.C: + return "", fmt.Errorf("document text extraction timed out after %s", settings.Timeout) + } +} diff --git a/server/platform/services/docextractor/docextractor_test.go b/server/platform/services/docextractor/docextractor_test.go index fbbe27868d..8805267473 100644 --- a/server/platform/services/docextractor/docextractor_test.go +++ b/server/platform/services/docextractor/docextractor_test.go @@ -8,7 +8,9 @@ import ( "errors" "io" "strings" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -213,6 +215,144 @@ func TestExtractWithExtraExtractors(t *testing.T) { }) } +type slowExtractor struct { + delay time.Duration +} + +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) { + time.Sleep(se.delay) + return "done", nil +} + +func TestExtractTimeout(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + data := []byte("hello world") + + t.Run("aborts a slow extraction once the timeout elapses", func(t *testing.T) { + start := time.Now() + text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: 50 * time.Millisecond}, []Extractor{&slowExtractor{delay: 10 * time.Second}}) + 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") + }) + + t.Run("returns the result when extraction finishes within the timeout", func(t *testing.T) { + text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: 5 * time.Second}, []Extractor{&slowExtractor{delay: 10 * time.Millisecond}}) + require.NoError(t, err) + require.Equal(t, "done", text) + }) + + t.Run("a zero timeout disables the bound", func(t *testing.T) { + text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: 0}, []Extractor{&slowExtractor{delay: 10 * time.Millisecond}}) + require.NoError(t, err) + require.Equal(t, "done", text) + }) + + t.Run("a panic in the detached extraction is converted to an error", func(t *testing.T) { + text, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader(data), ExtractSettings{Timeout: time.Second}, []Extractor{&panickingExtractor{}}) + require.Error(t, err) + require.Empty(t, text) + require.Contains(t, err.Error(), "panic") + }) +} + +type panickingExtractor struct{} + +func (pe *panickingExtractor) Name() string { return "panickingExtractor" } + +func (pe *panickingExtractor) Match(filename string) bool { return true } + +func (pe *panickingExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (string, error) { + panic("boom") +} + +type recordingCloser struct { + closed atomic.Bool +} + +func (c *recordingCloser) Close() error { + c.closed.Store(true) + return nil +} + +// blockingExtractor blocks inside Extract until release is closed, simulating a +// converter that is still using the reader after an extraction timeout fires. +type blockingExtractor struct { + started chan struct{} + release chan struct{} +} + +func (be *blockingExtractor) Name() string { return "blockingExtractor" } + +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 + return "done", nil +} + +func TestExtractReaderCloserOwnership(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + + t.Run("reader is closed only after the detached extraction finishes on timeout", func(t *testing.T) { + closer := &recordingCloser{} + be := &blockingExtractor{started: make(chan struct{}), release: make(chan struct{})} + settings := ExtractSettings{Timeout: 50 * time.Millisecond, ReaderCloser: closer} + + _, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader([]byte("hi")), settings, []Extractor{be}) + require.Error(t, err) + require.Contains(t, err.Error(), "timed out") + + // Wait (with a deadline) for the detached extraction to start so the + // test fails fast instead of hanging if it never runs. + select { + case <-be.started: + case <-time.After(2 * time.Second): + require.FailNow(t, "extraction did not start within the deadline") + } + // The extraction goroutine is still running, so closing the reader now + // would race with it; it must stay open. + require.False(t, closer.closed.Load(), "reader must not be closed while the extraction goroutine is still running") + + close(be.release) + require.Eventually(t, closer.closed.Load, 2*time.Second, 5*time.Millisecond, "reader should be closed once the extraction goroutine finishes") + }) + + t.Run("reader is closed on the synchronous path", func(t *testing.T) { + closer := &recordingCloser{} + _, err := ExtractWithExtraExtractors(logger, "file.txt", bytes.NewReader([]byte("hi")), ExtractSettings{ReaderCloser: closer}, []Extractor{&slowExtractor{delay: 0}}) + require.NoError(t, err) + require.True(t, closer.closed.Load(), "reader should be closed after synchronous extraction") + }) +} + +func TestDocumentMaxFileSize(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + + data, err := testutils.ReadTestFile("sample-doc.docx") + require.NoError(t, err) + + t.Run("a generous limit extracts the document content", func(t *testing.T) { + text, err := Extract(logger, "sample-doc.docx", bytes.NewReader(data), ExtractSettings{MaxFileSize: 10 * 1024 * 1024}) + require.NoError(t, err) + assert.Contains(t, text, "simple") + }) + + t.Run("a tiny limit prevents the document content from being extracted", func(t *testing.T) { + text, err := Extract(logger, "sample-doc.docx", bytes.NewReader(data), ExtractSettings{MaxFileSize: 16}) + require.NoError(t, err) + assert.NotContains(t, text, "simple") + }) +} + func TestArchiveMaxFileSize(t *testing.T) { t.Parallel() diff --git a/server/platform/services/docextractor/documents.go b/server/platform/services/docextractor/documents.go index a0c11d9606..7d2ad90fa2 100644 --- a/server/platform/services/docextractor/documents.go +++ b/server/platform/services/docextractor/documents.go @@ -10,6 +10,8 @@ import ( "strings" "code.sajari.com/docconv/v2" + + "github.com/mattermost/mattermost/server/v8/channels/utils" ) type documentExtractor struct{} @@ -36,7 +38,7 @@ func (de *documentExtractor) Match(filename string) bool { return ok } -func (de *documentExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (out string, outErr error) { +func (de *documentExtractor) Extract(filename string, r io.ReadSeeker, maxFileSize int64) (out string, outErr error) { defer func() { if r := recover(); r != nil { out = "" @@ -50,7 +52,14 @@ func (de *documentExtractor) Extract(filename string, r io.ReadSeeker, _ int64) return "", errors.New("unknown converter") } - text, _, err := converter(r) + // Bound how much data the converter is allowed to read so a small upload + // cannot expand into an unbounded amount of in-memory work. + var reader io.Reader = r + if maxFileSize > 0 { + reader = utils.NewLimitedReaderWithError(r, maxFileSize) + } + + text, _, err := converter(reader) if err != nil { return "", err } diff --git a/server/platform/services/docextractor/pdf.go b/server/platform/services/docextractor/pdf.go index 2c245c3d66..67deb34bc9 100644 --- a/server/platform/services/docextractor/pdf.go +++ b/server/platform/services/docextractor/pdf.go @@ -13,6 +13,8 @@ import ( "strings" "github.com/ledongthuc/pdf" + + "github.com/mattermost/mattermost/server/v8/channels/utils" ) type pdfExtractor struct{} @@ -29,7 +31,7 @@ func (pe *pdfExtractor) Match(filename string) bool { return supportedExtensions[extension] } -func (pe *pdfExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (out string, outErr error) { +func (pe *pdfExtractor) Extract(filename string, r io.ReadSeeker, maxFileSize int64) (out string, outErr error) { defer func() { if r := recover(); r != nil { out = "" @@ -42,7 +44,14 @@ func (pe *pdfExtractor) Extract(filename string, r io.ReadSeeker, _ int64) (out } defer f.Close() defer os.Remove(f.Name()) - size, err := io.Copy(f, r) + + // Bound how much data is copied to disk so a small upload cannot expand + // into an unbounded amount of temporary storage. + var src io.Reader = r + if maxFileSize > 0 { + src = utils.NewLimitedReaderWithError(r, maxFileSize) + } + size, err := io.Copy(f, src) if err != nil { return "", fmt.Errorf("error copying data into temporary file: %v", err) } diff --git a/server/platform/services/docextractor/pdf_test.go b/server/platform/services/docextractor/pdf_test.go index 722cbf6489..5f5ca7e8c7 100644 --- a/server/platform/services/docextractor/pdf_test.go +++ b/server/platform/services/docextractor/pdf_test.go @@ -50,3 +50,30 @@ func TestWrongPdfFile(t *testing.T) { _, err = extractor.Extract("sample-doc.pdf", bytes.NewReader(content), 0) require.Error(t, err) } + +func TestPdfMaxFileSize(t *testing.T) { + extractor := pdfExtractor{} + content, err := testutils.ReadTestFile("sample-doc.pdf") + require.NoError(t, err) + require.Greater(t, len(content), 16, "fixture must be larger than the tight limit under test") + + t.Run("a zero limit means unlimited and extracts the content", func(t *testing.T) { + text, err := extractor.Extract("sample-doc.pdf", bytes.NewReader(content), 0) + require.NoError(t, err) + require.Contains(t, text, "simple") + }) + + t.Run("a generous limit extracts the content", func(t *testing.T) { + text, err := extractor.Extract("sample-doc.pdf", bytes.NewReader(content), 10*1024*1024) + require.NoError(t, err) + require.Contains(t, text, "simple") + }) + + t.Run("a tight limit prevents extraction", func(t *testing.T) { + // The reader errors once it reads past the limit, so io.Copy to the + // temp file fails and no text is extracted. + text, err := extractor.Extract("sample-doc.pdf", bytes.NewReader(content), 16) + require.Error(t, err) + require.Empty(t, text) + }) +} diff --git a/server/public/model/config.go b/server/public/model/config.go index 53360b1df7..2a465f4449 100644 --- a/server/public/model/config.go +++ b/server/public/model/config.go @@ -1767,6 +1767,7 @@ type FileSettings struct { Directory *string `access:"environment_file_storage,write_restrictable,cloud_restrictable"` EnablePublicLink *bool `access:"site_public_links,cloud_restrictable"` ExtractContent *bool `access:"environment_file_storage,write_restrictable"` + ExtractContentTimeout *int `access:"environment_file_storage,write_restrictable"` // In seconds. 0 disables the timeout. ArchiveRecursion *bool `access:"environment_file_storage,write_restrictable"` PublicLinkSalt *string `access:"site_public_links,cloud_restrictable"` // telemetry: none InitialFont *string `access:"environment_file_storage,cloud_restrictable"` // telemetry: none @@ -1844,6 +1845,10 @@ func (s *FileSettings) SetDefaults(isUpdate bool) { s.ExtractContent = NewPointer(true) } + if s.ExtractContentTimeout == nil { + s.ExtractContentTimeout = NewPointer(10) + } + if s.ArchiveRecursion == nil { s.ArchiveRecursion = NewPointer(false) } @@ -4230,6 +4235,10 @@ func (s *FileSettings) isValid() *AppError { return NewAppError("Config.IsValid", "model.config.is_valid.max_file_size.app_error", nil, "", http.StatusBadRequest) } + if *s.ExtractContentTimeout < 0 { + return NewAppError("Config.IsValid", "model.config.is_valid.extract_content_timeout.app_error", nil, "", http.StatusBadRequest) + } + if !(*s.DriverName == ImageDriverLocal || *s.DriverName == ImageDriverS3) { return NewAppError("Config.IsValid", "model.config.is_valid.file_driver.app_error", nil, "", http.StatusBadRequest) } diff --git a/server/public/model/config_test.go b/server/public/model/config_test.go index e3f6a29d87..ada5ddd895 100644 --- a/server/public/model/config_test.go +++ b/server/public/model/config_test.go @@ -296,6 +296,38 @@ func TestFileSettingsDirectoryWhitespaceValidation(t *testing.T) { } } +func TestFileSettingsExtractContentTimeout(t *testing.T) { + t.Run("default is valid", func(t *testing.T) { + cfg := &Config{} + cfg.SetDefaults() + require.NotNil(t, cfg.FileSettings.ExtractContentTimeout) + assert.Equal(t, 10, *cfg.FileSettings.ExtractContentTimeout) + assert.Nil(t, cfg.FileSettings.isValid()) + }) + + t.Run("zero disables the timeout and is valid", func(t *testing.T) { + cfg := &Config{} + cfg.SetDefaults() + cfg.FileSettings.ExtractContentTimeout = NewPointer(0) + assert.Nil(t, cfg.FileSettings.isValid()) + }) + + t.Run("a positive value is valid", func(t *testing.T) { + cfg := &Config{} + cfg.SetDefaults() + cfg.FileSettings.ExtractContentTimeout = NewPointer(10) + assert.Nil(t, cfg.FileSettings.isValid()) + }) + + t.Run("a negative value is rejected", func(t *testing.T) { + cfg := &Config{} + cfg.SetDefaults() + cfg.FileSettings.ExtractContentTimeout = NewPointer(-1) + err := cfg.FileSettings.isValid() + require.NotNil(t, err) + assert.Equal(t, "model.config.is_valid.extract_content_timeout.app_error", err.Id) + }) +} func TestConfigDefaultSignatureAlgorithm(t *testing.T) { c1 := Config{} c1.SetDefaults() diff --git a/webapp/channels/src/components/admin_console/admin_definition.tsx b/webapp/channels/src/components/admin_console/admin_definition.tsx index 7b5b20144b..a30f111d36 100644 --- a/webapp/channels/src/components/admin_console/admin_definition.tsx +++ b/webapp/channels/src/components/admin_console/admin_definition.tsx @@ -1081,6 +1081,18 @@ const AdminDefinition: AdminDefinitionType = { it.not(it.userHasWritePermissionOnResource(RESOURCE_KEYS.ENVIRONMENT.FILE_STORAGE)), ), }, + { + type: 'number', + key: 'FileSettings.ExtractContentTimeout', + label: defineMessage({id: 'admin.image.extractContentTimeoutTitle', defaultMessage: 'Document content extraction timeout (seconds):'}), + help_text: defineMessage({id: 'admin.image.extractContentTimeoutDescription', defaultMessage: 'Maximum number of seconds spent extracting the searchable content of a single uploaded document. Extractions that exceed this limit are aborted to protect server performance. Set to 0 to disable the timeout.'}), + placeholder: defineMessage({id: 'admin.image.extractContentTimeoutExample', defaultMessage: '10'}), + validate: validators.minValue(0, defineMessage({id: 'admin.image.extractContentTimeout.minValue', defaultMessage: 'Timeout must be 0 or greater.'})), + isDisabled: it.any( + it.not(it.userHasWritePermissionOnResource(RESOURCE_KEYS.ENVIRONMENT.FILE_STORAGE)), + it.configIsFalse('FileSettings', 'ExtractContent'), + ), + }, { type: 'bool', key: 'FileSettings.ArchiveRecursion', diff --git a/webapp/channels/src/i18n/en.json b/webapp/channels/src/i18n/en.json index ffad609401..993c6d1ca5 100644 --- a/webapp/channels/src/i18n/en.json +++ b/webapp/channels/src/i18n/en.json @@ -1349,6 +1349,10 @@ "admin.image.enableProxyDescription": "When true, enables an image proxy for loading all Markdown images.", "admin.image.exportDirectoryDescription": "Directory to which files are written. If blank, defaults to ./data/.", "admin.image.extractContentDescription": "When enabled, supported document types are searchable by their content. Search results for existing documents may be incomplete until a data migration is executed.", + "admin.image.extractContentTimeout.minValue": "Timeout must be 0 or greater.", + "admin.image.extractContentTimeoutDescription": "Maximum number of seconds spent extracting the searchable content of a single uploaded document. Extractions that exceed this limit are aborted to protect server performance. Set to 0 to disable the timeout.", + "admin.image.extractContentTimeoutExample": "10", + "admin.image.extractContentTimeoutTitle": "Document content extraction timeout (seconds):", "admin.image.extractContentTitle": "Enable document search by content:", "admin.image.localDescription": "Directory to which files and images are written. If blank, defaults to ./data/.", "admin.image.localExample": "E.g.: \"./data/\"", diff --git a/webapp/platform/types/src/config.ts b/webapp/platform/types/src/config.ts index 02bcc5a30f..5f9687312e 100644 --- a/webapp/platform/types/src/config.ts +++ b/webapp/platform/types/src/config.ts @@ -548,6 +548,7 @@ export type FileSettings = { Directory: string; EnablePublicLink: boolean; ExtractContent: boolean; + ExtractContentTimeout: number; ArchiveRecursion: boolean; PublicLinkSalt: string; InitialFont: string;