From 6b31f9826da71dad4ee8c0491efba995a8f51440 Mon Sep 17 00:00:00 2001
From: Alexander Yastrebov <yastrebov.alex@gmail.com>
Date: Fri, 4 Aug 2023 02:47:41 +0200
Subject: [PATCH] net/http: fix request canceler leak on connection close

Due to a race condition persistConn could be closed without removing
request canceler.

Note that without the fix test occasionally passes and to demonstrate
the issue it has to be run multiple times, e.g. using -count=10.

Fixes #61708
---
 src/net/http/transport.go      |  1 +
 src/net/http/transport_test.go | 52 ++++++++++++++++++++++++++++++++++
 2 files changed, 53 insertions(+)

diff --git a/src/net/http/transport.go b/src/net/http/transport.go
index 35dfe908d82..c2376aa6615 100644
--- a/src/net/http/transport.go
+++ b/src/net/http/transport.go
@@ -2267,6 +2267,7 @@ func (pc *persistConn) readLoop() {
 			pc.t.cancelRequest(rc.cancelKey, rc.req.Context().Err())
 		case <-pc.closech:
 			alive = false
+			pc.t.setReqCanceler(rc.cancelKey, nil)
 		}
 
 		testHookReadLoopBeforeNextRead()
diff --git a/src/net/http/transport_test.go b/src/net/http/transport_test.go
index bcc26aa58e0..4ff26ff32af 100644
--- a/src/net/http/transport_test.go
+++ b/src/net/http/transport_test.go
@@ -6810,3 +6810,55 @@ func testRequestSanitization(t *testing.T, mode testMode) {
 		resp.Body.Close()
 	}
 }
+
+// Issue 61708
+func TestTransportAndServerSharedBodyReqCancelerCleanupOnConnectionClose(t *testing.T) {
+	run(t, testTransportAndServerSharedBodyReqCancelerCleanupOnConnectionClose, []testMode{http1Mode})
+}
+func testTransportAndServerSharedBodyReqCancelerCleanupOnConnectionClose(t *testing.T, mode testMode) {
+	const bodySize = 1 << 20
+
+	backend := newClientServerTest(t, mode, HandlerFunc(func(rw ResponseWriter, req *Request) {
+		io.Copy(rw, req.Body)
+	}))
+	t.Logf("Backend address: %s", backend.ts.Listener.Addr().String())
+
+	var proxy *clientServerTest
+	proxy = newClientServerTest(t, mode, HandlerFunc(func(rw ResponseWriter, req *Request) {
+		breq, _ := NewRequest("POST", backend.ts.URL, req.Body)
+
+		bresp, err := backend.c.Do(breq)
+		if err != nil {
+			t.Fatalf("Unexpected proxy outbound request error: %v", err)
+		}
+		defer bresp.Body.Close()
+
+		_, err = io.Copy(rw, bresp.Body)
+		if err == nil {
+			t.Fatalf("Expected proxy copy error")
+		}
+		t.Logf("Proxy copy error: %v", err)
+	}))
+	t.Logf("Proxy address: %s", proxy.ts.Listener.Addr().String())
+
+	req, _ := NewRequest("POST", proxy.ts.URL, io.LimitReader(neverEnding('a'), bodySize))
+	res, err := proxy.c.Do(req)
+	if err != nil {
+		t.Fatalf("Original request: %v", err)
+	}
+	// Close body without reading to trigger proxy copy error
+	res.Body.Close()
+
+	// Verify no outstanding requests after readLoop/writeLoop
+	// goroutines shut down.
+	waitCondition(t, 10*time.Millisecond, func(d time.Duration) bool {
+		n := backend.tr.NumPendingRequestsForTesting()
+		if n > 0 {
+			if d > 0 {
+				t.Logf("pending requests = %d after %v (want 0)", n, d)
+			}
+			return false
+		}
+		return true
+	})
+}