Início / Optitravel / Core

Job Queue

This module lives in includes/classes/Core/Queue and provides an async queue system for the app using the Cache class.

What you get

• Dynamic queues (emails, simulations, logs, etc).
• Multiple workers in parallel.
• Retries (max_attempts) per job.
• Failed jobs list per queue with automatic requeue by worker.
• Delayed jobs.
• Handler registry by job name.

Folder structure

• Queue.php: small facade to publish jobs.
• QueueManager.php: Cache operations (push, reserve, ack, retry, failed, delayed).
• QueueWorker.php: worker runtime loop and job execution.
• Worker.php: CLI entrypoint to run workers.
• Handlers.php: default map of job-name => callable.
• Handlers/: concrete handler classes.

Job model

Each job is stored in the Cache as JSON with fields like:

• id
• queue
• job
• payload
• attempts
• max_attempts
• created_at

Cache keys used per queue:

• queue:<name>:pending
• queue:<name>:processing
• queue:<name>:failed
• queue:<name>:exhausted
• queue:<name>:delayed

Built-in metrics keys (automatic counters):

• queue:<name>:metrics:processed_total
• queue:<name>:metrics:failed_total
• queue:<name>:metrics:exhausted_total

Publish jobs

Queue availability is controlled by QUEUE_ENABLED:

• QUEUE_ENABLED=1 (default): producers and workers are enabled.
• QUEUE_ENABLED=0: pushes throw a RuntimeException, and workers stop consuming without deleting existing jobs.

The worker CLI exits without starting when disabled. The explicit --restart command remains available.

Signature:

Queue::push(string $queue, string $job, array $payload = [], int $maxAttempts = 3, int $delaySeconds = 0): string

Use App\Core\Queue\Queue:

<?php

use App\Core\Queue\Queue;

// Immediate job to queue "emails"
$jobId = Queue::push(
    'emails',
    'send_email_struct',
    [
        'system_id' => '768',
        'email_struct' => $emailStruct,
    ]
);

// Delayed job to queue "simulations" (30 seconds) with 5 max attempts
Queue::push(
    'simulations',
    'run_parallel_search',
    ['query' => 'Lisbon hotels'],
    5,
    30
);

Run workers

Default mode (recommended)

If you run without arguments, worker will:

• listen to all queues (auto-discovery);
• load Handlers.php from this same folder.

php includes/classes/Core/Queue/Worker.php

Listen specific queues

php includes/classes/Core/Queue/Worker.php --queues=emails,simulations --sleep=1

Set explicit consumer name

php includes/classes/Core/Queue/Worker.php --queues=emails --consumer-name=emails-worker-01

Override handlers file

php includes/classes/Core/Queue/Worker.php --handlers=/path/to/custom/handlers.php

Graceful restart after deploy

After deploying new PHP code, request a local worker restart:

php includes/classes/Core/Queue/Worker.php --restart

The command writes a marker to the local Redis instance. Workers detect it between jobs, finish the current job, flush pending DB metrics, then exit. Use systemd or Supervisor to start replacement workers automatically.

The Queue Dashboard lists workers with a Redis heartbeat. A worker is kept visible for up to five minutes after its last heartbeat, which avoids hiding workers while they execute longer jobs. The dashboard can request a graceful restart for one worker or all workers connected to the same Redis instance.

Arguments

• --queues (optional): comma-separated queue names. If omitted, all queues.
• --handlers (optional): custom handlers map file. If omitted, local Handlers.php.
• --consumer-name (optional): explicit worker name for logs/observability. If omitted, auto-generated as <hostname>-<pid>-<rand4hex>.
• --sleep (optional): idle wait in seconds. Default 1.
• --auto-retry-failed (optional): 1 or 0. Default 0.
• --failed-retry-every (optional): seconds between failed requeue cycles. Default 60.
• --failed-retry-limit (optional): max jobs requeued per queue per cycle. Default 25.
• --failed-retry-delay (optional): delay in seconds when requeueing from failed. Default 60.
• --failed-retry-max-requeues (optional): max times a failed job can be requeued. Default 3.
• --restart: requests a graceful restart for all local workers without starting a worker.

Automatic failed retry

Workers can requeue failed jobs automatically when enabled with --auto-retry-failed=1.

Default behavior:

• Every 60 seconds, each worker checks failed jobs.
• It moves up to 25 failed jobs per queue.
• Requeued jobs return to delayed queue with exponential backoff (base 60s).
• Each job can be requeued from failed up to 3 times.
• After that, job is moved to exhausted (stops automatic retries).

This avoids immediate hot-loop retries of poison jobs and reduces the need for manual maintenance.

Built-in metrics

QueueManager updates counters automatically, so app code does not need to manage this:

• processed_total: incremented when a job is acknowledged.
• failed_total: incremented when a job reaches max attempts and is moved to failed.
• exhausted_total: incremented when a failed job reaches retry limit and is moved to exhausted.

You can read totals with:

• processedTotal(string $queue): int
• failedTotal(string $queue): int
• exhaustedTotal(string $queue): int

Optional DB aggregated metrics (batch flush)

Workers can also write aggregated execution metrics to central DB in batches (not per job).

Enable with env:

