From 012f8842eb64707e13de179c51a7f5b5f39ad243 Mon Sep 17 00:00:00 2001
From: loks0n <22452787+loks0n@users.noreply.github.com>
Date: Fri, 11 Sep 2026 18:22:33 +0100
Subject: [PATCH] fix(swoole): do not receive a new message after stop

The 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 <noreply@anthropic.com>
---
 src/Queue/Adapter/Swoole.php                  | 16 ++++++++++++++++
 tests/Queue/E2E/Adapter/SwooleRestartTest.php | 14 ++++++++++++++
 2 files changed, 30 insertions(+)

diff --git a/src/Queue/Adapter/Swoole.php b/src/Queue/Adapter/Swoole.php
index b6f4a79..bf5c3d8 100644
--- a/src/Queue/Adapter/Swoole.php
+++ b/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/tests/Queue/E2E/Adapter/SwooleRestartTest.php b/tests/Queue/E2E/Adapter/SwooleRestartTest.php
index 60504c1..85b3e2e 100644
--- a/tests/Queue/E2E/Adapter/SwooleRestartTest.php
+++ b/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;