Background jobs

Some work does not have to happen while a customer waits for a page: handing a mail to the mail server, writing a large export, pushing stock to a marketplace. Thelia runs that work through Symfony Messenger. Code dispatches a message, a handler does the work, and the configuration of the shop decides whether the handler runs at once or later, in a worker.

This page explains the model. The configuration, the commands and a complete module example are in the background jobs reference; running the workers on a server is covered in Running the workers.

What the core ships

The core requires symfony/messenger, symfony/doctrine-messenger, doctrine/dbal and symfony/scheduler. It does not require the Doctrine ORM or DoctrineBundle: DBAL is only there to serve the queue kept in the shop database.

There is one bus, messenger.bus.default, and four transports:

Transport	What it carries	Where it lives
async	The jobs of the shop: mail delivery, the messages of modules	MESSENGER_TRANSPORT_DSN, synchronous when empty
async_heavy	Back-office exports and imports	Derived from MESSENGER_TRANSPORT_DSN, or MESSENGER_HEAVY_TRANSPORT_DSN
failed	The jobs that failed every attempt (the failure transport)	The shop database by default, whatever the queue is; MESSENGER_FAILURE_TRANSPORT_DSN overrides it
scheduler_thelia	The recurring tasks of the thelia schedule	Generated by the schedule, nothing to configure

Nothing waits by default

MESSENGER_TRANSPORT_DSN is shipped empty in .env. An empty value makes async and async_heavy fall back to sync://: every message is handled immediately, in the request or the command that dispatched it, exactly as if the handler had been called directly. Nothing is stored and nothing waits for a worker, so a shop that runs no worker loses no job. A failure in the handler surfaces in the code that dispatched the message, as any exception would.

The shop opts in to a real queue by naming one:

# .env.local
MESSENGER_TRANSPORT_DSN=doctrine://default

From then on, messages are stored and wait for messenger:consume. Setting the variable without running a worker leaves mails undelivered and exports and imports pending, so the two go together.

DSN	Queue
empty	No queue, every job runs at once
doctrine://default	Table messenger_messages of the shop database
redis://host:6379/messages	A Redis stream, once symfony/redis-messenger is installed
amqp://user:password@host:5672/%2f/messages	An AMQP broker such as RabbitMQ, once symfony/amqp-messenger is installed

The core only ships the Doctrine transport. A Redis queue needs composer require symfony/redis-messenger and the redis PHP extension; an AMQP queue needs composer require symfony/amqp-messenger and the amqp extension.

Choosing a transport

Start with doctrine://default. It needs nothing to install and no server to run: the queue is a table of the shop database, saved with its backups, and the failed jobs are kept in that database whatever the queue is. Workers on several servers can share it, since they all reach the shop database. The jobs of a shop (mails, back-office exports and imports, the messages of a few modules) are well within what a table can serve.

Move to another transport for one of these reasons:

• the volume: modules that dispatch many messages, a push per product combination on a large catalog for instance, keep the workers reading and writing the table all day. Redis or a broker takes that load off the database the shop serves its pages from,
• the hosting already runs a Redis server or a RabbitMQ broker, with its monitoring and its backups: use it rather than adding a queue to the database,
• the team already operates AMQP for other applications and wants the shop jobs in the same place.

Redis is the shorter step: one package, one PHP extension, and the heavy queue is derived from the DSN. AMQP also needs MESSENGER_HEAVY_TRANSPORT_DSN set, see below. Whatever the transport, a Redis or AMQP server holds personal data and must stay private, see Personal data.

A second queue for heavy jobs

Exports and imports are routed to async_heavy, so a large import never holds up the mails waiting on async. Its DSN is derived from MESSENGER_TRANSPORT_DSN:

MESSENGER_TRANSPORT_DSN	async_heavy
empty	sync://, the job runs at once
doctrine://...	The same table, with queue_name=heavy
redis://host:port/<stream>/...	The stream <stream>_heavy, or messages_heavy when the DSN names no stream
anything else	The same DSN: both transports share one queue

