ENGCOM-8074: Cron cleanup repeatedly hits deadlocks on large environments where groups can overlap #28007

 - Merge Pull Request #28007 from driskell/magento2:Fix-cron-schedule-deadlock
 - Merged commits:
   1. aa12503
   2. b0b72b0
   3. 315bd3a
   4. 96029b5
   5. 7df183b
   6. 68d679b
   7. a901619
   8. dda5c72
   9. 4d4fa68
   10. e95d2fe
   11. 916bc93
   12. edd7753
   13. e70027f
   14. 92dc764
   15. d1b22ff
   16. a76482c

Author: magento-engcom-team

app/code/Magento/Cron/Model/DeadlockRetrier.php

@@ -17,6 +17,20 @@
  */
 class DeadlockRetrier implements DeadlockRetrierInterface
 {
+    /**
+     * @var \Psr\Log\LoggerInterface
+     */
+    private $logger;
+
+    /**
+     * @param \Psr\Log\LoggerInterface $logger
+     */
+    public function __construct(
+        \Psr\Log\LoggerInterface $logger
+    ) {
+        $this->logger = $logger;
+    }
+
     /**
      * @inheritdoc
      */
@@ -30,6 +44,7 @@ public function execute(callable $callback, AdapterInterface $connection)
             try {
                 return $callback();
             } catch (DeadlockException $e) {
+                $this->logger->warning(sprintf("Deadlock detected in cron: %s", $e->getMessage()));
                 continue;
             }
         }

app/code/Magento/Cron/Model/ResourceModel/Schedule.php

@@ -65,31 +65,47 @@ public function trySetJobStatusAtomic($scheduleId, $newStatus, $currentStatus)
     public function trySetJobUniqueStatusAtomic($scheduleId, $newStatus, $currentStatus)
     {
         $connection = $this->getConnection();
+        $connection->beginTransaction();
 
         // this condition added to avoid cron jobs locking after incorrect termination of running job
         $match = $connection->quoteInto(
             'existing.job_code = current.job_code ' .
-            'AND (existing.executed_at > UTC_TIMESTAMP() - INTERVAL 1 DAY OR existing.executed_at IS NULL) ' .
-            'AND existing.status = ?',
+            'AND existing.status = ? ' .
+            'AND (existing.executed_at > UTC_TIMESTAMP() - INTERVAL 1 DAY OR existing.executed_at IS NULL)',
             $newStatus
         );
 
+        // Select and lock all related schedules - this prevents deadlock in case cron overlaps and two jobs of
+        // the same code attempt to lock at the same time, and force them to serialize
         $selectIfUnlocked = $connection->select()
+            ->from(
+                ['current' => $this->getTable('cron_schedule')],
+                []
+            )
             ->joinLeft(
                 ['existing' => $this->getTable('cron_schedule')],
                 $match,
-                ['status' => new \Zend_Db_Expr($connection->quote($newStatus))]
+                ['existing.schedule_id']
             )
             ->where('current.schedule_id = ?', $scheduleId)
             ->where('current.status = ?', $currentStatus)
-            ->where('existing.schedule_id IS NULL');
-
-        $update = $connection->updateFromSelect($selectIfUnlocked, ['current' => $this->getTable('cron_schedule')]);
-        $result = $connection->query($update)->rowCount();
+            ->forUpdate(true);
 
-        if ($result == 1) {
-            return true;
+        $scheduleId = $connection->fetchOne($selectIfUnlocked);
+        if (!empty($scheduleId)) {
+            // Existing running schedule found
+            $connection->commit();
+            return false;
         }
-        return false;
+
+        // Mark our schedule as running
+        $connection->update(
+            $this->getTable('cron_schedule'),
+            ['status' => new \Zend_Db_Expr($connection->quote($newStatus))],
+            ['schedule_id = ?' => $scheduleId]
+        );
+
+        $connection->commit();
+        return true;
     }
 }

app/code/Magento/Cron/Observer/ProcessCronQueueObserver.php

@@ -9,6 +9,7 @@
  */
 namespace Magento\Cron\Observer;
 
+use Magento\Cron\Model\ResourceModel\Schedule\Collection as ScheduleCollection;
 use Magento\Cron\Model\Schedule;
 use Magento\Framework\App\State;
 use Magento\Framework\Console\Cli;
