internal/rproxy/limits_test.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -807,8 +808,10 @@ func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing.
 	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
 		_, _ = io.WriteString(w, greeting)
 	})
+	// The request time limit is far longer than dial's deadline, so that a busy
+	// host cannot use it up before the app's answer arrives.
 	addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
-		maxRequestTime: "3000",
+		maxRequestTime: "120000",
 	})
 
 	// Far more announced than sent, so that Go's server answers for the app
@@ -869,11 +872,11 @@ func TestAVisitorThatStopsALittleShortGetsTheAppsAnswerAtOnce(t *testing.T) {
 
 	standIn := startWAF(t, allow)
 	app := startApp(t, answerBeforeTheBody)
-	// The request time limit is longer than the 10 seconds the status and headers
-	// of an answer have to be sent, and than dial's deadline.
+	// Both time limits are far longer than dial's deadline, so that the answer
+	// has to come before either runs out, and a busy host cannot use them up.
 	addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
-		maxRequestTime:  "15000",
-		maxResponseTime: "300",
+		maxRequestTime:  "120000",
+		maxResponseTime: "60000",
 	})
 
 	conn := dial(t, addr)
@@ -910,6 +913,82 @@ func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing.
 	}
 }
 
+func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOff(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	const limit = 300 * time.Millisecond
+
+	// With two parts, rproxy fails to send the first and then waits for the
+	// second, which the app sends only once rproxy has ended the request to it.
+	for _, tc := range []struct {
+		name  string
+		parts []string
+	}{
+		{"with the body in one part", []string{greeting}},
+		{"with the body in two parts", []string{greeting[:5], greeting[5:]}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			headersRead := make(chan struct{})
+			// letAppSendBody also runs when the test ends, as the app's server waits
+			// for the app.
+			letAppSendBody := sync.OnceFunc(func() { close(headersRead) })
+			standIn := startWAF(t, allow)
+			// The app sends its body once the visitor has the status and headers and
+			// the response time limit has run out since, so that rproxy cannot send
+			// it. After each part, it reads the request's body, which ends when rproxy
+			// ends the request to it.
+			app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+				rc := http.NewResponseController(w)
+				_ = rc.EnableFullDuplex()
+
+				w.Header().Set(contentLengthHeader, strconv.Itoa(len(greeting)))
+
+				_ = rc.Flush()
+
+				<-headersRead
+				time.Sleep(limit)
+
+				for _, part := range tc.parts {
+					_, _ = io.WriteString(w, part)
+					_ = rc.Flush()
+					_, _ = io.Copy(io.Discard, r.Body)
+				}
+			})
+
+			t.Cleanup(letAppSendBody)
+
+			// The request time limit is far longer than dial's deadline.
+			addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
+				maxRequestTime:  "120000",
+				maxResponseTime: strconv.FormatInt(limit.Milliseconds(), 10),
+			})
+
+			conn := dial(t, addr)
+			sendAllButALittle(t, conn)
+
+			res, err := http.ReadResponse(bufio.NewReader(conn), nil)
+			if err != nil {
+				t.Fatalf("read the answer: %v", err)
+			}
+
+			letAppSendBody()
+
+			// If the connection stays open, dial's deadline ends this read with a
+			// timeout.
+			body, err := io.ReadAll(res.Body)
+			_ = res.Body.Close()
+
+			if len(body) != 0 || !errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Errorf("read %q (%v), want the connection closed", body, err)
+			}
+		})
+	}
+}
+
 func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) {
 	t.Parallel()
 

internal/rproxy/request.go

@@ -384,8 +384,19 @@ func (rq *request) refuse(status int) {
 func (rq *request) finish(ctx context.Context) {
 	rq.stopTimers()
 
-	if rq.refused.Load() == 0 {
-		rq.stopPassingOnBody()
+	// After a cut-off answer, the reading of a body that has not arrived whole
+	// ends now. Go's server would otherwise keep the connection open while it
+	// reads the rest. A flush fails if the time to send the answer ran out, and
+	// every flush after a failed one fails too. ReverseProxy ignores the failure,
+	// but panics if it was still waiting for more of the app's answer; finish runs
+	// either way. Before a status, a flush would send a 200; after a 101,
+	// ReverseProxy has taken over the connection.
+	switch {
+	case rq.out.status != 0 && rq.out.status != http.StatusSwitchingProtocols &&
+		errors.Is(rq.rc.Flush(), os.ErrDeadlineExceeded):
+		rq.stopPassingOnBody(time.Now())
+	case rq.refused.Load() == 0:
+		rq.stopPassingOnBody(rq.requestDeadline())
 	}
 
 	requestEnd := rq.endRequest()
@@ -473,10 +484,10 @@ func (rq *request) stopReadingBody(deadline time.Time) {
 }
 
 // stopPassingOnBody ends the transport's reading of a visitor's body that has
-// not arrived whole, and lets Go's server read the rest until
-// SWRP_MAX_REQUEST_TIME_MS runs out. Go's server reads it once the handler
-// returns, but with no deadline at all if another read of it is still waiting.
-func (rq *request) stopPassingOnBody() {
+// not arrived whole, and lets Go's server read the rest until deadline. Go's
+// server reads it once the handler returns, but with no deadline at all if
+// another read of it is still waiting.
+func (rq *request) stopPassingOnBody(deadline time.Time) {
 	if rq.body == nil || rq.body.received.Load() {
 		return
 	}
@@ -489,7 +500,7 @@ func (rq *request) stopPassingOnBody() {
 	rq.body.stopped = true
 	rq.body.mu.Unlock()
 
-	rq.stopReadingBody(rq.requestDeadline())
+	rq.stopReadingBody(deadline)
 }
 
 // startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the