fix(swoole): 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. Unit suite: 117 tests green.

🤖 Generated with Claude Code

[github-actions]

Thanks for contributing! This repository is a read-only mirror; development for this library happens in packages/queue in the utopia-php monorepo. Please open this pull request there instead.

[greptile-apps]

Confidence Score: 4/5

The production change appears correct, but the repository’s explicit testing requirement must be satisfied before merging by replacing the raw Redis assertion with the public broker API.

Findings

1.  Test Mirrors Redis Internals ▶

Fix with agent prompt

### Issue 1
tests/Queue/E2E/Adapter/SwooleRestartTest.php:163-168
The new `queued()` helper bypasses the broker API, hard-codes the Redis connection, and reconstructs the internal queue key for a raw `lLen()` call. This violates the repository directive to test observable behavior without mirroring source code or configuration. A harmless connection or key-format change would now fail this shutdown test even when the behavior remains correct. This repository requirement must be satisfied before merging; query the queue depth through `Redis::getQueueSize()` with a `Queue` instead.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Summary

• Releases the newly acquired slot and exits the receive loop when shutdown was requested.
• Preserves wait-group draining so existing handlers can finish before consumer cleanup.
• Extends the Swoole shutdown test with a second queued job, although its queue-depth assertion is improperly coupled to Redis internals.

_{Reviews (1) · Last reviewed commit: "fix(swoole): do not receive a new messag..."}

[greptile-apps]

Test Mirrors Redis Internals

The new queued() helper bypasses the broker API, hard-codes the Redis connection, and reconstructs the internal queue key for a raw lLen() call. This violates the repository directive to test observable behavior without mirroring source code or configuration. A harmless connection or key-format change would now fail this shutdown test even when the behavior remains correct. This repository requirement must be satisfied before merging; query the queue depth through Redis::getQueueSize() with a Queue instead.

Context Used: Call out and harshly judge implementation-coupled tests. We don't mirror source code, configuration, or version pins in assertions. We test observable behavior; use linters for syntax and schema checks. (source)

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/Queue/E2E/Adapter/SwooleRestartTest.php
Line: 163-168

Comment:
**Test Mirrors Redis Internals**

The new `queued()` helper bypasses the broker API, hard-codes the Redis connection, and reconstructs the internal queue key for a raw `lLen()` call. This violates the repository directive to test observable behavior without mirroring source code or configuration. A harmless connection or key-format change would now fail this shutdown test even when the behavior remains correct. This repository requirement must be satisfied before merging; query the queue depth through `Redis::getQueueSize()` with a `Queue` instead.

**Context Used:** Call out and harshly judge implementation-coupled tests. We don't mirror source code, configuration, or version pins in assertions. We test observable behavior; use linters for syntax and schema checks. ([source](https://app.greptile.com/review/custom-context?memory=instruction-0))

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!