• QUEUE_DB_METRICS_ENABLED=1
• QUEUE_DB_METRICS_FLUSH_EVERY_SECONDS=30 (default)
• QUEUE_DB_METRICS_FLUSH_MAX_ITEMS=200 (default)
• QUEUE_DB_METRICS_BUFFER_HARD_LIMIT=5000 (default)

Table is fixed to queue_metrics.

Recorded statuses:

• success
• failed_retry
• failed_terminal

The worker keeps an in-memory buffer and flushes when either:

• time threshold is reached; or
• buffer item threshold is reached.

This reduces DB writes while still preserving operational visibility.

Required table:

CREATE TABLE queue_metrics (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    minute_bucket DATETIME NOT NULL,
    queue_name VARCHAR(100) NOT NULL,
    job_name VARCHAR(190) NOT NULL,
    system_id VARCHAR(64) NULL,
    status ENUM('success','failed_retry','failed_terminal') NOT NULL,
    events_count INT UNSIGNED NOT NULL DEFAULT 0,
    duration_ms_total BIGINT UNSIGNED NOT NULL DEFAULT 0,
    attempts_total BIGINT UNSIGNED NOT NULL DEFAULT 0,
    created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (id),
    UNIQUE KEY uq_queue_metrics (minute_bucket, queue_name, job_name, system_id, status),
    KEY idx_queue_metrics_status_bucket (status, minute_bucket),
    KEY idx_queue_metrics_queue_bucket (queue_name, minute_bucket),
    KEY idx_queue_metrics_system_bucket (system_id, minute_bucket)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

Optional max queue length

You can define QUEUE_MAX_LEN in env to keep queue structures trimmed automatically (similar to Redis stream MAXLEN).

• QUEUE_MAX_LEN=0 (default): disabled (no trimming).
• QUEUE_MAX_LEN=10000: keeps up to 10k entries per list/zset key.

To reduce Redis overhead in high-throughput systems:

• QUEUE_TRIM_EVERY=10 (default): check trim every 10 writes. Set 1 for strict mode.
• QUEUE_TRIM_OVERFLOW_PCT=10 (default): trim only after passing max + 10% (hysteresis).

When enabled, QueueManager trims:

• pending, failed, exhausted lists with LTRIM.
• delayed zset by removing oldest score entries.

Important: trimming drops older jobs when limit is reached.

Parallelism

Yes, you can run multiple workers at the same time.

Examples:

# 3 workers for email-heavy workload
php includes/classes/Core/Queue/Worker.php --queues=emails
php includes/classes/Core/Queue/Worker.php --queues=emails
php includes/classes/Core/Queue/Worker.php --queues=emails

# 1 worker dedicated to simulations
php includes/classes/Core/Queue/Worker.php --queues=simulations

Redis pop operations used are atomic, so the same pending job is not consumed twice in normal operation.

Default handlers included

File: includes/classes/Core/Queue/Handlers.php

Mapped jobs:

• send_email_struct => EmailHandler::sendFromStruct
• write_content_log => LogHandler::writeContentInfo

Payload send_email_struct

[
    'system_id' => '1047',
    'email_struct' => $emailStruct,
    'e_fact' => false,
    'ref' => '',
    'certify_pdf' => false,
    'remove_files' => true,
    'save_log' => false,
    'smtp_debug' => '',
    'language' => 'pt',
]

Payload write_content_log

[
    'system_id' => '1047',
    'domain' => 'queue',
    'subject' => 'Async Process',
    'email' => 'ops@example.com',
    'content' => 'Job completed',
    'context' => ['job_id' => '123'],
]

Create new handlers

Step 1: create handler class

Example file includes/classes/Core/Queue/Handlers/SimulationHandler.php:

<?php declare(strict_types=1);

namespace App\Core\Queue\Handlers;

use RuntimeException;

class SimulationHandler
{
    public static function run(array $payload): void
    {
        $simulationId = (int) ($payload['simulation_id'] ?? 0);
        if ($simulationId <= 0) {
            throw new RuntimeException('Missing simulation_id');
        }

        // Do your simulation processing here.
    }
}

Step 2: register in Handlers.php

use App\Core\Queue\Handlers\SimulationHandler;

return [
    'run_simulation' => [SimulationHandler::class, 'run'],
];

Step 3: publish to a queue

Queue::push('simulations', 'run_simulation', ['simulation_id' => 987], 5);

Step 4: run a worker that can process that queue

php includes/classes/Core/Queue/Worker.php --queues=simulations

Handler contract and errors

• Handler callable receives:
• arg1: array $payload
• arg2: array $rawJob (optional)
• Throw exception to mark execution as failed.
• Worker increments attempts and retries until max_attempts.
• After max attempts, job goes to failed list.
• If failed auto-retry is enabled, job may be requeued from failed with backoff.
• Once failed_requeues reaches the configured limit, job moves to exhausted and is no longer retried.

Operational recommendations

• Create dedicated queues by domain (emails, billing, integrations, simulations).
• Use higher worker count for slow I/O queues.
• Keep handlers idempotent when possible.
• Keep payloads small; store large blobs in DB/storage and pass IDs.

---

↑ (Última atualização: 01/10/2026)