caddyhttp: preserve expired request-body read deadlines (#8178)

* caddyhttp: preserve expired request-body read deadlines

* caddyhttp: retain only expired drain deadlines

Track the last successfully installed read deadline and retain it after a terminal timeout only when draining is enabled and the deadline has expired. Failed deadline changes preserve the previous installed value.

Cover early and wrapped timeouts, non-draining requests, failed deadline installation and replacement, and cleanup retries.

* caddyhttp: shorten installed deadline field name

caddytest/integration/request_body_timeout_test.go

@@ -0,0 +1,133 @@
+package integration
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/caddyserver/caddy/v2/caddytest"
+)
+
+// A timed-out upload must get an error response without sending the remainder
+// of its body. HTTP/1 drains unread bodies smaller than 256 KiB before replying;
+// clearing the expired read deadline makes that drain wait for the client.
+func TestReverseProxyRequestBodyTimeout(t *testing.T) {
+	for _, proto := range []int{1, 2} {
+		for _, tc := range []struct {
+			name     string
+			timeouts string
+			trickle  bool
+			handler  string
+		}{
+			{"hard stalled", `"read_timeout": 300000000, "read_idle_timeout": 60000000000`, false, ""},
+			{"hard trickled", `"read_timeout": 300000000, "read_idle_timeout": 60000000000`, true, ""},
+			{"idle stalled", `"read_idle_timeout": 300000000`, false, ""},
+			{"minimum rate trickled", `"read_idle_timeout": 300000000, "read_min_rate": 65536`, true, ""},
+			{"route stalled", `"read_idle_timeout": 60000000000`, false, `{"handler": "timeouts", "read_timeout": 300000000},`},
+		} {
+			t.Run(fmt.Sprintf("HTTP%d/%s", proto, tc.name), func(t *testing.T) {
+				type uploadResult struct {
+					n   int64
+					err error
+				}
+				upstreamRead := make(chan uploadResult, 1)
+				upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					n, err := io.Copy(io.Discard, r.Body)
+					upstreamRead <- uploadResult{n, err}
+					w.WriteHeader(http.StatusNoContent)
+				}))
+				defer upstream.Close()
+				tester := caddytest.NewTester(t)
+				tester.InitServer(fmt.Sprintf(`{
+                    "admin": {"listen": "localhost:2999", "config": {"persist": false}},
+                    "apps": {
+                        "tls": {"certificates": {"load_files": [{
+                            "certificate": "/caddy.localhost.crt", "key": "/caddy.localhost.key"
+                        }]}},
+                        "http": {"grace_period": 1, "servers": {"srv0": {
+                            "listen": ["127.0.0.1:9443"],
+                            "automatic_https": {"disable": true},
+                            "tls_connection_policies": [{}],
+                            %s,
+                            "routes": [{"handle": [%s{"handler": "reverse_proxy",
+                                "upstreams": [{"dial": %q}]
+                            }]}]
+                        }}}
+                    }
+                }`, tc.timeouts, tc.handler, upstream.Listener.Addr().String()), "json")
+				if t.Failed() {
+					t.FailNow()
+				}
+				transport := tester.Client.Transport.(*http.Transport)
+				if proto == 1 {
+					transport.ForceAttemptHTTP2 = false
+				}
+				defer transport.CloseIdleConnections()
+				ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+				defer cancel()
+				pr, pw := io.Pipe()
+				defer pr.Close()
+				defer pw.Close()
+				stopClose := context.AfterFunc(ctx, func() { pr.CloseWithError(ctx.Err()) })
+				defer stopClose()
+				writerDone := make(chan struct{})
+				go func() {
+					defer close(writerDone)
+					if _, err := pw.Write(make([]byte, 1024)); err != nil {
+						return
+					}
+					if !tc.trickle {
+						return // Leave the pipe open so the request body stalls.
+					}
+					ticker := time.NewTicker(25 * time.Millisecond)
+					defer ticker.Stop()
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						case <-ticker.C:
+							if _, err := pw.Write([]byte("x")); err != nil {
+								return
+							}
+						}
+					}
+				}()
+				defer func() {
+					cancel()
+					pr.Close()
+					<-writerDone
+				}()
+				req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://a.caddy.localhost:9443", pr)
+				if err != nil {
+					t.Fatal(err)
+				}
+				req.ContentLength = 100 << 10 // Below net/http's post-handler drain limit.
+				resp, err := tester.Client.Do(req)
+				if err != nil {
+					t.Fatalf("did not receive a response to the timed-out upload: %v", err)
+				}
+				defer resp.Body.Close()
+				if resp.ProtoMajor != proto {
+					t.Fatalf("got HTTP/%d, want HTTP/%d", resp.ProtoMajor, proto)
+				}
+				// A read timeout can also cancel the request context, so the proxy
+				// can classify it as a canceled request rather than a gateway error.
+				if resp.StatusCode != 499 && resp.StatusCode != http.StatusBadGateway && resp.StatusCode != http.StatusGatewayTimeout {
+					t.Errorf("unexpected timeout response status: %d", resp.StatusCode)
+				}
+				select {
+				case result := <-upstreamRead:
+					if result.err == nil || result.n < 1024 || result.n >= req.ContentLength {
+						t.Errorf("upstream upload was not interrupted: read %d bytes, error %v", result.n, result.err)
+					}
+				case <-ctx.Done():
+					t.Fatal("upstream upload was not interrupted")
+				}
+			})
+		}
+	}
+}

