From 7dc6ac3687fcb9fe5fba35686878158d18bd3232 Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Thu, 5 Nov 2026 01:20:00 +0700 Subject: [PATCH] fix(observability-proxy): read a leftover request body under the deadline ParseForm reads no body on a GET or on a POST that is not a form. When such a request declared a body and never sent it, net/http drained it after the handler flushed the headers, with no read deadline, so the request held its slot for as long as the client kept the connection. The handler now reads the rest of the body itself, under the same deadline and size cap as the form. Draining was chosen over refusing a declared body so that clients which send one stay working. A body over the cap now answers 413 instead of a generic 400. Assisted-by: LLM Signed-off-by: ZverGuy --- internal/observabilityproxy/server.go | 14 ++++- .../server_deadline_test.go | 53 +++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/internal/observabilityproxy/server.go b/internal/observabilityproxy/server.go index 2e391442c..0f9e533b7 100644 --- a/internal/observabilityproxy/server.go +++ b/internal/observabilityproxy/server.go @@ -266,8 +266,12 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT return nil } -// readForm reads the body under the request timeout, so a client that -// sends it slowly holds its slot only that long. Once the body is read the +// readForm reads the whole body under the request timeout, so a client +// that sends it slowly holds its slot only that long. ParseForm leaves the +// body of a GET or of a POST that is not a form unread, and net/http +// drains a leftover body with no deadline once the headers go out, so the +// rest is read here as well; draining rather than refusing a declared +// body keeps any client that sends one working. Once the body is read the // deadline is lifted: the server's background read would take a later // expiry for a client that went away and cancel a tail. After a failed // read it stays, so the server's drain of the unread body is bounded too. @@ -278,7 +282,13 @@ func (s *Server) readForm(w http.ResponseWriter, r *http.Request) *HTTPError { _ = rc.SetReadDeadline(time.Now().Add(s.cfg.RequestTimeout)) } err := r.ParseForm() + if err == nil { + _, err = io.Copy(io.Discard, r.Body) + } + var tooLarge *http.MaxBytesError switch { + case errors.As(err, &tooLarge): + return errorf(http.StatusRequestEntityTooLarge, "request body larger than %d bytes", tooLarge.Limit) case err == nil: _ = rc.SetReadDeadline(time.Time{}) return nil diff --git a/internal/observabilityproxy/server_deadline_test.go b/internal/observabilityproxy/server_deadline_test.go index afd69ed30..ac62dd782 100644 --- a/internal/observabilityproxy/server_deadline_test.go +++ b/internal/observabilityproxy/server_deadline_test.go @@ -21,6 +21,7 @@ import ( "io" "net" "net/http" + "strings" "sync" "testing" "time" @@ -244,3 +245,55 @@ func TestServerAbortsTruncatedResponse(t *testing.T) { t.Fatal("a truncated store answer reached the client as a complete response") } } + +// TestServerUndeliveredBodyReleasesSlot declares a body that never comes +// on requests whose form parsing does not read it: a GET, a POST that is +// not a form, and a GET tail. net/http would drain that body with no +// deadline once the headers go out, so the slot must still come back. +func TestServerUndeliveredBodyReleasesSlot(t *testing.T) { + for name, raw := range map[string]string{ + "get query": "GET " + logsRoot + "query?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Length: 10\r\n\r\n", + "post text query": "POST " + logsRoot + "query?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Type: text/plain\r\nContent-Length: 10\r\n\r\n", + "get tail": "GET " + logsRoot + "tail?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Length: 10\r\n\r\n", + } { + t.Run(name, func(t *testing.T) { + lines := make(chan string, 1) + done := make(chan struct{}) + h := newHarness(t, Config{ + RequestTimeout: 200 * time.Millisecond, + TailIdleTimeout: 200 * time.Millisecond, + Limiter: LimiterConfig{MaxConcurrentPerUser: 1, MaxTails: 1}, + }, tailUpstream(lines, done)) + t.Cleanup(func() { close(done) }) + pipe := servePipe(t, h.server) + conn := pipe.send(t, raw) + held := func() bool { + h.server.limits.mu.Lock() + defer h.server.limits.mu.Unlock() + return h.server.limits.active["alice"]+h.server.limits.total > 0 + } + eventually(t, time.Second, held, "the request never took a slot") + if resp := readResponse(t, conn); resp.StatusCode != http.StatusRequestTimeout { + t.Fatalf("undelivered body answered %d, want 408", resp.StatusCode) + } + eventually(t, time.Second, func() bool { return !held() }, "an undelivered body keeps its slot") + }) + } +} + +func TestServerOversizedBody(t *testing.T) { + h := newHarness(t, Config{}, nil) + for _, ct := range []string{"application/x-www-form-urlencoded", "text/plain"} { + req, _ := http.NewRequest(http.MethodPost, h.proxy.URL+logsRoot+"query?query=*", strings.NewReader("query="+strings.Repeat("x", maxFormBytes))) + req.Header.Set("Authorization", "Bearer good") + req.Header.Set("Content-Type", ct) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusRequestEntityTooLarge { + t.Errorf("%s body over the cap answered %d, want 413", ct, resp.StatusCode) + } + } +}