rproxy: answer a visitor that stops sending its body after the app has answered (closes #298)

When the app answered before the visitor's body had arrived whole, Go's
server read up to 256 KiB of the rest before it sent the answer. A
visitor that stopped sending got nothing until SWRP_MAX_REQUEST_TIME_MS
ran out, and no bytes at all once that was past the 10 seconds the
status has to be sent.

Such an answer now carries Connection: close, which Go sends at once.
finish first ends the transport's read of the body: once the handler
returns, Go's server reads the rest with no deadline while another read
of it waits. requestBody.Close does nothing, so that ReverseProxy's
close does not wait on the visitor.

Model: opus-5-5

internal/rproxy/bodies.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"sync"
 	"sync/atomic"
 )
 
@@ -31,11 +32,24 @@ type requestBody struct {
 	received atomic.Bool
 	// bytes is how much of the body has been read.
 	bytes atomic.Int64
+
+	// mu is held for the whole of each Read, and guards stopped.
+	mu sync.Mutex
+	// stopped is true once the body no longer goes to the app; Read then reads
+	// no more.
+	stopped bool
 }
 
 // Read reads from the visitor's body, after the part held, which has been
 // counted already.
 func (b *requestBody) Read(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.stopped {
+		return 0, http.ErrBodyReadAfterClose
+	}
+
 	if len(b.held) > 0 {
 		n := copy(p, b.held)
 		b.held = b.held[n:]
@@ -60,9 +74,11 @@ func (b *requestBody) Read(p []byte) (int, error) {
 	return n, err
 }
 
-// Close closes the visitor's body.
+// Close does nothing: Go's server closes the visitor's body once the handler
+// returns. Closing it earlier, as the transport and ReverseProxy do, would wait
+// for a Read in progress and then read up to 256 KiB more of the body.
 func (b *requestBody) Close() error {
-	return b.body.Close()
+	return nil
 }
 
 // responseBody is the app's response body on its way to the visitor.

internal/rproxy/limits_test.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -378,12 +379,19 @@ type deadlineRecorder struct {
 
 	readDeadline  time.Time
 	writeDeadline time.Time
+	// stalled, if set, sends the part of the request body that never arrives. A
+	// read deadline that has passed ends a read of it, as on a connection.
+	stalled *io.PipeWriter
 }
 
 // SetReadDeadline notes deadline.
 func (w *deadlineRecorder) SetReadDeadline(deadline time.Time) error {
 	w.readDeadline = deadline
 
+	if w.stalled != nil && !deadline.IsZero() && !deadline.After(time.Now()) {
+		_ = w.stalled.CloseWithError(os.ErrDeadlineExceeded)
+	}
+
 	return nil
 }
 
@@ -422,7 +430,7 @@ func TestABodyLeftUnreadIsReadOnlyUntilTheRequestTimeLimit(t *testing.T) {
 	req.ContentLength = announced
 	req.Header.Set(contentLengthHeader, strconv.Itoa(announced))
 
-	w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+	w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder(), stalled: stop}
 
 	before := time.Now()
 
@@ -814,6 +822,94 @@ func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing.
 	}
 }
 
