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

Summary

Consumer::consume() reads one batch per poll signal. A signal comes either from an insert notification or from the poll timer. When messages are already in the queue (a backlog), no new inserts happen, so after reading one batch the consumer sleeps until the next timer tick, even though the queue still holds messages.

A queue of N messages therefore needs about N / batch × pollInterval to drain, while the CPU stays idle.

Where

src/Internal/functions.php L48–L68: the loop iterates over $polls. Each iteration reads one batch (readBatch(count: $config->batch)), handles it, commits, and goes back to waiting for the next poll signal.

Poll signals come from:

• TimeoutWatcher, every pollInterval (default 3 s);
• ChannelWatcher, on pgmq.q_<queue>.INSERT notifications, i.e. only when something is inserted.

Nothing triggers another read when a batch comes back full.

Reproduction

$queue = createQueue($pg, 'backlog');
$queue->sendBatch(array_fill(0, 25, new SendMessage('{}')));   // the backlog is already there

$consumed = 0;
$context = createConsumer($pg)->consume(
    static function (array $messages, ConsumeController $ctrl) use (&$consumed): void {
        $consumed += \count($messages);
        $ctrl->ack($messages);
        if ($consumed === 25) {
            $ctrl->stop();
        }
    },
    new ConsumeConfig(queue: 'backlog', batch: 10, pollInterval: TimeSpan::fromSeconds(10)),
);

$context->awaitCompletion(new TimeoutCancellation(2));   // CancelledException: needs ~20 s

The first batch is read immediately (the initial poll), the second one after 10 s, the third after 20 s.

Impact in practice

We hit this through thesis/message-bus, whose PgmqTransport has the same loop (thesis-php/message-bus#2). Our workload fans out ~450 messages into one queue at once, and the workers then read exactly batch messages per pollInterval: 2 workers × 10 messages / 5 s = 40 messages per 10 s, while handling a message takes ~14 ms. Waiting time in the queue was p50 56 s, p90 267 s, and the workers were idle ~97 % of the time.

Proposed fix

After a full batch, push the next poll request immediately: a full batch means there may be more. A partial batch means the queue is (most likely) empty, so the consumer goes back to waiting for a notification or the timer, as before.

if (\count($messages) === $config->batch && !$polls->isComplete()) {
    $polls->pushAsync(null)->ignore();
}

There are no extra queries when idle: an empty or partial read never re-polls.

PR: #21 (the same loop in thesis/message-bus: thesis-php/message-bus#2).