Persistent Queue

What It Is

Key Benefits

• Fault Tolerance & Durability: If a worker crashes mid-task or the entire server goes down, the job remains safely stored. Once the system recovers, it resumes processing without data loss.
• Asynchronous Execution: Heavy tasks run in the background, allowing your web application to respond instantly to users without making them wait for slow operations to finish.
• Load Leveling & Traffic Spikes: Queues act as a buffer during high-traffic surges, catching incoming jobs and feeding them to your workers at a steady, manageable pace so your database and external APIs don't get overwhelmed.
• Independent Scalability: You can scale your web servers and your background worker pools independently depending on where your resource bottlenecks lie.

When to Use Them (Common Use Cases)

• Transactional & Marketing Emails / SMS: Sending password resets, welcome emails, or bulk notifications via third-party providers (like Mailgun, SendGrid, or Twilio).
• File & Media Processing: Generating PDF reports, resizing uploaded profile images, converting video formats, or parsing large CSV/Excel imports.
• Third-Party API Integrations: Communicating with external webhooks, payment gateways (e.g., Stripe processing webhooks), or syncing data with external CRMs where network latency or API rate limits could slow down your app.
• Scheduled / Batch Operations: Running routine database cleanups, generating daily analytics summaries, or processing batch data updates overnight.

// In a controller: queue the job and answer the visitor straight away
dispatch(new SendWelcomeEmail($user->id));

return $this->view('auth/registered.twig');

Quick Start

1. Create the tables (once per project):

php rhapsody queue:install --migrate

This publishes a migration into db/migrations and runs it. Without --migrate it only publishes the file; run php rhapsody migrate yourself when you are ready.

2. Write a job:

namespace App\Jobs;

use Rhapsody\Core\Contracts\JobInterface;

final class SendWelcomeEmail implements JobInterface
{
    public function __construct(public readonly int $userId) {}

    public function handle(UserRepository $users, WelcomeMailer $mailer): void
    {
        $mailer->sendTo($users->find($this->userId));
    }
}

3. Dispatch it:

dispatch(new SendWelcomeEmail($user->id));

4. Run a worker:

php rhapsody queue:work

Writing Jobs

A job is a small class that implements JobInterface and has a public handle() method.

• handle() gets its dependencies from the container. Type-hint the services you need, exactly as you would in a controller. The worker calls it through Container::call(), so aliases, singletons and closure bindings all apply.
• The constructor holds what you dispatch with. Pass plain values: an id, a string, a number. Look the full record up inside handle() so the job always works with fresh data.
• No closures or open resources in constructor arguments. The job is stored as text, and a job that cannot be stored is refused when you dispatch it, with a message naming the class.

Reacting to a Permanent Failure

Implement FailableJobInterface to be told when a job has used up all its attempts. It runs once, after the job has been moved to the failed jobs.

use Rhapsody\Core\Contracts\FailableJobInterface;
use Throwable;

final class ChargeInvoice implements FailableJobInterface
{
    public function __construct(public readonly int $invoiceId) {}

    public function handle(Billing $billing): void
    {
        $billing->charge($this->invoiceId);
    }

    public function failed(Throwable $exception): void
    {
        error_log("Invoice {$this->invoiceId} could not be charged: " . $exception->getMessage());
    }
}

failed() is called directly, so it does not get container injection. If it throws, the error is logged and ignored.

Dispatching

use Rhapsody\Core\Queue\QueueManager;

// 1. The global helper
dispatch(new SendWelcomeEmail($id));
dispatch(new SendReminder($id), queue: 'emails', delay: 3600);

// 2. Inject the manager (easier to test and to swap)
public function __construct(private QueueManager $queue) {}

$jobId = $this->queue->dispatch(new SendWelcomeEmail($id));
$waiting = $this->queue->size('emails');

Both return the new job's id. A negative delay or an invalid queue name throws InvalidArgumentException.

Retries

If handle() throws, the job is put back and tried again after a wait. When it runs out of attempts it moves to the failed jobs. Tune this per job with attributes:

use Rhapsody\Core\Queue\Attributes\Backoff;
use Rhapsody\Core\Queue\Attributes\Timeout;
use Rhapsody\Core\Queue\Attributes\Tries;

#[Tries(5)]       // give up after 5 attempts
#[Backoff(30)]    // wait 30s, then 60s, 120s, 240s
#[Timeout(45)]    // abort an attempt that runs longer than 45s
final class ImportCatalog implements JobInterface { /* ... */ }

Running the Worker

