feat: worker ticking

caddy/config_test.go

@@ -284,3 +284,96 @@ func TestCreateUniqueWorkerNamesQualifiedByServer(t *testing.T) {
 	// workers without a server keep the numeric postfix behavior
 	require.Equal(t, "queue_2", app.createUniqueWorkerName(wc, ""))
 }
+
+func TestModuleWorkerWithTickConfiguration(t *testing.T) {
+	configWithTick := `
+	{
+		php {
+			worker ../testdata/worker-with-counter.php {
+				tick 60s health
+				tick each 1m aligned message
+				tick overlap 1h aligned message
+				tick idle 3s "HELLO THERE!!!!"
+			}
+		}
+	}`
+
+	d := caddyfile.NewTestDispenser(configWithTick)
+	module := &FrankenPHPModule{}
+
+	err := module.UnmarshalCaddyfile(d)
+	require.NoError(t, err)
+	require.Len(t, module.Workers, 1)
+
+	ticks := module.Workers[0].Ticks
+	require.Len(t, ticks, 4)
+	require.Equal(t, 60*time.Second, ticks[0].Interval)
+	require.Equal(t, "health", ticks[0].Message)
+	require.False(t, ticks[0].Aligned)
+	require.Equal(t, frankenphp.TickModeSynchronous, ticks[0].Mode)
+
+	require.Equal(t, time.Minute, ticks[1].Interval)
+	require.Equal(t, "message", ticks[1].Message)
+	require.True(t, ticks[1].Aligned)
+	require.Equal(t, frankenphp.TickModeEach, ticks[1].Mode)
+
+	require.Equal(t, time.Hour, ticks[2].Interval)
+	require.Equal(t, "message", ticks[2].Message)
+	require.True(t, ticks[2].Aligned)
+	require.Equal(t, frankenphp.TickModeOverlapping, ticks[2].Mode)
+
+	require.Equal(t, "HELLO THERE!!!!", ticks[3].Message)
+	require.False(t, ticks[3].Aligned)
+	require.Equal(t, frankenphp.TickModeIdle, ticks[3].Mode)
+	require.Equal(t, 3*time.Second, ticks[3].Interval)
+}
+
+func TestModuleWorkerWithInvalidTickConfiguration(t *testing.T) {
+	tests := []struct {
+		name   string
+		config string
+	}{
+		{
+			name: "missing message",
+			config: `{
+				php {
+					worker {
+						file ../testdata/worker-with-counter.php
+						tick 60s
+					}
+				}
+			}`,
+		},
+		{
+			name: "invalid interval",
+			config: `{
+				php {
+					worker {
+						file ../testdata/worker-with-counter.php
+						tick not-a-duration health
+					}
+				}
+			}`,
+		},
+		{
+			name: "each must come first",
+			config: `{
+				php {
+					worker {
+						file ../testdata/worker-with-counter.php
+						tick 60s health each
+					}
+				}
+			}`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			d := caddyfile.NewTestDispenser(tt.config)
+			module := &FrankenPHPModule{}
+			err := module.UnmarshalCaddyfile(d)
+			require.Error(t, err)
+		})
+	}
+}

caddy/workerconfig.go

