From 4bd508b4f7234aa91278e8582e7af558003f2d79 Mon Sep 17 00:00:00 2001 From: Kevin Dunglas Date: Mon, 20 Jul 2026 16:34:57 +0200 Subject: [PATCH] fix: spool queued request bodies to break HTTP/2 upload deadlock Concurrent uploads multiplexed on one HTTP/2 connection hang when there are more streams than PHP threads. A stream queued waiting for a thread keeps its flow-control window open while no one reads the body; enough queued streams exhaust the connection-level window and stall every stream on the connection, including those a thread is already serving. Drain a request body into a buffer (spilling past 2 MiB to a temp file) before it enters the queue, releasing the window so every stream is read by someone. Requests that get a thread immediately still stream live. Bodies overrunning a request_body max_size limit are rejected with 413 instead of reaching PHP truncated. Closes #1074 --- context.go | 10 ++ frankenphp.go | 3 +- requestbodyspool.go | 150 ++++++++++++++++++++++++++++++ requestbodyspool_internal_test.go | 102 ++++++++++++++++++++ requestbodyspool_test.go | 97 +++++++++++++++++++ threadregular.go | 10 +- worker.go | 10 +- 7 files changed, 379 insertions(+), 3 deletions(-) create mode 100644 requestbodyspool.go create mode 100644 requestbodyspool_internal_test.go create mode 100644 requestbodyspool_test.go diff --git a/context.go b/context.go index c582c6e453..e487dbbb4c 100644 --- a/context.go +++ b/context.go @@ -36,6 +36,11 @@ type frankenPHPContext struct { // Whether the request is already closed by us isDone bool + // Whether the request body has been drained into a buffer while queued + bodySpooled bool + // Releases the spooled body's backing temp file, if any + cleanupBody func() + responseWriter http.ResponseWriter responseController *http.ResponseController handlerParameters any @@ -132,6 +137,11 @@ func (fc *frankenPHPContext) closeContext() { close(fc.done) fc.isDone = true + + if fc.cleanupBody != nil { + fc.cleanupBody() + fc.cleanupBody = nil + } } // validate checks if the request should be outright rejected diff --git a/frankenphp.go b/frankenphp.go index ad2dedc42a..e6eaaa3b7f 100644 --- a/frankenphp.go +++ b/frankenphp.go @@ -53,6 +53,7 @@ var ( ErrInvalidRequestPath = ErrRejected{"invalid request path", http.StatusBadRequest} ErrInvalidContentLengthHeader = ErrRejected{"invalid Content-Length header", http.StatusBadRequest} ErrMaxWaitTimeExceeded = ErrRejected{"maximum request handling time exceeded", http.StatusServiceUnavailable} + ErrRequestBodyTooLarge = ErrRejected{"request body too large", http.StatusRequestEntityTooLarge} contextKey = contextKeyStruct{} serverHeader = []string{"FrankenPHP"} @@ -649,7 +650,7 @@ func go_read_post(threadIndex C.uintptr_t, cBuf *C.char, countBytes C.size_t) (r // deadline on a finalized HTTP/2 stream, dereferencing a nil pointer and // crashing the process. See https://github.com/php/frankenphp/issues/2535. var rc *http.ResponseController - if fc.requestBodyTimeout > 0 && !fc.isDone { + if fc.requestBodyTimeout > 0 && !fc.isDone && !fc.bodySpooled { if fc.responseController == nil { fc.responseController = http.NewResponseController(fc.responseWriter) } diff --git a/requestbodyspool.go b/requestbodyspool.go new file mode 100644 index 0000000000..a15ba76f4e --- /dev/null +++ b/requestbodyspool.go @@ -0,0 +1,150 @@ +package frankenphp + +import ( + "bytes" + "errors" + "io" + "log/slog" + "net/http" + "os" + "time" +) + +// bodySpoolMemoryThreshold is how much of a queued request body is kept in +// memory before spilling to a temp file. +const bodySpoolMemoryThreshold = 2 << 20 // 2 MiB + +// spoolRequestBody drains the request body and replaces it with a buffered copy. +// +// A request stalled in the thread queue keeps its HTTP/2 stream flow-control +// window open while no one reads the body. Enough stalled uploads multiplexed on +// a single connection exhaust the connection-level window and deadlock every +// stream on that connection, including those a thread is already serving. +// Draining a queued body up front releases the window and breaks the deadlock. +// See https://github.com/php/frankenphp/issues/1074. +// +// Only bodies with a known, non-zero Content-Length are spooled. Streaming +// requests (chunked or unknown length, e.g. long-lived uploads) keep their live +// stream and their current behavior. +// It returns ErrRequestBodyTooLarge (and rejects the request with 413) when a +// request_body max_size limit wraps the body and the client exceeds it; callers +// must stop handling the request in that case. +func (fc *frankenPHPContext) spoolRequestBody() error { + r := fc.request + if fc.bodySpooled || r == nil || r.Body == nil || r.Body == http.NoBody || r.ContentLength <= 0 { + return nil + } + + src := io.Reader(r.Body) + + // keep bounding slow uploads while draining, mirroring go_read_post + if fc.requestBodyTimeout > 0 && !fc.isDone && fc.responseWriter != nil { + if fc.responseController == nil { + fc.responseController = http.NewResponseController(fc.responseWriter) + } + src = &deadlineReader{r: r.Body, rc: fc.responseController, timeout: fc.requestBodyTimeout} + } + + sw := &spoolWriter{threshold: bodySpoolMemoryThreshold} + n, err := io.Copy(sw, src) + + if fc.requestBodyTimeout > 0 && fc.responseController != nil { + _ = fc.responseController.SetReadDeadline(time.Time{}) + } + _ = r.Body.Close() + + // A body larger than the configured max_size cannot be handled at all; + // reject it up front instead of feeding PHP a truncated request. + var maxBytesErr *http.MaxBytesError + if errors.As(err, &maxBytesErr) { + sw.cleanup() + fc.reject(ErrRequestBodyTooLarge) + + return ErrRequestBodyTooLarge + } + + if err != nil { + // The stream is already (partially) drained and cannot be replayed. + // Hand PHP whatever was read; a short read surfaces as EOF, matching + // what a live read would have produced after the same failure. + if fc.logger.Enabled(fc.request.Context(), slog.LevelWarn) { + fc.logger.LogAttrs(fc.request.Context(), slog.LevelWarn, "error while spooling request body", slog.Any("error", err)) + } + } + + fc.bodySpooled = true + r.ContentLength = n + + if sw.file == nil { + r.Body = io.NopCloser(bytes.NewReader(sw.buf.Bytes())) + + return nil + } + + if _, err := sw.file.Seek(0, io.SeekStart); err != nil { + sw.cleanup() + r.Body = io.NopCloser(bytes.NewReader(nil)) + r.ContentLength = 0 + + return nil + } + + r.Body = sw.file + fc.cleanupBody = sw.cleanup + + return nil +} + +// spoolWriter buffers in memory up to threshold, then spills the rest to a temp +// file so a large queued body never grows the heap unbounded. +type spoolWriter struct { + buf bytes.Buffer + file *os.File + threshold int +} + +func (s *spoolWriter) Write(p []byte) (int, error) { + if s.file == nil { + if s.buf.Len()+len(p) <= s.threshold { + return s.buf.Write(p) + } + + f, err := os.CreateTemp("", "frankenphp-upload-*") + if err != nil { + return 0, err + } + + s.file = f + if _, err := s.file.Write(s.buf.Bytes()); err != nil { + return 0, err + } + s.buf.Reset() + } + + return s.file.Write(p) +} + +// cleanup releases the backing temp file, if any. Safe to call when nothing spilled. +func (s *spoolWriter) cleanup() { + if s.file == nil { + return + } + + name := s.file.Name() + _ = s.file.Close() + _ = os.Remove(name) +} + +// deadlineReader resets the read deadline before every read to bound a stall +// without capping a steady upload. +type deadlineReader struct { + r io.Reader + rc *http.ResponseController + timeout time.Duration +} + +func (d *deadlineReader) Read(p []byte) (int, error) { + _ = d.rc.SetReadDeadline(time.Now().Add(d.timeout)) + + return d.r.Read(p) +} diff --git a/requestbodyspool_internal_test.go b/requestbodyspool_internal_test.go new file mode 100644 index 0000000000..a8bc0cd818 --- /dev/null +++ b/requestbodyspool_internal_test.go @@ -0,0 +1,102 @@ +package frankenphp + +import ( + "bytes" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newSpoolContext(body []byte) *frankenPHPContext { + fc := newFrankenPHPContext() + fc.logger = slog.Default() + fc.request = httptest.NewRequest("POST", "/", bytes.NewReader(body)) + + return fc +} + +// TestSpoolRequestBodyInMemory drains a small body into memory, leaving no temp +// file to clean up. +func TestSpoolRequestBodyInMemory(t *testing.T) { + body := bytes.Repeat([]byte("a"), 1024) + fc := newSpoolContext(body) + + fc.spoolRequestBody() + + assert.True(t, fc.bodySpooled) + assert.Nil(t, fc.cleanupBody, "a small body stays in memory") + assert.Equal(t, int64(len(body)), fc.request.ContentLength) + + got, err := io.ReadAll(fc.request.Body) + require.NoError(t, err) + assert.Equal(t, body, got) +} + +// TestSpoolRequestBodyToFile spills a body larger than the memory threshold to a +// temp file, serves identical bytes, and removes the file on cleanup. +func TestSpoolRequestBodyToFile(t *testing.T) { + body := bytes.Repeat([]byte("b"), bodySpoolMemoryThreshold+4096) + fc := newSpoolContext(body) + + fc.spoolRequestBody() + + assert.True(t, fc.bodySpooled) + require.NotNil(t, fc.cleanupBody, "a large body spills to a temp file") + assert.Equal(t, int64(len(body)), fc.request.ContentLength) + + got, err := io.ReadAll(fc.request.Body) + require.NoError(t, err) + assert.Equal(t, body, got) + + fc.cleanupBody() +} + +// TestSpoolRequestBodySkipsStreaming leaves a body of unknown length untouched so +// long-lived streaming uploads keep their live stream. +func TestSpoolRequestBodySkipsStreaming(t *testing.T) { + fc := newSpoolContext([]byte("streamed")) + fc.request.ContentLength = -1 + + fc.spoolRequestBody() + + assert.False(t, fc.bodySpooled) + got, err := io.ReadAll(fc.request.Body) + require.NoError(t, err) + assert.Equal(t, "streamed", string(got)) +} + +// TestSpoolRequestBodyIdempotent does not re-drain an already spooled body: a +// second call must not touch the already consumed stream. +func TestSpoolRequestBodyIdempotent(t *testing.T) { + body := bytes.Repeat([]byte("c"), 512) + fc := newSpoolContext(body) + + require.NoError(t, fc.spoolRequestBody()) + require.NoError(t, fc.spoolRequestBody()) + + assert.True(t, fc.bodySpooled) + got, err := io.ReadAll(fc.request.Body) + require.NoError(t, err) + assert.Equal(t, body, got) +} + +// TestSpoolRequestBodyRejectsOversized rejects a body that overruns a +// request_body max_size limit (an http.MaxBytesReader) with 413 instead of +// feeding PHP a truncated request. +func TestSpoolRequestBodyRejectsOversized(t *testing.T) { + fc := newSpoolContext(bytes.Repeat([]byte("d"), 4096)) + rec := httptest.NewRecorder() + fc.responseWriter = rec + fc.request.Body = http.MaxBytesReader(rec, fc.request.Body, 1024) + + err := fc.spoolRequestBody() + + require.ErrorIs(t, err, ErrRequestBodyTooLarge) + assert.False(t, fc.bodySpooled) + assert.Equal(t, http.StatusRequestEntityTooLarge, rec.Code) +} diff --git a/requestbodyspool_test.go b/requestbodyspool_test.go new file mode 100644 index 0000000000..d09118ed3b --- /dev/null +++ b/requestbodyspool_test.go @@ -0,0 +1,97 @@ +package frankenphp_test + +import ( + "bytes" + "fmt" + "io" + "net/http" + "os" + "sync" + "testing" + "time" + + "github.com/dunglas/frankenphp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestConcurrentUploadsHTTP2 reproduces php/frankenphp#1074: many request bodies +// multiplexed on a single HTTP/2 connection, with more streams than PHP threads. +// A queued stream that no thread reads keeps its flow-control window open; enough +// of them exhaust the connection window and deadlock every stream, including the +// ones a thread is serving. Draining queued bodies up front must let all of them +// complete. +func TestConcurrentUploadsHTTP2(t *testing.T) { + require.NoError(t, frankenphp.Init( + frankenphp.WithNumThreads(2), + frankenphp.WithMaxThreads(2), + )) + defer frankenphp.Shutdown() + + cwd, _ := os.Getwd() + handler := func(w http.ResponseWriter, r *http.Request) { + req, err := frankenphp.NewRequestWithContext(r, + frankenphp.WithRequestDocumentRoot(cwd+"/testdata/", false), + ) + require.NoError(t, err) + require.NoError(t, frankenphp.ServeHTTP(w, req)) + } + + addr, client := newH2CServer(t, handler) + + const ( + concurrency = 30 + bodySize = 512 << 10 // large enough to exhaust the connection window + ) + body := bytes.Repeat([]byte("x"), bodySize) + want := fmt.Sprintf("read=%d", bodySize) + + var wg sync.WaitGroup + errs := make(chan error, concurrency) + for i := range concurrency { + wg.Add(1) + go func() { + defer wg.Done() + + req, err := http.NewRequest(http.MethodPost, "http://"+addr+"/read-input.php", bytes.NewReader(body)) + if err != nil { + errs <- err + return + } + req.ContentLength = bodySize + req.Header.Set("Content-Type", "application/octet-stream") + + resp, err := client.Do(req) + if err != nil { + errs <- fmt.Errorf("request %d: %w", i, err) + return + } + defer func() { _ = resp.Body.Close() }() + + got, err := io.ReadAll(resp.Body) + if err != nil { + errs <- fmt.Errorf("request %d: %w", i, err) + return + } + if string(got) != want { + errs <- fmt.Errorf("request %d: got %q, want %q", i, got, want) + } + }() + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + close(errs) + for err := range errs { + assert.NoError(t, err) + } + case <-time.After(30 * time.Second): + t.Fatal("concurrent uploads deadlocked: streams stalled waiting for a PHP thread") + } +} diff --git a/threadregular.go b/threadregular.go index 283627334f..220f53054a 100644 --- a/threadregular.go +++ b/threadregular.go @@ -156,7 +156,15 @@ func handleRequestWithRegularPHPThreads(ch contextHolder) error { regularThreadMu.RUnlock() } - // if no thread was available, mark the request as queued and fan it out to all threads + // no thread was available: drain the body so a stalled request does not + // hold its HTTP/2 flow-control window open while queued (see #1074) + if err := ch.frankenPHPContext.spoolRequestBody(); err != nil { + metrics.StopRequest() + + return err + } + + // mark the request as queued and fan it out to all threads queuedRegularThreads.Add(1) metrics.QueuedRequest() diff --git a/worker.go b/worker.go index 75d7915252..c93a6546f9 100644 --- a/worker.go +++ b/worker.go @@ -238,7 +238,15 @@ func (worker *worker) handleRequest(ch contextHolder) error { worker.threadMutex.RUnlock() } - // if no thread was available, mark the request as queued and apply the scaling strategy + // no thread was available: drain the body so a stalled request does not + // hold its HTTP/2 flow-control window open while queued (see #1074) + if err := ch.frankenPHPContext.spoolRequestBody(); err != nil { + metrics.StopWorkerRequest(worker.name, time.Since(ch.frankenPHPContext.startedAt)) + + return err + } + + // mark the request as queued and apply the scaling strategy worker.queuedRequests.Add(1) metrics.QueuedWorkerRequest(worker.name)