Fix #347: Install SignalLoop signal handlers only while consuming

CHANGELOG.md

@@ -3,3 +3,4 @@
 ## 1.0.0 under development
 
 - Initial release.
+- Bug #347: `SignalLoop` installs signal handlers only while messages are consumed and restores the previous ones afterwards, instead of replacing them when it is created (@samdark)

docs/guide/en/loops.md

@@ -27,6 +27,19 @@ It supports:
 - Pause/resume via `SIGTSTP` and `SIGCONT`.
 - Optional soft memory limit (see below).
 
+To stop a worker gracefully, send `SIGTERM` to its process, for example:
+
+```shell
+kill -TERM <worker-pid>
+```
+
+The worker finishes the current message and leaves the remaining messages in the queue for the next worker.
+Use `SIGTSTP` to pause processing and `SIGCONT` to resume it.
+
+You can create or resolve `SignalLoop` in another console command without changing that command's signal handling.
+Existing application handlers are restored when queue processing ends, including when processing fails.
+While a queue worker is running, its graceful-shutdown behavior takes precedence over application handlers.
+
 ### `SimpleLoop`
 
 `\Yiisoft\Queue\Cli\SimpleLoop` is used by default when `ext-pcntl` is **not** available.
@@ -91,6 +104,31 @@ use Yiisoft\Queue\Cli\SignalLoop;
 $loop = new SignalLoop(memorySoftLimit: 256 * 1024 * 1024);
 ```
 
+When creating `RunCommand` or `ListenAllCommand` manually, pass the same loop instance to the command and its consumers.
+This keeps graceful shutdown active between queues. For example, with your configured consumer provider:
+
+```php
+use Yiisoft\Queue\Command\RunCommand;
+
+$command = new RunCommand($consumerProvider, $loop);
+```
+
+For your own execution loop, use `start()` and `finish()` to restore application signal handlers when work ends:
+
+```php
+$loop->start();
+try {
+    while ($loop->canContinue()) {
+        // Process one unit of work.
+    }
+} finally {
+    $loop->finish();
+}
+```
+
+If you call `canContinue()` without `start()`, the loop keeps its signal handlers for the rest of the process.
+Use the paired calls above when other application code needs to handle signals after your work ends.
+
 ## Writing a custom loop
 
 Implement `LoopInterface` and encapsulate your own stopping conditions:
@@ -102,3 +140,8 @@ Implement `LoopInterface` and encapsulate your own stopping conditions:
 
 The only requirement is that `canContinue()` returns `false` when the worker should stop.
 
+Implement `ScopedLoopInterface` when your custom loop needs setup and cleanup around message processing.
+Put setup in `start()` and cleanup in `finish()`; consumers and commands call these methods for you,
+even if processing throws. Support nested calls when consumers share a loop with a command: keep the loop active
+until the matching outermost `finish()`.
+

src/Cli/ScopedLoopInterface.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Cli;
+
+/**
+ * A loop that changes process-wide state, such as signal handlers, only while messages are consumed.
+ *
+ * {@see ScopedLoopInterface::start()} is called before consuming starts and {@see ScopedLoopInterface::finish()} after it ends. Calls may be nested, for example
+ * when `queue:listen-all` runs consumers that share the same loop: the state is changed by the outermost
+ * {@see ScopedLoopInterface::start()} and restored by the matching {@see ScopedLoopInterface::finish()}.
+ */
+interface ScopedLoopInterface extends LoopInterface
+{
+    /**
+     * Prepares the loop before consuming messages.
+     */
+    public function start(): void;
+
+    /**
+     * Restores what {@see ScopedLoopInterface::start()} changed after consuming messages.
+     */
+    public function finish(): void;
+}

src/Cli/SignalLoop.php

@@ -10,7 +10,15 @@
 use const SIGTERM;
 use const SIGTSTP;
 
-final class SignalLoop implements LoopInterface
+/**
+ * A loop that stops on `SIGHUP`, `SIGINT` and `SIGTERM`, and pauses on `SIGTSTP` until `SIGCONT`.
+ *
+ * Pass the same loop to your consumers and commands to finish the current message before stopping.
+ * Creating a loop preserves existing application signal handlers. Consumers restore them after processing ends.
+ * For a custom execution loop, pair {@see SignalLoop::start()} with {@see SignalLoop::finish()} in a `finally` block.
+ * Calling {@see SignalLoop::canContinue()} without starting a scope keeps the handlers for the rest of the process.
+ */
+final class SignalLoop implements ScopedLoopInterface
 {
     use SoftLimitTrait;
 
@@ -20,20 +28,40 @@
     protected bool $pause = false;
     protected bool $exit = false;
 
+    /**
+     * @var array<int, mixed>|null Handlers that were active before ours were installed, by signal. `null` if our
+     * handlers are not installed.
+     */
+    private ?array $previousHandlers = null;
+    private int $depth = 0;
+    private bool $installedByStart = false;
+
     /**
      * @param int $memorySoftLimit Soft RAM limit in bytes. The loop won't let you continue to execute the program if
      * soft limit is reached. Zero means no limit.
      */
-    public function __construct(protected int $memorySoftLimit = 0)
+    public function __construct(protected int $memorySoftLimit = 0) {}
+
+    public function start(): void
     {
-        foreach (self::SIGNALS_EXIT as $signal) {
-            pcntl_signal($signal, fn() => $this->exit = true);
+        $this->depth++;
+
+        if ($this->depth === 1 && $this->previousHandlers === null) {
+            $this->installHandlers();
+            $this->installedByStart = true;
         }
-        foreach (self::SIGNALS_SUSPEND as $signal) {
-            pcntl_signal($signal, fn() => $this->pause = true);
+    }
+
+    public function finish(): void
+    {
+        if ($this->depth === 0) {
+            return;
         }
-        foreach (self::SIGNALS_RESUME as $signal) {
-            pcntl_signal($signal, fn() => $this->pause = false);
+
+        $this->depth--;
+
+        if ($this->depth === 0 && $this->installedByStart) {
+            $this->restoreHandlers();
         }
     }
 
@@ -53,6 +81,10 @@
 
     protected function dispatchSignals(): bool
     {
+        if ($this->previousHandlers === null) {
+            $this->installHandlers();
+        }
+
         pcntl_signal_dispatch();
 
         // Wait for resume signal until the loop is suspended
@@ -68,4 +100,44 @@
     {
         return $this->memorySoftLimit;
     }
+
+    private function installHandlers(): void
+    {
+        $handlers = [];
+        foreach (self::SIGNALS_EXIT as $signal) {
+            $handlers[$signal] = function (): void {
+                $this->exit = true;
+            };
+        }
+        foreach (self::SIGNALS_SUSPEND as $signal) {
+            $handlers[$signal] = function (): void {
+                $this->pause = true;
+            };
+        }
+        foreach (self::SIGNALS_RESUME as $signal) {
+            $handlers[$signal] = function (): void {
+                $this->pause = false;
+            };
+        }
+
+        $this->previousHandlers = [];
+        foreach ($handlers as $signal => $handler) {
+            $this->previousHandlers[$signal] = pcntl_signal_get_handler($signal);
+            pcntl_signal($signal, $handler);
+        }
+    }
+
+    private function restoreHandlers(): void
+    {
+        // Deliver signals received so far to our handlers, so that they are not lost.
+        pcntl_signal_dispatch();
+
+        foreach ($this->previousHandlers ?? [] as $signal => $handler) {
+            /** @psalm-var callable():void|int $handler */
+            pcntl_signal($signal, $handler);
+        }
+
+        $this->previousHandlers = null;
+        $this->installedByStart = false;
+    }
 }

src/Command/ListenAllCommand.php

@@ -11,6 +11,7 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Cli\ScopedLoopInterface;
 use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
 
 #[AsCommand(
@@ -86,15 +87,25 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $limit = (int) $input->getOption('limit');
 
-        while ($this->loop->canContinue()) {
-            $hasMessages = false;
-            foreach ($consumers as $consumer) {
-                $hasMessages = $consumer->run($limit) > 0 || $hasMessages;
-            }
+        if ($this->loop instanceof ScopedLoopInterface) {
+            $this->loop->start();
+        }
 
-            if (!$hasMessages && $this->loop->canContinue()) {
-                /** @psalm-var 0|positive-int $pauseSeconds */
-                sleep($pauseSeconds);
+        try {
+            while ($this->loop->canContinue()) {
+                $hasMessages = false;
+                foreach ($consumers as $consumer) {
+                    $hasMessages = $consumer->run($limit) > 0 || $hasMessages;
+                }
+
+                if (!$hasMessages && $this->loop->canContinue()) {
+                    /** @psalm-var 0|positive-int $pauseSeconds */
+                    sleep($pauseSeconds);
+                }
+            }
+        } finally {
+            if ($this->loop instanceof ScopedLoopInterface) {
+                $this->loop->finish();
             }
         }
 

src/Command/RunCommand.php

@@ -10,6 +10,8 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Cli\ScopedLoopInterface;
 use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
 
 #[AsCommand(
@@ -20,6 +22,7 @@ final class RunCommand extends Command
 {
     public function __construct(
         private readonly QueueConsumerProviderInterface $queueProvider,
+        private readonly ?LoopInterface $loop = null,
     ) {
         parent::__construct();
     }
@@ -50,14 +53,27 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             $queueNames = $this->queueProvider->getConsumerQueueNames();
         }
 
-        /** @var string $queueName */
-        foreach ($queueNames as $queueName) {
-            $queueConsumer = $this->queueProvider->getConsumer($queueName);
+        $limit = (int) $input->getOption('limit');
 
-            $output->write("Processing queue $queueName... ");
-            $count = $queueConsumer->run((int) $input->getOption('limit'));
+        // One scope for all queues, so that a signal received between two of them doesn't reach the default handler.
+        if ($this->loop instanceof ScopedLoopInterface) {
+            $this->loop->start();
+        }
+
+        try {
+            /** @var string $queueName */
+            foreach ($queueNames as $queueName) {
+                $queueConsumer = $this->queueProvider->getConsumer($queueName);
+
+                $output->write("Processing queue $queueName... ");
+                $count = $queueConsumer->run($limit);
 
-            $output->writeln("Messages processed: $count.");
+                $output->writeln("Messages processed: $count.");
+            }
+        } finally {
+            if ($this->loop instanceof ScopedLoopInterface) {
+                $this->loop->finish();
+            }
         }
 
         return Command::SUCCESS;

src/QueueConsumer.php

@@ -8,6 +8,7 @@
 use Psr\Log\LoggerInterface;
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Cli\ScopedLoopInterface;
 use Yiisoft\Queue\Message\MessageInterface;
 use Yiisoft\Queue\Worker\WorkerInterface;
 
@@ -34,13 +35,18 @@ public function run(int $max = 0): int
         }
         $this->logger->debug('Start processing queue messages.');
         $count = 0;
-        $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool {
-            if (($max > 0 && $count >= $max) || !$this->handle($message)) {
-                return false;
-            }
-            $count++;
-            return true;
-        });
+        $this->startLoop();
+        try {
+            $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool {
+                if (($max > 0 && $count >= $max) || !$this->handle($message)) {
+                    return false;
+                }
+                $count++;
+                return true;
+            });
+        } finally {
+            $this->finishLoop();
+        }
         $this->logger->info('Processed {count} queue messages.', ['count' => $count]);
         return $count;
     }
@@ -52,10 +58,29 @@ public function listen(): void
             return;
         }
         $this->logger->info('Start listening to the queue.');
-        $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message));
+        $this->startLoop();
+        try {
+            $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message));
+        } finally {
+            $this->finishLoop();
+        }
         $this->logger->info('Finish listening to the queue.');
     }
 
+    private function startLoop(): void
+    {
+        if ($this->loop instanceof ScopedLoopInterface) {
+            $this->loop->start();
+        }
+    }
+
+    private function finishLoop(): void
+    {
+        if ($this->loop instanceof ScopedLoopInterface) {
+            $this->loop->finish();
+        }
+    }
+
     private function handle(MessageInterface $message): bool
     {
         if (!$this->loop->canContinue()) {