Cron cleanup repeatedly hits deadlocks on large environments

Fix failing tests

Author: ihor-sviziev

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(
@@ -674,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 = [];
@@ -703,7 +704,7 @@ private function cleanupDisabledJobs($groupId)
      * @param string $groupId
      * @return void
      */
-    private function cleanupRunningJobs($groupId)
+    private function cleanupRunningJobs(string $groupId): void
     {
         $scheduleResource = $this->_scheduleFactory->create()->getResource();
         $connection = $scheduleResource->getConnection();
@@ -716,13 +717,11 @@ private function cleanupRunningJobs($groupId)
                 'status' => \Magento\Cron\Model\Schedule::STATUS_ERROR,
                 'messages' => 'Time out'
             ],
-            $connection->quoteInto(
-                'status = ? ' .
-                'AND job_code IN (?) ' .
-                'AND (scheduled_at < UTC_TIMESTAMP() - INTERVAL 1 DAY)',
-                \Magento\Cron\Model\Schedule::STATUS_RUNNING,
-                array_keys($jobs[$groupId])
-            )
+            [
+                $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'
+            ]
         );
     }
 
@@ -803,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;
             }
@@ -826,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(
@@ -851,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/Observer/ProcessCronQueueObserverTest.php

@@ -1045,53 +1045,85 @@ public function testMissedJobsCleanedInTime()
         $scheduleMock->expects($this->exactly(10))->method('getResource')->willReturn($this->scheduleResourceMock);
         $this->scheduleFactoryMock->expects($this->exactly(11))->method('create')->willReturn($scheduleMock);
 
+        $connectionMock = $this->prepareConnectionMock($tableName);
+
+        $this->scheduleResourceMock->expects($this->exactly(6))
+            ->method('getTable')
+            ->with($tableName)
+            ->willReturn($tableName);
+        $this->scheduleResourceMock->expects($this->exactly(15))
+            ->method('getConnection')
+            ->willReturn($connectionMock);
+
+        $this->retrierMock->expects($this->exactly(5))
+            ->method('execute')
+            ->willReturnCallback(
+                function ($callback) {
+                    return $callback();
+                }
+            );
+
+        $this->cronQueueObserver->execute($this->observerMock);
+    }
+
+    /**
+     * @param string $tableName
+     * @return AdapterInterface|MockObject
+     */
+    private function prepareConnectionMock(string $tableName)
+    {
         $connectionMock = $this->getMockForAbstractClass(AdapterInterface::class);
 
         $connectionMock->expects($this->exactly(5))
             ->method('delete')
             ->withConsecutive(
-                [$tableName, ['status = ?' => 'pending', 'job_code in (?)' => ['test_job1']]],
-                [$tableName, ['status = ?' => 'success', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]],
-                [$tableName, ['status = ?' => 'missed', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]],
-                [$tableName, ['status = ?' => 'error', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]],
-                [$tableName, ['status = ?' => 'pending', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]]
+                [
+                    $tableName,
+                    ['status = ?' => 'pending', 'job_code in (?)' => ['test_job1']]
+                ],
+                [
+                    $tableName,
+                    ['status = ?' => 'success', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]
+                ],
+                [
+                    $tableName,
+                    ['status = ?' => 'missed', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]
+                ],
+                [
+                    $tableName,
+                    ['status = ?' => 'error', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]
+                ],
+                [
+                    $tableName,
+                    ['status = ?' => 'pending', 'job_code in (?)' => ['test_job1'], 'scheduled_at < ?' => null]
+                ]
             )
             ->willReturn(1);
 
-        $connectionMock->expects($this->once())
+        $connectionMock->expects($this->any())
             ->method('quoteInto')
-            ->with(
-                'status = ? AND job_code IN (?) AND (scheduled_at < UTC_TIMESTAMP() - INTERVAL 1 DAY)',
-                'running',
-                ['test_job1']
+            ->withConsecutive(
+                ['status = ?', \Magento\Cron\Model\Schedule::STATUS_RUNNING],
+                ['job_code IN (?)', ['test_job1']],
             )
-            ->willReturn('');
+            ->willReturnOnConsecutiveCalls(
+                "status = 'running'",
+                "job_code IN ('test_job1')"
+            );
 
         $connectionMock->expects($this->once())
             ->method('update')
             ->with(
                 $tableName,
                 ['status' => 'error', 'messages' => 'Time out'],
-                ''
+                [
+                    "status = 'running'",
+                    "job_code IN ('test_job1')",
+                    'scheduled_at < UTC_TIMESTAMP() - INTERVAL 1 DAY'
+                ]
             )
             ->willReturn(0);
 
-        $this->scheduleResourceMock->expects($this->exactly(6))
-            ->method('getTable')
-            ->with($tableName)
-            ->willReturn($tableName);
-        $this->scheduleResourceMock->expects($this->exactly(15))
-            ->method('getConnection')
-            ->willReturn($connectionMock);
-
-        $this->retrierMock->expects($this->exactly(5))
-            ->method('execute')
-            ->willReturnCallback(
-                function ($callback) {
-                    return $callback();
-                }
-            );
-
-        $this->cronQueueObserver->execute($this->observerMock);
+        return $connectionMock;
     }
 }

app/code/Magento/Cron/etc/db_schema_whitelist.json

@@ -12,10 +12,11 @@
         },
         "index": {
             "CRON_SCHEDULE_JOB_CODE": true,
-            "CRON_SCHEDULE_SCHEDULED_AT_STATUS": true
+            "CRON_SCHEDULE_SCHEDULED_AT_STATUS": true,
+            "CRON_SCHEDULE_JOB_CODE_STATUS_SCHEDULED_AT": true
         },
         "constraint": {
             "PRIMARY": true
         }
     }
-}
+}