job groups

application/Espo/Core/Job/AbstractQueueJob.php

@@ -29,11 +29,16 @@
 
 namespace Espo\Core\Job;
 
+use Espo\Core\Utils\DateTime;
+
 use Espo\ORM\EntityManager;
 
+use Espo\Entities\Job as JobEntity;
+
+use DateTimeImmutable;
 use RuntimeException;
 
-abstract class AbstractQueueJob implements Job
+abstract class AbstractQueueJob implements JobPreperable
 {
     protected $queue = null;
 
@@ -43,6 +48,8 @@ abstract class AbstractQueueJob implements Job
 
     private $entityManager;
 
+    private const SHIFT_PERIOD = '5 seconds';
+
     public function __construct(
         JobManager $jobManager,
         QueuePortionNumberProvider $portionNumberProvider,
@@ -61,6 +68,74 @@ abstract class AbstractQueueJob implements Job
 
         $limit = $this->portionNumberProvider->get($this->queue);
 
-        $this->jobManager->processQueue($this->queue, $limit);
+        $group = $data->get('group');
+
+        $this->jobManager->processQueue($this->queue, $group, $limit);
+    }
+
+    public function prepare(ScheduledJobData $data, DateTimeImmutable $executeTime): void
+    {
+        $groupList = [];
+
+        $shiftPeriod = self::SHIFT_PERIOD;
+
+        $query = $this->entityManager
+            ->getQueryBuilder()
+            ->select('group')
+            ->from(JobEntity::ENTITY_TYPE)
+            ->where([
+                'status' => JobStatus::PENDING,
+                'queue' => $this->queue,
+                'executeTime<=' => $executeTime
+                    ->modify($shiftPeriod)
+                    ->format(DateTime::SYSTEM_DATE_TIME_FORMAT),
+            ])
+            ->groupBy('group')
+            ->build();
+
+        $sth = $this->entityManager->getQueryExecutor()->execute($query);
+
+        while ($row = $sth->fetch()) {
+            $groupList[] = $row['group'] ?? null;
+        }
+
+        if (!count($groupList)) {
+            return;
+        }
+
+        foreach ($groupList as $group) {
+            $existingJob = $this->entityManager
+                ->getRDBRepository(JobEntity::ENTITY_TYPE)
+                ->select('id')
+                ->where([
+                    'scheduledJobId' => $data->getId(),
+                    'targetGroup' => $group,
+                    'status' => [
+                        JobStatus::RUNNING,
+                        JobStatus::READY,
+                    ]
+                ])
+                ->findOne();
+
+            if ($existingJob) {
+                continue;
+            }
+
+            $name = $data->getName();
+
+            if ($group) {
+                $name .= ' :: ' . $group;
+            }
+
+            $this->entityManager->createEntity(JobEntity::ENTITY_TYPE, [
+                'scheduledJobId' => $data->getId(),
+                'executeTime' => $executeTime->format(DateTime::SYSTEM_DATE_TIME_FORMAT),
+                'name' => $data->getName(),
+                'data' => [
+                    'group' => $group,
+                ],
+                'targetGroup' => $group,
+            ]);
+        }
     }
 }

application/Espo/Core/Job/JobManager.php

@@ -164,11 +164,12 @@ class JobManager
     /**
      * Process pending jobs from a specific queue. Jobs within a queue are processed one by one.
      */
