New in Symfony 8.2 Messenger Outbox and Claim Check

Transactional Outbox

Contributed by Nicolas Grekas in #65901

Saving changes in the database and sending a message to a broker (AMQP, Amazon SQS, etc.) is not atomic. If the transaction is rolled back, the message was already sent; if the broker is down, the changes are saved without the message.

The usual workaround is to route the message to the Doctrine transport, so the insert joins the transaction. The problem is that the database becomes the queue that every worker polls. The transactional outbox pattern stores the message in the database inside the transaction too, but then a dedicated worker (called the relay) reads those stored messages and sends them to the real broker. Your consumers keep reading from the broker, so they scale as usual.

In Symfony 8.2, this is the new outbox transport option, whose value is the name of another transport where messages are stored before being forwarded:

1
2
3
4
5
6
7
8
9
10

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            orders:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                # messages sent to "orders" are stored in "db_outbox" first
                outbox: db_outbox

            db_outbox: 'doctrine://default?queue_name=outbox'

Then, run a worker for each transport:

1
2
3
4
5

# the relay: forwards the stored messages to "orders" (it never runs handlers)
$ php bin/console messenger:consume db_outbox

# handles the messages sent to "orders" as usual
$ php bin/console messenger:consume orders

Things to keep in mind about the outbox:

• it must be a Doctrine transport using the same database connection where your application saves its own data;
• the delay of a message is applied while it waits in the outbox, so later it's forwarded without delay;
• when forwarding fails, the relay applies the retry strategy and the failure transport of the outbox transport. When handling fails, the retries go to the target transport directly, because they already went through the outbox;
• the order in which messages are forwarded is not guaranteed.

Claim Check

Contributed by Yanick Witschi in #65641

Brokers put a limit on the size of the messages they accept. Instead of making your messages smaller, the claim check pattern stores the big ones somewhere else and sends only a small reference to them.

Symfony 8.2 adds the claim_check transport option, which takes the cache pool used to store messages and the maximum size (in bytes) of the messages sent to the transport. It's configured per transport because every broker has a different limit:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# config/packages/messenger.yaml
framework:
    cache:
        pools:
            # a dedicated pool, reachable by producers and consumers
            messenger.claim_check.cache:
                adapter: cache.adapter.redis
                # claims are only removed when they expire
                default_lifetime: 604_800 # 7 days

    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                claim_check:
                    cache_pool: messenger.claim_check.cache
                    # keep some margin below the limit of your broker
                    max_size: 200_000

Messages whose encoded size (body + headers) stays below max_size are sent as usual. Bigger ones are stored in the cache pool and the transport only carries a small reference with a random identifier and a checksum. When the worker consumes that reference, Symfony loads the message back from the pool, verifies its checksum and handles it as if it had traveled through the transport. Any PSR-6 cache pool works, so you can use Redis, Valkey, Memcached, PDO, Doctrine DBAL and the other backends supported by the Cache component.

Keep in mind that the pool is now part of the delivery of your messages:

• use a pool that every producer and every consumer can reach, and don't share it with the rest of the application, because clearing it discards pending messages;
• the pool must define a default_lifetime, and it must outlive the messages (including their delay, all their retries and the wait in the failure transport);
• consuming a message whose claim expired fails with a ClaimCheckNotFoundException wrapped in a MessageDecodingFailedException, so it follows the usual retry and failure transport path.