Download the PHP package solophp/job-queue without Composer

JobQueue

A small, production-ready, database-backed job queue for PHP.

Correctness-first: atomic claim, visibility timeout for dead workers, exponential backoff on failures, no long-lived transactions. Small surface area, no framework lock-in.

Features

• Atomic claim via FOR UPDATE SKIP LOCKED (MySQL 8+, MariaDB 10.6+, PostgreSQL 9.5+); on SQLite the single-writer model provides the same guarantee.
• Visibility timeout — jobs whose worker died (crashed, OOM-killed, timed out) are automatically returned to the queue.
• Exponential backoff — retries are rescheduled with baseDelay * 2^retry_count, capped at 1 hour.
• No long transactions — handle() runs outside any transaction, so external side-effects cannot be rolled back.
• Typed jobs via JobInterface + PSR-11 dependency injection through createFromContainer.
• UTC timestamps — all DB times written in UTC regardless of the PHP default timezone.
• PSR-3 logging with full-fidelity error context — Throwables are forwarded under the PSR-3 exception key (trace + previous chain preserved for structured log sinks).
• Observability hooks via JobQueueListener — onClaimed / onCompleted / onFailed / onReclaimed events for metrics, tracing, or context enrichment.
• Long-running worker with signal-driven graceful shutdown and max-jobs / max-runtime / max-memory limits.
• Operational API — reclaimStuck() (callable from a separate cron), getFailedJobs(), retry().
• Optional deleteOnSuccess and LockGuard for preventing overlapping workers.

Requirements

• PHP >= 8.3
• ext-json
• doctrine/dbal ^4
• One of: MySQL 8+, MariaDB 10.6+, PostgreSQL 9.5+, SQLite 3.35+

Installation

Setup

Defining a job

Every job implements Solo\Contracts\JobQueue\JobInterface and provides a static createFromContainer(ContainerInterface $container, array $data): self factory. This lets the queue persist only job data (not services) and wire up dependencies at run time.

Pushing jobs

Monitoring progress

getStats() is the recommended way to track batch progress: push N jobs with a shared type, then poll getStats($type). No batch table, no extra schema — the type column is the grouping key.

Processing jobs

What processJobs() does, in order:

1. Reclaim stuck jobs — any job stuck in in_progress longer than lockTimeout is returned to pending (or marked failed if retries are exhausted).
2. Atomic claim — a short transaction selects up to $limit pending jobs with FOR UPDATE SKIP LOCKED and marks them in_progress. Other workers skip these rows immediately.
3. Run each job outside any transaction — handle() executes, then a single UPDATE marks the job completed or failed. Side-effects are never rolled back.

On a thrown exception the job is rescheduled with exponential backoff. After maxRetries attempts it becomes failed permanently.

Delivery semantics: at-least-once

Jobs can run more than once. handle() runs outside any transaction, and markCompleted() is a separate statement. If a worker crashes (or the DB connection drops) after handle() succeeds but before markCompleted() finishes, the job stays in_progress. Once lockTimeout elapses, reclaimStuck returns it to the queue and another worker runs handle() again.

This is the standard at-least-once guarantee. Make every handler idempotent — external side effects (emails sent, API calls, row inserts) must tolerate being repeated:

• Use deduplication keys on outbound APIs (idempotency tokens).
• Prefer INSERT ... ON CONFLICT / upserts over raw inserts.
• For emails, check a "sent" flag before sending.
• For financial operations, use transaction references.

If you cannot make a handler idempotent, keep side effects minimal and accept the trade-off.

Timezones

All internal timestamps (scheduled_at, locked_at, expires_at, created_at defaults) are written in UTC regardless of the PHP default timezone. Compare timestamps in UTC on the database side too. DateTimeImmutable arguments you pass to push()/addJob() are converted to UTC internally.

Long-running worker

For production, prefer the Worker daemon over a cron tick. It blocks on the queue, processes batches, sleeps when empty, and shuts down cleanly on SIGTERM / SIGINT so jobs in flight finish before the process exits. Restart limits (max-jobs / max-runtime / max-memory) let a supervisor (systemd, Supervisord, K8s) recycle the process to release memory and pick up fresh code.

If the pcntl extension is unavailable, signal handlers are silently skipped — exit relies on limits or stop().

Observability hooks

Implement Solo\Contracts\JobQueue\JobQueueListener to ship metrics, open tracing spans, or enrich your log context per job:

Listener exceptions propagate to the caller, so keep handlers cheap and non-throwing.

Operational API

Preventing overlapping workers

Use LockGuard in cron-driven workers to avoid two instances of the same script running concurrently. It uses flock() — atomic, cross-platform, auto-released by the OS on process exit.

Integration with Async Event-Dispatcher

JobQueue integrates with SoloPHP Async Event-Dispatcher:

API

Testing

Tests run against SQLite in-memory, covering atomic claim, retry/backoff, visibility timeout, type filtering, expiry, DI wiring, validation branches and logger integration. Schema SQL for MySQL and PostgreSQL is tested via mocked connections.

License

MIT — see LICENSE.