Queue Model

Overview

Logidav has separate queue mechanisms. The general Doctrine queue stores rows in mz_queue; its command polls those rows and invokes the processor named in each row. Some general product processors publish messages to RabbitMQ. A different RabbitMQ consumer handles status messages and updates mz_queue. Meduse-specific rows use a second table, mz_meduse_queue, with a separate command. Meduse's Laravel jobs and FTP file handoffs are additional mechanisms, not aliases for either Logidav table. These boundaries are visible in the mz_queue entity, the general processor command, the RabbitMQ publisher, and the separate mz_meduse_queue entity.

View diagram source

flowchart LR
    PRODUCERS[Logidav producers] --> APPDB[(mz_queue)]
    APPDB --> RUN[menzzo:queue:processor<br/>action=runQueues]
    RUN --> PROCESSORS[Container-resolved processors]
    PROCESSORS -->|selected product processors publish| PUTQ[(RabbitMQ<br/>logidav.api.product.put.queue)]
    STATUSQ[(RabbitMQ<br/>logidav.api.product.messages.exchange.queue)] --> CHECK[menzzo:queue:processor<br/>action=checkQueues]
    CHECK -->|status by bulk UUID| APPDB

    MEDDB[(mz_meduse_queue)] --> MEDRUN[meduse:queue:processor<br/>action=runQueues]
    MEDRUN --> MEDPROC[Meduse-specific processor]
    MEDPROC -->|code-defined file output| FTP[(Meduse FTP connector)]
    FTP -->|stock, price, and other file readers| SCHED[Meduse scheduled file commands]
    FTP -->|new-product import only| LARAVEL[Meduse Laravel queue job]

Diagram

The diagram describes source-level connections, not a verified deployment topology. See Meduse integration, Meduse connector folders, Meduse queue details, and scheduled Meduse connector jobs.

Task Lifecycle

There is no single lifecycle shared by SQL rows, broker messages, and file imports:

View diagram source

flowchart TD
    CREATE[Create mz_queue row<br/>Pending] --> RUN[runQueues selects eligible rows]
    RUN --> EXEC[Resolve row model and call processor]
    EXEC --> RESULT[Persist processor result and queue status]
    EXEC -->|some product processors publish| PUT[RabbitMQ product.put queue]
    PUT -. downstream status reply .-> STATUS[RabbitMQ product.messages.exchange.queue]
    STATUS --> CHECK[checkQueues consumes status]
    CHECK --> UPDATE[Update mz_queue row by Magento bulk UUID]

Diagram

runQueues selection and dispatch select rows by status, age, and optionally run_category, then group them by model before invoking the processor. A separate action can revisit eligible Processing or Error rows. Retry timing and completion semantics depend on the processor; do not infer a universal maximum-retry state machine from the table fields alone.

The separate mz_meduse_queue command selects Pending rows grouped by model and executes each model. Its Meduse-specific product processor marks rows and can write a CSV to the FTP syncProduct/pending folder. See Meduse queue details for the code-defined path and its producer limitation.

Queue Entity Fields

AppBundle\Entity\Queue maps to mz_queue. Its source entity defines these fields:

Field	Description
model	Container service/class identifier invoked by the queue processor.
status	Stored status; the entity lists Pending, Processing, Complete, Success, and Error.
body	JSON payload stored with the task.
run_category	Optional category used to select a processor run.
number_of_trials	Attempt counter used by processing and retry eligibility logic.
magento_bulk_uuid	Correlation identifier used to match a Magento status message to a queue row.
last_sent_at	Timestamp used by processor eligibility logic.
success_message, error_message	Processor or downstream status details.

MeduseBundle\Entity\MeduseQueue is a separate entity mapped to mz_meduse_queue; it has its own body, model, status, and result-message fields. Do not merge its status behavior with mz_queue.

QueueProcessorInterface

The QueueProcessorInterface sends queue entities to general SQL queue processors and requires a result array:

interface QueueProcessorInterface
{
    /** @param Queue[] $queues */
    public function execute(array $queues): array;
}