-    public function processQueue(string $queue, int $limit): void
+    public function processQueue(string $queue, ?string $group, int $limit): void
     {
         $params = QueueProcessorParams
             ::create()
             ->withQueue($queue)
+            ->withGroup($group)
             ->withLimit($limit)
             ->withUseProcessPool(false)
             ->withNoLock(true);

application/Espo/Core/Job/MetadataProvider.php

@@ -0,0 +1,62 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
+ * Website: https://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job;
+
+use Espo\Core\Utils\Metadata;
+
+class MetadataProvider
+{
+    private $metadata;
+
+    public function __construct(Metadata $metadata)
+    {
+        $this->metadata = $metadata;
+    }
+
+    /**
+     * @return string[]
+     */
+    public function getPreparableJobNameList(): array
+    {
+        $list = [];
+
+        $items = $this->metadata->get(['app', 'scheduledJobs']) ?? [];
+
+        foreach ($items as $name => $item) {
+            $isPreparable = $item['isPreparable'] ?? false;
+
+            if ($isPreparable) {
+                $list[] = $name;
+            }
+        }
+
+        return $list;
+    }
+}

application/Espo/Core/Job/QueueProcessor.php

@@ -77,6 +77,7 @@ class QueueProcessor
 
         $pendingJobList = $this->queueUtil->getPendingJobList(
             $this->params->getQueue(),
+            $this->params->getGroup(),
             $this->params->getLimit()
         );
 

application/Espo/Core/Job/QueueProcessorParams.php

@@ -37,6 +37,8 @@ class QueueProcessorParams
 
     private $queue = null;
 
+    private $group = null;
+
     private $limit = 0;
 
     public function withUseProcessPool(bool $useProcessPool): self
@@ -66,6 +68,15 @@ class QueueProcessorParams
         return $obj;
     }
 
+    public function withGroup(?string $group): self
+    {
+        $obj = clone $this;
+
+        $obj->group = $group;
+
+        return $obj;
+    }
+
     public function withLimit(int $limit): self
     {
         $obj = clone $this;
@@ -90,6 +101,11 @@ class QueueProcessorParams
         return $this->queue;
     }
 
+    public function getGroup(): ?string
+    {
+        return $this->group;
+    }
+
     public function getLimit(): int
     {
         return $this->limit;

application/Espo/Core/Job/QueueUtil.php

@@ -50,15 +50,22 @@ class QueueUtil
 
     private $scheduleUtil;
 
+    private $metadataProvider;
+
     private const NOT_EXISTING_PROCESS_PERIOD = 300;
 
     private const READY_NOT_STARTED_PERIOD = 60;
 
-    public function __construct(Config $config, EntityManager $entityManager, ScheduleUtil $scheduleUtil)
-    {
+    public function __construct(
+        Config $config,
+        EntityManager $entityManager,
+        ScheduleUtil $scheduleUtil,
+        MetadataProvider $metadataProvider
+    ) {
         $this->config = $config;
         $this->entityManager = $entityManager;
         $this->scheduleUtil = $scheduleUtil;
+        $this->metadataProvider = $metadataProvider;
     }
 
     public function isJobPending(string $id): bool
@@ -79,7 +86,10 @@ class QueueUtil
         return $job->get('status') === JobStatus::PENDING;
     }
 
-    public function getPendingJobList(?string $queue = null, int $limit = 0): Collection
+    /**
+     * @return JobEntity[]
+     */
+    public function getPendingJobList(?string $queue = null, ?string $group = null, int $limit = 0): Collection
     {
         $builder = $this->entityManager
             ->getRDBRepository(JobEntity::ENTITY_TYPE)
@@ -92,6 +102,7 @@ class QueueUtil
                 'targetType',
                 'methodName',
                 'serviceName',
+                'className',
                 'job',
                 'data',
             ])
@@ -99,6 +110,7 @@ class QueueUtil
                 'status' => JobStatus::PENDING,
                 'executeTime<=' => DateTimeUtil::getSystemNowString(),
                 'queue' => $queue,
+                'group' => $group,
             ])
             ->order('number');
 
@@ -137,15 +149,16 @@ class QueueUtil
         $list = [];
 
         $jobList = $this->entityManager
-            ->getRepository(JobEntity::ENTITY_TYPE)
+            ->getRDBRepository(JobEntity::ENTITY_TYPE)
             ->select(['scheduledJobId'])
+            ->leftJoin('scheduledJob')
             ->where([
                 'status' => [
                     JobStatus::RUNNING,
                     JobStatus::READY,
                 ],
                 'scheduledJobId!=' => null,
-                'targetId=' => null,
+                'scheduledJob.job!=' => $this->metadataProvider->getPreparableJobNameList(),
             ])
             ->order('executeTime')
             ->find();
