Repository: libp2p/go-libp2p
Pull request #2756 (2024-04-03 03:11:12)

libp2phttp: Return connection: close when doing http over streams
This tells clients not to reuse the stream. They should make new requests on new streams. Context https://github.com/libp2p/specs/pull/508/files#diff-8e568763d7d4e15f9465300afa93c121c7f8e399acfff4c2aee7e70ed1199401R125


Patch:
diff --git a/p2p/http/libp2phttp.go b/p2p/http/libp2phttp.go
index fbd8a64f..5e45d978 100644
--- a/p2p/http/libp2phttp.go
+++ b/p2p/http/libp2phttp.go
@@ -304,7 +304,7 @@ func (h *Host) Serve() error {
 		h.httpTransport.listenAddrs = append(h.httpTransport.listenAddrs, h.StreamHost.Addrs()...)
 
 		go func() {
-			errCh <- http.Serve(listener, h.ServeMux)
+			errCh <- http.Serve(listener, connectionCloseHeaderMiddleware(h.ServeMux))
 		}()
 	}
 
@@ -422,6 +422,9 @@ func (rt *streamRoundTripper) RoundTrip(r *http.Request) (*http.Response, error)
 		return nil, err
 	}
 
+	// Write connection: close header to ensure the stream is closed after the response
+	r.Header.Add("connection", "close")
+
 	go func() {
 		defer s.CloseWrite()
 		r.Write(s)
@@ -816,3 +819,11 @@ func (h *Host) RemovePeerMetadata(server peer.ID) {
 	}
 	h.peerMetadata.Remove(server)
 }
