Honour HeaderReceived's ReadTimeout per HTTP/2 stream

server.go

@@ -63,6 +63,7 @@ func (s *Server) ServeConn(c net.Conn) error {
 		writer:         make(chan *FrameHeader, 128),
 		reader:         make(chan *FrameHeader, 128),
 		maxRequestTime: s.s.ReadTimeout,
+		headerReceived: s.s.HeaderReceived,
 		maxIdleTime:    s.s.IdleTimeout,
 		maxBodySize:    s.s.MaxRequestBodySize,
 		streamBodies:   s.s.StreamRequestBody,

serverConn.go

@@ -108,6 +108,11 @@ type serverConn struct {
 
 	// maxRequestTime is the max time of a request over one single stream
 	maxRequestTime time.Duration
+	// headerReceived is fasthttp.Server.HeaderReceived. Only its
+	// ReadTimeout is honoured here: it replaces maxRequestTime for that one
+	// stream, measured from the stream's start. MaxRequestBodySize and
+	// WriteTimeout are not applied per stream.
+	headerReceived func(*fasthttp.RequestHeader) fasthttp.RequestConfig
 	pingInterval   time.Duration
 	// maxIdleTime is the max time a client can be connected without sending any REQUEST.
 	// As highlighted, PING/PONG frames are completely excluded.
@@ -409,7 +414,6 @@ func (sc *serverConn) handleStreams() {
 	}()
 
 	var strms Streams
-	var reqTimerArmed bool
 
 	// Handlers still reading a request body must not wait on a connection
 	// that is gone (Serve waits for every handler before it closes).
@@ -427,6 +431,48 @@ func (sc *serverConn) handleStreams() {
 		return !strm.processing || (strm.body != nil && strm.State() == StreamStateOpen)
 	}
 
+	// requestDeadline is when strm's request must have fully arrived: its
+	// own read timeout (headerReceived) or else the server's, from the
+	// stream's start. ok is false when neither is set.
+	requestDeadline := func(strm *Stream) (at time.Time, ok bool) {
+		d := sc.maxRequestTime
+		if strm.readTimeout > 0 {
+			d = strm.readTimeout
+		}
+		if d <= 0 {
+			return time.Time{}, false
+		}
+		return strm.startedAt.Add(d), true
+	}
+
+	// armRequestTimer arms the request timer for the earliest deadline of
+	// the streams still receiving their request, or leaves it disarmed when
+	// none is. A stale fire of the timer is harmless: nothing is due yet,
+	// and the timer re-arms from here.
+	armRequestTimer := func() {
+		var next time.Time
+		for _, strm := range strms {
+			if strm.origType != FrameHeaders || !receiving(strm) {
+				continue
+			}
+			if at, ok := requestDeadline(strm); ok && (next.IsZero() || at.Before(next)) {
+				next = at
+			}
+		}
+		if next.IsZero() {
+			return
+		}
+		when := time.Until(next)
+		if when < time.Millisecond {
+			when = time.Millisecond
+		}
+		sc.maxRequestTimer.Reset(when)
+
+		if sc.debug {
+			sc.logger.Printf("Next request will timeout in %f seconds\n", when.Seconds())
+		}
+	}
+
 	closeStream := func(strm *Stream) {
 		if strm.origType == FrameHeaders {
 			sc.freeSlot(strm)
@@ -500,20 +546,19 @@ loop:
 		case <-sc.closer:
 			break loop
 		case <-sc.maxRequestTimer.C:
-			reqTimerArmed = false
-
 			// maxRequestTime is a read timeout: it only applies to streams
 			// still receiving their request, a streamed body included.
 			// Streams whose request has fully arrived (long responses,
 			// SSE, gRPC streaming) are never timed out here — the handler
 			// owns its own lifetime.
 			var due []*Stream
 			for _, strm := range strms {
-				// the request is due if the startedAt time + maxRequestTime is in the past
-				isDue := time.Now().After(
-					strm.startedAt.Add(sc.maxRequestTime))
-				if !isDue {
-					break
+				// the request is due if its read deadline is in the past.
+				// Deadlines differ per stream (headerReceived), so streams
+				// are not in deadline order: check every one.
+				at, ok := requestDeadline(strm)
+				if !ok || !time.Now().After(at) {
+					continue
 				}
 
 				if !receiving(strm) {
@@ -549,25 +594,7 @@ loop:
 			// Pick the oldest not-yet-processing stream instead, clamp to a
 			// positive deadline, and disarm if none remain so the next new
 			// stream re-arms via the reader branch.
-			reqTimerArmed = false
-			if sc.maxRequestTime > 0 {
-				for _, strm := range strms {
-					if strm.origType != FrameHeaders || !receiving(strm) {
-						continue
-					}
-					reqTimerArmed = true
-					when := strm.startedAt.Add(sc.maxRequestTime).Sub(time.Now())
-					if when < time.Millisecond {
-						when = time.Millisecond
-					}
-					sc.maxRequestTimer.Reset(when)
-
-					if sc.debug {
-						sc.logger.Printf("Next request will timeout in %f seconds\n", when.Seconds())
-					}
-					break
-				}
-			}
+			armRequestTimer()
 		case strm := <-sc.done:
 			// A handler goroutine finished writing its response.
 			strm.processing = false
@@ -671,13 +698,11 @@ loop:
 					sc.logger.Printf("Stream %d created. Open streams: %d\n", strm.ID(), atomic.LoadInt32(&sc.openSlots))
 				}
 
-				if !reqTimerArmed && sc.maxRequestTime > 0 {
-					reqTimerArmed = true
-					sc.maxRequestTimer.Reset(sc.maxRequestTime)
-
-					if sc.debug {
-						sc.logger.Printf("Next request will timeout in %f seconds\n", sc.maxRequestTime.Seconds())
-					}
+				// Re-armed for every new stream: one with a longer read
+				// timeout (headerReceived) may hold the timer past this
+				// stream's deadline.
+				if sc.maxRequestTime > 0 {
+					armRequestTimer()
 				}
 			}
 
@@ -720,7 +745,13 @@ loop:
 				}
 			}
 
-			if err := sc.handleFrame(strm, fr); err != nil {
+			readTimeout := strm.readTimeout
+			err := sc.handleFrame(strm, fr)
+			if strm.readTimeout != readTimeout {
+				// headerReceived gave this request its own read timeout.
+				armRequestTimer()
+			}
+			if err != nil {
 				if errors.Is(err, errRequestBodyTooLarge) {
 					// Respond 413 and stop the upload with a complete
 					// response + RST_STREAM(NO_ERROR) (RFC 7540 §8.1).
@@ -940,6 +971,9 @@ func (sc *serverConn) handleFrame(strm *Stream, fr *FrameHeader) error {
 			// request alone.
 			if !strm.processing {
 				strm.ctx.Request.URI().SetSchemeBytes(strm.scheme)
+				if sc.headerReceived != nil {
+					strm.readTimeout = sc.headerReceived(&strm.ctx.Request.Header).ReadTimeout
+				}
 			}
 		}
 	case FrameData:

server_readtimeout_test.go

@@ -0,0 +1,83 @@
+package http2
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/valyala/fasthttp"
+)
+
+// slowBody yields n bytes, one small chunk per tick, so an upload takes
+// about n/chunk ticks.
+type slowBody struct {
+	left int
+	tick time.Duration
+}
+
+func (b *slowBody) Read(p []byte) (int, error) {
+	if b.left <= 0 {
+		return 0, io.EOF
+	}
+	time.Sleep(b.tick)
+	n := 1024
+	if n > b.left {
+		n = b.left
+	}
+	if n > len(p) {
+		n = len(p)
+	}
+	copy(p, bytes.Repeat([]byte{'x'}, n))
+	b.left -= n
+	return n, nil
+}
+
+// TestHeaderReceivedReadTimeoutPerStream: fasthttp.Server.HeaderReceived
+// gives one request its own read timeout over HTTP/2, as it does over
+// HTTP/1.1. A slow upload on the path it extends completes; the same upload
+// elsewhere is cut at the server's ReadTimeout.
+func TestHeaderReceivedReadTimeoutPerStream(t *testing.T) {
+	const serverTimeout = 300 * time.Millisecond
+
+	c := startH2(t, &fasthttp.Server{
+		Handler: func(ctx *fasthttp.RequestCtx) {
+			n, err := io.Copy(io.Discard, bodyOf(ctx))
+			if err != nil {
+				ctx.Error(err.Error(), fasthttp.StatusBadRequest)
+				return
+			}
+			ctx.SetBodyString(time.Duration(n).String())
+		},
+		StreamRequestBody: true,
+		ReadTimeout:       serverTimeout,
+		HeaderReceived: func(h *fasthttp.RequestHeader) fasthttp.RequestConfig {
+			if string(h.RequestURI()) == "/slow" {
+				return fasthttp.RequestConfig{ReadTimeout: 10 * time.Second}
+			}
+			return fasthttp.RequestConfig{}
+		},
+	}, ServerConfig{})
+
+	upload := func(path string) (int, error) {
+		// ~1 s of upload: over three times the server's ReadTimeout.
+		body := &slowBody{left: 20 * 1024, tick: 50 * time.Millisecond}
+		req, _ := http.NewRequest("POST", "http://localhost"+path, body)
+		req.ContentLength = int64(body.left)
+		res, err := c.Do(req)
+		if err != nil {
+			return 0, err
+		}
+		defer res.Body.Close()
+		_, _ = io.Copy(io.Discard, res.Body)
+		return res.StatusCode, nil
+	}
+
+	if status, err := upload("/slow"); err != nil || status != http.StatusOK {
+		t.Fatalf("upload on the extended path: status %d, err %v; want 200", status, err)
+	}
+	if status, err := upload("/other"); err == nil && status == http.StatusOK {
+		t.Fatal("upload past the server's ReadTimeout succeeded on a path HeaderReceived did not extend")
+	}
+}

stream.go

@@ -50,6 +50,10 @@ type Stream struct {
 	startedAt       time.Time
 	headersFinished bool
 
+	// readTimeout overrides the server's ReadTimeout for this one request
+	// (fasthttp.Server.HeaderReceived); 0 keeps the server's.
+	readTimeout time.Duration
+
 	// processing is true while the request handler runs in its own
 	// goroutine. Owned by the handleStreams loop.
 	processing bool
@@ -87,6 +91,7 @@ func NewStream(id uint32, win int32) *Stream {
 	strm.state = StreamStateIdle
 	strm.headersFinished = false
 	strm.startedAt = time.Time{}
+	strm.readTimeout = 0
 	strm.previousHeaderBytes = strm.previousHeaderBytes[:0]
 	strm.ctx = nil
 	strm.scheme = []byte("https")