Skip to main content
Version: Next

Background jobs reference

This page lists the settings and commands of the background jobs, then walks through a complete module that queues its own work. The model behind it (transports, retries, the classes a queue accepts) is explained in Background jobs. Running the workers in production is covered in Running the workers.

Configuration​

Every setting is an environment variable, set in .env.local or in the environment of the hosting.

VariableDefaultEffect
MESSENGER_TRANSPORT_DSNemptyThe queue of the async transport, from which the async_heavy queue is derived. Empty means no queue: every job runs at once. doctrine://default, a Redis DSN or an AMQP DSN names a queue that a worker consumes.
MESSENGER_HEAVY_TRANSPORT_DSNderivedOverrides the queue of async_heavy. Set it on AMQP: otherwise the exports and imports share the queue of async and its three retries.
MESSENGER_FAILURE_TRANSPORT_DSNdoctrine://default?queue_name=failedWhere the jobs that failed every attempt are kept.
THELIA_SCHEDULE_SALE_CHECK* * * * *Cron expression of sale:check-activation. Empty disables the task.
THELIA_SCHEDULE_MAINTENANCE_PURGE30 3 * * *Cron expression of maintenance:purge. Empty disables the task.
THELIA_SCHEDULE_FAILED_JOBS_PURGE0 4 * * *Cron expression of thelia:messenger:purge-failed. Empty disables the task.
THELIA_SCHEDULE_CURRENCY_RATESemptyCron expression of currency:update-rates. Off by default.
LOCK_DSNflock in the .env of the core, semaphore in the .env.dist of thelia-projectThe lock store that keeps two workers from running the same scheduled task.

doctrine://default is the only Doctrine connection name accepted: it is the shop database, reached with the settings of the Propel connection. The table messenger_messages is created by the installer and by the 3.3.0 update script.

A Redis or AMQP DSN needs its Messenger bridge, which the core does not ship: composer require symfony/redis-messenger (with the redis PHP extension) or composer require symfony/amqp-messenger (with the amqp extension).

How async_heavy derives its queue from MESSENGER_TRANSPORT_DSN is shown in A second queue for heavy jobs. The options of the Symfony transport go in the DSN, such as doctrine://default?redeliver_timeout=600; the heavy queue derived from it keeps them. What redeliver_timeout changes, and what it does not, is explained in When a worker stops.

The core configures Messenger for the whole application: its serializer, its bus, its failure_transport and the routing of its messages are loaded before the configuration of the project. The container parameter thelia.messenger.allowed_message_classes (an array, empty by default) lists the exact names of the message classes that a project queues from outside Thelia\ and the namespaces of the active modules: an application that queues its own App\Message\… classes gets a LogicException on dispatch until they are listed there. The core declares it empty (core/lib/Thelia/Config/Resources/parameters/messenger.php). Any other message class must also be taken by a handler, or have a parent or an interface a handler takes: the parameter thelia.messenger.handled_message_classes is filled when the container is built and is not meant to be set by hand. A handler that takes any message (* or object) does not count.

A project sets the list in config/services.yaml, which is loaded after the configuration of the core:

# config/services.yaml
parameters:
    thelia.messenger.allowed_message_classes:
        - App\Message\SyncInvoices

Transports​

TransportRetriesNotes
async3, after 30 s, 2 min and 8 min (multiplier 4, with jitter)Then the message goes to failed.
async_heavynone (max_retries: 0)Back-office exports and imports. A job that fails goes to failed at once.
failednoneReplay with messenger:failed:retry or from the back office.
scheduler_thelianoneThe recurring tasks of the thelia schedule.

The core routes Symfony\Component\Mailer\Messenger\SendEmailMessage to async, and Thelia\Domain\DataTransfer\Job\RunExportJob and Thelia\Domain\DataTransfer\Job\RunImportJob to async_heavy.

Commands​

CommandUse
messenger:consume async async_heavy --time-limit=3600 --memory-limit=256MRun a worker. It takes the mails first, then the exports and imports.
messenger:consume scheduler_thelia --time-limit=3600 --memory-limit=256MRun the recurring tasks.
messenger:statsCount the messages waiting in each transport.
messenger:failed:showList the failed jobs, or show one with its error.
messenger:failed:retryReplay failed jobs.
messenger:failed:removeDelete failed jobs.
messenger:stop-workersAsk every worker to stop after its current message.
thelia:messenger:purge-failedDelete the jobs in the failure transport for more than --older-than days (30 by default). maintenance:purge deletes those older than 30 days too.

All of them run through php Thelia, for example php Thelia messenger:stats.

Back office​

Configuration > System > Background jobs (/admin/configuration/background-jobs) shows:

  • whether a queue is configured, and how many jobs wait in it,
  • how many jobs failed, and the list of failed jobs, the newest first, with their description, the reason of the failure, the date and the number of attempts. The description is the describe() of a message implementing Thelia\Messenger\Message\DescribedJob, the subject and the recipients of a mail, or else the short name of the message class,
  • the recurring tasks whose last run failed, under "Recurring tasks that failed",
  • the last 20 exports and the last 20 imports of the administrator looking at it, of every administrator for a super-administrator.

The reason of a failed job is shown when it was written for the administrator (an exception implementing Thelia\Exception\UserFacingFailure) or when it is the answer of the mail server, with the credentials of any URL in it hidden (a password holding an @ is hidden whole). Either is cut to 2000 characters. Any other reason reads as a server error.

