Skip to content

Close each file after its transfer finished during directory upload/download #3514

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wty-Bryant wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Close each file after its transfer finished during directory upload/download #3514

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6963ab8
close files after transfer finish for keyprefix transfer
wty-Bryant Aug 12, 2026
fece570
add changelog
wty-Bryant Aug 12, 2026
155fb41
update some syntax
wty-Bryant Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions .changelog/ef7fcdde5f2e4043b83f0eaa64e9f031.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
]
}
100 changes: 61 additions & 39 deletions feature/s3/transfermanager/api_op_DownloadDirectory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -262,54 +269,67 @@ 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)
}
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)
Comment on lines +310 to +318

@Tongs2000 Tongs2000 Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Overall this looks great. One small corner case I wanted to confirm. In downloadSingleObject Since the defer runs before the caller's setErr, and the new setErr keeps the first non-cancellation error, if io.Copy fails and Close() also fails on the same object, the surfaced error will be the close error rather than the copy error. Which do we actually want to report? Same shape exists in uploadSingleFile.

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)
}

d.objectsDownloaded.Add(1)
d.emitter.ObjectsTransferred(ctx, n)
d.objectsFailed.Add(1)
return nil
}

d.objectsDownloaded.Add(1)
d.emitter.ObjectsTransferred(ctx, n)
return nil
}

func (d *directoryDownloader) freshContext(ctx context.Context) (context.Context, context.CancelFunc) {
Expand All @@ -323,7 +343,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 {
Expand Down
72 changes: 45 additions & 27 deletions feature/s3/transfermanager/api_op_UploadDirectory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -375,37 +380,48 @@ 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)
}

u.progressOnce.Do(func() {
u.emitter.Start(ctx, u.in)
})
u.filesUploaded.Add(1)
u.emitter.ObjectsTransferred(ctx, aws.ToInt64(out.ContentLength))
// 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))
return nil
}

func (u *directoryUploader) freshContext(ctx context.Context) (context.Context, context.CancelFunc) {
Expand All @@ -419,7 +435,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 {
Expand Down
Loading
Loading