rproxy: close the connection when a short answer is cut off (closes #318)

When SWRP_MAX_RESPONSE_TIME_MS ran out before the last of a short answer
was written, the flush failed. If the visitor had stopped short of its
announced body, Go's server then read the rest until
SWRP_MAX_REQUEST_TIME_MS ran out, keeping the connection open with the
answer cut off.

finish now flushes once more, and if that fails at its deadline, it ends
the reading of the body at once. ReverseProxy ignores the failed flush when
the app's answer came in one read, and panics when it was still waiting for
more; finish runs after both. Go's server never reuses a connection after a
failed write, so it closes it.

Model: opus-5-5

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
+	// Both time limits are far longer than dial's deadline, so that the answer
-	// of an answer have to be sent, and than dial's deadline.
+	// 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",
+		maxRequestTime:  "120000",
-		maxResponseTime: "300",
+		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 {
+	// After a cut-off answer, the reading of a body that has not arrived whole
-		rq.stopPassingOnBody()
+	// 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
+// not arrived whole, and lets Go's server read the rest until deadline. Go's
-// SWRP_MAX_REQUEST_TIME_MS runs out. Go's server reads it once the handler
+// server reads it once the handler returns, but with no deadline at all if
-// returns, but with no deadline at all if another read of it is still waiting.
+// another read of it is still waiting.
-func (rq *request) stopPassingOnBody() {
+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