The waiting count adds the jobs of async and of async_heavy, and counts them once when both transports have the same DSN. When failed is in the shop database, the default, the list is read in SQL, so the newest failures show however many there are.

Each failed job can be replayed or deleted. Replay takes the job out of failed first, then puts it back on the transport it failed on; if that fails, the job goes back to failed. A job that implements Thelia\Messenger\Message\ReplayableJob is put back as forReplay() returns it, so it starts afresh, and the same goes for a job replayed with messenger:failed:retry. A double click cannot send it twice. Without a queue, a replayed job runs at once, and a job that fails again stays in the failed list. A replay that fails shows "The job failed again. The details are in the server log." and the reason goes to the log. If putting it back in failed fails too (queue and database both down), the job is lost from the queues and a critical line of the log names it by its id and its class, and the two failures by their class, code and place: its description and its content may name a customer, and the log is kept far longer than the failed jobs.

A job the workers could no longer read (module turned off, class removed, content that no longer fits) is listed as Unreadable job: <reason> with its original class. Replayed, it would fail the same way, so it has no Replay button; delete it once you know what it was.

A recurring task of the thelia schedule that fails does not go to failed: it runs again at its next time. Its last failure is kept in the application cache until a run of the task goes through, or for a month after the date of that failure, and the screen lists it with the task, the reason and the date. A command that exits with an error code gives Command "<input>" exited with code "<code>". as its reason.

The "Details" link of an export or import is only shown to the administrator who started the job and to super-administrators.

The screen answers to the resource admin.configuration.background-jobs, given to no existing profile by the update: until it is granted, only super-administrators see the screen. Grant VIEW to read it, UPDATE to replay and DELETE to delete. The description of a mail names its recipient and a failure reason may quote personal data, so give it only to the profiles that need it.

Back-office exports and imports​

Exports and imports started from the back office are jobs. How they run, and why an import is a single transaction, is explained in Background jobs. This section lists what they expose and what they accept.

ExportImport
Transportasync_heavyasync_heavy
LauncherExportJobLauncher::launch(Export $export, string $serializerId, ?string $archiverId = null, ?Lang $language = null, bool $includeImages = false, bool $includeDocuments = false, ?array $rangeDate = null, ?int $adminId = null): ExportJobImportJobLauncher::launch(Import $import, File $file, string $originalName, ?Lang $language = null, ?int $adminId = null): ImportJob
MessageRunExportJob(int $exportJobId)RunImportJob(int $importJobId)
Rowexport_jobimport_job: status, file name, rows imported, refused rows with their reason, error
Without a queueRuns in the request, the file is served at onceRuns in the request, the page shows the rows imported and the rows refused
With a queue/admin/export/job/{id}: waiting, running with the rows written, done with the download, failed with the reason/admin/import/job/{id}: waiting, running, done with the rows changed and refused, failed with the reason
Replay of a failed jobStarts the export overStarts over from the first row
FilesDeleted by maintenance:purge once older than a dayKept in var/data-transfer/import/<Ymd>/ until the import is done, kept while it may be replayed
maintenance:purgeDeletes the done jobs older than 7 days, any other after 31 daysDeletes the done jobs older than 7 days, any other after 31 days, with their files
Right to launchVIEW on the export resourceUPDATE on the import resource; VIEW only shows the imports and their jobs
Job page accessThe administrator who started it and super-administrators, download includedThe administrator who started it and super-administrators

The launchers and the messages live in Thelia\Domain\DataTransfer\Job\. A job has one of the statuses of the enum Thelia\Domain\DataTransfer\Job\JobStatus: queued, running, done, failed. Both messages implement Thelia\Messenger\Message\DescribedJob, and are listed as Export #<id> and Import #<id> among the failed jobs.

The job pages​

A job page refreshes itself every 3 seconds, at most 100 times, then stops and offers a link to refresh it by hand; a "Stop refreshing" link stops it earlier. Other administrators than the one who started the job get a 403, even with the export or import right, and the lists of recent exports and imports only show their own jobs, except to super-administrators.

An administrator can launch 10 exports and imports in 10 minutes, the two counted together (rate limiter admin_data_transfer_launch, declared in the core framework configuration). A launch is only counted once its form is valid: a form sent without a file, with a file the server did not receive whole, with a file whose name or content the import refuses, or with an invalid option is not counted. Past the limit, the launch is refused with a message asking to wait a few minutes.

Without a queue, an import that refused more than 10 rows leads to its job page, which lists them all; with fewer, the form page shows them in its message. The description of an export or an import, written in HTML by its module, keeps only p, br, strong, b, em, i, u, ul, ol, li, code, pre, blockquote, span and a (with href and title, and only http, https and mailto links). Anything else, scripts, styles, media, id and name attributes included, is removed.

The reason of a failure​

The reason an administrator reads, on the job page and on the Background jobs screen, is the message of an exception implementing Thelia\Exception\UserFacingFailure, found anywhere in the chain of the failure, cut to 2000 characters. The core ones live in Thelia\Domain\DataTransfer\Exception: UploadRefusedException (an uploaded file refused), DataTransferNoDataFoundException, HandlerUnavailableException, MissingColumnsException and JobRefusedException. Thelia\Form\Exception\FormValidationException itself does not implement the interface.

