ensure http response body is closed (#1103)

* Ensure http response body is closed
  - When appropriate, consume all body data before close to allow connection reuse.

Author: gammazero

bitswap/network/httpnet/httpnet.go

@@ -624,6 +624,7 @@ func (ht *Network) connectToURL(ctx context.Context, p peer.ID, u network.Parsed
 	// FIXME: Storacha returns 410 for our probe.
 	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusGone {
 		log.Debugf("connect/ping request to %s %s succeeded: %d", p, req.URL, resp.StatusCode)
+		io.Copy(io.Discard, resp.Body) // read all body data so that connection can be reused
 		return resp.StatusCode, nil
 	}
 

filestore/fsrefstore.go

@@ -2,6 +2,7 @@ package filestore
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -249,6 +250,8 @@ func (f *FileManager) readURLDataObj(ctx context.Context, m mh.Multihash, d *pb.
 	if err != nil {
 		return nil, &CorruptReferenceError{StatusFileError, err}
 	}
+	defer res.Body.Close()
+
 	if res.StatusCode != http.StatusOK && res.StatusCode != http.StatusPartialContent {
 		return nil, &CorruptReferenceError{
 			StatusFileError,
@@ -258,12 +261,12 @@ func (f *FileManager) readURLDataObj(ctx context.Context, m mh.Multihash, d *pb.
 
 	outbuf := make([]byte, d.GetSize())
 	_, err = io.ReadFull(res.Body, outbuf)
-	if err == io.EOF || err == io.ErrUnexpectedEOF {
-		return nil, &CorruptReferenceError{StatusFileChanged, err}
-	} else if err != nil {
+	if err != nil {
+		if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+			return nil, &CorruptReferenceError{StatusFileChanged, err}
+		}
 		return nil, &CorruptReferenceError{StatusFileError, err}
 	}
-	res.Body.Close()
 
 	// Work with CIDs for this, as they are a nice wrapper and things
 	// will not break if multihashes underlying types change.

gateway/backend_car_fetcher.go

@@ -64,6 +64,8 @@ func (ps *remoteCarFetcher) Fetch(ctx context.Context, path path.ImmutablePath,
 	if err != nil {
 		return err
 	}
+	defer resp.Body.Close()
+	defer io.Copy(io.Discard, resp.Body) // read all body data so that connection can be reused
 
 	if resp.StatusCode != http.StatusOK {
 		errData, err := io.ReadAll(resp.Body)
@@ -75,12 +77,7 @@ func (ps *remoteCarFetcher) Fetch(ctx context.Context, path path.ImmutablePath,
 		return fmt.Errorf("http error from car gateway: %s: %w", resp.Status, err)
 	}
 
-	err = cb(path, resp.Body)
-	if err != nil {
-		resp.Body.Close()
-		return err
-	}
-	return resp.Body.Close()
+	return cb(path, resp.Body)
 }
 
 func (ps *remoteCarFetcher) getRandomGatewayURL() string {

routing/http/client/client.go

@@ -667,6 +667,7 @@ func (c *Client) GetClosestPeers(ctx context.Context, key cid.Cid) (peers iter.R
 	var skipBodyClose bool
 	defer func() {
 		if !skipBodyClose {
+			io.Copy(io.Discard, resp.Body) // Drain body for connection reuse
 			resp.Body.Close()
 		}
 	}()