feat: per-handler error channel and delayed retry for async endpoints (fixes #659)

[dgafka]

Why is this change proposed?

Originally reported in #659: configuring withDefaultErrorChannel(...) on a #[KafkaConsumer] (or any inbound channel adapter) crashed at runtime with MessageHeaderDoesNotExistsException because the framework assumed every polled message carried POLLED_CHANNEL_NAME, which inbound channel adapters don't set. While fixing that, two related gaps surfaced: there was no way to scope an Error Channel or a delayed-retry policy to a single async handler when multiple handlers shared a transport, and dead-letter replay couldn't route messages back to inbound-adapter handlers because the framework had no header carrying the actual request channel.

This PR fixes the original crash and turns the existing asynchronousExecution: [...] parameter on #[Asynchronous] into a real per-handler error-handling story — #[ErrorChannel] and the new #[DelayedRetry] both work there, with full Dead Letter replay support for inbound channel adapters.

Usage examples

Per-handler Error Channel for handlers sharing one transport

final class OrderProjectors
{
    #[Asynchronous('orders', asynchronousExecution: [new ErrorChannel('paymentsErrors')])]
    #[CommandHandler('order.process_payment', 'paymentsHandler')]
    public function processPayment(ProcessPayment $command): void { /* ... */ }

    #[Asynchronous('orders', asynchronousExecution: [new ErrorChannel('shippingErrors')])]
    #[CommandHandler('order.ship', 'shippingHandler')]
    public function ship(ShipOrder $command): void { /* ... */ }
}

#[DelayedRetry] — retries with backoff and dead letter, per handler

#[Asynchronous('orders', asynchronousExecution: [
    new DelayedRetry(
        initialDelayMs: 1000,
        multiplier: 2,
        maxAttempts: 3,
        deadLetterChannel: 'dbal_dead_letter',
    ),
])]
#[CommandHandler('order.charge', 'chargeHandler')]
public function charge(ChargeOrder $command): void { /* ... */ }

#[DelayedRetry] on a gateway

#[DelayedRetry(initialDelayMs: 1000, multiplier: 2, maxAttempts: 3, deadLetterChannel: 'dbal_dead_letter')]
interface ResilientCommandBus extends CommandBus
{
}

Combining attributes (e.g. skip transactions for a notification handler)

#[Asynchronous('orders', asynchronousExecution: [
    new WithoutDatabaseTransaction(),
    new ErrorChannel('emailDeliveryErrors'),
])]
#[CommandHandler('order.send_confirmation_email', 'emailHandler')]
public function sendConfirmation(SendConfirmation $command): void { /* ... */ }

Inbound Channel Adapter (#[KafkaConsumer]) with #[ErrorChannel]

Failures are forwarded to the configured Error Channel — typically a Dead Letter for later replay:

#[ErrorChannel('dbal_dead_letter')]
#[KafkaConsumer(endpointId: 'orderConsumer', topics: ['orders'])]
public function handle(Order $payload): void { /* ... */ }

Inbound Channel Adapter (#[KafkaConsumer]) with #[InstantRetry] + #[ErrorChannel]

In-process retries before forwarding the failure to the Error Channel:

#[InstantRetry(retryTimes: 2)]
#[ErrorChannel('dbal_dead_letter')]
#[KafkaConsumer(endpointId: 'orderConsumer', topics: ['orders'])]
public function handle(Order $payload): void { /* ... */ }

Inbound Channel Adapter (#[Scheduled]) with #[ErrorChannel]

#[ErrorChannel('schedulerErrors')]
#[Scheduled(requestChannelName: 'pollPayments', endpointId: 'paymentsPoller')]
#[Poller(fixedRateInMilliseconds: 60000)]
public function poll(): PaymentsBatch { /* ... */ }

Inbound Channel Adapter (#[Scheduled]) with #[InstantRetry] + #[ErrorChannel]

#[InstantRetry(retryTimes: 3)]
#[ErrorChannel('schedulerErrors')]
#[Scheduled(requestChannelName: 'pollPayments', endpointId: 'paymentsPoller')]
#[Poller(fixedRateInMilliseconds: 60000)]
public function poll(): PaymentsBatch { /* ... */ }

⚠️ #[DelayedRetry] is not supported on Inbound Channel Adapters. They consume from external systems (Kafka, AMQP, scheduled tasks) and have no source Message Channel to reschedule into.

Use case scenarios

1. Different SLAs per handler on a shared queue — payment handlers route to a Dead Letter for manual review, while shipping handlers retry with backoff. Both consume from the same orders channel; no need to split queues for different policies.
2. Kafka consumer with default error channel — withDefaultErrorChannel(DbalDeadLetterBuilder::STORE_CHANNEL) now works for #[KafkaConsumer] (this was the original crash). Failed messages are stored in the DBAL Dead Letter and DeadLetterGateway::replyAll() correctly replays them back to the same Kafka consumer endpoint.
3. Resilient gateways — extend CommandBus with #[DelayedRetry(...)] once and reuse it across an application: callers transparently get retries + dead-letter without each handler having to implement try/catch.

Failure flow with #[DelayedRetry]

sequenceDiagram
    participant User
    participant CommandBus
    participant AsyncChannel as Async Channel
    participant Consumer as Polling Consumer
    participant Handler
    participant Retry as DelayedRetryErrorHandler
    participant DLQ as Dead Letter

    User->>CommandBus: send(Command)
    CommandBus->>AsyncChannel: enqueue
    Consumer->>AsyncChannel: poll
    AsyncChannel-->>Consumer: Message
    Consumer->>Handler: invoke
    Handler--xConsumer: throws
    Consumer->>Retry: failed Message
    alt retries remaining
        Retry->>AsyncChannel: reschedule with delay
    else retries exhausted
        Retry->>DLQ: store
        User->>DLQ: replyAll()
        DLQ->>Handler: replay (via INBOUND_REQUEST_CHANNEL or POLLED_CHANNEL_NAME)
    end

Loading

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.