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 and ReverseProxy ignored it. 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. Go does that read even after
an aborting panic.

forward now checks for a failed flush once ReverseProxy returns. It
then records a 504 refusal, ends the reading of the body at once, and
panics so that Go's server closes the connection, as SPEC.md says a
cut-off answer does.

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
-	// 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,63 @@ func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing.
 	}
 }
 
+func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOff(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	const limit = 300 * time.Millisecond
+
+	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.
+	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)
+
+		_, _ = io.WriteString(w, greeting)
+		_ = 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

@@ -195,6 +195,20 @@ func (rq *request) forward(ctx context.Context) {
 	rq.startRequestTimer()
 	rq.upstreamStart = time.Now()
 	reverseProxy.ServeHTTP(rq.out, out)
+
+	// ReverseProxy ignores a failed flush, so it misses a short answer cut off when
+	// SWRP_MAX_RESPONSE_TIME_MS runs out; once a flush has failed, every later one
+	// fails too. Go's server would then keep the connection open while it reads
+	// the rest of a body that has not arrived whole. So that reading ends now, and
+	// the refusal keeps finish from extending it to SWRP_MAX_REQUEST_TIME_MS. The
+	// panic has Go's server close the connection, and log nothing; the deferred
+	// finish still runs.
+	if rq.out.status != http.StatusSwitchingProtocols &&
+		errors.Is(rq.rc.Flush(), os.ErrDeadlineExceeded) {
+		rq.refuse(http.StatusGatewayTimeout)
+		rq.stopPassingOnBody(time.Now())
+		panic(http.ErrAbortHandler)
+	}
 }
 
 // rewrite makes the request the app receives: the visitor's request, unchanged,
@@ -385,7 +399,7 @@ func (rq *request) finish(ctx context.Context) {
 	rq.stopTimers()
 
 	if rq.refused.Load() == 0 {
-		rq.stopPassingOnBody()
+		rq.stopPassingOnBody(rq.requestDeadline())
 	}
 
 	requestEnd := rq.endRequest()
@@ -473,10 +487,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 +503,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