fix(queue): do not receive a new message after stop

packages/queue/src/Queue/Adapter/Swoole.php

@@ -234,6 +234,14 @@ protected function consumeBound(
         while (!$this->isStopped()) {
             $slots->push(true);
 
+            // The push blocks while every slot is busy, so a stop that landed
+            // during a handler is first seen here. Receiving now would claim
+            // a message the process was told not to take.
+            if ($this->isStopped()) {
+                $slots->pop();
+                break;
+            }
+
             $message = $this->nextMessage($errorCallback);
 
             if (!$message instanceof Message) {
@@ -287,6 +295,14 @@ protected function run(
         while (!$this->isStopped()) {
             $slots->push(true);
 
+            // The push blocks while every slot is busy, so a stop that landed
+            // during a handler is first seen here. Receiving now would claim
+            // a message the process was told not to take.
+            if ($this->isStopped()) {
+                $slots->pop();
+                break;
+            }
+
             $message = $this->nextMessageFrom($errorCallback, $queue, $consumer);
 
             if (!$message instanceof Message) {

packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php

@@ -112,34 +112,44 @@ public function testSimultaneousWorkerExitsRestoreTheWholePool(): void
 
     public static function stopSignals(): \Iterator
     {
-        yield 'SIGTERM' => [SIGTERM];
-        yield 'SIGINT' => [SIGINT];
+        // One queue per worker takes the single-queue loop, two the
+        // multi-queue one; both must drain the same way.
+        foreach ([1, 2] as $queues) {
+            yield "SIGTERM, {$queues} queue(s)" => [SIGTERM, $queues];
+            yield "SIGINT, {$queues} queue(s)" => [SIGINT, $queues];
+        }
     }
 
     #[DataProvider('stopSignals')]
-    public function testShutdownDrainsJobWithoutRestartingWorkers(int $signal): void
+    public function testShutdownDrainsJobWithoutRestartingWorkers(int $signal, int $queues): void
     {
-        $this->start(3);
+        $this->start(3, $queues);
         $ready = $this->waitFor('ready', 3);
         $this->events = [];
         $this->publish(0, 'slow');
+        $this->publish(0, 'slow');
         $this->waitFor('started', 1);
         $this->assertTrue(proc_terminate($this->process, $signal));
         $this->waitFor('exited', 1);
+        // The job in flight finishes; the one behind it stays put. Before the
+        // stop flag was re-checked after the slot came free, the loop went
+        // straight back to receive and pulled it, and a job accepted after
+        // SIGTERM is one more the grace period has to cover.
         $this->assertCount(1, array_filter($this->events, fn(array $e): bool => $e['event'] === 'processed'));
+        $this->assertSame(1, $this->queued(0), 'A message published behind the in-flight job must still be on the queue after the drain');
         $this->assertCount(3, array_filter($this->events, fn(array $e): bool => $e['event'] === 'stopped'));
         $this->assertCount(0, array_filter($this->events, fn(array $e): bool => $e['event'] === 'ready'));
         foreach ($ready as $worker) {
             $this->assertFalse(Process::kill($worker['pid'], 0), 'Supervisor must reap every child before returning');
         }
     }
 
-    private function start(int $workers): void
+    private function start(int $workers, int $queues = 1): void
     {
         $this->namespace = 'restart-' . bin2hex(random_bytes(8));
         $this->log = tempnam(sys_get_temp_dir(), 'queue-restart-');
         $this->process = proc_open(
-            [PHP_BINARY, '-d', 'display_errors=0', '-d', 'log_errors=1', '-d', 'error_log=/dev/stderr', __DIR__ . '/../../servers/Swoole/restart.php', $this->namespace, (string) $workers],
+            [PHP_BINARY, '-d', 'display_errors=0', '-d', 'log_errors=1', '-d', 'error_log=/dev/stderr', __DIR__ . '/../../servers/Swoole/restart.php', $this->namespace, (string) $workers, (string) $queues],
             [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w'], 2 => ['file', $this->log, 'a']],
             $pipes,
         );
@@ -154,6 +164,13 @@ private function publish(int $worker, string $mode): void
         $this->assertTrue($broker->publish(new Queue('worker-' . $worker, $this->namespace), ['mode' => $mode]));
     }
 
+    private function queued(int $worker): int
+    {
+        $broker = new Redis(new Connection('127.0.0.1', 16379), new Connection('127.0.0.1', 16379));
+
+        return $broker->getQueueSize(new Queue('worker-' . $worker, $this->namespace));
+    }
+
     private function waitFor(string $event, int $count): array
     {
         $deadline = microtime(true) + 10;

packages/queue/tests/Queue/servers/Swoole/restart.php

@@ -12,15 +12,18 @@
 use Utopia\Queue\Server;
 
 [$script, $namespace, $workers] = $_SERVER['argv'];
+// A third argument above 1 registers extra queues per worker, so the adapter
+// takes its multi-queue loop instead of the single-queue one.
+$queues = (int) ($_SERVER['argv'][3] ?? 1);
 $adapter = new Swoole(
     fn(): \Utopia\Queue\Broker\Redis => new Redis(new Connection('127.0.0.1', 16379), new Connection('127.0.0.1', 16379)),
     (int) $workers,
     $namespace,
 );
 $server = new Server($adapter);
-$server->workerStart()->inject('workerId')->action(function (string $workerId) use ($server, $adapter): void {
+$server->workerStart()->inject('workerId')->action(function (string $workerId) use ($server, $adapter, $queues): void {
     ini_set('memory_limit', '64M');
-    $server->job('worker-' . $workerId)->inject('message')->action(function (Message $message) use ($workerId, $adapter): void {
+    $handler = function (Message $message) use ($workerId, $adapter): void {
         $mode = $message->getPayload()['mode'];
         if ($mode === 'fatal') {
             echo strlen(str_repeat('x', 128 * 1024 * 1024)) . PHP_EOL;
@@ -35,7 +38,11 @@
             Coroutine::sleep(0.5);
         }
         echo json_encode(['event' => 'processed', 'worker' => $workerId, 'pid' => getmypid()]) . PHP_EOL;
-    });
+    };
+    $server->job('worker-' . $workerId)->inject('message')->action($handler);
+    for ($extra = 1; $extra < $queues; $extra++) {
+        $server->job('worker-' . $workerId . '-' . $extra)->inject('message')->action($handler);
+    }
     echo json_encode(['event' => 'ready', 'worker' => $workerId, 'pid' => getmypid()]) . PHP_EOL;
 });
 $server->workerStop()->inject('workerId')->action(function (string $workerId): void {