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

[levivannoort]

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.

[levivannoort]

Paired with #277, which is the other half of the same wedge: this one stops a permanent failure from consuming a retry chain, #277 stops a transient one's chain from being charged against consumer throughput. Either alone leaves half of it.

[levivannoort]

A live instance of this, measured in Appwrite Cloud's fra1 staging cluster on 2026-09-16 (read off
/jsz on the NATS cluster, two samples four minutes apart).

v1-region-manager receives cache-purge jobs naming regions that have no configured hostname. The
handler throws before a request is built, so the failure is permanent at attempt 1 — but reject()
cannot say so, and the message buys the full ladder (ackWait=30, backoff=30,120,600,
maxDeliver=5) before it dead-letters.

What that looks like on the consumer:

• Q_V1-REGION-MANAGER: 44 messages, all 44 in ack_pending, all 44 redelivered, num_pending 0.
  Flat across the sample (delivered 8924 → 8925): a fixed set of permanently-failing messages
  cycling the ladder, not a backlog draining.
• 44 of the consumer's 60 maxAckPending slots held, against 3 replicas × 30 coroutines of capacity.
• Q_V1-REGION-MANAGER_DEAD grew to 236 in about 2.5 hours, and every one of those messages spent
  roughly 20 minutes of wall clock and five delivery attempts to reach a verdict that was knowable on
  the first.

So the cost is not only the wasted attempts — it is that the in-flight budget is a fixed fleet-wide
number, and permanently-failing messages hold slots in it for the whole chain. A queue with enough of
them stops draining the work that could succeed, which is what happened here: the caller-side fix
(appwrite-labs/cloud#5859) had to skip these at the handler precisely because the broker offers no way
to fail one terminally.

A terminal verdict on reject() would have turned 5 attempts × 44 messages into 1 × 44, and freed the
slots immediately rather than 20 minutes later.