Skip to content

Navigation Menu

Sign in
Appearance settings
Sign up
Appearance settings

[Messenger][WIP] Sending messages in batches - #62600

#62600
Open
JoppeDC wants to merge 4 commits into
symfony:8.2symfony/symfony:8.2from
JoppeDC:batch-messenger-pocJoppeDC/symfony:batch-messenger-pocCopy head branch name to clipboard
Open

[Messenger][WIP] Sending messages in batches#62600
JoppeDC wants to merge 4 commits into
symfony:8.2symfony/symfony:8.2from
JoppeDC:batch-messenger-pocJoppeDC/symfony:batch-messenger-pocCopy head branch name to clipboard

Conversation

@JoppeDC

@JoppeDC JoppeDC commented Dec 1, 2025

Copy link
Copy Markdown
Contributor
Q A
Branch? 8.1
Bug fix? no
New feature? yes
Deprecations? Possibly?
Issues Fix #60748
License MIT

Note: This PR is a proof of concept. It is in no way, shape or form supposed to be a final ready for production fix. This PR is supposed to get the conversation going. Please, provide feedback where needed, as this feature turned out to be more complex than i initially imagined 😀 This is also the reason the code is littered with comments

Batch Dispatch for Symfony Messenger

Allows dispatching multiple messages in a single batch, enabling transports to optimize delivery (e.g., bulk inserts to a queue). This allows you to optimize message sending (Eg. Single SQS request for 10 messages, where you pay per-request)

How it works

$bus->dispatchBatch([new OrderCreated(1), new OrderCreated(2), new OrderCreated(3)]);

Step by step:

  1. MessageBus::dispatchBatch() creates a BatchCollector and generates a unique batch ID
  2. Each message goes through the regular middleware chain individually, with a BatchStamp attached (contains batchId, index, size, and collector reference)
  3. SendMessageMiddleware detects the collector in BatchStamp, so it skips sending the message and instead adds it to the collector
  4. After all messages are processed, BatchCollector::flush() groups envelopes by transport and calls sendBatch() (or falls back to individual send() calls)
  5. Events are dispatched: SendMessageToTransportsEvent per message, MessageSentToTransportsEvent per sent message, BatchSentToTransportsEvent once for the whole batch

Notes

  1. Error handling - If one transport in a multi-transport batch fails, partial sends may have occurred. But that issue currently also exists for single messages
  2. Batch size limits - I have now blocked this on the collector level, to avoid partial flushes.

@carsonbot carsonbot added Status: Needs Review Feature Messenger RFC RFC = Request For Comments (proposals about features that you want to be discussed) labels Dec 1, 2025
@carsonbot carsonbot added this to the 8.1 milestone Dec 1, 2025
@carsonbot carsonbot changed the title [Messenger] [RFC] Sending messages in batches [Messenger] Sending messages in batches Dec 1, 2025
@JoppeDC
JoppeDC force-pushed the batch-messenger-poc branch 2 times, most recently from 59d8706 to 90f61db Compare December 1, 2025 16:35
@JoppeDC
JoppeDC force-pushed the batch-messenger-poc branch from 90f61db to f77ca1b Compare December 1, 2025 16:37
Some transports will have a max batch limit (eg. SQS)
Either the transport has to make sub batches, or the messageBus should fail when trying to send bigger batches
The issue with the transport making sub batches, is that you have the chance that half the batch fails and half succeeds.
eg.
dispatchBatch(20 messages)
- transport sends 10
- transport tries to send 10 again, but fails

now half the batch got sent, but we have no clue which ones.
IMO, best to block this.
private string $batchId,
private int $batchIndex,
private int $batchSize,
private ?BatchCollector $collector = null,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

non-nullable?

public function addPendingSend(string $alias, SenderInterface $sender, Envelope $envelope, int $index): void
{
$this->pending[$alias] ??= ['sender' => $sender, 'items' => []];
$this->pending[$alias]['items'][] = ['envelope' => $envelope, 'index' => $index];

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
$this->pending[$alias]['items'][] = ['envelope' => $envelope, 'index' => $index];
$this->pending[$alias]['items'][$index] = $envelope;

?

*
* @author Joppe De Cuyper <hello@joppe.dev>
*
* @internal

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

im fine with making it publicly available, for other maybe-implementions

otherwise we can argue whole BatchStamp is internal

@ro0NL

ro0NL commented Dec 1, 2025

Copy link
Copy Markdown
Contributor

it might be interesting to update DoctrineTransport as such

@JoppeDC

JoppeDC commented Dec 1, 2025

Copy link
Copy Markdown
Contributor Author

it might be interesting to update DoctrineTransport as such

Yeah that was the next thing i was looking at. This interface sounds nice and easy on paper, but already ran into trouble when implementing. Not all transports that support batching, also support returning the ids of each item in the badge. So TransportMessageIdStamp can't be promised depending on the transport used

@ro0NL

ro0NL commented Dec 1, 2025

Copy link
Copy Markdown
Contributor

So TransportMessageIdStamp can't be promised

it sounds generally reasonable, there's no current contract enforcing it either as well

it SHOULD be implemented if possible

@wouterj wouterj changed the title [Messenger] Sending messages in batches [WIP][Messenger] Sending messages in batches Dec 1, 2025
@carsonbot carsonbot changed the title [WIP][Messenger] Sending messages in batches [Messenger][WIP] Sending messages in batches Dec 1, 2025
@nicolas-grekas nicolas-grekas modified the milestones: 8.1, 8.2 May 6, 2026
@nicolas-grekas

Copy link
Copy Markdown
Member

Thanks for the RFC, and thanks for your patience: eight months without a core answer on a PR that explicitly asked for one is on us.

Yes to the concept. The need is real and cannot be met in userland without bypassing the bus, and the design solves the part I expected to sink it. Running the normal per-message middleware stack and deferring only the send keeps routing, stamps and envelopes behaving as usual, with no batch-aware middleware to invent.

Four things before it is worth polishing:

  • Implement SQS. It is the reason [Messenger] Batch send for SQS #60748 exists, and it is where the hard part lives, since SendMessageBatch reports success and failure per entry. Doctrine alone does not exercise that.
  • Chunk automatically instead of throwing. getMaxBatchSize() plus BatchSizeExceededException makes application code responsible for knowing the transport's limit, 10 for SQS and unbounded for Doctrine, so swapping transport breaks the caller. The collector should split the batch itself.
  • Decide what a partially failed batch returns. Same question from the other side, and SQS will force it.
  • Reconsider BatchStamp carrying a live BatchCollector. Nothing gets serialized since it is non-sendable, but a stamp holding a mutable service rather than a value object will draw fire in review.

Shrinking the public surface where you can would help as well. Six new classes and interfaces is a lot to commit to; if BatchSenderInterface plus the collector are enough and the rest can be internal or dropped, that is a much easier sell.

The limitation you already found yourself is fine: not every batching transport can return per-message ids, so TransportMessageIdStamp cannot be promised. Document it as expected rather than working around it.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Feature Messenger RFC RFC = Request For Comments (proposals about features that you want to be discussed) Status: Needs Work

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Messenger] Batch send for SQS

5 participants

Morty Proxy This is a proxified and sanitized view of the page, visit original site.