@@ -83,7 +84,7 @@ class ProcessCronQueueObserver implements ObserverInterface
     const MAX_RETRIES = 5;
 
     /**
-     * @var \Magento\Cron\Model\ResourceModel\Schedule\Collection
+     * @var ScheduleCollection
      */
     protected $_pendingSchedules;
 
@@ -278,12 +279,12 @@ function ($groupId) use ($currentTime) {
      *
      * It should be taken by standalone (child) process, not by the parent process.
      *
-     * @param int $groupId
+     * @param string $groupId
      * @param callable $callback
      *
      * @return void
      */
-    private function lockGroup($groupId, callable $callback)
+    private function lockGroup(string $groupId, callable $callback): void
     {
         if (!$this->lockManager->lock(self::LOCK_PREFIX . $groupId, self::LOCK_TIMEOUT)) {
             $this->logger->warning(
@@ -399,7 +400,7 @@ function () use ($schedule) {
      * @param string $jobName
      * @return void
      */
-    private function startProfiling(string $jobName = '')
+    private function startProfiling(string $jobName = ''): void
     {
         $this->statProfiler->clear();
         $this->statProfiler->start(
@@ -416,7 +417,7 @@ private function startProfiling(string $jobName = '')
      * @param string $jobName
      * @return void
      */
-    private function stopProfiling(string $jobName = '')
+    private function stopProfiling(string $jobName = ''): void
     {
         $this->statProfiler->stop(
             sprintf(self::CRON_TIMERID, $jobName),
@@ -445,9 +446,9 @@ private function getProfilingStat(string $jobName): string
      * Return job collection from data base with status 'pending'.
      *
      * @param string $groupId
-     * @return \Magento\Cron\Model\ResourceModel\Schedule\Collection
+     * @return ScheduleCollection
      */
-    private function getPendingSchedules($groupId)
+    private function getPendingSchedules(string $groupId): ScheduleCollection
     {
         $jobs = $this->_config->getJobs();
         $pendingJobs = $this->_scheduleFactory->create()->getCollection();
@@ -462,7 +463,7 @@ private function getPendingSchedules($groupId)
      * @param string $groupId
      * @return $this
      */
-    private function generateSchedules($groupId)
+    private function generateSchedules(string $groupId): self
     {
         /**
          * check if schedule generation is needed
@@ -533,13 +534,13 @@ protected function _generateJobs($jobs, $exists, $groupId)
      * @param int $currentTime
      * @return void
      */
-    private function cleanupJobs($groupId, $currentTime)
+    private function cleanupJobs(string $groupId, int $currentTime): void
     {
         // check if history cleanup is needed
         $lastCleanup = (int)$this->_cache->load(self::CACHE_KEY_LAST_HISTORY_CLEANUP_AT . $groupId);
         $historyCleanUp = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_CLEANUP_EVERY);
         if ($lastCleanup > $this->dateTime->gmtTimestamp() - $historyCleanUp * self::SECONDS_IN_MINUTE) {
-            return $this;
+            return;
         }
         // save time history cleanup was ran with no expiration
         $this->_cache->save(
@@ -550,6 +551,7 @@ private function cleanupJobs($groupId, $currentTime)
         );
 
         $this->cleanupDisabledJobs($groupId);
+        $this->cleanupRunningJobs($groupId);
 
         $historySuccess = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_SUCCESS);
         $historyFailure = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_FAILURE);
@@ -673,7 +675,7 @@ protected function getScheduleTimeInterval($groupId)
      * @param string $groupId
      * @return void
      */
-    private function cleanupDisabledJobs($groupId)
+    private function cleanupDisabledJobs(string $groupId): void
     {
         $jobs = $this->_config->getJobs();
         $jobsToCleanup = [];
@@ -696,6 +698,33 @@ private function cleanupDisabledJobs($groupId)
         }
     }
 
+    /**
+     * Cleanup jobs that were left in a running state due to an unexpected stop
+     *
+     * @param string $groupId
+     * @return void
+     */
+    private function cleanupRunningJobs(string $groupId): void
+    {
+        $scheduleResource = $this->_scheduleFactory->create()->getResource();
+        $connection = $scheduleResource->getConnection();
+
+        $jobs = $this->_config->getJobs();
+
+        $connection->update(
+            $scheduleResource->getTable('cron_schedule'),
+            [
+                'status' => \Magento\Cron\Model\Schedule::STATUS_ERROR,
+                'messages' => 'Time out'
+            ],
+            [
+                $connection->quoteInto('status = ?', \Magento\Cron\Model\Schedule::STATUS_RUNNING),
+                $connection->quoteInto('job_code IN (?)', array_keys($jobs[$groupId])),
+                'scheduled_at < UTC_TIMESTAMP() - INTERVAL 1 DAY'
+            ]
+        );
+    }
+
     /**
      * Get cron expression of cron job.
      *
@@ -773,13 +802,13 @@ private function isGroupInFilter($groupId): bool
      * @param array $jobsRoot
      * @param int $currentTime
      */
-    private function processPendingJobs($groupId, $jobsRoot, $currentTime)
+    private function processPendingJobs(string $groupId, array $jobsRoot, int $currentTime): void
     {
-        $procesedJobs = [];
+        $processedJobs = [];
         $pendingJobs = $this->getPendingSchedules($groupId);
         /** @var Schedule $schedule */
         foreach ($pendingJobs as $schedule) {
-            if (isset($procesedJobs[$schedule->getJobCode()])) {
+            if (isset($processedJobs[$schedule->getJobCode()])) {
                 // process only on job per run
                 continue;
             }
@@ -796,7 +825,7 @@ private function processPendingJobs($groupId, $jobsRoot, $currentTime)
             $this->tryRunJob($scheduledTime, $currentTime, $jobConfig, $schedule, $groupId);
 
             if ($schedule->getStatus() === Schedule::STATUS_SUCCESS) {
-                $procesedJobs[$schedule->getJobCode()] = true;
+                $processedJobs[$schedule->getJobCode()] = true;
             }
 
             $this->retrier->execute(
@@ -821,7 +850,7 @@ private function tryRunJob($scheduledTime, $currentTime, $jobConfig, $schedule,
     {
         // use sha1 to limit length
         // phpcs:ignore Magento2.Security.InsecureFunction
-        $lockName =  self::LOCK_PREFIX . md5($groupId . '_' . $schedule->getJobCode());
+        $lockName = self::LOCK_PREFIX . md5($groupId . '_' . $schedule->getJobCode());
 
         try {
             for ($retries = self::MAX_RETRIES; $retries > 0; $retries--) {

app/code/Magento/Cron/Test/Unit/Model/DeadlockRetrierTest.php

@@ -13,6 +13,7 @@
 use Magento\Framework\DB\Adapter\AdapterInterface;
 use Magento\Framework\DB\Adapter\DeadlockException;
 use PHPUnit\Framework\MockObject\MockObject;
+use Psr\Log\LoggerInterface;
 
 class DeadlockRetrierTest extends \PHPUnit\Framework\TestCase
 {
@@ -27,6 +28,11 @@ class DeadlockRetrierTest extends \PHPUnit\Framework\TestCase
      */
     private $adapterMock;
 
+    /**
+     * @var LoggerInterface|MockObject
+     */
+    private $loggerMock;
+
     /**
      * @var AbstractModel|MockObject
      */
@@ -38,8 +44,9 @@ class DeadlockRetrierTest extends \PHPUnit\Framework\TestCase
     protected function setUp(): void
     {
         $this->adapterMock = $this->getMockForAbstractClass(AdapterInterface::class);
+        $this->loggerMock = $this->getMockForAbstractClass(LoggerInterface::class);
         $this->modelMock = $this->createMock(AbstractModel::class);
-        $this->retrier = new DeadlockRetrier();
+        $this->retrier = new DeadlockRetrier($this->loggerMock);
     }
 
     /**
@@ -75,6 +82,8 @@ public function testRetry(): void
         $this->modelMock->expects($this->exactly(DeadlockRetrierInterface::MAX_RETRIES))
             ->method('getId')
             ->willThrowException(new DeadlockException());
+        $this->loggerMock->expects($this->exactly(DeadlockRetrierInterface::MAX_RETRIES - 1))
+            ->method('warning');
 
         $this->retrier->execute(
             function () {
@@ -95,6 +104,8 @@ public function testRetrySecond(): void
         $this->modelMock->expects($this->at(1))
             ->method('getId')
             ->willReturn(2);
+        $this->loggerMock->expects($this->once())
+            ->method('warning');
 
         $this->retrier->execute(
             function () {