diff --git a/SPEC.md b/SPEC.md index b5fa734..4616085 100644 --- a/SPEC.md +++ b/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 -unchanged but for these headers: +same way. + +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 diff --git a/internal/rproxy/bodies.go b/internal/rproxy/bodies.go index 4487562..f80c374 100644 --- a/internal/rproxy/bodies.go +++ b/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. diff --git a/internal/rproxy/limits_test.go b/internal/rproxy/limits_test.go index 9b24130..a60b9ae 100644 --- a/internal/rproxy/limits_test.go +++ b/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() diff --git a/internal/rproxy/request.go b/internal/rproxy/request.go index 7ea9226..c7684ef 100644 --- a/internal/rproxy/request.go +++ b/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() }