chore(storage): fix gRPC reader size/remain logic (#4662)

Correctly sets the size to the size of the whole object and
remain to the number of bytes to be read. Adds checks for
Size and Remain to the integration tests.

I used "chore" because this is still experimental, non-exported
code.

storage/integration_test.go

@@ -793,13 +793,19 @@ func TestIntegration_ObjectReadGRPC(t *testing.T) {
 
 	obj := gc.Bucket(grpcBucketName).Object(name)
 
-	// Using a negative length to indicate reading to the end.
-	r, err := obj.NewRangeReader(ctx, 0, -1)
+	r, err := obj.NewReader(ctx)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer r.Close()
 
+	if size := r.Size(); size != int64(len(content)) {
+		t.Errorf("got size = %v, want %v", size, len(content))
+	}
+	if rem := r.Remain(); rem != int64(len(content)) {
+		t.Errorf("got %v bytes remaining, want %v", rem, len(content))
+	}
+
 	b := new(bytes.Buffer)
 	b.Grow(len(content))
 
@@ -816,6 +822,10 @@ func TestIntegration_ObjectReadGRPC(t *testing.T) {
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Errorf("got(-),want(+):\n%s", diff)
 	}
+
+	if rem := r.Remain(); rem != 0 {
+		t.Errorf("got %v bytes remaining, want 0", rem)
+	}
 }
 
 func TestIntegration_ObjectReadChunksGRPC(t *testing.T) {
@@ -850,6 +860,13 @@ func TestIntegration_ObjectReadChunksGRPC(t *testing.T) {
 	}
 	defer r.Close()
 
+	if size := r.Size(); size != int64(len(content)) {
+		t.Errorf("got size = %v, want %v", size, len(content))
+	}
+	if rem := r.Remain(); rem != int64(len(content)) {
+		t.Errorf("got %v bytes remaining, want %v", rem, len(content))
+	}
+
 	bufSize := len(content)
 	buf := make([]byte, bufSize)
 
@@ -868,6 +885,10 @@ func TestIntegration_ObjectReadChunksGRPC(t *testing.T) {
 		offset += n
 	}
 
+	if rem := r.Remain(); rem != 0 {
+		t.Errorf("got %v bytes remaining, want 0", rem)
+	}
+
 	// TODO: Verify content with the checksums.
 }
 
@@ -905,6 +926,13 @@ func TestIntegration_ObjectReadRelativeToEndGRPC(t *testing.T) {
 	}
 	defer r.Close()
 
+	if size := r.Size(); size != int64(len(content)) {
+		t.Errorf("got size = %v, want %v", size, len(content))
+	}
+	if rem := r.Remain(); rem != int64(offset) {
+		t.Errorf("got %v bytes remaining, want %v", rem, offset)
+	}
+
 	b := new(bytes.Buffer)
 	b.Grow(offset)
 
@@ -921,6 +949,10 @@ func TestIntegration_ObjectReadRelativeToEndGRPC(t *testing.T) {
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Errorf("got(-),want(+):\n%s", diff)
 	}
+
+	if rem := r.Remain(); rem != 0 {
+		t.Errorf("got %v bytes remaining, want 0", rem)
+	}
 }
 
 func TestIntegration_ObjectReadPartialContentGRPC(t *testing.T) {

storage/reader.go

@@ -233,6 +233,8 @@ func (o *ObjectHandle) NewRangeReader(ctx context.Context, offset, length int64)
 		if !strings.HasPrefix(cr, "bytes ") || !strings.Contains(cr, "/") {
 			return nil, fmt.Errorf("storage: invalid Content-Range %q", cr)
 		}
+		// Content range is formatted <first byte>-<last byte>/<total size>. We take
+		// the total size.
 		size, err = strconv.ParseInt(cr[strings.LastIndex(cr, "/")+1:], 10, 64)
 		if err != nil {
 			return nil, fmt.Errorf("storage: invalid Content-Range %q", cr)
@@ -514,9 +516,8 @@ func (o *ObjectHandle) newRangeReaderWithGRPC(ctx context.Context, offset, lengt
 	// object metadata.
 	msg := res.response
 	obj := msg.GetMetadata()
-	// This is the size of the content the Reader will convey. It can be the
-	// entire object, or just the size of the request range.
-	size := msg.GetContentRange().GetCompleteLength()
+	// This is the size of the entire object, even if only a range was requested.
+	size := obj.GetSize()
 
 	r.Attrs = ReaderObjectAttrs{
 		Size:            size,
@@ -527,9 +528,16 @@ func (o *ObjectHandle) newRangeReaderWithGRPC(ctx context.Context, offset, lengt
 		Metageneration:  obj.GetMetageneration(),
 		Generation:      obj.GetGeneration(),
 	}
-	if cr := msg.GetContentRange(); cr != nil {
+
+	r.size = size
+	cr := msg.GetContentRange()
+	if cr != nil {
 		r.Attrs.StartOffset = cr.GetStart()
+		r.remain = cr.GetEnd() - cr.GetStart() + 1
+	} else {
+		r.remain = size
 	}
+
 	// Only support checksums when reading an entire object, not a range.
 	if msg.GetObjectChecksums().Crc32C != nil && offset == 0 && length == 0 {
 		r.wantCRC = msg.GetObjectChecksums().GetCrc32C()
@@ -539,8 +547,6 @@ func (o *ObjectHandle) newRangeReaderWithGRPC(ctx context.Context, offset, lengt
 	// Store the content from the first Recv in the client buffer for reading
 	// later.
 	r.leftovers = msg.GetChecksummedData().GetContent()
-	r.remain = size
-	r.size = size
 
 	return r, nil
 }