feat: background workers = non-HTTP workers with shared state

Background workers are long-running PHP workers outside the HTTP cycle that
observe their environment and publish configuration via set_vars/get_vars.
HTTP workers read this state per-request with version-based caching.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: nicolas-grekas

background_worker.go

@@ -0,0 +1,312 @@
+package frankenphp
+
+// #include "frankenphp.h"
+import "C"
+import (
+	"fmt"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+)
+
+type backgroundWorkerState struct {
+	varsPtr     unsafe.Pointer // *C.HashTable, persistent, managed by C
+	varsVersion atomic.Uint64  // incremented on each set_vars call
+	mu          sync.RWMutex
+	ready       chan struct{}
+	readyOnce   sync.Once
+}
+
+type BackgroundWorkerRegistry struct {
+	entrypoint     string
+	autoStartNames []string // names to start at boot (from Caddyfile match)
+	maxWorkers     int      // max lazy-started instances (0 = unlimited)
+	mu             sync.Mutex
+	workers        map[string]*backgroundWorkerState
+}
+
+func NewBackgroundWorkerRegistry(entrypoint string) *BackgroundWorkerRegistry {
+	return &BackgroundWorkerRegistry{
+		entrypoint: entrypoint,
+		workers:    make(map[string]*backgroundWorkerState),
+	}
+}
+
+func (registry *BackgroundWorkerRegistry) AddAutoStartNames(names ...string) {
+	registry.autoStartNames = append(registry.autoStartNames, names...)
+}
+
+func (registry *BackgroundWorkerRegistry) SetMaxWorkers(max int) {
+	registry.maxWorkers = max
+}
+
+func (registry *BackgroundWorkerRegistry) reserve(name string) (*backgroundWorkerState, bool, error) {
+	registry.mu.Lock()
+	defer registry.mu.Unlock()
+
+	if bgw := registry.workers[name]; bgw != nil {
+		return bgw, true, nil
+	}
+
+	if registry.maxWorkers > 0 && len(registry.workers) >= registry.maxWorkers {
+		return nil, false, fmt.Errorf("cannot start background worker %q: limit of %d reached - increase max_threads on the catch-all background worker or declare it as a named worker with match", name, registry.maxWorkers)
+	}
+
+	bgw := &backgroundWorkerState{
+		ready: make(chan struct{}),
+	}
+	registry.workers[name] = bgw
+
+	return bgw, false, nil
+}
+
+func (registry *BackgroundWorkerRegistry) remove(name string, bgw *backgroundWorkerState) {
+	registry.mu.Lock()
+	defer registry.mu.Unlock()
+
+	if registry.workers[name] == bgw {
+		delete(registry.workers, name)
+	}
+}
+
+var (
+	lockedVarsMu    sync.Mutex
+	lockedVarsStack = make(map[int][][]*backgroundWorkerState)
+)
+
+func startBackgroundWorker(thread *phpThread, bgWorkerName string) error {
+	if bgWorkerName == "" {
+		return fmt.Errorf("background worker name must not be empty")
+	}
+
+	registry := getRegistry(thread)
+	if registry == nil || registry.entrypoint == "" {
+		return fmt.Errorf("no background worker configured in this php_server")
+	}
+
+	return startBackgroundWorkerWithRegistry(registry, bgWorkerName)
+}
+
+func startBackgroundWorkerWithRegistry(registry *BackgroundWorkerRegistry, bgWorkerName string) error {
+	bgw, exists, err := registry.reserve(bgWorkerName)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return nil
+	}
+
+	worker, err := newWorker(workerOpt{
+		name:                   bgWorkerName,
+		fileName:               registry.entrypoint,
+		num:                    1,
+		env:                    PrepareEnv(nil),
+		watch:                  []string{},
+		maxConsecutiveFailures: -1,
+	})
+	if err != nil {
+		registry.remove(bgWorkerName, bgw)
+
+		return fmt.Errorf("failed to create background worker: %w", err)
+	}
+
+	worker.httpEnabled = false
+	worker.backgroundWorker = bgw
+	worker.backgroundRegistry = registry
+
+	bgWorkerThread := getInactivePHPThread()
+	if bgWorkerThread == nil {
+		registry.remove(bgWorkerName, bgw)
+
+		return fmt.Errorf("no available PHP thread for background worker (increase max_threads)")
+	}
+
+	scalingMu.Lock()
+	workers = append(workers, worker)
+	scalingMu.Unlock()
+
+	convertToWorkerThread(bgWorkerThread, worker)
+
+	if globalLogger.Enabled(globalCtx, slog.LevelInfo) {
+		globalLogger.LogAttrs(globalCtx, slog.LevelInfo, "background worker started", slog.String("name", bgWorkerName))
+	}
+
+	return nil
+}
+
+// startAutoBackgroundWorkers finds all registries with auto-start names and starts them.
+// Called before initWorkers so background workers have a head start on set_vars.
+func startAutoBackgroundWorkers(workerOpts []workerOpt) error {
+	seen := make(map[*BackgroundWorkerRegistry]struct{})
+	for _, w := range workerOpts {
+		if w.backgroundRegistry != nil {
+			if _, ok := seen[w.backgroundRegistry]; !ok {
+				seen[w.backgroundRegistry] = struct{}{}
+				if err := w.backgroundRegistry.startAutoWorkers(); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+// startAutoWorkers starts all background workers configured with match names.
+func (registry *BackgroundWorkerRegistry) startAutoWorkers() error {
+	for _, name := range registry.autoStartNames {
+		if err := startBackgroundWorkerWithRegistry(registry, name); err != nil {
+			return fmt.Errorf("failed to auto-start background worker %q: %w", name, err)
+		}
+	}
+	return nil
+}
+
+func getRegistry(thread *phpThread) *BackgroundWorkerRegistry {
+	if handler, ok := thread.handler.(*workerThread); ok && handler.worker.backgroundRegistry != nil {
+		return handler.worker.backgroundRegistry
+	}
+	if fc, ok := fromContext(thread.context()); ok {
+		return fc.backgroundRegistry
+	}
+
+	return nil
+}
+
+//export go_frankenphp_start_background_worker
+func go_frankenphp_start_background_worker(threadIndex C.uintptr_t, name *C.char, nameLen C.size_t) *C.char {
+	bgWorkerName := C.GoStringN(name, C.int(nameLen))
+
+	if err := startBackgroundWorker(phpThreads[threadIndex], bgWorkerName); err != nil {
+		return C.CString(err.Error())
+	}
+
+	return nil
+}
+
+// go_frankenphp_worker_wait_and_get waits for background workers to be ready and returns
+// their persistent HashTable pointers. outPtrs must point to a C-allocated array
+// of nameCount void* slots.
+//
+//export go_frankenphp_worker_wait_and_get
+func go_frankenphp_worker_wait_and_get(threadIndex C.uintptr_t, names **C.char, nameLens *C.size_t, nameCount C.int, timeoutMs C.int, outPtrs *unsafe.Pointer, callerVersions *C.ulong, outVersions *C.ulong) *C.char {
+	registry := getRegistry(phpThreads[threadIndex])
+	if registry == nil {
+		return C.CString("no background worker configured in this php_server")
+	}
+
+	n := int(nameCount)
+	nameSlice := unsafe.Slice(names, n)
+	nameLenSlice := unsafe.Slice(nameLens, n)
+	ptrSlice := unsafe.Slice(outPtrs, n)
+
+	sks := make([]*backgroundWorkerState, n)
+	goNames := make([]string, n)
+	for i := 0; i < n; i++ {
+		goNames[i] = C.GoStringN(nameSlice[i], C.int(nameLenSlice[i]))
+		registry.mu.Lock()
+		sks[i] = registry.workers[goNames[i]]
+		registry.mu.Unlock()
+		if sks[i] == nil {
+			return C.CString("background worker not found: " + goNames[i])
+		}
+	}
+
+	// Fast path: check if all versions match - skip lock + copy entirely
+	if callerVersions != nil {
+		callerVSlice := unsafe.Slice(callerVersions, n)
+		allMatch := true
+		for i, sk := range sks {
+			if uint64(callerVSlice[i]) != sk.varsVersion.Load() {
+				allMatch = false
+				break
+			}
+		}
+		if allMatch {
+			// Versions unchanged - caller can reuse cached data
+			for i := 0; i < n; i++ {
+				ptrSlice[i] = nil
+			}
+			return nil
+		}
+	}
+
+	timeout := time.Duration(timeoutMs) * time.Millisecond
+	done := make(chan error, 1)
+	go func() {
+		timer := time.NewTimer(timeout)
+		defer timer.Stop()
+		for i, sk := range sks {
+			select {
+			case <-sk.ready:
+				// background worker has called set_vars
+			case <-timer.C:
+				done <- fmt.Errorf("timeout waiting for background worker: %s", goNames[i])
+				return
+			}
+		}
+		done <- nil
+	}()
+
+	if err := <-done; err != nil {
+		return C.CString(err.Error())
+	}
+
+	outVSlice := unsafe.Slice(outVersions, n)
+	for i, sk := range sks {
+		sk.mu.RLock()
+		ptrSlice[i] = sk.varsPtr
+		outVSlice[i] = C.ulong(sk.varsVersion.Load())
+	}
+
+	// Push locked background workers onto a stack so release_vars can unlock them
+	// without re-lookup. Stack supports reentrant get_vars calls.
+	lockedVarsMu.Lock()
+	lockedVarsStack[int(threadIndex)] = append(lockedVarsStack[int(threadIndex)], sks)
+	lockedVarsMu.Unlock()
+
+	return nil
+}
+
+//export go_frankenphp_worker_release_vars
+func go_frankenphp_worker_release_vars(threadIndex C.uintptr_t) {
+	lockedVarsMu.Lock()
+	stack := lockedVarsStack[int(threadIndex)]
+	if len(stack) == 0 {
+		lockedVarsMu.Unlock()
+		return
+	}
+	sks := stack[len(stack)-1]
+	lockedVarsStack[int(threadIndex)] = stack[:len(stack)-1]
+	lockedVarsMu.Unlock()
+
+	for _, sk := range sks {
+		sk.mu.RUnlock()
+	}
+}
+
+//export go_frankenphp_worker_set_vars
+func go_frankenphp_worker_set_vars(threadIndex C.uintptr_t, varsPtr unsafe.Pointer, oldPtr *unsafe.Pointer) *C.char {
+	thread := phpThreads[threadIndex]
+
+	handler, ok := thread.handler.(*workerThread)
+	if !ok || handler.worker.httpEnabled || handler.worker.backgroundWorker == nil {
+		return C.CString("frankenphp_worker_set_vars() can only be called from a background worker")
+	}
+
+	sk := handler.worker.backgroundWorker
+
+	sk.mu.Lock()
+	*oldPtr = sk.varsPtr
+	sk.varsPtr = varsPtr
+	sk.varsVersion.Add(1)
+	sk.mu.Unlock()
+
+	sk.readyOnce.Do(func() {
+		handler.markBackgroundReady()
+		close(sk.ready)
+	})
+
+	return nil
+}