diff --git a/jobs/import_process/worker.go b/jobs/import_process/worker.go index c5414ff4a0..e9d29a6414 100644 --- a/jobs/import_process/worker.go +++ b/jobs/import_process/worker.go @@ -63,6 +63,14 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) model.Worker { } defer importFile.Close() + // The import is a long running operation, try to cancel any timeouts attached to the reader. + type TimeoutCanceler interface{ CancelTimeout() bool } + if tc, ok := importFile.(TimeoutCanceler); ok { + if !tc.CancelTimeout() { + appContext.Logger().Warn("Could not cancel the timeout for the file reader. The import may fail due to a timeout.") + } + } + importZipReader, err := zip.NewReader(importFile.(io.ReaderAt), importFileSize) if err != nil { return model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.open_file", nil, "", http.StatusInternalServerError).Wrap(err) diff --git a/shared/filestore/s3store.go b/shared/filestore/s3store.go index ff6fcd246f..60132cd830 100644 --- a/shared/filestore/s3store.go +++ b/shared/filestore/s3store.go @@ -208,26 +208,41 @@ func (b *S3FileBackend) MakeBucket() error { // s3WithCancel is a wrapper struct which cancels the context // when the object is closed. type s3WithCancel struct { - *s3.Object + io.ReadSeekCloser + timer *time.Timer cancel context.CancelFunc } func (sc *s3WithCancel) Close() error { + sc.timer.Stop() sc.cancel() - return sc.Object.Close() + return sc.ReadSeekCloser.Close() +} + +// CancelTimeout attempts to cancel the timeout for this reader. It allows calling +// code to ignore the timeout in case of longer running operations. The methods returns +// false if the timeout has already fired. +func (sc *s3WithCancel) CancelTimeout() bool { + return sc.timer.Stop() } // Caller must close the first return value func (b *S3FileBackend) Reader(path string) (ReadCloseSeeker, error) { path = filepath.Join(b.pathPrefix, path) - ctx, cancel := context.WithTimeout(context.Background(), b.timeout) + ctx, cancel := context.WithCancel(context.Background()) minioObject, err := b.client.GetObject(ctx, b.bucket, path, s3.GetObjectOptions{}) if err != nil { cancel() return nil, errors.Wrapf(err, "unable to open file %s", path) } - return &s3WithCancel{Object: minioObject, cancel: cancel}, nil + sc := &s3WithCancel{ + ReadSeekCloser: minioObject, + timer: time.AfterFunc(b.timeout, cancel), + cancel: cancel, + } + + return sc, nil } func (b *S3FileBackend) ReadFile(path string) ([]byte, error) { diff --git a/shared/filestore/s3store_test.go b/shared/filestore/s3store_test.go index 60c765ad0b..5222226a5b 100644 --- a/shared/filestore/s3store_test.go +++ b/shared/filestore/s3store_test.go @@ -10,12 +10,14 @@ import ( "encoding/base64" "errors" "fmt" + "io" "net/http/httptest" "net/http/httputil" "net/url" "os" "strings" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -196,3 +198,105 @@ func TestInsecureMakeBucket(t *testing.T) { func newTLSProxyServer(backend *url.URL) *httptest.Server { return httptest.NewTLSServer(httputil.NewSingleHostReverseProxy(backend)) } + +func TestS3WithCancel(t *testing.T) { + // Some of these tests use time.Sleep to wait for the timeout to expire. + // They are run in parallel to reduce wait times. + + t.Run("zero timeout", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(0, nil) + + time.Sleep(10 * time.Millisecond) // give the context time to cancel + + require.False(t, r.CancelTimeout()) + require.Error(t, ctx.Err()) + }) + + t.Run("timeout", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(50*time.Millisecond, nil) + + time.Sleep(100 * time.Millisecond) // give the context time to cancel + + require.False(t, r.CancelTimeout()) + require.Error(t, ctx.Err()) + }) + + t.Run("timeout cancel", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(50*time.Millisecond, nil) + + time.Sleep(10 * time.Millisecond) // give the context time to cancel + + require.True(t, r.CancelTimeout()) + require.NoError(t, ctx.Err()) + + time.Sleep(100 * time.Millisecond) // wait for the original (canceled) timeout to expire + + require.False(t, r.CancelTimeout()) + require.NoError(t, ctx.Err()) + require.NoError(t, r.Close()) + }) + + t.Run("timeout closed", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(50*time.Millisecond, nil) + + time.Sleep(10 * time.Millisecond) // give the context time to cancel + + require.True(t, r.CancelTimeout()) + require.NoError(t, ctx.Err()) + require.NoError(t, r.Close()) + + time.Sleep(100 * time.Millisecond) // wait for the original (canceled) timeout to expire + + require.False(t, r.CancelTimeout()) + require.Error(t, ctx.Err()) + require.NoError(t, r.Close()) + }) + + t.Run("close cancel close", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(50*time.Millisecond, nil) + + time.Sleep(10 * time.Millisecond) // give the context time to cancel + + require.True(t, r.CancelTimeout()) + require.NoError(t, r.Close()) + require.Error(t, ctx.Err()) + require.False(t, r.CancelTimeout()) + require.Error(t, ctx.Err()) + require.NoError(t, r.Close()) + }) + + t.Run("close error", func(t *testing.T) { + t.Parallel() + r, ctx := newMockS3WithCancel(50*time.Millisecond, errors.New("test error")) + + time.Sleep(10 * time.Millisecond) // give the context time to cancel + + require.NoError(t, ctx.Err()) + require.Error(t, r.Close()) + require.False(t, r.CancelTimeout()) + require.Error(t, ctx.Err()) + }) +} + +func newMockS3WithCancel(timeout time.Duration, closeErr error) (*s3WithCancel, context.Context) { + ctx, cancel := context.WithCancel(context.Background()) + return &s3WithCancel{ + ReadSeekCloser: fauxCloser{strings.NewReader("testdata"), closeErr}, + timer: time.AfterFunc(timeout, cancel), + cancel: cancel, + }, ctx +} + +type fauxCloser struct { + io.ReadSeeker + closeErr error +} + +func (fc fauxCloser) Close() error { + return fc.closeErr +}