New in Symfony 8.2 Messenger Flows

Carrying Context Across a Flow

Contributed by Nicolas Grekas in #65899

When a handler dispatches a follow-up message, that message starts with an empty envelope. So data like a tenant ID or a trace ID often ends up as a property of every message in the flow, only to pass it along.

In Symfony 8.2, stamps that implement the new PropagatedStampInterface are copied automatically to every message dispatched while handling the message that carries them. In addition, handlers can now get stamps (or the whole Envelope) as arguments after the message:

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

use Symfony\Component\Messenger\Stamp\PropagatedStampInterface;

class TenantStamp implements PropagatedStampInterface
{
    public function __construct(
        public readonly string $id,
    ) {
    }
}

$bus->dispatch(new ImportCatalog($path), [new TenantStamp($tenant->id)]);

// the ImportCatalog handler dispatches ImportRow messages without dealing
// with tenants, but they still receive the TenantStamp
#[AsMessageHandler]
class ImportRowHandler
{
    public function __invoke(ImportRow $row, TenantStamp $tenant): void
    {
        $this->catalogs->for($tenant->id)->add($row);
    }
}

Identity Stamps

Contributed by Nicolas Grekas in #65899

When a flow involves many messages, debugging it requires knowing which messages belong to it and which message dispatched each one. Symfony 8.2 can add this information automatically with the new identity_stamps option:

1
2
3
4

# config/packages/messenger.yaml
framework:
    messenger:
        identity_stamps: true

When enabled, every message gets a MessageIdStamp with a unique ID that is kept across retries and replays. Messages dispatched by a handler also get a CausationStamp with the ID of the message being handled, and all the messages of the flow share the same CorrelationStamp (the following example uses short IDs for readability; real IDs are 32-character random strings):

1
2
3

ImportCatalog   id: a1   correlation: a1
├── ImportRow   id: b2   correlation: a1   causation: a1
└── ImportRow   id: c3   correlation: a1   causation: a1

As with any other stamp, handlers can get them as arguments (e.g. to add the correlation ID to your logs). The option is disabled by default because it changes the serialized envelopes. It's also required by the Messenger inbox.

Chaining Messages

Contributed by Nicolas Grekas in #65903

The new ChainStamp lets you define the sequence of a flow in a single place. Each message is dispatched only after the previous one has been handled by all its handlers:

1
2
3
4
5
6
7
8
9
10

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\ChainStamp;
use Symfony\Component\Messenger\Stamp\DelayStamp;

$bus->dispatch(ChainStamp::envelope(
    new ProcessPodcast($id),
    new OptimizePodcast($id),
    // pass an Envelope to add stamps to a step
    new Envelope(new ReleasePodcast($id), [new DelayStamp(3_600_000)]),
));

The remaining steps travel with the message inside the stamp, so a retried step continues the chain from where it stopped. A step without its own routing goes to the transport that the previous step was received from, so each step is retried independently. If a step fails, the chain stops.

Dispatching a Message When Another One Fails

Contributed by Nicolas Grekas in #65917

The new DispatchOnFailureStamp defines a message to dispatch when the message it's attached to fails permanently (after all its retries in a worker, or when the exception is thrown in a synchronous dispatch). Combined with chains, you can define compensation logic for the entire flow:

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

use Symfony\Component\Messenger\Stamp\ChainStamp;
use Symfony\Component\Messenger\Stamp\DispatchOnFailureStamp;
use Symfony\Component\Messenger\Stamp\ErrorDetailsStamp;
use Symfony\Component\Messenger\Stamp\FailedMessageStamp;

$bus->dispatch(
    ChainStamp::envelope(new ChargeCard($orderId), new ShipOrder($orderId))
        ->with(new DispatchOnFailureStamp(new CancelOrder($orderId)))
);

#[AsMessageHandler]
class CancelOrderHandler
{
    public function __invoke(CancelOrder $message, FailedMessageStamp $failed, ErrorDetailsStamp $error): void
    {
        // $failed->getMessage() is the ChargeCard or ShipOrder message that failed
    }
}

Each step of the chain passes the stamp to the next one, so CancelOrder is dispatched only once, whichever step fails. This stamp also replaces the failure transport: the failed message is only sent to the failure transport when CancelOrder can't be dispatched.

Retries and Failure Transport for Synchronous Messages

Contributed by Nicolas Grekas in #65902

Retries and the failure transport were only available to messages handled by workers. Messages handled with the sync:// transport (e.g. a payment webhook processed during the request) weren't retried, and the retry_strategy and failure_transport options were silently ignored.

In Symfony 8.2, the sync:// transport supports both features with the new retry and failure_transport options:

1
2
3
4
5
6
7
8
9
10
11
12

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            webhooks:
                dsn: 'sync://?retry=true&failure_transport=true'
                retry_strategy:
                    max_retries: 2
                failure_transport: failed

        routing:
            App\Webhook\PaymentReceived: webhooks

Failed messages are retried immediately, ignoring the delays of the retry strategy. If the message still fails, it's sent to the failure transport instead of throwing an exception, so you can inspect it with messenger:failed:show and retry it with messenger:failed:retry as usual.

Stamps in Scheduled Tasks

Contributed by Zoran Shefot Bogoevski in #66782

Adding stamps to the messages of scheduled tasks required a custom schedule provider. In Symfony 8.2, the #[AsCronTask] and #[AsPeriodicTask] attributes include a new stamps option. List the stamp class names and, when a stamp needs constructor arguments, map the class to those arguments:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Scheduler\Attribute\AsCronTask;

#[AsCronTask('0 2 * * *', transports: 'async', stamps: [
    DelayStamp::class => 5000,
    TenantStamp::class => ['id' => 'acme'],
])]
class GenerateDailyReport
{
    public function __invoke(): void
    {
        // ...
    }
}

Stamps used when sending messages (such as DelayStamp) only apply when the task goes through a transport (via the transports or use_messenger_routing options).