From 6963ab89291441cbe7bb7ea858f61b2c6c3a0339 Mon Sep 17 00:00:00 2001 From: wty-Bryant Date: Wed, 12 Aug 2026 18:00:38 -0400 Subject: [PATCH 1/3] close files after transfer finish for keyprefix transfer --- .../api_op_DownloadDirectory.go | 102 +++++++---- .../transfermanager/api_op_UploadDirectory.go | 73 +++++--- .../download_directory_test.go | 166 ++++++++++++++++++ .../transfermanager/upload_directory_test.go | 154 ++++++++++++++++ 4 files changed, 429 insertions(+), 66 deletions(-) diff --git a/feature/s3/transfermanager/api_op_DownloadDirectory.go b/feature/s3/transfermanager/api_op_DownloadDirectory.go index c151f49b8aa0..aaa9eb7dc965 100644 --- a/feature/s3/transfermanager/api_op_DownloadDirectory.go +++ b/feature/s3/transfermanager/api_op_DownloadDirectory.go @@ -17,6 +17,13 @@ import ( s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" ) +// createFileFn is a thin indirection over os.Create used when writing +// downloaded objects to disk. It exists so tests can verify that every file +// handle DownloadDirectory creates is subsequently closed, guarding against a +// file-handle leak analogous to the one fixed for aws/aws-sdk-go-v2#3512. In +// prod env it is exactly os.Create. +var createFileFn = os.Create + // DownloadDirectoryInput represents a request to the DownloadDirectory() call type DownloadDirectoryInput struct { // Bucket where objects are downloaded from @@ -262,54 +269,69 @@ func (d *directoryDownloader) downloadObject(ctx context.Context, ch chan object continue } - input := &GetObjectInput{ - Bucket: d.in.Bucket, - Key: aws.String(data.key), - } - if d.in.Callback != nil { - d.in.Callback.UpdateRequest(input) + if err := d.downloadSingleObject(ctx, data); err != nil { + d.setErr(err) } - out, err := d.c.GetObject(ctx, input) + } +} + +func (d *directoryDownloader) downloadSingleObject(ctx context.Context, data objectEntry) error { + input := &GetObjectInput{ + Bucket: d.in.Bucket, + Key: aws.String(data.key), + } + if d.in.Callback != nil { + d.in.Callback.UpdateRequest(input) + } + out, err := d.c.GetObject(ctx, input) + if err != nil { + err = d.failurePolicy.OnDownloadFailed(d.in, input, err) if err != nil { - err = d.failurePolicy.OnDownloadFailed(d.in, input, err) - if err != nil { - d.setErr(fmt.Errorf("error when heading info of object %s: %v", data.key, err)) - } else { - d.objectsFailed.Add(1) - } - continue + return fmt.Errorf("error when heading info of object %s: %v", data.key, err) + } else { + d.objectsFailed.Add(1) + return nil } + } - d.progressOnce.Do(func() { - d.emitter.Start(ctx, d.in) - }) + d.progressOnce.Do(func() { + d.emitter.Start(ctx, d.in) + }) - err = os.MkdirAll(filepath.Dir(data.path), 0755) - if err != nil { - d.setErr(fmt.Errorf("error when creating directory for file %s: %v", data.path, err)) - continue + err = os.MkdirAll(filepath.Dir(data.path), 0755) + if err != nil { + return fmt.Errorf("error when creating directory for file %s: %v", data.path, err) + } + + file, err := createFileFn(data.path) + if err != nil { + return fmt.Errorf("error when creating file %s: %v", data.path, err) + } + var fileCopyFail bool + defer func() { + if err := file.Close(); err != nil { + d.setErr(fmt.Errorf("error when closing file %s: %v", data.path, err)) } - file, err := os.Create(data.path) - if err != nil { - d.setErr(fmt.Errorf("error when creating file %s: %v", data.path, err)) - continue + if fileCopyFail { + os.Remove(data.path) // only remove the file if the copy failed } - n, err := io.Copy(file, out.Body) + }() + n, err := io.Copy(file, out.Body) + if err != nil { + fileCopyFail = true + // where s3.GetObject is really called, must be handled by failure policy + err = d.failurePolicy.OnDownloadFailed(d.in, input, err) if err != nil { - // where s3.GetObject is really called, must be handled by failure policy - err = d.failurePolicy.OnDownloadFailed(d.in, input, err) - if err != nil { - d.setErr(fmt.Errorf("error when getting object and writing to local file %s: %v", data.path, err)) - } else { - d.objectsFailed.Add(1) - } - os.Remove(data.path) - continue + return fmt.Errorf("error when getting object and writing to local file %s: %v", data.path, err) + } else { + d.objectsFailed.Add(1) + return nil } - - d.objectsDownloaded.Add(1) - d.emitter.ObjectsTransferred(ctx, n) } + + d.objectsDownloaded.Add(1) + d.emitter.ObjectsTransferred(ctx, n) + return nil } func (d *directoryDownloader) freshContext(ctx context.Context) (context.Context, context.CancelFunc) { @@ -323,7 +345,9 @@ func (d *directoryDownloader) setErr(err error) { d.mu.Lock() defer d.mu.Unlock() - d.err = err + if d.err == nil || (isCancellationError(d.err) && !isCancellationError(err)) { + d.err = err + } } func (d *directoryDownloader) getErr() error { diff --git a/feature/s3/transfermanager/api_op_UploadDirectory.go b/feature/s3/transfermanager/api_op_UploadDirectory.go index c8fad4cf9bd9..831267f091fd 100644 --- a/feature/s3/transfermanager/api_op_UploadDirectory.go +++ b/feature/s3/transfermanager/api_op_UploadDirectory.go @@ -313,6 +313,11 @@ func (u *directoryUploader) traverseFolder(path string) ([]string, error) { if err != nil { return []string{}, err } + defer func() { + if err := f.Close(); err != nil { + u.setErr(fmt.Errorf("error when closing folder %s: %v", path, err)) + } + }() subFiles, err := f.ReadDir(0) if err != nil { return []string{}, err @@ -375,37 +380,49 @@ func (u *directoryUploader) uploadFile(ctx context.Context, ch chan fileEntry) { if u.getErr() != nil { continue } - f, err := os.Open(data.path) - if err != nil { - u.setErr(fmt.Errorf("error when opening file %s: %v", data.path, err)) - continue - } - input := &UploadObjectInput{ - Bucket: u.in.Bucket, - Key: aws.String(data.key), - Body: f, + + if err := u.uploadSingleFile(ctx, data); err != nil { + u.setErr(err) } - if u.in.Callback != nil { - u.in.Callback.UpdateRequest(input) + } +} + +func (u *directoryUploader) uploadSingleFile(ctx context.Context, data fileEntry) error { + f, err := os.Open(data.path) + if err != nil { + return fmt.Errorf("error when opening file %s: %v", data.path, err) + } + defer func() { + if err := f.Close(); err != nil { + u.setErr(fmt.Errorf("error when closing file %s: %v", data.path, err)) } - out, err := u.c.UploadObject(ctx, input) + }() + input := &UploadObjectInput{ + Bucket: u.in.Bucket, + Key: aws.String(data.key), + Body: f, + } + if u.in.Callback != nil { + u.in.Callback.UpdateRequest(input) + } + out, err := u.c.UploadObject(ctx, input) + if err != nil { + err = u.failurePolicy.OnUploadFailed(u.in, input, err) if err != nil { - err = u.failurePolicy.OnUploadFailed(u.in, input, err) - if err != nil { - u.setErr(fmt.Errorf("error when uploading file %s: %v", data.path, err)) - } else { - // this failed object is ignored, just increase the failure count - u.filesFailed.Add(1) - } - continue + return fmt.Errorf("error when uploading file %s: %v", data.path, err) + } else { + // this failed object is ignored, just increase the failure count + u.filesFailed.Add(1) + return nil } - - u.progressOnce.Do(func() { - u.emitter.Start(ctx, u.in) - }) - u.filesUploaded.Add(1) - u.emitter.ObjectsTransferred(ctx, aws.ToInt64(out.ContentLength)) } + + u.progressOnce.Do(func() { + u.emitter.Start(ctx, u.in) + }) + u.filesUploaded.Add(1) + u.emitter.ObjectsTransferred(ctx, aws.ToInt64(out.ContentLength)) + return nil } func (u *directoryUploader) freshContext(ctx context.Context) (context.Context, context.CancelFunc) { @@ -419,7 +436,9 @@ func (u *directoryUploader) setErr(err error) { u.mu.Lock() defer u.mu.Unlock() - u.err = err + if u.err == nil || (isCancellationError(u.err) && !isCancellationError(err)) { + u.err = err + } } func (u *directoryUploader) getErr() error { diff --git a/feature/s3/transfermanager/download_directory_test.go b/feature/s3/transfermanager/download_directory_test.go index 585bc467c9d4..cc2cf5ab0095 100644 --- a/feature/s3/transfermanager/download_directory_test.go +++ b/feature/s3/transfermanager/download_directory_test.go @@ -11,6 +11,7 @@ import ( "runtime" "sort" "strings" + "sync" "testing" "github.com/aws/aws-sdk-go-v2/internal/awstesting" @@ -700,3 +701,168 @@ func TestDownloadDirectoryWithContextCanceled(t *testing.T) { t.Errorf("expected error message to contain %q, but did not %q", e, a) } } + +// createdFileCapture records the *os.File handles that DownloadDirectory +// creates via createFileFn, so a test can assert after the call returns that +// each one was closed. A closed *os.File returns an error from Stat, so a nil +// error means the handle leaked. It is safe for concurrent use by the download +// workers. +type createdFileCapture struct { + mu sync.Mutex + files []*os.File +} + +func (c *createdFileCapture) capture(f *os.File) { + c.mu.Lock() + c.files = append(c.files, f) + c.mu.Unlock() +} + +func (c *createdFileCapture) count() int { + return len(c.files) +} + +// stillOpen returns the names of captured files that are still open. A closed +// *os.File returns an error from Stat, so a nil error means the handle leaked. +func (c *createdFileCapture) stillOpen() []string { + var open []string + for _, f := range c.files { + if _, err := f.Stat(); err == nil { + open = append(open, f.Name()) + } + } + sort.Strings(open) + return open +} + +// TestDownloadDirectoryClosesCreatedFiles verifies that DownloadDirectory +// closes every file handle it creates before returning - on the success path +// and when individual object downloads fail and are ignored. Regression test +// for a file-handle leak analogous to aws/aws-sdk-go-v2#3512. +func TestDownloadDirectoryClosesCreatedFiles(t *testing.T) { + _, filename, _, _ := runtime.Caller(0) + root := filepath.Join(filepath.Dir(filename), "testdata") + + cases := map[string]struct { + destination string + objectsLists [][]s3types.Object + continuationTokens []string + getobjectFn func(*s3testing.TransferManagerLoggingClient, *s3.GetObjectInput) (*s3.GetObjectOutput, error) + failurePolicy DownloadDirectoryFailurePolicy + expectCreated int + expectErr string + }{ + "single object": { + destination: "close-single-object", + objectsLists: [][]s3types.Object{ + {{Key: aws.String("foo/bar")}}, + }, + expectCreated: 1, + }, + "multiple objects with subdirs": { + destination: "close-multiple-objects", + objectsLists: [][]s3types.Object{ + { + {Key: aws.String("foo/bar")}, + {Key: aws.String("baz")}, + {Key: aws.String("foo/zoo/bar")}, + {Key: aws.String("foo/zoo/oii/bababoii")}, + }, + }, + expectCreated: 4, + }, + "multiple objects paginated": { + destination: "close-multiple-objects-paginated", + objectsLists: [][]s3types.Object{ + {{Key: aws.String("foo/bar")}, {Key: aws.String("baz")}}, + {{Key: aws.String("foo/zoo/bar")}, {Key: aws.String("foo/zoo/oii/bababoii")}}, + {{Key: aws.String("foo/zoo/baz")}, {Key: aws.String("foo/zoo/oii/yee")}}, + }, + continuationTokens: []string{"token1", "token2"}, + expectCreated: 6, + }, + "created files are closed when some downloads fail and are ignored": { + destination: "close-error-ignored", + objectsLists: [][]s3types.Object{ + { + {Key: aws.String("foo/bar")}, + {Key: aws.String("baz")}, + {Key: aws.String("foo/zoo/bar")}, + {Key: aws.String("foo/zoo/oii/bababoii")}, + }, + }, + getobjectFn: func(c *s3testing.TransferManagerLoggingClient, in *s3.GetObjectInput) (*s3.GetObjectOutput, error) { + if key := aws.ToString(in.Key); key == "foo/zoo/bar" || key == "baz" { + return nil, fmt.Errorf("mocking error") + } + return &s3.GetObjectOutput{ + Body: io.NopCloser(bytes.NewReader(c.Data)), + ContentLength: aws.Int64(int64(len(c.Data))), + PartsCount: aws.Int32(c.PartsCount), + ETag: aws.String(etag), + }, nil + }, + failurePolicy: IgnoreDownloadFailurePolicy{}, + // GetObject only performs a HeadObject up front and streams the + // body lazily, so os.Create runs for all four objects. foo/zoo/bar + // and baz then fail during io.Copy (body read), so their files are + // created, closed, and removed; the other two download normally. + // Every created file must be closed regardless of the outcome. + expectCreated: 4, + }, + } + + for name, c := range cases { + t.Run(name, func(t *testing.T) { + capture := &createdFileCapture{} + prev := createFileFn + createFileFn = func(name string) (*os.File, error) { + f, err := prev(name) + if err == nil { + capture.capture(f) + } + return f, err + } + defer func() { createFileFn = prev }() + + s3Client, _ := s3testing.NewDownloadDirectoryClient() + s3Client.ListObjectsData = c.objectsLists + s3Client.ContinuationTokens = c.continuationTokens + if c.getobjectFn == nil { + s3Client.GetObjectFn = s3testing.PartGetObjectFn + } else { + s3Client.GetObjectFn = c.getobjectFn + } + s3Client.Data = make([]byte, 0) + s3Client.PartsCount = 1 + mgr := New(s3Client) + + dstPath := filepath.Join(root, c.destination) + defer os.RemoveAll(dstPath) + + req := &DownloadDirectoryInput{ + Bucket: aws.String("mock-bucket"), + Destination: aws.String(dstPath), + FailurePolicy: c.failurePolicy, + } + + _, err := mgr.DownloadDirectory(context.Background(), req) + if err != nil { + if c.expectErr == "" { + t.Fatalf("expect no error, got %v", err) + } else if !strings.Contains(err.Error(), c.expectErr) { + t.Fatalf("expect %s to be contained in %v", c.expectErr, err) + } + } else if c.expectErr != "" { + t.Fatalf("expect error %s, got none", c.expectErr) + } + + if e, a := c.expectCreated, capture.count(); e != a { + t.Fatalf("expected DownloadDirectory to create %d file(s) under %s, captured %d", e, dstPath, a) + } + if open := capture.stillOpen(); len(open) != 0 { + t.Fatalf("expected all created files under %s to be closed after DownloadDirectory returned, but these remain open: %v", dstPath, open) + } + }) + } +} diff --git a/feature/s3/transfermanager/upload_directory_test.go b/feature/s3/transfermanager/upload_directory_test.go index d06f6229ed48..d459ed2abb23 100644 --- a/feature/s3/transfermanager/upload_directory_test.go +++ b/feature/s3/transfermanager/upload_directory_test.go @@ -9,6 +9,7 @@ import ( "runtime" "sort" "strings" + "sync" "testing" "github.com/aws/aws-sdk-go-v2/internal/awstesting" @@ -645,3 +646,156 @@ func TestUploadDirectoryWithContextCanceled(t *testing.T) { t.Errorf("expected error message to contain %q, but did not %q", e, a) } } + +// openFileCapture records the *os.File bodies that UploadDirectory opens. It +// implements PutRequestCallback, which the directory uploader invokes for every +// object with input.Body still set to the raw *os.File it opened (before that +// body is read/replaced for the actual PutObject call). This lets a black-box +// test observe the exact set of file handles opened and, after UploadDirectory +// returns, assert that each one was closed - a regression guard for +// aws/aws-sdk-go-v2#3512. It is safe for concurrent use by the upload workers. +type openFileCapture struct { + mu sync.Mutex + files []*os.File +} + +// UpdateRequest implements PutRequestCallback, capturing the opened *os.File. +func (c *openFileCapture) UpdateRequest(in *UploadObjectInput) { + if f, ok := in.Body.(*os.File); ok { + c.mu.Lock() + c.files = append(c.files, f) + c.mu.Unlock() + } +} + +func (c *openFileCapture) count() int { + return len(c.files) +} + +// stillOpen returns the names of captured files that are still open. A closed +// *os.File returns an error from Stat, so a nil error means the handle leaked. +func (c *openFileCapture) stillOpen() []string { + var open []string + for _, f := range c.files { + if _, err := f.Stat(); err == nil { + open = append(open, f.Name()) + } + } + sort.Strings(open) + return open +} + +// TestUploadDirectoryClosesOpenedFiles verifies that UploadDirectory closes +// every file handle it opens before returning, across recursive/non-recursive +// walks and even when individual object uploads fail. Regression test for +// aws/aws-sdk-go-v2#3512. +func TestUploadDirectoryClosesOpenedFiles(t *testing.T) { + _, filename, _, _ := runtime.Caller(0) + root := filepath.Join(filepath.Dir(filename), "testdata") + + cases := map[string]struct { + source string + recursive bool + followSymLinks bool + keyPrefix string + putobjectFunc func(*s3testing.TransferManagerLoggingClient, *s3.PutObjectInput) (*s3.PutObjectOutput, error) + preprocessFunc func(string) (func(), error) + failurePolicy UploadDirectoryFailurePolicy + expectFilesTraversed int + expectErr string + }{ + "single file recursively": { + source: filepath.Join(root, "single-file-dir"), + recursive: true, + expectFilesTraversed: 1, + }, + "multi file at root recursively": { + source: filepath.Join(root, "multi-file-at-root"), + recursive: true, + expectFilesTraversed: 3, + }, + "multi file with subdir recursively": { + source: filepath.Join(root, "multi-file-with-subdir"), + recursive: true, + expectFilesTraversed: 4, + }, + "multi file with subdir non-recursively": { + source: filepath.Join(root, "multi-file-with-subdir"), + expectFilesTraversed: 2, + }, + "multi file with subdir and symlink recursively": { + source: filepath.Join(root, "multi-file-contain-symlink"), + recursive: true, + followSymLinks: true, + preprocessFunc: func(string) (func(), error) { + symlinkPath1 := filepath.Join(root, "multi-file-contain-symlink", "to", "the", "symFoo") + symlinkPath2 := filepath.Join(root, "multi-file-contain-symlink", "to", "symBar") + postprocessFunc := func() { + os.Remove(symlinkPath1) + os.Remove(symlinkPath2) + } + if err := os.Symlink(filepath.Join(root, "dstFile1"), symlinkPath1); err != nil { + return postprocessFunc, err + } + if err := os.Symlink(filepath.Join(root, "dstDir1"), symlinkPath2); err != nil { + return postprocessFunc, err + } + return postprocessFunc, nil + }, + expectFilesTraversed: 6, + }, + "files are closed even when every upload fails and is ignored": { + source: filepath.Join(root, "multi-file-with-subdir"), + recursive: true, + putobjectFunc: func(*s3testing.TransferManagerLoggingClient, *s3.PutObjectInput) (*s3.PutObjectOutput, error) { + return nil, fmt.Errorf("banned key") + }, + expectFilesTraversed: 4, + failurePolicy: IgnoreUploadFailurePolicy{}, + }, + } + + for name, c := range cases { + t.Run(name, func(t *testing.T) { + capture := &openFileCapture{} + + s3Client, _ := s3testing.NewUploadDirectoryClient([]string{"UploadPart", "CompleteMultipartUpload"}) + s3Client.PutObjectFn = c.putobjectFunc + mgr := New(s3Client) + + if c.preprocessFunc != nil { + postprocessFunc, err := c.preprocessFunc(root) + defer postprocessFunc() + if err != nil { + t.Fatalf("error when preprocessing: %v", err) + } + } + + req := &UploadDirectoryInput{ + Bucket: aws.String("mock-bucket"), + Source: aws.String(c.source), + Recursive: aws.Bool(c.recursive), + FollowSymbolicLinks: aws.Bool(c.followSymLinks), + KeyPrefix: aws.String(c.keyPrefix), + Callback: capture, + FailurePolicy: c.failurePolicy, + } + + _, err := mgr.UploadDirectory(context.Background(), req) + if c.expectErr == "" { + if err != nil { + t.Fatalf("expect no error, got %v", err) + } + } else if err == nil || !strings.Contains(err.Error(), c.expectErr) { + t.Fatalf("expect error containing %q, got %v", c.expectErr, err) + } + + if e, a := c.expectFilesTraversed, capture.count(); e != a { + t.Fatalf("expected UploadDirectory to open %d file under %s, captured %d", e, c.source, a) + } + if open := capture.stillOpen(); len(open) != 0 { + t.Fatalf("expected all opened files under %s to be closed after UploadDirectory returned, but these remain open: %v", c.source, open) + } + }) + } +} From fece570b75d0472451233e4d6c36111993f56d91 Mon Sep 17 00:00:00 2001 From: wty-Bryant Date: Wed, 12 Aug 2026 18:07:21 -0400 Subject: [PATCH 2/3] add changelog --- .changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 .changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json diff --git a/.changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json b/.changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json new file mode 100644 index 000000000000..e6857d98885d --- /dev/null +++ b/.changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json @@ -0,0 +1,8 @@ +{ + "id": "ef7fcdde-5f2e-4043-b83f-0eaa64e9f031", + "type": "bugfix", + "description": "Close each file transferred during directory upload/download to avoid leaking file handles", + "modules": [ + "feature/s3/transfermanager" + ] +} \ No newline at end of file From 155fb41863068cb78c064a6dfcfe3969d61eccf5 Mon Sep 17 00:00:00 2001 From: wty-Bryant Date: Wed, 12 Aug 2026 18:22:22 -0400 Subject: [PATCH 3/3] update some syntax --- feature/s3/transfermanager/api_op_DownloadDirectory.go | 10 ++++------ feature/s3/transfermanager/api_op_UploadDirectory.go | 7 +++---- 2 files changed, 7 insertions(+), 10 deletions(-) diff --git a/feature/s3/transfermanager/api_op_DownloadDirectory.go b/feature/s3/transfermanager/api_op_DownloadDirectory.go index aaa9eb7dc965..b55d5683ee58 100644 --- a/feature/s3/transfermanager/api_op_DownloadDirectory.go +++ b/feature/s3/transfermanager/api_op_DownloadDirectory.go @@ -288,10 +288,9 @@ func (d *directoryDownloader) downloadSingleObject(ctx context.Context, data obj err = d.failurePolicy.OnDownloadFailed(d.in, input, err) if err != nil { return fmt.Errorf("error when heading info of object %s: %v", data.key, err) - } else { - d.objectsFailed.Add(1) - return nil } + d.objectsFailed.Add(1) + return nil } d.progressOnce.Do(func() { @@ -323,10 +322,9 @@ func (d *directoryDownloader) downloadSingleObject(ctx context.Context, data obj err = d.failurePolicy.OnDownloadFailed(d.in, input, err) if err != nil { return fmt.Errorf("error when getting object and writing to local file %s: %v", data.path, err) - } else { - d.objectsFailed.Add(1) - return nil } + d.objectsFailed.Add(1) + return nil } d.objectsDownloaded.Add(1) diff --git a/feature/s3/transfermanager/api_op_UploadDirectory.go b/feature/s3/transfermanager/api_op_UploadDirectory.go index 831267f091fd..c7db6da7ca58 100644 --- a/feature/s3/transfermanager/api_op_UploadDirectory.go +++ b/feature/s3/transfermanager/api_op_UploadDirectory.go @@ -410,11 +410,10 @@ func (u *directoryUploader) uploadSingleFile(ctx context.Context, data fileEntry err = u.failurePolicy.OnUploadFailed(u.in, input, err) if err != nil { return fmt.Errorf("error when uploading file %s: %v", data.path, err) - } else { - // this failed object is ignored, just increase the failure count - u.filesFailed.Add(1) - return nil } + // this failed object is ignored, just increase the failure count + u.filesFailed.Add(1) + return nil } u.progressOnce.Do(func() {