[MM-69242] prevent Global Relay export panic on attachment read failure (release-10.11) (#37106)
Automatic Merge
Этот коммит содержится в:
Vishal
@@ -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)
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Ссылка в новой задаче
Block a user