+
+func connectionCloseHeaderMiddleware(next http.Handler) http.Handler {
+	// Sets connection: close. It's preferable to not reuse streams for HTTP.
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Connection", "close")
+		next.ServeHTTP(w, r)
+	})
+}
diff --git a/p2p/transport/quic/cmd/client/main.go b/p2p/transport/quic/cmd/client/main.go
index 5736d895..78736253 100644
--- a/p2p/transport/quic/cmd/client/main.go
+++ b/p2p/transport/quic/cmd/client/main.go
@@ -1,11 +1,20 @@
 package main
 
 import (
+	"context"
+	"crypto/rand"
 	"fmt"
+	"io"
 	"log"
 	"os"
 
-	cmdlib "github.com/libp2p/go-libp2p/p2p/transport/quic/cmd/lib"
+	ic "github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/core/peer"
+	libp2pquic "github.com/libp2p/go-libp2p/p2p/transport/quic"
+	"github.com/libp2p/go-libp2p/p2p/transport/quicreuse"
+
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/quic-go/quic-go"
 )
 
 func main() {
@@ -13,7 +22,57 @@ func main() {
 		fmt.Printf("Usage: %s <multiaddr> <peer id>", os.Args[0])
 		return
 	}
-	if err := cmdlib.RunClient(os.Args[1], os.Args[2]); err != nil {
+	if err := run(os.Args[1], os.Args[2]); err != nil {
 		log.Fatalf(err.Error())
 	}
 }
+
+func run(raddr string, p string) error {
+	peerID, err := peer.Decode(p)
+	if err != nil {
+		return err
+	}
+	addr, err := ma.NewMultiaddr(raddr)
+	if err != nil {
+		return err
+	}
+	priv, _, err := ic.GenerateECDSAKeyPair(rand.Reader)
+	if err != nil {
+		return err
+	}
+
+	reuse, err := quicreuse.NewConnManager(quic.StatelessResetKey{}, quic.TokenGeneratorKey{})
+	if err != nil {
+		return err
+	}
+	t, err := libp2pquic.NewTransport(priv, reuse, nil, nil, nil)
+	if err != nil {
+		return err
+	}
+
+	log.Printf("Dialing %s\n", addr.String())
+	conn, err := t.Dial(context.Background(), addr, peerID)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	str, err := conn.OpenStream(context.Background())
+	if err != nil {
+		return err
+	}
+	defer str.Close()
+	const msg = "Hello world!"
+	log.Printf("Sending: %s\n", msg)
+	if _, err := str.Write([]byte(msg)); err != nil {
+		return err
+	}
+	if err := str.CloseWrite(); err != nil {
+		return err
+	}
+	data, err := io.ReadAll(str)
+	if err != nil {
+		return err
+	}
+	log.Printf("Received: %s\n", data)
+	return nil
+}
diff --git a/p2p/transport/quic/cmd/lib/lib.go b/p2p/transport/quic/cmd/lib/lib.go
deleted file mode 100644
index fd2a270a..00000000
--- a/p2p/transport/quic/cmd/lib/lib.go
+++ /dev/null
@@ -1,129 +0,0 @@
-package cmdlib
-
-import (
-	"context"
-	"crypto/rand"
-	"fmt"
-	"io"
-	"log"
-
-	ic "github.com/libp2p/go-libp2p/core/crypto"
-	"github.com/libp2p/go-libp2p/core/peer"
-	tpt "github.com/libp2p/go-libp2p/core/transport"
-	libp2pquic "github.com/libp2p/go-libp2p/p2p/transport/quic"
-	"github.com/libp2p/go-libp2p/p2p/transport/quicreuse"
-
-	ma "github.com/multiformats/go-multiaddr"
-	"github.com/quic-go/quic-go"
-)
-
-func RunClient(raddr string, p string) error {
-	peerID, err := peer.Decode(p)
-	if err != nil {
-		return err
-	}
-	addr, err := ma.NewMultiaddr(raddr)
-	if err != nil {
-		return err
-	}
-	priv, _, err := ic.GenerateECDSAKeyPair(rand.Reader)
-	if err != nil {
-		return err
-	}
-
-	reuse, err := quicreuse.NewConnManager(quic.StatelessResetKey{}, quic.TokenGeneratorKey{})
-	if err != nil {
-		return err
-	}
-	t, err := libp2pquic.NewTransport(priv, reuse, nil, nil, nil)
-	if err != nil {
-		return err
-	}
-
-	log.Printf("Dialing %s\n", addr.String())
-	conn, err := t.Dial(context.Background(), addr, peerID)
-	if err != nil {
-		return err
-	}
-	defer conn.Close()
-	str, err := conn.OpenStream(context.Background())
-	if err != nil {
-		return err
-	}
-	defer str.Close()
-	const msg = "Hello world!"
-	log.Printf("Sending: %s\n", msg)
-	if _, err := str.Write([]byte(msg)); err != nil {
-		return err
-	}
-	if err := str.CloseWrite(); err != nil {
-		return err
-	}
-	data, err := io.ReadAll(str)
-	if err != nil {
-		return err
-	}
-	log.Printf("Received: %s\n", data)
-	return nil
-}
-
-func RunServer(port string, location chan peer.AddrInfo) error {
-	addr, err := ma.NewMultiaddr(fmt.Sprintf("/ip4/0.0.0.0/udp/%s/quic-v1", port))
-	if err != nil {
-		return err
-	}
-	priv, _, err := ic.GenerateECDSAKeyPair(rand.Reader)
-	if err != nil {
-		return err
-	}
-	peerID, err := peer.IDFromPrivateKey(priv)
-	if err != nil {
-		return err
-	}
-
-	reuse, err := quicreuse.NewConnManager(quic.StatelessResetKey{}, quic.TokenGeneratorKey{})
-	if err != nil {
-		return err
-	}
-	t, err := libp2pquic.NewTransport(priv, reuse, nil, nil, nil)
-	if err != nil {
-		return err
-	}
-
-	ln, err := t.Listen(addr)
-	if err != nil {
-		return err
-	}
-	fmt.Printf("Listening. Now run: go run cmd/client/main.go %s %s\n", ln.Multiaddr(), peerID)
-	if location != nil {
-		location <- peer.AddrInfo{ID: peerID, Addrs: []ma.Multiaddr{ln.Multiaddr()}}
-	}
-	for {
-		conn, err := ln.Accept()
-		if err != nil {
-			return err
-		}
-		log.Printf("Accepted new connection from %s (%s)\n", conn.RemotePeer(), conn.RemoteMultiaddr())
-		go func() {
-			if err := handleConn(conn); err != nil {
-				log.Printf("handling conn failed: %s", err.Error())
-			}
-		}()
-	}
-}
-
-func handleConn(conn tpt.CapableConn) error {
-	str, err := conn.AcceptStream()
-	if err != nil {
-		return err
-	}
-	data, err := io.ReadAll(str)
-	if err != nil {
-		return err
-	}
-	log.Printf("Received: %s\n", data)
-	if _, err := str.Write(data); err != nil {
-		return err
-	}
-	return str.Close()
-}
diff --git a/p2p/transport/quic/cmd/server/main.go b/p2p/transport/quic/cmd/server/main.go
index ef3f90a4..222841d4 100644
--- a/p2p/transport/quic/cmd/server/main.go
+++ b/p2p/transport/quic/cmd/server/main.go
@@ -1,11 +1,20 @@
 package main
 
 import (
+	"crypto/rand"
 	"fmt"
+	"io"
 	"log"
 	"os"
 
-	cmdlib "github.com/libp2p/go-libp2p/p2p/transport/quic/cmd/lib"
+	ic "github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/core/peer"
+	tpt "github.com/libp2p/go-libp2p/core/transport"
+	libp2pquic "github.com/libp2p/go-libp2p/p2p/transport/quic"
+	"github.com/libp2p/go-libp2p/p2p/transport/quicreuse"
+
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/quic-go/quic-go"
 )
 
 func main() {
@@ -13,7 +22,65 @@ func main() {
 		fmt.Printf("Usage: %s <port>", os.Args[0])
 		return
 	}
-	if err := cmdlib.RunServer(os.Args[1], nil); err != nil {
+	if err := run(os.Args[1]); err != nil {
 		log.Fatalf(err.Error())
 	}
 }
+
+func run(port string) error {
+	addr, err := ma.NewMultiaddr(fmt.Sprintf("/ip4/0.0.0.0/udp/%s/quic", port))
+	if err != nil {
+		return err
+	}
+	priv, _, err := ic.GenerateECDSAKeyPair(rand.Reader)
+	if err != nil {
+		return err
+	}
+	peerID, err := peer.IDFromPrivateKey(priv)
+	if err != nil {
+		return err
+	}
+
+	reuse, err := quicreuse.NewConnManager(quic.StatelessResetKey{}, quic.TokenGeneratorKey{})
+	if err != nil {
+		return err
+	}
+	t, err := libp2pquic.NewTransport(priv, reuse, nil, nil, nil)
+	if err != nil {
+		return err
+	}
+
+	ln, err := t.Listen(addr)
+	if err != nil {
+		return err
+	}
+	fmt.Printf("Listening. Now run: go run cmd/client/main.go %s %s\n", ln.Multiaddr(), peerID)
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			return err
+		}
+		log.Printf("Accepted new connection from %s (%s)\n", conn.RemotePeer(), conn.RemoteMultiaddr())
+		go func() {
+			if err := handleConn(conn); err != nil {
+				log.Printf("handling conn failed: %s", err.Error())
+			}
+		}()
+	}
+}
+
+func handleConn(conn tpt.CapableConn) error {
+	str, err := conn.AcceptStream()
+	if err != nil {
+		return err
+	}
+	data, err := io.ReadAll(str)
+	if err != nil {
+		return err
+	}
+	log.Printf("Received: %s\n", data)
+	if _, err := str.Write(data); err != nil {
+		return err
+	}
+	return str.Close()
+}


