Compare commits

...
Author SHA1 Message Date
sneak 387744ab50 rproxy: close the connection when a short answer is cut off (closes #318)
check / check (push) Successful in 3m44s
When SWRP_MAX_RESPONSE_TIME_MS ran out before the last of a short
answer was written, the flush failed and ReverseProxy ignored it. If
the visitor had stopped short of its announced body, Go's server then
read the rest until SWRP_MAX_REQUEST_TIME_MS ran out, keeping the
connection open with the answer cut off. Go does that read even after
an aborting panic.

forward now checks for a failed flush once ReverseProxy returns. It
then records a 504 refusal, ends the reading of the body at once, and
panics so that Go's server closes the connection, as SPEC.md says a
cut-off answer does.

Model: opus-5-5
2026-11-11 02:34:56 +00:00
sneak 78047e52a9 rproxy tests: a cut-off answer must close the connection, and the stop-short tests no longer race the clock
A new test fails on the current code: when the response time limit
runs out before rproxy can write the app's short body, and the visitor
has stopped a little short of its announced length, the connection
stays open until the request time limit instead of closing.

The test that failed under load used a 300 ms response time limit,
which a busy host can use up between the app's headers and its body.
It and the test that answers a visitor that stops sending now use
limits far longer than the test's 10-second read deadline.

Model: opus-5-5
2026-11-11 02:34:56 +00:00
2 changed files with 85 additions and 11 deletions

No files matched your search

+65 -5
View File
@@ -12,6 +12,7 @@ import (
"os" "os"
"strconv" "strconv"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
"time" "time"
@@ -807,8 +808,10 @@ func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing.
app := startApp(t, func(w http.ResponseWriter, _ *http.Request) { app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, greeting) _, _ = io.WriteString(w, greeting)
}) })
// The request time limit is far longer than dial's deadline, so that a busy
// host cannot use it up before the app's answer arrives.
addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
maxRequestTime: "3000", maxRequestTime: "120000",
}) })
// Far more announced than sent, so that Go's server answers for the app // Far more announced than sent, so that Go's server answers for the app
@@ -869,11 +872,11 @@ func TestAVisitorThatStopsALittleShortGetsTheAppsAnswerAtOnce(t *testing.T) {
standIn := startWAF(t, allow) standIn := startWAF(t, allow)
app := startApp(t, answerBeforeTheBody) app := startApp(t, answerBeforeTheBody)
// The request time limit is longer than the 10 seconds the status and headers // Both time limits are far longer than dial's deadline, so that the answer
// of an answer have to be sent, and than dial's deadline. // has to come before either runs out, and a busy host cannot use them up.
addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
maxRequestTime: "15000", maxRequestTime: "120000",
maxResponseTime: "300", maxResponseTime: "60000",
}) })
conn := dial(t, addr) conn := dial(t, addr)
@@ -910,6 +913,63 @@ func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing.
} }
} }
func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOff(
t *testing.T,
) {
t.Parallel()
const limit = 300 * time.Millisecond
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.
app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
rc := http.NewResponseController(w)
_ = rc.EnableFullDuplex()
w.Header().Set(contentLengthHeader, strconv.Itoa(len(greeting)))
_ = rc.Flush()
<-headersRead
time.Sleep(limit)
_, _ = io.WriteString(w, greeting)
_ = 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",
maxResponseTime: strconv.FormatInt(limit.Milliseconds(), 10),
})
conn := dial(t, addr)
sendAllButALittle(t, conn)
res, err := http.ReadResponse(bufio.NewReader(conn), nil)
if err != nil {
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)
}
}
func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) { func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) {
t.Parallel() t.Parallel()
+20 -6
View File
@@ -195,6 +195,20 @@ func (rq *request) forward(ctx context.Context) {
rq.startRequestTimer() rq.startRequestTimer()
rq.upstreamStart = time.Now() rq.upstreamStart = time.Now()
reverseProxy.ServeHTTP(rq.out, out) reverseProxy.ServeHTTP(rq.out, out)
// ReverseProxy ignores a failed flush, so it misses a short answer cut off when
// SWRP_MAX_RESPONSE_TIME_MS runs out; once a flush has failed, every later one
// fails too. Go's server would then keep the connection open while it reads
// the rest of a body that has not arrived whole. So that reading ends now, and
// the refusal keeps finish from extending it to SWRP_MAX_REQUEST_TIME_MS. The
// panic has Go's server close the connection, and log nothing; the deferred
// finish still runs.
if rq.out.status != http.StatusSwitchingProtocols &&
errors.Is(rq.rc.Flush(), os.ErrDeadlineExceeded) {
rq.refuse(http.StatusGatewayTimeout)
rq.stopPassingOnBody(time.Now())
panic(http.ErrAbortHandler)
}
} }
// rewrite makes the request the app receives: the visitor's request, unchanged, // rewrite makes the request the app receives: the visitor's request, unchanged,
@@ -385,7 +399,7 @@ func (rq *request) finish(ctx context.Context) {
rq.stopTimers() rq.stopTimers()
if rq.refused.Load() == 0 { if rq.refused.Load() == 0 {
rq.stopPassingOnBody() rq.stopPassingOnBody(rq.requestDeadline())
} }
requestEnd := rq.endRequest() requestEnd := rq.endRequest()
@@ -473,10 +487,10 @@ func (rq *request) stopReadingBody(deadline time.Time) {
} }
// stopPassingOnBody ends the transport's reading of a visitor's body that has // 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 // not arrived whole, and lets Go's server read the rest until deadline. Go's
// SWRP_MAX_REQUEST_TIME_MS runs out. Go's server reads it once the handler // server reads it once the handler returns, but with no deadline at all if
// returns, but with no deadline at all if another read of it is still waiting. // another read of it is still waiting.
func (rq *request) stopPassingOnBody() { func (rq *request) stopPassingOnBody(deadline time.Time) {
if rq.body == nil || rq.body.received.Load() { if rq.body == nil || rq.body.received.Load() {
return return
} }
@@ -489,7 +503,7 @@ func (rq *request) stopPassingOnBody() {
rq.body.stopped = true rq.body.stopped = true
rq.body.mu.Unlock() rq.body.mu.Unlock()
rq.stopReadingBody(rq.requestDeadline()) rq.stopReadingBody(deadline)
} }
// startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the // startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the