MESSENGER_HEAVY_TRANSPORT_DSN overrides the derived value. On AMQP, or any queue other than Doctrine and Redis, set it: sharing the queue of async, the exports and imports are read by the worker of async and tried again three times like its jobs. A worker started with messenger:consume async async_heavy takes the mails first, then the heavy jobs; a shop can also run one worker per transport.

The queue in the shop database

doctrine://default opens a DBAL connection with the settings of the Propel connection, its PDO options and attributes included (TLS certificates, for instance), so the queue lives in the same database as the shop, with no extra service. Only the connection name default is accepted.

The messenger_messages table is created by the installer and by the 3.3.0 update script. The transport never creates it on the fly.

The DBAL connection is a second connection to the database, independent from Propel. A message dispatched inside a Propel transaction is written to the queue at once, and stays there even if that transaction is rolled back. A worker can then pick up a message about a row that was never committed. Dispatch after the commit, never inside the transaction.

Events or messages

Thelia events and messages answer different needs.

An event, dispatched through the event dispatcher, tells every listener that something happens, and every listener runs before dispatch() returns. Listeners can change the event, stop its propagation, or make the action fail. Events stay synchronous whatever the configuration of the queue: the event dispatcher is not a message bus. See the events reference.

A message asks for one piece of work to be done by its handler, now or later, without the caller depending on the result. Use a message when:

• the work is slow or calls a remote service (a mail server, a marketplace, an ERP),
• the caller does not need its result to answer,
• the work can be retried on its own if it fails.

The two combine. A listener that reacts to an event but does not need to do the work in the request dispatches a message, and the handler does the work:

public function onStockUpdated(ProductSaleElementUpdateEvent $event): void
{
    $this->bus->dispatch(new PushStock($event->getProductSaleElementId()));
}

The listener stays fast, and the handler gets the retries and the failure handling described below.

Retries and failed jobs

When a handler throws on the async transport, the message is retried three times, after 30 seconds, 2 minutes and 8 minutes (a multiplier of 4, plus the random jitter Symfony adds). After the last attempt it goes to the failed transport. A message is never lost and never retried in a loop. async_heavy has no retries (max_retries: 0): a message that fails there goes to failed at once, since running a whole export or import again unchanged would fail the same way.

A handler that knows a retry cannot help, for instance because the remote service rejected the data, throws Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException and the message goes to failed at once.

failed is MESSENGER_FAILURE_TRANSPORT_DSN, which defaults to doctrine://default?queue_name=failed: by default the failed jobs are kept in the shop database, whatever queue async uses, so the back office has a single place to read them from. They stay there until someone replays them, removes them, or a purge deletes them after 30 days (thelia:messenger:purge-failed and maintenance:purge). When failed is in the shop database, replaying or deleting a job from the back office acts once, even when two administrators, or a double click, act on the same job at the same moment. Replaying the same job from the back office and with messenger:failed:retry at the same moment is not guarded: see Known limits.

Messenger writes the exception of a failed or retried job to the log of the workers, and a database error quotes the values of a row. The core takes the exception out of these log lines, on the messenger channel, and names it by its class, code and place in the code instead. The back office shows the reason of a failed job only when it was written for the administrator, or when it is the answer of the mail server with its credentials hidden; anything else reads as a server error. The rule is detailed in the reference.

Because a failed job can be replayed, and because a retry runs a handler that may have done part of its work, a handler must be idempotent: run twice, its effect happens once. The reference shows how in Idempotence and replay.

What a worker sees

A handler running in a worker serves no page: there is no visitor, no cart and no HTTP request of a browser. Each message gets a request of its own on the address of the shop (DEFAULT_URI, its folder included), with an empty session, so the loops of a template a handler renders, a mail of a module for instance, can run. Nothing a page sets up runs for it: the session language is the default language, the request is not one of the back office, and no customer or administrator is restored from a cookie. The request goes once the message is handled or has failed, and a message the worker skipped never hands its request to the next.

