feat: batched outbox forwarding for combined message channels (Enterprise)

[dgafka]

Why is this change proposed?

Why

An outbox in front of a broker channel relays one message per poll cycle, so a single consumer run barely moves a busy queue, and a deep backlog makes each cycle progressively slower. This change turns the outbox into a true relay: an opt-in publishing process drains messages in batches straight from the database and pushes them forward in bulk, without consuming them through a message channel or deserializing payloads. Relaying 10 000 messages into a durable Dbal target drops from ~82s to ~1s (and to ~0.35s as a single in-memory-bound batch). Delivery stays at-least-once: a failed delivery is released for redelivery without duplicating already-delivered messages, and connection failures roll the whole cycle back for a clean retry.

Resulting flow

graph TD
    A[Polling tick of batch publishing endpoint] --> B[Claim up to batch size rows per outbox channel<br/>FOR UPDATE SKIP LOCKED on PostgreSQL, claim markers elsewhere]
    B --> C[Group rows by routing slip target, payloads stay serialized]
    C --> D{Target supports<br/>high-throughput publishing?}
    D -- yes --> E[Single BatchMessage per target, one multi-row insert]
    D -- no --> F[Per-message publish]
    E --> G[Delete delivered rows, commit per-channel transaction]
    F --> G
    G --> H{Delivery failed?}
    H -- release/ignore strategy --> I[Release only failed rows for redelivery, or drop on ignore]
    H -- stop or connection failure --> J[Rollback restores claims instantly, clean retry]

Loading

Out of scope

Batch forwarding for non-Dbal outbox sources — configuring it for any other channel fails at compile time.

Example

#[ServiceContext]
public function channels(): array
{
    return [
        OutboxForwardingMessageChannel::create('orders', DbalBackedMessageChannelBuilder::create('outbox'), 'orderProcessing')
            ->withMaxForwardingBatchSize(100),
        AmqpBackedMessageChannelBuilder::create('orderProcessing')
            ->withHighThroughputPublishing(),
    ];
}

OutboxForwardingMessageChannel extends CombinedMessageChannel and is the whole definition of the relay: exactly one Dbal backed source (the builder may be embedded, registering the outbox channel along the way and making a non database source unrepresentable) and one target channel. Reusing the outbox inside a plain CombinedMessageChannel, pointing an endpoint or output channel at it, or configuring it without a Dbal source all fail at compile time; forwarding channels sharing one outbox must agree on batch size, endpoint id and failure strategy.

One process can relay multiple outboxes — including ones in different databases — by sharing an endpoint id: OutboxForwardingMessageChannel::create('orders', 'ordersOutbox', 'ordersTarget')->withEndpointId('outboxPublisher') for each flow, then running the outboxPublisher consumer.

With multi-tenancy each tenant already owns an outbox in its own database; the publisher reuses polling consumer tenant propagation, so every run drains the next tenant's outbox in round robin and forwards to that tenant's target — no extra configuration beyond the standard MultiTenantConfiguration.

Failure handling per strategy

The strategy applied to failed deliveries is configured on the forwarding itself; when not set, it inherits from the outbox channel (default RESEND):

OutboxForwardingMessageChannel::create('orders', 'outbox', 'orderProcessing')
    ->withFinalFailureStrategy(FinalFailureStrategy::STOP);

Strategy	Behaviour when delivery of a message fails
RESEND (default) / RELEASE	Only the failed rows return to the outbox marked as redelivered and are retried on the next run; successfully delivered rows of the same batch stay delivered — no duplicates
IGNORE	Failed rows are dropped from the outbox; the rest of the batch stays delivered
STOP	The exception propagates and the publisher stops; the batch transaction rolls back, so every row returns to the outbox — messages already handed to the target within that batch may be delivered again after restart (at-least-once)
any + connection failure	The cycle always rolls back and claims restore instantly regardless of the configured strategy, so a restarted process retries the full batch cleanly

The error channel is not involved in relay failures — the outbox itself is the retry store, so failed messages never leave the delivery guarantee for a dead-end.

Benchmark results

Draining a 10 000 message outbox (PostgreSQL source, warmed consumer, phpbench mode; all broker targets receive whole batches via high-throughput publishing):

Scenario	Total time	Per message	Speedup
Message by message (without licence)	76.6 s	~7.7 ms	1×

Scenario	Total time	Per message	Speedup
Batched, size 100 → in-memory target	0.83 s	~0.083 ms	~92×
Batched, single batch of 10 000 → in-memory target	0.44 s	~0.044 ms	~174×
Batched, size 100 → Dbal target	0.91 s	~0.091 ms	~84×
Batched, size 100 → RabbitMQ target	0.87 s	~0.087 ms	~88×
Batched, size 100 → Kafka target	0.58 s	~0.058 ms	~132×
Batched, size 100 → Redis target	0.59 s	~0.059 ms	~129×
Batched, size 100 → SQS target	3.01 s	~0.30 ms	~25×

Every broker receives 10 000 durable messages in under a second except SQS, which is bound by its API's 10-message batch limit (~1000 HTTP round trips). The message-by-message path also degrades as the backlog deepens (per-message cost grew from ~2.4 ms at 200 messages to ~7.7 ms at 10 000), while the batched path improves with volume as fixed per-run overhead amortizes. Batch size is the throughput knob: one large batch outperforms a hundred 100-row cycles at the cost of holding the whole batch in memory within one transaction.

Description of Changes

• New opt-in OutboxForwardingMessageChannel (Enterprise) — a Combined Message Channel of exactly one Dbal backed outbox and one target — replaces the outbox channel's consumer with a standalone polling endpoint executing SQL directly: claim, group by routing-slip target, publish, delete/release — all inside an explicit per-channel transaction; rows relay in wire format, so serialization happens once on the producing side and format conversion belongs to the target channel
• Compile-time safety: outbox forwarding requires an Enterprise licence; a non-Dbal or unknown source, reuse of the outbox inside a plain Combined Message Channel, usage as execution or output channel, and conflicting settings between flows sharing an outbox all fail bootstrap
• Failure handling per configurable final failure strategy (withFinalFailureStrategy on the forwarding configuration, inheriting from the outbox channel when not set): release only failed rows, ignore drops them, stop and connection failures roll back the transaction so claims restore instantly; crashed processes' claims are swept for redelivery
• Channel-level withAsyncPublishing() renamed to withHighThroughputPublishing() across DBAL, AMQP, Kafka, Redis and SQS channel builders; finalFailureStrategy is now correctly passed to DBAL, SQS and Redis inbound message converters (pre-existing gap, AMQP already did)
• Cross-cutting performance: header mapping short-circuits scalars and match-all mappings, batch entries in wire format skip intermediate message construction on the outbound path, and the Dbal batch insert binds parameters directly without the doctrine type registry
• Multi-tenant setups publish in round robin across tenant outboxes per run, routing each batch to the drained tenant's target; the publishing endpoint carries WithoutDatabaseTransaction, now honoured consistently by the transaction, object manager and deduplication interceptors (also removing a per-tick deduplication insert every relay paid)
• Warmed 10k-message benchmark comparing message-by-message against batched draining, including batch-size and durable-target variants

Pull Request Contribution Terms

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