Repository: aws/aws-sdk-go (go)

Issue:
S3 `Downloader` potentially downloads corrupted object when the it is split into multiple parts
### Describe the bug

* We have a system that's constantly updating an object stored in an S3 bucket
* The object is downloaded by multiple clients concurrently
* In our particular case, the stored object is gzipped but the issue applies to all sorts of files
* When the object grows to over 5MB (which causes the downloader to download the object in multiple parts), we encountered an issue where the second part is downloaded from a different version of the object, therefore corrupting the download


### Expected Behavior

The latest version of the object is downloaded

### Current Behavior

An error is encountered from time to time, e.g.,

```
2023/09/12 14:54:39
panic: flate: corrupt input before offset 1089400

goroutine 1 [running]:
main.fetch(0x14000146640)
        /Users/kevinqiu/src/tmp/s3bugrepro/main.go:36 +0x2a0
main.main()
        /Users/kevinqiu/src/tmp/s3bugrepro/main.go:57 +0xd0
exit status 2
```

or
```
panic: gzip: invalid checksum
```

With logging turned on, it's observed that when a later part is being downloaded and when the object is updated before that, the later chunk from a different version is downloaded and therefore corrupting the output.



### Reproduction Steps

Minimally reproducible example:
### Producer
The producer is simply a script that uploads a gzipped file (greater than 5MB) to a bucket constantly
```bash
#! /bin/bash
while true; do
    files="0 1 2 3"
    for i in $files; do
        aws s3 cp $i s3://$BUCKET/TEST
        sleep 1
    done
done
```

### Consumer
```go
package main

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
)

func fetch(downloader *s3manager.Downloader) {
	buff := &aws.WriteAtBuffer{}
	goi := &s3.GetObjectInput{
		Bucket: aws.String("BUCKET"),  // replace with the real bucket name
		Key:    aws.String("TEST"),
	}
	_, err := downloader.Download(buff, goi)
	if err != nil {
		panic(err)
	}

	b := buff.Bytes()
	br := bytes.NewReader(b)
	gzr, err := gzip.NewReader(br)
	if err != nil {
		panic(err)
	}

	uzb, err := io.ReadAll(gzr)
	if err != nil && err != io.EOF {
		panic(err)
	}

	fmt.Printf("Size=%v\n", len(uzb))
}

func main() {
	s, err := session.NewSession(&aws.Config{
		Region:   aws.String("us-east-1"),
		LogLevel: aws.LogLevel(aws.LogDebugWithHTTPBody),
	})

	if err != nil {
		panic(err)
	}

	downloader := s3manager.NewDownloader(s, func(downloader *s3manager.Downloader) {
		downloader.PartSize = 1024 * 512  // set to a small chunk size so the problem can be reproduced sooner
	})

	for {
		fetch(downloader)
		time.Sleep(2 * time.Second)
	}
}
```

### Possible Solution

