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

[loks0n]

Problem

Both Swoole consume loops (consumeBound() and run()) check isStopped() 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 lands mid-job is first noticed after the slot comes free, and by then the loop has already gone back to receive() and claimed the next message.

Measured on cloud's compose stack with _APP_WORKERS_NUM=3 and six 5s jobs queued: SIGTERM during the first three, and each process still took one more, so the drain ran for 10s instead of 5s. Every handler process accepts one job after being told to stop. With cloud's 30s terminationGracePeriodSeconds and no reap() on most Redis queues, a long build or delete accepted after SIGTERM is SIGKILLed and its claim is stranded on processing.

Fix

Re-check the flag once the slot is held; pop it and leave without receiving. A message that receive() already returned before the flag flipped is still processed and committed as before.

Test

SwooleRestartTest::testShutdownDrainsJobWithoutRestartingWorkers 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 the fix (actual size 2 matches expected size 1), green after. bin/monorepo test queue passes; Pint passes.

Supersedes utopia-php/queue#88, closed.

🤖 Generated with Claude Code

[greptile-apps]

Confidence Score: 5/5

The PR appears safe to merge; no blocking or non-blocking correctness issue remains.

Summary

• Re-checks the stop flag immediately after acquiring a concurrency slot in both consumption paths.
• Releases the acquired slot before leaving during shutdown.
• Verifies that active work drains while queued work remains pending for both signals and one-queue or multi-queue configurations.

_{Reviews (2) · Last reviewed commit: "test(queue): cover the multi-queue drain..."}

[github-actions]

Benchmark results

queue — workload shapes across both concurrency axes (4 cores, 600 messages, median of 3)

workload	shape	redis msg/s	nats msg/s
io	1p x 1c	37	38
io	1p x 4c	149	153
io	4p x 1c	148	152
cpu	1p x 1c	160	172
cpu	1p x 4c	175	178
cpu	4p x 1c	384	403
mixed	1p x 1c	62	64
mixed	1p x 4c	164	174
mixed	4p x 1c	242	252

Np x Mc = N consumer processes x M handler coroutines. io yields and should follow
coroutines; cpu does not and should follow processes; mixed is in between.

Shared CI runners — treat absolute numbers as rough, compare modes within a run. Commit c123375.