Any other failure reads "The job failed because of a server error. The details are in the server log.", and the log line of the job names the exception by its class, code and place in the code. The job page passes the reason through the translator, so the fixed messages are shown in the language of the administrator.

A module whose export or import throws an exception meant for the administrator implements the interface on it:

final class WarehouseRefusedException extends \RuntimeException implements \Thelia\Exception\UserFacingFailure
{
}

Some refusals of the core, all JobRefusedException:

  • an export whose year is not four digits or whose month is not 1 to 12: "The dates of the export are not valid.",
  • an export whose format is no longer available on the server: The format "<format>" is no longer available on this server.,
  • an import row whose save leaves the transaction unable to commit, even when a module catches the failure: Row <n> could not be saved: nothing was imported.

A job the queue refuses at dispatch is recorded as failed with "The job could not be queued. The details are in the server log.", and the file of such an import is deleted. A job whose row was deleted fails for good and stays in failed. A job deleted from the failed jobs before it ever ran is marked failed with "Deleted from the failed jobs before it ran.", so its row no longer reads as waiting for a worker.

What an import accepts​

The file is checked in the request, before anything is queued. Its extension must be one the import handlers accept, and its content is read: an archive must match the type of its archiver, anything else must be text. A file that does not parse in its format is refused with a message asking to check its content.

An archive is looked into before it is extracted, both when it is uploaded and when it is imported (Thelia\Domain\DataTransfer\ArchiveInspector). It is refused when it holds more than 1000 entries, folders included, more than 512 MB once extracted, a name that is absolute or contains .., or a link, zip and tar alike. What the extraction really wrote is measured against the same 512 MB, and the extracted copy is removed once the import is over, whatever came of it. ImportHandler takes the ArchiveInspector as a required constructor argument: code that builds or extends it passes one.

A row the import refuses (a missing combination, an invalid GTIN) is listed with the job as a refusal, and the other rows are kept. A refusal that is not valid UTF-8 is stored with the invalid bytes replaced. The refusals are kept up to 60,000 bytes; past that, the first ones are kept and a last line says <n> more rows were refused. A stock or a price must be a number, given as text or as a number, below 10^10 in absolute value: a list, an object, a missing price or anything else refuses the row with its reason, and the price of a row is checked before anything is written for it.

The file is moved out of the upload directory to var/data-transfer/import/, not to the cache, which a deployment empties. Its path is stored relative to the project, its name is cut to 100 characters, and the import only reads a file that lies in that directory.

Files and retention​

  • An import file is deleted once the import is done. Without a queue a failed import can never be replayed, so its file is deleted at once; with a queue it is kept while the job may be replayed.
  • Deleting a failed import from the back office deletes its file too (Thelia\Messenger\Event\FailedJobRemovedEvent), unless the import is still marked running. messenger:failed:remove and thelia:messenger:purge-failed leave the file to maintenance:purge.
  • maintenance:purge deletes the done jobs created more than 7 days ago, and any other job (failed, queued, running) after 31 days: a day past the 30 days a failed message is kept, so a replayed job still finds its row. The file of an import goes with its row only when it lies inside var/data-transfer/import. The purge also removes the files of that directory older than 31 days, such as the extracted copy of an archive left by a killed worker, and its empty directories once they are more than a day old.
  • A file that cannot be deleted is logged as a warning and left to the purge; the job is still reported as it ended.

Recurring tasks​

The thelia schedule runs these commands when a worker consumes scheduler_thelia:

CommandVariableDefault
sale:check-activationTHELIA_SCHEDULE_SALE_CHECKevery minute
maintenance:purgeTHELIA_SCHEDULE_MAINTENANCE_PURGE03:30
thelia:messenger:purge-failedTHELIA_SCHEDULE_FAILED_JOBS_PURGE04:00
currency:update-ratesTHELIA_SCHEDULE_CURRENCY_RATESoff

currency:update-rates is off because it overwrites every rate, including those set by hand. sale:check-activation switches a sale back on when it was turned off by hand inside its dates, as the command always did.

A shop that keeps these commands in its crontab must not also consume scheduler_thelia, or each task runs twice. Choose one of the two.

A module adds a task to the same schedule with a Symfony Scheduler attribute and schedule: 'thelia':

  • #[AsCronTask('0 2 * * *', schedule: 'thelia')] for a cron expression,
  • #[AsPeriodicTask('1 hour', schedule: 'thelia')] for an interval.

The attribute goes on a console command class, or on a service class with method: naming the method to call. The example below schedules a command.

A complete module: StockSync​

StockSync pushes the stock of each product combination to a remote warehouse API. When a combination is saved in the back office, a listener dispatches a message, and the handler calls the API. A nightly task pushes every combination again, to catch changes made elsewhere.

With an empty MESSENGER_TRANSPORT_DSN, the push happens during the save. With a queue, the save returns at once and a worker does the push, with retries if the API is down.

Without a queue, an exception thrown by the handler reaches the code that dispatched the message: the save of the combination in the back office fails with it. A shop that relies on a remote API this way should run a queue.

The first line of each block below names the file it goes in; it is not part of the file.