When `GetObjectInput.versionId` is not provided by the user (which means getting the latest object version), send a request to first figure out the latest version of the object, and then [set the `versionId` in the subsequent `downloadChunk` method](https://github.com/aws/aws-sdk-go/blob/main/service/s3/s3manager/download.go#L421).


### Additional Information/Context

_No response_

### SDK version used

1.44

### Environment details (Version of Go (`go version`)? OS name and version, etc.)

1.20

Pull request description:
Feat GetObject Version Control
This pr fixes #4986, which is a port from the [fix](https://github.com/aws/aws-sdk-go-v2/pull/3050) on go v2


Patch:
diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
index 09d05635e..ea5f741be 100644
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@@ -5,3 +5,5 @@
 ### SDK Bugs
 * Fix improper use of printf-style functions.
   * Required for Go 1.24.
+* `service/s3/s3manager`: Abort multipart download if object is modified during download
+  * Fixes [4986](https://github.com/aws/aws-sdk-go/issues/4986)
diff --git a/service/s3/s3manager/download.go b/service/s3/s3manager/download.go
index 37eb7dd52..1c0473181 100644
--- a/service/s3/s3manager/download.go
+++ b/service/s3/s3manager/download.go
@@ -290,13 +290,15 @@ type downloader struct {
 	in *s3.GetObjectInput
 	w  io.WriterAt
 
-	wg sync.WaitGroup
-	m  sync.Mutex
+	wg   sync.WaitGroup
+	m    sync.Mutex
+	once sync.Once
 
 	pos        int64
 	totalBytes int64
 	written    int64
 	err        error
+	etag       string
 
 	partBodyMaxRetries int
 }
@@ -424,6 +426,9 @@ func (d *downloader) downloadChunk(chunk dlchunk) error {
 
 	// Get the next byte range of data
 	in.Range = aws.String(chunk.ByteRange())
+	if in.VersionId == nil && d.etag != "" {
+		in.IfMatch = aws.String(d.etag)
+	}
 
 	var n int64
 	var err error
@@ -466,6 +471,9 @@ func (d *downloader) tryDownloadChunk(in *s3.GetObjectInput, w io.Writer) (int64
 		return 0, err
 	}
 	d.setTotalBytes(resp) // Set total if not yet set.
+	d.once.Do(func() {
+		d.etag = aws.StringValue(resp.ETag)
+	})
 
 	var src io.Reader = resp.Body
 	if d.cfg.BufferProvider != nil {


Test patch:
diff --git a/service/s3/s3manager/download_test.go b/service/s3/s3manager/download_test.go
index 00ff8d785..b9060d669 100644
--- a/service/s3/s3manager/download_test.go
+++ b/service/s3/s3manager/download_test.go
@@ -30,6 +30,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 )
 
+var etag string = "myetag"
+
 func dlLoggingSvc(data []byte) (*s3.S3, *[]string, *[]string) {
 	var m sync.Mutex
 	names := []string{}
@@ -203,6 +205,93 @@ func dlLoggingSvcWithErrReader(cases []testErrReader) (*s3.S3, *[]string) {
 	return svc, &names
 }
 
+func dlLoggingSvcWithVersions(data []byte) (*s3.S3, *[]string, *[]string, *[]string) {
+	var m sync.Mutex
+	versions := []string{}
+	etags := []string{}
+	names := []string{}
+
+	svc := s3.New(unit.Session)
+	svc.Handlers.Send.Clear()
+	svc.Handlers.Send.PushBack(func(r *request.Request) {
+		m.Lock()
+		defer m.Unlock()
+
+		names = append(names, r.Operation.Name)
+		versions = append(versions, aws.StringValue(r.Params.(*s3.GetObjectInput).VersionId))
+		etags = append(etags, aws.StringValue(r.Params.(*s3.GetObjectInput).IfMatch))
+
+		rerng := regexp.MustCompile(`bytes=(\d+)-(\d+)`)
+		rng := rerng.FindStringSubmatch(r.HTTPRequest.Header.Get("Range"))
+		start, _ := strconv.ParseInt(rng[1], 10, 64)
+		fin, _ := strconv.ParseInt(rng[2], 10, 64)
+		fin++
+
+		if fin > int64(len(data)) {
+			fin = int64(len(data))
+		}
+
+		bodyBytes := data[start:fin]
+		r.HTTPResponse = &http.Response{
+			StatusCode: 200,
+			Body:       ioutil.NopCloser(bytes.NewReader(bodyBytes)),
+			Header:     http.Header{},
+		}
+		r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d",
+			start, fin-1, len(data)))
+		r.HTTPResponse.Header.Set("Content-Length", fmt.Sprintf("%d", len(bodyBytes)))
+		r.HTTPResponse.Header.Set("Etag", etag)
+	})
+
+	return svc, &names, &versions, &etags
+}
+
+func dlLoggingSvcWithVersionMismatch(t *testing.T) *s3.S3 {
+	var m sync.Mutex
+	reqCount := int64(0)
+	body := bytes.NewReader(make([]byte, s3manager.DefaultDownloadPartSize))
+	svc := s3.New(unit.Session)
+	svc.Handlers.Send.Clear()
+	svc.Handlers.Send.PushBack(func(r *request.Request) {
+		m.Lock()
+		defer m.Unlock()
+
+		statusCode := http.StatusOK
+		var eTag *string
+		switch atomic.LoadInt64(&reqCount) {
+		case 0:
+			if a := r.Params.(*s3.GetObjectInput).IfMatch; a != nil {
+				t.Errorf("expect no Etag in first request, got %s", aws.StringValue(a))
+				statusCode = http.StatusBadRequest
+			} else {
+				eTag = aws.String(etag)
+			}
+		case 1:
+			// Give a chance for the multipart chunks to be queued up
+			time.Sleep(1 * time.Second)
+			// mock the precondition error when object is synchronously updated
+			statusCode = http.StatusPreconditionFailed
+		default:
+			if a := aws.StringValue(r.Params.(*s3.GetObjectInput).IfMatch); a != etag {
+				t.Errorf("expect subrequests' IfMatch header to be %s, got %s", etag, a)
+				statusCode = http.StatusBadRequest
+			}
+		}
+		r.HTTPResponse = &http.Response{
+			StatusCode: statusCode,
+			Body:       ioutil.NopCloser(body),
+			Header:     http.Header{},
+		}
+		r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes 0-%d/%d",
+			body.Len()-1, body.Len()*10))
+		r.HTTPResponse.Header.Set("Content-Length", fmt.Sprintf("%d", body.Len()))
+		r.HTTPResponse.Header.Set("Etag", aws.StringValue(eTag))
+		atomic.AddInt64(&reqCount, 1)
+	})
+
+	return svc
+}
+
 func TestDownloadOrder(t *testing.T) {
 	s, names, ranges := dlLoggingSvc(buf12MB)
 
@@ -526,6 +615,92 @@ func TestDownloadPartBodyRetry_FailRetry(t *testing.T) {
 	}
 }
 