The generic command resolves the row's model from the Symfony container and calls execute() with the grouped queue entities. Register each processor so the container can resolve its identifier. The Meduse-specific ProductMeduseDataUpdateProcessor also has an execute() method, but it reads mz_meduse_queue itself and emits an FTP file; it is not the generic RabbitMQ consumer.

Queue Processor Command

The command names below document source interfaces; they are not run instructions. Any production use requires approval for the exact command and target.

The general SQL queue command is QueueProcessorCommand:

Reference only

This example is not an instruction to run the command. Production use requires approval for this exact command and target.

php bin/console menzzo:queue:processor --action=runQueues

Its source also provides a distinct RabbitMQ status-consumer action:

Reference only

This example is not an instruction to run the command. Production use requires approval for this exact command and target.

php bin/console menzzo:queue:processor --action=checkQueues

runQueues polls mz_queue, resolves processors, and persists their results. checkQueues consumes logidav.api.product.messages.exchange.queue; its callback interprets messages as Magento status updates, updates mz_queue by magento_bulk_uuid, and acknowledges each message. (consumer; status update and acknowledgement)

The similarly named Meduse queue command is separate (source):

Reference only

This example is not an instruction to run the command. Production use requires approval for this exact command and target.

php bin/console meduse:queue:processor --action=runQueues

It polls mz_meduse_queue, not AppBundle\Entity\Queue. In the inspected source, QueueService::addQueue() temporarily forces Meduse-queue insertion off, and MeduseQueueService is unimplemented. The separate Meduse table and processor therefore should not be presented as the active destination of normal addQueue() calls.

Concurrency Prevention with LockableTrait

Both the general queue command and the Meduse queue command use Symfony's LockableTrait to avoid overlapping invocations. The general command releases its lock in a finally block. This source evidence does not establish that every cronjob uses the trait or how commands are scheduled in a deployment.

Cronjobs vs Queues

Mechanism	Source-backed behavior
General SQL queue	menzzo:queue:processor --action=runQueues processes eligible mz_queue rows when invoked. The source does not establish a deployment cadence.
RabbitMQ publisher	Selected general product processors publish named messages, including logidav.api.product.put.queue.
RabbitMQ status consumer	menzzo:queue:processor --action=checkQueues runs a consume/wait loop for logidav.api.product.messages.exchange.queue.
Meduse SQL queue	meduse:queue:processor --action=runQueues selects pending mz_meduse_queue records by model.
Meduse FTP polling	Meduse scheduled commands scan connector folders at configured intervals; source schedules include one-minute product commands, a five-minute price reader, and a thirty-minute stock reader. (schedule source; scheduled jobs)
Meduse Laravel jobs	New-product file imports are pushed to a Laravel queue named process-new-product. Its backend is selected by QUEUE_CONNECTION; source does not prove the deployed adapter.

Avoid describing the SQL command as a continuously running RabbitMQ consumer, or claiming a production supervisor/cron cadence without deployment evidence.

Which Services Use Queues

Source path	Mechanism established in source
ProductDataUpdateProcessor	Processes mz_queue rows and publishes product update JSON to RabbitMQ queue logidav.api.product.put.queue.
QueueProcessorCommand::checkQueues	Consumes Magento status messages from logidav.api.product.messages.exchange.queue and updates correlated mz_queue rows.
ProductMeduseDataUpdateProcessor	Reads pending mz_meduse_queue rows and can write a CSV to Meduse FTP syncProduct/pending. QueueService::addQueue() currently disables routing inserts to that table.
Meduse GenerateSyncJobsForCommand / ProcessNewProduct	Reads Logidav new-product JSON files and pushes a Laravel job to process-new-product; the backend depends on deployment configuration.

The previous page also named SaleService, ShipmentService, ProductQtyLogService, and ERP services as queue users. Those paths were not traced in this Meduse-focused source audit, so their exact producers and processors remain unverified here. For the Meduse integration's file and queue boundaries, see Meduse connectors, Meduse queues, and scheduled Meduse jobs.

For operational diagnosis, see the Queue Debugging Runbook and Replay Failed Jobs.