snapshot of https://github.com/espocrm/espocrm.git (master, 2026-09-18; blobs > 131072B omitted)

application/Espo/Core/Job/QueueProcessor/Params.php

@@ -0,0 +1,116 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM – Open Source CRM application.
+ * Copyright (C) 2014-2026 EspoCRM, Inc.
+ * Website: https://www.espocrm.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://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 Affero General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU Affero General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job\QueueProcessor;
+
+class Params
+{
+    private ?string $queue = null;
+    private ?string $group = null;
+    private int $limit = 0;
+    /** @var ?Params[] */
+    private ?array $subQueueParamsList = null;
+    private float $weight = 1.0;
+
+    public function withQueue(?string $queue): self
+    {
+        $obj = clone $this;
+        $obj->queue = $queue;
+
+        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;
+        $obj->limit = $limit;
+
+        return $obj;
+    }
+
+    public function withWeight(float $weight): self
+    {
+        $obj = clone $this;
+        $obj->weight = $weight;
+
+        return $obj;
+    }
+
+    /**
+     * @param ?Params[] $subQueueParamsList
+     */
+    public function withSubQueueParamsList(?array $subQueueParamsList): self
+    {
+        $obj = clone $this;
+        $obj->subQueueParamsList = $subQueueParamsList;
+
+        return $obj;
+    }
+
+    public function getQueue(): ?string
+    {
+        return $this->queue;
+    }
+
+    public function getGroup(): ?string
+    {
+        return $this->group;
+    }
+
+    public function getLimit(): int
+    {
+        return $this->limit;
+    }
+
+    public function getWeight(): float
+    {
+        return $this->weight;
+    }
+
+    /**
+     * @return ?Params[]
+     */
+    public function getSubQueueParamsList(): ?array
+    {
+        return $this->subQueueParamsList;
+    }
+
+    public static function create(): self
+    {
+        return new self();
+    }
+}

application/Espo/Core/Job/QueueProcessor/Picker.php

@@ -0,0 +1,134 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM – Open Source CRM application.
+ * Copyright (C) 2014-2026 EspoCRM, Inc.
+ * Website: https://www.espocrm.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://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 Affero General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU Affero General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job\QueueProcessor;
+
+use Espo\Core\Job\QueueUtil;
+use Espo\Entities\Job;
+use RuntimeException;
+
+/**
+ * Picks jobs for a portion distributing by weights if needed.
+ */
+class Picker
+{
+    public function __construct(
+        private QueueUtil $queueUtil,
+    ) {}
+
+    /**
+     * @param Params $params
+     * @return iterable<Job>
+     */
+    public function pick(Params $params): iterable
+    {
+        $paramsList = $params->getSubQueueParamsList();
+
+        if (!$paramsList) {
+            return $this->queueUtil->getPendingJobs($params);
+        }
+
+        $groups = [];
+
+        foreach ($paramsList as $itemParams) {
+            $groups[] = iterator_to_array($this->queueUtil->getPendingJobs($itemParams));
+        }
+
+        return $this->pickJobsRecursively($paramsList, $groups, $params->getLimit());
+    }
+
+    /**
+     * @param Params[] $paramsList,
+     * @param Job[][] $groups
+     * @return Job[]
+     */
+    private function pickJobsRecursively(
+        array $paramsList,
+        array $groups,
+        int $limit,
+    ): array {
+
+        $totalWeight = array_reduce($paramsList, fn ($c, $it) => $c + $it->getWeight(), 0.0);
+
+        /** @var Job[][] $leftovers */
+        $leftovers = [];
+        $output = [];
+
+        foreach ($paramsList as $i => $itemParams) {
+            if (!array_key_exists($i, $groups)) {
+                throw new RuntimeException();
+            }
+
+            $jobs = $groups[$i];
+            $weight = $itemParams->getWeight();
+
+            $portion = (int) round($weight / $totalWeight * $limit);
+
+            $pickedJobs = [];
+
+            while (count($pickedJobs) < $portion) {
+                if (count($jobs) === 0) {
+                    break;
+                }
+
+                $pickedJobs[] = array_shift($jobs);
+            }
+
+            $output = array_merge($output, $pickedJobs);
+
+            $leftovers[] = $jobs;
+        }
+
+        $left = $limit - count($output);
+        $leftoverCount = array_reduce($leftovers, fn ($c, $it) => $c + count($it), 0);
+
+        if ($left && $leftoverCount) {
+            foreach ($leftovers as $i => $jobs) {
+                if (count($jobs) === 0) {
+                    unset($leftovers[$i]);
+                    unset($paramsList[$i]);
+                }
+            }
+
+            $leftovers = array_values($leftovers);
+            $paramsList = array_values($paramsList);
+
+            $rest = $this->pickJobsRecursively(
+                paramsList: $paramsList,
+                groups: $leftovers,
+                limit: $leftoverCount,
+            );
+
+            $output = array_merge($output, $rest);
+            $output = array_slice($output, 0, $limit);
+        }
+
+        return $output;
+    }
+}

