Create LimitAttemptsExtension

Author: sheldonreiff

docs/en/index.rst

@@ -85,6 +85,13 @@ Create a Job class::
     {
         use LogTrait;
 
+        /**
+         * The maximum number of times the job may be attempted.
+         * 
+         * @var int|null
+         */
+        public static $maxAttempts = 3;
+
         public function execute(Message $message): ?string
         {
             $id = $message->getArgument('id');
@@ -118,6 +125,14 @@ The job **may** also return a null value, which is interpreted as
 ``Processor::ACK``. Failure to respond with a valid type will result in an
 interpreted message failure and requeue of the message.
 
+Properties:
+
+- ``maxAttempts``: The maximum number of times the job may be requeued as a result
+  of an exception or by explicitly returning ``Processor::REQUEUE``. If
+  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.
+
 Queueing
 --------
 
@@ -244,6 +259,7 @@ This shell can take a few different options:
 - ``--logger`` (default: ``stdout``): Name of a configured logger
 - ``--max-jobs`` (default: ``null``): Maximum number of jobs to process. Worker will exit after limit is reached.
 - ``--max-runtime`` (default: ``null``): Maximum number of seconds to run. Worker will exit after limit is reached.
+- ``--max-attempts`` (default: ``null``): Maximum number of times each job will be attempted. Maximum attempts defined on a job will override this value.
 - ``--verbose`` or ``-v`` (default: ``null``): Provide verbose output, displaying the current values for:
 
   - Max Iterations

src/Command/WorkerCommand.php

@@ -22,6 +22,7 @@
 use Cake\Console\ConsoleOptionParser;
 use Cake\Core\Configure;
 use Cake\Log\Log;
+use Cake\Queue\Consumption\LimitAttemptsExtension;
 use Cake\Queue\Consumption\LimitConsumedMessagesExtension;
 use Cake\Queue\Queue\Processor;
 use Cake\Queue\QueueManager;
@@ -86,6 +87,12 @@ public function getOptionParser(): ConsoleOptionParser
             'default' => null,
             'short' => 'r',
         ]);
+        $parser->addOption('max-attempts', [
+            'help' => 'Maximum number of times each job will be attempted.'
+                . ' Maximum attempts defined on a job will override this value.',
+            'default' => null,
+            'short' => 'a',
+        ]);
         $parser->setDescription(
             'Runs a queue worker that consumes from the named queue.'
         );
@@ -104,6 +111,7 @@ protected function getQueueExtension(Arguments $args, LoggerInterface $logger):
     {
         $extensions = [
             new LoggerExtension($logger),
+            new LimitAttemptsExtension((int)$args->getOption('max-attempts') ?: null),
         ];
 
         if (!is_null($args->getOption('max-jobs'))) {

src/Consumption/LimitAttemptsExtension.php

@@ -0,0 +1,80 @@
+<?php
+declare(strict_types=1);
+
+namespace Cake\Queue\Consumption;
+
+use Cake\Queue\Job\Message;
+use Enqueue\Consumption\Context\MessageResult;
+use Enqueue\Consumption\MessageResultExtensionInterface;
+use Enqueue\Consumption\Result;
+use Interop\Queue\Processor;
+
+class LimitAttemptsExtension implements MessageResultExtensionInterface
+{
+    /**
+     * The property key used to set the number of times a message was attempted.
+     *
+     * @var string
+     */
+    public const ATTEMPTS_PROPERTY = 'attempts';
+
+    /**
+     * The maximum number of times a job may be attempted. $maxAttempts defined on a
+     * Job will override this value.
+     *
+     * @var int|null
+     */
+    protected $maxAttempts;
+
+    /**
+     * @param int|null $maxAttempts The maximum number of times a job may be attempted.
+     * @return void
+     */
+    public function __construct(?int $maxAttempts = null)
+    {
+        $this->maxAttempts = $maxAttempts;
+    }
+
+    /**
+     * @param \Enqueue\Consumption\Context\MessageResult $context The result of the message after it was processed.
+     * @return void
+     */
+    public function onResult(MessageResult $context): void
+    {
+        if ($context->getResult() !== Processor::REQUEUE) {
+            return;
+        }
+
+        $message = $context->getMessage();
+
+        $jobMessage = new Message($message, $context->getContext());
+
+        $maxAttempts = $jobMessage->getMaxAttempts() ?? $this->maxAttempts;
+
+        if ($maxAttempts === null) {
+            return;
+        }
+
+        $attemptNumber = $message->getProperty(self::ATTEMPTS_PROPERTY, 0) + 1;
+
+        if ($attemptNumber >= $maxAttempts) {
+            $context->changeResult(
+                Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts))
+            );
+
+            return;
+        }
+
+        $newMessage = clone $message;
+        $newMessage->setProperty(self::ATTEMPTS_PROPERTY, $attemptNumber);
+
+        $queueContext = $context->getContext();
+        $producer = $queueContext->createProducer();
+        $consumer = $context->getConsumer();
+        $producer->send($consumer->getQueue(), $newMessage);
+
+        $context->changeResult(
+            Result::reject('A copy of the message was sent with an incremented attempt count.')
+        );
+    }
+}

