merge: v5-stability into v5.x

Brings the job-state, delay and redis adapter fixes plus the Pest suite onto the
release branch. Worker.php was touched by both sides and merged without conflict:
the crash integration and the recovery sweep sit in different parts of run().

# Conflicts:
#	src/Worker.php

Author: mychidarko

alchemy.yml

@@ -0,0 +1,29 @@
+app:
+  - src
+
+tests:
+  engine: pest
+  parallel: false
+  paths:
+    - tests
+  files:
+    - '*.test.php'
+  coverage:
+    local: false
+    actions: false
+
+lint:
+  preset: PSR12
+
+actions:
+  run:
+    - lint
+  os:
+    - ubuntu-latest
+  php:
+    versions:
+      - '8.2'
+      - '8.3'
+  events:
+    - push
+    - pull_request

composer.json

@@ -33,10 +33,23 @@
     "require-dev": {
         "leafs/db": "*",
         "leafs/redis": "*",
+        "pestphp/pest": "*",
         "friendsofphp/php-cs-fixer": "^3.0"
     },
     "scripts": {
-        "format": "vendor/bin/php-cs-fixer fix --config=.php_cs.dist.php --allow-risky=yes"
+        "format": "vendor/bin/php-cs-fixer fix --config=.php_cs.dist.php --allow-risky=yes",
+        "alchemy": "@php vendor/bin/alchemy all",
+        "test": "@php vendor/bin/alchemy test",
+        "lint": "@php vendor/bin/alchemy lint",
+        "fmt": "@php vendor/bin/alchemy fmt",
+        "analyse": "@php vendor/bin/alchemy analyse",
+        "refactor": "@php vendor/bin/alchemy refactor",
+        "ci": "@php vendor/bin/alchemy ci"
+    },
+    "config": {
+        "allow-plugins": {
+            "pestphp/pest-plugin": true
+        }
     },
     "require": {
         "php": "^8.2",

phpunit.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="vendor/phpunit/phpunit/phpunit.xsd"
+         bootstrap="vendor/autoload.php"
+         colors="true">
+    <testsuites>
+        <testsuite name="Queue">
+            <directory suffix=".test.php">tests</directory>
+        </testsuite>
+    </testsuites>
+</phpunit>

src/Job.php

@@ -14,7 +14,7 @@ abstract class Job implements Dispatchable
     /**
      * Data to pass to the job
      */
-    protected static $data = [];
+    protected array $data = [];
 
     /**
      * Queue instance
@@ -43,24 +43,10 @@ abstract class Job implements Dispatchable
     protected $delayBeforeRetry = 0;
 
     /**
-     * Number of seconds to wait before archiving a job that has not yet been processed
+     * Number of seconds to wait before archiving a job that has not yet been processed.
+     * Set to 0 to never expire.
      */
-    protected $expire = 60;
-
-    /**
-     * Force the worker to process the job, even if it has expired or has reached its maximum number of retries
-     */
-    protected $force = false;
-
-    /**
-     * The maximum amount of memory the job is allowed to consume (MB)
-     */
-    protected $memory = 128;
-
-    /**
-     * The number of seconds a child process can run before being killed.
-     */
-    protected $timeout = 60;
+    protected $expire = 3600;
 
     /**
      * The maximum number of times the job may be attempted.
@@ -83,16 +69,11 @@ public function fromQueue($job, $config, $queue)
         $this->job = $job;
         $this->queue = $queue;
 
-        if (isset($config['data'])) {
-            static::$data = $config['data'];
-        }
+        $this->data = $config['data'] ?? [];
 
         $this->delay = $config['delay'] ?? 0;
         $this->delayBeforeRetry = $config['delayBeforeRetry'] ?? 0;
-        $this->expire = $config['expire'] ?? 60;
-        $this->force = $config['force'] ?? false;
-        $this->memory = $config['memory'] ?? 128;
-        $this->timeout = $config['timeout'] ?? 60;
+        $this->expire = $config['expire'] ?? 3600;
         $this->tries = $config['tries'] ?? 3;
 
         return $this;
@@ -108,21 +89,16 @@ public function getJobId()
         return $this->job['id'];
     }
 
-    /**
-     * Handle delay for job
-     */
-    public function handleDelay()
-    {
-        echo "Job #{$this->job['id']} is delayed for {$this->delay} seconds\n";
-        sleep($this->delay);
-    }
-
     /**
      * Check if job has expired
      */
     public function hasExpired()
     {
-        return $this->job['created_at'] < (time() - $this->expire);
+        if ((int) $this->expire === 0) {
+            return false;
+        }
+
+        return ((int) ($this->job['created_at'] ?? time())) < (time() - $this->expire);
     }
 
     /**
@@ -131,7 +107,7 @@ public function hasExpired()
     public function handleExpiry()
     {
         echo "Job #{$this->job['id']} has expired\n";
-        $this->queue->pop($this->job['id']);
+        $this->setStatus('expired');
     }
 
     /**
@@ -151,12 +127,41 @@ public function setStatus($status)
     }
 
     /**
-     * Retry job
+     * Retry job. If the retry limit has been reached, the job
+     * is marked as failed with the exception recorded instead.
+     *
+     * @param \Throwable|null $exception The exception that caused the failure
+     */
+    public function retry(?\Throwable $exception = null)
+    {
+        if (($this->job['retry_count'] + 1) >= $this->tries) {
+            $this->fail($exception);
+
+            return;
+        }
+
+        $this->queue->retryFailedJob(
+            $this->job['id'],
+            $this->job['retry_count'],
+            $this->delayBeforeRetry ?? 0
+        );
+    }
+
+    /**
+     * Mark the job as permanently failed and record the exception
+     *
+     * @param \Throwable|null $exception The exception that caused the failure
      */
-    public function retry()
+    public function fail(?\Throwable $exception = null)
     {
-        sleep($this->delayBeforeRetry ?? 0);
-        $this->queue->retryFailedJob($this->job['id'], $this->job['retry_count']);
+        $exceptionDump = null;
+
+        if ($exception) {
+            $trace = explode("\n", $exception->getTraceAsString());
+            $exceptionDump = $exception->getMessage() . "\n" . implode("\n", array_slice($trace, 0, 5));
+        }
+
+        $this->queue->markJobAsFailed($this->job['id'], $exceptionDump);
     }
 
     /**
@@ -167,30 +172,29 @@ public function retry()
      */
     public function release($delay = 0)
     {
-        sleep($delay);
-
         $this->queue->push([
             'class' => $this->job['class'],
             'status' => 'pending',
             'retry_count' => $this->job['retry_count'] + 1,
+            'available_at' => time() + $delay,
             'config' => json_encode([
                 'delay' => $this->delay,
                 'delayBeforeRetry' => $this->delayBeforeRetry,
                 'expire' => $this->expire,
-                'force' => $this->force,
-                'memory' => $this->memory,
-                'timeout' => $this->timeout,
                 'tries' => $this->tries,
-                'data' => static::$data,
+                'data' => $this->data,
             ]),
         ]);
+
+        $this->removeFromQueue();
     }
 
     public static function with($data)
     {
-        static::$data[] = $data;
+        $instance = new static();
+        $instance->data = [$data];
 
-        return new static();
+        return $instance;
     }
 
     public function stack()
@@ -204,18 +208,15 @@ public function getConfig()
             'delay' => $this->delay,
             'delayBeforeRetry' => $this->delayBeforeRetry,
             'expire' => $this->expire,
-            'force' => $this->force,
-            'memory' => $this->memory,
-            'timeout' => $this->timeout,
             'tries' => $this->tries,
-            'data' => static::$data,
+            'data' => $this->data,
         ];
     }
 
     public function trigger()
     {
         $this->queue->setJobStatus($this->job['id'], 'processing');
-        $this->handle(...static::$data);
+        $this->handle(...$this->data);
     }
 
     public function removeFromQueue()

