net/http: retry idempotent HTTP reqs on dead reused conns

If we try to reuse a connection that the server is in the process of
closing, we may end up successfully writing out our request (or a
portion of our request) only to find a connection error when we try to
read from (or finish writing to) the socket. This manifests as an EOF
returned from the Transport's RoundTrip.

The issue, among others, is described in golang#4677.

This change follows some of the Chromium guidelines for retrying
idempotent requests only when the connection has been already been used
successfully and no header data has yet been received for the response.

Change-Id: I1ca630b944f0ed7ec1d3d46056a50fb959481a16

Author: bgentry

src/net/http/transport.go

@@ -221,18 +221,48 @@ func (t *Transport) RoundTrip(req *Request) (resp *Response, err error) {
 		return nil, err
 	}
 
-	// Get the cached or newly-created connection to either the
-	// host (for http or https), the http proxy, or the http proxy
-	// pre-CONNECTed to https server.  In any case, we'll be ready
-	// to send it requests.
-	pconn, err := t.getConn(req, cm)
-	if err != nil {
-		t.setReqCanceler(req, nil)
-		req.closeBody()
-		return nil, err
-	}
+	var pconn *persistConn
+	for {
+		// Get the cached or newly-created connection to either the
+		// host (for http or https), the http proxy, or the http proxy
+		// pre-CONNECTed to https server.  In any case, we'll be ready
+		// to send it requests.
+		pconn, err = t.getConn(req, cm)
+		if err != nil {
+			t.setReqCanceler(req, nil)
+			req.closeBody()
+			return nil, err
+		}
 
-	return pconn.roundTrip(treq)
+		resp, err = pconn.roundTrip(treq)
+		if err != nil {
+			if brhErr, ok := err.(beforeRespHeaderError); ok {
+				err = brhErr.error // unwrap the custom error in case we return it
+				if pconn.isReused() && (req.Method == "GET" || req.Method == "HEAD") {
+					// If we try to reuse a connection that the server is in the process of
+					// closing, we may end up successfully writing out our request (or a
+					// portion of our request) only to find a connection error when we try to
+					// read from (or finish writing to) the socket.
+
+					// There can be a race between the socket pool checking checking whether a
+					// socket is still connected, receiving the FIN, and sending/reading data
+					// on a reused socket.  If we receive the FIN between the connectedness
+					// check and writing/reading from the socket, we may first learn the socket
+					// is disconnected when we get a ERR_SOCKET_NOT_CONNECTED.  This will most
+					// likely happen when trying to retrieve its IP address.
+					// See http://crbug.com/105824 for more details.
+
+					// We resend a request only if we reused a keep-alive connection and
+					// did not yet receive any header data. This automatically prevents an
+					// infinite resend loop because we'll run out of the cached keep-alive
+					// connections eventually.
+					continue
+				}
+			}
+		}
+		break
+	}
+	return resp, err
 }
 
 // RegisterProtocol registers a new protocol with scheme.
