Merge pull request #28 from netlogix/feat/in-memory-retry-for-scheduling

fix: Improve database task scheduling retry mechanism

Author: paxuclus

Classes/Domain/Scheduler.php

@@ -269,30 +269,37 @@ protected function scheduleJob(ScheduledJob $job): void
                        job      = :job,
                        claimed  = :claimed
             MySQL;
-        $this->dbal
-            ->executeQuery(
-                $statement,
-                [
-                    'groupname' => $job->getGroupName(),
-                    'identifier' => $job->getIdentifier(),
-                    'duedate' => $job->getDuedate(),
-                    'queue' => $job->getQueueName(),
-                    'job' => serialize($job->getJob()),
-                    'incarnation' => $job->getIncarnation(),
-                    'claimed' => $job->getClaimed(),
-                    'running' => $job->isRunning() ? 1 : 0,
-                ],
-                [
-                    'groupname' => Types::STRING,
-                    'identifier' => Types::STRING,
-                    'duedate' => Types::DATETIME_IMMUTABLE,
-                    'queue' => Types::STRING,
-                    'job' => Types::BLOB,
-                    'incarnation' => Types::INTEGER,
-                    'claimed' => Types::STRING,
-                    'running' => Types::BOOLEAN,
-                ]
-            );
+
+        (new Retry())
+            /**
+             * @see http://backoffcalculator.com/?attempts=5&rate=1&interval=0.1
+             */
+            ->withExponentialBackoff(retryInterval: 0.05, maxRetries: 5)
+            ->onExceptionsOfType(RetryableException::class)
+            ->task(fn() => $this->dbal
+                ->executeQuery(
+                    $statement,
+                    [
+                        'groupname' => $job->getGroupName(),
+                        'identifier' => $job->getIdentifier(),
+                        'duedate' => $job->getDuedate(),
+                        'queue' => $job->getQueueName(),
+                        'job' => serialize($job->getJob()),
+                        'incarnation' => $job->getIncarnation(),
+                        'claimed' => $job->getClaimed(),
+                        'running' => $job->isRunning() ? 1 : 0,
+                    ],
+                    [
+                        'groupname' => Types::STRING,
+                        'identifier' => Types::STRING,
+                        'duedate' => Types::DATETIME_IMMUTABLE,
+                        'queue' => Types::STRING,
+                        'job' => Types::BLOB,
+                        'incarnation' => Types::INTEGER,
+                        'claimed' => Types::STRING,
+                        'running' => Types::BOOLEAN,
+                    ]
+                ));
         // TODO: Find a way to "trigger queueing" without cronjobs. Maybe "dynamic cronjobs" like "at".
         // TODO: On Shutdown: Add queueing job to job queue.
     }

Tests/Functional/SchedulingTest.php

@@ -3,8 +3,10 @@
 
 namespace Netlogix\JobQueue\Scheduled\Tests\Functional;
 
+use Doctrine\DBAL\Exception\DeadlockException;
 use Netlogix\JobQueue\Scheduled\Domain\Model\ScheduledJob;
 use Netlogix\JobQueue\Scheduled\Domain\Scheduler;
+use Netlogix\JobQueue\Scheduled\Service\Connection;
 
 class SchedulingTest extends TestCase
 {
@@ -103,4 +105,39 @@ public function Scheduled_jobs_contain_due_dates(): void
             $scheduledJob->getDuedate()
         );
     }
+
+    /**
+     * @test
+     * @depends A_scheduled_job_gets_persisted
+     */
+    public function Scheduling_jobs_retries_RetryableExceptions(): void
+    {
+        $connection = self::createMock(Connection::class);
+        $connection->expects(self::any())
+            ->method('executeQuery')
+            ->willThrowException(self::createStub(DeadlockException::class));
+
+        $this->scheduler->injectConnection($connection);
+
+        $start = microtime(true);
+        try {
+            $this->scheduler->schedule(
+                new ScheduledJob(
+                    self::getJobQueueJob(),
+                    self::getQueueName(),
+                    self::getDueDate(),
+                    Scheduler::DEFAULT_GROUP_NAME
+                )
+            );
+        } catch (DeadlockException $e) {
+        }
+        $end = microtime(true);
+        $delta = $end - $start;
+
+        self::assertInstanceOf(DeadlockException::class, $e);
+
+        // guesstimated value is about 1.5
+        self::assertGreaterThan(1, $delta);
+        self::assertLessThan(2, $delta);
+    }
 }