feat: Poll every group of an installation in one process

Classes/Domain/AbstractScheduler.php

@@ -15,8 +15,11 @@
 use Neos\Flow\Annotations as Flow;
 use Throwable;
 
-use function array_filter;
-use function in_array;
+use function array_fill_keys;
+use function array_key_exists;
+use function array_keys;
+use function array_values;
+use function implode;
 use function sprintf;
 
 abstract class AbstractScheduler implements Scheduler
@@ -27,23 +30,108 @@ abstract class AbstractScheduler implements Scheduler
     protected $dbal;
 
     /**
-     * @var string[]
+     * @var TimeBaseForDueDateCalculation
      */
-    protected array $activeGroupNames = [self::DEFAULT_GROUP_NAME];
+    protected TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation;
+
+    protected GroupRepository $groupRepository;
 
     /**
-     * @var TimeBaseForDueDateCalculation
+     * Step 1 of claiming a job: tag one due row with the claim value.
+     *
+     * One branch per group, each limited to its own oldest due job, combined by
+     * UNION ALL and narrowed down to a single row afterwards. Every branch is an
+     * equality lookup on "groupname" and therefore keeps using idx_for_update,
+     * which a "groupname IN (…)" would give up: duedate is only ordered within
+     * one group, so sorting across groups would need a filesort.
+     *
+     * Group names are bound as :g0 … :gn, see claimParameters().
      */
-    protected TimeBaseForDueDateCalculation $timeBaseForDueDateCalculation;
+    abstract protected function buildClaimQuery(string ...$groupNames): string;
+
+    /**
+     * Step 2 of claiming a job: read the row that carries the claim value.
+     *
+     * Filters on the claim value alone - which group was hit is unknown until
+     * the row has been read, and a claim is a UUID.
+     */
+    abstract protected function buildSelectQuery(): string;
+
+    /**
+     * Step 3 of claiming a job: unlock the row so parallel processes may
+     * overwrite the claim value again. Unrelated to release().
+     */
+    abstract protected function buildReleaseQuery(): string;
+
+    abstract protected function buildScheduleQuery(): string;
+
+    abstract protected function buildResetStaleJobsQuery(): string;
+
+    protected function buildIsScheduledQuery(): string
+    {
+        $tableName = ScheduledJob::TABLE_NAME;
+
+        return /** @lang MySQL */ <<<"MySQL"
+            SELECT 1 FROM {$tableName}
+            WHERE identifier = :identifier
+              AND groupname = :groupname
+              AND claimed = ''
+            MySQL;
+    }
+
+    protected function buildDeleteJobQuery(): string
+    {
+        $tableName = ScheduledJob::TABLE_NAME;
+
+        return /** @lang MySQL */ <<<"MySQL"
+            DELETE FROM {$tableName}
+            WHERE groupname = :groupname
+              AND identifier = :identifier
+              AND claimed = :claimed
+            MySQL;
+    }
+
+    /**
+     * Fallback of release(): the row was rescheduled while running, so it must
+     * not be deleted but freed for the next run.
+     */
+    protected function buildFreeJobQuery(): string
+    {
+        $tableName = ScheduledJob::TABLE_NAME;
+
+        return /** @lang MySQL */ <<<"MySQL"
+            UPDATE {$tableName}
+            SET running = 0,
+                activity = NOW()
+            WHERE groupname = :groupname
+              AND identifier = :identifier
+              AND claimed = ''
+            MySQL;
+    }
+
+    protected function buildFailQuery(): string
+    {
+        $tableName = ScheduledJob::TABLE_NAME;
+
+        return /** @lang MySQL */ <<<"MySQL"
+            UPDATE {$tableName}
+            SET claimed = :failed,
+                running = 0,
+                activity = NOW()
+            WHERE identifier = :identifier
+            MySQL;
+    }
 
