From 36617ecc47cf19ffbe6c0cca15cf52676a12d21c Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Fri, 11 Sep 2026 18:37:28 +0100 Subject: [PATCH 1/2] fix(queue): do not receive a new message after stop The Swoole consume loops checked the stop flag only at the top of the loop, before blocking on the concurrency slot. At maxCoroutines=1 the push blocks for the whole handler, so a SIGTERM that landed mid-job was first noticed only after the slot came free, and by then the loop had already gone back to receive() and claimed the next message. Every handler process took one more job after being told to stop, so a rolling restart drained for two job durations per process, and a long job accepted after SIGTERM was SIGKILLed at the pod's grace period with its claim stranded. Re-check the flag once the slot is held and leave without receiving. The restart test now publishes a second slow job behind the in-flight one and asserts it is still on the queue after the drain; seen red on both SIGTERM and SIGINT before this change. Co-Authored-By: Claude Fable 5.1 --- packages/queue/src/Queue/Adapter/Swoole.php | 16 ++++++++++++++++ .../Queue/E2E/Adapter/SwooleRestartTest.php | 14 ++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/packages/queue/src/Queue/Adapter/Swoole.php b/packages/queue/src/Queue/Adapter/Swoole.php index b6f4a79ec..bf5c3d81a 100644 --- a/packages/queue/src/Queue/Adapter/Swoole.php +++ b/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) { diff --git a/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php b/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php index 60504c159..85b3e2ec0 100644 --- a/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php +++ b/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php @@ -123,10 +123,16 @@ public function testShutdownDrainsJobWithoutRestartingWorkers(int $signal): void $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) { @@ -154,6 +160,14 @@ 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 + { + $redis = new \Redis(); + $redis->connect('127.0.0.1', 16379); + + return (int) $redis->lLen($this->namespace . '.queue.worker-' . $worker); + } + private function waitFor(string $event, int $count): array { $deadline = microtime(true) + 10; From c1211c06b50b80af8401189459fd5aa4f58da5a4 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Fri, 11 Sep 2026 18:45:45 +0100 Subject: [PATCH 2/2] test(queue): cover the multi-queue drain and read depth through the broker The restart fixture takes a queue count, so the drain test runs once against the single-queue loop and once against the multi-queue one; the multi-queue cases go red when only the run() check is removed. Remaining depth is read through Broker\Redis::getQueueSize() rather than the list key it happens to use. Co-Authored-By: Claude Fable 5.1 --- .../Queue/E2E/Adapter/SwooleRestartTest.php | 21 +++++++++++-------- .../tests/Queue/servers/Swoole/restart.php | 13 +++++++++--- 2 files changed, 22 insertions(+), 12 deletions(-) diff --git a/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php b/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php index 85b3e2ec0..b90a0f327 100644 --- a/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php +++ b/packages/queue/tests/Queue/E2E/Adapter/SwooleRestartTest.php @@ -112,14 +112,18 @@ 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'); @@ -140,12 +144,12 @@ public function testShutdownDrainsJobWithoutRestartingWorkers(int $signal): void } } - 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, ); @@ -162,10 +166,9 @@ private function publish(int $worker, string $mode): void private function queued(int $worker): int { - $redis = new \Redis(); - $redis->connect('127.0.0.1', 16379); + $broker = new Redis(new Connection('127.0.0.1', 16379), new Connection('127.0.0.1', 16379)); - return (int) $redis->lLen($this->namespace . '.queue.worker-' . $worker); + return $broker->getQueueSize(new Queue('worker-' . $worker, $this->namespace)); } private function waitFor(string $event, int $count): array diff --git a/packages/queue/tests/Queue/servers/Swoole/restart.php b/packages/queue/tests/Queue/servers/Swoole/restart.php index f545fd5db..35c7d3e01 100644 --- a/packages/queue/tests/Queue/servers/Swoole/restart.php +++ b/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 {