e2e/tests: add TestProxyReuseConnection

In order to reuse HTTP connection the body must be drained.

e2e/tests/proxy_test.go

@@ -12,15 +12,19 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"math/rand"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
 	"runtime"
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/saucelabs/forwarder/e2e/forwarder"
@@ -287,3 +291,47 @@ func TestProxyUpstream(t *testing.T) {
 		t.Fatalf("%s via header not found", forwarder.UpstreamProxyServiceName)
 	}
 }
+
+func TestProxyReuseConnection(t *testing.T) {
+	c := newClient(t, "http://wronghost", func(tr *http.Transport) {
+		var (
+			d      net.Dialer
+			dialed atomic.Bool
+		)
+		tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
+			if dialed.CompareAndSwap(false, true) {
+				return d.DialContext(ctx, network, addr)
+			}
+			return nil, errors.New("only one dial is allowed")
+		}
+	})
+
+	for i := 0; i < 2; i++ {
+		r, w := io.Pipe()
+
+		written := make(chan struct{})
+		go func() {
+			zeros := make([]byte, 1024*1024)
+			if _, err := w.Write(zeros); err != nil {
+				t.Error(err)
+			}
+			if err := w.Close(); err != nil {
+				t.Error(err)
+			}
+			close(written)
+		}()
+
+		res := c.GET("/", func(req *http.Request) {
+			req.Body = r
+		})
+		if res.StatusCode != http.StatusBadGateway {
+			t.Fatalf("Expected status %d, got %d", http.StatusBadGateway, res.StatusCode)
+		}
+
+		select {
+		case <-written:
+		case <-time.After(10 * time.Second):
+			t.Fatal("timed-out waiting for body read")
+		}
+	}
+}