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

Change-Id: I9029d7d65cf602dd29ee1b2a87a77a73e99d9c92
GitHub-Last-Rev: 6b31f98
GitHub-Pull-Request: #61745
Reviewed-on: https://go-review.googlesource.com/c/go/+/515796
Reviewed-by: Bryan Mills <[email protected]>
Reviewed-by: Damien Neil <[email protected]>
Run-TryBot: Bryan Mills <[email protected]>
TryBot-Result: Gopher Robot <[email protected]>

Author: AlexanderYastrebov

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()

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
+	})
+}