diff --git a/server/enterprise/message_export/global_relay_export/attachment_read_test.go b/server/enterprise/message_export/global_relay_export/attachment_read_test.go new file mode 100644 index 0000000000..413001c2f7 --- /dev/null +++ b/server/enterprise/message_export/global_relay_export/attachment_read_test.go @@ -0,0 +1,317 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.enterprise for license information. + +package global_relay_export + +import ( + "bytes" + "context" + "encoding/base64" + "errors" + "fmt" + "io" + "strings" + "testing" + "testing/synctest" + + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/utils/fileutils" + "github.com/mattermost/mattermost/server/v8/enterprise/message_export/shared" + "github.com/mattermost/mattermost/server/v8/platform/shared/filestore" + "github.com/mattermost/mattermost/server/v8/platform/shared/templates" +) + +// --- fault-injection helpers ------------------------------------------------ + +// scriptedReader serves data[pos:], honouring Seek so a resumed read (re-open + Seek past +// the bytes already streamed) continues from the right offset, exactly like the S3/local +// backends. When failAfter >= 0 it returns an error once pos reaches that offset, modelling +// a transient read failure mid-stream (e.g. an S3 timeout). failAfter == 0 fails immediately +// with no progress; failAfter < 0 reads cleanly to EOF. +type scriptedReader struct { + data []byte + pos int + failAfter int +} + +var _ filestore.ReadCloseSeeker = (*scriptedReader)(nil) + +func (r *scriptedReader) Read(p []byte) (int, error) { + if r.failAfter >= 0 && r.pos >= r.failAfter { + return 0, errors.New("simulated transient S3 read failure mid-stream") + } + if r.pos >= len(r.data) { + return 0, io.EOF + } + end := len(r.data) + if r.failAfter >= 0 && r.failAfter < end { + end = r.failAfter + } + n := copy(p, r.data[r.pos:end]) + r.pos += n + return n, nil +} + +func (r *scriptedReader) Close() error { return nil } + +func (r *scriptedReader) Seek(offset int64, whence int) (int64, error) { + if whence != io.SeekStart { + return 0, fmt.Errorf("scriptedReader: unsupported whence %d", whence) + } + r.pos = int(offset) + return offset, nil +} + +// scriptedBackend is a filestore.FileBackend whose Reader is driven by a per-path factory. +// generateEmail exercises Reader plus FileExists (consulted on a read failure to tell a +// genuinely-missing object apart from a transient read error); the rest of the embedded +// interface is nil and unused. +type scriptedBackend struct { + filestore.FileBackend + readers map[string]func() (filestore.ReadCloseSeeker, error) + // missing lists paths that FileExists should report as absent, modelling an object that + // opens lazily (S3/MinIO) but no longer exists. Paths not listed report as present. + missing map[string]bool +} + +func (b *scriptedBackend) Reader(path string) (filestore.ReadCloseSeeker, error) { + if fn, ok := b.readers[path]; ok { + return fn() + } + return nil, fmt.Errorf("scriptedBackend: no reader registered for %q", path) +} + +func (b *scriptedBackend) FileExists(path string) (bool, error) { + return !b.missing[path], nil +} + +// scriptedReaderFactory returns a Reader factory that fails at the given offsets on +// successive opens: the i-th open uses failAt[i] (see scriptedReader.failAfter). Opens past +// the end of failAt read cleanly to EOF. The same data is served every open, so a resumed +// read reconstructs the full content. +func scriptedReaderFactory(data []byte, failAt ...int) func() (filestore.ReadCloseSeeker, error) { + var call int + return func() (filestore.ReadCloseSeeker, error) { + failAfter := -1 + if call < len(failAt) { + failAfter = failAt[call] + } + call++ + return &scriptedReader{data: data, failAfter: failAfter}, nil + } +} + +// healthyReader serves the given content cleanly in a single uninterrupted read. +func healthyReader(content []byte) func() (filestore.ReadCloseSeeker, error) { + return scriptedReaderFactory(content) +} + +// openFails models a file that cannot be opened at all (e.g. deleted from the store). +func openFails() (filestore.ReadCloseSeeker, error) { + return nil, errors.New("file does not exist") +} + +// assertAttachmentPresent checks that content was written into the email as an attachment. +// gomail stores attachments base64-encoded, and MIME wraps that base64 across lines, so we +// drop the line breaks and look for the unwrapped encoding — letting content be any length. +func assertAttachmentPresent(t *testing.T, out *bytes.Buffer, content []byte, msg string) { + t.Helper() + unwrapped := strings.NewReplacer("\r", "", "\n", "").Replace(out.String()) + require.Contains(t, unwrapped, base64.StdEncoding.EncodeToString(content), msg) +} + +func newChannelExport(files ...*model.FileInfo) *ChannelExport { + return &ChannelExport{ + ChannelId: "channelid1234567890123456", + ChannelName: "test-channel", + ChannelDisplayName: "Test Channel", + ChannelType: model.ChannelTypeDirect, + Participants: []ParticipantRow{ + {JoinExport: shared.JoinExport{UserId: "userid", UserEmail: "participant@example.com"}}, + }, + uploadedFiles: files, + } +} + +// runGenerateEmail calls generateEmail inside a testing/synctest bubble so the exponential +// backoff between read retries uses a fake clock and the retries don't actually sleep. Any +// panic is recovered so a regression (e.g. the MM-69242 nil-pointer that crashes the whole +// server) shows up as a clear test failure rather than crashing the test binary. +func runGenerateEmail(t *testing.T, backend filestore.FileBackend, ce *ChannelExport) (warnings int, out *bytes.Buffer, genErr error) { + t.Helper() + return runGenerateEmailCtx(context.Background(), t, backend, ce) +} + +// runGenerateEmailCtx is runGenerateEmail with a caller-supplied context, used to exercise +// job cancellation mid-retry. +func runGenerateEmailCtx(ctx context.Context, t *testing.T, backend filestore.FileBackend, ce *ChannelExport) (warnings int, out *bytes.Buffer, genErr error) { + t.Helper() + templatesDir, ok := fileutils.FindDir("templates") + require.True(t, ok, "could not locate the server templates dir") + templatesContainer, err := templates.New(templatesDir) + require.NoError(t, err) + require.NotNil(t, templatesContainer) + + // The logger (and its async logr goroutine) is created OUTSIDE the bubble so it + // isn't tracked as a bubble goroutine. + rctx := request.TestContext(t).WithContext(ctx) + out = &bytes.Buffer{} + + synctest.Test(t, func(t *testing.T) { + var recovered any + func() { + defer func() { recovered = recover() }() + warnings, genErr = generateEmail(rctx, backend, ce, templatesContainer, out) + }() + if recovered != nil { + t.Fatalf("MM-69242 regression: generateEmail panicked (in production this crashes "+ + "the entire server): %v", recovered) + } + }) + + return warnings, out, genErr +} + +// --- tests ------------------------------------------------------------------ + +// A transient read failure that clears within the retry budget must not fail the export: +// the attachment is resumed on a later attempt and its full content lands in the email. +func TestGenerateEmail_TransientReadRecoversOnRetry(t *testing.T) { + flakyContent := []byte("flaky attachment content that must survive a mid-stream failure and resume") + healthyContent := []byte("healthy attachment content") + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "flaky.bin", Path: "data/flaky.bin"}, + &model.FileInfo{Id: "file2", Name: "healthy.bin", Path: "data/healthy.bin"}, + ) + backend := &scriptedBackend{readers: map[string]func() (filestore.ReadCloseSeeker, error){ + // First open fails after 20 bytes; the retry re-opens, seeks to 20, and finishes — + // proving the resume reconstructs the full content, not just non-empty output. + "data/flaky.bin": scriptedReaderFactory(flakyContent, 20), + "data/healthy.bin": healthyReader(healthyContent), + }} + + warnings, out, genErr := runGenerateEmail(t, backend, ce) + + require.NoError(t, genErr, "a transient failure that clears within the retry budget should succeed") + require.Equal(t, 0, warnings) + assertAttachmentPresent(t, out, healthyContent, "the healthy attachment's content should be in the email") + assertAttachmentPresent(t, out, flakyContent, "the resumed attachment's full content should be in the email") +} + +// A read failure that persists past the retry budget fails the batch (so the job retries) +// instead of shipping an incomplete export — and, critically, never panics. The healthy +// attachment after the failing one is exactly what triggered the MM-69242 nil-deref. +func TestGenerateEmail_PersistentReadFailsBatch(t *testing.T) { + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "fails.bin", Path: "data/fails.bin"}, + &model.FileInfo{Id: "file2", Name: "healthy.bin", Path: "data/healthy.bin"}, + ) + backend := &scriptedBackend{readers: map[string]func() (filestore.ReadCloseSeeker, error){ + // Fails immediately (no progress) on every attempt, exhausting the stall budget. + "data/fails.bin": scriptedReaderFactory(make([]byte, 200), 0, 0, 0), + "data/healthy.bin": healthyReader([]byte("healthy attachment content")), + }} + + warnings, _, genErr := runGenerateEmail(t, backend, ce) + + require.Error(t, genErr, "a persistent attachment read failure should fail the batch (so the job retries)") + require.Equal(t, 0, warnings, "a read failure is an error, not a skipped/missing-file warning") +} + +// A genuinely missing attachment (open fails AND FileExists confirms it's gone) keeps the +// prior MM-62493 behavior: warn, increment the warning count, skip it, don't fail the batch. +func TestGenerateEmail_MissingAttachmentSkipped(t *testing.T) { + healthyContent := []byte("healthy attachment content") + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "missing.bin", Path: "data/missing.bin"}, + &model.FileInfo{Id: "file2", Name: "healthy.bin", Path: "data/healthy.bin"}, + ) + backend := &scriptedBackend{ + readers: map[string]func() (filestore.ReadCloseSeeker, error){ + "data/missing.bin": openFails, + "data/healthy.bin": healthyReader(healthyContent), + }, + missing: map[string]bool{"data/missing.bin": true}, // FileExists confirms it's gone + } + + warnings, out, genErr := runGenerateEmail(t, backend, ce) + + require.NoError(t, genErr, "a missing file should be skipped, not fail the batch") + require.Equal(t, 1, warnings, "the missing file should be counted as a warning") + assertAttachmentPresent(t, out, healthyContent, "the surviving attachment should still be exported") +} + +// A transient OPEN failure (backend.Reader errors, but the file still exists) must NOT be +// silently skipped as "missing": it is retried and, if it persists, fails the batch so the +// job retries — otherwise a transient infrastructure hiccup at open time could drop an +// attachment from a compliance export and still report success (MM-69338). +func TestGenerateEmail_TransientOpenFailureFailsBatch(t *testing.T) { + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "openflaky.bin", Path: "data/openflaky.bin"}, + &model.FileInfo{Id: "file2", Name: "healthy.bin", Path: "data/healthy.bin"}, + ) + backend := &scriptedBackend{ + readers: map[string]func() (filestore.ReadCloseSeeker, error){ + "data/openflaky.bin": openFails, // open fails on every attempt... + "data/healthy.bin": healthyReader([]byte("healthy attachment content")), + }, + // ...but FileExists reports the object still present, so it's a transient hiccup, not a + // deletion (no entry in `missing` ⇒ FileExists returns true). + } + + warnings, _, genErr := runGenerateEmail(t, backend, ce) + + require.Error(t, genErr, "a transient open failure on an existing file should fail the batch, not be skipped") + require.Equal(t, 0, warnings, "a transient open failure is an error, not a skipped/missing-file warning") +} + +// On S3/MinIO a deleted object is not detected when the reader is opened (minio-go's +// GetObject is lazy); the "no such key" only surfaces as a read error on the first Read. Such +// a file must be treated as missing — skipped with a warning, batch not failed and not +// retried — exactly like a local-backend open failure (MM-62493). Modeled +// here by a reader that opens but fails its first read, with FileExists reporting it absent. +func TestGenerateEmail_ReadNotFoundSkipped(t *testing.T) { + healthyContent := []byte("healthy attachment content") + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "s3missing.bin", Path: "data/s3missing.bin"}, + &model.FileInfo{Id: "file2", Name: "healthy.bin", Path: "data/healthy.bin"}, + ) + backend := &scriptedBackend{ + readers: map[string]func() (filestore.ReadCloseSeeker, error){ + "data/s3missing.bin": scriptedReaderFactory(make([]byte, 200), 0), // opens, first read fails + "data/healthy.bin": healthyReader(healthyContent), + }, + missing: map[string]bool{"data/s3missing.bin": true}, // FileExists reports it gone + } + + warnings, out, genErr := runGenerateEmail(t, backend, ce) + + require.NoError(t, genErr, "a read-time not-found must be skipped, not fail the batch") + require.Equal(t, 1, warnings, "the missing file should be counted as a warning") + assertAttachmentPresent(t, out, healthyContent, "the surviving attachment should still be exported") +} + +// A job cancelled while a read is backing off must abort promptly with the context error +// (not sleep through the backoff, not panic). +func TestGenerateEmail_ContextCancelledDuringRetry(t *testing.T) { + ce := newChannelExport( + &model.FileInfo{Id: "file1", Name: "fails.bin", Path: "data/fails.bin"}, + ) + backend := &scriptedBackend{readers: map[string]func() (filestore.ReadCloseSeeker, error){ + "data/fails.bin": scriptedReaderFactory(make([]byte, 200), 0, 0, 0), + }} + + // Cancelled before the first backoff: the retry loop hits the first read failure, then + // the cancellation wins the backoff select immediately. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + warnings, _, genErr := runGenerateEmailCtx(ctx, t, backend, ce) + + require.Error(t, genErr, "a cancelled job should fail rather than ship a partial export") + require.ErrorIs(t, genErr, context.Canceled) + require.Equal(t, 0, warnings) +} diff --git a/server/enterprise/message_export/global_relay_export/global_relay_export.go b/server/enterprise/message_export/global_relay_export/global_relay_export.go index 070696245b..014cb8edb2 100644 --- a/server/enterprise/message_export/global_relay_export/global_relay_export.go +++ b/server/enterprise/message_export/global_relay_export/global_relay_export.go @@ -6,6 +6,7 @@ package global_relay_export import ( "archive/zip" "encoding/json" + "errors" "fmt" "io" "mime" @@ -32,6 +33,17 @@ const ( GlobalRelayChannelIDHeader = "X-Mattermost-ChannelID" GlobalRelayChannelTypeHeader = "X-Mattermost-ChannelType" MaxEmailsPerConnection = 400 + + // maxAttachmentReadAttempts bounds how many consecutive read attempts that make no + // forward progress we tolerate before giving up. Attempts that do make progress reset + // this budget (and the backoff), so a large attachment can still complete over a flaky + // connection. If the budget is exhausted, the batch is failed and the job is retried. + maxAttachmentReadAttempts = 3 + + // attachmentReadBackoff is the initial delay before retrying a stalled attachment read; + // it doubles after each stalled attempt (exponential backoff) and resets once a retry + // makes progress. + attachmentReadBackoff = 1 * time.Second ) // MaxEmailBytes is a var because it needs to be set in tests. Otherwise it shouldn't be touched. @@ -264,7 +276,7 @@ func generateEmail(rctx request.CTX, fileAttachmentBackend filestore.FileBackend htmlBody, err := channelExportToHTML(rctx, channelExport, templates) if err != nil { - return warningCount, fmt.Errorf("unable to generate eml file data: %w", err) + return warningCount, fmt.Errorf("unable to render the channel export to HTML: %w", err) } subject := fmt.Sprintf("Mattermost Compliance Export: %s", channelExport.ChannelDisplayName) @@ -295,34 +307,177 @@ func generateEmail(rctx request.CTX, fileAttachmentBackend filestore.FileBackend m.SetBody("text/plain", txtBody) m.AddAlternative("text/html", htmlMessage) + // attachmentReadErr captures a genuine attachment read/write failure that we must + // NOT surface to gomail. gomail v2.3.1 stores any error returned by a copy closure + // and then nil-derefs while writing the *next* attachment, which panics and + // (because workers don't recover) crashes the whole server (MM-69242). So the + // closure always returns nil and we fail the batch here, after WriteTo. + var attachmentReadErr error + for _, fileInfo := range channelExport.uploadedFiles { path := fileInfo.Path m.Attach(fileInfo.Name, gomail.SetCopyFunc(func(writer io.Writer) error { - var reader filestore.ReadCloseSeeker - reader, err = fileAttachmentBackend.Reader(path) - if err != nil { + missing, readErr := streamAttachmentForExport(rctx, fileAttachmentBackend, path, writer) + switch { + case missing: + // The attachment no longer exists in the store (confirmed via FileExists). + // Warn and skip so a single deleted file can't block the export (MM-62493). rctx.Logger().Warn("File not found for export", mlog.String("filename", path)) warningCount += 1 - return nil - } - defer reader.Close() - - _, err = io.Copy(writer, reader) - if err != nil { - return fmt.Errorf("unable to add attachment to the Global Relay export: %w", err) + case readErr != nil: + // A read/write failure that persisted across retries. Record it and fail the + // batch after WriteTo so the job retries instead of shipping an incomplete export. + rctx.Logger().Error("Failed to read attachment for Global Relay export after retries", + mlog.String("filename", path), mlog.Err(readErr)) + attachmentReadErr = errors.Join(attachmentReadErr, fmt.Errorf("attachment %q: %w", path, readErr)) } + // Always return nil: an error here poisons gomail's writer and panics on the + // next attachment (MM-69242). We fail the batch after WriteTo instead. return nil })) } - _, err = m.WriteTo(w) - if err != nil { - return warningCount, fmt.Errorf("unable to generate eml file data: %w", err) + if _, err = m.WriteTo(w); err != nil { + return warningCount, fmt.Errorf("unable to write the eml message: %w", err) + } + if attachmentReadErr != nil { + return warningCount, fmt.Errorf("unable to read one or more attachments for the eml message: %w", attachmentReadErr) } return warningCount, nil } +// errAttachmentStreamFatal wraps a streaming failure that retrying cannot fix — a failure +// writing to the output (gomail) stream, or a failed resume Seek. The caller fails the batch +// rather than retrying or skipping. +var errAttachmentStreamFatal = errors.New("attachment stream cannot be retried") + +// streamAttachmentForExport streams the attachment at path directly into dst (the gomail +// writer), retrying transient failures (e.g. an S3 timeout, whether it surfaces when opening +// the reader or mid-read). Each retry re-opens the backend reader and Seeks past the bytes +// already written, so a retry resumes rather than re-downloads: memory stays constant (an S3 +// Seek is a ranged GET, not a fresh download) instead of buffering a whole, up to +// ~MaxEmailBytes, attachment. Only attempts that make NO forward progress count against the +// retry budget, so a large attachment can still complete over a flaky connection as long as +// each retry advances; a cancelled job aborts promptly via the context. +// +// An open or read failure is classified, not assumed missing: it returns missing=true only +// when FileExists confirms the object is genuinely gone (the caller skips it, preserving +// MM-62493). A failure on a file that still exists — a transient infrastructure hiccup at open +// or read time, indistinguishable from a deletion by error alone — is retried and, if it +// persists, returned as an error so the batch fails rather than silently dropping an +// attachment from a compliance export (MM-69338). +// +// NOTE: a failed stream may have already written a partial attachment to dst. That is safe +// only because the caller fails the whole batch on a non-nil error, so the incomplete output +// is discarded and the job retries; the closure must NOT return this error to gomail (MM-69242). +func streamAttachmentForExport(rctx request.CTX, backend filestore.FileBackend, path string, dst io.Writer) (missing bool, err error) { + var written int64 + backoff := attachmentReadBackoff + for stalled := 0; stalled < maxAttachmentReadAttempts; { + var n int64 + n, err = streamAttachmentOnce(backend, path, dst, written) + written += n + + if err == nil { + return false, nil + } + if errors.Is(err, errAttachmentStreamFatal) { + // Output-stream write failure or a failed resume Seek: retrying can't help, so + // fail the batch. + return false, err + } + + // An open or read failure — both retryable, but first tell a genuinely-missing file + // apart from a transient hiccup. If the object is gone (and we've emitted nothing yet), + // skip it so a single deleted file can't block the export forever (preserves MM-62493). + // Anything else — it still exists, or the existence check itself failed — is treated as + // transient: retried, then failed, so a transient open/read error can't silently drop an + // attachment from a compliance export (MM-69338). On S3/MinIO a deleted object isn't even + // detected on open (minio-go's GetObject is lazy), so this read-time check is what makes + // the skip work there at all. + if written == 0 { + if exists, existsErr := backend.FileExists(path); existsErr == nil && !exists { + return true, nil + } + } + + if n > 0 { + // Made forward progress: the next attempt resumes further along. Reset the + // stall budget and backoff so a flaky connection can still finish a large file. + stalled = 0 + backoff = attachmentReadBackoff + } else { + stalled++ + if stalled >= maxAttachmentReadAttempts { + break + } + } + + // Transient failure: back off (exponentially) before retrying, but bail out promptly + // if the job is being cancelled rather than sleeping through it. + rctx.Logger().Warn("Transient error streaming attachment for Global Relay export; backing off before retry", + mlog.String("filename", path), mlog.Int("bytesRead", written), + mlog.Duration("backoff", backoff), mlog.Err(err)) + + select { + case <-time.After(backoff): + case <-rctx.Context().Done(): + return false, rctx.Context().Err() + } + + backoff *= 2 + } + + return false, err +} + +// streamAttachmentOnce makes a single open+copy attempt, resuming past resumeFrom bytes so a +// retry continues rather than re-downloads. It returns the bytes copied in this attempt. A nil +// error means the attachment streamed fully. An error wrapping errAttachmentStreamFatal is not +// retryable (output-stream write failure or a failed resume Seek); any other error is a +// retryable open/read failure that the caller classifies as missing-vs-transient. +func streamAttachmentOnce(backend filestore.FileBackend, path string, dst io.Writer, resumeFrom int64) (int64, error) { + reader, err := backend.Reader(path) + if err != nil { + // Open failure: retryable. The caller checks FileExists to tell a deleted file + // (skip) from a transient hiccup (retry, then fail the batch). + return 0, err + } + defer reader.Close() + + if resumeFrom > 0 { + // Resume where the previous attempt left off instead of re-reading from the start. + if _, err = reader.Seek(resumeFrom, io.SeekStart); err != nil { + return 0, fmt.Errorf("%w: seeking to resume offset %d: %w", errAttachmentStreamFatal, resumeFrom, err) + } + } + + rd := &readErrorReader{Reader: reader} + n, err := io.Copy(dst, rd) + if err != nil && rd.readErr == nil { + // io.Copy failed writing to the output stream, not reading the attachment. + return n, fmt.Errorf("%w: %w", errAttachmentStreamFatal, err) + } + return n, err +} + +// readErrorReader wraps a reader and remembers the last non-EOF read error. It lets the +// caller of io.Copy tell a failed attachment read (retryable) apart from a failed write to +// the output stream (not retryable), which io.Copy collapses into a single error. +type readErrorReader struct { + io.Reader + readErr error +} + +func (r *readErrorReader) Read(p []byte) (int, error) { + n, err := r.Reader.Read(p) + if err != nil && err != io.EOF { + r.readErr = err + } + return n, err +} + func getParticipantEmails(channelExport *ChannelExport) []string { participantEmails := make([]string, 0, len(channelExport.Participants)) for _, participant := range channelExport.Participants {