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

* Bound document content extraction to prevent upload DoS (MM-69098) Extraction of uploaded Office documents ran with no time or size bound and acquired a shared worker-pool slot on the request goroutine, so a low-privilege user could keep cheap-to-upload but expensive-to-extract documents in flight and degrade file uploads for every user on the server. - Enforce a configurable per-extraction timeout (FileSettings.ExtractContentTimeout, default 10s) and honor the previously-ignored size argument in the document and PDF extractors to bound the work performed. - Run extraction on a dedicated, bounded worker pool with non-blocking submit (GoExtraction) so saturating it can never block the request goroutines that dispatch uploads; overflow is skipped and backfilled by the ExtractContent job. - Route the file and resumable-upload extraction dispatch sites through the new pool instead of the shared GoBuffered/Go pools. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Use require.Fail instead of t.Fatal in extraction pool test The mattermost-govet tFatal analyzer (run by check-style) forbids t.Fatal in tests in favor of testify assertions. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * ci: retrigger workflows (transient container init failure) Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Use sync.WaitGroup.Go for extraction workers Satisfies the golangci-lint waitgroupgo/modernize analyzer flagged by check-style at goroutines.go:58. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Fix file close race with detached extraction goroutine On timeout, extractWithTimeout returns while the converter may still be reading the input on a detached goroutine. ExtractContentFromFileInfo previously closed the file via defer as soon as Extract returned, racing with (and closing the file out from under) that goroutine. Transfer close ownership to docextractor via ExtractSettings.ReaderCloser: the reader is now closed only after extraction actually finishes - in the detached goroutine on the timeout path, or after Extract returns on the synchronous path. ExtractContentFromFileInfo no longer closes the file itself. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Document that extraction timeout bounds wait time, not CPU Clarify in ExtractSettings.Timeout and extractWithTimeout that on timeout the docconv converter keeps running on a detached goroutine (docconv is not context-aware), so the timeout bounds how long an extraction occupies a worker slot, not the CPU it consumes. MaxFileSize is the primary bound on per-extraction work. Note load-shedding as a possible future improvement. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Add tests for PDF size cap and ExtractContentTimeout validation - TestPdfMaxFileSize exercises the LimitedReaderWithError branch in pdf.go: a tight MaxFileSize errors out before extraction, while zero and generous limits extract normally. - TestFileSettingsExtractContentTimeout asserts ExtractContentTimeout=-1 fails IsValid while 0 (disabled) and 10 (default) pass. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Harden extraction shutdown, panic safety, and timeout input validation - stopExtractionWorkers now drains the queue after signaling stop so a worker cannot dequeue and run buffered tasks during shutdown. - The detached extraction goroutine recovers panics and converts them to errors, so a panic in an extractor cannot crash the server. - Admin console enforces a client-side minimum of 0 for ExtractContentTimeout (validators.minValue). - Test hardening: bounded wait for the detached extractor to start, plus a test that a panic is surfaced as an error. Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * ci: retrigger workflows (transient Docker Hub image pull timeout) Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> * Correct misleading extraction-skip logs and comments The queue-full warning logs and GoExtraction doc comments claimed skipped content would be backfilled by a 'periodic ExtractContent job'. That job is registered with a nil scheduler, so it never runs automatically -- it only runs when an admin manually triggers a content extraction job (mmctl extract). Reword the logs and comments to state that skipped files stay unsearchable until an admin runs a content extraction job, so we don't imply automatic recovery that doesn't exist. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Julien Tant <JulienTant@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-10-28 02:36:08 +08:00 · 2026-06-10 23:12:51 +00:00
parent 3535473d96
commit 471fd8d1dd
19 changed files with 514 additions and 13 deletions
@@ -920,12 +920,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
@@ -1187,12 +1189,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
@@ -1757,10 +1761,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")
@@ -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
+	}
+}
@@ -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")
+	})
+}
@@ -108,6 +108,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
@@ -156,6 +163,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 		startTime:           time.Now(),
 		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),
 		},
@@ -442,6 +451,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 	ps.searchConfigListenerId = searchConfigListenerId
 	ps.searchLicenseListenerId = searchLicenseListenerId

+	ps.startExtractionWorkers()
+
 	return ps, nil
 }

@@ -570,6 +581,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
@@ -884,6 +884,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",
@@ -349,12 +349,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
@@ -11642,6 +11642,10 @@
    "id": "model.config.is_valid.export_azure_timeout.app_error",
    "translation": "Invalid timeout value {{.Value}}. Should be a positive number."
  },
+  {
+    "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', 'amazons3', or 'azureblob'."
@@ -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)
+	}
+}
@@ -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()

@@ -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
 	}
@@ -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)
 	}
@@ -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)
+	})
+}
@@ -1811,6 +1811,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
@@ -1907,6 +1908,10 @@ func (s *FileSettings) SetDefaults(isUpdate bool) {
 		s.ExtractContent = new(true)
 	}

+	if s.ExtractContentTimeout == nil {
+		s.ExtractContentTimeout = new(10)
+	}
+
 	if s.ArchiveRecursion == nil {
 		s.ArchiveRecursion = new(false)
 	}
@@ -4594,6 +4599,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 || *s.DriverName == ImageDriverAzure) {
 		return NewAppError("Config.IsValid", "model.config.is_valid.file_driver.app_error", nil, "", http.StatusBadRequest)
 	}
@@ -346,6 +346,39 @@ func TestFileSettingsAzureRequestTimeoutBounds(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 TestFileSettingsAzureAuthMode(t *testing.T) {
 	t.Run("defaults to shared_key", func(t *testing.T) {
 		cfg := &Config{}