http2: don't PING a responsive server when resetting a stream

When sending a RST_STREAM for a canceled request, we sometimes send
a PING frame along with the reset to confirm that the server is responsive
and has received the reset.

Sending too many PINGs trips denial-of-service detection on some servers,
causing them to close a connection with an ENHANCE_YOUR_CALM error.

Do not send a PING frame along with an RST_STREAM if the connection
has displayed signs of life since the canceled request began.
Specifically, if we've received any stream-related frames since the
request was sent, assume the server is responsive and do not send a PING.

We still send a PING if a request is canceled and no stream-related
frames have been received from the server since the request was first
sent.

For golang/go#76296

Change-Id: I1be3532febf9ac99d65e9cd35346c02306db5f9d
Reviewed-on: https://go-review.googlesource.com/c/net/+/720300
Reviewed-by: Nicholas Husin <husin@google.com>
Reviewed-by: Nicholas Husin <nsh@golang.org>
LUCI-TryBot-Result: Go LUCI <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>

Author: neild

http2/clientconn_test.go

@@ -295,9 +295,12 @@ func (b *testRequestBody) closeWithError(err error) {
 // (Note that the RoundTrip won't complete until response headers are received,
 // the request times out, or some other terminal condition is reached.)
 func (tc *testClientConn) roundTrip(req *http.Request) *testRoundTrip {
+	ctx, cancel := context.WithCancel(req.Context())
+	req = req.WithContext(ctx)
 	rt := &testRoundTrip{
-		t:     tc.t,
-		donec: make(chan struct{}),
+		t:      tc.t,
+		donec:  make(chan struct{}),
+		cancel: cancel,
 	}
 	tc.roundtrips = append(tc.roundtrips, rt)
 	go func() {
@@ -367,6 +370,7 @@ type testRoundTrip struct {
 	respErr error
 	donec   chan struct{}
 	id      atomic.Uint32
+	cancel  context.CancelFunc
 }
 
 // streamID returns the HTTP/2 stream ID of the request.

http2/transport.go

@@ -376,6 +376,13 @@ type ClientConn struct {
 	// completely unresponsive connection.
 	pendingResets int
 
+	// readBeforeStreamID is the smallest stream ID that has not been followed by
+	// a frame read from the peer. We use this to determine when a request may
+	// have been sent to a completely unresponsive connection:
+	// If the request ID is less than readBeforeStreamID, then we have had some
+	// indication of life on the connection since sending the request.
+	readBeforeStreamID uint32
+
 	// reqHeaderMu is a 1-element semaphore channel controlling access to sending new requests.
 	// Write to reqHeaderMu to lock it, read from it to unlock.
 	// Lock reqmu BEFORE mu or wmu.
@@ -1654,6 +1661,8 @@ func (cs *clientStream) cleanupWriteRequest(err error) {
 	}
 	bodyClosed := cs.reqBodyClosed
 	closeOnIdle := cc.singleUse || cc.doNotReuse || cc.t.disableKeepAlives() || cc.goAway != nil
+	// Have we read any frames from the connection since sending this request?
+	readSinceStream := cc.readBeforeStreamID > cs.ID
 	cc.mu.Unlock()
 	if mustCloseBody {
 		cs.reqBody.Close()
@@ -1685,8 +1694,10 @@ func (cs *clientStream) cleanupWriteRequest(err error) {
 				//
 				// This could be due to the server becoming unresponsive.
 				// To avoid sending too many requests on a dead connection,
-				// we let the request continue to consume a concurrency slot
-				// until we can confirm the server is still responding.
+				// if we haven't read any frames from the connection since
+				// sending this request, we let it continue to consume
+				// a concurrency slot until we can confirm the server is
+				// still responding.
 				// We do this by sending a PING frame along with the RST_STREAM
 				// (unless a ping is already in flight).
 				//
@@ -1697,7 +1708,7 @@ func (cs *clientStream) cleanupWriteRequest(err error) {
 				// because it's short lived and will probably be closed before
 				// we get the ping response.
 				ping := false
-				if !closeOnIdle {
+				if !closeOnIdle && !readSinceStream {
 					cc.mu.Lock()
 					// rstStreamPingsBlocked works around a gRPC behavior:
 					// see comment on the field for details.
@@ -2784,6 +2795,7 @@ func (rl *clientConnReadLoop) streamByID(id uint32, headerOrData bool) *clientSt
 		// See comment on ClientConn.rstStreamPingsBlocked for details.
 		rl.cc.rstStreamPingsBlocked = false
 	}
+	rl.cc.readBeforeStreamID = rl.cc.nextStreamID
 	cs := rl.cc.streams[id]
 	if cs != nil && !cs.readAborted {
 		return cs

http2/transport_test.go

@@ -2563,9 +2563,6 @@ func testTransportReturnsUnusedFlowControl(t testing.TB, oneDataFrame bool) {
 			}
 			return true
 		},
-		func(f *PingFrame) bool {
-			return true
-		},
 		func(f *WindowUpdateFrame) bool {
 			if !oneDataFrame && !sentAdditionalData {
 				t.Fatalf("Got WindowUpdateFrame, don't expect one yet")
@@ -5567,6 +5564,8 @@ func TestTransport1xxLimits(t *testing.T) {
 	}
 }
 
+// TestTransportSendPingWithReset verifies that when a request to an unresponsive server
+// is canceled, it continues to consume a concurrency slot until the server responds to a PING.
 func TestTransportSendPingWithReset(t *testing.T) { synctestTest(t, testTransportSendPingWithReset) }
 func testTransportSendPingWithReset(t testing.TB) {
 	tc := newTestClientConn(t, func(tr *Transport) {
@@ -5578,33 +5577,25 @@ func testTransportSendPingWithReset(t testing.TB) {
 
 	// Start several requests.
 	var rts []*testRoundTrip
-	for i := 0; i < maxConcurrent+1; i++ {
+	for i := range maxConcurrent + 1 {
 		req := must(http.NewRequest("GET", "https://dummy.tld/", nil))
 		rt := tc.roundTrip(req)
 		if i >= maxConcurrent {
 			tc.wantIdle()
 			continue
 		}
 		tc.wantFrameType(FrameHeaders)
-		tc.writeHeaders(HeadersFrameParam{
-			StreamID:   rt.streamID(),
-			EndHeaders: true,
-			BlockFragment: tc.makeHeaderBlockFragment(
-				":status", "200",
-			),
-		})
-		rt.wantStatus(200)
 		rts = append(rts, rt)
 	}
 
 	// Cancel one request. We send a PING frame along with the RST_STREAM.
-	rts[0].response().Body.Close()
+	rts[0].cancel()
 	tc.wantRSTStream(rts[0].streamID(), ErrCodeCancel)
 	pf := readFrame[*PingFrame](t, tc)
 	tc.wantIdle()
 
 	// Cancel another request. No PING frame, since one is in flight.
-	rts[1].response().Body.Close()
+	rts[1].cancel()
 	tc.wantRSTStream(rts[1].streamID(), ErrCodeCancel)
 	tc.wantIdle()
 
@@ -5613,16 +5604,55 @@ func testTransportSendPingWithReset(t testing.TB) {
 	tc.writePing(true, pf.Data)
 	tc.wantFrameType(FrameHeaders)
 	tc.wantIdle()
+}
 
-	// Receive a byte of data for the remaining stream, which resets our ability
-	// to send pings (see comment on ClientConn.rstStreamPingsBlocked).
-	tc.writeData(rts[2].streamID(), false, []byte{0})
+// TestTransportNoPingAfterResetWithFrames verifies that when a request to a responsive
+// server is canceled (specifically: when frames have been received from the server
+// in the time since the request was first sent), the request is immediately canceled and
+// does not continue to consume a concurrency slot.
+func TestTransportNoPingAfterResetWithFrames(t *testing.T) {
+	synctestTest(t, testTransportNoPingAfterResetWithFrames)
+}
+func testTransportNoPingAfterResetWithFrames(t testing.TB) {
+	tc := newTestClientConn(t, func(tr *Transport) {
+		tr.StrictMaxConcurrentStreams = true
+	})
 
-	// Cancel the last request. We send another PING, since none are in flight.
-	rts[2].response().Body.Close()
-	tc.wantRSTStream(rts[2].streamID(), ErrCodeCancel)
-	tc.wantFrameType(FramePing)
+	const maxConcurrent = 1
+	tc.greet(Setting{SettingMaxConcurrentStreams, maxConcurrent})
+
+	// Start request #1.
+	// The server immediately responds with request headers.
+	req1 := must(http.NewRequest("GET", "https://dummy.tld/", nil))
+	rt1 := tc.roundTrip(req1)
+	tc.wantFrameType(FrameHeaders)
+	tc.writeHeaders(HeadersFrameParam{
+		StreamID:   rt1.streamID(),
+		EndHeaders: true,
+		BlockFragment: tc.makeHeaderBlockFragment(
+			":status", "200",
+		),
+	})
+	rt1.wantStatus(200)
+
+	// Start request #2.
+	// The connection is at its concurrency limit, so this request is not yet sent.
+	req2 := must(http.NewRequest("GET", "https://dummy.tld/", nil))
+	rt2 := tc.roundTrip(req2)
 	tc.wantIdle()
+
+	// Cancel request #1.
+	// This frees a concurrency slot, and request #2 is sent.
+	rt1.cancel()
+	tc.wantRSTStream(rt1.streamID(), ErrCodeCancel)
+	tc.wantFrameType(FrameHeaders)
+
+	// Cancel request #2.
+	// We send a PING along with the RST_STREAM, since no frames have been received
+	// since this request was sent.
+	rt2.cancel()
+	tc.wantRSTStream(rt2.streamID(), ErrCodeCancel)
+	tc.wantFrameType(FramePing)
 }
 
 // Issue #70505: gRPC gets upset if we send more than 2 pings per HEADERS/DATA frame