Driver-based queue: Redis and memory drivers, priority, unique jobs, backoff, leases and crash recovery

.github/workflows/tests.yml

@@ -15,6 +15,55 @@ jobs:
 
     name: PHP 8.5
 
+    services:
+      mysql:
+        image: mysql:8.0
+        env:
+          MYSQL_ROOT_PASSWORD: password
+          MYSQL_DATABASE: queue_test
+        ports:
+          - 3306:3306
+        options: >-
+          --health-cmd="mysqladmin ping -h 127.0.0.1 -ppassword"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=10
+
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_PASSWORD: password
+          POSTGRES_DB: queue_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd="pg_isready -U postgres"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=10
+
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd="redis-cli ping"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=10
+
+    env:
+      # A backend that is down must fail the run, not silently skip its tests.
+      QUEUE_TEST_REQUIRE_BACKENDS: 1
+      QUEUE_TEST_REDIS_URL: redis://127.0.0.1:6379
+      QUEUE_TEST_MYSQL_DSN: mysql:host=127.0.0.1;port=3306;dbname=queue_test;charset=utf8mb4
+      QUEUE_TEST_MYSQL_USER: root
+      QUEUE_TEST_MYSQL_PASS: password
+      QUEUE_TEST_PGSQL_DSN: pgsql:host=127.0.0.1;port=5432;dbname=queue_test
+      QUEUE_TEST_PGSQL_USER: postgres
+      QUEUE_TEST_PGSQL_PASS: password
+      QUEUE_TEST_PGSQL_SCHEMA: queue_test
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -23,7 +72,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: '8.5'
-          extensions: dom, curl, libxml, mbstring, zip
+          extensions: dom, curl, libxml, mbstring, zip, pdo_sqlite, pdo_mysql, pdo_pgsql, pcntl, posix
           ini-values: error_reporting=E_ALL
           tools: composer:v2
           coverage: none

composer.json

@@ -17,7 +17,8 @@
         "doppar/framework": "4.*",
         "mockery/mockery": "^1.6",
         "phpstan/phpstan": "^2.1",
