SPEC.md

@@ -2304,8 +2304,14 @@ no report. Its log line has status `0` and no `disposition`.
 
 `rproxy` sends the app at `SWRP_UPSTREAM_URL` the visitor's request: the part of
 the body it holds, then the rest, streaming. The app's answer comes back the
-same way. Apart from what Go's HTTP code changes, the request reaches the app
+same way.
-unchanged but for these headers:
+
+An app may answer before the visitor has sent its whole body. The visitor gets
+that answer at once, and the connection closes after it. Once the answer is
+complete, the rest of the body no longer goes to the app.
+
+Apart from what Go's HTTP code changes, the request reaches the app unchanged
+but for these headers:
 
 - `X-Forwarded-For`, with the address of `rproxy`'s TCP peer added, when that
   peer is inside `SWRP_TRUSTED_PROXIES`. The peer's other forwarded headers pass

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
+		// Once the app has answered, Go's server can read what is left of a body
-		// each close the visitor's body, and closing it reads what is left of a
+		// under 256 KiB without requestBody.Read, before an answer rproxy gives
-		// body under 256 KiB without requestBody.Read, so waiting is false while
+		// itself, so waiting is false while that read waits for the visitor.
-		// that read waits for the visitor. That read ends now, before refuse:
+		// 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
+// responseReceived is called once the app has sent its whole answer.
-// 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.
 func (rq *request) responseReceived() {
-	if rq.body != nil && !rq.body.received.Load() {
-		return
-	}
-
 	rq.stopTimers()
 }