local/modules/StockSync/
├── StockSync.php
├── composer.json
├── Config/
│   ├── config.xml
│   ├── module.xml
│   ├── schema.xml
│   └── TheliaMain.sql          # generated by module:generate:sql
├── Command/
│   └── PushAllStockCommand.php
├── EventListener/
│   └── StockChangedListener.php
├── Message/
│   └── PushStock.php
├── MessageHandler/
│   └── PushStockHandler.php
├── Model/                       # generated by module:generate:model
└── Tests/
    ├── PushStockHandlerTest.php
    └── StockChangedListenerTest.php

The module class​

configureServices() registers every class of the module, which makes the handler, the listener and the command discovered. configureContainer() routes the message to async, in the same way the core routes its own messages. postActivation() creates the tables.

// local/modules/StockSync/StockSync.php
<?php

declare(strict_types=1);

namespace StockSync;

use Propel\Runtime\Connection\ConnectionInterface;
use StockSync\Message\PushStock;
use Symfony\Component\DependencyInjection\Loader\Configurator\ContainerConfigurator;
use Symfony\Component\DependencyInjection\Loader\Configurator\ServicesConfigurator;
use Thelia\Core\Install\Database;
use Thelia\Module\BaseModule;

final class StockSync extends BaseModule
{
    public const DOMAIN_NAME = 'stocksync';

    public static function configureServices(ServicesConfigurator $servicesConfigurator): void
    {
        $servicesConfigurator->load(self::getModuleCode().'\\', __DIR__)
            ->exclude([
                __DIR__.'/I18n/*',
                __DIR__.'/Model/*',
                __DIR__.'/Message/*',
                __DIR__.'/Tests/*',
                __DIR__.'/StockSync.php',
            ])
            ->autowire()
            ->autoconfigure();
    }

    public static function configureContainer(ContainerConfigurator $containerConfigurator): void
    {
        $containerConfigurator->extension('framework', [
            'messenger' => [
                'routing' => [
                    PushStock::class => 'async',
                ],
            ],
        ], prepend: true);
    }

    public function postActivation(?ConnectionInterface $con = null): void
    {
        if (!self::getConfigValue('is_initialized', false)) {
            (new Database($con))->insertSql(null, [__DIR__.'/Config/TheliaMain.sql']);

            self::setConfigValue('is_initialized', true);
        }
    }
}

Message/ is excluded from the service scan: a message is data, not a service. Tests/ is excluded too: a test class is not a service, and PHPUnit is not installed in production. The message class may also carry the route itself, with #[AsMessage('async')]. A message routed nowhere runs in the request that dispatched it.

A shop can also route the message itself, in config/packages/messenger.yaml, which takes precedence over the prepended module configuration:

# config/packages/messenger.yaml
framework:
    messenger:
        routing:
            'StockSync\Message\PushStock': async

A queue of its own​

Routed to async, the pushes wait behind the mails of the shop. A module whose work is plentiful can give it a transport of its own, still synchronous by default:

public static function configureContainer(ContainerConfigurator $containerConfigurator): void
{
    $containerConfigurator->parameters()->set('stock_sync.inline_transport_dsn', 'sync://');
    $containerConfigurator->extension('framework', [
        'messenger' => [
            'transports' => [
                'stock_sync' => '%env(default:stock_sync.inline_transport_dsn:STOCK_SYNC_TRANSPORT_DSN)%',
            ],
            'routing' => [
                PushStock::class => 'stock_sync',
            ],
        ],
    ], prepend: true);
}

With STOCK_SYNC_TRANSPORT_DSN empty or unset, the pushes stay synchronous. A shop that wants them queued sets, for instance, STOCK_SYNC_TRANSPORT_DSN=doctrine://default?queue_name=stock_sync and runs a worker of its own with php Thelia messenger:consume stock_sync: the workers of the shop do not consume it. Pushes that fail every attempt still go to the failed transport of the shop, and the back office lists them there, but it does not count the pushes waiting in stock_sync: watch that queue with messenger:stats stock_sync.

Metadata and dependencies​

<!-- local/modules/StockSync/Config/module.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<module xmlns="http://thelia.net/schema/dic/module"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://thelia.net/schema/dic/module http://thelia.net/schema/dic/module/module-2_2.xsd">
    <fullnamespace>StockSync\StockSync</fullnamespace>
    <descriptive locale="en_US">
        <title>Stock sync</title>
        <subtitle>Pushes the stock of each combination to a warehouse API</subtitle>
    </descriptive>
    <languages>
        <language>en_US</language>
    </languages>
    <version>1.0.0</version>
    <author>
        <name>Your Name</name>
        <email>you@example.com</email>
    </author>
    <type>classic</type>
    <thelia>3.3.0</thelia>
    <stability>prod</stability>
</module>
{
    "name": "your-vendor/stock-sync-module",
    "type": "thelia-module",
    "license": "MIT",
    "require": {
        "thelia/installer": "~1.1",
        "symfony/http-client": "^7.4"
    },
    "extra": {
        "installer-name": "StockSync"
    },
    "autoload": {
        "psr-4": {
            "StockSync\\": ""
        }
    }
}

Config/config.xml declares nothing, since configureServices() registers the services, but the module:generate:* commands refuse a module without it:

<!-- local/modules/StockSync/Config/config.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<config xmlns="http://thelia.net/schema/dic/config"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://thelia.net/schema/dic/config http://thelia.net/schema/dic/config/thelia-1.0.xsd">
</config>

The tables​

