Adhere to request.Context when roundtripping on a stream (#2796)

* Adhere to request.Context when roundtripping on a stream

This allows HTTP request timeout and cancellation to work.

Fixes #2533

* Attempt to cancel the read by setting the read deadline

* Don't spawn a goroutine in read

---------

Co-authored-by: Marco Munizaga <git@marcopolo.io>

p2p/http/libp2phttp.go

@@ -434,9 +434,13 @@ func (rt *streamRoundTripper) RoundTrip(r *http.Request) (*http.Response, error)
 		}
 	}()
 
-	// TODO: Adhere to the request.Context
+	if deadline, ok := r.Context().Deadline(); ok {
+		s.SetReadDeadline(deadline)
+	}
+
 	resp, err := http.ReadResponse(bufio.NewReader(s), r)
 	if err != nil {
+		s.Close()
 		return nil, err
 	}
 	resp.Body = &streamReadCloser{resp.Body, s}

p2p/http/libp2phttp_test.go

@@ -15,6 +15,8 @@ import (
 	"math/big"
 	"net"
 	"net/http"
+	"net/url"
+	"os"
 	"reflect"
 	"strings"
 	"testing"
@@ -108,6 +110,63 @@ func TestHTTPOverStreamsSendsConnectionClose(t *testing.T) {
 	}
 }
 
+func TestHTTPOverStreamsContextAndClientTimeout(t *testing.T) {
+	const clientTimeout = 200 * time.Millisecond
+
+	serverHost, err := libp2p.New(
+		libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/0/quic-v1"),
+	)
+	require.NoError(t, err)
+
+	httpHost := libp2phttp.Host{StreamHost: serverHost}
+	httpHost.SetHTTPHandler("/hello/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(2 * clientTimeout)
+		w.Write([]byte("hello"))
+	}))
+
+	// Start server
+	go httpHost.Serve()
+	defer httpHost.Close()
+
+	// Start client
+	clientHost, err := libp2p.New(libp2p.NoListenAddrs)
+	require.NoError(t, err)
+	clientHost.Connect(context.Background(), peer.AddrInfo{
+		ID:    serverHost.ID(),
+		Addrs: serverHost.Addrs(),
+	})
+
+	clientRT, err := (&libp2phttp.Host{StreamHost: clientHost}).NewConstrainedRoundTripper(peer.AddrInfo{ID: serverHost.ID()})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "/hello/", nil)
+	require.NoError(t, err)
+
+	client := &http.Client{Transport: clientRT}
+	_, err = client.Do(req)
+	require.Error(t, err)
+	require.ErrorIs(t, err, os.ErrDeadlineExceeded)
+	t.Log("OK, deadline exceeded waiting for response as expected")
+
+	// Make another request, this time using http.Client.Timeout.
+	clientRT, err = (&libp2phttp.Host{StreamHost: clientHost}).NewConstrainedRoundTripper(peer.AddrInfo{ID: serverHost.ID()})
+	require.NoError(t, err)
+
+	client = &http.Client{
+		Transport: clientRT,
+		Timeout:   clientTimeout,
+	}
+
+	_, err = client.Get("/hello/")
+	require.Error(t, err)
+	var uerr *url.Error
+	require.ErrorAs(t, err, &uerr)
+	require.True(t, uerr.Timeout())
+	t.Log("OK, timed out waiting for response as expected")
+}
+
 func TestHTTPOverStreamsReturnsConnectionClose(t *testing.T) {
 	serverHost, err := libp2p.New(
 		libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/0/quic-v1"),