From e66caa3e7157c3be8f16d9c6e681642eaf58d284 Mon Sep 17 00:00:00 2001 From: Alma Tuck Date: Wed, 7 Oct 2026 15:00:35 -0600 Subject: [PATCH] feat(server): honour HeaderReceived's ReadTimeout per HTTP/2 stream fasthttp.Server.HeaderReceived lets a server give one request its own read timeout over HTTP/1.1. Over HTTP/2 every stream got the server's ReadTimeout, so a long upload on HTTP/2 could not be given more time without raising it for every request. The server now calls HeaderReceived when a stream's headers end; a non-zero ReadTimeout replaces the server's for that stream, measured from the stream's start. The request timer arms for the earliest deadline of the streams still receiving, and re-arms on every new stream, since one long deadline may otherwise hold the timer past a shorter one. Only ReadTimeout is honoured; MaxRequestBodySize and WriteTimeout are not applied per stream. Co-Authored-By: Claude Opus 5.5 --- server.go | 1 + serverConn.go | 104 ++++++++++++++++++++++++------------- server_readtimeout_test.go | 83 +++++++++++++++++++++++++++++ stream.go | 5 ++ 4 files changed, 158 insertions(+), 35 deletions(-) create mode 100644 server_readtimeout_test.go diff --git a/server.go b/server.go index 926c5d5..58784b3 100644 --- a/server.go +++ b/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, diff --git a/serverConn.go b/serverConn.go index ae036e1..39b325a 100644 --- a/serverConn.go +++ b/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,8 +546,6 @@ 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, @@ -509,11 +553,12 @@ loop: // 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: diff --git a/server_readtimeout_test.go b/server_readtimeout_test.go new file mode 100644 index 0000000..89d0b9f --- /dev/null +++ b/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") + } +} diff --git a/stream.go b/stream.go index 9a7d37b..7a70a92 100644 --- a/stream.go +++ b/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")