fix(swoole): do not receive a new message after stop - #88
Conversation
Sorry, something went wrong.
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>
|
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. |
Sorry, something went wrong.
|
Sorry, something went wrong.
| 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); |
There was a problem hiding this 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)
Prompt To Fix With AIThis 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!
Sorry, something went wrong.
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