Skip to content

Keep the message that hits the limit in the queue - #340

Open
KalimeroMK wants to merge 1 commit into
yiisoft:masterfrom
KalimeroMK:fix-limit-message-loss
Open

KalimeroMK wants to merge 1 commit into
yiisoft:masterfrom
KalimeroMK:fix-limit-message-loss

Conversation

@KalimeroMK

@KalimeroMK KalimeroMK commented Oct 12, 2026 •

Copy link
Copy Markdown
Q A
Is bugfix? ✔️
New feature? ❌
Breaks BC? ❌
Tests added? ✔️

What does this PR do?

queue:run --limit=1 over two queued messages prints Messages processed: 1. and leaves the queue empty. The second message is discarded without ever reaching a handler.

$producer->push(new GenericMessage('simple', null));
$producer->push(new GenericMessage('simple', null));

$consumer->run(1);                  // 1
$adapter->getMessagesList();        // [] — expected 1 message still queued

Cause

QueueConsumer::run() returns false from the callback for the message that trips the limit, without handling it. InMemoryAdapter::runExisting() unsets the message anyway, before it looks at what the callback returned:

$result = $handlerCallback($this->messages[$this->current]);
unset($this->messages[$this->current]);   // stubs/InMemoryAdapter.php:54
$this->current++;

Both shipped adapters already read false as "not consumed":

  • queue-redis — if ($result) { $this->provider->delete(...); } (Adapter.php:35)
  • queue-amqp — if ($this->messageConsumed = $callback($message)) { basic_ack(); } else { basic_nack($tag, false, true); } (ExistingMessagesConsumer.php:42)

So the stub is the outlier, and AdapterInterface never said which reading was intended — it only said "Returns false if it cannot continue handling messages".

Fix

InMemoryAdapter keeps a message the callback refused, and runExisting()/subscribe() state the contract the other adapters already implement. QueueConsumer is unchanged.

Not addressed here

One bit cannot carry both "was this message consumed?" and "should I continue?". When the loop stops after a message was handled (SignalLoop on SIGTERM), the callback returns false for a message that was processed, so it is kept and processed again on the next run. Every adapter behaves this way today; splitting the two signals is an AdapterInterface change and belongs in its own issue: #346.

Validation

Both new assertions fail without the fix (actual size 0 matches expected size 1, actual size 1 matches expected size 2) and pass with it. Full suite: 280 tests, 516 assertions, OK. Psalm and PHP CS Fixer clean.

@codecov

codecov Bot commented Oct 12, 2026 •

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.58%. Comparing base (7c79c3b) to head (a258088).

Additional details and impacted files
@@            Coverage Diff            @@
##             master     #340   +/-   ##
=========================================
  Coverage     94.58%   94.58%           
  Complexity      373      373           
=========================================
  Files            59       59           
  Lines           978      978           
=========================================
  Hits            925      925           
  Misses           53       53           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@KalimeroMK

Copy link
Copy Markdown
Author

Pushed a follow-up: the fix left return false inside the callback reachable only when the loop asks to stop, which no test covered (codecov caught it, -1 line). Adding that test showed run() returned one less than it handled in that case — the message is always handled by the time the callback decides whether to continue — so the count moved out of the early return.

`queue:run --limit=1` over two queued messages reported one processed and left the queue empty:
`InMemoryAdapter::runExisting()` removed the message that tripped the limit even though the
callback returned `false` without handling it.

Both shipped adapters already treat `false` as "not consumed" — `queue-redis` deletes only when
the callback returned `true`, `queue-amqp` nacks with requeue otherwise — so the stub was the
outlier. Spell that contract out on `runExisting()` and `subscribe()`.
@KalimeroMK
KalimeroMK force-pushed the fix-limit-message-loss branch from 9a65e0f to a258088 Compare October 12, 2026 22:14
@KalimeroMK KalimeroMK changed the title Fix message loss when QueueConsumer::run() hits the limit Keep the message that hits the limit in the queue Oct 12, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant