Add an option to delay job execution when adding to the queue (#52)

* add an option to delay job execution when adding to the queue

* add delay exception test

* fix phpcpd

* tests: make sure the job is added to the queue with the proper delay

Author: michalsn

docs/running-queues.md

@@ -12,7 +12,7 @@ This will cause command to check for the new jobs every 10 seconds if the queue
 
 ### With CRON
 
-Using queues with CRON is more challenging, but definitely doable. You can use command like this:
+Using queues with CRON is more challenging but definitely doable. You can use command like this:
 
     php spark queue:work emails -max-jobs 20 --stop-when-empty
 
@@ -63,6 +63,22 @@ But we can also run the worker like this:
 
 This way, worker will consume jobs with the `low` priority and then with `high`. The order set in the config file is override.
 
+### Delaying jobs
+
+Normally, when we add jobs to a queue, they are run in the order in which we added them to the queue (FIFO - first in, first out).
+Of course, there are also priorities, which we described in the previous section. But what about the scenario where we want to run a job, but not earlier than in 5 minutes?
+
+This is where job delay comes into play. We measure the delay in seconds.
+
+```php
+// This job will be run not sooner than in 5 minutes
+service('queue')->setDelay(5 * MINUTE)->push('emails', 'email', ['message' => 'Email sent no sooner than 5 minutes from now']);
+```
+
+Note that there is no guarantee that the job will run exactly in 5 minutes. If many new jobs are added to the queue (without a delay), it may take a long time before the delayed job is actually executed.
+
+We can also combine delayed jobs with priorities.
+
 ### Running many instances of the same queue
 
 As mentioned above, sometimes we may want to have multiple instances of the same command running at the same time. The queue is safe to use in that scenario with all databases as long as you keep the `skipLocked` to `true` in the config file. Only for SQLite3 driver, this setting is not relevant as it provides atomicity without the need for explicit concurrency control.

phpunit.xml.dist

@@ -12,7 +12,7 @@
          stopOnFailure="false"
          stopOnIncomplete="false"
          stopOnSkipped="false"
-         cacheDirectory=".phpunit.cache"
+         cacheDirectory="build/.phpunit.cache"
          beStrictAboutCoverageMetadata="true">
 
     <coverage includeUncoveredFiles="true">

src/Exceptions/QueueException.php

@@ -51,4 +51,9 @@ public static function forIncorrectQueuePriority(string $priority, string $queue
     {
         return new self(lang('Queue.incorrectQueuePriority', [$priority, $queue]));
     }
+
+    public static function forIncorrectDelayValue(): static
+    {
+        return new self(lang('Queue.incorrectDelayValue'));
+    }
 }

src/Handlers/BaseHandler.php

@@ -29,6 +29,7 @@ abstract class BaseHandler
 {
     protected QueueConfig $config;
     protected ?string $priority = null;
+    protected ?int $delay       = null;
 
     abstract public function name(): string;
 
@@ -62,6 +63,20 @@ public function setPriority(string $priority): static
         return $this;
     }
 
+    /**
+     * Set delay for job queue (in seconds).
+     */
+    public function setDelay(int $delay): static
+    {
+        if ($delay < 0) {
+            throw QueueException::forIncorrectDelayValue();
+        }
+
+        $this->delay = $delay;
+
+        return $this;
+    }
+
     /**
      * Retry failed job.
      *

src/Handlers/DatabaseHandler.php

@@ -56,10 +56,10 @@ public function push(string $queue, string $job, array $data): bool
             'priority'     => $this->priority,
             'status'       => Status::PENDING->value,
             'attempts'     => 0,
-            'available_at' => Time::now(),
+            'available_at' => Time::now()->addSeconds($this->delay ?? 0),
         ]);
 
-        $this->priority = null;
+        $this->priority = $this->delay = null;
 
         return $this->jobModel->insert($queueJob, false);
     }

src/Handlers/PredisHandler.php

@@ -64,19 +64,21 @@ public function push(string $queue, string $job, array $data): bool
 
         helper('text');
 
+        $availableAt = Time::now()->addSeconds($this->delay ?? 0);
+
         $queueJob = new QueueJob([
             'id'           => random_string('numeric', 16),
             'queue'        => $queue,
             'payload'      => new Payload($job, $data),
             'priority'     => $this->priority,
             'status'       => Status::PENDING->value,
             'attempts'     => 0,
-            'available_at' => Time::now(),
+            'available_at' => $availableAt,
         ]);
 
-        $result = $this->predis->zadd("queues:{$queue}:{$this->priority}", [json_encode($queueJob) => Time::now()->timestamp]);
+        $result = $this->predis->zadd("queues:{$queue}:{$this->priority}", [json_encode($queueJob) => $availableAt->timestamp]);
 
-        $this->priority = null;
+        $this->priority = $this->delay = null;
 
         return $result > 0;
     }

src/Handlers/RedisHandler.php

@@ -81,19 +81,21 @@ public function push(string $queue, string $job, array $data): bool
 
         helper('text');
 
+        $availableAt = Time::now()->addSeconds($this->delay ?? 0);
+
         $queueJob = new QueueJob([
             'id'           => random_string('numeric', 16),
             'queue'        => $queue,
             'payload'      => new Payload($job, $data),
             'priority'     => $this->priority,
             'status'       => Status::PENDING->value,
             'attempts'     => 0,
-            'available_at' => Time::now(),
+            'available_at' => $availableAt,
         ]);
 
-        $result = (int) $this->redis->zAdd("queues:{$queue}:{$this->priority}", Time::now()->timestamp, json_encode($queueJob));
+        $result = (int) $this->redis->zAdd("queues:{$queue}:{$this->priority}", $availableAt->timestamp, json_encode($queueJob));
 
-        $this->priority = null;
+        $this->priority = $this->delay = null;
 
         return $result > 0;
     }

src/Language/en/Queue.php

@@ -24,4 +24,5 @@
     'incorrectPriorityFormat' => 'The priority name should consists only lowercase letters.',
     'tooLongPriorityName'     => 'The priority name is too long. It should be no longer than 64 letters.',
     'incorrectQueuePriority'  => 'This queue has incorrectly defined priority: "{0}" for the queue: "{1}".',
+    'incorrectDelayValue'     => 'The number of seconds of delay must be a positive integer.',
 ];

tests/DatabaseHandlerTest.php

@@ -112,6 +112,9 @@ public function testPushWithPriority(): void
         ]);
     }
 
+    /**
+     * @throws ReflectionException
+     */
     public function testPushAndPopWithPriority(): void
     {
         Time::setTestNow('2023-12-29 14:15:16');
@@ -148,6 +151,38 @@ public function testPushAndPopWithPriority(): void
         $this->assertSame($payload, $result->payload);
     }
 
+    /**
+     * @throws Exception
+     */
+    public function testPushWithDelay(): void
+    {
+        Time::setTestNow('2023-12-29 14:15:16');
+
+        $handler = new DatabaseHandler($this->config);
+        $result  = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key' => 'value']);
+
+        $this->assertTrue($result);
+
+        $availableAt = 1703859376;
+
+        $this->seeInDatabase('queue_jobs', [
+            'queue'        => 'queue-delay',
+            'payload'      => json_encode(['job' => 'success', 'data' => ['key' => 'value']]),
+            'available_at' => $availableAt,
+        ]);
+
+        $this->assertEqualsWithDelta(MINUTE, $availableAt - Time::now()->getTimestamp(), 1);
+    }
+
+    public function testPushWithDelayException(): void
+    {
+        $this->expectException(QueueException::class);
+        $this->expectExceptionMessage('The number of seconds of delay must be a positive integer.');
+
+        $handler = new DatabaseHandler($this->config);
+        $handler->setDelay(-60);
+    }
+
     /**
      * @throws ReflectionException
      */

tests/PredisHandlerTest.php

@@ -102,6 +102,27 @@ public function testPushWithPriority(): void
         $this->assertSame(['key' => 'value'], $queueJob->payload['data']);
     }
 
+    /**
+     * @throws ReflectionException
+     */
+    public function testPushWithDelay(): void
+    {
+        Time::setTestNow('2023-12-29 14:15:16');
+
+        $handler = new PredisHandler($this->config);
+        $result  = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key' => 'value']);
+
+        $this->assertTrue($result);
+
+        $predis = self::getPrivateProperty($handler, 'predis');
+        $this->assertSame(1, $predis->zcard('queues:queue-delay:default'));
+
+        $task     = $predis->zrangebyscore('queues:queue-delay:default', '-inf', Time::now()->addSeconds(MINUTE)->timestamp, ['limit' => [0, 1]]);
+        $queueJob = new QueueJob(json_decode((string) $task[0], true));
+        $this->assertSame('success', $queueJob->payload['job']);
+        $this->assertSame(['key' => 'value'], $queueJob->payload['data']);
+    }
+
     public function testPushException(): void
     {
         $this->expectException(QueueException::class);