@@ -805,6 +835,7 @@ type persistConn struct {
 	numExpectedResponses int
 	closed               bool // whether conn has been closed
 	broken               bool // an error has happened on this connection; marked broken so it's not reused.
+	reused               bool // whether conn has had successful request/response and is being reused.
 	// mutateHeaderFunc is an optional func to modify extra
 	// headers on each outbound request before it's written. (the
 	// original Request given to RoundTrip is not modified)
@@ -819,6 +850,22 @@ func (pc *persistConn) isBroken() bool {
 	return b
 }
 
+// isReused reports whether this connection is in a known broken state.
+func (pc *persistConn) isReused() bool {
+	pc.lk.Lock()
+	r := pc.reused
+	pc.lk.Unlock()
+	return r
+}
+
+// markReused marks this connection as having been successfully used for a request and response.
+func (pc *persistConn) markReused() bool {
+	pc.lk.Lock()
+	pc.reused = true
+	pc.lk.Unlock()
+	return true
+}
+
 func (pc *persistConn) cancelRequest() {
 	pc.conn.Close()
 }
@@ -840,6 +887,9 @@ func (pc *persistConn) readLoop() {
 
 	for alive {
 		pb, err := pc.br.Peek(1)
+		if err != nil {
+			err = beforeRespHeaderError{err}
+		}
 
 		pc.lk.Lock()
 		if pc.numExpectedResponses == 0 {
@@ -910,13 +960,15 @@ func (pc *persistConn) readLoop() {
 					err == nil &&
 					!pc.sawEOF &&
 					pc.wroteRequest() &&
+					pc.markReused() &&
 					pc.t.putIdleConn(pc)
 			}
 		}
 
 		if alive && !hasBody {
 			alive = !pc.sawEOF &&
 				pc.wroteRequest() &&
+				pc.markReused() &&
 				pc.t.putIdleConn(pc)
 		}
 
@@ -1028,6 +1080,12 @@ func (e *httpError) Temporary() bool { return true }
 var errTimeout error = &httpError{err: "net/http: timeout awaiting response headers", timeout: true}
 var errClosed error = &httpError{err: "net/http: transport closed before response was received"}
 
+// beforeRespHeaderError is used to indicate when an IO error has occurred before
+// any header data was received.
+type beforeRespHeaderError struct {
+	error
+}
+
 func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
 	pc.t.setReqCanceler(req.Request, pc.cancelRequest)
 	pc.lk.Lock()
@@ -1086,7 +1144,7 @@ WaitResponse:
 		select {
 		case err := <-writeErrCh:
 			if err != nil {
-				re = responseAndError{nil, err}
+				re = responseAndError{nil, beforeRespHeaderError{err}}
 				pc.close()
 				break WaitResponse
 			}

src/net/http/transport_test.go

@@ -2096,6 +2096,67 @@ type closerFunc func() error
 
 func (f closerFunc) Close() error { return f() }
 
+// Issue 4677. If we try to reuse a connection that the server is in the
+// process of closing, we may end up successfully writing out our request (or a
+// portion of our request) only to find a connection error when we try to read
+// from (or finish writing to) the socket.
+//
+// NOTE: we resend a request only if the request is idempotent, we reused a
+// keep-alive connection, and we haven't yet received any header data.  This
+// automatically prevents an infinite resend loop because we'll run out of the
+// cached keep-alive connections eventually.
+func TestRetryIdempotentRequestsOnError(t *testing.T) {
+	defer afterTest(t)
+
+	ts := httptest.NewServer(HandlerFunc(func(w ResponseWriter, r *Request) {
+	}))
+	defer ts.Close()
+
+	tr := &Transport{}
+	c := &Client{Transport: tr}
+	var (
+		mu sync.Mutex // guards t
+		wg sync.WaitGroup
+	)
+
+	fatal := func(err error) {
+		mu.Lock()
+		defer mu.Unlock()
+		t.Fatal(err)
+	}
+
+	// open 2 conns
+	wg.Add(2)
+	for i := 0; i < 2; i++ {
+		go func() {
+			defer wg.Done()
+			res, err := c.Get(ts.URL)
+			if err != nil {
+				fatal(err)
+			}
+			res.Body.Close()
+		}()
+	}
+	wg.Wait()
+
+	wg.Add(2)
+	waitc := make(chan struct{})
+	for i := 0; i < 2; i++ {
+		go func() {
+			defer wg.Done()
+			<-waitc
+			res, err := c.Get(ts.URL)
+			if err != nil {
+				fatal(err)
+			}
+			res.Body.Close()
+		}()
+	}
+	ts.CloseClientConnections()
+	close(waitc)
+	wg.Wait()
+}
+
 // Issue 6981
 func TestTransportClosesBodyOnError(t *testing.T) {
 	if runtime.GOOS == "plan9" {