fix: give net/http, Prometheus, OTel and slog copies of the request strings they keep, and give Adapt's request its headers (#732, #720)

.github/workflows/ci.yml

@@ -445,6 +445,45 @@ jobs:
       - name: middleware/otel
         run: go test -race -count=1 -timeout=120s ./...
         working-directory: middleware/otel
+      # celeris#720, #732: Adapt must hand net/http the request headers, and
+      # Adapt, adapters.WrapMiddleware and the metrics, otel and logger
+      # middleware must hand net/http, Prometheus, the OTel SDK and slog
+      # copies of the request strings those keep, not views of the
+      # connection's receive buffer. The steps above run these packages, but
+      # without -v, and each of these tests drops its io_uring arms without a
+      # word when the probe finds no ring, which happens while another test
+      # binary's rings are still being freed. So they run here by name, one
+      # package at a time, with CELERIS_REQUIRE_IOURING_WORKERS=1 turning a
+      # missing io_uring arm into a failure, and an exact tally: the seven
+      # top-level tests and all 35 arms (std, and epoll and io_uring with sync
+      # and async handlers; 14 of them io_uring) must PASS, with no SKIP line,
+      # and every go test must exit 0. A failing test does not stop the step
+      # early, and the step runs even when a step above failed.
+      - name: celeris#720, #732 request strings handed to net/http, Prometheus, OTel, slog (io_uring arms required)
+        if: ${{ !cancelled() }}
+        shell: bash
+        env:
+          CELERIS_REQUIRE_IOURING_WORKERS: "1"
+        run: |
+          set -o pipefail
+          echo "memlock (KiB): $(ulimit -l)"
+          : > /tmp/c732.log
+          failed=0
+          go test -race -count=1 -timeout=300s -v -run '^(TestAdaptRequestCarriesHeaders|TestAdaptKeptRequestStringsSurviveNextRequest)$' . 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1))
+          go test -race -count=1 -timeout=300s -v -run '^TestWrapMiddlewareKeptStringsSurviveNextRequest$' ./middleware/adapters/ 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1))
+          go test -race -count=1 -timeout=300s -v -run '^TestKeptRecordSurvivesNextRequest$' ./middleware/logger/ 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1))
+          (cd middleware/metrics && go test -race -count=1 -timeout=300s -v -run '^(TestLabelValuesSurviveNextRequest|TestLabelValuesSurviveConnectionReuse)$' .) 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1))
+          (cd middleware/otel && go test -race -count=1 -timeout=300s -v -run '^TestAttributesSurviveNextRequest$' .) 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1))
+          tests='TestAdaptRequestCarriesHeaders|TestAdaptKeptRequestStringsSurviveNextRequest|TestWrapMiddlewareKeptStringsSurviveNextRequest|TestKeptRecordSurvivesNextRequest|TestLabelValuesSurviveNextRequest|TestLabelValuesSurviveConnectionReuse|TestAttributesSurviveNextRequest'
+          top=$(grep -cE "^--- PASS: (${tests}) \(" /tmp/c732.log || true)
+          arms=$(grep -cE "^    --- PASS: (${tests})/" /tmp/c732.log || true)
+          uring=$(grep -cE "^    --- PASS: (${tests})/io_uring" /tmp/c732.log || true)
+          skipped=$(grep -cE '^[[:space:]]*--- SKIP' /tmp/c732.log || true)
+          echo "celeris#720/#732 tests: top-level PASS $top (want 7), arm PASS $arms (want 35), io_uring arm PASS $uring (want 14), SKIP lines $skipped, go test runs failed $failed (want 0)"
+          if [ "$top" -ne 7 ] || [ "$arms" -ne 35 ] || [ "$uring" -ne 14 ] || [ "$skipped" -ne 0 ] || [ "$failed" -ne 0 ]; then
+            echo "expected all seven tests and all 35 arms (14 of them io_uring) to PASS with no SKIP line"
+            exit 1
+          fi
       - name: middleware/protobuf
         run: go test -race -count=1 -timeout=120s ./...
         working-directory: middleware/protobuf

adapt_request_linux_test.go