+// answerBeforeTheBody is an app that sends its whole answer, greeting, before it
+// reads the request's body, and then reads the body.
+func answerBeforeTheBody(w http.ResponseWriter, r *http.Request) {
+	rc := http.NewResponseController(w)
+	_ = rc.EnableFullDuplex()
+
+	w.Header().Set(contentLengthHeader, strconv.Itoa(len(greeting)))
+	_, _ = io.WriteString(w, greeting)
+	_ = rc.Flush()
+
+	_, _ = io.Copy(io.Discard, r.Body)
+}
+
+// sendAllButALittle sends a POST whose body stops short of its announced
+// length, by less than Go's server reads of a body left unread before it
+// answers.
+func sendAllButALittle(t *testing.T, conn net.Conn) {
+	t.Helper()
+
+	send(t, conn, post+announcedLength(2*questionBodyMaxBytes)+
+		strings.Repeat("x", questionBodyMaxBytes+1))
+}
+
+// readAnswer reads the whole answer from conn.
+func readAnswer(t *testing.T, conn net.Conn) answered {
+	t.Helper()
+
+	res, err := http.ReadResponse(bufio.NewReader(conn), nil)
+	if err != nil {
+		t.Fatalf("read the answer: %v", err)
+	}
+
+	body, err := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+
+	if err != nil {
+		t.Fatalf("read the answer's body: %v", err)
+	}
+
+	return answered{res.StatusCode, res.Header, string(body)}
+}
+
+func TestAVisitorThatStopsALittleShortGetsTheAppsAnswerAtOnce(t *testing.T) {
+	t.Parallel()
+
+	standIn := startDiscriminator(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.
+	addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
+		maxRequestTime:  "15000",
+		maxResponseTime: "300",
+	})
+
+	conn := dial(t, addr)
+	sendAllButALittle(t, conn)
+
+	wantGreeting(t, readAnswer(t, conn))
+}
+
+func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing.T) {
+	t.Parallel()
+
+	const limit = 2 * time.Second
+
+	standIn := startDiscriminator(t, allow)
+	app := startApp(t, answerBeforeTheBody)
+	addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
+		maxRequestTime: strconv.FormatInt(limit.Milliseconds(), 10),
+	})
+
+	conn := dial(t, addr)
+	start := time.Now()
+
+	sendAllButALittle(t, conn)
+	wantGreeting(t, readAnswer(t, conn))
+
+	// If the connection stays open, dial's deadline ends this read with a timeout.
+	rest, err := io.ReadAll(conn)
+	if err != nil || len(rest) != 0 {
+		t.Errorf("read %q (%v), want the connection closed", rest, err)
+	}
+
+	if closed := time.Since(start); closed < limit {
+		t.Errorf("the connection closed after %s, want %s at least", closed, limit)
+	}
+}
+
 func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) {
 	t.Parallel()
 

internal/rproxy/request.go

@@ -273,6 +273,13 @@ func (rq *request) modifyResponse(res *http.Response) error {
 		rq.out.Header()["Content-Type"] = nil
 	}
 
+	// Go's server reads up to 256 KiB of a body left unread before it sends an
+	// answer, and so waits on a visitor that has stopped sending. With this
+	// header it sends the answer at once, and closes the connection after it.
+	if rq.body != nil && !rq.body.received.Load() {
+		rq.out.Header().Set("Connection", "close")
+	}
+
 	res.Body = &responseBody{body: limitBody(res.Body, maxBytes), rq: rq}
 	rq.out.afterStatus = rq.setResponseDeadline
 
@@ -375,7 +382,7 @@ func (rq *request) finish(ctx context.Context) {
 	rq.stopTimers()
 
 	if rq.refused.Load() == 0 {
-		rq.stopReadingBody(rq.requestDeadline())
+		rq.stopPassingOnBody()
 	}
 
 	requestEnd := rq.endRequest()
@@ -462,6 +469,26 @@ func (rq *request) stopReadingBody(deadline time.Time) {
 	_ = rq.rc.SetReadDeadline(deadline)
 }
 
+// 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() {
+	if rq.body == nil || rq.body.received.Load() {
+		return
+	}
+
+	// A deadline that has passed ends a Read that waits, and the lock waits for
+	// that Read to return.
+	_ = rq.rc.SetReadDeadline(time.Now())
+
+	rq.body.mu.Lock()
+	rq.body.stopped = true
+	rq.body.mu.Unlock()
+
+	rq.stopReadingBody(rq.requestDeadline())
+}
+
 // startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the
 // request's headers being read until the app has been sent the whole request.
 func (rq *request) startRequestTimer() {
@@ -490,10 +517,10 @@ func (rq *request) requestTimedOut() {
 	}
 
 	if rq.body == nil || !rq.body.waiting.Load() {
-		// Once the app has answered, the transport, ReverseProxy and Go's server
-		// each close the visitor's body, and closing it reads what is left of a
-		// body under 256 KiB without requestBody.Read, so waiting is false while
-		// that read waits for the visitor. That read ends now, before refuse:
+		// Once the app has answered, Go's server can read what is left of a body
+		// under 256 KiB without requestBody.Read, before an answer rproxy gives
+		// itself, so waiting is false while that read waits for the visitor.
+		// That read ends now, before refuse:
 		// should rproxy still answer itself, answer then sets the deadline it
 		// gives its own answers.
 		if !rq.answerStarted.IsZero() {
@@ -557,17 +584,8 @@ func (rq *request) responseTimedOut() {
 	}
 }
 
-// responseReceived is called once the app has sent its whole answer. An app may
-// answer before the visitor has sent its whole body, as with a 404 it sends
-// without reading it. The time limits then run on: until the transport's read of
-// the body returns, Go's server cannot send the answer, and
-// SWRP_MAX_REQUEST_TIME_MS is what ends that read for a visitor that stops
-// sending.
+// responseReceived is called once the app has sent its whole answer.
 func (rq *request) responseReceived() {
-	if rq.body != nil && !rq.body.received.Load() {
-		return
-	}
-
 	rq.stopTimers()
 }