MM-44542: Add timeout to S3 operations (#20648)

https://mattermost.atlassian.net/browse/MM-44542

```release-note
We add a new config setting AmazonS3RequestTimeoutMilliseconds under
FileSettings which sets a timeout for requests to AWS S3.

By default, the timeout is at 30 seconds.
```

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

i18n/en.json

@@ -7983,6 +7983,10 @@
     "id": "model.config.is_valid.allow_cookies_for_subdomains.app_error",
     "translation": "Allowing cookies for subdomains requires SiteURL to be set."
   },
+  {
+    "id": "model.config.is_valid.amazons3_timeout.app_error",
+    "translation": "Invalid timeout value {{.Value}}. Should be a positive number."
+  },
   {
     "id": "model.config.is_valid.atmos_camo_image_proxy_options.app_error",
     "translation": "Invalid RemoteImageProxyOptions for atmos/camo. Must be set to your shared key."

model/config.go

@@ -1429,6 +1429,7 @@ type FileSettings struct {
 	AmazonS3SignV2                     *bool   `access:"environment_file_storage,write_restrictable,cloud_restrictable"`
 	AmazonS3SSE                        *bool   `access:"environment_file_storage,write_restrictable,cloud_restrictable"`
 	AmazonS3Trace                      *bool   `access:"environment_file_storage,write_restrictable,cloud_restrictable"`
+	AmazonS3RequestTimeoutMilliseconds *int64  `access:"environment_file_storage,write_restrictable,cloud_restrictable"` // telemetry: none
 }
 
 func (s *FileSettings) SetDefaults(isUpdate bool) {
@@ -1532,6 +1533,10 @@ func (s *FileSettings) SetDefaults(isUpdate bool) {
 	if s.AmazonS3Trace == nil {
 		s.AmazonS3Trace = NewBool(false)
 	}
+
+	if s.AmazonS3RequestTimeoutMilliseconds == nil {
+		s.AmazonS3RequestTimeoutMilliseconds = NewInt64(30000)
+	}
 }
 
 func (s *FileSettings) ToFileBackendSettings(enableComplianceFeature bool, skipVerify bool) filestore.FileBackendSettings {
@@ -1553,6 +1558,7 @@ func (s *FileSettings) ToFileBackendSettings(enableComplianceFeature bool, skipV
 		AmazonS3SignV2:                     s.AmazonS3SignV2 != nil && *s.AmazonS3SignV2,
 		AmazonS3SSE:                        s.AmazonS3SSE != nil && *s.AmazonS3SSE && enableComplianceFeature,
 		AmazonS3Trace:                      s.AmazonS3Trace != nil && *s.AmazonS3Trace,
+		AmazonS3RequestTimeoutMilliseconds: *s.AmazonS3RequestTimeoutMilliseconds,
 		SkipVerify:                         skipVerify,
 	}
 }
@@ -3416,6 +3422,10 @@ func (s *FileSettings) isValid() *AppError {
 		return NewAppError("Config.IsValid", "model.config.is_valid.image_decoder_concurrency.app_error", map[string]any{"Value": *s.MaxImageDecoderConcurrency}, "", http.StatusBadRequest)
 	}
 
+	if *s.AmazonS3RequestTimeoutMilliseconds <= 0 {
+		return NewAppError("Config.IsValid", "model.config.is_valid.amazons3_timeout.app_error", map[string]any{"Value": *s.MaxImageDecoderConcurrency}, "", http.StatusBadRequest)
+	}
+
 	return nil
 }
 

shared/filestore/filesstore.go

@@ -53,6 +53,7 @@ type FileBackendSettings struct {
 	AmazonS3SSE                        bool
 	AmazonS3Trace                      bool
 	SkipVerify                         bool
+	AmazonS3RequestTimeoutMilliseconds int64
 }
 
 func (settings *FileBackendSettings) CheckMandatoryS3Fields() error {

shared/filestore/filesstore_test.go

@@ -82,6 +82,7 @@ func runBackendTest(t *testing.T, encrypt bool) {
 			AmazonS3PathPrefix:                 "",
 			AmazonS3SSL:                        false,
 			AmazonS3SSE:                        encrypt,
+			AmazonS3RequestTimeoutMilliseconds: 5000,
 		},
 	})
 }

