Merge pull request #108 from sheldonreiff/unique-jobs

Add support for unique jobs

Author: markstory

docs/en/index.rst

@@ -59,6 +59,16 @@ The following configuration should be present in the config array of your **conf
 
             // Whether to store failed jobs in the queue_failed_jobs table. default: false
             'storeFailedJobs' => true,
+
+            // (optional) The cache configuration for storing unique job ids. `duration`
+            // should be greater than the maximum length of time any job can be expected
+            // to remain on the queue. Otherwise, duplicate jobs may be
+            // possible. Defaults to +24 hours. Note that `File` engine is only suitable
+            // for local development.
+            // See https://book.cakephp.org/4/en/core-libraries/caching.html#configuring-cache-engines.
+            'uniqueCache' => [
+                'engine' => 'File',
+            ],
         ]
     ],
     // ...
@@ -114,6 +124,13 @@ A simple job that logs received messages would look like::
          */
         public static $maxAttempts = 3;
 
+        /**
+         * Whether there should be only one instance of a job on the queue at a time. (optional property)
+         * 
+         * @var bool
+         */
+        public static $shouldBeUnique = false;
+
         public function execute(Message $message): ?string
         {
             $id = $message->getArgument('id');
@@ -154,6 +171,12 @@ Job Properties:
   provided, this value will override the value provided in the worker command
   line option ``--max-attempts``. If a value is not provided by the job or by
   the command line option, the job may be requeued an infinite number of times.
+- ``shouldBeUnique``: If ``true``, only one instance of the job, identified by
+  it's class, method, and data, will be allowed to be present on the queue at a
+  time. Subsequent pushes will be silently dropped. This is useful for
+  idempotent operations where consecutive job executions have no benefit. For
+  example, refreshing calculated data. If ``true``, the ``uniqueCache``
+  configuration must be set.
 
 Queueing
 --------

src/Command/JobCommand.php

@@ -17,6 +17,7 @@
 namespace Cake\Queue\Command;
 
 use Bake\Command\SimpleBakeCommand;
+use Cake\Console\Arguments;
 use Cake\Console\ConsoleOptionParser;
 
 class JobCommand extends SimpleBakeCommand
@@ -47,6 +48,20 @@ public function template(): string
         return 'Cake/Queue.job';
     }
 
+    /**
+     * @inheritDoc
+     */
+    public function templateData(Arguments $arguments): array
+    {
+        $parentData = parent::templateData($arguments);
+
+        $data = [
+            'isUnique' => $arguments->getOption('unique'),
+        ];
+
+        return array_merge($parentData, $data);
+    }
+
     /**
      * Gets the option parser instance and configures it.
      *
@@ -61,6 +76,11 @@ public function buildOptionParser(ConsoleOptionParser $parser): ConsoleOptionPar
             ->setDescription('Bake a queue job class.')
             ->addArgument('name', [
                 'help' => 'The name of the queue job class to create.',
+            ])
+            ->addOption('unique', [
+                'help' => 'Whether there should be only one instance of a job on the queue at a time.',
+                'boolean' => true,
+                'default' => false,
             ]);
     }
 }

src/Command/WorkerCommand.php

@@ -25,6 +25,7 @@
 use Cake\Log\Log;
 use Cake\Queue\Consumption\LimitAttemptsExtension;
 use Cake\Queue\Consumption\LimitConsumedMessagesExtension;
+use Cake\Queue\Consumption\RemoveUniqueJobIdFromCacheExtension;
 use Cake\Queue\Listener\FailedJobsListener;
 use Cake\Queue\Queue\Processor;
 use Cake\Queue\QueueManager;
@@ -132,6 +133,7 @@ protected function getQueueExtension(Arguments $args, LoggerInterface $logger):
         $extensions = [
             new LoggerExtension($logger),
             $limitAttempsExtension,
+            new RemoveUniqueJobIdFromCacheExtension('Cake/Queue.queueUnique'),
         ];
 
         if (!is_null($args->getOption('max-jobs'))) {

src/Consumption/RemoveUniqueJobIdFromCacheExtension.php

@@ -0,0 +1,53 @@
+<?php
+declare(strict_types=1);
+
+namespace Cake\Queue\Consumption;
+
+use Cake\Cache\Cache;
+use Cake\Queue\Job\Message;
+use Cake\Queue\QueueManager;
+use Enqueue\Consumption\Context\MessageResult;
+use Enqueue\Consumption\MessageResultExtensionInterface;
+
+class RemoveUniqueJobIdFromCacheExtension implements MessageResultExtensionInterface
+{
+    /**
+     * Cache engine name.
+     *
+     * @var string
+     */
+    protected $cache;
+
+    /**
+     * @param string $cache Cache engine name.
+     * @return void
+     */
+    public function __construct(string $cache)
+    {
+        $this->cache = $cache;
+    }
+
+    /**
+     * @param \Enqueue\Consumption\Context\MessageResult $context The result of the message after it was processed.
+     * @return void
+     */
+    public function onResult(MessageResult $context): void
+    {
+        $message = $context->getMessage();
+
+        $jobMessage = new Message($message, $context->getContext());
+
+        [$class, $method] = $jobMessage->getTarget();
+
+        /** @psalm-suppress InvalidPropertyFetch */
+        if (empty($class::$shouldBeUnique)) {
+            return;
+        }
+
+        $data = $jobMessage->getArgument();
+
+        $uniqueId = QueueManager::getUniqueId($class, $method, $data);
+
+        Cache::delete($uniqueId, $this->cache);
+    }
+}