Test patch:
diff --git a/p2p/http/example_test.go b/p2p/http/example_test.go
index 2e90f8cc..719badea 100644
--- a/p2p/http/example_test.go
+++ b/p2p/http/example_test.go
@@ -57,7 +57,7 @@ func ExampleHost_withAStockGoHTTPClient() {
 }
 
 func ExampleHost_listenOnHTTPTransportAndStreams() {
-	serverStreamHost, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/0/quic-v1"))
+	serverStreamHost, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/50124/quic-v1"))
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -65,18 +65,14 @@ func ExampleHost_listenOnHTTPTransportAndStreams() {
 
 	server := libp2phttp.Host{
 		InsecureAllowHTTP: true, // For our example, we'll allow insecure HTTP
-		ListenAddrs:       []ma.Multiaddr{ma.StringCast("/ip4/127.0.0.1/tcp/0/http")},
+		ListenAddrs:       []ma.Multiaddr{ma.StringCast("/ip4/127.0.0.1/tcp/50124/http")},
 		StreamHost:        serverStreamHost,
 	}
 	go server.Serve()
 	defer server.Close()
 
-	for _, a := range server.Addrs() {
-		_, transport := ma.SplitLast(a)
-		fmt.Printf("Server listening on transport: %s\n", transport)
-	}
-	// Output: Server listening on transport: /quic-v1
-	// Server listening on transport: /http
+	fmt.Println("Server listening on:", server.Addrs())
+	// Output: Server listening on: [/ip4/127.0.0.1/udp/50124/quic-v1 /ip4/127.0.0.1/tcp/50124/http]
 }
 
 func ExampleHost_overLibp2pStreams() {
diff --git a/p2p/http/libp2phttp_test.go b/p2p/http/libp2phttp_test.go
index 57dd0cc8..41de2876 100644
--- a/p2p/http/libp2phttp_test.go
+++ b/p2p/http/libp2phttp_test.go
@@ -68,6 +68,84 @@ func TestHTTPOverStreams(t *testing.T) {
 	require.Equal(t, "hello", string(body))
 }
 
+func TestHTTPOverStreamsSendsConnectionClose(t *testing.T) {
+	serverHost, err := libp2p.New(
+		libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/0/quic-v1"),
+	)
+	require.NoError(t, err)
+
+	httpHost := libp2phttp.Host{StreamHost: serverHost}
+
+	connectionHeaderVal := make(chan string, 1)
+	httpHost.SetHTTPHandlerAtPath("/hello", "/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("hello"))
+		connectionHeaderVal <- r.Header.Get("Connection")
+	}))
+
+	// Start server
+	go httpHost.Serve()
+	defer httpHost.Close()
+
+	// run client
+	clientHost, err := libp2p.New(libp2p.NoListenAddrs)
+	require.NoError(t, err)
+	clientHost.Connect(context.Background(), peer.AddrInfo{
+		ID:    serverHost.ID(),
+		Addrs: serverHost.Addrs(),
+	})
+	clientHttpHost := libp2phttp.Host{StreamHost: clientHost}
+	rt, err := clientHttpHost.NewConstrainedRoundTripper(peer.AddrInfo{ID: serverHost.ID()})
+	require.NoError(t, err)
+	client := &http.Client{Transport: rt}
+	_, err = client.Get("/")
+	require.NoError(t, err)
+
+	select {
+	case val := <-connectionHeaderVal:
+		require.Equal(t, "close", strings.ToLower(val))
+	case <-time.After(5 * time.Second):
+		t.Fatal("timeout waiting for connection header")
+	}
+}
+
+func TestHTTPOverStreamsReturnsConnectionClose(t *testing.T) {
+	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.SetHTTPHandlerAtPath("/hello", "/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		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(),
+	})
+
+	s, err := clientHost.NewStream(context.Background(), serverHost.ID(), libp2phttp.ProtocolIDForMultistreamSelect)
+	require.NoError(t, err)
+	_, err = s.Write([]byte("GET / HTTP/1.1\r\nHost: \r\n\r\n"))
+	require.NoError(t, err)
+
+	out := make([]byte, 1024)
+	n, err := s.Read(out)
+	if err != io.EOF {
+		require.NoError(t, err)
+	}
+
+	require.Contains(t, strings.ToLower(string(out[:n])), "connection: close")
+}
+
 func TestRoundTrippers(t *testing.T) {
 	serverHost, err := libp2p.New(
 		libp2p.ListenAddrStrings("/ip4/127.0.0.1/udp/0/quic-v1"),
diff --git a/p2p/transport/quic/cmd/lib/lib_test.go b/p2p/transport/quic/cmd/lib/lib_test.go
deleted file mode 100644
index 1058b078..00000000
--- a/p2p/transport/quic/cmd/lib/lib_test.go
+++ /dev/null
@@ -1,30 +0,0 @@
-package cmdlib
-
-import (
-	"testing"
-
-	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/multiformats/go-multiaddr"
-)
-
-func TestCmd(t *testing.T) {
-	serverLocation := make(chan peer.AddrInfo)
-	go RunServer("0", serverLocation)
-
-	l := <-serverLocation
-
-	ip, rest := multiaddr.SplitFirst(l.Addrs[0])
-	if ip.Protocol().Code == multiaddr.P_IP4 && ip.Value() == "0.0.0.0" {
-		// Windows can't dial to 0.0.0.0 so replace with localhost
-		var err error
-		ip, err = multiaddr.NewComponent("ip4", "127.0.0.1")
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	err := RunClient(multiaddr.Join(ip, rest).String(), l.ID.String())
-	if err != nil {
-		t.Fatal(err)
-	}
-}