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..e7eed7a 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, $clock), + ); + } + 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..d6735e2 --- /dev/null +++ b/src/Listener/StopWorkerOnRestartSignalListener.php @@ -0,0 +1,68 @@ +clock = $clock ?? new SystemClock(); + } + + public function onWorkerStarted(): void + { + $this->startTime = $this->clock->now()->getTimestamp(); + } + + 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..73052c2 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, $clock)], [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..3d57df5 --- /dev/null +++ b/tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php @@ -0,0 +1,151 @@ +file = sys_get_temp_dir() . '/' . uniqid('worker-restart-', true); + $this->clock = new TestClock(); + $this->startTime = $this->clock->now()->getTimestamp(); + } + + 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, clock: $this->clock); + $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, $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, clock: $this->clock); + $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]); + + 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(); + + // 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(), + ); + } +}