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

Author: dunglas

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

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

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

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

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