stock_sync_state remembers the last quantity pushed for each combination; it is what makes the handler idempotent. stock_sync_log keeps one line per push. The foreign key points to a core table, so the core schema is declared as an external schema, for reference only: without it, module:generate:sql cannot resolve product_sale_elements.

<!-- local/modules/StockSync/Config/schema.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<database defaultIdMethod="native" name="TheliaMain" namespace="StockSync\Model">

    <!-- The core tables a foreign key points to, read for reference only. -->
    <external-schema filename="local/config/schema.xml" referenceOnly="true"/>

    <table name="stock_sync_state">
        <column name="product_sale_elements_id" primaryKey="true" required="true" type="INTEGER"/>
        <column name="pushed_quantity" required="true" type="FLOAT"/>
        <column name="pushed_at" required="true" type="TIMESTAMP"/>

        <foreign-key foreignTable="product_sale_elements" onDelete="CASCADE">
            <reference local="product_sale_elements_id" foreign="id"/>
        </foreign-key>
    </table>

    <table name="stock_sync_log">
        <column autoIncrement="true" name="id" primaryKey="true" required="true" type="INTEGER"/>
        <column name="product_sale_elements_id" required="true" type="INTEGER"/>
        <column name="quantity" required="true" type="FLOAT"/>
        <column name="created_at" required="true" type="TIMESTAMP"/>
    </table>
</database>

Generate the models and the SQL file:

php Thelia module:generate:model StockSync
php Thelia module:generate:sql StockSync

The message​

The message carries an id, nothing else. It is a class of the module namespace, so the queue accepts it, and its only property is a promoted scalar, so the Symfony serializer writes and reads it as JSON.

// local/modules/StockSync/Message/PushStock.php
<?php

declare(strict_types=1);

namespace StockSync\Message;

use Thelia\Messenger\Message\DescribedJob;

final readonly class PushStock implements DescribedJob
{
    public function __construct(
        public int $productSaleElementsId,
    ) {
    }

    public function describe(): string
    {
        return \sprintf('Stock push of combination #%d', $this->productSaleElementsId);
    }
}

Never put a Propel model in a message. It cannot be serialized as JSON, and by the time the worker reads the message the row may have changed or been deleted.

Thelia\Messenger\Message\DescribedJob gives the job the line the Background jobs screen shows for it among the failed jobs. A message that does not implement it is listed by the short name of its class (PushStock), which tells the administrator nothing about which combination failed. describe() runs when the screen is displayed, on the message read back from the queue.

The description is shown to every administrator who may read the screen, and it stays as long as the failed job does. Never put personal data in it: no name, e-mail address or postal address of a customer, no order content. Give ids and references, which the administrator can look up with their own rights.

Thelia\Messenger\Message\ReplayableJob is for a message that carries a state a replay must reset, such as the number of times it was postponed: its forReplay() returns the message to put back on the queue when the job is replayed, from the back office or with messenger:failed:retry. PushStock carries only an id, so it has nothing to reset and does not implement it.

The handler​

The handler re-reads the combination, compares it with what was pushed last time, calls the API with the absolute quantity, then records the push. It assumes nothing about a request: the shop URL comes from DEFAULT_URI, and the locale of the product title is the default language of the shop, chosen explicitly.

// local/modules/StockSync/MessageHandler/PushStockHandler.php
<?php

declare(strict_types=1);

namespace StockSync\MessageHandler;

use Propel\Runtime\Propel;
use StockSync\Message\PushStock;
use StockSync\Model\StockSyncLog;
use StockSync\Model\StockSyncState;
use StockSync\Model\StockSyncStateQuery;
use StockSync\StockSync;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
use Symfony\Contracts\HttpClient\HttpClientInterface;
use Thelia\Model\Lang;
use Thelia\Model\ProductSaleElementsQuery;

#[AsMessageHandler]
final readonly class PushStockHandler
{
    public function __construct(
        private HttpClientInterface $httpClient,
        #[Autowire(env: 'DEFAULT_URI')]
        private string $shopUrl,
    ) {
    }

    public function __invoke(PushStock $message): void
    {
        $productSaleElements = ProductSaleElementsQuery::create()->findPk($message->productSaleElementsId);

        // Deleted since the message was dispatched: nothing left to push.
        if (null === $productSaleElements) {
            return;
        }

        $quantity = (float) $productSaleElements->getQuantity();
        $state = StockSyncStateQuery::create()->findPk($productSaleElements->getId());

        // Already pushed with this quantity: a replay, or a second message for the same change.
        if (null !== $state && $state->getPushedQuantity() === $quantity) {
            return;
        }

        $apiUrl = (string) StockSync::getConfigValue('api_url');

        if ('' === $apiUrl) {
            throw new UnrecoverableMessageHandlingException('StockSync has no api_url configured.');
        }

        $locale = Lang::getDefaultLanguage()->getLocale();
        $product = $productSaleElements->getProduct();
        $product->setLocale($locale);

        $response = $this->httpClient->request('PUT', rtrim($apiUrl, '/').'/stock/'.rawurlencode((string) $productSaleElements->getRef()), [
            'auth_bearer' => (string) StockSync::getConfigValue('api_token'),
            'json' => [
                'quantity' => $quantity,
                'title' => $product->getTitle(),
                'locale' => $locale,
                'source' => rtrim($this->shopUrl, '/'),
            ],
        ]);

        $status = $response->getStatusCode();

        // The API refused the data: retrying would get the same answer.
        if ($status >= 400 && $status < 500 && 429 !== $status) {
            throw new UnrecoverableMessageHandlingException(\sprintf('The warehouse refused %s with HTTP %d.', $productSaleElements->getRef(), $status));
        }

        // A 5xx or a 429: throw, so Messenger retries later.
        if ($status >= 300) {
            throw new \RuntimeException(\sprintf('The warehouse answered HTTP %d for %s.', $status, $productSaleElements->getRef()));
        }

        $this->recordPush($productSaleElements->getId(), $quantity, $state);
    }

    private function recordPush(int $productSaleElementsId, float $quantity, ?StockSyncState $state): void
    {
        $connection = Propel::getConnection('TheliaMain');
        $connection->beginTransaction();

        try {
            $state ??= (new StockSyncState())->setProductSaleElementsId($productSaleElementsId);
            $state
                ->setPushedQuantity($quantity)
                ->setPushedAt(new \DateTimeImmutable())
                ->save($connection);

            (new StockSyncLog())
                ->setProductSaleElementsId($productSaleElementsId)
                ->setQuantity($quantity)
                ->setCreatedAt(new \DateTimeImmutable())
                ->save($connection);

            $connection->commit();
        } catch (\Throwable $exception) {
            $connection->rollBack();

            throw $exception;
        }
    }
}