@@ -363,10 +376,12 @@ class QueueUtil
         $duplicateJobList = $this->entityManager
             ->getRepository(JobEntity::ENTITY_TYPE)
             ->select(['scheduledJobId'])
+            ->leftJoin('scheduledJob')
             ->where([
                 'scheduledJobId!=' => null,
                 'status' => JobStatus::PENDING,
                 'executeTime<=' => DateTimeUtil::getSystemNowString(),
+                'scheduledJob.job!=' => $this->metadataProvider->getPreparableJobNameList(),
                 'targetId' => null,
             ])
             ->groupBy(['scheduledJobId'])

application/Espo/Entities/Job.php

@@ -77,6 +77,22 @@ class Job extends Entity
         return $this->get('targetId');
     }
 
+    /**
+     * Get a group.
+     */
+    public function getGroup(): ?string
+    {
+        return $this->get('group');
+    }
+
+    /**
+     * Get a queue.
+     */
+    public function getQueue(): ?string
+    {
+        return $this->get('queue');
+    }
+
     /**
      * Get data.
      */

application/Espo/Resources/metadata/app/metadata.json

@@ -21,6 +21,7 @@
         ["app", "massActions", "__ANY__", "implementationClassName"],
         ["app", "actions", "__ANY__", "implementationClassName"],
         ["app", "fieldProcessing"],
+        ["app", "scheduledJobs"],
         ["selectDefs"],
         ["recordDefs"],
         ["aclDefs"],

application/Espo/Resources/metadata/app/scheduledJobs.json

@@ -0,0 +1,17 @@
+{
+    "ProcessJobQueueQ0": {
+        "isPreparable": true
+    },
+    "ProcessJobQueueQ1": {
+        "isPreparable": true
+    },
+    "ProcessJobQueueE0": {
+        "isPreparable": true
+    },
+    "CheckEmailAccounts": {
+        "isPreparable": true
+    },
+    "CheckInboundEmails": {
+        "isPreparable": true
+    }
+}

application/Espo/Resources/metadata/entityDefs/Job.json

@@ -14,7 +14,8 @@
                 "Failed": "danger",
                 "Running": "warning",
                 "Ready": "warning"
-            }
+            },
+            "maxLength": 16
         },
         "executeTime": {
             "type": "datetime",
@@ -64,6 +65,16 @@
             "maxLength": 36,
             "default": null
         },
+        "group": {
+            "type": "varchar",
+            "maxLength": 64,
+            "default": null
+        },
+        "targetGroup": {
+            "type": "varchar",
+            "maxLength": 64,
+            "default": null
+        },
         "startedAt": {
             "type": "datetime",
             "hasSeconds": true
@@ -117,6 +128,12 @@
         },
         "status": {
             "columns": ["status", "deleted"]
+        },
+        "statusScheduledJobId": {
+            "columns": ["status", "scheduledJobId"]
+        },
+        "queueGroupExecuteTime": {
+            "columns": ["queue", "status", "group", "executeTime"]
         }
     }
 }

tests/integration/Espo/Core/Job/JobTest.php

@@ -58,20 +58,86 @@ class JobTest extends \tests\integration\Core\BaseTestCase
         $this->entityManager = $this->getContainer()->get('entityManager');
     }
 
-    public function testProcessQueue(): void
+    public function testProcessQueueNoGroup(): void
     {
         $job = $this->entityManager->createEntity('Job', [
             'job' => 'Dummy',
             'queue' => 'q0',
         ]);
 
-        $this->jobManager->processQueue('q0', 10);
+        $this->jobManager->processQueue('q0', null, 10);
 
         $jobReloaded = $this->entityManager->getEntity('Job', $job->id);
 
         $this->assertEquals(JobStatus::SUCCESS, $jobReloaded->getStatus());
     }
 