application/Espo/Core/Job/QueueProcessor/QueueProcessors/DefaultQueueProcessor.php

@@ -0,0 +1,74 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM – Open Source CRM application.
+ * Copyright (C) 2014-2026 EspoCRM, Inc.
+ * Website: https://www.espocrm.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://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 Affero General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU Affero General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job\QueueProcessor\QueueProcessors;
+
+use Espo\Core\Job\AsyncPoolFactory;
+use Espo\Core\Job\ConfigDataProvider;
+use Espo\Core\Job\QueueProcessor;
+use Espo\Core\Utils\Log;
+use Espo\Core\Job\QueueProcessor\Params;
+
+/**
+ * @internal
+ */
+class DefaultQueueProcessor implements QueueProcessor
+{
+    public function __construct(
+        private ProcessPoolQueueProcessor $processPoolQueueProcessor,
+        private SequentialQueueProcessor $sequentialQueueProcessor,
+        private AsyncPoolFactory $asyncPoolFactory,
+        private Log $log,
+        private ConfigDataProvider $configDataProvider,
+    ) {}
+
+    public function process(Params $params): void
+    {
+        if ($this->inParallel()) {
+            $this->processPoolQueueProcessor->process($params);
+        } else {
+            $this->sequentialQueueProcessor->process($params);
+        }
+    }
+
+    private function inParallel(): bool
+    {
+        if (!$this->configDataProvider->runInParallel()) {
+            return false;
+        }
+
+        if (!$this->asyncPoolFactory->isSupported()) {
+            $this->log->warning("Enabled `jobRunInParallel` parameter requires pcntl and posix extensions.");
+
+            return false;
+        }
+
+        return true;
+    }
+}

application/Espo/Core/Job/QueueProcessor/QueueProcessors/ProcessPoolQueueProcessor.php

