Skip to content

feat: declared background workers + frankenphp_get_worker_handle() #2617

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nicolas-grekas wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: declared background workers + frankenphp_get_worker_handle() #2617

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
279 changes: 279 additions & 0 deletions bgworker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,279 @@
package frankenphp_test

import (
"bytes"
"net/http"
"os"
"path/filepath"
"testing"
"time"

"github.com/dunglas/frankenphp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

// requireFileEventually asserts that `path` appears on disk before the
// deadline. Wraps require.Eventually so call sites stay short.
func requireFileEventually(t testing.TB, path string, msgAndArgs ...any) {
t.Helper()
require.Eventually(t, func() bool {
_, err := os.Stat(path)
return err == nil
}, 5*time.Second, 25*time.Millisecond, msgAndArgs...)
}

// TestBackgroundWorkerLifecycle boots a background worker that touches a
// sentinel file then parks on its handle. It proves the bg worker runs
// (sentinel appears) and that Shutdown returns within a reasonable time.
// The test asserts on Shutdown timing, so it manages Shutdown itself
// instead of using initServers' t.Cleanup hook.
func TestBackgroundWorkerLifecycle(t *testing.T) {
tmp := t.TempDir()
sentinel := filepath.Join(tmp, "bg-lifecycle.sentinel")

require.NoError(t, frankenphp.Init(
frankenphp.WithWorkers("bg-lifecycle", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerEnv(map[string]string{"BG_SENTINEL": sentinel}),
),
frankenphp.WithNumThreads(2),
))

requireFileEventually(t, sentinel, "background worker did not touch sentinel")

done := make(chan struct{})
go func() {
frankenphp.Shutdown()
close(done)
}()

select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatalf("Shutdown did not return within 10s")
}
}

// TestBackgroundWorkerCrashRestarts boots a worker that exit(1)s on its
// first run and touches a "restarted" sentinel on its second run. The
// sentinel proves the crash-restart loop fired.
func TestBackgroundWorkerCrashRestarts(t *testing.T) {
tmp := t.TempDir()
crashMarker := filepath.Join(tmp, "bg-crash.marker")
restarted := filepath.Join(tmp, "bg-crash.restarted")

initServers(t,
frankenphp.WithWorkers("bg-crash", "testdata/bgworker/crash.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerEnv(map[string]string{
"BG_CRASH_MARKER": crashMarker,
"BG_RESTARTED_SENTINEL": restarted,
}),
),
frankenphp.WithNumThreads(2),
)

requireFileEventually(t, restarted, "background worker did not restart after crash")
}

// TestBackgroundWorkerOnServer scopes a background worker to a Server. It
// proves that the worker inherits the server env (the sentinel directory is
// declared on the server, not on the worker), that FRANKENPHP_WORKER holds
// the worker name, and that the worker does not intercept HTTP requests
// served by the same server.
func TestBackgroundWorkerOnServer(t *testing.T) {
tmp := t.TempDir()

server, err := frankenphp.NewServer(
testDataDir,
frankenphp.WithServerName("sidekick-server"),
frankenphp.WithServerEnv(map[string]string{"BG_SENTINEL_DIR": tmp}),
)
require.NoError(t, err)

globalSentinel := filepath.Join(tmp, "global.sentinel")
initServers(t,
frankenphp.WithServer(server),
frankenphp.WithWorkers("jobs", "testdata/bgworker/named.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerServerScope(server),
),
// a global worker may reuse the name: names are scoped to their server
frankenphp.WithWorkers("jobs", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerEnv(map[string]string{"BG_SENTINEL": globalSentinel}),
),
frankenphp.WithNumThreads(3),
)

// named.php touches "<BG_SENTINEL_DIR>/<FRANKENPHP_WORKER>": the script sees
// the declared name, not the server-qualified one used by metrics and logs
requireFileEventually(t, filepath.Join(tmp, "jobs"), "background worker did not touch its per-name sentinel")
requireFileEventually(t, globalSentinel, "the global worker sharing the name did not start")

body := serverGet(t, server, "http://example.com/index.php")
assert.Contains(t, body, "I am by birth a Genevese", "the server must still serve regular requests")
}

