Merge pull request #37 from netlogix/fix/postgres-reset-stale

Author: HenrikVogel

Classes/Command/SchedulerCommandController.php

@@ -61,49 +61,7 @@ public function resetStaleJobsCommand(
         string $groupName,
         int $minutes = 10
     ): void {
-        $tableName = ScheduledJob::TABLE_NAME;
-        $platform =  $this->connection->getDbal()->getDatabasePlatform();
-        if ($platform instanceof MySqlPlatform)
-        {
-            $sql = <<<MySQL
-                UPDATE {$tableName}
-                SET running = 0,
-                    claimed = '',
-                    incarnation = incarnation + 1
-                WHERE running = 1
-                  AND claimed NOT LIKE 'failed(%)'
-                  AND groupname = :groupName
-                  AND activity < NOW() - INTERVAL :minutes MINUTE
-            MySQL;
-        }
-        else if ($platform instanceof PostgreSqlPlatform || $platform instanceof PostgreSQL94Platform)
-        {
-            $sql = <<<PostgreSQL
-                UPDATE {$tableName}
-                SET running = FALSE,
-                    claimed = '',
-                    incarnation = incarnation + 1
-                WHERE running = TRUE
-                  AND claimed NOT LIKE 'failed(%)'
-                  AND groupname = :groupName
-                  AND activity < NOW() - make_interval(mins => :minutes)
-            PostgreSQL;
-        } else {
-            throw new \RuntimeException("unsupported database platform " . $this->connection->getDbal()->getDatabasePlatform()->getName());
-        }
-
-        $freed = $this->connection->executeQuery(
-            sql: $sql,
-            params: [
-                'groupName' => $groupName,
-                'minutes' => max($minutes, 1),
-            ],
-            types: [
-                'groupName' => Types::STRING,
-                'minutes' => Types::SMALLINT,
-            ],
-        )->rowCount();
-
+        $freed = $this->scheduler->resetStaleJobs($groupName, $minutes);
         if ($freed) {
             $this->outputLine('Freed ' . $freed . ' stale jobs.');
         }

Classes/Domain/AbstractScheduler.php

@@ -5,6 +5,7 @@
 namespace Netlogix\JobQueue\Scheduled\Domain;
 
 use DateTimeImmutable;
+use Doctrine\DBAL\Exception;
 use Doctrine\DBAL\Exception\RetryableException;
 use Doctrine\DBAL\Types\Types;
 use InvalidArgumentException;
@@ -40,6 +41,7 @@ abstract class AbstractScheduler implements Scheduler
     protected const SELECT_QUERY = "";
     protected const RELEASE_QUERY = "";
     protected const SCHEDULE_QUERY = "";
+    protected const RESET_STALE_JOBS_QUERY = "";
 
     public function injectConnection(Connection $connection): void
     {
@@ -274,6 +276,28 @@ public function activity(ScheduledJob $job): void
             );
     }
 
+    /**
+     * Reset stale jobs that have not changed for too long.
+     *
+     * @param string $groupName Free jobs in this group only
+     * @param int $minutes Count jobs as stale if their last activity was more than these many minutes ago
+     * @throws Exception
+     * @return int Number of freed jobs
+     */
+    public function resetStaleJobs(string $groupName, int $minutes): int {
+        return $this->dbal->executeQuery(
+            sql: static::RESET_STALE_JOBS_QUERY,
+            params: [
+                'groupName' => $groupName,
+                'minutes' => max($minutes, 1),
+            ],
+            types: [
+                'groupName' => Types::STRING,
+                'minutes' => Types::SMALLINT,
+            ],
+        )->rowCount();
+    }
+
     protected function scheduleJob(ScheduledJob $job): void
     {
         $this->validateGroupName($job->getGroupName());

Classes/Domain/MySQLScheduler.php

@@ -110,4 +110,15 @@ class MySQLScheduler extends AbstractScheduler  {
                        END
             MySQL;
 
-}
+    protected const RESET_STALE_JOBS_QUERY = <<<MySQL
+        UPDATE netlogix_jobqueue_scheduled_job
+        SET running = 0,
+            claimed = '',
+            incarnation = incarnation + 1
+        WHERE running = 1
+          AND claimed NOT LIKE 'failed(%)'
+          AND groupname = :groupName
+          AND activity < NOW() - INTERVAL :minutes MINUTE
+        MySQL;
+
+}

Classes/Domain/PostgreSQLScheduler.php

@@ -75,4 +75,15 @@ class PostgreSQLScheduler extends AbstractScheduler {
             END;
         PostgreSQL;
 
-}
+    protected const RESET_STALE_JOBS_QUERY = <<<PostgreSQL
+        UPDATE netlogix_jobqueue_scheduled_job
+        SET running = 0,
+            claimed = '',
+            incarnation = incarnation + 1
+        WHERE running = 1
+          AND claimed NOT LIKE 'failed(%)'
+          AND groupname = :groupName
+          AND activity < NOW() - make_interval(mins => :minutes)
+        PostgreSQL;
+
+}

Classes/Domain/Scheduler.php

@@ -22,4 +22,6 @@ public function fail(ScheduledJob $job, string $reason): void;
 
     public function activity(ScheduledJob $job): void;
 
-}
+    public function resetStaleJobs(string $groupName, int $minutes): int;
+
+}