modules/caddyhttp/idletimeout.go

@@ -16,8 +16,10 @@ package caddyhttp
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net/http"
+	"os"
 	"sync"
 	"time"
 
@@ -106,8 +108,10 @@ func (d *IdleDeadline) restore(o idleOverride) {
 // IdleTimeoutReader wraps a request body with IdleDeadline, resetting
 // the read deadline before every Read call instead of bounding the
 // whole body transfer with a single hard deadline.
-// The deadline is cleared after an error (most likely io.EOF) is encountered when reading the body
-// to prevent context canceled error for h1 requests.
+// The deadline is cleared after a non-timeout error (most likely io.EOF) is
+// encountered when reading the body to prevent context canceled errors for h1
+// requests. An expired deadline is kept when DrainDeadline is set to bound
+// net/http's post-handler drain.
 // see: https://github.com/caddyserver/caddy/issues/8103
 type IdleTimeoutReader struct {
 	io.ReadCloser
@@ -136,6 +140,7 @@ type IdleTimeoutReader struct {
 	mu          sync.Mutex
 	unsupported bool
 	armed       armedDeadline
+	installed   time.Time // Last deadline successfully installed through Ctrl.
 	finished    bool
 	terminalErr error
 }
@@ -186,8 +191,9 @@ func (r *IdleTimeoutReader) Read(p []byte) (int, error) {
 // HandlerDone prevents later body reads from using the response controller.
 // If the body is not finished and DrainDeadline is set, it leaves an idle
 // deadline armed for net/http's post-handler drain. Otherwise, it clears a
-// deadline left from a finished body or put back between reads. It must be
-// called before the installing handler returns.
+// deadline left from a finished body or put back between reads, unless the
+// body timed out with an expired drain deadline. It must be called before the
+// installing handler returns.
 func (r *IdleTimeoutReader) HandlerDone() {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -227,7 +233,7 @@ func (r *IdleTimeoutReader) Override(timeout time.Duration, minRate int64) (rest
 
 // rearmLocked moves an armed deadline to what the current settings give.
 func (r *IdleTimeoutReader) rearmLocked() {
-	if r.armed == deadlineIdle && !r.finished && !r.unsupported {
+	if r.armed == deadlineIdle && !r.finished && !r.unsupported && r.terminalErr == nil {
 		r.setDeadlineLocked("could not set read deadline")
 	}
 }
@@ -249,6 +255,7 @@ func (r *IdleTimeoutReader) setReadDeadlineLocked(deadline time.Time, logMessage
 		}
 		return false
 	}
+	r.installed = deadline
 	return true
 }
 
@@ -278,6 +285,13 @@ func (r *IdleTimeoutReader) clearDeadlineLocked() {
 	if r.armed == deadlineNone {
 		return
 	}
+	if r.DrainDeadline && !r.installed.IsZero() && !time.Now().Before(r.installed) &&
+		errors.Is(r.terminalErr, os.ErrDeadlineExceeded) {
+		// HTTP/1 drains a small unread body before writing the response.
+		// Clearing its expired deadline would let a stalled client block
+		// that drain, and therefore the error response, indefinitely.
+		return
+	}
 	if !r.setReadDeadlineLocked(time.Time{}, "could not clear read deadline") {
 		return
 	}

modules/caddyhttp/idletimeout_test.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"testing/iotest"
 	"time"
 
 	"github.com/stretchr/testify/assert"
@@ -146,6 +147,133 @@ func TestIdleTimeoutReaderClearsDeadlineAtEOF(t *testing.T) {
 	assert.True(t, deadlines[len(deadlines)-1].IsZero())
 }
 
+func TestIdleTimeoutReaderTerminalError(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		err          error
+		keepDeadline bool
+	}{
+		{"timeout", os.ErrDeadlineExceeded, true},
+		{"wrapped timeout", fmt.Errorf("read body: %w", os.ErrDeadlineExceeded), true},
+		{"EOF", io.EOF, false},
+		{"other error", errors.New("read failed"), false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			deadline := time.Now().Add(-time.Second)
+			w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+			r := &IdleTimeoutReader{
+				ReadCloser:    io.NopCloser(iotest.ErrReader(tc.err)),
+				Ctrl:          http.NewResponseController(w),
+				Deadline:      IdleDeadline{Timeout: time.Hour, HardDeadline: deadline},
+				Logger:        zap.NewNop(),
+				DrainDeadline: true,
+			}
+			_, err := r.Read(make([]byte, 1))
+			require.ErrorIs(t, err, tc.err)
+			want := time.Time{}
+			if tc.keepDeadline {
+				want = deadline
+			}
+			deadlines := w.snapshot()
+			require.NotEmpty(t, deadlines)
+			assert.Equal(t, want, deadlines[len(deadlines)-1], "after Read")
+			r.HandlerDone()
+			deadlines = w.snapshot()
+			assert.Equal(t, want, deadlines[len(deadlines)-1], "after HandlerDone")
+
+			_, err = r.Read(make([]byte, 1))
+			require.ErrorIs(t, err, tc.err)
+			r.HandlerDone()
+			assert.Equal(t, deadlines, w.snapshot(), "late reads and repeated completion must not change deadlines")
+		})
+	}
+}
+
+// Only an expired deadline successfully installed for a draining request may
+// survive a timeout; a failed replacement must not change that decision.
+func TestIdleTimeoutReaderTimeoutDeadlineCleanup(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		drain       bool
+		initial     time.Duration
+		replacement time.Duration
+		failAt      int
+		wrapped     bool
+		wantClear   bool
+		wantCalls   int
+	}{
+		{name: "early timeout", drain: true, initial: time.Hour, wantClear: true, wantCalls: 2},
+		{name: "wrapped early timeout", drain: true, initial: time.Hour, wrapped: true, wantClear: true, wantCalls: 2},
+		{name: "expired without drain", initial: -time.Hour, wantClear: true, wantCalls: 2},
+		{name: "initial installation fails", drain: true, initial: -time.Hour, failAt: 1, wantCalls: 1},
+		{name: "failed expired replacement", drain: true, initial: time.Hour, replacement: -time.Hour, failAt: 2, wantClear: true, wantCalls: 3},
+		{name: "failed future replacement", drain: true, initial: -time.Hour, replacement: time.Hour, failAt: 2, wantCalls: 2},
+		{name: "failed clear retries at completion", drain: true, initial: time.Hour, failAt: 2, wantClear: true, wantCalls: 3},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder(), failAt: tc.failAt}
+			terminalErr := error(os.ErrDeadlineExceeded)
+			if tc.wrapped {
+				terminalErr = fmt.Errorf("read body: %w", terminalErr)
+			}
+			r := &IdleTimeoutReader{
+				ReadCloser: io.NopCloser(iotest.ErrReader(terminalErr)),
+				Ctrl:       http.NewResponseController(w),
+				Deadline:   IdleDeadline{Timeout: tc.initial},
+				Logger:     zap.NewNop(), DrainDeadline: tc.drain,
+			}
+			// Install once, then exercise a failed replacement during Read.
+			if tc.replacement != 0 {
+				r.setDeadlineLocked("initial deadline")
+				r.Deadline.Timeout = tc.replacement
+			}
+			_, err := r.Read(make([]byte, 1))
+			require.ErrorIs(t, err, terminalErr)
+			afterRead := w.snapshot()
+			if tc.wantClear {
+				require.True(t, afterRead[len(afterRead)-1].IsZero(), "must attempt clearing after Read")
+			}
+			r.HandlerDone()
+			deadlines := w.snapshot()
+			require.Len(t, deadlines, tc.wantCalls)
+			if tc.wantClear {
+				assert.True(t, deadlines[len(deadlines)-1].IsZero())
+			} else if tc.failAt != 1 {
+				assert.True(t, deadlines[0].Before(time.Now()), "the successfully installed deadline expired")
+			}
+			_, err = r.Read(make([]byte, 1))
+			require.ErrorIs(t, err, terminalErr)
+			r.HandlerDone()
+			assert.Equal(t, deadlines, w.snapshot(), "late reads and completion must leave deadlines alone")
+		})
+	}
+}
+
+func TestIdleTimeoutReaderTimeoutAfterOverride(t *testing.T) {
+	w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+	r := &IdleTimeoutReader{
+		ReadCloser:    io.NopCloser(iotest.ErrReader(os.ErrDeadlineExceeded)),
+		Ctrl:          http.NewResponseController(w),
+		Deadline:      IdleDeadline{Timeout: time.Hour},
+		Logger:        zap.NewNop(),
+		DrainDeadline: true,
+	}
+	restore := r.Override(-time.Second, 0)
+	_, err := r.Read(make([]byte, 1))
+	require.ErrorIs(t, err, os.ErrDeadlineExceeded)
+	deadlines := w.snapshot()
+	require.NotEmpty(t, deadlines)
+	expired := deadlines[0]
+	require.True(t, expired.Before(time.Now()))
+	assert.Equal(t, expired, deadlines[len(deadlines)-1])
+
+	// Unwinding a route-level timeout must not rearm the longer server timeout.
+	restore()
+	r.HandlerDone()
+	deadlines = w.snapshot()
+	assert.Equal(t, expired, deadlines[len(deadlines)-1])
+}
+
 func TestIdleTimeoutReaderHandlerDone(t *testing.T) {
 	body := &blockingReadCloser{
 		started: make(chan struct{}),