feat: add workerStop handler, graceful exits

docker-compose.yml

@@ -3,8 +3,9 @@ services:
     container_name: tests
     build: .
     volumes:
-      - ./src:/usr/local/src/src
-      - ./tests:/usr/local/src/tests
+      - ./vendor:/usr/src/code/vendor
+      - ./src:/usr/src/code/src
+      - ./tests:/usr/src/code/tests
     depends_on:
       - swoole
       - swoole-amqp
@@ -16,8 +17,9 @@ services:
     build: ./tests/Queue/servers/Swoole/.
     command: php /usr/src/code/tests/Queue/servers/Swoole/worker.php
     volumes:
-      - ./src:/usr/local/src/src
-      - ./tests:/usr/local/src/tests
+      - ./vendor:/usr/src/code/vendor
+      - ./src:/usr/src/code/src
+      - ./tests:/usr/src/code/tests
     depends_on:
       - redis
 
@@ -26,8 +28,9 @@ services:
     build: ./tests/Queue/servers/SwooleRedisCluster/.
     command: php /usr/src/code/tests/Queue/servers/SwooleRedisCluster/worker.php
     volumes:
-      - ./src:/usr/local/src/src
-      - ./tests:/usr/local/src/tests
+      - ./vendor:/usr/src/code/vendor
+      - ./src:/usr/src/code/src
+      - ./tests:/usr/src/code/tests
     depends_on:
       redis-cluster-0:
         condition: service_healthy
@@ -37,8 +40,9 @@ services:
     build: ./tests/Queue/servers/AMQP/.
     command: php /usr/src/code/tests/Queue/servers/AMQP/worker.php
     volumes:
-      - ./src:/usr/local/src/src
-      - ./tests:/usr/local/src/tests
+      - ./vendor:/usr/src/code/vendor
+      - ./src:/usr/src/code/src
+      - ./tests:/usr/src/code/tests
     depends_on:
       amqp:
         condition: service_healthy
@@ -48,8 +52,9 @@ services:
     build: ./tests/Queue/servers/Workerman/.
     command: php /usr/src/code/tests/Queue/servers/Workerman/worker.php start
     volumes:
-      - ./src:/usr/local/src/src
-      - ./tests:/usr/local/src/tests
+      - ./vendor:/usr/src/code/vendor
+      - ./src:/usr/src/code/src
+      - ./tests:/usr/src/code/tests
     depends_on:
       - redis
 

src/Queue/Adapter/Swoole.php

@@ -2,14 +2,19 @@
 
 namespace Utopia\Queue\Adapter;
 
+use Swoole\Constant;
 use Swoole\Process\Pool;
