fix: reuse created-once channels so queued requests don't get stuck on replaced channels (#2470)

fixes #2469

Author: henderkes

.github/actions/watcher/action.yaml

@@ -18,7 +18,8 @@ runs:
     - if: steps.cache-watcher.outputs.cache-hit != 'true'
       name: Compile e-dant/watcher
       run: |
-        mkdir watcher
+        mkdir -p watcher
+        rm -rf watcher/*
         gh release download --repo e-dant/watcher -A tar.gz -O - | tar -xz -C watcher --strip-components 1
         cd watcher
         cmake -S . -B build -DCMAKE_BUILD_TYPE=Release

frankenphp.go

@@ -32,6 +32,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 	"unsafe"
@@ -66,7 +67,8 @@ var (
 
 	metrics Metrics = nullMetrics{}
 
-	maxWaitTime          time.Duration
+	// atomic: read by in-flight requests while a reload may rewrite it
+	maxWaitTime          atomic.Int64
 	maxRequestsPerThread int
 )
 
@@ -275,7 +277,7 @@ func Init(options ...Option) error {
 		metrics = opt.metrics
 	}
 
-	maxWaitTime = opt.maxWaitTime
+	maxWaitTime.Store(int64(opt.maxWaitTime))
 	maxRequestsPerThread = opt.maxRequests
 
 	if opt.maxIdleTime > 0 {
@@ -317,7 +319,10 @@ func Init(options ...Option) error {
 		return err
 	}
 
-	regularRequestChan = make(chan contextHolder)
+	// reused across reloads so queued requests aren't orphaned on a stale channel
+	if regularRequestChan == nil {
+		regularRequestChan = make(chan contextHolder)
+	}
 	regularThreads = make([]*phpThread, 0, opt.numThreads-workerThreadCount)
 	for i := 0; i < opt.numThreads-workerThreadCount; i++ {
 		convertToRegularThread(getInactivePHPThread())

phpmainthread_test.go

@@ -166,6 +166,65 @@ func TestTransitionThreadsWhileDoingRequests(t *testing.T) {
 	transitionsWG.Wait()
 }
 
+// Test https://github.com/php/frankenphp/issues/2469
+// A request queued for a thread when the server reloads must be served
+// by the new threads, not rejected with ErrMaxWaitTimeExceeded.
+func TestQueuedRequestSurvivesReload(t *testing.T) {
+	t.Cleanup(Shutdown)
+
+	initOpts := []Option{
+		WithNumThreads(1),
+		WithMaxThreads(1),
+		WithMaxWaitTime(5 * time.Second),
+	}
+	assert.NoError(t, Init(initOpts...))
+
+	doRequest := func(query string) error {
+		r := httptest.NewRequest("GET", "http://localhost/sleep.php?"+query, nil)
+		w := httptest.NewRecorder()
+		req, err := NewRequestWithContext(r, WithRequestDocumentRoot(testDataPath, false))
+		if err != nil {
+			return err
+		}
+
+		return ServeHTTP(w, req)
+	}
+
+	// Occupy the single PHP thread so the next request has to wait in the queue.
+	started := make(chan struct{})
+	go func() {
+		close(started)
+		_ = doRequest("sleep=800")
+	}()
+	<-started
+	time.Sleep(100 * time.Millisecond)
+
+	queued := make(chan error, 1)
+	go func() {
+		queued <- doRequest("sleep=0")
+	}()
+
+	// Wait until the second request is genuinely queued waiting for a thread.
+	deadline := time.Now().Add(2 * time.Second)
+	for queuedRegularThreads.Load() == 0 {
+		if !assert.True(t, time.Now().Before(deadline), "request was never queued") {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	// Reload while the request is queued, exactly like the Caddy module does.
+	Shutdown()
+	assert.NoError(t, Init(initOpts...))
+
+	select {
+	case err := <-queued:
+		assert.NoError(t, err, "a queued request must not be rejected by a reload")
+	case <-time.After(15 * time.Second):
+		t.Fatal("the queued request never completed after the reload")
+	}
+}
+
 func TestFinishBootingAWorkerScript(t *testing.T) {
 	setupGlobals(t)
 

scaling.go

@@ -35,16 +35,19 @@ var (
 )
 
 func initAutoScaling(mainThread *phpMainThread) {
+	// reused across reloads so queued requests aren't orphaned on a stale channel
+	if scaleChan == nil {
+		scaleChan = make(chan *frankenPHPContext)
+	}
+
 	if mainThread.maxThreads <= mainThread.numThreads {
-		scaleChan = nil
 		return
 	}
 
 	done := mainThread.done
 	mstate := mainThread.state
 
 	scalingMu.Lock()
-	scaleChan = make(chan *frankenPHPContext)
 	maxScaledThreads := mainThread.maxThreads - mainThread.numThreads
 	autoScaledThreads = make([]*phpThread, 0, maxScaledThreads)
 	scalingMu.Unlock()

threadregular.go

@@ -6,6 +6,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/dunglas/frankenphp/internal/state"
 )
@@ -169,7 +170,7 @@ func handleRequestWithRegularPHPThreads(ch contextHolder) error {
 			return nil
 		case scaleChan <- ch.frankenPHPContext:
 			// the request has triggered scaling, continue to wait for a thread
-		case <-timeoutChan(maxWaitTime):
+		case <-timeoutChan(time.Duration(maxWaitTime.Load())):
 			// the request has timed out stalling
 			queuedRegularThreads.Add(-1)
 			metrics.DequeuedRequest()

worker.go

@@ -330,7 +330,7 @@ func (worker *worker) handleRequest(ch contextHolder) error {
 			return nil
 		case workerScaleChan <- ch.frankenPHPContext:
 			// the request has triggered scaling, continue to wait for a thread
-		case <-timeoutChan(maxWaitTime):
+		case <-timeoutChan(time.Duration(maxWaitTime.Load())):
 			// the request has timed out stalling
 			worker.queuedRequests.Add(-1)
 			metrics.DequeuedWorkerRequest(worker.name)