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/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()