mirror of
https://github.com/caddyserver/caddy.git
synced 2026-11-11 19:37:55 +00:00
reverseproxy: keep empty unknown-length request bodies retryable
The consumed-stream retry guard treated any read of a streaming request body as consumption. Go's transport probes unknown-length bodies of GET, HEAD, and similar requests before sending them, and HTTP/3 reports bodyless GET and HEAD requests with an unknown length, so every such request lost its retries once an upstream connection was established. A first read that returns io.EOF without data no longer counts as consumption. Later attempts send http.NoBody, while bodies that yield data, fail, or are interrupted mid-read remain non-retryable. Follow-up to de1bcf6d-457c-4348-9d4e-5eeb41cc120b. Code written by GitHub Copilot (GPT-6.1 Sol, Claude Opus 5.5); reviewed by mholt. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
1 parent
0c7629e52b
commit
7a58576d34
2 files changed
+107
-8
No files matched your search
@@ -341,6 +341,68 @@ func TestResponseRetryStatusCode(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestResponseRetryUnknownLengthEmptyBody verifies that a bodyless request with
|
||||
// an unknown length (as HTTP/3 GET requests have) remains retryable even though
|
||||
// the transport probes its body, while a non-empty streaming body does not.
|
||||
func TestResponseRetryUnknownLengthEmptyBody(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
body string
|
||||
wantCode int
|
||||
wantBad int32
|
||||
wantGood int32
|
||||
}{
|
||||
{body: "", wantCode: http.StatusOK, wantBad: 1, wantGood: 1},
|
||||
{body: "synthetic", wantCode: http.StatusBadGateway, wantBad: 1, wantGood: 0},
|
||||
} {
|
||||
t.Run(fmt.Sprintf("body=%q", tc.body), func(t *testing.T) {
|
||||
var badCalls, goodCalls atomic.Int32
|
||||
badServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
badCalls.Add(1)
|
||||
w.WriteHeader(http.StatusBadGateway)
|
||||
}))
|
||||
t.Cleanup(badServer.Close)
|
||||
goodServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
goodCalls.Add(1)
|
||||
if got, err := io.ReadAll(r.Body); err != nil || len(got) != 0 {
|
||||
t.Errorf("good upstream body = %q, %v", got, err)
|
||||
}
|
||||
w.Write([]byte("ok"))
|
||||
}))
|
||||
t.Cleanup(goodServer.Close)
|
||||
|
||||
retryMatch := caddyhttp.MatcherSets{
|
||||
caddyhttp.MatcherSet{
|
||||
newExpressionMatcher(t, "{http.reverse_proxy.status_code} == 502"),
|
||||
},
|
||||
}
|
||||
// RoundRobin picks index 1 first, then 0
|
||||
h := minimalHandlerWithRetryMatch(1, retryMatch,
|
||||
&Upstream{Host: new(Host), Dial: goodServer.Listener.Addr().String()},
|
||||
&Upstream{Host: new(Host), Dial: badServer.Listener.Addr().String()})
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "http://example.com/", io.NopCloser(strings.NewReader(tc.body)))
|
||||
req.ContentLength = -1
|
||||
req = prepareTestRequest(req)
|
||||
rec := httptest.NewRecorder()
|
||||
err := h.ServeHTTP(rec, req, caddyhttp.HandlerFunc(func(http.ResponseWriter, *http.Request) error {
|
||||
return nil
|
||||
}))
|
||||
|
||||
gotCode := rec.Code
|
||||
if herr, ok := err.(caddyhttp.HandlerError); ok {
|
||||
gotCode = herr.StatusCode
|
||||
}
|
||||
if gotCode != tc.wantCode {
|
||||
t.Errorf("status = %d, want %d (err=%v)", gotCode, tc.wantCode, err)
|
||||
}
|
||||
if badCalls.Load() != tc.wantBad || goodCalls.Load() != tc.wantGood {
|
||||
t.Errorf("upstream calls: bad=%d good=%d, want bad=%d good=%d",
|
||||
badCalls.Load(), goodCalls.Load(), tc.wantBad, tc.wantGood)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestResponseRetryHeader verifies that response header matching triggers
|
||||
// retries via a CEL expression checking {rp.header.*}
|
||||
func TestResponseRetryHeader(t *testing.T) {
|
||||
|
||||
@@ -170,7 +170,7 @@ type Handler struct {
|
||||
// only when the request already meets the transport's framing requirements.
|
||||
// Only fully buffered bodies can be replayed after an upstream reads them.
|
||||
// Bodies reaching a positive limit remain streaming and cannot be retried
|
||||
// after reading starts; -1 buffers without a size limit.
|
||||
// after data is read from them; -1 buffers without a size limit.
|
||||
RequestBuffers int64 `json:"request_buffers,omitempty"`
|
||||
|
||||
// If nonzero, the entire response body up to this size will be read
|
||||
@@ -598,11 +598,15 @@ func (h *Handler) Cleanup() error {
|
||||
|
||||
// bodyNopCloserIfNotRead wraps a request body to prevent closing if not read, i.e., when
|
||||
// dialing to upstream fails.
|
||||
// It will close the body as normal if the body is read.
|
||||
// It will close the body as normal if the body is read. A body that is empty on
|
||||
// its first read (such as an HTTP/3 GET, which has an unknown length and is
|
||||
// probed by the transport) is not considered consumed, so it remains retryable.
|
||||
type bodyNopCloserIfNotRead struct {
|
||||
io.ReadCloser
|
||||
mu sync.Mutex
|
||||
read bool
|
||||
reading int // reads in progress
|
||||
consumed bool // data, a read error, or an interrupted read was observed
|
||||
empty bool // EOF was reached without consuming any data
|
||||
closed bool
|
||||
closeOnce sync.Once
|
||||
closeErr error
|
||||
@@ -614,28 +618,56 @@ func (b *bodyNopCloserIfNotRead) Read(p []byte) (int, error) {
|
||||
b.mu.Unlock()
|
||||
return 0, net.ErrClosed
|
||||
}
|
||||
b.read = true
|
||||
if b.empty {
|
||||
b.mu.Unlock()
|
||||
return 0, io.EOF
|
||||
}
|
||||
b.reading++
|
||||
b.mu.Unlock()
|
||||
return b.ReadCloser.Read(p)
|
||||
|
||||
n, err := b.ReadCloser.Read(p)
|
||||
|
||||
b.mu.Lock()
|
||||
b.reading--
|
||||
if n > 0 || (err != nil && err != io.EOF) {
|
||||
b.consumed = true
|
||||
} else if err == io.EOF && !b.consumed {
|
||||
b.empty = true
|
||||
}
|
||||
b.mu.Unlock()
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (b *bodyNopCloserIfNotRead) Close() error {
|
||||
b.closeOnce.Do(func() {
|
||||
b.mu.Lock()
|
||||
b.closed = true
|
||||
read := b.read
|
||||
// closing the stream is the only way to interrupt a blocked read,
|
||||
// after which the stream can no longer be replayed
|
||||
if b.reading > 0 {
|
||||
b.consumed = true
|
||||
}
|
||||
consumed := b.consumed
|
||||
b.mu.Unlock()
|
||||
if read {
|
||||
if consumed {
|
||||
b.closeErr = b.ReadCloser.Close()
|
||||
}
|
||||
})
|
||||
return b.closeErr
|
||||
}
|
||||
|
||||
// didRead reports whether the stream was (or may have been) partially consumed.
|
||||
func (b *bodyNopCloserIfNotRead) didRead() bool {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.read
|
||||
return b.consumed
|
||||
}
|
||||
|
||||
// isEmpty reports whether the stream ended without yielding any data.
|
||||
func (b *bodyNopCloserIfNotRead) isEmpty() bool {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.empty && !b.consumed
|
||||
}
|
||||
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
|
||||
@@ -726,6 +758,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
|
||||
var attemptBody *bodyNopCloserIfNotRead
|
||||
if bufferedReqBody != nil {
|
||||
clonedReq.Body = &bufferedBodyReader{owner: bufferedReqBody}
|
||||
} else if streamingReqBody == http.NoBody {
|
||||
clonedReq.Body = http.NoBody
|
||||
} else if streamingReqBody != nil {
|
||||
attemptBody = &bodyNopCloserIfNotRead{ReadCloser: streamingReqBody}
|
||||
clonedReq.Body = attemptBody
|
||||
@@ -739,6 +773,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
|
||||
attemptBody.Close()
|
||||
if attemptBody.didRead() {
|
||||
done = true
|
||||
} else if attemptBody.isEmpty() {
|
||||
// the stream is drained but nothing was lost; replay it as empty
|
||||
streamingReqBody = http.NoBody
|
||||
}
|
||||
}
|
||||
if done {
|
||||
|
||||
Reference in new issue
Block a user