utopia-php / utopia-php/monorepo

Consumer::reject() has no terminal verdict, so a permanent failure buys the whole retry chain and an in-flight slot for all of it

Open
#276 2 comments 0 reactions 1 assignee View on GitHub

@levivannoort is already working on this.

Since Sep 15, 2026.

enhancement
Dominant language
PHP
Stars
3
Forks
4
Avg merge
12h 25m
Merged PRs (30d)
103

Description

Consumer::reject() takes a message and nothing else:

// packages/queue/src/Queue/Consumer.php:16
/** Mark a message as failed. */
public function reject(Queue $queue, Message $message): void;

So Broker\Nats::reject() has one branch available, and it is the same one for every failure:

// packages/queue/src/Queue/Broker/Nats.php:745
if ($numDelivered >= $this->maxDeliver) {
    // Exhausted: park on the dead stream and drop it from the work stream.
    $this->commandsJs()->publish($this->deadSubject($queue), $jsMessage->getData());
    $onCommands->term('max deliveries exceeded');

    return;
}

// Redeliver later (AckWait/NAK); a crashed worker is reclaimed the same way.
$onCommands->nak($this->backoffFor($numDelivered));

A NAK'd-with-delay message stays delivered-but-unacked for the whole of its backoff, so it holds
one maxAckPending slot while it sleeps. That is correct for a transient failure — the slot is
the reservation that stops the same work being handed to a second worker. It is wrong for a
failure that will never succeed: a message whose handler threw Region hostname not configured
in 4ms buys 30 + 120 + 600 + 600 ≈ 1,250s of slot before it dead-letters, and the caller has
no way to say so.

The caller does know. The handler that threw it knows the difference between "MySQL was
unreachable" and "this tenant has no database", "401 Invalid ingestion signature", or "this
region has no hostname configured". That knowledge is thrown away twice over:

// packages/queue/src/Queue/Adapter.php:374
} catch (\Throwable $error) {
    // The work did not happen, so hand the message back to be retried.
    try {
        $consumer->reject($queue, $message);
    } catch (\Throwable) {
    }

    $this->report($errorCallback, $error, $message);

reject() runs before report(), so an application's error hook sees the Throwable after
the redelivery decision has already been made. There is no seam at which a consumer of this
library can shorten a doomed message's life.

Why it matters

In cloud's staging this saturated two queues in two days and neither recovered until its input
dried up. Q_V1-REGION-MANAGER on a maxAckPending of 60: num_ack_pending pinned at 60 for
3.5 hours, all 60 redeliveries, num_pending 1,390 never delivered once, handlers finishing in
4ms. Three /jsz reads 45s apart returned byte-identical counters. The healthy messages behind
the poison ones — a different region, which would have succeeded — could not be delivered at
all, and the backlog drained only by dead-lettering at ~1.9 msg/min.

On Broker\Redis the same input was log noise: no backoff and no in-flight ceiling, so a poison
message costs a slot for milliseconds. This is a semantic regression in the NATS broker relative
to the Redis one, for the class of failure that is guaranteed to recur.

The workaround available today is to shorten the retry budget, which trades a transient
failure's chance of succeeding for a permanent one's blast radius. They are the same knob
pointed in opposite directions, and there is no setting that is right for both.

Fix

Give reject() a terminal verdict, and let Broker\Nats::reject() take the term() +
dead-stream branch it already has immediately instead of NAK-with-backoff:

  • a marker the handler raises — Utopia\Queue\PermanentFailure (or an interface a caller's own
    exception can implement, which avoids forcing a wrapper at every throw site), and
  • Adapter::runPhases() passing that verdict into reject().

Broker\Redis::reject() does the same minus the delay, so the two brokers keep matching
semantics.

This restores Redis's cost profile for the poison case without shortening the retry budget for
transient ones, which is the only version of the fix that does not force the trade above.

Adding a parameter to Consumer::reject() is a BC break for implementers. An optional third
argument, or a separate rejectPermanently() defaulting to reject(), keeps it to a minor.

Found while working through the cloud NATS migration —
appwrite-labs/cloud#5815 (blocker 11). Related: #198.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.