diff --git a/caddy/app.go b/caddy/app.go index fcee129180..f5bb578ca7 100644 --- a/caddy/app.go +++ b/caddy/app.go @@ -23,6 +23,7 @@ import ( var ( options []frankenphp.Option optionsMU sync.RWMutex + activeApp atomic.Pointer[FrankenPHPApp] ) // EXPERIMENTAL: RegisterWorkers provides a way for extensions to register frankenphp.Workers @@ -147,6 +148,7 @@ func (f *FrankenPHPApp) Start() error { // if FrankenPHP is currently running, shut it down first // this will happen in admin API reloads and caddy tests + activeApp.Store(f) frankenphp.Shutdown() if err := frankenphp.Init(f.opts...); err != nil { return err diff --git a/caddy/module.go b/caddy/module.go index 20dcec9ee2..caeac9be36 100644 --- a/caddy/module.go +++ b/caddy/module.go @@ -195,6 +195,10 @@ func (f *FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ c } } + if app := activeApp.Load(); app != nil && app != f.app { + return app.serveReloadedRequest(w, r) + } + ctx := r.Context() repl := ctx.Value(caddy.ReplacerCtxKey).(*caddy.Replacer) @@ -226,6 +230,11 @@ func (f *FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ c } err := f.server.ServeHTTP(w, r, opts...) + if errors.Is(err, frankenphp.ErrNotRunning) { + if app := activeApp.Load(); app != nil && app != f.app { + return app.serveReloadedRequest(w, r) + } + } if _, rejected := errors.AsType[frankenphp.ErrRejected](err); err != nil && !rejected { return caddyhttp.Error(http.StatusInternalServerError, err) @@ -234,6 +243,37 @@ func (f *FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ c return nil } +// Old HTTP handlers can still receive requests while the replacement PHP app +// starts. Wait for it, then route the original request through its HTTP server. +func (f *FrankenPHPApp) serveReloadedRequest(w http.ResponseWriter, r *http.Request) error { + select { + case <-f.started: + case <-r.Context().Done(): + return r.Context().Err() + case <-time.After(10 * time.Second): + return caddyhttp.Error(http.StatusServiceUnavailable, frankenphp.ErrNotRunning) + } + if !f.hasStarted.Load() { + return caddyhttp.Error(http.StatusServiceUnavailable, frankenphp.ErrNotRunning) + } + + previous := r.Context().Value(caddyhttp.ServerCtxKey).(*caddyhttp.Server) + for _, server := range f.httpApp.Servers { + if !slices.Equal(server.Listen, previous.Listen) { + continue + } + + original := r.Context().Value(caddyhttp.OriginalRequestCtxKey).(http.Request) + r = r.Clone(r.Context()) + r.Method, r.RequestURI = original.Method, original.RequestURI + *r.URL = *original.URL + server.ServeHTTP(w, r) + return nil + } + + return caddyhttp.Error(http.StatusServiceUnavailable, frankenphp.ErrNotRunning) +} + // UnmarshalCaddyfile implements caddyfile.Unmarshaler. func (f *FrankenPHPModule) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { for d.Next() { diff --git a/scaling.go b/scaling.go index dd21a7e37c..16646912f8 100644 --- a/scaling.go +++ b/scaling.go @@ -93,6 +93,11 @@ func scaleWorkerThread(worker *worker, done chan struct{}, mstate *state.ThreadS return } + // Do not scale a worker from a previous PHP runtime. + if worker.done != done { + return + } + thread, err := addWorkerThread(worker) if err != nil { if globalLogger.Enabled(globalCtx, slog.LevelWarn) { diff --git a/worker.go b/worker.go index 388dfbd031..4123b41739 100644 --- a/worker.go +++ b/worker.go @@ -27,6 +27,7 @@ type worker struct { maxThreads int requestOptions []RequestOption requestChan chan *frankenPHPContext + done <-chan struct{} threads []*phpThread threadMutex sync.RWMutex maxConsecutiveFailures int @@ -64,6 +65,8 @@ func initWorkers(opts []workerOpt) error { return err } + w.done = mainThread.done + totalThreadsToStart += w.num workers = append(workers, w) workersByName[w.name] = w @@ -277,6 +280,13 @@ func (worker *worker) handleRequest(fc *frankenPHPContext) error { return nil case workerScaleChan <- fc: // the request has triggered scaling, continue to wait for a thread + case <-worker.done: + worker.queuedRequests.Add(-1) + metrics.DequeuedWorkerRequest(worker.name) + metrics.StopWorkerRequest(worker.name, time.Since(fc.startedAt)) + + // No thread accepted the request; the caller can retry after a reload. + return ErrNotRunning case <-timeoutChan(time.Duration(maxWaitTime.Load())): // the request has timed out stalling worker.queuedRequests.Add(-1)