FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix(swoole): do not receive a new message after stop by loks0n · Pull Request #88 · utopia-php/queue · GitHub

/ queue Public

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’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
loks0n wants to merge 1 commit into from
+30 −0
Conversation 3 Commits 1 Checks 2 Files changed 2
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 .php  (2) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
16 changes: 16 additions & 0 deletions src/Queue/Adapter/Swoole.php
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
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
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
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 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.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

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

private function waitFor(string $event, int $count): array
{
$deadline = microtime(true) + 10;
Expand Down
Back | FazBrowse Home | New Git URL