requestbody: idle-reset ReadTimeout/WriteTimeout, add MinRate/MaxWriteChunk

ReadTimeout/WriteTimeout set a single deadline once, so any transfer
running longer than the timeout got cut regardless of whether it was
actually stalled - the same bug the server-wide timeouts had before
switching to idle-reset. Reuse caddyhttp.IdleTimeoutReader/Writer here
too, giving per-route granularity nginx/Apache have via location/
directory scoping and Caddy's server-wide timeouts don't: a route
matching this handler can now set its own idle window independently
from the rest of the server block.

modules/caddyhttp/requestbody/caddyfile.go

@@ -15,6 +15,7 @@
 package requestbody
 
 import (
+	"strconv"
 	"time"
 
 	"github.com/dustin/go-humanize"
@@ -57,6 +58,17 @@ func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error)
 			}
 			rb.ReadTimeout = timeout
 
+		case "read_min_rate":
+			var rateStr string
+			if !h.AllArgs(&rateStr) {
+				return nil, h.ArgErr()
+			}
+			rate, err := strconv.ParseInt(rateStr, 10, 64)
+			if err != nil {
+				return nil, h.Errf("parsing read_min_rate: %v", err)
+			}
+			rb.ReadMinRate = rate
+
 		case "write_timeout":
 			var timeoutStr string
 			if !h.AllArgs(&timeoutStr) {
@@ -68,6 +80,28 @@ func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error)
 			}
 			rb.WriteTimeout = timeout
 