@@ -1,8 +1,10 @@
 package caddy
 
 import (
+	"fmt"
 	"path/filepath"
 	"strconv"
+	"time"
 
 	"github.com/caddyserver/caddy/v2"
 	"github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
@@ -22,6 +24,7 @@ import (
 type workerConfig struct {
 	mercureContext
 
+	// Name for the worker. Default: the absolute path of the worker file, postfixed with a number if the name is already used.
 	// Name for the worker. Default: the absolute path of the worker file, postfixed with a number if the name is already used.
 	Name string `json:"name,omitempty"`
 	// FileName sets the path to the worker script.
@@ -38,10 +41,26 @@ type workerConfig struct {
 	MatchPath []string `json:"match_path,omitempty"`
 	// MaxConsecutiveFailures sets the maximum number of consecutive failures before panicking (defaults to 6, set to -1 to never panick)
 	MaxConsecutiveFailures int `json:"max_consecutive_failures,omitempty"`
+	// Ticks configures periodic internal messages sent to the worker.
+	Ticks []*tickConfig `json:"ticks,omitempty"`
 
 	options []frankenphp.WorkerOption
 }
 
+type tickConfig struct {
+	Interval time.Duration       `json:"interval"`
+	Message  string              `json:"message"`
+	Aligned  bool                `json:"aligned,omitempty"`
+	Mode     frankenphp.TickMode `json:"mode,omitempty"`
+}
+
+var tickModes = map[string]frankenphp.TickMode{
+	"sync":    frankenphp.TickModeSynchronous,
+	"overlap": frankenphp.TickModeOverlapping,
+	"each":    frankenphp.TickModeEach,
+	"idle":    frankenphp.TickModeIdle,
+}
+
 func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) {
 	wc := workerConfig{}
 	if d.NextArg() {
@@ -139,8 +158,15 @@ func unmarshalWorker(d *caddyfile.Dispenser) (workerConfig, error) {
 			}
 
 			wc.MaxConsecutiveFailures = v
+		case "tick":
+			tick, err := parseTickConfig(d)
+			if err != nil {
+				return wc, d.WrapErr(err)
+			}
+
+			wc.Ticks = append(wc.Ticks, tick)
 		default:
-			return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, max_consecutive_failures, max_threads", v)
+			return wc, wrongSubDirectiveError("worker", "name, file, num, env, watch, match, tick, max_consecutive_failures, max_threads", v)
 		}
 	}
 
@@ -175,5 +201,47 @@ func (wc *workerConfig) toWorkerOptions() ([]frankenphp.WorkerOption, error) {
 		}
 		opts = append(opts, frankenphp.WithWorkerMatcher(matchFunc.Match))
 	}
+
+	if len(wc.Ticks) > 0 {
+		for _, t := range wc.Ticks {
+			opts = append(opts, frankenphp.WithWorkerTicks(t.Mode, t.Interval, t.Message, t.Aligned))
+		}
+	}
+
 	return opts, nil
 }
+
+// parse the configuration for recurring ticks to the worker
+// tick 1s "Hello, world!"
+// tick overlap aligned 1m "Hello, world!"
+func parseTickConfig(d *caddyfile.Dispenser) (*tickConfig, error) {
+	args := d.RemainingArgs()
+	if len(args) < 2 {
+		return nil, d.ArgErr()
+	}
+
+	mode := frankenphp.TickModeSynchronous
+	if m, ok := tickModes[args[0]]; ok {
+		mode, args = m, args[1:]
+	}
+
+	aligned := false
+	if len(args) > 2 && args[len(args)-2] == "aligned" {
+		aligned = true
+		args = append(args[:len(args)-2], args[len(args)-1])
+	}
+
+	if len(args) != 2 {
+		return nil, d.ArgErr()
+	}
+
+	interval, err := time.ParseDuration(args[0])
+	if err != nil {
+		return nil, fmt.Errorf("tick interval must be a valid duration, received: %s (%s)", args[0], err)
+	}
+	if interval <= 0 {
+		return nil, fmt.Errorf("tick interval must be positive, received: %s (%s)", args[0], interval)
+	}
+
+	return &tickConfig{Interval: interval, Message: args[1], Aligned: aligned, Mode: mode}, nil
+}

docs/config.md

@@ -111,6 +111,7 @@ You can also explicitly configure FrankenPHP using the [global option](https://c
 			watch <path> # Sets the path to watch for file changes. Can be specified more than once for multiple paths.
 			name <name> # Sets the name of the worker, used in logs and metrics. Default: absolute path of worker file
 			max_consecutive_failures <num> # Sets the maximum number of consecutive failures before the worker is considered unhealthy, -1 means the worker will always restart. Default: 6.
+			tick <mode> <interval> [aligned] <message> # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align ticks to the start of each interval. Mode can be sync, overlap, each and idle. Can be specified more than once.
 		}
 	}
 }
@@ -198,6 +199,8 @@ php_server [<matcher>] {
 		watch <path> # Sets the path to watch for file changes. Can be specified more than once for multiple paths.
 		env <key> <value> # Sets an extra environment variable to the given value. Can be specified more than once for multiple environment variables. Environment variables for this worker are also inherited from the php_server parent, but can be overwritten here.
 		match <path> # match the worker to a path pattern. Overrides try_files and can only be used in the php_server directive.
+		tick <interval> [aligned] <message> # Sends a periodic message to the worker via frankenphp_handle_request(). Interval must be a duration (e.g. 60s, 1m). Use aligned to align ticks to the start of each interval. Can be specified more than once.
+		tick <mode> <interval> [aligned] <message> # Like tick, with an explicit mode (sync, overlap, each, idle).
 	}
 	worker <other_file> <num> # Can also use the short form like in the global frankenphp block.
 }

docs/worker.md

