Skip to content

Poll again right after a full batch instead of waiting for the next signal - #21

Open
iGrog wants to merge 1 commit into
thesis-php:0.1.xfrom
iGrog:fix/consumer-drains-backlog
Open

iGrog wants to merge 1 commit into
thesis-php:0.1.xfrom
iGrog:fix/consumer-drains-backlog

Conversation

@iGrog

@iGrog iGrog commented Oct 7, 2026 •

Copy link
Copy Markdown

Fixes #19.

A backlog that is already in the queue produces no further insert notifications, so after reading one batch the consumer waited for the poll timer: a queue of N messages took ~N / batch × pollInterval to drain.

Change

Internal\consume(): after a batch that came back full (count === $config->batch), push the next poll request right away. A partial or empty batch does not re-poll, so idle behaviour and query count are unchanged. A consumer stopped from the handler ($ctrl->stop()) is respected: $polls->isComplete().

Test

testConsumerDrainsBacklogWithoutWaitingForPollInterval: 25 messages already in the queue, batch: 10, pollInterval: 10 s; all must be consumed within 2 s.

  • without the fix: Amp\CancelledException (the 2nd and 3rd batches wait 10 s and 20 s);
  • with the fix: passes.

Edge cases checked with a stress harness (real Postgres, pgmq 1.11.1 and 1.13.0)

scenario result
backlog of 0 / 5 / 10 / 30 / 35 messages, batch: 10, pollInterval: 10 s all drained in ≤ 0.06 s (without the fix: 30 and 35 time out)
backlog of 500, 4 consumers 0.15 s, no duplicates
consumer stopped in the middle of a backlog (handler 10 ms per message) stops in 0.01 s, nothing handled after the stop
handler throws in the middle of a batch error surfaces, no busy loop; the batch is rolled back and redelivered
producers keep inserting while a backlog of 500 drains (2 consumers) all 800 delivered, no duplicates ²
20 messages delayed by 3 s mixed with 25 visible ones, pollInterval: 2 s visible ones in 0.07 s; delayed ones at 4.1 s (≤ delay + pollInterval)
idle consumer, pollInterval: 1 s no busy loop: 13 transactions in 3 s

² When several separate transactions insert into the queue within the 30 ms notify throttle window, the last of them get no notification (pgmq notifies on the leading edge only) and wait for the poll timer. This happens with or without this change; details in thesis-php/message-bus#4.

Checks (run locally in the project's Docker images)

PHP 8.3 PHP 8.4 PHP 8.5
tests, highest deps ✅ 36 ✅ 36 ✅ 36
tests, lowest deps ✅ 36 ✅ 36 ✅ 36 ¹
phpstan, highest / lowest ✅ / ✅ ✅ / ✅ ✅ / ✅

fixer-check, rector-check, composer-validate, composer-normalize-check, deps-analyze: ✅

¹ Two PHP deprecations from amphp/amp (lowest version) on PHP 8.5: Increment on non-numeric string in FutureState.php:70. They are present on 0.1.x as well, and the exit code is 0.

…ignal

A backlog that is already in the queue produces no further insert notifications,
so after reading one batch the consumer slept until the poll interval elapsed:
a queue of N messages took ~N/batch * pollInterval to drain.
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.

Consumer reads one batch per poll signal: a backlog drains at batch / pollInterval

1 participant