Support delay in Mailer when using Messenger

[bobvandevijver]

I have an action which imports a list of accounts, each of which will receive an e-mail that the account has been created. Unfortunately, I'm still using SMTP as transport (which is my only option), which means I need to throttle my e-mails. However, I'm using the Mailer component with Messenger (configured with doctrine transport), which means I should be able to something with the scheduled messages.

As far I can see, I have two options for that:

• Limit the amount of consumed messages by the worker, but also make sure it is not restarted automatically directly after each exit. This would require a more complex worker configuration outside of Symfony to be configured during deployment.
• Delay the message when putting it on the message bus using the DelayStamp

The latter would have my preference, but unfortunately this is not possible directly with the Mailer as I cannot supply the stamp when calling send. As workaround I can create the SendEmailMessage myself, but as it is marked as internal I obviously really shouldn't.

That's why I propose to add stamp support to the MailerInterface, with default null. When provided while the bus is not configured, it should log a warning (or throw an exception, to be discussed). This way you have more control over the created message, and there are probably some other scenarios where this might be beneficial as well.

Example

/**
 * @throws TransportExceptionInterface
 * @param StampInterface[] $stamps
 */
public function send(RawMessage $message, Envelope $envelope = null, array $stamps = null): void;

Obviously I can make a PR if this is something that is considered to be valuable for the Symfony ecosystem, but I'm looking for input/verification first 😄

[arjanfrans]

I recently ran into the problem that we are getting rate limited(using AWS SES as a provider). The possibility of adding delay stamps to messages could help us work around that issue.

Perhaps an addition to the configuration to throttle the amount of mails being sent in a given time span.

[bobvandevijver]

Maybe we could use the new RateLimiter component to implement this for configurable message types?

[fabpot]

@arjanfrans For AWS SES, it will be built-in soon I suppose (/cc @jderusse). And we can do the same for all HTTP based transports. For all transports extending AbstractTransport (including SMTP), we do have a build-in rate limiting feature (see AbstractTransport::setMaxPerSecond()), but it does not work well when using messenger. We should probably move to use the new rate limiter component instead. Anyone willing to give it a try?

[fabpot]

@wouterj Looks like the new rate limiter component can be useful in many ways :)

[wouterj]

I haven't read this thread carefully, but I've talked about this with @jderusse yesterday on Slack. There is a difference between adhering to an external rate limiter (which requires rate throttling) and maintaining your own rate limit (which is the use-case for the RateLimiter component). I don't think using your own rate limit to cope with an external one is necessarily the best option. Rate throttling requires techniques like exponential backoff and other backoff algorithms, not "rate limiting".

It might be interesting for the Mailer, Messenger and HttpClient components to create a "rate throttler" in the RateLimiter component (e.g. a generic retry class with backoff abstraction). A rate limiter can function as one possible back-off implementation and e.g. exponential backoff can as well.

please note that I'm enjoying life with 0 open Symfony PRs until the 5.2 stable release - and I'm far from a rate limiting/throttling expert -, so anyone should feel welcome to start working on this topic (there seems to be a lot of interest).

[bobvandevijver]

Although that is a valid argument, to solve my initial problem having just a rate limiter would suffice as I know the actual limit of the mail service I'm using.

If you want to effectively use throttling, you should have the knowledge that your message was rejected due to throttling of your mail provider. And I believe that might be quite hard to consistently detect: maybe the HTTP providers do have a dedicated response for it (maybe even with the actual limit it), but SMTP doesn't as far a I know (correct me if I'm wrong).

That's why I believe both (limiting and throttling) can be effective tools for the Messenger component. The latter is already available in the retry_strategy when combined with the multiplier option for separate messages (even though it is very basic).

Adding basic rate limiting on transport level would be valuable anyways in my opinion, not only for mail or something else with an external service. If I find time, I will try to look into adding it.

[jderusse]

We have several tools to solve this issue

By prevent failure:

• leverage the new RateLimiter component

this requires knowing how the server is configured (bucket size, fill rate, ...)

By retrying failure