src/Job/Message.php

@@ -117,7 +117,8 @@ public function getCallable()
     /**
      * Get the target class and method.
      *
-     * @return array
+     * @return array{string, string}
+     * @psalm-return array{class-string, string}
      */
     public function getTarget(): array
     {
@@ -165,6 +166,7 @@ public function getMaxAttempts(): ?int
 
         $class = $target[0];
 
+        /** @psalm-suppress InvalidPropertyFetch */
         return $class::$maxAttempts ?? null;
     }
 

src/Listener/FailedJobsListener.php

@@ -16,7 +16,6 @@
  */
 namespace Cake\Queue\Listener;
 
-use Cake\Event\EventInterface;
 use Cake\Event\EventListenerInterface;
 use Cake\ORM\Exception\PersistenceFailedException;
 use Cake\ORM\Locator\LocatorAwareTrait;
@@ -39,10 +38,10 @@ public function implementedEvents(): array
     }
 
     /**
-     * @param \Cake\Event\EventInterface $event EventInterface
+     * @param object $event EventInterface.
      * @return void
      */
-    public function storeFailedJob(EventInterface $event): void
+    public function storeFailedJob($event): void
     {
         /** @var \Cake\Queue\Job\Message $jobMessage */
         $jobMessage = $event->getSubject();

src/QueueManager.php

@@ -17,6 +17,7 @@
 namespace Cake\Queue;
 
 use BadMethodCallException;
+use Cake\Cache\Cache;
 use Cake\Core\App;
 use Cake\Log\Log;
 use Enqueue\Client\Message as ClientMessage;
@@ -119,6 +120,16 @@ public static function setConfig($key, $config = null): void
             }
         }
 
+        if (!empty($config['uniqueCache'])) {
+            $cacheDefaults = [
+                'duration' => '+24 hours',
+            ];
+
+            $cacheConfig = array_merge($cacheDefaults, $config['uniqueCache']);
+
+            Cache::setConfig('Cake/Queue.queueUnique', $cacheConfig);
+        }
+
         /** @psalm-suppress InvalidPropertyAssignmentValue */
         static::$_config[$key] = $config;
     }
@@ -209,7 +220,33 @@ public static function push($className, array $data = [], array $options = []):
 
         $name = $options['config'] ?? 'default';
 
