From 896765c16e792edb0442998dea8af07e0f83d429 Mon Sep 17 00:00:00 2001 From: David Badura Date: Sun, 4 Oct 2026 09:36:32 +0200 Subject: [PATCH 1/2] Add restart signal listener to restart workers after a deployment StopWorkerOnRestartSignalListener stops the worker once a given file is touched after the worker has started. Touching the file during a deployment makes all running workers restart with the new code. It can be enabled with the new restartSignalFile option in create(). --- docs/events.md | 8 +- docs/getting-started.md | 38 ++++++-- src/DefaultWorker.php | 11 ++- .../StopWorkerOnRestartSignalListener.php | 63 ++++++++++++++ tests/Unit/DefaultWorkerTest.php | 5 +- .../StopWorkerOnRestartSignalListenerTest.php | 86 +++++++++++++++++++ 6 files changed, 198 insertions(+), 13 deletions(-) create mode 100644 src/Listener/StopWorkerOnRestartSignalListener.php create mode 100644 tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php diff --git a/docs/events.md b/docs/events.md index 44d289c..17a10a8 100644 --- a/docs/events.md +++ b/docs/events.md @@ -27,18 +27,18 @@ $eventDispatcher->addListener( ``` All [limits](getting-started.md#limits) are implemented as event subscribers (`StopWorkerOnIterationLimitListener`, `StopWorkerOnMemoryLimitListener`, -`StopWorkerOnTimeLimitListener`, `StopWorkerOnSignalListener`), +`StopWorkerOnTimeLimitListener`, `StopWorkerOnSignalListener`, `StopWorkerOnRestartSignalListener`), so you can add your own stop conditions the same way: ```php use Patchlevel\Worker\Event\WorkerRunningEvent; use Symfony\Component\EventDispatcher\EventSubscriberInterface; -final class StopWorkerOnNewDeploymentListener implements EventSubscriberInterface +final class StopWorkerOnMaintenanceModeListener implements EventSubscriberInterface { public function onWorkerRunning(WorkerRunningEvent $event): void { - if (new_version_deployed()) { + if (maintenance_mode_enabled()) { $event->worker->stop(); } } @@ -56,7 +56,7 @@ use Patchlevel\Worker\DefaultWorker; use Symfony\Component\EventDispatcher\EventDispatcher; $eventDispatcher = new EventDispatcher(); -$eventDispatcher->addSubscriber(new StopWorkerOnNewDeploymentListener()); +$eventDispatcher->addSubscriber(new StopWorkerOnMaintenanceModeListener()); $worker = DefaultWorker::create( $job, diff --git a/docs/getting-started.md b/docs/getting-started.md index b319e7f..4e88e1c 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -37,12 +37,13 @@ and rethrows the exception, so the process exits with an error. All options are optional. Without limits the worker runs until it is stopped via `$stop()`, `$worker->stop()` or a SIGTERM/SIGINT signal. -| Option | Type | Description | -|-----------------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------| -| `runLimit` | `int` | Stop after this number of iterations. | -| `memoryLimit` | `string` | Stop when memory usage exceeds this value, e.g. `128MB` or `128M`. Supported units: `B`, `K`/`KB`, `M`/`MB`, `G`/`GB` (case-insensitive, 1024-based). | -| `timeLimit` | `int` | Stop after this number of seconds. | -| `heartbeatFile` | `string` | Touch this file on start and after every iteration, see [heartbeat](#heartbeat). | +| Option | Type | Description | +|---------------------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------| +| `runLimit` | `int` | Stop after this number of iterations. | +| `memoryLimit` | `string` | Stop when memory usage exceeds this value, e.g. `128MB` or `128M`. Supported units: `B`, `K`/`KB`, `M`/`MB`, `G`/`GB` (case-insensitive, 1024-based). | +| `timeLimit` | `int` | Stop after this number of seconds. | +| `restartSignalFile` | `string` | Stop when this file is touched after the worker has started, see [restart after a deployment](#restart-after-a-deployment). | +| `heartbeatFile` | `string` | Touch this file on start and after every iteration, see [heartbeat](#heartbeat). | Limits are checked after each iteration. When a limit is exceeded, the worker logs the reason and stops gracefully. @@ -71,6 +72,31 @@ Without `ext-pcntl` this feature is not available. If you need to react to other signals, register the `StopWorkerOnSignalListener` with your own list of signals on a custom event dispatcher, see [events & listeners](events.md). +## Restart after a deployment + +Long-running workers keep the code they were started with. To pick up a new release, +set the `restartSignalFile` option and touch that file during the deployment. +Every worker that was started before the file was touched stops after its current iteration, +and the process manager starts it again with the new code: + +```php +use Patchlevel\Worker\DefaultWorker; + +$worker = DefaultWorker::create( + $job, + ['restartSignalFile' => 'var/worker-restart'], + $logger, +); +``` +```bash +touch var/worker-restart +``` +:::note +The file is checked after every iteration and only its modification time matters, +so the file can stay in place and be touched again on the next deployment. +All workers that should restart have to see the same file, e.g. on a shared volume. +::: + ## Sleep `run()` takes a sleep timer in milliseconds (default: `1000`): diff --git a/src/DefaultWorker.php b/src/DefaultWorker.php index 44e565e..8cfddc4 100644 --- a/src/DefaultWorker.php +++ b/src/DefaultWorker.php @@ -11,6 +11,7 @@ use Patchlevel\Worker\Listener\HeartbeatListener; use Patchlevel\Worker\Listener\StopWorkerOnIterationLimitListener; use Patchlevel\Worker\Listener\StopWorkerOnMemoryLimitListener; +use Patchlevel\Worker\Listener\StopWorkerOnRestartSignalListener; use Patchlevel\Worker\Listener\StopWorkerOnSignalListener; use Patchlevel\Worker\Listener\StopWorkerOnTimeLimitListener; use Psr\Clock\ClockInterface; @@ -100,8 +101,8 @@ public function stop(): void } /** - * @param Closure(Closure):(bool|void) $job - * @param array{runLimit?: (positive-int|null), memoryLimit?: (string|null), timeLimit?: (positive-int|null), heartbeatFile?: (string|null)} $options + * @param Closure(Closure):(bool|void) $job + * @param array{runLimit?: (positive-int|null), memoryLimit?: (string|null), timeLimit?: (positive-int|null), restartSignalFile?: (string|null), heartbeatFile?: (string|null)} $options */ public static function create( Closure $job, @@ -134,6 +135,12 @@ public static function create( ); } + if (isset($options['restartSignalFile'])) { + $eventDispatcher->addSubscriber( + new StopWorkerOnRestartSignalListener($options['restartSignalFile'], $logger), + ); + } + if (isset($options['heartbeatFile'])) { $eventDispatcher->addSubscriber( new HeartbeatListener($options['heartbeatFile'], $logger), diff --git a/src/Listener/StopWorkerOnRestartSignalListener.php b/src/Listener/StopWorkerOnRestartSignalListener.php new file mode 100644 index 0000000..3e78d07 --- /dev/null +++ b/src/Listener/StopWorkerOnRestartSignalListener.php @@ -0,0 +1,63 @@ +startTime = time(); + } + + public function onWorkerRunning(WorkerRunningEvent $event): void + { + clearstatcache(true, $this->file); + + if (!is_file($this->file)) { + return; + } + + $modifiedTime = filemtime($this->file); + + if ($modifiedTime === false || $modifiedTime <= $this->startTime) { + return; + } + + $this->logger?->info( + 'Worker stopped due to restart signal from {file}', + ['file' => $this->file], + ); + + $event->worker->stop(); + } + + /** @return array */ + public static function getSubscribedEvents(): array + { + return [ + WorkerStartedEvent::class => 'onWorkerStarted', + WorkerRunningEvent::class => 'onWorkerRunning', + ]; + } +} diff --git a/tests/Unit/DefaultWorkerTest.php b/tests/Unit/DefaultWorkerTest.php index a3c4498..5e40c66 100644 --- a/tests/Unit/DefaultWorkerTest.php +++ b/tests/Unit/DefaultWorkerTest.php @@ -12,6 +12,7 @@ use Patchlevel\Worker\Listener\HeartbeatListener; use Patchlevel\Worker\Listener\StopWorkerOnIterationLimitListener; use Patchlevel\Worker\Listener\StopWorkerOnMemoryLimitListener; +use Patchlevel\Worker\Listener\StopWorkerOnRestartSignalListener; use Patchlevel\Worker\Listener\StopWorkerOnSignalListener; use Patchlevel\Worker\Listener\StopWorkerOnTimeLimitListener; use Patchlevel\Worker\Tests\ReturnCallback; @@ -147,12 +148,13 @@ public function testOptions(): void $clock = new TestClock(); - $invokationCount = $this->exactly(5); + $invokationCount = $this->exactly(6); $invokationParameters = [ [new StopWorkerOnSignalListener(logger: $logger)], [new StopWorkerOnIterationLimitListener(10, $logger)], [new StopWorkerOnMemoryLimitListener(Bytes::parseFromString('10KB'), $logger)], [new StopWorkerOnTimeLimitListener(20, $logger, $clock)], + [new StopWorkerOnRestartSignalListener('var/worker-restart', $logger)], [new HeartbeatListener('/tmp/worker-heartbeat', $logger)], ]; @@ -172,6 +174,7 @@ static function ($stop): void { 'runLimit' => 10, 'memoryLimit' => '10KB', 'timeLimit' => 20, + 'restartSignalFile' => 'var/worker-restart', 'heartbeatFile' => '/tmp/worker-heartbeat', ], $logger, diff --git a/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php b/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php new file mode 100644 index 0000000..2ca9dbe --- /dev/null +++ b/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php @@ -0,0 +1,86 @@ +file = sys_get_temp_dir() . '/' . uniqid('worker-restart-', true); + } + + protected function tearDown(): void + { + if (!is_file($this->file)) { + return; + } + + unlink($this->file); + } + + public function testShouldNotStopWithoutFile(): void + { + $worker = $this->createMock(Worker::class); + $worker + ->expects($this->never()) + ->method('stop'); + + $listener = new StopWorkerOnRestartSignalListener($this->file); + $listener->onWorkerStarted(); + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + } + + public function testShouldNotStopIfFileWasTouchedBeforeStart(): void + { + $worker = $this->createMock(Worker::class); + $worker + ->expects($this->never()) + ->method('stop'); + + touch($this->file, time() - 10); + + $listener = new StopWorkerOnRestartSignalListener($this->file); + $listener->onWorkerStarted(); + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + } + + public function testShouldStopIfFileWasTouchedAfterStart(): void + { + $worker = $this->createMock(Worker::class); + $worker + ->expects($this->once()) + ->method('stop'); + + $logger = $this->createMock(LoggerInterface::class); + $logger + ->expects($this->once()) + ->method('info') + ->with('Worker stopped due to restart signal from {file}', ['file' => $this->file]); + + $listener = new StopWorkerOnRestartSignalListener($this->file, $logger); + $listener->onWorkerStarted(); + + touch($this->file, time() + 10); + + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + } +} From 56e81fb3a0f2128354dc33ba6e1e7b4d12104793 Mon Sep 17 00:00:00 2001 From: David Badura Date: Mon, 5 Oct 2026 09:21:46 +0200 Subject: [PATCH 2/2] Use clock in restart signal listener and cover escaped mutants The start time now comes from the PSR-20 clock, which makes the check for a file touched in the same second as the start deterministic in tests. The stat cache test changes the file from another process, since touching it from within the test process clears the cache anyway. --- src/DefaultWorker.php | 2 +- .../StopWorkerOnRestartSignalListener.php | 9 ++- tests/Unit/DefaultWorkerTest.php | 2 +- .../StopWorkerOnRestartSignalListenerTest.php | 77 +++++++++++++++++-- 4 files changed, 80 insertions(+), 10 deletions(-) diff --git a/src/DefaultWorker.php b/src/DefaultWorker.php index 8cfddc4..e7eed7a 100644 --- a/src/DefaultWorker.php +++ b/src/DefaultWorker.php @@ -137,7 +137,7 @@ public static function create( if (isset($options['restartSignalFile'])) { $eventDispatcher->addSubscriber( - new StopWorkerOnRestartSignalListener($options['restartSignalFile'], $logger), + new StopWorkerOnRestartSignalListener($options['restartSignalFile'], $logger, $clock), ); } diff --git a/src/Listener/StopWorkerOnRestartSignalListener.php b/src/Listener/StopWorkerOnRestartSignalListener.php index 3e78d07..d6735e2 100644 --- a/src/Listener/StopWorkerOnRestartSignalListener.php +++ b/src/Listener/StopWorkerOnRestartSignalListener.php @@ -6,28 +6,33 @@ use Patchlevel\Worker\Event\WorkerRunningEvent; use Patchlevel\Worker\Event\WorkerStartedEvent; +use Patchlevel\Worker\SystemClock; +use Psr\Clock\ClockInterface; use Psr\Log\LoggerInterface; use Symfony\Component\EventDispatcher\EventSubscriberInterface; use function clearstatcache; use function filemtime; use function is_file; -use function time; final class StopWorkerOnRestartSignalListener implements EventSubscriberInterface { private int $startTime = 0; + private readonly ClockInterface $clock; + /** @param string $file the worker stops if this file is touched after the worker has started */ public function __construct( private readonly string $file, private readonly LoggerInterface|null $logger = null, + ClockInterface|null $clock = null, ) { + $this->clock = $clock ?? new SystemClock(); } public function onWorkerStarted(): void { - $this->startTime = time(); + $this->startTime = $this->clock->now()->getTimestamp(); } public function onWorkerRunning(WorkerRunningEvent $event): void diff --git a/tests/Unit/DefaultWorkerTest.php b/tests/Unit/DefaultWorkerTest.php index 5e40c66..73052c2 100644 --- a/tests/Unit/DefaultWorkerTest.php +++ b/tests/Unit/DefaultWorkerTest.php @@ -154,7 +154,7 @@ public function testOptions(): void [new StopWorkerOnIterationLimitListener(10, $logger)], [new StopWorkerOnMemoryLimitListener(Bytes::parseFromString('10KB'), $logger)], [new StopWorkerOnTimeLimitListener(20, $logger, $clock)], - [new StopWorkerOnRestartSignalListener('var/worker-restart', $logger)], + [new StopWorkerOnRestartSignalListener('var/worker-restart', $logger, $clock)], [new HeartbeatListener('/tmp/worker-heartbeat', $logger)], ]; diff --git a/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php b/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php index 2ca9dbe..3d57df5 100644 --- a/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php +++ b/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php @@ -5,27 +5,40 @@ namespace Patchlevel\Worker\Tests\Unit\Listener; use Patchlevel\Worker\Event\WorkerRunningEvent; +use Patchlevel\Worker\Event\WorkerStartedEvent; use Patchlevel\Worker\Listener\StopWorkerOnRestartSignalListener; +use Patchlevel\Worker\Tests\TestClock; use Patchlevel\Worker\Worker; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; use Psr\Log\LoggerInterface; use function is_file; +use function proc_close; +use function proc_open; +use function sprintf; use function sys_get_temp_dir; -use function time; use function touch; use function uniqid; use function unlink; +use function var_export; + +use const PHP_BINARY; #[CoversClass(StopWorkerOnRestartSignalListener::class)] final class StopWorkerOnRestartSignalListenerTest extends TestCase { private string $file; + private TestClock $clock; + + private int $startTime; + protected function setUp(): void { $this->file = sys_get_temp_dir() . '/' . uniqid('worker-restart-', true); + $this->clock = new TestClock(); + $this->startTime = $this->clock->now()->getTimestamp(); } protected function tearDown(): void @@ -44,7 +57,7 @@ public function testShouldNotStopWithoutFile(): void ->expects($this->never()) ->method('stop'); - $listener = new StopWorkerOnRestartSignalListener($this->file); + $listener = new StopWorkerOnRestartSignalListener($this->file, clock: $this->clock); $listener->onWorkerStarted(); $listener->onWorkerRunning(new WorkerRunningEvent($worker)); } @@ -56,9 +69,23 @@ public function testShouldNotStopIfFileWasTouchedBeforeStart(): void ->expects($this->never()) ->method('stop'); - touch($this->file, time() - 10); + touch($this->file, $this->startTime - 10); + + $listener = new StopWorkerOnRestartSignalListener($this->file, clock: $this->clock); + $listener->onWorkerStarted(); + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + } + + public function testShouldNotStopIfFileWasTouchedAtStart(): void + { + $worker = $this->createMock(Worker::class); + $worker + ->expects($this->never()) + ->method('stop'); + + touch($this->file, $this->startTime); - $listener = new StopWorkerOnRestartSignalListener($this->file); + $listener = new StopWorkerOnRestartSignalListener($this->file, clock: $this->clock); $listener->onWorkerStarted(); $listener->onWorkerRunning(new WorkerRunningEvent($worker)); } @@ -76,11 +103,49 @@ public function testShouldStopIfFileWasTouchedAfterStart(): void ->method('info') ->with('Worker stopped due to restart signal from {file}', ['file' => $this->file]); - $listener = new StopWorkerOnRestartSignalListener($this->file, $logger); + touch($this->file, $this->startTime + 10); + + $listener = new StopWorkerOnRestartSignalListener($this->file, $logger, $this->clock); + $listener->onWorkerStarted(); + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + } + + public function testShouldStopIfFileWasTouchedByAnotherProcess(): void + { + $worker = $this->createMock(Worker::class); + $worker + ->expects($this->once()) + ->method('stop'); + + touch($this->file, $this->startTime - 10); + + $listener = new StopWorkerOnRestartSignalListener($this->file, clock: $this->clock); $listener->onWorkerStarted(); - touch($this->file, time() + 10); + // A deployment touches the file from another process, which doesn't clear the stat cache + // of this process. The delay makes sure the first check below caches the old modification time. + $process = proc_open( + [PHP_BINARY, '-r', sprintf('usleep(200000); touch(%s, %d);', var_export($this->file, true), $this->startTime + 10)], + [], + $pipes, + ); + self::assertIsResource($process); + + $listener->onWorkerRunning(new WorkerRunningEvent($worker)); + + proc_close($process); $listener->onWorkerRunning(new WorkerRunningEvent($worker)); } + + public function testSubscribedEvents(): void + { + self::assertSame( + [ + WorkerStartedEvent::class => 'onWorkerStarted', + WorkerRunningEvent::class => 'onWorkerRunning', + ], + StopWorkerOnRestartSignalListener::getSubscribedEvents(), + ); + } }