• throw a RecoverableException and let symfony/messenger retry the message (beware @bobvandevijver by default, messenger will retry failing message 3 times and stops, use RecoverableException to avoid that)
• leverage the new RetryableHttpClient but is not (yet?) compatible with AWS: sadly returns a 400 response with a custom payload (while the Response's body is not sent to the RetryDecider). We may need to improve the RetryableClient for such case

note: I solved this issue for a personal project with a listener and redis

<?php

declare(strict_types=1);

namespace App\EventListener;

use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Mailer\Event\MessageEvent;

class ThrottlerMailerListener implements EventSubscriberInterface
{
    private $throttlerLimit;
    private $redis;

    public function __construct(int $throttlerLimit, \Redis $redis)
    {
        $this->throttlerLimit = $throttlerLimit;
        $this->redis = $redis;
    }

    public function __invoke(): void
    {
        while (!$this->canSend()) {
            usleep(100000);
        }
    }

    public static function getSubscribedEvents()
    {
        return [
            MessageEvent::class => '__invoke',
        ];
    }

    private function canSend()
    {
        $t = (string) time();
        $responses = $this->redis->multi()
            ->incr($t)
            ->expire($t, 59) // 59 here does not matter, it's just to clean things
            ->exec()
        ;

        return $responses[0] <= $this->throttlerLimit;
    }
}

[Aerendir]

I came on this thread because I was searching for a solution to implement a rate limiting in Messenger.

My use case is the calling of an external API that limits the amount of calls I can send per minute.

It is very similar to the rate limiting imposed by AWS SES, but for an HTTP endpoint.

If we put together the two use cases, we have

• the same need (verifying the limit is not yet reached),
• handled in the same way (Messenger Component)
• BUT applied to two different “media”: HTTP (HttpClient Component) and SMTP (Mailer Component).

Given that the touch point between the two is Messenger, I will speak about it (because the problem can also be solved, as seen in previous comments, at HttpClient/Mailer Component level).

I found basically two ways of avoiding throttling due to a reached limit:

1. Implement the logic directly in the handler;
2. Implementing a custom middleware.

Implementing the logic directly in the Handler

It is really simple:

use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;

final class TestHandler implements MessageHandlerInterface
{
    private MessageBusInterface $messageBus;

    public function __construct(MessageBusInterface $messageBus)
    {
        $this->messageBus = $messageBus;
    }

    public function __invoke(TestMessage $message): bool
    {
        // ... Do what you like to check the current status of your quota
        if ($quotaIsReached) {
            $rescheduleMessage = new TestMessage(...);
            $this->messageBus->dispatch($rescheduleEnvelope,  [new DelayStamp($quota['refill_in_milliseconds'])]);
            
            return true;
        }

        // ... Send the message
        
        return true;
    }
}

This is a good starting point but has a drawback: what if we need the same checking logic in two different handlers?

We have to copy and paste some lines of code in the second handler, duplicating the code and we need to change something, we have to change it in two different handlers each time (and remember we have to!).

So, here come the second approach I found: the use of a middleware.

Implementing the logic in a Middleware

This is the solution I prefer as, once created, it is extremely usable.

This is the middleware:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;

final class TestMiddleware implements MiddlewareInterface
{
    private MessageBusInterface $messageBus;

    public function __construct(MessageBusInterface $messageBus)
    {
        $this->messageBus = $messageBus;
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        /** @var MyRateLimitedStamp|null $rateLimitedStamp */
        $rateLimitedStamp = $envelope->last(MyRateLimitedStamp::class);
        if (null === $rateLimitedStamp) {
            return $stack->next()->handle($envelope, $stack);
        }

        $receivedStamp = $envelope->last(ReceivedStamp::class);
        if (null === $receivedStamp) {
            return $stack->next()->handle($envelope, $stack);
        }

        // ... Do your checking logic here

        if (false === $quotaIsReached) {
            return $stack->next()->handle($envelope, $stack);
        }

        // We reached the limit
        $rescheduleEnvelope = new TestMessage(...);
        $this->messageBus->dispatch($rescheduleEnvelope, [new DelayStamp($quota['refill_in_milliseconds']), new MyRateLimitedStamp()]);

        return $envelope;
    }
}

Then configure it:

# config/packages/messenger.yaml

framework:
    ...
        buses:
            messenger.bus.default:
                middleware:
                    - 'App\Messenger\Middleware\TestMiddleware'

Done: now you can be sure the api will not be called if the limit is reached (almost... because a race condition on milliseconds is anyway possible).

This last example can be improved using interfaces (not tested yet: I will do it tomorrow):

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        // Do the check using the interface
        if (false === $envelope->getMessage() instanceof AwsMessageInterface::class) {
            return $stack->next()->handle($envelope, $stack);
        }

        $receivedStamp = $envelope->last(ReceivedStamp::class);

        // ...

        // We don't need to set the stamp anymore
        $rescheduleEnvelope = new TestMessage(...);
        $this->messageBus->dispatch($rescheduleEnvelope, [new DelayStamp($quota['refill_in_milliseconds'])]);

This approach has the advantage that we:

• Don't have to add a new stamp each time we create the message
• We use a single interace to manage rate limiting and queues:

framework:
    messenger:
        ...
        transports:
            aws: '%env(MESSENGER_TRANSPORT_DSN)%'
        routing:
            # Route your messages to the transports
            'App\Messenger\Queues\AwsMessageInterface': aws

Hope this will help.

[carsonbot]

Thank you for this suggestion.
There has not been a lot of activity here for a while. Would you still like to see this feature?

[bobvandevijver]

Even with the nice suggestions posted already, I still believe it would have added value to be able to rate limit a certain messenger queue with some configuration.

[hracik]

Also looking for solution to limit "consume" for certain messages, but I see a problem with dispatching same message again.. problem is the order of messages is broken, the latest message in queue became newest, just because limit was reached at the moment. Better solution for my needs will be to keep current message and delay it. Something like soft-fail. Messenger should always retry soft-fail, but when normal fail occurs apply usual retry strategy.