The handler still runs in a long process that handles one message after another, so the core resets the state of the shop before each message:

• settings are read again from the database, so a change made in the back office since the previous message is seen,
• the active languages, the default country and currency, the module configuration, the rewritten URLs, the price currency, the taxes of a rule, the return statuses and the results of the loops are forgotten,
• the translator is set back to the default language.

Before each message, the Propel connection is also checked and reopened if MySQL closed it after its wait_timeout. Propel instance pooling is disabled for the whole life of the worker, so a handler never reads a model object kept in memory by a previous message.

Once a message is handled or has failed, and before the worker acknowledges it, the DBAL connection of the queue in the shop database is closed. The acknowledgement, or the retry, goes through a connection opened afresh, which MySQL cannot have closed while a long job kept the worker busy.

Symfony also resets the services tagged kernel.reset between messages, unless the worker is started with --no-reset.

What a worker does not read again is its container: the mail settings and the handlers of the modules active when it started. Clearing the cache of the shop (saving the mail settings, turning a module on or off, thelia:cache:clear) therefore asks every worker to stop once its current message is done, as messenger:stop-workers does, and its supervisor starts it again on the new cache. A clear asked for by a job (a module handler that saves a setting), or by the command of a recurring task the worker runs, stops the worker running it once the job is acknowledged, and the cache is cleared when that worker's command ends, an exception included: clearing it in the middle would delete the files the worker still loads its own listeners from. A worker stopped by a fatal error or killed loses the clear: run php Thelia thelia:cache:clear then. A clear of the image or document cache leaves the workers alone. The signal is written in the application cache pools, so workers running on another server only see it when THELIA_CACHE_DSN points to a cache server they share; otherwise, run php Thelia messenger:stop-workers on their server.

A handler therefore never relies on what a visitor would have set. It receives ids, reads its data again, chooses the locale explicitly, and builds absolute URLs from DEFAULT_URI.

What the core sends through the queue

The core routes three jobs: mails to async, exports and imports to async_heavy.

E-mails

Symfony\Component\Mailer\Messenger\SendEmailMessage is routed to async. The mail is rendered in the request, where the language and the shop URL are known; only the delivery to the mail server is deferred. With a queue, a slow or unreachable mail server no longer holds up the page that sends the mail, and a mail that could not be delivered ends up in failed, ready to be sent again.

The order history line email_sent is written when the mail server accepted the mail. A listener on the Symfony Mailer SentMessageEvent reads the X-Thelia-Order-Id and X-Thelia-Message-Code headers of the mail. These headers travel through the queue with the mail and are removed right before it is handed to the mail server, so the customer never receives them. With a worker, the line appears when the worker delivered the mail, and its author is system.

A queue may deliver a job twice: when a worker stops after the mail server accepted the mail but before the job was acknowledged, the transport redelivers the job once its redeliver_timeout is over. MailerFactory gives each mail it sends an identifier in an X-Thelia-Mail-Id header, a new one on each send, even when the same Email object is sent again. A worker keeps the identifiers of the mails the server accepted for a week, and does not send a mail whose identifier it kept. Workers on several servers share them when THELIA_CACHE_DSN points to a cache server they share. A mail sent with the Symfony mailer directly, not through MailerFactory, is not guarded: see Known limits.

With a queue, MailerFactory::send*OrFail() only throws when the mail could not be built or queued. A delivery failure happens later, in the worker. MailerFactory::sendNow() hands a mail to the mail server at once, queue or not: the back office uses it to test the mail configuration, whose only point is the answer of the server.

A queued mail holds everything it is sent with, and names no file of the server:

• an attachment given by its path (attachFromPath()) is read when the mail is queued and stored in the queue with its content. It is held in the memory of the request, and a large one is bounded by the max_allowed_packet of the database,
• a mail is read back from the queue only when its class is exactly RawMessage, Message or Email, or a TemplatedEmail or NotificationEmail already rendered, and when every part is one Symfony builds a mail with. A part that names a file, in the mail or in a mail it carries (MessagePart), a part of another class, or a part that throws when it is looked at, makes the job an UndecodableJob, never sent,
• a mail that would be refused is refused when it is queued, with a LogicException. Symfony queues a TemplatedEmail before rendering it, and the worker renders nothing: render it first (BodyRenderer), or send it without a queue.

Back-office exports and imports

An export or an import started from the back office is a job. The launcher (Thelia\Domain\DataTransfer\Job\ExportJobLauncher or ImportJobLauncher) records a row in export_job or import_job and dispatches RunExportJob or RunImportJob, routed to async_heavy. The row is what the back office reads. Without a queue, the job runs in the request and the page shows its outcome at once. With a queue, the page of the job follows it: waiting, running, done, or failed with its reason.

An import file is checked in the request, before anything is queued, and moved to var/data-transfer/import/. It is kept outside the cache on purpose: a deployment empties the cache, and a queued import would lose its file.

An import runs in a single database transaction. Stopped half way, by an error, a killed worker or a deployment, it leaves the catalog as it was, and its outcome is written to its row in the same transaction, so the catalog and the row never disagree. The cost is that the rows it writes stay locked until it ends: an order or a back-office edit touching one of them waits, then fails after innodb_lock_wait_timeout. Run a large stock or price import outside trading hours. When the code that launches an import already holds a transaction, the import works inside it and leaves the commit or the rollback to that code, so do not launch an import inside a transaction of your own. The listeners of IMPORT_BEGIN, IMPORT_FINISHED and IMPORT_SUCCESS run inside the transaction too: what they write is cancelled with the import, what they send is not, and a job they queue in the shop database is committed at once, so a worker may run it before the import is committed.

A job never runs twice at the same time, and a finished job never runs again. A worker claims the row of the job before running it. A message that finds its job still running, because another worker holds it or because its worker died, does not run it: it is queued again to look at the job later. The job is taken over only once its row has shown no progress for an hour, so a long export or import that is still working is never started a second time. What this means for an operator after a worker is killed is in Running the workers.

A job that fails goes to failed at once, with no retry: running a whole export or import again unchanged would fail the same way. Replaying it from the failed jobs starts it over, an import from its first row.

The reason a job failed is shown to the administrator only when it was written for them, by an exception implementing Thelia\Exception\UserFacingFailure. Any other failure may quote SQL, the values of a row, paths or host names: the administrator reads that the job failed because of a server error, and the log names the exception by its class, code and place in the code, never by its message.

An export or an import is one job for one file. It is not split into smaller jobs: a file is written by a single writer, and an import has to be atomic. Splitting work into smaller jobs fits work made of independent items, as in a module that pushes one message per product combination. It does not fit the rows of a file: do not dispatch one message per row.

The launchers, the rights, the limits of an uploaded file and the retention of the jobs and their files are in the reference.

Recurring tasks

The recurring tasks of the shop are declared in a Symfony Scheduler schedule named thelia (Thelia\Scheduler\TheliaSchedule). A worker consuming scheduler_thelia runs them; until one does, nothing runs, and a shop can keep its crontab instead. The schedule is stateful and locked through LOCK_DSN, so two workers never run a task twice, and a worker that restarts after a missed run catches up the last one only. Across servers this holds only when both are shared: LOCK_DSN must name a store every worker reaches (not flock nor semaphore, which lock one server only), and THELIA_CACHE_DSN a cache server they all use; otherwise each server runs the tasks on its own, and a cache clear forgets the last runs. The tasks and their settings are listed in the reference.