+func TestDownloadWithVersionID(t *testing.T) {
+	s, names, versions, etags := dlLoggingSvcWithVersions(buf12MB)
+	d := s3manager.NewDownloaderWithClient(s)
+
+	w := &aws.WriteAtBuffer{}
+	n, err := d.Download(w, &s3.GetObjectInput{
+		Bucket:    aws.String("bucket"),
+		Key:       aws.String("key"),
+		VersionId: aws.String("vid"),
+	})
+
+	if err != nil {
+		t.Fatalf("expect no error, got %v", err)
+	}
+
+	if e, a := int64(len(buf12MB)), n; e != a {
+		t.Errorf("expect %d buffer length, got %d", e, a)
+	}
+
+	expectCalls := []string{"GetObject", "GetObject", "GetObject"}
+	if e, a := expectCalls, *names; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v API calls, got %v", e, a)
+	}
+
+	expectVersions := []string{"vid", "vid", "vid"}
+	if e, a := expectVersions, *versions; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v version ids, got %v", e, a)
+	}
+
+	expectETags := []string{"", "", ""}
+	if e, a := expectETags, *etags; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v etags, got %v", e, a)
+	}
+}
+
+func TestDownloadWithETags(t *testing.T) {
+	s, names, versions, etags := dlLoggingSvcWithVersions(buf12MB)
+	d := s3manager.NewDownloaderWithClient(s)
+
+	w := &aws.WriteAtBuffer{}
+	n, err := d.Download(w, &s3.GetObjectInput{
+		Bucket: aws.String("bucket"),
+		Key:    aws.String("key"),
+	})
+
+	if err != nil {
+		t.Fatalf("expect no error, got %v", err)
+	}
+
+	if e, a := int64(len(buf12MB)), n; e != a {
+		t.Errorf("expect %d buffer length, got %d", e, a)
+	}
+
+	expectCalls := []string{"GetObject", "GetObject", "GetObject"}
+	if e, a := expectCalls, *names; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v API calls, got %v", e, a)
+	}
+
+	expectVersions := []string{"", "", ""}
+	if e, a := expectVersions, *versions; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v version ids, got %v", e, a)
+	}
+
+	expectETags := []string{"", etag, etag}
+	if e, a := expectETags, *etags; !reflect.DeepEqual(e, a) {
+		t.Errorf("expect %v etags, got %v", e, a)
+	}
+}
+
+func TestDownloadWithVersionMismatch(t *testing.T) {
+	s := dlLoggingSvcWithVersionMismatch(t)
+	d := s3manager.NewDownloaderWithClient(s)
+
+	w := &aws.WriteAtBuffer{}
+	_, err := d.Download(w, &s3.GetObjectInput{
+		Bucket: aws.String("bucket"),
+		Key:    aws.String("key"),
+	})
+
+	if err == nil {
+		t.Fatalf("expect error, got none")
+	} else if e, a := "PreconditionFailed", err.Error(); !strings.Contains(a, e) {
+		t.Fatalf("expect error message to contain %s, but did not %s", e, a)
+	}
+}
+
 func TestDownloadWithContextCanceled(t *testing.T) {
 	d := s3manager.NewDownloader(unit.Session)