fix(queue): keep the consumer alive when the broker fails

[loks0n]

The bug

Both consume loops called receive() unguarded:

while (!$this->isStopped()) {
    $message = $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT);
    if (!$message instanceof Message) { continue; }
    ...
}

Anything the broker throws unwinds out of consume() and ends the worker. A single unreachable moment costs the process, and whatever was in flight goes with it.

Note the asymmetry this creates with the method right below it, whose docblock already promises the opposite:

process() — Never throws: a failed handler is rejected and reported to $errorCallback

Message handling was made total. Message acquisition was not.

Why it looked fine

Nothing at that level retried. What resembled resilience came from the connection pool underneath: reconnectAttempts × reconnectSleep delayed each failed receive() by tens of seconds, which behaved like a backoff in front of the missing guard.

#94 removes those pool retries, which turns the slow stall into a crash loop. But the guard was missing either way — a consumer should survive a broker outage regardless of what its pool happens to do — so this stands on its own and is worth landing first.

The fix

Both loops route through one helper on the base class:

protected function nextMessage(callable $errorCallback): ?Message
{
    try {
        return $this->consumer->receive($this->queue, static::RECEIVE_TIMEOUT);
    } catch (\Throwable $error) {
        // A reporting hook that throws must not cost the worker either.
        try {
            $errorCallback(null, $error);
        } catch (\Throwable) {
        }

        sleep(static::RECEIVE_BACKOFF);

        return null;
    }
}

Returning null lands on the existing continue, so the loop carries on and the worker resumes when the broker returns.

One helper rather than a try/catch in each loop, because Swoole overrides consume() and the Workerman path inherits the base. sleep() is coroutine-hooked under Swoole, so it yields rather than blocking the worker; under the plain adapter it blocks, which is correct for a single-threaded consumer.

Reporting is the consumer's decision

The failure goes to the existing $errorCallback rather than to error_log. Its signature is already nullable, and Server already guards for it:

function (?Message $message, Throwable $th): void {
    $this->context()->set('error', fn() => $th);
    if ($message instanceof \Utopia\Queue\Message) {   // already handles null
        $this->context()->set('message', fn() => $message);
    }
    foreach ($this->errorHooks as $hook) { ... }
}

So this is the case that shape was for. A consumer logs a broker failure through the same Server::error() hook it already registers for handler failures, instead of the library deciding on its behalf — no new API, and nothing writes to stderr behind the caller's back.

The callback is invoked defensively: a reporting hook that throws must not cost the worker either, which is the whole point of the change. consume()'s callbacks are now documented, including that $errorCallback can receive a null message.

Test

Drives the Swoole loop with a consumer that fails twice, then delegates to a working broker, asserting both failures reach the callback with a null message. Verified against the unguarded code:

before:  Fatal error: Uncaught RuntimeException: broker unreachable
after:   OK (1 test, 4 assertions)

Registered in the unit suite — it needs no Redis, using the existing InMemoryConnection double.

bin/monorepo check queue, test queue and validate all pass.

Scope

RECEIVE_BACKOFF is a flat 1s. Escalating backoff would be kinder to a broker that is down for a long stretch, but a fixed pause is the smaller change and matches the existing RECEIVE_TIMEOUT shape; worth revisiting if it ever proves noisy.

I did not add a second test for the base Adapter::consume() path. It calls the same helper, and covering it would mean a concrete Adapter stub implementing four abstract methods to exercise one shared line.

🤖 Generated with Claude Code