Batch the consume path

Consumer::receive() (src/Queue/Consumer.php:10) returns one message per call, and every broker implements it that way. enqueueMany() already batches on both brokers (#229 made the NATS side one round trip per window), so the publish path is batched and the consume path is not.

Broker\Redis::receive() costs 5 round trips per message: one BRPOP (Broker/Redis.php:57) plus four claim writes (:95-98). Broker\Nats::fetchOne() (Broker/Nats.php:973) is fetch(1, ...), hardcoded.

This is the bottleneck on cloud's stats-usage, which profiles as BRPOP + json_decode rather than handler work.

Shape

A capability interface, not a change to Consumer — the package's established idiom, alongside Exclusive, Bounded and Stateful:

namespace Utopia\Queue\Consumer;

interface Batched
{
    /** @return list<Message> */
    public function receiveBatch(Queue $queue, int $timeout, int $max): array;
}

Three contract points, each a bug if missed: only the first message may block (a fill-the-batch wait turns a sparse queue into a $timeout-latency queue); every returned message is claimed and owed its own ack, no batch ack; nothing may be lost between the pop and the claim.

receive() stays mandatory, so every existing consumer and test fake keeps working, and Adapter falls back to it for anything not Batched.

Redis

Blocking brPop for the first message — latency unchanged — then a non-blocking tail-pop for the rest. Claim writes collapse from 4N to N+3 using leftPushMany (already on Connection.php:26) and a new incrementBy. Net: N+5 round trips per batch instead of 5N, i.e. ~1.3 per message at N=16 instead of 5.

Two methods added to Connection (rightPopMany, incrementBy), which is six implementations plus a LockingTest::operationProvider() yield each. Deliberately not adding pipeline(): phpredis \RedisCluster cannot pipeline and Connection\Locking has no vocabulary for holding its lock across more than one command.

Rejected: LMPOP (raises the floor to Redis 7.0 for one round trip per batch; a silent ERR unknown command on an older production Redis is a crash-loop) and an atomic Lua claim (the claim spans five hash slots, and hash-tagging renames live keys, orphaning in-flight processing/jobs entries across the deploy — and there is a live cluster E2E path).

Sharpest detail: between the tail-pop and the processing write, N messages exist only in worker memory. A throw there loses them outright. Pop, then claim, then decode — and a failed claim write must push the batch back with rightPushMany before the error propagates.

NATS

Consumer::fetch(N, $timeout) in packages/nats loops until the batch fills or the deadline passes (JetStream/Consumer.php:89-95), so fetch(16, 2.0) against three waiting messages burns the full two seconds. Naive batching makes NATS slower. Use what is already there: no_wait sends no expires and the server answers 404 immediately, so pull() becomes a batched noWait priority poll, a blocking fetch of exactly 1, then a noWait top-up. Write the latency guard test first.

Per-message ack isolation already holds — inFlight is keyed by pid — so one poison message in a batch of 16 does not take the other 15.

Adapter

Adapter\Swoole::consumeBound() reserves one coroutine slot before receiving; the invariant is that a message is never removed from the broker without a slot already reserved for it (SwooleConcurrencyTest.php:42). Batched, it reserves n and asks for n — the consume loop is the only pusher on the slot channel, so the extra pushes cannot block. At maxCoroutines = 1 it degenerates to exactly today's code, which is the regression proof.

Server::job() gains int $batch = 1 and refuses batch > maxCoroutines, which is what keeps the Bounded/maxAckPending guard at Server.php:473 arithmetically correct. Sequential adapters and Adapter\KubernetesJob stay single.

Note the cap is the gate. Batch size is bounded by free coroutine slots, so a worker at maxCoroutines = 1 gains nothing. cloud's worker-stats-usage is one of those today — raising it comes first and is measured separately.

Tracked on appwrite-labs/cloud#5815.