-        $config = static::getConfig($name);
+        $config = static::getConfig($name) + [
+            'logger' => null,
+        ];
+
+        $logger = $config['logger'] ? Log::engine($config['logger']) : null;
+
+        /** @psalm-suppress InvalidPropertyFetch */
+        if (!empty($class::$shouldBeUnique)) {
+            if (!Cache::getConfig('Cake/Queue.queueUnique')) {
+                throw new InvalidArgumentException(
+                    "$class::\$shouldBeUnique is set to `true` but `uniqueCache` configuration is missing."
+                );
+            }
+
+            $uniqueId = static::getUniqueId($class, $method, $data);
+
+            if (Cache::read($uniqueId, 'Cake/Queue.queueUnique')) {
+                if ($logger) {
+                    $logger->debug(
+                        "An identical instance of $class already exists on the queue. This push will be ignored."
+                    );
+                }
+
+                return;
+            }
+        }
+
         $queue = $options['queue'] ?? $config['queue'] ?? 'default';
 
         $message = new ClientMessage([
@@ -237,5 +274,31 @@ public static function push($className, array $data = [], array $options = []):
 
         $client = static::engine($name);
         $client->sendEvent($queue, $message);
+
+        /** @psalm-suppress InvalidPropertyFetch */
+        if (!empty($class::$shouldBeUnique)) {
+            $uniqueId = static::getUniqueId($class, $method, $data);
+
+            Cache::add($uniqueId, true, 'Cake/Queue.queueUnique');
+        }
+    }
+
+    /**
+     * @param string $class Class name
+     * @param string $method Method name
+     * @param array $data Message data
+     * @return string
+     */
+    public static function getUniqueId(string $class, string $method, array $data): string
+    {
+        sort($data);
+
+        $hashInput = implode([
+            $class,
+            $method,
+            json_encode($data),
+        ]);
+
+        return hash('md5', $hashInput);
     }
 }

templates/bake/job.twig

@@ -34,6 +34,15 @@ class {{ name }}Job implements JobInterface
      */
     public static $maxAttempts = 3;
 
+{% if isUnique %}
+    /**
+     * Whether there should be only one instance of a job on the queue at a time. (optional property)
+     * 
+     * @var bool
+     */
+    public static $shouldBeUnique = true;
+
+{% endif %}
     /**
      * Executes logic for {{ name }}Job
      *

tests/TestCase/Consumption/LimitAttemptsExtensionTest.php

@@ -40,7 +40,7 @@ public static function dropConfigs()
 
     public function testMessageShouldBeRequeuedIfMaxAttemptsIsNotSet()
     {
-        $consume = $this->setupQeueue();
+        $consume = $this->setupQueue();
 
         QueueManager::push(RequeueJob::class);
 
@@ -52,7 +52,7 @@ public function testMessageShouldBeRequeuedIfMaxAttemptsIsNotSet()
 
     public function testFailedEventIsFiredWhenMaxAttemptsIsExceeded()
     {
-        $consume = $this->setupQeueue();
+        $consume = $this->setupQueue();
 
         QueueManager::push(MaxAttemptsIsThreeJob::class, []);
 
@@ -63,7 +63,7 @@ public function testFailedEventIsFiredWhenMaxAttemptsIsExceeded()
 
     public function testMessageShouldBeRequeuedUntilMaxAttemptsIsReached()
     {
-        $consume = $this->setupQeueue();
+        $consume = $this->setupQueue();
 
         QueueManager::push(MaxAttemptsIsThreeJob::class, []);
 
@@ -74,7 +74,7 @@ public function testMessageShouldBeRequeuedUntilMaxAttemptsIsReached()
 
     public function testMessageShouldBeRequeuedIfGlobalMaxAttemptsIsNotSet()
     {
-        $consume = $this->setupQeueue();
+        $consume = $this->setupQueue();
 
         QueueManager::push(RequeueJob::class);
 
@@ -86,7 +86,7 @@ public function testMessageShouldBeRequeuedIfGlobalMaxAttemptsIsNotSet()
 
     public function testMessageShouldBeRequeuedUntilGlobalMaxAttemptsIsReached()
     {
-        $consume = $this->setupQeueue([3]);
+        $consume = $this->setupQueue([3]);
 
         QueueManager::push(MaxAttemptsIsThreeJob::class, ['succeedAt' => 10]);
 
@@ -95,7 +95,7 @@ public function testMessageShouldBeRequeuedUntilGlobalMaxAttemptsIsReached()
         $this->assertDebugLogContainsExactly('MaxAttemptsIsThreeJob is requeueing', 3);
     }
 
-    protected function setupQeueue($extensionArgs = [])
+    protected function setupQueue($extensionArgs = [])
     {
         Log::setConfig('debug', [
             'className' => 'Array',