php rhapsody queue:work                          # process the "default" queue
php rhapsody queue:work --queue=high,default     # "high" is always checked first
php rhapsody queue:work --stop-when-empty        # drain the queue, then exit

On a Server You Control: Run It as a Daemon

Keep the worker running with a process manager such as Supervisor, which restarts it whenever it stops. The limit options let it recycle itself regularly.

[program:rhapsody-queue]
command=php /var/www/app/rhapsody queue:work --queue=high,default --max-time=3600
directory=/var/www/app
autostart=true
autorestart=true
numprocs=2
process_name=%(program_name)s_%(process_num)02d
stopwaitsecs=70

Run as many workers as you like: each job is claimed by exactly one of them. stopwaitsecs gives a worker time to finish its current job when the process manager stops it.

On Shared Hosting: Run It from Cron

No process manager? Start a worker every minute. It processes whatever is waiting and exits.

* * * * * cd /path/to/app && php rhapsody queue:work --stop-when-empty --max-time=55 >> /dev/null 2>&1

If a run is still busy when the next one starts, that is fine: the two simply share the work.

Failed Jobs

A job that has used all its attempts, or whose stored data can no longer be read, is moved to the failed_jobs table together with the error and stack trace. Every failure is also written to storage/logs/queue.log.

php rhapsody queue:failed          # list them
php rhapsody queue:retry 12        # put one back on its queue
php rhapsody queue:retry all       # put them all back
php rhapsody queue:flush           # delete them all (asks first; --force skips the question)

A retried job starts again with a fresh attempt count.

Commands

Good to Know

• A job may run more than once. If a worker dies after finishing the work but before the job is marked done, the job is handed out again. Write jobs so that running twice does no harm (check whether the email was already sent, use an upsert instead of an insert).
• Dispatch after you commit. Dispatch once the data the job needs is saved. A worker can pick the job up almost immediately, and it will not see changes that are still inside an open transaction.
• Keep #[Timeout] below the retry window. A reservation that is older than 90 seconds is treated as abandoned and the job is handed to another worker. The default job timeout is 60 seconds, which fits; if you raise it, raise the driver's retryAfter too (see below).
• Timeouts and a graceful stop need pcntl. On Linux and macOS the worker aborts a job that exceeds its timeout, and finishes its current job before exiting on SIGTERM or Ctrl+C. Windows (for example a local WAMP setup) has no pcntl: jobs run without a time limit, and stopping the worker mid-job leaves the job to be retried once its reservation expires.
• A crashing job cannot loop forever. Every time a job is picked up it counts as an attempt. A job that keeps killing its worker still ends up in the failed jobs once its attempts are used.
• Jobs are signed. Each job is stored with a signature made from your APP_KEY. A row that was altered, or inserted by hand, is rejected and moved to the failed jobs. If you rotate the key, list the old one in APP_PREVIOUS_KEYS so jobs already waiting can still be read.
• Renaming or deleting a job class makes waiting jobs of that class unreadable, and they go to the failed jobs. Let the queue drain first, or keep the old class until it has.
• Restart workers after a deploy. A running worker keeps the old code in memory. Stop it (a process manager will start a fresh one) or let --max-time recycle it. Cron workers pick up new code on their own.
• Queue order. Within a queue, jobs run oldest first. To make some work more urgent, give it its own queue and list that queue first: --queue=high,default.
• Works on every supported database. MySQL, PostgreSQL, SQL Server and SQLite use the same tables and the same queries. The priority and meta columns are reserved for add-on modules and are not used by core.
• If the tables are missing, the worker and dispatch() stop with a message telling you to run queue:install --migrate.

Using a Different Database Connection

By default the queue uses the same database as the rest of your app. To point it somewhere else, or to change the retry window, bind your own QueueInterface in your project's bootstrap.php. The worker, dispatch() and every queue command pick it up automatically.

use Rhapsody\Core\Bootstrap\ServiceRegistrar;
use Rhapsody\Core\Contracts\QueueInterface;
use Rhapsody\Core\Queue\DatabaseQueue;

return static function (ServiceRegistrar $services): void {
    $services->singleton(QueueInterface::class, fn () => new DatabaseQueue(
        static fn (): PDO => new PDO($dsn, $user, $password, [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]),
        retryAfter: 120,
    ));
};

The jobs and failed_jobs tables must exist in that database. QueueInterface is a small contract (push, pop, release, delete, fail, plus the failed-job methods), so a Redis or SQS driver can be dropped in the same way.

Related Documentation

• Deferred Tasks
• The Service Container
• Console Commands (CLI)
• Encryption
• Logging