s3/transfermanager: fix panic in concurrentReader when len(p) is smaller than cap(p)

.changelog/891e26459bd94f7b9c5da4571ef3876e.json

@@ -0,0 +1,8 @@
+{
+    "id": "891e2645-9bd9-4f7b-9c5d-a4571ef3876e",
+    "type": "bugfix",
+    "description": "Fix the concurrent reader panic when buffer len is smaller than capacity",
+    "modules": [
+        "feature/s3/transfermanager"
+    ]
+}

feature/s3/transfermanager/concurrent_reader.go

@@ -200,15 +200,15 @@ type outChunk struct {
 }
 
 func (r *concurrentReader) read(p []byte) (int, error) {
-	if cap(p) == 0 {
+	if len(p) == 0 {
 		return 0, nil
 	}
 
 	var written int
 
 	partSize := r.partSize
 	minIndex := int32(r.written / partSize)
-	maxIndex := min(int32((r.written+int64(cap(p))-1)/partSize), atomic.LoadInt32(&r.capacity)-1)
+	maxIndex := min(int32((r.written+int64(len(p))-1)/partSize), atomic.LoadInt32(&r.capacity)-1)
 	for i := minIndex; i <= maxIndex; i++ {
 		if e := r.getErr(); e != nil && e != io.EOF {
 			r.clean()
@@ -255,7 +255,7 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 
 		index := r.partSize*int64(oc.index) - r.written
 
-		if index < int64(cap(p)) {
+		if index < int64(len(p)) {
 			n, err := oc.body.Read(p[index:])
 			oc.cur += int64(n)
 			written += n

feature/s3/transfermanager/concurrent_reader_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+
 	"io"
 	"math"
 	"math/rand"
@@ -15,6 +16,58 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 )
 
+func TestConcurrentReaderReadUsesSliceLenForBounds(t *testing.T) {
+	tests := map[string]func(*concurrentReader){
+		"buffered chunk": func(r *concurrentReader) {
+			r.capacity = 2
+			r.receiveCount = 2
+			r.buf[1] = &outChunk{
+				body:   bytes.NewReader([]byte("chunk")),
+				index:  1,
+				length: int64(len("chunk")),
+			}
+		},
+		"received chunk": func(r *concurrentReader) {
+			r.capacity = 2
+			r.receiveCount = 1
+			r.ch <- outChunk{
+				body:   bytes.NewReader([]byte("chunk")),
+				index:  1,
+				length: int64(len("chunk")),
+			}
+		},
+	}
+
+	for name, setup := range tests {
+		t.Run(name, func(t *testing.T) {
+			r := &concurrentReader{
+				partSize:   8,
+				partsCount: 2,
+				buf:        make(map[int32]*outChunk),
+				ch:         make(chan outChunk, 1),
+			}
+			setup(r)
+
+			p := make([]byte, 4, 9)
+			n, err := r.read(p)
+			if err != nil {
+				t.Fatalf("expect no error, got %v", err)
+			}
+			if n != 0 {
+				t.Fatalf("expect no bytes read into short slice, got %d", n)
+			}
+
+			chunk, ok := r.buf[1]
+			if !ok {
+				t.Fatal("expect chunk to remain buffered")
+			}
+			if chunk.cur != 0 {
+				t.Fatalf("expect chunk cursor to remain unchanged, got %d", chunk.cur)
+			}
+		})
+	}
+}
+
 func TestConcurrentReader(t *testing.T) {
 	cases := map[string]struct {
 		partSize     int64