@@ -0,0 +1,132 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM – Open Source CRM application.
+ * Copyright (C) 2014-2026 EspoCRM, Inc.
+ * Website: https://www.espocrm.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://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 Affero General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU Affero General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job\QueueProcessor\QueueProcessors;
+
+use Espo\Core\Job\AsyncPoolFactory;
+use Espo\Core\Job\ConfigDataProvider;
+use Espo\Core\Job\Job\Status;
+use Espo\Core\Job\JobTask;
+use Espo\Core\Job\QueueProcessor;
+use Espo\Core\Job\QueueProcessor\Params;
+use Espo\Core\Job\QueueProcessor\Picker;
+use Espo\Core\Job\QueueUtil;
+use Espo\Core\ORM\EntityManager;
+use Espo\Entities\Job;
+use Spatie\Async\Pool as AsyncPool;
+
+class ProcessPoolQueueProcessor implements QueueProcessor
+{
+    public function __construct(
+        private QueueUtil $queueUtil,
+        private AsyncPoolFactory $asyncPoolFactory,
+        private EntityManager $entityManager,
+        private Picker $picker,
+        private ConfigDataProvider $configDataProvider,
+    ) {}
+
+    public function process(Params $params): void
+    {
+        $noLock = $this->skipLock($params);
+
+        $pool = $this->asyncPoolFactory->create();
+
+        foreach ($this->picker->pick($params) as $job) {
+            $this->processJob($noLock, $job, $pool);
+        }
+
+        $pool->wait();
+    }
+
+    private function processJob(bool $noLock, Job $job, AsyncPool $pool): void
+    {
+        $lockTable = !$noLock && $job->getScheduledJobId();
+
+        if ($lockTable) {
+            // MySQL doesn't allow to lock non-existent rows. We resort to locking an entire table.
+            $this->entityManager->getLocker()->lockExclusive(Job::ENTITY_TYPE);
+        }
+
+        $skip = $this->toSkip($noLock, $job);
+
+        if ($skip) {
+            if ($lockTable) {
+                $this->entityManager->getLocker()->rollback();
+            }
+
+            return;
+        }
+
+        $this->prepareJob($job);
+
+        if ($lockTable) {
+            $this->entityManager->getLocker()->commit();
+        }
+
+        $this->runJob($job, $pool);
+    }
+
+    private function toSkip(bool $noLock, Job $job): bool
+    {
+        if (!$noLock && !$this->queueUtil->isJobPending($job)) {
+            return true;
+        }
+
+        if ($this->queueUtil->isScheduledJobRunning($job)) {
+            return true;
+        }
+
+        return false;
+    }
+
+    private function prepareJob(Job $job): void
+    {
+        $job
+            // Needed for failing not started.
+            ->setStartedAtNow()
+            ->setStatus(Status::READY);
+
+        $this->entityManager->saveEntity($job);
+    }
+
+    private function runJob(Job $job, AsyncPool $pool): void
+    {
+        $task = new JobTask($job->getId());
+
+        $pool->add($task);
+    }
+
+    private function skipLock(Params $params): bool
+    {
+        return
+            $params->getGroup() !== null ||
+            $params->getQueue() !== null ||
+            $this->configDataProvider->noTableLocking();
+    }
+}

application/Espo/Core/Job/QueueProcessor/QueueProcessors/SequentialQueueProcessor.php

@@ -0,0 +1,83 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM – Open Source CRM application.
+ * Copyright (C) 2014-2026 EspoCRM, Inc.
+ * Website: https://www.espocrm.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://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 Affero General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU Affero General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Job\QueueProcessor\QueueProcessors;
+
+use Espo\Core\Job\Job\Status;
+use Espo\Core\Job\JobRunner;
+use Espo\Core\Job\QueueProcessor;
+use Espo\Core\Job\QueueProcessor\Params;
+use Espo\Core\Job\QueueProcessor\Picker;
+use Espo\Core\Job\QueueUtil;
+use Espo\Core\ORM\EntityManager;
+use Espo\Core\Utils\System;
+use Espo\Entities\Job;
+
+class SequentialQueueProcessor implements QueueProcessor
+{
+    public function __construct(
+        private QueueUtil $queueUtil,
+        private JobRunner $jobRunner,
+        private EntityManager $entityManager,
+        private Picker $picker,
+    ) {}
+
+    public function process(Params $params): void
+    {
+        foreach ($this->picker->pick($params) as $job) {
+            $this->processJob($job);
+        }
+    }
+
+    private function processJob(Job $job): void
+    {
+        if ($this->toSkip($job)) {
+            return;
+        }
+
+        $this->prepareJob($job);
+
+        $this->jobRunner->run($job);
+    }
+
+    private function toSkip(Job $job): bool
+    {
+        return $this->queueUtil->isScheduledJobRunning($job);
+    }
+
+    private function prepareJob(Job $job): void
+    {
+        $job
+            ->setStartedAtNow()
+            ->setStatus(Status::RUNNING)
+            ->setPid(System::getPid());
+
+        $this->entityManager->saveEntity($job);
+    }
+}