fix: route extension requests to their worker, refuse messages once stopped

requestoptions.go

@@ -205,6 +205,15 @@ func WithRequestBodyTimeout(timeout time.Duration) RequestOption {
 	}
 }
 
+// withWorker dispatches the request on a worker directly, no lookup
+func withWorker(w *worker) RequestOption {
+	return func(o *frankenPHPContext) error {
+		o.worker = w
+
+		return nil
+	}
+}
+
 // WithWorkerName sets the worker that should handle the request
 func WithWorkerName(name string) RequestOption {
 	return func(o *frankenPHPContext) error {

workerextension.go

@@ -24,27 +24,45 @@ type extensionWorkers struct {
 	internalWorker *worker
 }
 
+// server is the one the worker belongs to, the fallback one for a global
+// worker, nil before Init()
+func (w *extensionWorkers) server() *Server {
+	switch {
+	case w.internalWorker == nil:
+		return nil
+	case w.internalWorker.server != nil:
+		return w.internalWorker.server
+	default:
+		return fallbackServer
+	}
+}
+
 // EXPERIMENTAL: SendRequest sends an HTTP request to the worker and writes the response to the provided ResponseWriter.
 func (w *extensionWorkers) SendRequest(rw http.ResponseWriter, r *http.Request) error {
-	fr, err := NewRequestWithContext(
-		r,
-		WithOriginalRequest(r),
-		WithWorkerName(w.name),
-	)
-
-	if err != nil {
-		return err
+	server := w.server()
+	if server == nil {
+		return ErrNotRunning
 	}
 
-	return ServeHTTP(rw, fr)
+	// the worker itself, the name may be the empty one it was declared with
+	return server.ServeHTTP(rw, r, WithOriginalRequest(r), withWorker(w.internalWorker))
 }
 
 func (w *extensionWorkers) NumThreads() int {
+	if w.internalWorker == nil {
+		return 0
+	}
+
 	return w.internalWorker.countThreads()
 }
 
 // EXPERIMENTAL: SendMessage sends a message to the worker and waits for a response.
 func (w *extensionWorkers) SendMessage(ctx context.Context, message any, rw http.ResponseWriter) (any, error) {
+	// the worker stays referenced past Shutdown(), with no thread left to take the message
+	if server := w.server(); server == nil || !server.isRegistered.Load() {
+		return nil, ErrNotRunning
+	}
+
 	fc := newContextFromMessage(message, rw, ctx, w.internalWorker)
 	err := w.internalWorker.handleRequest(fc)
 

workerextension_test.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -77,6 +78,43 @@ func TestWorkerExtensionSendMessage(t *testing.T) {
 	assert.Equal(t, "received message: Hello Workers", ret)
 }
 
+// an extension worker without a name is still the one SendRequest()
+// dispatches on
+func TestWorkerExtensionSendRequestWithoutName(t *testing.T) {
+	t.Cleanup(Shutdown)
+
+	externalWorker, o := WithExtensionWorkers("", "testdata/worker.php", 1)
+	require.NoError(t, Init(o))
+
+	w := httptest.NewRecorder()
+	require.NoError(t, externalWorker.SendRequest(w, httptest.NewRequest("GET", "http://example.com/index.php", nil)))
+
+	body, err := io.ReadAll(w.Result().Body)
+	require.NoError(t, err)
+	assert.Contains(t, string(body), "Requests handled: 0")
+}
+
+// SendMessage() after Shutdown() reports the stopped runtime instead of
+// waiting for a thread
+func TestWorkerExtensionSendMessageAfterShutdown(t *testing.T) {
+	externalWorker, o := WithExtensionWorkers("extensionWorkers", "testdata/message-worker.php", 1)
+	require.NoError(t, Init(o))
+	Shutdown()
+
+	errChan := make(chan error, 1)
+	go func() {
+		_, err := externalWorker.SendMessage(t.Context(), "Hello Workers", nil)
+		errChan <- err
+	}()
+
+	select {
+	case err := <-errChan:
+		require.ErrorIs(t, err, ErrNotRunning)
+	case <-time.After(2 * time.Second):
+		t.Fatal("SendMessage() did not return after Shutdown()")
+	}
+}
+
 func TestErrorIf2WorkersHaveSameName(t *testing.T) {
 	_, o1 := WithExtensionWorkers("duplicateWorker", "testdata/worker.php", 1)
 	_, o2 := WithExtensionWorkers("duplicateWorker", "testdata/worker2.php", 1)