MJ Convento

/

Messenger, and the queue we should have built two years earlier

1,931 words, about 9 min read

The table was called pending_jobs and when I finally dropped it on 3 August it had 1,184,902 rows, of which 41 were actually pending. The rest were finished rows nobody had ever deleted, because the cleanup was a separate cron that had been failing silently since a MySQL user permission change in March. That is the honest summary of five years of background work on this platform: a table, a cron, and a cleanup cron that was broken.

We replaced it with Symfony Messenger over about three weeks in July. The booking platform runs Symfony 4.4 LTS, the internal ops app went to 5.1 in June, and both now push their background work through the same shape of code. I want to write down what the migration was actually like, including the parts where the cron table was not as stupid as it looks.

What the cron table actually was

Every minute, a command woke up and did roughly this:

SELECT id, type, payload, tries
FROM pending_jobs
WHERE status = 'pending' AND run_after <= NOW()
ORDER BY id
LIMIT 50;

Then it marked those 50 rows as running, looped over them, and called a handler chosen by a switch on type. It worked. It ran for five years. What it did not have was every property that makes a queue a queue.

There was no locking, so when the platform went from one worker box to three in 2018 it started double-sending partner notifications. The fix at the time was a SELECT ... FOR UPDATE wrapped around the claim, which turned the jobs table into a lock convoy every minute at exactly the moment the booking traffic peaked. There was no backoff: a failing job retried every minute forever, so one partner endpoint returning 503 for an hour produced sixty attempts and sixty log lines per job. There was no dead letter anywhere, so a permanently broken job sat in the table burning a slot. And because the dispatcher was a cron, the floor on latency was 60 seconds. A customer clicking “resend label” waited up to a minute for something that takes 300 ms.

The tries column is the part that stings. Two of the three things a queue needs were already in that table, written badly: retry counting and a delay. The third one, what to do when tries runs out, was never written at all.

The message and the handler

Messenger’s shape is two classes. A message, which is a plain object with no interface and no base class, and a handler, which is a class with an __invoke() method type hinted with the message. On PHP 7.4 the message gets typed properties and is about as small as an object can be:

namespace AppMessage;

final class GenerateWaybillPdf
{
    private int $waybillId;
    private string $requestedBy;

    public function __construct(int $waybillId, string $requestedBy)
    {
        $this->waybillId = $waybillId;
        $this->requestedBy = $requestedBy;
    }

    public function getWaybillId(): int
    {
        return $this->waybillId;
    }

    public function getRequestedBy(): string
    {
        return $this->requestedBy;
    }
}

The handler implements MessageHandlerInterface, which is an empty marker interface. That is all of it, and it is worth knowing that is all of it: there is no annotation, no tag to write by hand, no attribute. Autoconfiguration sees the interface, reads the type hint on __invoke(), and wires the handler to the message class. On PHP 7.4 there could not be an attribute for this anyway, so the marker interface is not a workaround, it is the mechanism. php bin/console debug:messenger prints what it decided.

namespace AppMessageHandler;

use AppMessageGenerateWaybillPdf;
use AppRepositoryWaybillRepository;
use AppWaybillPdfRenderer;
use DoctrineORMEntityManagerInterface;
use PsrLogLoggerInterface;
use SymfonyComponentMessengerExceptionUnrecoverableMessageHandlingException;
use SymfonyComponentMessengerHandlerMessageHandlerInterface;

final class GenerateWaybillPdfHandler implements MessageHandlerInterface
{
    private WaybillRepository $waybills;
    private PdfRenderer $renderer;
    private EntityManagerInterface $em;
    private LoggerInterface $logger;

    public function __construct(
        WaybillRepository $waybills,
        PdfRenderer $renderer,
        EntityManagerInterface $em,
        LoggerInterface $logger
    ) {
        $this->waybills = $waybills;
        $this->renderer = $renderer;
        $this->em = $em;
        $this->logger = $logger;
    }

    public function __invoke(GenerateWaybillPdf $message): void
    {
        $waybill = $this->waybills->find($message->getWaybillId());

        if (null === $waybill) {
            throw new UnrecoverableMessageHandlingException(
                sprintf('Waybill %d no longer exists', $message->getWaybillId())
            );
        }

        if (null !== $waybill->getPdfKey()) {
            $this->logger->info('waybill pdf already rendered, skipping', [
                'waybill' => $waybill->getId(),
            ]);

            return;
        }

        $waybill->setPdfKey($this->renderer->renderAndStore($waybill));
        $this->em->flush();
    }
}

Two things in there are the whole lesson of the migration and I will come back to both: the early return when the work is already done, and UnrecoverableMessageHandlingException, which tells the worker not to retry.

Transports, and why we started with Doctrine

A transport is a DSN. The Messenger documentation hands you doctrine://default, amqp://guest:guest@localhost:5672/%2f/messages and a Redis form, and the Flex recipe puts them in .env commented out. We started on Doctrine deliberately, because it introduced no new infrastructure on day one and let us argue about message design instead of about RabbitMQ. The Doctrine transport creates a messenger_messages table by itself unless you set auto_setup: false, takes a queue_name option so several logical transports can share one table, and has a redeliver_timeout that defaults to 3600 seconds.

