[Feature] No broker actor caps redelivery or has a dead-letter path, so a poison message hot-loops on AMQP requeue, redelivers forever on JetStream and sits in the Redis PEL with nothing to reclaim it

[pathosDev]

Use case

Every broker actor in src/io/broker/ gives the application a way to say "I could not process this" — negativeAcknowledgment on Kafka and AMQP, nak on JetStream, a missing acknowledgment on Redis Streams. None of them has anywhere for the message to go. There is no retry counter, no attempt cap, no dead-letter destination, and nothing that distinguishes "the database was briefly down" from "this payload will never parse".

The consequences are protocol-specific and all of them are production-relevant:

• AMQP hot-loops. nack defaults to requeue: true, so a permanently-malformed message returns to the queue immediately and is redelivered at once. With prefetch: 1 and a single consumer that is a tight loop between the broker and the actor, at full CPU, forever. No exchange is declared as a DLX in assertQueue either, so RabbitMQ has nowhere to route it even if the application nacked with requeue: false.
• JetStream redelivers forever. The consumer config the actor builds has no max_deliver; NATS defaults to unlimited. A message that is nak'd on every delivery is redelivered until the stream's retention policy removes it. (term() is exposed per message, so an application can opt out one message at a time — there is just no policy that does it automatically.)
• Redis Streams neither redelivers nor drops. An unacked entry sits in the group's pending-entries list with nothing to reclaim it. The code says so in a comment.
• Kafka re-delivers on rebalance with no attempt counter, so the same record can cycle indefinitely across restarts.

What is missing is one policy, applied uniformly: cap the attempts, then route the message somewhere an operator can look at it.

Proposed shape

A shared, opt-in deadLetter block in BrokerCommonOptionsType, implemented once in BrokerActor and specialised by each subclass at the one point it settles a message:

readonly deadLetter?: {
  /** Attempts before the message is considered poison.  Default 5. */
  readonly maxAttempts?: number;
  /** What to do when the cap is reached. */
  readonly action?: 'drop' | 'dead-letter' | 'protocol';   // default 'dead-letter'
  /** Where dead-lettered payloads go when `action: 'dead-letter'`. */
  readonly target?: ActorRef<PoisonMessage<unknown>>;
};

with 'protocol' meaning "use the broker's own facility", which is where the per-subclass work is:

• AMQP — declare a DLX/DLQ pair alongside assertQueue when deadLetter is configured, and nack with requeue: false once the cap is reached so the broker routes it. queueOptions already exists as the place to thread deadLetterExchange through.
• JetStream — set max_deliver on the consumer (the field is simply absent from ConsumerAddConfig today) and call term() rather than nak() at the cap.
• Kafka — count attempts per (topic, partition, offset), and at the cap either produce to a <topic>.DLT and commit, or drop-and-commit; the pendingCommits map is already keyed exactly right for the counter.
• Redis Streams — this needs [Feature] RedisStreams PEL-reclaim (XAUTOCLAIM/XCLAIM) command #462 (XAUTOCLAIM/XCLAIM) first; with reclaim in place, an entry whose delivery_count exceeds the cap is XACKed and copied to a dead-letter stream.
• MQTT — nothing to do until the QoS/ack story is resolved (filed separately in this batch).

The framework already has system.deadLetters and #433 is the durable-DLQ-with-replay track, so action: 'dead-letter' should route there rather than inventing a second inspection surface.

Evidence for the current state

AMQP nacks back into the queue by default, src/io/broker/AmqpActor.ts:195-201:

src/io/broker/AmqpActor.ts:195-201
  private onNegativeAcknowledgment(command: NegativeAcknowledgmentCommand): void {
    const raw = this.pendingAcks.get(command.delivery.ackToken);
    if (raw && this.channel) {
      try { this.channel.nack(raw, false, command.requeue ?? true); } catch { /* ignore */ }
      this.pendingAcks.delete(command.delivery.ackToken);
    }
  }

and the queue is asserted with no dead-letter routing, src/io/broker/AmqpActor.ts:105-109:

src/io/broker/AmqpActor.ts:105-109
    for (const binding of this.options.bindings ?? []) {
      await this.channel.assertQueue(binding.queue, binding.queueOptions ?? { durable: true });
      if (binding.exchange) {
        await this.channel.bindQueue(binding.queue, binding.exchange, binding.routingKey ?? '');
      }

The JetStream consumer config has no max_deliver field to set, src/io/broker/JetStreamActor.ts:637-646:

src/io/broker/JetStreamActor.ts:637-646
type ConsumerAddConfig = {
  durable_name: string;
  ack_policy?: 'explicit' | 'none' | 'all';
  ack_wait?: number;
  filter_subject?: string;
  max_ack_pending?: number;
  deliver_policy?: 'all' | 'last' | 'new' | 'by_start_sequence' | 'by_start_time';
  opt_start_seq?: number;
  opt_start_time?: string;
};

Redis Streams states the gap in its own comment, src/io/broker/RedisStreamsActor.ts:131-141:

src/io/broker/RedisStreamsActor.ts:131-141
  /**
   * Fire-and-forget — awaiting the `XACK` would stall the mailbox behind a
   * broker round-trip.  A failure is logged and the entry stays in the
   * group's pending list; nothing reclaims it yet (see #462).
   */
  private onAcknowledgment(command: AcknowledgmentCommand): void {
    if (this.redis && this.options.consumerGroup) {
      void this.redis.xack(command.stream, this.options.consumerGroup.group, command.id)
        .catch((e: Error) => this.log.warn(`xack failed: ${e.message}`));
    }
  }

Kafka's nack path is a warn and a rejection, with no counter, src/io/broker/KafkaActor.ts:450-460:

src/io/broker/KafkaActor.ts:450-460
  private onNegativeAcknowledgment(command: NegativeAcknowledgmentCommand): void {
    const key = pendingKey(command.topic, command.partition, command.offset);
    const pending = this.pendingCommits.get(key);
    if (!pending) return;
    this.log.warn(
      `KafkaActor: nack for ${key}${command.reason ? ` (${command.reason})` : ''} — `
      + `re-delivery will happen on next rebalance`,
    );
    pending.fail(new Error(command.reason ?? 'KafkaActor: nack from handler'));
    this.pendingCommits.delete(key);
  }

Acceptance

• A deadLetter block exists in BrokerCommonOptionsType with maxAttempts, an action and a target, resolvable from HOCON like every other broker option.
• AMQP declares a DLX/DLQ when configured and nacks with requeue: false at the cap, so a poison message cannot hot-loop.
• max_deliver is settable on the JetStream consumer and is set from maxAttempts.
• Kafka counts attempts per record and stops re-delivering at the cap.
• Redis Streams acks and dead-letters an entry past the cap (depends on [Feature] RedisStreams PEL-reclaim (XAUTOCLAIM/XCLAIM) command #462).
• Delivery-attempt counts are visible — a metric per broker actor, and the count carried on the inbound message type.
• Docs (EN + DE) explain the poison-message model per protocol.

Verification status

Found in the ten-lens production-readiness review of 2026-08-05 (v0.13.0) and re-verified before filing: confirmed by reading. No broker driver is installed in this tree and the review does not start brokers, so the AMQP requeue loop was not observed live; every element of the finding is a textual absence checkable in the source — the requeue ?? true default, the assertQueue call with no dead-letter arguments, the ConsumerAddConfig type with no max_deliver member, and the Redis comment stating that nothing reclaims the PEL. The one correction to the original finding: JetStream does expose term() per message (src/io/broker/JetStreamActor.ts:632), so an application can terminate a single poison message by hand; what is missing is the automatic cap, not the primitive.

Adjacent issues: #462 (RedisStreams PEL reclaim) is a prerequisite for the Redis leg and is not a duplicate. #433 (persistent DLQ with inspection and replay) is the destination this would route to. #222 (poison-pill quarantine after N failures on the same message digest) is the mailbox-level analogue and shares the counting idea but not the protocol integration. #650/#875 are the projection-side failure strategy — same problem shape, different subsystem. #670 (unify the broker subscribe/unsubscribe vocabulary) is the sibling consolidation and should probably land alongside this one.

Part of the production-readiness review batch — tracked in #913.