src/Job/Message.php

@@ -94,6 +94,20 @@ public function getCallable()
             return $this->callable;
         }
 
+        $target = $this->getTarget();
+
+        $this->callable = Closure::fromCallable([new $target[0](), $target[1]]);
+
+        return $this->callable;
+    }
+
+    /**
+     * Get the target class and method.
+     *
+     * @return array
+     */
+    protected function getTarget(): array
+    {
         $target = $this->parsedBody['class'] ?? null;
 
         if (!is_array($target) || count($target) !== 2) {
@@ -103,9 +117,7 @@ public function getCallable()
             ));
         }
 
-        $this->callable = Closure::fromCallable([new $target[0](), $target[1]]);
-
-        return $this->callable;
+        return $target;
     }
 
     /**
@@ -129,6 +141,20 @@ public function getArgument($key = null, $default = null)
         return Hash::get($data, $key, $default);
     }
 
+    /**
+     * The maximum number of attempts allowed by the job.
+     *
+     * @return null|int
+     */
+    public function getMaxAttempts(): ?int
+    {
+        $target = $this->getTarget();
+
+        $class = $target[0];
+
+        return $class::$maxAttempts ?? null;
+    }
+
     /**
      * @return string
      * @psalm-suppress InvalidToString

templates/bake/job.twig

@@ -27,6 +27,13 @@ use Interop\Queue\Processor;
  */
 class {{ name }}Job implements JobInterface
 {
+    /**
+     * The maximum number of times the job may be attempted.
+     * 
+     * @var int|null
+     */
+    public static $maxAttempts = 3;
+
     /**
      * Executes logic for {{ name }}Job
      *

tests/TestCase/Command/WorkerCommandTest.php

@@ -19,9 +19,11 @@
 use Cake\Core\Configure;
 use Cake\Log\Log;
 use Cake\Queue\QueueManager;
+use Cake\Queue\Test\TestCase\DebugLogTrait;
 use Cake\TestSuite\ConsoleIntegrationTestTrait;
 use Cake\TestSuite\TestCase;
 use TestApp\Job\LogToDebugJob;
+use TestApp\Job\RequeueJob;
 use TestApp\WelcomeMailerListener;
 
 /**
@@ -32,6 +34,7 @@
 class WorkerCommandTest extends TestCase
 {
     use ConsoleIntegrationTestTrait;
+    use DebugLogTrait;
 
     public function setUp(): void
     {
@@ -240,16 +243,31 @@ public function testQueueProcessesJobWithOtherQueue()
         $this->assertDebugLogContains('Debug job was run');
     }
 
-    protected function assertDebugLogContains($expected): void
+    /**
+     * Test max-attempts option
+     *
+     * @runInSeparateProcess
+     */
+    public function testQueueProcessesJobWithMaxAttempts()
     {
-        $log = Log::engine('debug');
-        $found = false;
-        foreach ($log->read() as $line) {
-            if (strpos($line, $expected) !== false) {
-                $found = true;
-                break;
-            }
-        }
-        $this->assertTrue($found, "Did not find `{$expected}` in logs.");
+        $config = [
+            'queue' => 'default',
+            'url' => 'file:///' . TMP . DS . 'queue',
+            'receiveTimeout' => 100,
+        ];
+        Configure::write('Queue', ['default' => $config]);
+
+        Log::setConfig('debug', [
+            'className' => 'Array',
+            'levels' => ['notice', 'info', 'debug'],
+        ]);
+
+        QueueManager::setConfig('default', $config);
+        QueueManager::push(RequeueJob::class);
+        QueueManager::drop('default');
+
+        $this->exec('queue worker --max-attempts=3 --max-jobs=1 --logger=debug --verbose');
+
+        $this->assertDebugLogContainsExactly('RequeueJob is requeueing', 3);
     }
 }