+use Utopia\CLI\Console;
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
 
 class Swoole extends Adapter
 {
     protected Pool $pool;
 
+    /** @var callable */
+    private $onStop;
+
     public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
     {
         parent::__construct($workerNum, $queue, $namespace);
@@ -21,19 +26,59 @@ public function __construct(Consumer $consumer, int $workerNum, string $queue, s
     public function start(): self
     {
         $this->pool->set(['enable_coroutine' => true]);
+
+        // Register signal handlers in the main process before starting pool
+        if (extension_loaded('pcntl')) {
+            pcntl_signal(SIGTERM, function () {
+                Console::info("[Swoole] Received SIGTERM, initiating graceful shutdown...");
+                $this->stop();
+            });
+
+            pcntl_signal(SIGINT, function () {
+                Console::info("[Swoole] Received SIGINT, initiating graceful shutdown...");
+                $this->stop();
+            });
+
+            // Enable async signals
+            pcntl_async_signals(true);
+        } else {
+            Console::warning("[Swoole] pcntl extension is not loaded, worker will not shutdown gracefully.");
+        }
+
         $this->pool->start();
         return $this;
     }
 
     public function stop(): self
     {
+        if ($this->onStop) {
+            call_user_func($this->onStop);
+        }
+
+        Console::info("[Swoole] Shutting down process pool...");
         $this->pool->shutdown();
+        Console::success("[Swoole] Process pool stopped.");
         return $this;
     }
 
     public function workerStart(callable $callback): self
     {
-        $this->pool->on('WorkerStart', function (Pool $pool, string $workerId) use ($callback) {
+        $this->pool->on(Constant::EVENT_WORKER_START, function (Pool $pool, string $workerId) use ($callback) {
+            // Register signal handlers in each worker process for graceful shutdown
+            if (extension_loaded('pcntl')) {
+                pcntl_signal(SIGTERM, function () use ($workerId) {
+                    Console::info("[Worker] Worker {$workerId} received SIGTERM, closing consumer...");
+                    $this->consumer->close();
+                });
+
+                pcntl_signal(SIGINT, function () use ($workerId) {
+                    Console::info("[Worker] Worker {$workerId} received SIGINT, closing consumer...");
+                    $this->consumer->close();
+                });
+
+                pcntl_async_signals(true);
+            }
+
             call_user_func($callback, $workerId);
         });
 
@@ -42,7 +87,8 @@ public function workerStart(callable $callback): self
 
     public function workerStop(callable $callback): self
     {
-        $this->pool->on('WorkerStart', function (Pool $pool, string $workerId) use ($callback) {
+        $this->onStop = $callback;
+        $this->pool->on(Constant::EVENT_WORKER_STOP, function (Pool $pool, string $workerId) use ($callback) {
             call_user_func($callback, $workerId);
         });
 

src/Queue/Broker/AMQP.php

@@ -131,6 +131,7 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
 
     public function close(): void
     {
+        $this->channel?->stopConsume();
         $this->channel?->getConnection()?->close();
     }
 

src/Queue/Server.php

@@ -53,6 +53,11 @@ class Server
      */
     protected Hook $workerStartHook;
 
+    /**
+     * Hook that is called when worker stops
+     */
+    protected Hook $workerStopHook;
+
     /**
      * @var array
      */
@@ -96,7 +101,7 @@ public function getResource(string $name, bool $fresh = false): mixed
     {
         if (!\array_key_exists($name, $this->resources) || $fresh || self::$resourcesCallbacks[$name]['reset']) {
             if (!\array_key_exists($name, self::$resourcesCallbacks)) {
-                throw new Exception('Failed to find resource: "' . $name . '"');
+                throw new Exception("Failed to find resource: $name");
             }
 
             $this->resources[$name] = \call_user_func_array(
@@ -213,49 +218,23 @@ public function start(): self
             $this->adapter->workerStart(function (string $workerId) {
                 Console::success("[Worker] Worker {$workerId} is ready!");
                 self::setResource('workerId', fn () => $workerId);
-                if (!is_null($this->workerStartHook)) {
+                if ($this->workerStartHook !== null) {
                     call_user_func_array($this->workerStartHook->getAction(), $this->getArguments($this->workerStartHook));
                 }
 
-                while (true) {
-                    $this->adapter->consumer->consume(
-                        $this->adapter->queue,
-                        function (Message $message) {
-                            $receivedAtTimestamp = microtime(true);
-                            Console::info("[Job] Received Job ({$message->getPid()}).");
-                            try {
-                                $waitDuration = microtime(true) - $message->getTimestamp();
-                                $this->jobWaitTime->record($waitDuration);
-
-                                $this->resources = [];
-                                self::setResource('message', fn () => $message);
-                                if ($this->job->getHook()) {
-                                    foreach ($this->initHooks as $hook) { // Global init hooks
-                                        if (in_array('*', $hook->getGroups())) {
-                                            $arguments = $this->getArguments($hook, $message->getPayload());
-                                            \call_user_func_array($hook->getAction(), $arguments);
-                                        }
-                                    }
-                                }
-
-                                foreach ($this->job->getGroups() as $group) {
-                                    foreach ($this->initHooks as $hook) { // Group init hooks
-                                        if (in_array($group, $hook->getGroups())) {
-                                            $arguments = $this->getArguments($hook, $message->getPayload());
-                                            \call_user_func_array($hook->getAction(), $arguments);
-                                        }
-                                    }
-                                }
-
-                                return \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
-                            } finally {
-                                $processDuration = microtime(true) - $receivedAtTimestamp;
-                                $this->processDuration->record($processDuration);
-                            }
-                        },
-                        function (Message $message) {
+                $this->adapter->consumer->consume(
+                    $this->adapter->queue,
+                    function (Message $message) {
+                        $receivedAtTimestamp = microtime(true);
+                        Console::info("[Job] Received Job ({$message->getPid()}).");
+                        try {
+                            $waitDuration = microtime(true) - $message->getTimestamp();
+                            $this->jobWaitTime->record($waitDuration);
+
+                            $this->resources = [];
+                            self::setResource('message', fn () => $message);
                             if ($this->job->getHook()) {
-                                foreach ($this->shutdownHooks as $hook) { // Global init hooks
+                                foreach ($this->initHooks as $hook) { // Global init hooks
                                     if (in_array('*', $hook->getGroups())) {
                                         $arguments = $this->getArguments($hook, $message->getPayload());
                                         \call_user_func_array($hook->getAction(), $arguments);
@@ -264,27 +243,65 @@ function (Message $message) {
                             }
 
                             foreach ($this->job->getGroups() as $group) {
-                                foreach ($this->shutdownHooks as $hook) { // Group init hooks
+                                foreach ($this->initHooks as $hook) { // Group init hooks
                                     if (in_array($group, $hook->getGroups())) {
                                         $arguments = $this->getArguments($hook, $message->getPayload());
                                         \call_user_func_array($hook->getAction(), $arguments);
                                     }
                                 }
                             }
-                            Console::success("[Job] ({$message->getPid()}) successfully run.");
-                        },
-                        function (?Message $message, Throwable $th) {
-                            Console::error("[Job] ({$message?->getPid()}) failed to run.");
-                            Console::error("[Job] ({$message?->getPid()}) {$th->getMessage()}");
 
-                            self::setResource('error', fn () => $th);
+                            return \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
+                        } finally {
+                            $processDuration = microtime(true) - $receivedAtTimestamp;
+                            $this->processDuration->record($processDuration);
+                        }
+                    },
+                    function (Message $message) {
+                        if ($this->job->getHook()) {
+                            foreach ($this->shutdownHooks as $hook) { // Global init hooks
+                                if (in_array('*', $hook->getGroups())) {
+                                    $arguments = $this->getArguments($hook, $message->getPayload());
+                                    \call_user_func_array($hook->getAction(), $arguments);
+                                }
+                            }
+                        }
 
-                            foreach ($this->errorHooks as $hook) {
-                                ($hook->getAction())(...$this->getArguments($hook));
+                        foreach ($this->job->getGroups() as $group) {
+                            foreach ($this->shutdownHooks as $hook) { // Group init hooks
+                                if (in_array($group, $hook->getGroups())) {
+                                    $arguments = $this->getArguments($hook, $message->getPayload());
+                                    \call_user_func_array($hook->getAction(), $arguments);
+                                }
                             }
-                        },
-                    );
+                        }
+                        Console::success("[Job] ({$message->getPid()}) successfully run.");
+                    },
+                    function (?Message $message, Throwable $th) {
+                        Console::error("[Job] ({$message?->getPid()}) failed to run.");
+                        Console::error("[Job] ({$message?->getPid()}) {$th->getMessage()}");
+
+                        self::setResource('error', fn () => $th);
+
+                        foreach ($this->errorHooks as $hook) {
+                            $hook->getAction()(...$this->getArguments($hook));
+                        }
+                    },
+                );
+            });
+
+            $this->adapter->workerStop(function ($workerId) {
+                Console::info("[Worker] Worker {$workerId} stopping...");
+                self::setResource('workerId', fn () => $workerId);
+
+                // Call user-defined workerStop hook if set
+                if ($this->workerStopHook !== null) {
+                    call_user_func_array($this->workerStopHook->getAction(), $this->getArguments($this->workerStopHook));
                 }
+
+                // Close consumer connection
+                $this->adapter->consumer->close();
+                Console::success("[Worker] Worker {$workerId} stopped gracefully.");
             });
 
             $this->adapter->start();
@@ -320,27 +337,23 @@ public function getWorkerStart(): Hook
 
     /**
      * Is called when a Worker stops.
-     * @param callable|null $callback
-     * @return self
-     * @throws Exception
+     * @return Hook
      */
-    public function workerStop(?callable $callback = null): self
+    public function workerStop(): Hook
     {
-        try {
-            $this->adapter->workerStop(function (string $workerId) use ($callback) {
-                Console::success("[Worker] Worker {$workerId} is ready!");
-                if (!is_null($callback)) {
-                    call_user_func($callback);
-                }
-            });
-        } catch (Throwable $error) {
-            self::setResource('error', fn () => $error);
-            foreach ($this->errorHooks as $hook) {
-                call_user_func_array($hook->getAction(), $this->getArguments($hook));
-            }
-        }
+        $hook = new Hook();
+        $hook->groups(['*']);
+        $this->workerStopHook = $hook;
+        return $hook;
+    }
 
-        return $this;
+    /**
+     * Returns Worker stops hook.
+     * @return ?Hook
+     */
+    public function getWorkerStop(): ?Hook
+    {
+        return $this->workerStopHook;
     }
 
     /**
@@ -355,7 +368,7 @@ protected function getArguments(Hook $hook, array $payload = []): array
         $arguments = [];
         foreach ($hook->getParams() as $key => $param) { // Get value from route or request object
             $value = $payload[$key] ?? $param['default'];
-            $value = ($value === '' || is_null($value)) ? $param['default'] : $value;
+            $value = ($value === '' || $value === null) ? $param['default'] : $value;
 
             $this->validate($key, $param, $value);
             $hook->setParamValue($key, $value);
@@ -384,7 +397,7 @@ protected function getArguments(Hook $hook, array $payload = []): array
      */
     protected function validate(string $key, array $param, mixed $value): void
     {
-        if ('' !== $value && !is_null($value)) {
+        if ('' !== $value && $value !== null) {
             $validator = $param['validator']; // checking whether the class exists
 
             if (\is_callable($validator)) {
@@ -399,7 +412,7 @@ protected function validate(string $key, array $param, mixed $value): void
                 throw new Exception('Invalid ' .$key . ': ' . $validator->getDescription(), 400);
             }
         } elseif (!$param['optional']) {
-            throw new Exception('Param "' . $key . '" is not optional.', 400);
+            throw new Exception("Param $key is not optional.", 400);
         }
     }
 

tests/Queue/servers/AMQP/Dockerfile

@@ -1,3 +1,5 @@
 FROM phpswoole/swoole:php8.3-alpine
 
-RUN apk add autoconf build-base
+RUN apk add autoconf build-base
+
+RUN docker-php-ext-install pcntl