diff --git a/.changelog/891e26459bd94f7b9c5da4571ef3876e.json b/.changelog/891e26459bd94f7b9c5da4571ef3876e.json new file mode 100644 index 000000000000..e09c1ee24bae --- /dev/null +++ b/.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" + ] +} \ No newline at end of file diff --git a/feature/s3/transfermanager/concurrent_reader.go b/feature/s3/transfermanager/concurrent_reader.go index 244477e9d37a..9ba7d19900ed 100644 --- a/feature/s3/transfermanager/concurrent_reader.go +++ b/feature/s3/transfermanager/concurrent_reader.go @@ -200,7 +200,7 @@ type outChunk struct { } func (r *concurrentReader) read(p []byte) (int, error) { - if cap(p) == 0 { + if len(p) == 0 { return 0, nil } @@ -208,7 +208,7 @@ func (r *concurrentReader) read(p []byte) (int, error) { 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 diff --git a/feature/s3/transfermanager/concurrent_reader_test.go b/feature/s3/transfermanager/concurrent_reader_test.go index 1fb44c42e32b..4302531387c9 100644 --- a/feature/s3/transfermanager/concurrent_reader_test.go +++ b/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