+    public function testProcessQueueGroupAll(): void
+    {
+        $job1 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+        ]);
+
+        $job2 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+            'group' => 'group-1',
+        ]);
+
+        $job3 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+            'group' => 'group-1',
+        ]);
+
+        $this->jobManager->process();
+
+        $job1Reloaded = $this->entityManager->getEntity('Job', $job1->getId());
+        $job2Reloaded = $this->entityManager->getEntity('Job', $job2->getId());
+        $job3Reloaded = $this->entityManager->getEntity('Job', $job3->getId());
+
+        $this->assertEquals(JobStatus::SUCCESS, $job1Reloaded->getStatus());
+        $this->assertEquals(JobStatus::SUCCESS, $job2Reloaded->getStatus());
+        $this->assertEquals(JobStatus::SUCCESS, $job3Reloaded->getStatus());
+    }
+
+    public function testProcessQueueGroupSeparate(): void
+    {
+        $job1 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+        ]);
+
+        $job2 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+            'group' => 'group-1',
+        ]);
+
+        $job3 = $this->entityManager->createEntity('Job', [
+            'job' => 'Dummy',
+            'queue' => 'q0',
+            'group' => 'group-1',
+        ]);
+
+        $this->jobManager->processQueue('q0', 'group-1', 100);
+
+        $job1Reloaded = $this->entityManager->getEntity('Job', $job1->getId());
+        $job2Reloaded = $this->entityManager->getEntity('Job', $job2->getId());
+        $job3Reloaded = $this->entityManager->getEntity('Job', $job3->getId());
+
+        $this->assertEquals(JobStatus::PENDING, $job1Reloaded->getStatus());
+        $this->assertEquals(JobStatus::SUCCESS, $job2Reloaded->getStatus());
+        $this->assertEquals(JobStatus::SUCCESS, $job3Reloaded->getStatus());
+
+        $this->jobManager->processQueue('q0', null, 100);
+
+        $job1Reloaded2 = $this->entityManager->getEntity('Job', $job1->getId());
+
+        $this->assertEquals(JobStatus::SUCCESS, $job1Reloaded2->getStatus());
+    }
+
     public function testRunJobById(): void
     {
         $job = $this->entityManager->createEntity('Job', [

tests/unit/Espo/Core/Job/MetadataProviderTest.php

@@ -0,0 +1,70 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2021 Yurii Kuznietsov, Taras Machyshyn, Oleksii Avramenko
+ * Website: https://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace tests\unit\Espo\Core\Job;
+
+use Espo\Core\{
+    Job\MetadataProvider,
+    Utils\Metadata,
+};
+
+class MetadataProviderTest extends \PHPUnit\Framework\TestCase
+{
+    private $metadata;
+
+    protected function setUp(): void
+    {
+        $this->metadata = $this->createMock(Metadata::class);
+    }
+
+    public function testGetPreparableJobNameList()
+    {
+        $this->metadata
+            ->method('get')
+            ->with(['app', 'scheduledJobs'])
+            ->willReturn([
+                'Test1' => [
+                    'isPreparable' => true,
+                ],
+                'Test2' => [
+                    'isPreparable' => true,
+                ],
+                'Test3' => [],
+                'Test4' => [
+                    'isPreparable' => false,
+                ],
+            ]);
+
+        $provider = new MetadataProvider($this->metadata);
+
+        $list = $provider->getPreparableJobNameList();
+
+        $this->assertEquals(['Test1', 'Test2'], $list);
+    }
+}

tests/unit/Espo/Core/Job/QueueProcessorParamsTest.php

@@ -60,11 +60,14 @@ class QueueProcessorParamsTest extends \PHPUnit\Framework\TestCase
             ->withLimit(10)
             ->withUseProcessPool(true)
             ->withNoLock(true)
+            ->withGroup('group-0')
             ->withQueue('q0');
 
         $this->assertTrue($params->useProcessPool());
         $this->assertTrue($params->noLock());
 
         $this->assertEquals('q0', $params->getQueue());
+
+        $this->assertEquals('group-0', $params->getGroup());
     }
 }