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 <maximbel2003@gmail.com>

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

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)
+		}
+	}
+}