-    #[Flow\InjectConfiguration(path: 'staleJobTimeout')]
-    protected int $staleJobTimeoutSecs;
+    protected function buildActivityQuery(): string
+    {
+        $tableName = ScheduledJob::TABLE_NAME;
 
-    protected const CLAIM_QUERY = "";
-    protected const SELECT_QUERY = "";
-    protected const RELEASE_QUERY = "";
-    protected const SCHEDULE_QUERY = "";
-    protected const RESET_STALE_JOBS_QUERY = "";
+        return /** @lang MySQL */ <<<"MySQL"
+            UPDATE {$tableName}
+            SET activity = NOW()
+            WHERE identifier = :identifier
+            MySQL;
+    }
 
     public function injectConnection(Connection $connection): void
     {
@@ -55,13 +143,9 @@ public function injectTimeBaseForDueDateCalculation(TimeBaseForDueDateCalculatio
         $this->timeBaseForDueDateCalculation = $timeBaseForDueDateCalculation;
     }
 
-    public function injectSettings(array $settings)
+    public function injectGroupRepository(GroupRepository $groupRepository): void
     {
-        $activeGroupNames = array_filter($settings['groupNames'] ?? []);
-        $this->activeGroupNames = array_keys($activeGroupNames);
-        if (!$this->activeGroupNames) {
-            $this->activeGroupNames = [self::DEFAULT_GROUP_NAME];
-        }
+        $this->groupRepository = $groupRepository;
     }
 
     public function schedule(ScheduledJob $job, ScheduledJob ...$jobs): void
@@ -75,14 +159,8 @@ public function schedule(ScheduledJob $job, ScheduledJob ...$jobs): void
     public function isScheduled(string $groupName, string $identifier): bool
     {
         $this->validateGroupName($groupName);
-        $tableName = ScheduledJob::TABLE_NAME;
 
-        $statement = <<<"MySQL"
-            SELECT 1 FROM {$tableName}
-            WHERE identifier = :identifier
-              AND groupname = :groupname
-              AND claimed = ''
-            MySQL;
+        $statement = $this->buildIsScheduledQuery();
 
         return $this->dbal->fetchOne($statement, ['identifier' => $identifier, 'groupname' => $groupName]) !== false;
     }
@@ -92,43 +170,50 @@ public function ping(): void
         $this->dbal->ping();
     }
 
-    public function next(string $groupName): ?ScheduledJob
+    public function next(string $groupName, string ...$furtherGroupNames): ?ScheduledJob
     {
-        $this->validateGroupName($groupName);
+        // array_values(): named arguments end up in the variadic as string keys,
+        // which would break the positional :g0 … :gn binding.
+        $groupNames = array_values([$groupName, ...$furtherGroupNames]);
+        foreach ($groupNames as $name) {
+            $this->validateGroupName($name);
+        }
         $claim = Algorithms::generateUUID();
 
-        $claimQuery = static::CLAIM_QUERY;
-        $this->dbal
+        $groupParameters = self::claimParameters($groupNames);
+
+        $claimedRows = $this->dbal
             ->executeQuery(
-                sql: $claimQuery,
+                sql: $this->buildClaimQuery(...$groupNames),
                 params: [
                     'now' => $this->timeBaseForDueDateCalculation->getNow(),
-                    'groupname' => $groupName,
                     'claimed' => $claim,
+                    ...$groupParameters,
                 ],
                 types: [
                     'now' => Types::DATETIME_IMMUTABLE,
-                    'groupname' => Types::STRING,
                     'claimed' => Types::STRING,
+                    ...array_fill_keys(array_keys($groupParameters), Types::STRING),
                 ],
                 logContext: fn (Throwable $throwable, int $incarnation) => [
                     'claim' => $claim,
-                    'groupName' => $groupName,
+                    'groupName' => implode(', ', $groupNames),
                     'step' => 'claim',
                 ]
-            );
+            )
+            ->rowCount();
 
-        $select = static::SELECT_QUERY;
+        if ($claimedRows === 0) {
+            return null;
+        }
 
         $row = $this->dbal
             ->executeQuery(
-                sql: $select,
+                sql: $this->buildSelectQuery(),
                 params: [
-                    'groupname' => $groupName,
                     'claimed' => $claim,
                 ],
                 types: [
-                    'groupname' => Types::STRING,
                     'claimed' => Types::STRING,
                 ]
             )
@@ -138,17 +223,16 @@ public function next(string $groupName): ?ScheduledJob
             return null;
         }
 
-        $release = static::RELEASE_QUERY;
+        // groupname is CHAR(36): PostgreSQL returns it space-padded
+        $groupName = rtrim((string) $row['groupname']);
 
         $this->dbal
             ->executeQuery(
-                sql: $release,
+                sql: $this->buildReleaseQuery(),
                 params: [
-                    'groupname' => $groupName,
                     'claimed' => $claim,
                 ],
                 types: [
-                    'groupname' => Types::STRING,
                     'claimed' => Types::STRING,
                 ],
                 logContext: fn (Throwable $throwable, int $incarnation) => [
@@ -170,20 +254,28 @@ public function next(string $groupName): ?ScheduledJob
         );
     }
 
