rproxy tests: a cut-off answer closes the connection, also after its first part reached the visitor

The app now sends its late parts only once rproxy has ended the request to
it, not after a sleep timed against the limit. With the body in one part it
stalls after its headers; with two parts, after the first part, which
reaches the visitor. Both cases fail on the current code: nothing cuts the
answer off, and the connection stays open. The 2-second limit lets rproxy
pass on a part sent at once, even on a busy host.

A new test, which passes on the current code, has every flush fail with no
refusal, so that only the failed-flush check in finish ends the reading of
the visitor's body.

Model: opus-5-5

internal/rproxy/limits_test.go

@@ -12,7 +12,6 @@ import (
 	"os"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -383,6 +382,20 @@ type deadlineRecorder struct {
 	// 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
+	// flushError, if set, is what every flush returns, and nothing is flushed.
+	flushError error
+}
+
+// FlushError, which http.ResponseController calls to flush, returns flushError
+// if it is set, and otherwise flushes.
+func (w *deadlineRecorder) FlushError() error {
+	if w.flushError != nil {
+		return w.flushError
+	}
+
+	w.Flush()
+
+	return nil
 }
 
 // SetReadDeadline notes deadline.
@@ -448,6 +461,45 @@ func TestABodyLeftUnreadIsReadOnlyUntilTheRequestTimeLimit(t *testing.T) {
 	}
 }
 
+func TestABodyLeftUnreadIsReadNoMoreOnceTheTimeToSendTheAnswerRunsOut(t *testing.T) {
+	t.Parallel()
+
+	standIn := startWAF(t, allow)
+	app := startApp(t, answerBeforeTheBody)
+	_, _, server := startServer(t, standIn.URL, app.URL, nil)
+
+	// The visitor sends the first part and a little more, and then nothing.
+	stalled, stop := io.Pipe()
+
+	t.Cleanup(func() { _ = stop.Close() })
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/",
+		io.MultiReader(bytes.NewReader(make([]byte, questionBodyMaxBytes+1)), stalled))
+	req.ContentLength = 2 * questionBodyMaxBytes
+	req.Header.Set(contentLengthHeader, strconv.Itoa(2*questionBodyMaxBytes))
+
+	// Every flush fails, as once the time to send the answer has run out. The
+	// default time limits are far longer than the test, so rproxy records no
+	// refusal.
+	w := &deadlineRecorder{
+		ResponseRecorder: httptest.NewRecorder(),
+		stalled:          stop,
+		flushError:       os.ErrDeadlineExceeded,
+	}
+
+	before := time.Now()
+
+	server.Handler.ServeHTTP(w, req)
+
+	if w.Code != http.StatusOK {
+		t.Errorf("status %d, want the app's 200", w.Code)
+	}
+
+	if w.readDeadline.Before(before) || w.readDeadline.After(time.Now()) {
+		t.Errorf("the reading of the body ends at %s, want now", w.readDeadline)
+	}
+}
+
 func TestAnAnswerRProxyGivesItselfHasTheResponseTimeToBeSent(t *testing.T) {
 	t.Parallel()
 
@@ -918,49 +970,45 @@ func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOf
 ) {
 	t.Parallel()
 
-	const limit = 300 * time.Millisecond
+	// rproxy passes on a part sent at once well within this limit, even on a busy
+	// host.
+	const limit = 2 * time.Second
 
-	// With two parts, rproxy fails to send the first and then waits for the
-	// second, which the app sends only once rproxy has ended the request to it.
+	// The app sends the parts in atOnce with its status and headers. It sends the
+	// parts in late only once rproxy has ended the request to it, which rproxy
+	// does when it cuts off the answer, so that those never reach the visitor.
 	for _, tc := range []struct {
-		name  string
-		parts []string
+		name         string
+		atOnce, late []string
 	}{
-		{"with the body in one part", []string{greeting}},
-		{"with the body in two parts", []string{greeting[:5], greeting[5:]}},
+		{"with the body in one part", nil, []string{greeting}},
+		{"with the body in two parts", []string{greeting[:5]}, []string{greeting[5:]}},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			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. After each part, it reads the request's body, which ends when rproxy
-			// ends the request to 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)))
 
+				for _, part := range tc.atOnce {
+					_, _ = io.WriteString(w, part)
+				}
+
 				_ = rc.Flush()
 
-				<-headersRead
-				time.Sleep(limit)
+				// The request's body ends when rproxy ends the request to the app.
+				_, _ = io.Copy(io.Discard, r.Body)
 
-				for _, part := range tc.parts {
+				for _, part := range tc.late {
 					_, _ = io.WriteString(w, part)
 					_ = 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",
@@ -975,15 +1023,14 @@ func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOf
 				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)
+			want := strings.Join(tc.atOnce, "")
+			if string(body) != want || !errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Errorf("read %q (%v), want %q and the connection closed", body, err, want)
 			}
 		})
 	}