+		case "write_min_rate":
+			var rateStr string
+			if !h.AllArgs(&rateStr) {
+				return nil, h.ArgErr()
+			}
+			rate, err := strconv.ParseInt(rateStr, 10, 64)
+			if err != nil {
+				return nil, h.Errf("parsing write_min_rate: %v", err)
+			}
+			rb.WriteMinRate = rate
+
+		case "max_write_chunk":
+			var sizeStr string
+			if !h.AllArgs(&sizeStr) {
+				return nil, h.ArgErr()
+			}
+			size, err := humanize.ParseBytes(sizeStr)
+			if err != nil {
+				return nil, h.Errf("parsing max_write_chunk: %v", err)
+			}
+			rb.MaxWriteChunk = int(size)
+
 		case "set":
 			var setStr string
 			if !h.AllArgs(&setStr) {

modules/caddyhttp/requestbody/requestbody.go

@@ -22,7 +22,6 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 
 	"github.com/caddyserver/caddy/v2"
 	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
@@ -38,12 +37,40 @@ type RequestBody struct {
 	// If more bytes are read, an error with HTTP status 413 is returned.
 	MaxSize int64 `json:"max_size,omitempty"`
 
+	// How long to allow a read from the request body to stall before
+	// aborting the connection, reset on every successful read (like the
+	// server-wide read_idle_timeout, but scoped to routes matching this
+	// handler). If zero, no idle timeout is applied here.
 	// EXPERIMENTAL. Subject to change/removal.
 	ReadTimeout time.Duration `json:"read_timeout,omitempty"`
 
+	// ReadMinRate requires the client to sustain at least this many
+	// bytes/second, averaged from the start of the read, or the
+	// connection is aborted (Apache mod_reqtimeout's MinRate). Only
+	// takes effect if ReadTimeout is also set.
+	// EXPERIMENTAL. Subject to change/removal.
+	ReadMinRate int64 `json:"read_min_rate,omitempty"`
+
+	// How long to allow a write to the client to stall before aborting
+	// the connection, reset on every successful write (like the
+	// server-wide write_idle_timeout, but scoped to routes matching this
+	// handler). If zero, no idle timeout is applied here.
 	// EXPERIMENTAL. Subject to change/removal.
 	WriteTimeout time.Duration `json:"write_timeout,omitempty"`
 
+	// WriteMinRate is like ReadMinRate, but for writes to the client.
+	// Only takes effect if WriteTimeout is also set.
+	// EXPERIMENTAL. Subject to change/removal.
+	WriteMinRate int64 `json:"write_min_rate,omitempty"`
+
+	// MaxWriteChunk bounds how many bytes a single underlying write
+	// operation is allowed to cover, so WriteTimeout/WriteMinRate can
+	// actually apply between chunks of a large response instead of
+	// being bounded by one deadline for the whole thing. If zero,
+	// caddyhttp.DefaultMaxWriteChunk is used.
+	// EXPERIMENTAL. Subject to change/removal.
+	MaxWriteChunk int `json:"max_write_chunk,omitempty"`
+
 	// This field permit to replace body on the fly
 	// EXPERIMENTAL. Subject to change/removal.
 	Set string `json:"set,omitempty"`
@@ -86,18 +113,30 @@ func (rb RequestBody) ServeHTTP(w http.ResponseWriter, r *http.Request, next cad
 	if rb.ReadTimeout > 0 || rb.WriteTimeout > 0 {
 		//nolint:bodyclose
 		rc := http.NewResponseController(w)
+		start := time.Now()
 		if rb.ReadTimeout > 0 {
-			if err := rc.SetReadDeadline(time.Now().Add(rb.ReadTimeout)); err != nil {
-				if c := rb.logger.Check(zapcore.ErrorLevel, "could not set read deadline"); c != nil {
-					c.Write(zap.Error(err))
-				}
+			r.Body = &caddyhttp.IdleTimeoutReader{
+				ReadCloser: r.Body,
+				Ctrl:       rc,
+				Deadline: caddyhttp.IdleDeadline{
+					Start:   start,
+					Timeout: rb.ReadTimeout,
+					MinRate: rb.ReadMinRate,
+				},
+				Logger: rb.logger,
 			}
 		}
 		if rb.WriteTimeout > 0 {
-			if err := rc.SetWriteDeadline(time.Now().Add(rb.WriteTimeout)); err != nil {
-				if c := rb.logger.Check(zapcore.ErrorLevel, "could not set write deadline"); c != nil {
-					c.Write(zap.Error(err))
-				}
+			w = &caddyhttp.IdleTimeoutWriter{
+				ResponseWriterWrapper: &caddyhttp.ResponseWriterWrapper{ResponseWriter: w},
+				Ctrl:                  rc,
+				Deadline: caddyhttp.IdleDeadline{
+					Start:   start,
+					Timeout: rb.WriteTimeout,
+					MinRate: rb.WriteMinRate,
+				},
+				MaxChunk: rb.MaxWriteChunk,
+				Logger:   rb.logger,
 			}
 		}
 	}

modules/caddyhttp/requestbody/requestbody_test.go

@@ -0,0 +1,108 @@
+// Copyright 2015 Matthew Holt and The Caddy Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package requestbody
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/caddyserver/caddy/v2/modules/caddyhttp"
+)
+
+// pacedReader emits chunkCount chunks of chunkSize bytes, sleeping delay
+// before each one, simulating a client that trickles a request body.
+type pacedReader struct {
+	delay      time.Duration
+	chunkSize  int
+	chunkCount int
+}
+
+func (p *pacedReader) Read(b []byte) (int, error) {
+	if p.chunkCount <= 0 {
+		return 0, io.EOF
+	}
+	time.Sleep(p.delay)
+	p.chunkCount--
+	n := copy(b, make([]byte, p.chunkSize))
+	return n, nil
+}
+
+// noError adapts a caddyhttp.Handler to a plain http.Handler for httptest.NewServer.
+func noError(h caddyhttp.Handler) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if err := h.ServeHTTP(w, r); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+	}
+}
+
+func TestRequestBody_ReadTimeoutIsIdleReset(t *testing.T) {
+	const timeout = 150 * time.Millisecond
+
+	rb := RequestBody{ReadTimeout: timeout}
+	rb.logger = zap.NewNop()
+
+	srv := httptest.NewServer(noError(caddyhttp.HandlerFunc(func(w http.ResponseWriter, r *http.Request) error {
+		return rb.ServeHTTP(w, r, caddyhttp.HandlerFunc(func(w http.ResponseWriter, r *http.Request) error {
+			_, err := io.Copy(io.Discard, r.Body)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusRequestTimeout)
+				return nil
+			}
+			w.WriteHeader(http.StatusOK)
+			return nil
+		}))
+	})))
+	defer srv.Close()
+
+	// each gap is well under timeout, but the cumulative transfer time
+	// is well over it; a hard (non-idle-reset) deadline would kill this
+	body := &pacedReader{delay: timeout / 4, chunkSize: 8, chunkCount: 8}
+	resp, err := http.Post(srv.URL, "application/octet-stream", body)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+}
+
+func TestRequestBody_WriteMaxChunkOverride(t *testing.T) {
+	const size = 10000
+	const maxChunk = 100
+
+	rb := RequestBody{WriteTimeout: time.Second, MaxWriteChunk: maxChunk}
+	rb.logger = zap.NewNop()
+
+	srv := httptest.NewServer(noError(caddyhttp.HandlerFunc(func(w http.ResponseWriter, r *http.Request) error {
+		return rb.ServeHTTP(w, r, caddyhttp.HandlerFunc(func(w http.ResponseWriter, r *http.Request) error {
+			_, err := w.Write(make([]byte, size))
+			return err
+		}))
+	})))
+	defer srv.Close()
+
+	resp, err := http.Get(srv.URL)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+
+	n, err := io.Copy(io.Discard, resp.Body)
+	require.NoError(t, err)
+	assert.EqualValues(t, size, n)
+}