shared/filestore/s3store.go

@@ -37,6 +37,7 @@ type S3FileBackend struct {
 	trace      bool
 	client     *s3.Client
 	skipVerify bool
+	timeout    time.Duration
 }
 
 type S3FileBackendAuthError struct {
@@ -79,6 +80,7 @@ func (s *S3FileBackendNoBucketError) Error() string {
 
 // NewS3FileBackend returns an instance of an S3FileBackend.
 func NewS3FileBackend(settings FileBackendSettings) (*S3FileBackend, error) {
+	timeout := time.Duration(settings.AmazonS3RequestTimeoutMilliseconds) * time.Millisecond
 	backend := &S3FileBackend{
 		endpoint:   settings.AmazonS3Endpoint,
 		accessKey:  settings.AmazonS3AccessKeyId,
@@ -91,6 +93,7 @@ func NewS3FileBackend(settings FileBackendSettings) (*S3FileBackend, error) {
 		encrypt:    settings.AmazonS3SSE,
 		trace:      settings.AmazonS3Trace,
 		skipVerify: settings.SkipVerify,
+		timeout:    timeout,
 	}
 	cli, err := backend.s3New()
 	if err != nil {
@@ -167,8 +170,10 @@ func (b *S3FileBackend) TestConnection() error {
 	// If a path prefix is present, we attempt to test the bucket by listing objects under the path
 	// and just checking the first response. This is because the BucketExists call is only at a bucket level
 	// and sometimes the user might only be allowed access to the specified path prefix.
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
 	if b.pathPrefix != "" {
-		obj := <-b.client.ListObjects(context.Background(), b.bucket, s3.ListObjectsOptions{Prefix: b.pathPrefix})
+		obj := <-b.client.ListObjects(ctx, b.bucket, s3.ListObjectsOptions{Prefix: b.pathPrefix})
 		if obj.Err != nil {
 			typedErr := s3.ToErrorResponse(obj.Err)
 			if typedErr.Code != bucketNotFound {
@@ -177,7 +182,7 @@ func (b *S3FileBackend) TestConnection() error {
 			exists = false
 		}
 	} else {
-		exists, err = b.client.BucketExists(context.Background(), b.bucket)
+		exists, err = b.client.BucketExists(ctx, b.bucket)
 		if err != nil {
 			return &S3FileBackendAuthError{DetailedError: "unable to check if the S3 bucket exists"}
 		}
@@ -191,27 +196,45 @@ func (b *S3FileBackend) TestConnection() error {
 }
 
 func (b *S3FileBackend) MakeBucket() error {
-	err := b.client.MakeBucket(context.Background(), b.bucket, s3.MakeBucketOptions{Region: b.region})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	err := b.client.MakeBucket(ctx, b.bucket, s3.MakeBucketOptions{Region: b.region})
 	if err != nil {
 		return errors.Wrap(err, "unable to create the s3 bucket")
 	}
 	return nil
 }
 
+// s3WithCancel is a wrapper struct which cancels the context
+// when the object is closed.
+type s3WithCancel struct {
+	*s3.Object
+	cancel context.CancelFunc
+}
+
+func (sc *s3WithCancel) Close() error {
+	sc.cancel()
+	return sc.Object.Close()
+}
+
 // Caller must close the first return value
 func (b *S3FileBackend) Reader(path string) (ReadCloseSeeker, error) {
 	path = filepath.Join(b.pathPrefix, path)
-	minioObject, err := b.client.GetObject(context.Background(), b.bucket, path, s3.GetObjectOptions{})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	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 minioObject, nil
+	return &s3WithCancel{Object: minioObject, cancel: cancel}, nil
 }
 
 func (b *S3FileBackend) ReadFile(path string) ([]byte, error) {
 	path = filepath.Join(b.pathPrefix, path)
-	minioObject, err := b.client.GetObject(context.Background(), b.bucket, path, s3.GetObjectOptions{})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	minioObject, err := b.client.GetObject(ctx, b.bucket, path, s3.GetObjectOptions{})
 	if err != nil {
 		return nil, errors.Wrapf(err, "unable to open file %s", path)
 	}
@@ -227,7 +250,9 @@ func (b *S3FileBackend) ReadFile(path string) ([]byte, error) {
 func (b *S3FileBackend) FileExists(path string) (bool, error) {
 	path = filepath.Join(b.pathPrefix, path)
 
-	_, err := b.client.StatObject(context.Background(), b.bucket, path, s3.StatObjectOptions{})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	_, err := b.client.StatObject(ctx, b.bucket, path, s3.StatObjectOptions{})
 	if err == nil {
 		return true, nil
 	}
@@ -243,7 +268,9 @@ func (b *S3FileBackend) FileExists(path string) (bool, error) {
 func (b *S3FileBackend) FileSize(path string) (int64, error) {
 	path = filepath.Join(b.pathPrefix, path)
 
-	info, err := b.client.StatObject(context.Background(), b.bucket, path, s3.StatObjectOptions{})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	info, err := b.client.StatObject(ctx, b.bucket, path, s3.StatObjectOptions{})
 	if err != nil {
 		return 0, errors.Wrapf(err, "unable to get file size for %s", path)
 	}
@@ -254,7 +281,9 @@ func (b *S3FileBackend) FileSize(path string) (int64, error) {
 func (b *S3FileBackend) FileModTime(path string) (time.Time, error) {
 	path = filepath.Join(b.pathPrefix, path)
 
-	info, err := b.client.StatObject(context.Background(), b.bucket, path, s3.StatObjectOptions{})
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	info, err := b.client.StatObject(ctx, b.bucket, path, s3.StatObjectOptions{})
 	if err != nil {
 		return time.Time{}, errors.Wrapf(err, "unable to get modification time for file %s", path)
 	}
@@ -281,7 +310,9 @@ func (b *S3FileBackend) CopyFile(oldPath, newPath string) error {
 		dstOpts.Encryption = encrypt.NewSSE()
 	}
 
-	if _, err := b.client.CopyObject(context.Background(), dstOpts, srcOpts); err != nil {
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	if _, err := b.client.CopyObject(ctx, dstOpts, srcOpts); err != nil {
 		return errors.Wrapf(err, "unable to copy file from %s to %s", oldPath, newPath)
 	}
 
@@ -307,11 +338,15 @@ func (b *S3FileBackend) MoveFile(oldPath, newPath string) error {
 		dstOpts.Encryption = encrypt.NewSSE()
 	}
 
-	if _, err := b.client.CopyObject(context.Background(), dstOpts, srcOpts); err != nil {
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	if _, err := b.client.CopyObject(ctx, dstOpts, srcOpts); err != nil {
 		return errors.Wrapf(err, "unable to copy the file to %s to the new destination", newPath)
 	}
 
-	if err := b.client.RemoveObject(context.Background(), b.bucket, oldPath, s3.RemoveObjectOptions{}); err != nil {
+	ctx2, cancel2 := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel2()
+	if err := b.client.RemoveObject(ctx2, b.bucket, oldPath, s3.RemoveObjectOptions{}); err != nil {
 		return errors.Wrapf(err, "unable to remove the file old file %s", oldPath)
 	}
 
@@ -327,8 +362,10 @@ func (b *S3FileBackend) WriteFile(fr io.Reader, path string) (int64, error) {
 		contentType = "binary/octet-stream"
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
 	options := s3PutOptions(b.encrypt, contentType)
-	info, err := b.client.PutObject(context.Background(), b.bucket, path, fr, -1, options)
+	info, err := b.client.PutObject(ctx, b.bucket, path, fr, -1, options)
 	if err != nil {
 		return info.Size, errors.Wrapf(err, "unable write the data in the file %s", path)
 	}
@@ -338,7 +375,9 @@ func (b *S3FileBackend) WriteFile(fr io.Reader, path string) (int64, error) {
 
 func (b *S3FileBackend) AppendFile(fr io.Reader, path string) (int64, error) {
 	fp := filepath.Join(b.pathPrefix, path)
-	if _, err := b.client.StatObject(context.Background(), b.bucket, fp, s3.StatObjectOptions{}); err != nil {
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	if _, err := b.client.StatObject(ctx, b.bucket, fp, s3.StatObjectOptions{}); err != nil {
 		return 0, errors.Wrapf(err, "unable to find the file %s to append the data", path)
 	}
 
@@ -352,9 +391,18 @@ func (b *S3FileBackend) AppendFile(fr io.Reader, path string) (int64, error) {
 	options := s3PutOptions(b.encrypt, contentType)
 	sse := options.ServerSideEncryption
 	partName := fp + ".part"
-	info, err := b.client.PutObject(context.Background(), b.bucket, partName, fr, -1, options)
+	ctx2, cancel2 := context.WithTimeout(context.Background(), b.timeout)
-	defer b.client.RemoveObject(context.Background(), b.bucket, partName, s3.RemoveObjectOptions{})
+	defer cancel2()
-	if info.Size > 0 {
+	info, err := b.client.PutObject(ctx2, b.bucket, partName, fr, -1, options)
+	if err != nil {
+		return 0, errors.Wrapf(err, "unable append the data in the file %s", path)
+	}
+	defer func() {
+		ctx4, cancel4 := context.WithTimeout(context.Background(), b.timeout)
+		defer cancel4()
+		b.client.RemoveObject(ctx4, b.bucket, partName, s3.RemoveObjectOptions{})
+	}()
+
 	src1Opts := s3.CopySrcOptions{
 		Bucket: b.bucket,
 		Object: fp,
@@ -368,19 +416,21 @@ func (b *S3FileBackend) AppendFile(fr io.Reader, path string) (int64, error) {
 		Object:     fp,
 		Encryption: sse,
 	}
-		_, err = b.client.ComposeObject(context.Background(), dstOpts, src1Opts, src2Opts)
+	ctx3, cancel3 := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel3()
+	_, err = b.client.ComposeObject(ctx3, dstOpts, src1Opts, src2Opts)
 	if err != nil {
 		return 0, errors.Wrapf(err, "unable append the data in the file %s", path)
 	}
 	return info.Size, nil
-	}
 
-	return 0, errors.Wrapf(err, "unable append the data in the file %s", path)
 }
 
 func (b *S3FileBackend) RemoveFile(path string) error {
 	path = filepath.Join(b.pathPrefix, path)
-	if err := b.client.RemoveObject(context.Background(), b.bucket, path, s3.RemoveObjectOptions{}); err != nil {
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	if err := b.client.RemoveObject(ctx, b.bucket, path, s3.RemoveObjectOptions{}); err != nil {
 		return errors.Wrapf(err, "unable to remove the file %s", path)
 	}
 
@@ -420,7 +470,9 @@ func (b *S3FileBackend) listDirectory(path string, recursion bool) ([]string, er
 		Recursive: recursion,
 	}
 	var paths []string
-	for object := range b.client.ListObjects(context.Background(), b.bucket, opts) {
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel()
+	for object := range b.client.ListObjects(ctx, b.bucket, opts) {
 		if object.Err != nil {
 			return nil, errors.Wrapf(object.Err, "unable to list the directory %s", path)
 		}
@@ -449,8 +501,13 @@ func (b *S3FileBackend) RemoveDirectory(path string) error {
 		Prefix:    filepath.Join(b.pathPrefix, path),
 		Recursive: true,
 	}
-	list := b.client.ListObjects(context.Background(), b.bucket, opts)
+	ctx, cancel := context.WithTimeout(context.Background(), b.timeout)
-	objectsCh := b.client.RemoveObjects(context.Background(), b.bucket, getPathsFromObjectInfos(list), s3.RemoveObjectsOptions{})
+	defer cancel()
+	list := b.client.ListObjects(ctx, b.bucket, opts)
+
+	ctx2, cancel2 := context.WithTimeout(context.Background(), b.timeout)
+	defer cancel2()
+	objectsCh := b.client.RemoveObjects(ctx2, b.bucket, getPathsFromObjectInfos(list), s3.RemoveObjectsOptions{})
 	for err := range objectsCh {
 		if err.Err != nil {
 			return errors.Wrapf(err.Err, "unable to remove the directory %s", path)

shared/filestore/s3store_test.go

@@ -4,8 +4,11 @@
 package filestore
 
 import (
+	"bytes"
+	"context"
 	"crypto/rand"
 	"encoding/base64"
+	"errors"
 	"fmt"
 	"net/http/httptest"
 	"net/http/httputil"
@@ -73,6 +76,7 @@ func TestMakeBucket(t *testing.T) {
 		AmazonS3PathPrefix:                 "",
 		AmazonS3SSL:                        false,
 		SkipVerify:                         false,
+		AmazonS3RequestTimeoutMilliseconds: 5000,
 	}
 
 	fileBackend, err := NewS3FileBackend(cfg)
@@ -82,6 +86,51 @@ func TestMakeBucket(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestTimeout(t *testing.T) {
+	s3Host := os.Getenv("CI_MINIO_HOST")
+	if s3Host == "" {
+		s3Host = "localhost"
+	}
+
+	s3Port := os.Getenv("CI_MINIO_PORT")
+	if s3Port == "" {
+		s3Port = "9000"
+	}
+
+	s3Endpoint := fmt.Sprintf("%s:%s", s3Host, s3Port)
+
+	// Generate a random bucket name
+	b := make([]byte, 30)
+	rand.Read(b)
+	bucketName := base64.StdEncoding.EncodeToString(b)
+	bucketName = strings.ToLower(bucketName)
+	bucketName = strings.Replace(bucketName, "+", "", -1)
+	bucketName = strings.Replace(bucketName, "/", "", -1)
+
+	cfg := FileBackendSettings{
+		DriverName:                         ImageDriverS3,
+		AmazonS3AccessKeyId:                MinioAccessKey,
+		AmazonS3SecretAccessKey:            MinioSecretKey,
+		AmazonS3Bucket:                     bucketName,
+		AmazonS3Endpoint:                   s3Endpoint,
+		AmazonS3Region:                     "",
+		AmazonS3PathPrefix:                 "",
+		AmazonS3SSL:                        false,
+		SkipVerify:                         false,
+		AmazonS3RequestTimeoutMilliseconds: 0,
+	}
+
+	fileBackend, err := NewS3FileBackend(cfg)
+	require.NoError(t, err)
+
+	err = fileBackend.MakeBucket()
+	require.True(t, errors.Is(err, context.DeadlineExceeded))
+
+	path := "tests/" + randomString() + ".png"
+	_, err = fileBackend.WriteFile(bytes.NewReader([]byte("testimage")), path)
+	require.True(t, errors.Is(err, context.DeadlineExceeded))
+}
+
 func TestInsecureMakeBucket(t *testing.T) {
 	s3Host := os.Getenv("CI_MINIO_HOST")
 	if s3Host == "" {
@@ -129,6 +178,7 @@ func TestInsecureMakeBucket(t *testing.T) {
 				AmazonS3PathPrefix:                 "",
 				AmazonS3SSL:                        true,
 				SkipVerify:                         testCase.skipVerify,
+				AmazonS3RequestTimeoutMilliseconds: 5000,
 			}
 
 			fileBackend, err := NewS3FileBackend(cfg)