-        "phpunit/phpunit": "^13.3"
+        "phpunit/phpunit": "^13.3",
+        "predis/predis": "^3.3"
     },
     "autoload": {
         "psr-4": {
@@ -37,6 +38,9 @@
         "php": "^8.5",
         "opis/closure": "^4.5"
     },
+    "suggest": {
+        "predis/predis": "Required to use the \"redis\" queue driver."
+    },
     "prefer-stable": true,
     "scripts": {
         "analyse": "vendor/bin/phpstan analyse --memory-limit=1G"

config/queue.php

@@ -0,0 +1,67 @@
+<?php
+
+return [
+
+    /*
+    |--------------------------------------------------------------------------
+    | Default Queue Connection
+    |--------------------------------------------------------------------------
+    |
+    | The connection used when a job does not ask for one. Choose one of the
+    | connections defined below.
+    |
+    */
+
+    'default' => env('QUEUE_CONNECTION', 'database'),
+
+    /*
+    |--------------------------------------------------------------------------
+    | Queue Connections
+    |--------------------------------------------------------------------------
+    |
+    | Supported drivers: "database", "redis", "memory".
+    |
+    | "lease" is how many seconds a worker may hold a job before it is handed
+    | to another worker, which is how jobs of crashed workers are recovered.
+    | Set it above the longest time a job needs; jobs with a timeout renew it
+    | automatically while they run.
+    |
+    | Register your own driver with Queue::extend('name', fn ($config, $clock) => ...).
+    |
+    */
+
+    'connections' => [
+
+        'database' => [
+            'driver' => 'database',
+            // Database connection to use; null uses the default connection.
+            'connection' => null,
+            'table' => 'queue_jobs',
+            'failed_table' => 'failed_jobs',
+            'lease' => 90,
+        ],
+
+        'redis' => [
+            'driver' => 'redis',
+            // Requires predis/predis. Same shape as the "redis" cache store.
+            'connection' => env('REDIS_URL', 'redis://127.0.0.1:6379'),
+            'options' => [
+                'parameters' => [
+                    'password' => env('REDIS_PASSWORD', null),
+                    'database' => env('REDIS_DB', 0),
+                ],
+            ],
+            // The braces are a Redis Cluster hash tag; keep them.
+            'prefix' => '{doppar_queue}',
+            'lease' => 90,
+        ],
+
+        // Keeps jobs in memory for the current process only. For tests.
+        'memory' => [
+            'driver' => 'memory',
+            'lease' => 90,
+        ],
+
+    ],
+
+];

src/Attributes/Queueable.php

@@ -13,12 +13,18 @@ class Queueable
      * @param int|null $delayFor
      * @param string|null $onQueue
      * @param int|null $timeout
+     * @param int|null $priority
+     * @param string|null $onConnection
+     * @param int|array<int, int>|null $backoff
      */
     public function __construct(
         public ?int $tries = null,
         public ?int $retryAfter = null,
         public ?int $delayFor = null,
         public ?string $onQueue = null,
-        public ?int $timeout = null
+        public ?int $timeout = null,
+        public ?int $priority = null,
+        public ?string $onConnection = null,
+        public int|array|null $backoff = null
     ) {}
 }

src/Commands/Concerns/ReadsOptions.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Doppar\Queue\Commands\Concerns;
+
+trait ReadsOptions
+{
+    /**
+     * Read a command option as a non-empty string, or null when it was not given.
+     *
+     * @param string $key
+     * @return string|null
+     */
+    protected function stringOption(string $key): ?string
+    {
+        $value = $this->option($key);
+
+        return is_string($value) && $value !== '' ? $value : null;
+    }
+}

src/Commands/QueueFailedCommand.php

@@ -3,35 +3,37 @@
 namespace Doppar\Queue\Commands;
 
 use Phaseolies\Console\Schedule\Command;
-use Doppar\Queue\Models\FailedJob;
+use Doppar\Queue\Commands\Concerns\ReadsOptions;
+use Doppar\Queue\QueueManager;
 
 class QueueFailedCommand extends Command
 {
+    use ReadsOptions;
+
     /**
-     * The name of the console command.
+     * The name and signature of the console command.
      *
      * @var string
      */
-    protected $name = 'queue:failed';
+    protected $name = 'queue:failed {--connection=}';
 
     /**
-     * The command description.
+     * The console command description.
      *
      * @var string
      */
     protected $description = 'List all failed jobs';
 
     /**
-     * Execute the console command
-     * Example: php pool queue:failed
+     * Execute the console command.
      *
      * @return int
      */
     public function handle(): int
     {
-        $failedJobs = FailedJob::orderBy('failed_at', 'desc')->get();
+        $failedJobs = app(QueueManager::class)->connection($this->stringOption('connection'))->failedJobs();
 
-        if ($failedJobs->isEmpty()) {
+        if ($failedJobs === []) {
             $this->info("No failed jobs found.");
             return Command::SUCCESS;
         }
@@ -41,20 +43,18 @@ public function handle(): int
         $table->setHeaders(['ID', 'Job', 'Queue', 'Failed At']);
 
         foreach ($failedJobs as $job) {
-            $payload = $job->payload;
-            $data = unserialize($payload);
+            $data = @unserialize($job->payload);
 
             $jobClass = null;
-            if ($data && isset($data['job']) && is_object($data['job'])) {
+            if (is_array($data) && isset($data['job']) && is_object($data['job'])) {
                 $jobClass = get_class($data['job']);
             }
 
-            $failedAt = date('Y-m-d H:i:s', $job->failed_at);
             $table->addRow([
                 $job->id,
                 $jobClass,
                 $job->queue,
-                $failedAt
+                date('Y-m-d H:i:s', $job->failedAt),
             ]);
         }
 

src/Commands/QueueFlushCommand.php

@@ -3,58 +3,50 @@
 namespace Doppar\Queue\Commands;
 
 use Phaseolies\Console\Schedule\Command;
-use Doppar\Queue\Models\FailedJob;
+use Doppar\Queue\Commands\Concerns\ReadsOptions;
+use Doppar\Queue\QueueManager;
 
 class QueueFlushCommand extends Command
 {
+    use ReadsOptions;
+
     /**
-     * The name of the console command.
+     * The name and signature of the console command.
      *
      * @var string
      */
-    protected $name = 'queue:flush {--id=}';
+    protected $name = 'queue:flush {--id=} {--connection=}';
 
     /**
-     * The command description.
+     * The console command description.
      *
      * @var string
      */
     protected $description = 'Delete failed job(s) by ID or all if no ID is provided';
 
     /**
-     * Execute the console command
-     * Example: php pool queue:flush --id=1
+     * Execute the console command.
      *
      * @return int
      */
     public function handle(): int
     {
-        $id = $this->option('id');
+        $driver = app(QueueManager::class)->connection($this->stringOption('connection'));
+        $id = $this->stringOption('id');
 
         if ($id) {
-            return $this->flushJobById($id);
-        }
+            if (!$driver->forgetFailed($id)) {
+                $this->error("Failed job with ID {$id} not found.");
+                return Command::FAILURE;
+            }
 
-        FailedJob::query()
-            ->cursor(function (FailedJob $failedJob) {
-                $failedJob->delete();
-                $this->info("✔ Job with ID {$failedJob->id} has been deleted.");
-            });
-
-        return Command::SUCCESS;
-    }
-
-    protected function flushJobById(int $id): int
-    {
-        $failedJob = FailedJob::find($id);
+            $this->info("✔ Job with ID {$id} has been deleted.");
 
-        if (!$failedJob) {
-            $this->error("Failed job with ID {$id} not found.");
-            return Command::FAILURE;
+            return Command::SUCCESS;
         }
 
-        $failedJob->delete();
-        $this->info("✔ Job with ID {$failedJob->id} has been deleted.");
+        $count = $driver->flushFailed();
+        $this->info("✔ {$count} failed job(s) deleted.");
 
         return Command::SUCCESS;
     }