Merge pull request magento#5639 from magento-thunder/MC-25132

Fixed issues:
- MC-25132: Fix deadlocks related to crons / consumers

Author: Yevhen Miroshnychenko

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

@@ -0,0 +1,39 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\Cron\Model;
+
+use Magento\Framework\DB\Adapter\AdapterInterface;
+use Magento\Framework\DB\Adapter\DeadlockException;
+
+/**
+ * Retrier for DB actions
+ *
+ * If some action throw an exceptions, try
+ */
+class DeadlockRetrier implements DeadlockRetrierInterface
+{
+    /**
+     * @inheritdoc
+     */
+    public function execute(callable $callback, AdapterInterface $connection)
+    {
+        if ($connection->getTransactionLevel() !== 0) {
+            return $callback();
+        }
+
+        for ($retries = self::MAX_RETRIES - 1; $retries > 0; $retries--) {
+            try {
+                return $callback();
+            } catch (DeadlockException $e) {
+                continue;
+            }
+        }
+
+        return $callback();
+    }
+}

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

@@ -0,0 +1,33 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\Cron\Model;
+
+use Magento\Framework\DB\Adapter\AdapterInterface;
+
+/**
+ * Retrier Interface
+ */
+interface DeadlockRetrierInterface
+{
+    /**
+     * Maximum numbers of attempts
+     */
+    public const MAX_RETRIES = 5;
+
+    /**
+     * Runs callback function
+     *
+     * If $callback throws an exception DeadlockException,
+     * this callback will be run maximum self::MAX_RETRIES times or less.
+     *
+     * @param callable $callback
+     * @param AdapterInterface $connection
+     * @return mixed
+     */
+    public function execute(callable $callback, AdapterInterface $connection);
+}

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

@@ -49,9 +49,9 @@ public function trySetJobStatusAtomic($scheduleId, $newStatus, $currentStatus)
     }
 
     /**
-     * Sets schedule status only if no existing schedules with the same job code
-     * have that status.  This is used to implement locking for cron jobs.
+     * Sets schedule status only if no existing schedules with the same job code have that status.
      *
+     * This is used to implement locking for cron jobs.
      * If the schedule is currently in $currentStatus and there are no existing
      * schedules with the same job code and $newStatus, set the schedule to
      * $newStatus and return true. Otherwise, return false.

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

@@ -56,6 +56,11 @@ class Schedule extends \Magento\Framework\Model\AbstractModel
      */
     private $dateTimeFactory;
 
+    /**
+     * @var DeadlockRetrierInterface
+     */
+    private $retrier;
+
     /**
      * @param \Magento\Framework\Model\Context $context
      * @param \Magento\Framework\Registry $registry
@@ -64,6 +69,7 @@ class Schedule extends \Magento\Framework\Model\AbstractModel
      * @param array $data
      * @param TimezoneInterface|null $timezoneConverter
      * @param DateTimeFactory|null $dateTimeFactory
+     * @param DeadlockRetrierInterface $retrier
      */
     public function __construct(
         \Magento\Framework\Model\Context $context,
@@ -72,11 +78,13 @@ public function __construct(
         \Magento\Framework\Data\Collection\AbstractDb $resourceCollection = null,
         array $data = [],
         TimezoneInterface $timezoneConverter = null,
-        DateTimeFactory $dateTimeFactory = null
+        DateTimeFactory $dateTimeFactory = null,
+        DeadlockRetrierInterface $retrier = null
     ) {
         parent::__construct($context, $registry, $resource, $resourceCollection, $data);
         $this->timezoneConverter = $timezoneConverter ?: ObjectManager::getInstance()->get(TimezoneInterface::class);
         $this->dateTimeFactory = $dateTimeFactory ?: ObjectManager::getInstance()->get(DateTimeFactory::class);
+        $this->retrier = $retrier ?: ObjectManager::getInstance()->get(DeadlockRetrierInterface::class);
     }
 
     /**
@@ -251,21 +259,42 @@ public function getNumeric($value)
     }
 
     /**
-     * Lock the cron job so no other scheduled instances run simultaneously.
+     * Sets a job to STATUS_RUNNING only if it is currently in STATUS_PENDING.
      *
-     * Sets a job to STATUS_RUNNING only if it is currently in STATUS_PENDING
-     * and no other jobs of the same code are currently in STATUS_RUNNING.
      * Returns true if status was changed and false otherwise.
      *
      * @return boolean
      */
     public function tryLockJob()
     {
-        if ($this->_getResource()->trySetJobUniqueStatusAtomic(
-            $this->getId(),
-            self::STATUS_RUNNING,
-            self::STATUS_PENDING
-        )) {
+        /** @var \Magento\Cron\Model\ResourceModel\Schedule $scheduleResource */
+        $scheduleResource = $this->_getResource();
+
+        // Change statuses from running to error for terminated jobs
+        $this->retrier->execute(
+            function () use ($scheduleResource) {
+                return $scheduleResource->getConnection()->update(
+                    $scheduleResource->getTable('cron_schedule'),
+                    ['status' => self::STATUS_ERROR],
+                    ['job_code = ?' => $this->getJobCode(), 'status = ?' => self::STATUS_RUNNING]
+                );
+            },
+            $scheduleResource->getConnection()
+        );
+
+        // Change status from pending to running for ran jobs
+        $result = $this->retrier->execute(
+            function () use ($scheduleResource) {
+                return $scheduleResource->trySetJobStatusAtomic(
+                    $this->getId(),
+                    self::STATUS_RUNNING,
+                    self::STATUS_PENDING
+                );
+            },
+            $scheduleResource->getConnection()
+        );
+
+        if ($result) {
             $this->setStatus(self::STATUS_RUNNING);
             return true;
         }