[KalimeroMK]

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 Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.59%. Comparing base (7c79c3b) to head (8a4bf0b).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #340      +/-   ##
============================================
+ Coverage     94.58%   94.59%   +0.01%     
- Complexity      373      375       +2     
============================================
  Files            59       59              
  Lines           978      981       +3     
============================================
+ Hits            925      928       +3     
  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]

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.

[samdark]

The limit fix itself works, but two things need changing:

1. A handled message is processed again after a stop. QueueConsumer::handle() runs the worker and then returns canContinue(). When the loop stops after a message was handled (SIGTERM, memory limit), the callback returns false, and InMemoryAdapter now keeps that message, so the next run handles it again: [1, 2] becomes [1, 1, 2]. testMessageIsNotConsumedWhenLoopCannotContinue() asserts 0 processed while the handler did run once; asserting executionTimes shows it. Checking the loop before processing a message (#346) fixes this, so I'd land both together.

2. The new AdapterInterface contract doesn't hold yet. It says false means "not consumed, keep it", but QueueConsumer returns false for a consumed message on stop, and queue-amqp's subscribe() acks regardless of the callback result. The old "Returns false if it cannot continue handling messages" sentence is still there next to it. "FIFO order" also isn't true for queue-db, which orders by priority, then id.

[KalimeroMK]

• The loop is checked before the message is handled, so false is only ever a message left untouched — e2b2182
• FIFO claim and the subscribe() contract sentence dropped — e2b2182
• queue:listen-all no longer waits out --pause after the loop asked to stop — f966853

Should #346 close with this, or do you still want the callback to return an enum?

[samdark]

Please merge this together with yiisoft/queue-amqp#143: with only one of the two, a worker stopping on its memory limit either loses the current message (this PR first) or processes it twice (queue-amqp#143 first). Also, please restore the line in AdapterInterface::subscribe() saying that returning false means the message was not consumed.

[KalimeroMK]

• subscribe() says again that false means the message was not consumed — 8a4bf0b