fix: flush

Author: Tolsee

caddy/caddy_test.go

@@ -973,14 +973,15 @@ func TestMaxWaitTimeWorker(t *testing.T) {
 		{
 			skip_install_trust
 			admin localhost:2999
-		    metrics
 			http_port `+testPort+`
+			metrics
 
 			frankenphp {
+				num_threads 2
 				max_wait_time 1ns
 				worker {
-					name service
 					num 1
+					name service
 					file ../testdata/sleep.php
 				}
 			}
@@ -1001,24 +1002,31 @@ func TestMaxWaitTimeWorker(t *testing.T) {
 	wg.Add(10)
 	for i := 0; i < 10; i++ {
 		go func() {
-			statusCode := getStatusCode("http://localhost:"+testPort+"/sleep.php?sleep=100", t)
-			fmt.Printf("Status code: %d\n", statusCode)
+			statusCode := getStatusCode("http://localhost:"+testPort+"/sleep.php?sleep=10000&iteration=1", t)
 			if statusCode == http.StatusGatewayTimeout {
 				success.Store(true)
 			}
 			wg.Done()
 		}()
 	}
 	wg.Wait()
-
 	require.True(t, success.Load(), "At least one request should have failed with a 504 Gateway Timeout status")
+
+	// Fetch metrics
+	resp, err := http.Get("http://localhost:2999/metrics")
+	require.NoError(t, err, "failed to fetch metrics")
+	defer resp.Body.Close()
+
+	// Read and parse metrics
+	metrics := new(bytes.Buffer)
+	_, err = metrics.ReadFrom(resp.Body)
+
 	expectedMetrics := `
 	# TYPE frankenphp_worker_queue_depth gauge
-	frankenphp_worker_queue_depth{worker="service"} 0
+	frankenphp_worker_queue_depth{worker="service"} 9
 	`
 
 	ctx := caddy.ActiveContext()
-	fmt.Printf("Metrics: %d\n", testutil.CollectAndCount(ctx.GetMetricsRegistry(), "frankenphp_worker_queue_depth"))
 	require.NoError(t,
 		testutil.GatherAndCompare(
 			ctx.GetMetricsRegistry(),

context.go

@@ -152,7 +152,10 @@ func (fc *frankenPHPContext) reject(statusCode int, message string) {
 	if rw != nil {
 		rw.WriteHeader(statusCode)
 		_, _ = rw.Write([]byte(message))
-		rw.(http.Flusher).Flush()
+		
+		if f, ok := rw.(http.Flusher); ok {
+    		f.Flush()
+		}
 	}
 
 	fc.closeContext()

worker.go

@@ -225,6 +225,7 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) {
 		case scaleChan <- fc:
 			// the request has triggered scaling, continue to wait for a thread
 		case <-timeoutChan(maxWaitTime):
+			// metrics.DequeuedWorkerRequest(worker.name)
 			// the request has timed out stalling
 			fc.reject(504, "Gateway Timeout")
 			return