@@ -0,0 +1,299 @@
+//go:build linux
+
+package celeris_test
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/goceleris/celeris"
+	celerisengine "github.com/goceleris/celeris/engine"
+	"github.com/goceleris/celeris/probe"
+)
+
+// TestAdaptRequestCarriesHeaders pins celeris#720.
+//
+// On epoll and io_uring the H1 parser builds the header slice lazily: it is
+// filled the first time something reads a header. Adapt copied the headers
+// from the slice without filling it, so on a route with no header read
+// before it, the *http.Request had no headers at all. Nothing runs before
+// the adapted handler here.
+func TestAdaptRequestCarriesHeaders(t *testing.T) {
+	for _, a := range keptArms(t) {
+		t.Run(a.name, func(t *testing.T) {
+			got := make(chan http.Header, 1)
+			addr, stop := startKeptServer(t, func() *celeris.Server {
+				srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async})
+				srv.POST("/h", celeris.Adapt(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					got <- r.Header.Clone()
+					_, _ = w.Write([]byte("ok"))
+				})))
+				return srv
+			})
+			defer stop()
+
+			conn, br := keptDial(t, addr)
+			defer func() { _ = conn.Close() }()
+			keptRoundTrip(t, conn, br, "POST /h HTTP/1.1\r\nHost: h.example\r\nX-Id: id-1\r\nAuthorization: Bearer t0k3n\r\nCookie: sid=s1\r\nContent-Type: application/json\r\nContent-Length: 2\r\n\r\n{}")
+			h := <-got
+			want := map[string]string{
+				"X-Id":          "id-1",
+				"Authorization": "Bearer t0k3n",
+				"Cookie":        "sid=s1",
+				"Content-Type":  "application/json",
+			}
+			var wrong []string
+			for k, v := range want {
+				if h.Get(k) != v {
+					wrong = append(wrong, fmt.Sprintf("%s=%q (want %q)", k, h.Get(k), v))
+				}
+			}
+			sort.Strings(wrong)
+			t.Logf("KEPT720 arm=%s header keys=%d wrong=%d", a.name, len(h), len(wrong))
+			if len(wrong) > 0 {
+				t.Errorf("the adapted handler's request is missing headers: %s", strings.Join(wrong, ", "))
+			}
+		})
+	}
+}
+
+// TestAdaptKeptRequestStringsSurviveNextRequest pins the Adapt site of
+// celeris#732.
+//
+// net/http lets a handler keep the request's strings after ServeHTTP
+// returns. On epoll and io_uring the method, path, query, Host and header
+// strings Adapt handed over were views of the connection's receive buffer,
+// which the engine reuses for the connection's next request. Three requests
+// with the same layout and different values go over one keep-alive
+// connection; the handler keeps its request's strings, and after the third
+// request each kept set must still read its own request. The methods are
+// ones the H1 parser does not intern, the header name "111-111" is one
+// net/http's canonicalization returns unchanged (it has no letters), and no
+// request has a query (with one, the URL was a new string and the path a
+// copy by accident), so none of them is copied on the way in.
+func TestAdaptKeptRequestStringsSurviveNextRequest(t *testing.T) {
+	methods := []string{"TRACE", "PURGE", "MKCOL"}
+	for _, a := range keptArms(t) {
+		t.Run(a.name, func(t *testing.T) {
+			kept := make(chan map[string]string, len(methods))
+			addr, stop := startKeptServer(t, func() *celeris.Server {
+				srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async})
+				h := celeris.Adapt(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					// Legal in net/http: keep the request's strings.
+					got := map[string]string{
+						"method": r.Method,
+						"path":   r.URL.Path,
+						"query":  r.URL.RawQuery,
+						"host":   r.Host,
+						"x-id":   r.Header.Get("X-Id"),
+					}
+					for k := range r.Header {
+						if strings.HasSuffix(k, "-111") || strings.HasSuffix(k, "-222") || strings.HasSuffix(k, "-333") {
+							got["digit-key"] = k
+						}
+					}
+					kept <- got
+					_, _ = w.Write([]byte("ok"))
+				}))
+				for _, m := range methods {
+					srv.Handle(m, "/a/:id", h)
+				}
+				return srv
+			})
+			defer stop()
+
+			conn, br := keptDial(t, addr)
+			defer func() { _ = conn.Close() }()
+			var want []map[string]string
+			for i, m := range methods {
+				v := strings.Repeat(string(rune('a'+i)), 4)
+				d := strings.Repeat(strconv.Itoa(i+1), 3)
+				keptRoundTrip(t, conn, br, m+" /a/"+v+" HTTP/1.1\r\nHost: h-"+v+".example\r\nX-Id: id-"+v+"\r\n"+d+"-"+d+": k\r\n\r\n")
+				want = append(want, map[string]string{
+					"method": m, "path": "/a/" + v, "query": "", "host": "h-" + v + ".example",
+					"x-id": "id-" + v, "digit-key": d + "-" + d,
+				})
+			}
+			keptCompare(t, a.name, "KEPT732ADAPT", want, kept)
+		})
+	}
+}
+
+// keptCompare reads one kept map per request from kept and reports every
+// field that no longer reads its own request.
+func keptCompare(t *testing.T, arm, tag string, want []map[string]string, kept <-chan map[string]string) {
+	t.Helper()
+	var wrong []string
+	for i, w := range want {
+		var got map[string]string
+		select {
+		case got = <-kept:
+		case <-time.After(10 * time.Second):
+			t.Fatalf("request %d: the handler did not report", i+1)
+		}
+		keys := make([]string, 0, len(w))
+		for k := range w {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			if got[k] != w[k] {
+				wrong = append(wrong, fmt.Sprintf("request %d %s: %q (want %q)", i+1, k, got[k], w[k]))
+			}
+		}
+	}
+	t.Logf("%s arm=%s requests=%d wrong=%d", tag, arm, len(want), len(wrong))
+	if len(wrong) > 0 {
+		t.Errorf("strings kept from a request read other bytes after the connection's later requests:\n  %s", strings.Join(wrong, "\n  "))
+	}
+}
+
+type keptArm struct {
+	name   string
+	engine celeris.EngineType
+	async  bool
+}
+
+// keptArms returns std, and epoll and io_uring with sync and async
+// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable
+// io_uring fails the test instead of dropping the io_uring arms.
+func keptArms(t *testing.T) []keptArm {
+	t.Helper()
+	arms := []keptArm{
+		{"std", celeris.Std, false},
+		{"epoll", celeris.Epoll, false},
+		{"epoll-async", celeris.Epoll, true},
+	}
+	if ok, p := keptProbeIOUring(); ok {
+		arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true})
+	} else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" {
+		t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion)
+	} else {
+		t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion)
+	}
+	return arms
+}
+
+// keptProbeIOUring probes the kernel's io_uring support. With
+// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is
+// retried for up to 10 s: the probe's ring can fail with ENOMEM against
+// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of
+// another test binary of the same user, are still charged
+// (engine/iouring/ring_budget_linux_test.go).
+func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) {
+	p = probe.Probe()
+	usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers
+	if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" {
+		return usable, p
+	}
+	for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); {
+		time.Sleep(10 * time.Millisecond)
+		p = probe.Probe()
+		usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers
+	}
+	return usable, p
+}
+
+// startKeptServer starts the server mk builds on a fresh loopback listener
+// and returns its address and a shutdown closure. A start that fails only
+// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see
+// keptProbeIOUring) is retried with a new server for up to 10 s.
+func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		s := mk()
+		ln, err := net.Listen("tcp", "127.0.0.1:0")
+		if err != nil {
+			t.Fatal(err)
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan error, 1)
+		go func() { done <- s.StartWithListenerAndContext(ctx, ln) }()
+		addr, err := keptWaitReady(s, done)
+		if err == nil {
+			return addr, func() { cancel(); <-done }
+		}
+		cancel()
+		_ = ln.Close()
+		if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) {
+			time.Sleep(2 * time.Millisecond)
+			continue
+		}
+		t.Fatalf("server did not start: %v", err)
+	}
+}
+
+func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) {
+	deadline := time.Now().Add(30 * time.Second)
+	for time.Now().Before(deadline) {
+		select {
+		case err := <-done:
+			if err == nil {
+				err = errors.New("start returned before the server was ready")
+			}
+			return "", err
+		default:
+		}
+		if a := s.Addr(); a != nil {
+			if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil {
+				_ = c.Close()
+				return a.String(), nil
+			}
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return "", errors.New("server not ready within 30s")
+}
+
+func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) {
+	t.Helper()
+	conn, err := net.DialTimeout("tcp", addr, 2*time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = conn.SetDeadline(time.Now().Add(20 * time.Second))
+	return conn, bufio.NewReader(conn)
+}
+
+// keptRoundTrip writes one request and reads its response; it fails the
+// test unless the status is 2xx.
+func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) {
+	t.Helper()
+	if _, err := conn.Write([]byte(req)); err != nil {
+		t.Fatal(err)
+	}
+	status, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read status: %v", err)
+	}
+	n := 0
+	for {
+		line, err := br.ReadString('\n')
+		if err != nil {
+			t.Fatalf("read header: %v", err)
+		}
+		if line == "\r\n" {
+			break
+		}
+		if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") {
+			n, _ = strconv.Atoi(strings.TrimSpace(v))
+		}
+	}
+	if _, err := br.Discard(n); err != nil {
+		t.Fatalf("read body: %v", err)
+	}
+	if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' {
+		t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0])
+	}
+}