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

Author: AlexanderYastrebov

src/net/http/transport.go

@@ -2254,6 +2254,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()

src/net/http/transport_test.go

@@ -6750,3 +6750,61 @@ 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)
+			return
+		}
+		defer bresp.Body.Close()
+
+		_, err = io.Copy(rw, bresp.Body)
+		if err == nil {
+			t.Fatalf("Expected proxy copy error")
+			return
+		}
+		t.Logf("Proxy copy error: %v", err)
+		// Could be one of:
+		// read tcp backend.c->backend.ts: use of closed network connection
+		// write tcp proxy.ts->proxy.c: write: broken pipe
+		// write tcp proxy.ts->proxy.c: write: connection reset by peer
+	}))
+	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
+	})
+}