A task of the schedule that fails does not go to the failure transport: Messenger only sends there what came from a queue, and the task runs again at its next time anyway. The core logs the failure and keeps the last one of each task, in the application cache, until a run of that task goes through, or for a month after the date of that failure, so a task taken off the schedule does not stay listed forever. The Background jobs screen lists them under "Recurring tasks that failed", see the reference.

Which classes a queue accepts

A queued message is data that is turned back into objects when a worker reads it, and whoever can write to the queue would choose their classes. Thelia never uses PHP serialization for this. Envelopes are JSON, written and read by the Symfony serializer through Thelia\Messenger\Serializer\AllowedClassesSerializer, which checks the classes before building anything:

• a message class must be under Thelia\, under the namespace of an active module (the namespace named after the module code, such as MyModule\Message\SyncStock), Symfony\Component\Mailer\Messenger\SendEmailMessage, or listed by its exact name in the container parameter thelia.messenger.allowed_message_classes,
• outside that parameter, a handler of the application must also take the message class, or one of its parents or interfaces. The handled classes are collected when the container is built, so no other class of those namespaces is ever built from a queue. A handler that takes any message (* or object) does not count,
• a stamp must be a Messenger stamp or come from those same namespaces, and SerializerStamp is refused, as it would let the queue change how the rest of the envelope is read.
• a message that runs a console command, a process, an HTTP request or a scheduled service call, or that dispatches another message (the classes under Symfony\Component\Console\Messenger\, Process\Messenger\, HttpClient\Messenger\, Scheduler\Messenger\, Cache\Messenger\ and Messenger\Message\), is never built from a queue, even when a project lists it: whoever writes to the queue would run anything on the server. The schedule of the shop is not affected, as its messages never go through a queue.

Whether a message class belongs to an active module is read from the database each time. When MySQL closed the connection of the worker during a long wait, the Propel connection is opened again once and the module is read again.

The check runs when a message is read from a queue and when it is written to one. A module dispatching to a real queue a class the workers would refuse gets a LogicException ("is not one the shop queues") at dispatch time, not a queue that never empties. Without a queue (sync://), the message is handled in the same call and never serialized, so the check does not run: a message that no handler takes fails with Symfony's NoHandlerForMessageException instead, also at dispatch time.

A job already queued can still become unreadable: its class is refused by the check, or it can no longer be built because its module was turned off, its class was removed, or its content no longer fits the class. The worker then reads it as Thelia\Messenger\Message\UndecodableJob, which keeps the original class (originalType), the reason and the original body, instead of letting it be dropped. Whatever the inner serializer throws while building the message, the reason is a fixed one, Its content no longer fits the class <class>., and the exception is named in the log by its class, code and place: a message that cannot be built never stops the worker, however often it is delivered. The handler of an UndecodableJob sends the job to failed at once. The Background jobs screen lists it as Unreadable job: <reason> with its original class, and it can be deleted there.

For a module author, this means:

• message classes live in the module namespace,
• each message class has a handler: a message that no handler takes fails at dispatch, with a LogicException when it goes to a queue, with a NoHandlerForMessageException without one,
• they can be written and read as JSON by the Symfony serializer: public or promoted constructor properties with scalar types, such as ids and codes,
• they never carry a Propel model. The handler reads the row again from its id,
• they implement Thelia\Messenger\Message\DescribedJob, so the back office lists a failed job by a line that says what it was, with no personal data in it. See the message of the example module.

A class from outside these namespaces, such as an application class under App\, is accepted only once a project adds its exact name to thelia.messenger.allowed_message_classes. The check for files a mail would make the worker read only applies to Symfony's SendEmailMessage: a listed class that carries an Email of its own must refuse attachments by path itself.

Personal data

A queue holds personal data: a queued mail carries the address of its recipient and the content of the order, and a failed job keeps everything it was dispatched with. How long each is kept, and how to protect a Redis or AMQP server, is in Personal data.