@@ -256,3 +256,36 @@ while (\frankenphp_handle_request($handler)) {
 
 When writing worker scripts, make sure to reset any request-specific state between requests.
 Frameworks like [Symfony](symfony.md) and [Laravel Octane](laravel.md) take care of resetting most state for you, but you may still need to reset your own services. With Symfony, services that hold request-specific state should implement [`Symfony\Contracts\Service\ResetInterface`](https://github.com/symfony/contracts/blob/main/Service/ResetInterface.php) so they're reset by the kernel between requests.
+
+## Ticking
+
+Workers can also be triggered repeatedly with a message.
+
+```caddyfile
+worker /path/to/worker.php {
+    tick 10s "Hello Worker" # send "Hello Worker" every 10s
+}
+```
+
+In the worker script, the function passed to `frankenphp_handle_request()` will receive the message directly as an argument every 10s:
+
+```php
+while(frankenphp_handle_request(function(string $message = "") {
+    match($message){
+        'Hello Worker' => handleMessage()
+        default => handleRequest() # if the worker also handles regular HTTP requests
+    }
+})){}
+```
+
+The interval must be a [Go duration](https://pkg.go.dev/time#ParseDuration) such as `60s`, `1m`, or `5m`.
+Add the `aligned` keyword to align ticks to the start of each interval (e.g. `tick 1m aligned minutely` runs at the start of every minute). Available modes for ticking are: "sync", "overlap", "each" and "idle".
+
+```caddyfile
+worker /path/to/worker {
+    tick sync 10s "message" # send a single tick each 10s, wait for completion in-between ticks
+    tick overlap 10s "message" # send a single tick each 10s, don't wait for completion
+    tick each 10s "message" # send ticks to each active thread every 10s, don't wait for completion
+    tick idle 10s "message" # send ticks to each active thread that has been idle for more than 10-13.3s
+}
+```

frankenphp.go

@@ -359,6 +359,8 @@ func Init(options ...Option) error {
 		}
 	}
 
+	initTicks()
+
 	return nil
 }
 
@@ -380,6 +382,7 @@ func shutdown() {
 		fn()
 	}
 
+	shutdownTicks()
 	drainWatchers()
 	drainPHPThreads()
 	unregisterServers()
@@ -629,7 +632,7 @@ func go_sapi_flush(threadIndex C.uintptr_t) bool {
 func go_read_post(threadIndex C.uintptr_t, cBuf *C.char, countBytes C.size_t) (readBytes C.size_t) {
 	fc := phpThreads[threadIndex].handler.frankenPHPContext()
 
-	if fc.responseWriter == nil {
+	if fc.responseWriter == nil || fc.request == nil {
 		return 0
 	}
 

frankenphp_test.go

@@ -1425,3 +1425,27 @@ func testOpcachePreload(t *testing.T, opts *testOptions) {
 		assert.Equal(t, "I am preloaded", body)
 	}, opts)
 }
+
+func TestTicks(t *testing.T) {
+	logger, buf := newTestLogger(t)
+	require.NoError(t, frankenphp.Init(
+		frankenphp.WithLogger(logger),
+		frankenphp.WithWorkers("tick-worker", "testdata/worker-with-counter.php", 1,
+			frankenphp.WithWorkerTicks(frankenphp.TickModeSynchronous, 100*time.Microsecond, "tick", false),
+		),
+	))
+	t.Cleanup(frankenphp.Shutdown)
+
+	i := 0
+	for {
+		output := buf.String()
+		if strings.Contains(output, "requests:1") {
+			break
+		}
+		time.Sleep(500 * time.Microsecond)
+		i++
+		if i > 10000 { // 5s timeout
+			t.Fatal("timed out without recording a worker tick")
+		}
+	}
+}

options.go

@@ -50,6 +50,7 @@ type workerOpt struct {
 	requestOptions         []RequestOption
 	watch                  []string
 	matchRequest           func(*http.Request) bool
+	ticks                  []*tick
 	maxConsecutiveFailures int
 	extensionWorkers       *extensionWorkers
 	onThreadReady          func(int)
@@ -239,6 +240,20 @@ func WithWorkerServerScope(s *Server) WorkerOption {
 	}
 }
 
+// WithWorkerTicks configures a periodic message sent to the worker via frankenphp_handle_request().
+func WithWorkerTicks(mode TickMode, interval time.Duration, message string, aligned bool) WorkerOption {
+	return func(w *workerOpt) error {
+		w.ticks = append(w.ticks, &tick{
+			interval: interval,
+			message:  message,
+			aligned:  aligned,
+			mode:     mode,
+		})
+
+		return nil
+	}
+}
+
 // WithWorkerMaxFailures sets the maximum number of consecutive failures before panicking
 func WithWorkerMaxFailures(maxFailures int) WorkerOption {
 	return func(w *workerOpt) error {