fix(storage): close HTTP2 response bodies to prevent flow-control leaks (#14324)

Fixes an issue where HTTP2 stream connections might exhaust flow-control
quota and cause INTERNAL_ERROR in long-lived connections.

- Modifies `httpReader.Read` to properly close the previous stream body
before reopening it.
- Adds a defer block in `parseReadResponse` to ensure response bodies
aren't leaked when an error is returned during header parsing.
- Ensures bodies are closed when stream discarding (`io.CopyN`) or
`X-Goog-Generation` header parsing fails in `readerReopen`. These fixes
apply equally to both the JSON and XML reader code paths.

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: krishnamd-jkp <230955344+krishnamd-jkp@users.noreply.github.com>

Author: krishnamd-jkp

storage/http_client.go

@@ -1424,12 +1424,16 @@ func (r *httpReader) Read(p []byte) (int, error) {
 		// Read failed (likely due to connection issues), but we will try to reopen
 		// the pipe and continue. Send a ranged read request that takes into account
 		// the number of bytes we've already seen.
+
+		// Close the current body before retrying. Otherwise we leave a
+		// connection open and the retry might fail due to quota issues.
+		r.body.Close()
+
 		res, err := r.reopen(r.seen)
 		if err != nil {
 			// reopen already retries
 			return n, err
 		}
-		r.body.Close()
 		r.body = res.Body
 	}
 	return n, nil
@@ -1523,6 +1527,7 @@ func readerReopen(ctx context.Context, header http.Header, params *newRangeReade
 			if params.gen < 0 && res.Header.Get("X-Goog-Generation") != "" {
 				gen64, err := strconv.ParseInt(res.Header.Get("X-Goog-Generation"), 10, 64)
 				if err != nil {
+					res.Body.Close()
 					return err
 				}
 				params.gen = gen64
@@ -1536,8 +1541,12 @@ func readerReopen(ctx context.Context, header http.Header, params *newRangeReade
 	}
 }
 
-func parseReadResponse(res *http.Response, params *newRangeReaderParams, reopen func(int64) (*http.Response, error)) (*Reader, error) {
-	var err error
+func parseReadResponse(res *http.Response, params *newRangeReaderParams, reopen func(int64) (*http.Response, error)) (r *Reader, err error) {
+	defer func() {
+		if err != nil {
+			res.Body.Close()
+		}
+	}()
 	var (
 		size        int64 // total size of object, even if a range was requested.
 		checkCRC    bool
@@ -1547,20 +1556,23 @@ func parseReadResponse(res *http.Response, params *newRangeReaderParams, reopen
 	if res.StatusCode == http.StatusPartialContent {
 		cr := strings.TrimSpace(res.Header.Get("Content-Range"))
 		if !strings.HasPrefix(cr, "bytes ") || !strings.Contains(cr, "/") {
-			return nil, fmt.Errorf("storage: invalid Content-Range %q", cr)
+			err = fmt.Errorf("storage: invalid Content-Range %q", cr)
+			return nil, err
 		}
 		// 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)
+			err = fmt.Errorf("storage: invalid Content-Range %q: %w", cr, err)
+			return nil, err
 		}
 
 		dashIndex := strings.Index(cr, "-")
 		if dashIndex >= 0 {
 			startOffset, err = strconv.ParseInt(cr[len("bytes="):dashIndex], 10, 64)
 			if err != nil {
-				return nil, fmt.Errorf("storage: invalid Content-Range %q: %w", cr, err)
+				err = fmt.Errorf("storage: invalid Content-Range %q: %w", cr, err)
+				return nil, err
 			}
 		}
 	} else {