framework:
    messenger:
        failure_transport: failed

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high
                retry_strategy:
                    max_retries: 4
                    delay: 2000
                    multiplier: 4
                    max_delay: 120000
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: default
            failed: 'doctrine://default?queue_name=failed'

        routing:
            'AppMessageNotifyPartnerStatusChange': async_priority_high
            'AppMessageGenerateWaybillPdf': async

        buses:
            messenger.bus.default:
                middleware:
                    - doctrine_ping_connection
                    - doctrine_close_connection

Yes, that is a queue in MySQL 5.7, and the thing I was most worried about turned out fine at our volume: about 34,000 messages a day, peak 900 an hour. The consumer polls, so you trade a little database load for not running a broker. We moved the high priority transport to AMQP in the third week anyway, once the partner notification volume made the polling interval visible, and the only change was the DSN. That portability is the strongest argument for the component. Note that in 5.1 the transports were split into separate packages, so the ops app needed symfony/doctrine-messenger and symfony/amqp-messenger in composer.json where 4.4 had them in the component.

One transport fact that is easy to skip and expensive to learn late: by default, messages are serialized with PHP’s native serialize() and unserialize(). That means a queued message is a serialized instance of your class. Rename the class, or remove a property, while messages are sitting in the transport, and those messages will not decode after the deploy. We renamed one message class in week two, deployed, and got a handful of undecodable rows. Since then, message classes are treated as a wire format: additive changes only, and drain before renaming.

Retries mean your handlers have to be idempotent

Messenger retries on any uncaught exception. The default retry_strategy is three attempts with a 1000 ms delay and a multiplier of 2, so one second, then two, then four, and max_delay of 0 means no ceiling. After the retries are exhausted the message is discarded, unless you configure a failure_transport, in which case it lands there and you get real commands:

php bin/console messenger:failed:show
php bin/console messenger:failed:show 4471 -vv
php bin/console messenger:failed:retry 4471 4472 --force
php bin/console messenger:failed:remove 4471

The first time I ran messenger:failed:show in production I felt slightly sick. Nineteen messages, with stack traces, four of them from a handler that had been quietly eating its own exception in the cron version and reporting success. Five years of that table, and nobody had ever had a list of what failed and why.

Retries are also the thing that forces a change in how you write the work itself. A retry will happen. Not might. Between the retry strategy, a worker killed mid-handle by a deploy, and the Doctrine transport’s redeliver_timeout handing a stuck message back after an hour, every handler has to assume it may run twice on the same message. That is why the handler above checks getPdfKey() before rendering, and why the two handlers that send things to partners now write a row keyed on the message identity before the outbound call and check for it on entry. For genuinely permanent failures, throwing UnrecoverableMessageHandlingException stops the retry loop immediately, which is the right answer for “this entity was deleted” and the wrong answer for anything network shaped. On 5.1 there is also a RecoverableMessageHandlingException to force a retry; on 4.4 we do not have it, so the ops app uses it and the platform does not.

Long-running PHP workers leak, so we do not let them live

This is the part nobody warns you about when they tell you to stop using cron. PHP’s request lifecycle cleans up after you. A worker has no requests. Doctrine’s entity manager accumulates an identity map, our PDF renderer holds buffers, and a worker started on Monday is not the same process by Thursday. The documentation says it plainly: do not let workers run forever.

So the worker is deliberately mortal. messenger:consume takes --limit, --memory-limit and --time-limit, and Supervisor restarts what exits:

[program:messenger-consume]
command=php /var/www/app/bin/console messenger:consume async_priority_high async --time-limit=3600 --memory-limit=192M
user=www-data
numprocs=4
startsecs=0
autostart=true
autorestart=true
process_name=%(program_name)s_%(process_num)02d

Passing two transports in that order means the worker drains high priority messages first and only looks at async when high is empty. Four processes per box on two boxes has been enough; our p95 from dispatch to handled is now 1.9 seconds against the old floor of 60. On deploy, the release script runs messenger:stop-workers, which signals each worker to finish its current message and exit so Supervisor starts fresh ones on the new code. That command works through the app cache pool, so if your cache adapter is per-process the signal goes nowhere, which cost me an afternoon of wondering why workers were running last week’s handlers. The doctrine_ping_connection middleware handles the other classic worker problem, a MySQL connection that timed out while the queue was empty overnight.

The honest version of why this did not happen in 2018, when the component first appeared: Messenger arrived in Symfony 4.1 marked experimental, and the changelog for 4.3 is a wall of breaking changes, including three new methods on ReceiverInterface and a rewritten error handling model. The component only stopped being marked experimental in 4.4. Waiting was defensible. Waiting until the cleanup cron had been broken for five months was not, and the 1.18 million rows were the bill.

What is still open: the failure transport is a MySQL table nobody looks at unless I look at it, so the next thing I write is an alert on its row count. And I have not decided what to do about ordering. Messenger gives you none, and there are two message types where I think we have been getting away with luck.

Sources