// TestBackgroundWorkerValidation covers the declaration-time errors.
func TestBackgroundWorkerValidation(t *testing.T) {
t.Cleanup(frankenphp.Shutdown)

t.Run("name is required", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("", "testdata/bgworker/basic.php", 1, frankenphp.WithWorkerBackground()),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "must have an explicit name")
})

t.Run("num must be >= 1", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("bg-zero", "testdata/bgworker/basic.php", 0, frankenphp.WithWorkerBackground()),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "must declare num >= 1")
})

t.Run("names are unique within a server", func(t *testing.T) {
// a global and a server-scoped worker may share a name (see
// TestBackgroundWorkerOnServer), two workers of one server may not
server, err := frankenphp.NewServer(testDataDir)
require.NoError(t, err)
err = frankenphp.Init(
frankenphp.WithServer(server),
frankenphp.WithWorkers("bg-shared", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerServerScope(server),
),
frankenphp.WithWorkers("bg-shared", "testdata/bgworker/named.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerServerScope(server),
),
frankenphp.WithNumThreads(3),
)
require.ErrorContains(t, err, "two workers in a server cannot have the same name")
})

t.Run("early return without the handle fails startup", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("bg-early", "testdata/bgworker/early-return.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerMaxFailures(2),
),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "frankenphp_get_worker_handle")
})

t.Run("fetching the handle without waiting on it fails startup", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("bg-no-wait", "testdata/bgworker/fetch-no-wait.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerMaxFailures(2),
),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "waiting on its handle")
})

t.Run("max_threads is rejected", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("bg-scaled", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerMaxThreads(2),
),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "cannot set max_threads")
})

t.Run("an unregistered server scope is rejected", func(t *testing.T) {
unregistered, err := frankenphp.NewServer(testDataDir)
require.NoError(t, err)
err = frankenphp.Init(
frankenphp.WithWorkers("bg-orphan", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerServerScope(unregistered),
),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "not passed to WithServer()")
})

t.Run("request matchers are rejected", func(t *testing.T) {
err := frankenphp.Init(
frankenphp.WithWorkers("bg-matched", "testdata/bgworker/basic.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerMatcher(func(*http.Request) bool { return true }),
),
frankenphp.WithNumThreads(2),
)
require.ErrorContains(t, err, "cannot match requests")
})
}

// TestBackgroundWorkerParksOnRead checks that a blocking read on the handle
// is a wait too: Init() returns only once the worker is ready, and the EOF
// of the drain unblocks the read so Shutdown() returns promptly.
func TestBackgroundWorkerParksOnRead(t *testing.T) {
tmp := t.TempDir()
sentinel := filepath.Join(tmp, "bg-read.sentinel")

require.NoError(t, frankenphp.Init(
frankenphp.WithWorkers("bg-read", "testdata/bgworker/read.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerEnv(map[string]string{"BG_SENTINEL": sentinel}),
),
frankenphp.WithNumThreads(2),
))
requireFileEventually(t, sentinel, "background worker parked on a read did not start")

done := make(chan struct{})
go func() {
frankenphp.Shutdown()
close(done)
}()
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("Shutdown did not return within 10s: the read did not observe EOF")
}
}

// TestBackgroundWorkerRestartDrainsParkedScript checks that RestartWorkers()
// wakes a parked background script through the drain and re-runs it.
func TestBackgroundWorkerRestartDrainsParkedScript(t *testing.T) {
tmp := t.TempDir()
countFile := filepath.Join(tmp, "bg-count.log")

initServers(t,
frankenphp.WithWorkers("bg-count", "testdata/bgworker/count.php", 1,
frankenphp.WithWorkerBackground(),
frankenphp.WithWorkerEnv(map[string]string{"BG_COUNT_FILE": countFile}),
),
frankenphp.WithNumThreads(2),
)
runs := func() int {
b, _ := os.ReadFile(countFile)
return bytes.Count(b, []byte("\n"))
}
require.Eventually(t, func() bool { return runs() == 1 }, 5*time.Second, 25*time.Millisecond, "background worker did not start")

frankenphp.RestartWorkers()

require.Eventually(t, func() bool { return runs() == 2 }, 5*time.Second, 25*time.Millisecond, "background worker was not re-run after the restart")
}