tests/TestCase/Consumption/LimitAttemptsExtensionTest.php

@@ -0,0 +1,104 @@
+<?php
+declare(strict_types=1);
+
+namespace Cake\Queue\Test\TestCase\Job;
+
+use Cake\Log\Log;
+use Cake\Queue\Consumption\LimitAttemptsExtension;
+use Cake\Queue\Consumption\LimitConsumedMessagesExtension;
+use Cake\Queue\Queue\Processor as QueueProcessor;
+use Cake\Queue\QueueManager;
+use Cake\Queue\Test\TestCase\DebugLogTrait;
+use Cake\TestSuite\TestCase;
+use Enqueue\Consumption\ChainExtension;
+use Psr\Log\NullLogger;
+use TestApp\Job\MaxAttemptsIsThreeJob;
+use TestApp\Job\RequeueJob;
+
+class LimitAttemptsExtensionTest extends TestCase
+{
+    use DebugLogTrait;
+
+    /**
+     * @beforeClass
+     * @after
+     */
+    public static function dropConfigs()
+    {
+        Log::drop('debug');
+        QueueManager::drop('default');
+    }
+
+    public function testMessageShouldBeRequeuedIfMaxAttemptsIsNotSet()
+    {
+        $consume = $this->setupQeueue();
+
+        QueueManager::push(RequeueJob::class);
+
+        $consume();
+
+        $count = $this->debugLogCount('RequeueJob is requeueing');
+        $this->assertGreaterThanOrEqual(10, $count);
+    }
+
+    public function testMessageShouldBeRequeuedUntilMaxAttemptsIsReached()
+    {
+        $consume = $this->setupQeueue();
+
+        QueueManager::push(MaxAttemptsIsThreeJob::class, []);
+
+        $consume();
+
+        $this->assertDebugLogContainsExactly('MaxAttemptsIsThreeJob is requeueing', 3);
+    }
+
+    public function testMessageShouldBeRequeuedIfGlobalMaxAttemptsIsNotSet()
+    {
+        $consume = $this->setupQeueue();
+
+        QueueManager::push(RequeueJob::class);
+
+        $consume();
+
+        $count = $this->debugLogCount('RequeueJob is requeueing');
+        $this->assertGreaterThanOrEqual(10, $count);
+    }
+
+    public function testMessageShouldBeRequeuedUntilGlobalMaxAttemptsIsReached()
+    {
+        $consume = $this->setupQeueue([3]);
+
+        QueueManager::push(MaxAttemptsIsThreeJob::class, ['succeedAt' => 10]);
+
+        $consume();
+
+        $this->assertDebugLogContainsExactly('MaxAttemptsIsThreeJob is requeueing', 3);
+    }
+
+    protected function setupQeueue($extensionArgs = [])
+    {
+        Log::setConfig('debug', [
+            'className' => 'Array',
+            'levels' => ['debug'],
+        ]);
+
+        QueueManager::setConfig('default', [
+            'url' => 'file:///' . TMP . DS . uniqid('queue'),
+            'receiveTimeout' => 100,
+        ]);
+
+        $client = QueueManager::engine('default');
+
+        $processor = new QueueProcessor(new NullLogger());
+        $client->bindTopic('default', $processor);
+
+        $extension = new ChainExtension([
+            new LimitConsumedMessagesExtension(1),
+            new LimitAttemptsExtension(...$extensionArgs),
+        ]);
+
+        return function () use ($client, $extension) {
+            $client->consume($extension);
+        };
+    }
+}