rproxy: read the rest of a Connection: close visitor's body after an early answer (closes #373)

Go's server reads the body left unread once the handler returns, but not
for a request with Connection: close: it closes the connection with the
rest unread, and the visitor, still sending, gets a reset. rproxy now
reads that rest itself once it has answered, up to the 256 KiB Go's
server reads, until the deadline already set on the connection.

Model: opus-5-5

internal/rproxy/limits_test.go

@@ -131,6 +131,67 @@ func TestADenyOutranksTheBodySizeLimit(t *testing.T) {
 	}
 }
 
+func TestAVisitorSendingConnectionCloseReadsAnAnswerBeforeTheRestOfItsBody(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name   string
+		waf    http.HandlerFunc
+		status int
+	}{
+		{"over the size limit", allow, http.StatusRequestEntityTooLarge},
+		{"denied", deny, http.StatusForbidden},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			standIn := startWAF(t, tc.waf)
+			app := startApp(t, func(http.ResponseWriter, *http.Request) {
+				t.Error("the app was asked")
+			})
+			addr, _ := startRProxy(t, standIn.URL, app.URL,
+				map[string]string{maxRequestSize: "1K"})
+
+			const size = 129 << 10
+
+			conn := dial(t, addr)
+			send(t, conn, post+"Connection: close\r\n"+
+				announcedLength(size)+
+				strings.Repeat("x", questionBodyMaxBytes))
+
+			reader := bufio.NewReader(conn)
+
+			res, err := http.ReadResponse(reader, nil)
+			if err != nil {
+				t.Fatalf("read the answer: %v", err)
+			}
+
+			_, _ = io.Copy(io.Discard, res.Body)
+			_ = res.Body.Close()
+
+			if res.StatusCode != tc.status {
+				t.Errorf("status %d, want %d", res.StatusCode, tc.status)
+			}
+
+			// The rest goes in small parts, so that a reset fails a later one.
+			for sent := questionBodyMaxBytes; sent < size; sent += 1 << 10 {
+				send(t, conn, strings.Repeat("x", 1<<10))
+			}
+
+			rest, err := io.ReadAll(reader)
+			if err != nil || len(rest) != 0 {
+				t.Errorf(
+					"after the answer, read %q and %v, want the end",
+					rest,
+					err,
+				)
+			}
+		})
+	}
+}
+
 func TestABodyNotAnnouncedOverItsLimitIsRefused(t *testing.T) {
 	t.Parallel()
 

internal/rproxy/request.go

@@ -38,6 +38,10 @@ const plainText = "text/plain; charset=utf-8"
 // and the visitor then gets a closed connection and no answer.
 const minSendTime = 10 * time.Second
 
+// maxBodyReadAfterAnswer is the most of a body left unread that Go's server
+// reads once the handler returns, for a visitor without Connection: close.
+const maxBodyReadAfterAnswer = 256 << 10
+
 // errResponseTooLate refuses an app's answer that arrives once
 // SWW_RPROXY_MAX_RESPONSE_TIME_MS has run out.
 var errResponseTooLate = errors.New(
@@ -338,8 +342,8 @@ func (rq *request) answer(status int) {
 	}
 
 	// A visitor found too slow is read no more; any other may go on sending until
-	// its time is up, so that Go's server can read the rest of the body and end
+	// its time is up, so that the rest of the body is read and the request ends
-	// the request cleanly.
+	// cleanly: by Go's server, or by readRestOfBody.
 	deadline := rq.requestDeadline()
 	if status == http.StatusRequestTimeout {
 		deadline = time.Now()
@@ -523,6 +527,22 @@ func (rq *request) stopReadingBody(deadline time.Time) {
 	_ = rq.rc.SetReadDeadline(deadline)
 }
 
+// readRestOfBody reads, once the visitor has been answered, the rest of a body
+// that has not arrived whole, when the visitor sent Connection: close. Go's
+// server reads it for any other visitor, but closes this visitor's connection
+// with the rest unread: the visitor, still sending, then gets a reset and may
+// never read the answer. The read ends at the deadline already set on the
+// connection, and after as much as Go's server reads.
+func (rq *request) readRestOfBody() {
+	if !rq.in.Close || rq.body == nil || rq.body.received.Load() ||
+		rq.out.status == 0 ||
+		rq.out.status == http.StatusSwitchingProtocols {
+		return
+	}
+
+	_, _ = io.CopyN(io.Discard, rq.in.Body, maxBodyReadAfterAnswer)
+}
+
 // 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 deadline. Go's
 // server reads it once the handler returns, but with no deadline at all if

internal/rproxy/rproxy.go

@@ -169,6 +169,7 @@ func newWAFTransport() *http.Transport {
 // it writes the request's log line and sends its report.
 func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	rq := h.newRequest(w, r)
+	defer rq.readRestOfBody()
 	defer rq.finish(r.Context())
 
 	if r.Method == http.MethodGet && r.URL.Path == waf.HealthPath {