// TestGetWorkerHandleOutsideBackgroundWorker checks the function throws on a
// regular request thread instead of handing out a stream.
func TestGetWorkerHandleOutsideBackgroundWorker(t *testing.T) {
server, err := frankenphp.NewServer(testDataDir)
require.NoError(t, err)
initServers(t, frankenphp.WithServer(server), frankenphp.WithNumThreads(1))

body := serverGet(t, server, "http://example.com/handle-outside.php")

assert.Contains(t, body, "can only be called from a background worker")
}
51 changes: 8 additions & 43 deletions caddy/app.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ import (
"github.com/caddyserver/caddy/v2/caddyconfig/httpcaddyfile"
"github.com/caddyserver/caddy/v2/modules/caddyhttp"
"github.com/dunglas/frankenphp"
"github.com/dunglas/frankenphp/internal/fastabs"
)

var (
Expand Down Expand Up @@ -60,15 +59,14 @@ type FrankenPHPApp struct {
// EXPERIMENTAL: MaxRequests sets the maximum number of requests a PHP thread handles before restarting (0 = unlimited)
MaxRequests int `json:"max_requests,omitempty"`

opts []frankenphp.Option
metrics frankenphp.Metrics
ctx context.Context
logger *slog.Logger
modules []*FrankenPHPModule
usedWorkerNames map[string]bool
httpApp *caddyhttp.App
hasStarted atomic.Bool
started chan any
opts []frankenphp.Option
metrics frankenphp.Metrics
ctx context.Context
logger *slog.Logger
modules []*FrankenPHPModule
httpApp *caddyhttp.App
hasStarted atomic.Bool
started chan any
}

var errIni = errors.New(`"php_ini" must be in the format: php_ini "<key>" "<value>"`)
Expand Down Expand Up @@ -133,7 +131,6 @@ func (f *FrankenPHPApp) Start() error {
// register global workers
for _, w := range f.Workers {
w.FileName = repl.ReplaceKnown(w.FileName, "")
w.Name = f.createUniqueWorkerName(w, "")
opts, err := w.toWorkerOptions()
if err != nil {
return err
Expand Down Expand Up @@ -224,7 +221,6 @@ func (f *FrankenPHPApp) registerModule(repl *caddy.Replacer, module *FrankenPHPM

for _, w := range module.Workers {
w.FileName = repl.ReplaceKnown(w.FileName, "")
w.Name = f.createUniqueWorkerName(w, serverName)
workerOptions, err := w.toWorkerOptions()
if err != nil {
return err
Expand All @@ -236,37 +232,6 @@ func (f *FrankenPHPApp) registerModule(repl *caddy.Replacer, module *FrankenPHPM
return nil
}

// avoid name collisions for workers
// on collision, a name is first qualified with the server name
// ("<serverName>:<name>") before falling back to a numeric postfix
func (f *FrankenPHPApp) createUniqueWorkerName(wc workerConfig, serverName string) string {
if f.usedWorkerNames == nil {
f.usedWorkerNames = make(map[string]bool)
}

if wc.Name == "" {
wc.Name, _ = fastabs.FastAbs(wc.FileName)
}

name := wc.Name
suffix := 0
for {
if _, ok := f.usedWorkerNames[name]; !ok {
f.usedWorkerNames[name] = true
break
}
if serverName != "" {
name = serverName + ":" + wc.Name
serverName = ""
continue
}
suffix++
name = fmt.Sprintf("%s_%d", wc.Name, suffix)
}

return name
}

// UnmarshalCaddyfile implements caddyfile.Unmarshaler.
func (f *FrankenPHPApp) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
for d.Next() {
Expand Down
Loading
Loading