src/Queue.php

@@ -81,20 +81,22 @@ public function getNextJob()
     /**
      * Mark job as failed
      * @param string|int $id The id of the job to mark as failed
+     * @param string|null $exception The exception that caused the failure
      */
-    public function markJobAsFailed($id)
+    public function markJobAsFailed($id, $exception = null)
     {
-        $this->adapter->markJobAsFailed($id);
+        $this->adapter->markJobAsFailed($id, $exception);
     }
 
     /**
      * Retry failed job
      * @param string|int $id The id of the job to retry
      * @param string|int $retryCount The number of times the job has been retried
+     * @param int $delay Seconds to wait before the job becomes available again
      */
-    public function retryFailedJob($id, $retryCount = 0)
+    public function retryFailedJob($id, $retryCount = 0, $delay = 0)
     {
-        $this->adapter->retryFailedJob($id, $retryCount);
+        $this->adapter->retryFailedJob($id, $retryCount, $delay);
     }
 
     /**

src/Queue/Adapters/Adapter.php

@@ -49,13 +49,23 @@ public function getNextJob();
 
     /**
      * Mark job as failed
+     * @param string|int $id The id of the job to mark as failed
+     * @param string|null $exception The exception that caused the failure
      */
-    public function markJobAsFailed($id);
+    public function markJobAsFailed($id, $exception = null);
 
     /**
      * Retry failed job
+     * @param string|int $id The id of the job to retry
+     * @param string|int $retryCount The number of times the job has been retried
+     * @param int $delay Seconds to wait before the job becomes available again
      */
-    public function retryFailedJob($id, $retryCount);
+    public function retryFailedJob($id, $retryCount, $delay = 0);
+
+    /**
+     * Reset jobs stuck in 'processing' (eg. after a worker crash) back to 'pending'
+     */
+    public function resetStuckJobs();
 
     /**
      * Disconnect