Bound document content extraction time and decouple it from uploads (MM-69098) (#36856) (#37041)

Automatic Merge

e2e-tests/cypress/tests/support/api/on_prem_default_config.json

@@ -197,6 +197,7 @@
         "Directory": "./data/",
         "EnablePublicLink": false,
         "ExtractContent": true,
+        "ExtractContentTimeout": 10,
         "ArchiveRecursion": false,
         "PublicLinkSalt": "",
         "InitialFont": "nunito-bold.ttf",

e2e-tests/playwright/lib/src/server/default_config.ts

@@ -305,6 +305,7 @@ const defaultServerConfig: AdminConfig = {
         Directory: './data/',
         EnablePublicLink: false,
         ExtractContent: true,
+        ExtractContentTimeout: 10,
         ArchiveRecursion: false,
         PublicLinkSalt: '',
         InitialFont: 'nunito-bold.ttf',

server/channels/app/file.go

@@ -858,12 +858,14 @@ func (a *App) UploadFileX(rctx request.CTX, channelID, name string, input io.Rea
 
 	if *a.Config().FileSettings.ExtractContent && t.ExtractContent {
 		infoCopy := *t.fileinfo
-		a.Srv().GoBuffered(func() {
+		if !a.Srv().GoExtraction(func() {
 			err := a.ExtractContentFromFileInfo(rctx, &infoCopy)
 			if err != nil {
 				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))
+		}
 	}
 
 	return t.fileinfo, nil
@@ -1125,12 +1127,14 @@ func (a *App) DoUploadFileExpectModification(rctx request.CTX, now time.Time, ra
 	// 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(rctx, &infoCopy)
 			if err != nil {
 				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))
+		}
 	}
 
 	return info, data, nil
@@ -1631,10 +1635,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")

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
+	}
+}

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")
+	})
+}

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
@@ -152,6 +159,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),
 		},
@@ -438,6 +447,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 	ps.searchConfigListenerId = searchConfigListenerId
 	ps.searchLicenseListenerId = searchLicenseListenerId
 
+	ps.startExtractionWorkers()
+
 	return ps, nil
 }
 
@@ -554,6 +565,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

server/channels/app/server.go

@@ -827,6 +827,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",

server/channels/app/upload.go

@@ -348,12 +348,14 @@ func (a *App) UploadData(rctx 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(rctx, &infoCopy)
 			if err != nil {
 				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))
+		}
 	}
 
 	// delete upload session

server/i18n/en.json

@@ -10856,6 +10856,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'."

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)
+	}
+}

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()
 

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
 	}

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)
 	}

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)
+	})
+}

server/public/model/config.go

@@ -1784,6 +1784,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
@@ -1861,6 +1862,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)
 	}
@@ -4396,6 +4401,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)
 	}

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()

webapp/channels/src/components/admin_console/admin_definition.tsx

@@ -1163,6 +1163,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',

webapp/channels/src/i18n/en.json

@@ -1367,6 +1367,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 <link>until a data migration is executed</link>.",
+  "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/\"",

webapp/platform/types/src/config.ts

@@ -559,6 +559,7 @@ export type FileSettings = {
     Directory: string;
     EnablePublicLink: boolean;
     ExtractContent: boolean;
+    ExtractContentTimeout: number;
     ArchiveRecursion: boolean;
     PublicLinkSalt: string;
     InitialFont: string;