+    /**
+     * Binds group names as :g0 … :gn, matching the branches of the claim query.
+     *
+     * @param list<string> $groupNames
+     * @return array<string, string>
+     */
+    protected static function claimParameters(array $groupNames): array
+    {
+        $parameters = [];
+        foreach ($groupNames as $index => $groupName) {
+            $parameters['g' . $index] = $groupName;
+        }
+
+        return $parameters;
+    }
+
     public function release(ScheduledJob $job): void
     {
         if ($job->getClaimed() === '') {
             throw new InvalidArgumentException('Cannot release unclaimed jobs', 1657027508);
         }
-        $tableName = ScheduledJob::TABLE_NAME;
-
-        $delete = /** @lang MySQL */
-            <<<"MySQL"
-                DELETE FROM {$tableName}
-                WHERE groupname = :groupname
-                  AND identifier = :identifier
-                  AND claimed = :claimed
-                MySQL;
+        $delete = $this->buildDeleteJobQuery();
         $deleteResult = $this->dbal
             ->executeQuery(
                 sql: $delete,
@@ -194,15 +286,7 @@ public function release(ScheduledJob $job): void
                 ]
             );
         if ($deleteResult->rowCount() === 0) {
-            $free = /** @lang MySQL */
-                <<<MySQL
-                    UPDATE {$tableName}
-                    SET running = 0,
-                        activity = NOW()
-                    WHERE groupname = :groupname
-                      AND identifier = :identifier
-                      AND claimed = ''
-                    MySQL;
+            $free = $this->buildFreeJobQuery();
             $this->dbal
                 ->executeQuery(
                     sql: $free,
@@ -222,16 +306,7 @@ public function fail(ScheduledJob $job, string $reason): void
 
         $reason = substr($reason, 0, 36);
 
-        $tableName = ScheduledJob::TABLE_NAME;
-
-        $update = /** @lang MySQL */
-            <<<MySQL
-                UPDATE {$tableName}
-                SET claimed = :failed,
-                    running = 0,
-                    activity = NOW()
-                WHERE identifier = :identifier
-                MySQL;
+        $update = $this->buildFailQuery();
         $this->dbal
             ->executeQuery(
                 sql: $update,
@@ -262,14 +337,7 @@ public function emitFailed(ScheduledJob $job, string $reason): void
 
     public function activity(ScheduledJob $job): void
     {
-        $tableName = ScheduledJob::TABLE_NAME;
-
-        $update = /** @lang MySQL */
-            <<<"MySQL"
-                UPDATE {$tableName}
-                SET activity = NOW()
-                WHERE identifier = :identifier
-                MySQL;
+        $update = $this->buildActivityQuery();
         $this->dbal
             ->executeQuery(
                 sql: $update,
@@ -283,23 +351,22 @@ public function activity(ScheduledJob $job): void
     }
 
     /**
-     * Reset stale jobs that have not changed for too long.
+     * Reset stale jobs of one group that have not changed for too long.
+     *
+     * One statement per group: the threshold is the group's own staleJobTimeout,
+     * and a single UPDATE cannot apply a different one per row.
      *
-     * @param string $groupName Free jobs in this group only
-     * @param int|null $minutes Only free jobs that are stale for at least this many minutes. @deprecated Use staleJobTimeout configuration setting instead.
      * @return int Number of freed jobs
      * @throws Exception
      */
-    public function resetStaleJobs(
-        string $groupName,
-        ?int $minutes = null
-    ): int {
+    public function resetStaleJobs(string $groupName): int
+    {
         return $this->dbal
             ->executeQuery(
-                sql: static::RESET_STALE_JOBS_QUERY,
+                sql: $this->buildResetStaleJobsQuery(),
                 params: [
                     'groupName' => $groupName,
-                    'seconds' => max($minutes === null ? $this->staleJobTimeoutSecs : $minutes * 60, 1),
+                    'seconds' => $this->groupRepository->get($groupName)->getStaleJobTimeout(),
                 ],
                 types: [
                     'groupName' => Types::STRING,
@@ -311,7 +378,7 @@ public function resetStaleJobs(
     protected function scheduleJob(ScheduledJob $job): void
     {
         $this->validateGroupName($job->getGroupName());
-        $statement = static::SCHEDULE_QUERY;
+        $statement = $this->buildScheduleQuery();
 
         $this->dbal
             ->executeQuery(
@@ -347,7 +414,7 @@ protected function scheduleJob(ScheduledJob $job): void
 
     protected function validateGroupName(string $groupName): void
     {
-        if (!in_array($groupName, $this->activeGroupNames, true)) {
+        if (!array_key_exists($groupName, $this->groupRepository->active())) {
             throw new InvalidArgumentException(\sprintf('Group name "%s" is not active', $groupName), 1721393320);
         }
     }
@@ -357,8 +424,4 @@ public function getConnection(): Connection
         return $this->dbal;
     }
 
-    public function getStaleJobTimeoutSeconds(): int
-    {
-        return $this->staleJobTimeoutSecs;
-    }
 }