fix(observability-proxy): admit a request before reading its body

The form was parsed before the concurrency slot was taken, and the
listener sets no body-read deadline, so a client that never finished a
POST body held a handler outside every limit for as long as it liked.
Tenant workloads reach the Service directly, so nginx buffering does
not cover this path.

The slot is now taken first and the body is read under the request
timeout; a body that does not arrive in time answers 408 and frees the
slot. The deadline is lifted once the body is read, because the
server's background read would take a later expiry for a client that
went away and cancel a live tail.

Assisted-by: LLM
Signed-off-by: ZverGuy <maximbel2003@gmail.com>

internal/observabilityproxy/server.go

@@ -25,6 +25,7 @@ import (
 	"log/slog"
 	"math"
 	"net/http"
+	"os"
 	"strconv"
 	"strings"
 	"time"
@@ -159,15 +160,6 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT
 	if !ok {
 		return errorf(http.StatusNotFound, "not found")
 	}
-	r.Body = http.MaxBytesReader(w, r.Body, maxFormBytes)
-	if err := r.ParseForm(); err != nil {
-		return errorf(http.StatusBadRequest, "invalid form: %v", err)
-	}
-	params, herr := BuildLogsParams(endpoint, r.Form, route.Namespace, s.now(), s.cfg.Logs)
-	if herr != nil {
-		return herr
-	}
-
 	tail := endpoint == "tail"
 	acquire := s.limits.acquire
 	if tail {
@@ -178,6 +170,13 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT
 		return errorf(http.StatusTooManyRequests, "too many concurrent requests")
 	}
 	defer release()
+	if herr := s.readForm(w, r); herr != nil {
+		return herr
+	}
+	params, herr := BuildLogsParams(endpoint, r.Form, route.Namespace, s.now(), s.cfg.Logs)
+	if herr != nil {
+		return herr
+	}
 
 	store, err := s.stores.Resolve(route.Namespace)
 	if err != nil {
@@ -233,6 +232,29 @@ 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
+// 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.
+func (s *Server) readForm(w http.ResponseWriter, r *http.Request) *HTTPError {
+	r.Body = http.MaxBytesReader(w, r.Body, maxFormBytes)
+	rc := http.NewResponseController(w)
+	if s.cfg.RequestTimeout > 0 {
+		_ = rc.SetReadDeadline(time.Now().Add(s.cfg.RequestTimeout))
+	}
+	err := r.ParseForm()
+	switch {
+	case err == nil:
+		_ = rc.SetReadDeadline(time.Time{})
+		return nil
+	case errors.Is(err, os.ErrDeadlineExceeded):
+		return errorf(http.StatusRequestTimeout, "request body not received within %s", s.cfg.RequestTimeout)
+	default:
+		return errorf(http.StatusBadRequest, "invalid form: %v", err)
+	}
+}
+
 func (s *Server) authorize(r *http.Request, route Route) (string, *HTTPError) {
 	token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
 	if !ok {

internal/observabilityproxy/server_deadline_test.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2026 The Cozystack 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 observabilityproxy
+
+import (
+	"bufio"
+	"net"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+)
+
+// pipeListener hands the server one end of a net.Pipe per dial. A pipe has
+// no buffer, so a client that stops reading blocks the server's next write
+// at once, as a full TCP window eventually does.
+type pipeListener struct {
+	conns  chan net.Conn
+	closed chan struct{}
+	once   sync.Once
+}
+
+type pipeAddr struct{}
+
+func (pipeAddr) Network() string { return "pipe" }
+func (pipeAddr) String() string  { return "pipe" }
+
+func servePipe(t *testing.T, handler http.Handler) *pipeListener {
+	t.Helper()
+	l := &pipeListener{conns: make(chan net.Conn), closed: make(chan struct{})}
+	srv := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second}
+	go func() { _ = srv.Serve(l) }()
+	t.Cleanup(func() { _ = srv.Close() })
+	return l
+}
+
+func (l *pipeListener) Accept() (net.Conn, error) {
+	select {
+	case c := <-l.conns:
+		return c, nil
+	case <-l.closed:
+		return nil, net.ErrClosed
+	}
+}
+
+func (l *pipeListener) Close() error {
+	l.once.Do(func() { close(l.closed) })
+	return nil
+}
+
+func (l *pipeListener) Addr() net.Addr { return pipeAddr{} }
+
+// send opens a connection and writes raw to it without waiting for the
+// server to read all of it.
+func (l *pipeListener) send(t *testing.T, raw string) net.Conn {
+	t.Helper()
+	client, server := net.Pipe()
+	t.Cleanup(func() { _ = client.Close() })
+	select {
+	case l.conns <- server:
+	case <-time.After(2 * time.Second):
+		t.Fatal("pipe server is not accepting")
+	}
+	go func() { _, _ = client.Write([]byte(raw)) }()
+	return client
+}
+
+func eventually(t *testing.T, within time.Duration, cond func() bool, msg string) {
+	t.Helper()
+	deadline := time.Now().Add(within)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatal(msg)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func readResponse(t *testing.T, conn net.Conn) *http.Response {
+	t.Helper()
+	got := make(chan *http.Response, 1)
+	go func() {
+		resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
+		if err != nil {
+			close(got)
+			return
+		}
+		got <- resp
+	}()
+	select {
+	case resp, ok := <-got:
+		if !ok {
+			t.Fatal("reading the response failed")
+		}
+		return resp
+	case <-time.After(3 * time.Second):
+		t.Fatal("no response")
+	}
+	return nil
+}
+
+// TestServerSlowBodyHoldsSlotUntilDeadline sends a POST whose body never
+// completes: it must count against the user's concurrency, and the read
+// must give up at the request timeout instead of holding the handler.
+func TestServerSlowBodyHoldsSlotUntilDeadline(t *testing.T) {
+	h := newHarness(t, Config{
+		RequestTimeout: 200 * time.Millisecond,
+		Limiter:        LimiterConfig{MaxConcurrentPerUser: 1},
+	}, nil)
+	pipe := servePipe(t, h.server)
+	conn := pipe.send(t, "POST "+logsRoot+"query HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\n"+
+		"Content-Type: application/x-www-form-urlencoded\r\nContent-Length: 100\r\n\r\nquery=")
+
+	eventually(t, time.Second, func() bool {
+		return h.status("GET", logsRoot+"query", "good") == http.StatusTooManyRequests
+	}, "a request with an unfinished body does not hold a concurrency slot")
+	if resp := readResponse(t, conn); resp.StatusCode != http.StatusRequestTimeout {
+		t.Fatalf("unfinished body answered %d, want 408", resp.StatusCode)
+	}
+	eventually(t, 2*time.Second, func() bool {
+		return h.status("GET", logsRoot+"query", "good") == http.StatusOK
+	}, "the slot of an unfinished body was not released")
+}