Add restart signal listener to restart workers after a deployment

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,

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`):

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;
@@ -40,7 +41,7 @@
     }
 
     /** @param positive-int|0 $sleepTimer in milliseconds */
     public function run(int $sleepTimer = 1000): void
     {
         $this->logger?->debug('Worker starting');
 
@@ -64,21 +65,21 @@
                 $this->eventDispatcher->dispatch(new WorkerRunningEvent($this));
 
                 if ($this->shouldStop) {
                     break;
                 }
 
                 if ($didWork) {
                     continue;
                 }
 
                 $sleepFor = max($sleepTimer - $ranTime, 0);
 
                 if ($sleepFor <= 0) {
                     continue;
                 }
 
                 $this->logger?->debug('Worker sleep for {sleepTimer}ms', ['sleepTimer' => $sleepFor]);
                 usleep($sleepFor * 1000);
             }
         } catch (Throwable $exception) {
             throw $exception;
@@ -100,8 +101,8 @@
     }
 
     /**
-     * @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 @@
             );
         }
 
+        if (isset($options['restartSignalFile'])) {
+            $eventDispatcher->addSubscriber(
+                new StopWorkerOnRestartSignalListener($options['restartSignalFile'], $logger, $clock),
+            );
+        }
+
         if (isset($options['heartbeatFile'])) {
             $eventDispatcher->addSubscriber(
                 new HeartbeatListener($options['heartbeatFile'], $logger),

src/Listener/StopWorkerOnRestartSignalListener.php

@@ -0,0 +1,68 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\Worker\Listener;
+
+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;
+
+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 = $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<class-string, string> */
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerStartedEvent::class => 'onWorkerStarted',
+            WorkerRunningEvent::class => 'onWorkerRunning',
+        ];
+    }
+}

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,

tests/Unit/Listener/StopWorkerOnRestartSignalListenerTest.php

@@ -0,0 +1,151 @@
+<?php
+
+declare(strict_types=1);
+
+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 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
+    {
+        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(),
+        );
+    }
+}