StockSync::getConfigValue() is read again for every message, because the worker forgets the module configuration between messages: a token changed in the back office is used by the next push without restarting the worker.

The two writes of recordPush() share an explicit transaction on the Propel connection, so the state and the log never disagree.

The listener​

The core saves a combination in its PRODUCT_UPDATE_PRODUCT_SALE_ELEMENT listener, at priority 128, and commits its transaction there. The module listens to the same event at a lower priority, so it runs after the commit, and dispatches the message.

// local/modules/StockSync/EventListener/StockChangedListener.php
<?php

declare(strict_types=1);

namespace StockSync\EventListener;

use StockSync\Message\PushStock;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use Thelia\Core\Event\ProductSaleElement\ProductSaleElementUpdateEvent;
use Thelia\Core\Event\TheliaEvents;

final readonly class StockChangedListener implements EventSubscriberInterface
{
    public function __construct(
        private MessageBusInterface $bus,
    ) {
    }

    public static function getSubscribedEvents(): array
    {
        return [
            // Below the core listener (128), which writes and commits the change.
            TheliaEvents::PRODUCT_UPDATE_PRODUCT_SALE_ELEMENT => ['onStockChanged', 0],
        ];
    }

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

The queue in the shop database uses its own connection, so a message dispatched before the commit stays queued even if the transaction is rolled back. Dispatching after the action has committed avoids a push for a change that never happened. If your own code wraps the action in a larger transaction, dispatch after that outer commit instead.

Stock also changes outside this event, when an order is placed for instance. The nightly task below covers those changes; a module that needs them sooner listens to the events concerned in the same way.

The recurring task​

#[AsCronTask] adds the command to the thelia schedule. It runs at 02:00 when a worker consumes scheduler_thelia, and can be run by hand like any command.

// local/modules/StockSync/Command/PushAllStockCommand.php
<?php

declare(strict_types=1);

namespace StockSync\Command;

use StockSync\Message\PushStock;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Scheduler\Attribute\AsCronTask;
use Thelia\Model\ProductSaleElementsQuery;

#[AsCommand(name: 'stock-sync:push-all', description: 'Queue a stock push for every product combination.')]
#[AsCronTask('0 2 * * *', schedule: 'thelia')]
final class PushAllStockCommand extends Command
{
    public function __construct(
        private readonly MessageBusInterface $bus,
    ) {
        parent::__construct();
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $ids = ProductSaleElementsQuery::create()->select('Id')->find();

        foreach ($ids as $id) {
            $this->bus->dispatch(new PushStock((int) $id));
        }

        $output->writeln(\sprintf('%d stock push(es) queued.', \count($ids)));

        return self::SUCCESS;
    }
}

The command only dispatches messages. With a queue, the pushes are spread over the async worker; without one, they run during the command. Combinations whose quantity did not change are skipped by the handler, so the nightly run only calls the API for real changes.

Trying it​

With the files above in local/modules/StockSync/, a local stand-in for the warehouse API is enough to watch a push go through the queue.

  1. Install and activate the module. Activation creates its two tables:

    composer dump-autoload
    php Thelia module:refresh
    php Thelia module:activate StockSync

  2. Start a fake warehouse that answers 200 and keeps what it receives:

    mkdir -p /tmp/warehouse
    echo '<?php file_put_contents("/tmp/warehouse/received.log", $_SERVER["REQUEST_URI"]." ".file_get_contents("php://input")."\n", FILE_APPEND);' > /tmp/warehouse/index.php
    php -S 127.0.0.1:8099 -t /tmp/warehouse /tmp/warehouse/index.php

  3. Give the module its two settings. The example has no configuration page; a module setting is stored in module_config and its value in module_config_i18n, under en_US when no locale is given:

    INSERT INTO module_config (module_id, name, created_at, updated_at)
        SELECT m.id, s.name, NOW(), NOW()
        FROM module m, (SELECT 'api_url' AS name UNION ALL SELECT 'api_token') s
        WHERE m.code = 'StockSync'
          AND NOT EXISTS (SELECT 1 FROM module_config c WHERE c.module_id = m.id AND c.name = s.name);
    INSERT INTO module_config_i18n (id, locale, value)
        SELECT c.id, 'en_US', IF(c.name = 'api_url', 'http://127.0.0.1:8099', 'test-token')
        FROM module_config c JOIN module m ON m.id = c.module_id
        WHERE m.code = 'StockSync' AND c.name IN ('api_url', 'api_token')
    ON DUPLICATE KEY UPDATE value = VALUES(value);

    Run again, it changes the two values without adding rows.

  4. With MESSENGER_TRANSPORT_DSN=doctrine://default in .env.local and no worker running, change the stock of a combination in the back office. The page answers at once and the messages wait: "Save all combinations" saves every row of the product, so one message per combination is queued, and the handler skips those whose quantity did not change:

    php Thelia messenger:stats async

  5. Run a worker until the queue is empty, then read what the warehouse received and what the module recorded:

    php Thelia messenger:consume async --time-limit=10 -vv
    cat /tmp/warehouse/received.log
    SELECT * FROM stock_sync_log ORDER BY id DESC LIMIT 1;

Without a queue, the push happens during the save of the stock, in the same request. Stop the fake warehouse and change the stock again to see a retry: the worker logs the failure, waits 30 seconds, and sends the message to failed after three more attempts.

Turning the module off​

A queue accepts the messages of active modules only. Messages of StockSync still waiting when the module is turned off can no longer be read: the worker reads each of them as Thelia\Messenger\Message\UndecodableJob and sends it to failed. Nothing is lost and nothing loops. The Background jobs screen lists them as Unreadable job: <reason> with the original class StockSync\Message\PushStock. Delete them there, and let the nightly task push the stock again once the module is back.

To avoid this, wait for php Thelia messenger:stats async to reach zero before turning the module off. The same happens when a message class is renamed or removed, or when its constructor changes so that queued content no longer fits: keep the old class until the queue is empty.

Idempotence and replay​

A handler can run more than once for the same message:

  • a retry after a failure that happened halfway, for instance when the API accepted the push but the database write that follows failed,
  • a replay from the failed jobs, by messenger:failed:retry or from the back office,
  • a worker killed hard (SIGKILL, out of memory) while handling the message: the Doctrine transport redelivers it after redeliver_timeout, one hour by default,
  • two messages for the same row, dispatched by two saves in a row.

Write every handler so that running it twice has the effect of running it once:

  • carry ids in the message, and read the current state of the row in the handler,
  • check whether the work is already done before doing it (stock_sync_state above),
  • send absolute values to remote services (a quantity, a status), never deltas such as "remove 2", which a replay would apply twice,
  • record what was done in the same transaction as the other writes of the handler,
  • return without error when the row no longer exists or is in a state that makes the job pointless. Throwing would only send the message to failed for nothing.

Use UnrecoverableMessageHandlingException when a retry cannot succeed (data refused, configuration missing). The message goes to failed at once, where it can be replayed once the cause is fixed.

Test the replay: run the handler twice in a test and check that the effect happened once, as in Testing a message, its handler and its dispatch.

Known limits​

Two cases are not guarded by the core in this version:

  • A failed job replayed from the back office and with messenger:failed:retry at the same moment may be sent twice: the command reads the job without holding it while it asks what to do with it. Replay a job from one place at a time.
  • A mail sent with the Symfony mailer directly, instead of MailerFactory, carries no identifier: when the queue hands its job out again (a worker stopped between the mail server accepting the mail and the job being acknowledged), it is sent a second time. Send the mails of a module through MailerFactory, which a worker never delivers twice.

Both are a reason more to keep every handler idempotent.

Testing a message, its handler and its dispatch​

Three things are worth a test in a module that queues work: that the right message is dispatched, that the handler does its work, and that running the handler twice has the effect of running it once. A fourth test checks that the queue of the shop accepts the message, which otherwise only shows when a real queue refuses it.

Test them apart. The tests of the core run without a queue: tests/bootstrap.php empties MESSENGER_TRANSPORT_DSN and MESSENGER_HEAVY_TRANSPORT_DSN, so async is synchronous and a message dispatched through the bus of the shop is handled at once, by the real handler. A module test that goes through that bus cannot tell "dispatched" from "handled", and its handler calls the real API. Give the listener a bus of its own instead, and call the handler directly.

The dispatch, without a database​

The listener receives a MessageBus that sends PushStock to an InMemoryTransport and has no handler. getSent() returns what was queued; nothing ran.

// local/modules/StockSync/Tests/StockChangedListenerTest.php
<?php

declare(strict_types=1);

namespace StockSync\Tests;

use PHPUnit\Framework\TestCase;
use StockSync\EventListener\StockChangedListener;
use StockSync\Message\PushStock;
use Symfony\Component\DependencyInjection\ServiceLocator;
use Symfony\Component\Messenger\MessageBus;
use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;
use Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport;
use Symfony\Component\Messenger\Transport\Sender\SendersLocator;
use Thelia\Core\Event\ProductSaleElement\ProductSaleElementUpdateEvent;
use Thelia\Model\Product;

final class StockChangedListenerTest extends TestCase
{
    public function testSavingACombinationQueuesOnePushForIt(): void
    {
        $queue = new InMemoryTransport();

        // A bus that sends PushStock to the in-memory queue and handles nothing.
        $bus = new MessageBus([
            new SendMessageMiddleware(new SendersLocator(
                [PushStock::class => ['stock_sync']],
                new ServiceLocator(['stock_sync' => static fn (): InMemoryTransport => $queue]),
            )),
        ]);

        (new StockChangedListener($bus))->onStockChanged(new ProductSaleElementUpdateEvent(new Product(), 42));

        $sent = $queue->getSent();
        self::assertCount(1, $sent);
        self::assertEquals(new PushStock(42), $sent[0]->getMessage());
    }
}

The handler and its replay, with the test database​

The handler reads the database, so its test extends Thelia\Test\IntegrationTestCase, which runs each test in a transaction rolled back at the end (see Testing). The handler is built by hand with a MockHttpClient, which answers in place of the warehouse and counts the requests. Calling it twice with the same message is the replay: the second call must not reach the warehouse, and only one push is logged.

// local/modules/StockSync/Tests/PushStockHandlerTest.php
<?php

declare(strict_types=1);

namespace StockSync\Tests;

use StockSync\Message\PushStock;
use StockSync\MessageHandler\PushStockHandler;
use StockSync\Model\StockSyncLogQuery;
use StockSync\StockSync;
use Symfony\Component\HttpClient\MockHttpClient;
use Symfony\Component\HttpClient\Response\MockResponse;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
use Thelia\Messenger\Serializer\AllowedClassesSerializer;
use Thelia\Model\ProductSaleElements;
use Thelia\Test\IntegrationTestCase;

final class PushStockHandlerTest extends IntegrationTestCase
{
    private ProductSaleElements $combination;

    protected function setUp(): void
    {
        parent::setUp();

        $factory = $this->createFixtureFactory();
        $product = $factory->product($factory->category(), $factory->taxRule(), $factory->currency());

        $this->combination = $product->getDefaultSaleElements();
        $this->combination->setQuantity(7)->save($this->getPropelConnection());

        StockSync::setConfigValue('api_url', 'https://warehouse.test');
        StockSync::setConfigValue('api_token', 'test-token');
    }

    public function testThePushSendsTheQuantityOfTheCombination(): void
    {
        $warehouse = new MockHttpClient(new MockResponse('', ['http_code' => 200]));

        $this->handler($warehouse)(new PushStock($this->combination->getId()));

        self::assertSame(1, $warehouse->getRequestsCount());
        self::assertSame(1, $this->pushesLogged());
    }

    public function testAReplayCallsTheWarehouseOnce(): void
    {
        $warehouse = new MockHttpClient([
            new MockResponse('', ['http_code' => 200]),
            new MockResponse('', ['http_code' => 200]),
        ]);
        $handler = $this->handler($warehouse);

        $handler(new PushStock($this->combination->getId()));
        $handler(new PushStock($this->combination->getId()));

        self::assertSame(1, $warehouse->getRequestsCount());
        self::assertSame(1, $this->pushesLogged());
    }

    public function testARefusalIsNotRetried(): void
    {
        $warehouse = new MockHttpClient(new MockResponse('', ['http_code' => 422]));

        $this->expectException(UnrecoverableMessageHandlingException::class);

        $this->handler($warehouse)(new PushStock($this->combination->getId()));
    }

    public function testTheQueueOfTheShopAcceptsTheMessage(): void
    {
        $serializer = $this->getService(AllowedClassesSerializer::class);

        $decoded = $serializer->decode($serializer->encode(new Envelope(new PushStock(42))));

        self::assertEquals(new PushStock(42), $decoded->getMessage());
    }

    private function handler(MockHttpClient $warehouse): PushStockHandler
    {
        return new PushStockHandler($warehouse, 'https://shop.test');
    }

    private function pushesLogged(): int
    {
        return StockSyncLogQuery::create()
            ->filterByProductSaleElementsId($this->combination->getId())
            ->count();
    }
}

testTheQueueOfTheShopAcceptsTheMessage() writes and reads the message with the serializer of the queues, Thelia\Messenger\Serializer\AllowedClassesSerializer. It fails with The message class "StockSync\Message\PushStock" is not one the shop queues when the message has no handler, lives outside the namespace of the module, or the module is not active in the test database.

Running the tests​

In a checkout of thelia/thelia, the module in local/modules/StockSync is registered and activated in the test database by test:prepare, which also runs its postActivation() and so creates its tables. Prepare the test database once the module is there, then run its tests with the same wrapper as the core suites:

ddev exec composer test:prepare
ddev exec composer phpc ./vendor/bin/phpunit local/modules/StockSync/Tests

phpc sets APP_ENV=test and unsets the DATABASE_* and MESSENGER_* variables DDEV injects, so the tests use the test database and no queue (see Testing).

A project created with thelia/thelia-project ships bin/test-prepare, but neither PHPUnit nor a PHPUnit configuration. To run the tests of a module there, require phpunit/phpunit as a dev dependency and give the project a phpunit.xml.dist and a tests/bootstrap.php like those of thelia/thelia.

Check that a test can fail before trusting it: remove the "already pushed" check of the handler and testAReplayCallsTheWarehouseOnce() fails; remove the dispatch() of the listener and testSavingACombinationQueuesOnePushForIt() fails.