Skip to content

fix(swoole): do not receive a new message after stop #88

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
loks0n wants to merge 1 commit into from
Closed

fix(swoole): do not receive a new message after stop #88

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions src/Queue/Adapter/Swoole.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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) {
Expand Down
14 changes: 14 additions & 0 deletions tests/Queue/E2E/Adapter/SwooleRestartTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
Comment on lines +163 to +168

@greptile-apps greptile-apps Bot Sep 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 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!

Fix in Claude Code Fix in Codex

}

private function waitFor(string $event, int $count): array
{
$deadline = microtime(true) + 10;
Expand Down