Lock only running cron record

A single transaction is used to lock, run and update the cron record.
Only the cron run is locked so other cron processes could run other cron
tasks.
When fixing transaction, we ensure that the same cron record is selected again
otherwise we restart with an empty transaction setup the selection process.
If a cron record fails to execute, it is ignored from the selection for this
run.
The run stops once there is no cron record that can be selected.

Closes #13158

trytond/CHANGELOG

@@ -1,3 +1,4 @@
+* Lock only running cron record
 * Add option to skip tests that require network
 * Add language support per field in export data
 * Add translated descriptor for Char and Text fields

trytond/trytond/ir/cron.py

@@ -5,6 +5,8 @@ import logging
 import time
 
 from dateutil.relativedelta import relativedelta
+from sql import Literal
+from sql.conditionals import Coalesce
 
 from trytond import backend
 from trytond.config import config
@@ -157,61 +159,84 @@ class Cron(DeactivableMixin, ModelSQL, ModelView):
 
     @classmethod
     def run(cls, db_name):
-        transaction = Transaction()
         logger.info('cron started for "%s"', db_name)
         now = datetime.datetime.now()
         retry = config.getint('database', 'retry')
-        with transaction.start(
-                db_name, 0, context={'_skip_warnings': True},
-                _lock_tables=[cls._table]):
-            pool = Pool()
-            Error = pool.get('ir.error')
-            crons = cls.search(['OR',
-                    ('next_call', '<=', now),
-                    ('next_call', '=', None),
-                    ])
+        count = 0
+        current_task_id = None
+        transaction_extras = {}
+        skip_task_ids = [-1]
+        while True:
+            if count:
+                time.sleep(0.02 * (retry - count))
+            with Transaction().start(
+                    db_name, 0, context={'_skip_warnings': True},
+                    **transaction_extras) as transaction:
+                pool = Pool()
+                Error = pool.get('ir.error')
+                table = cls.__table__()
+                database = transaction.database
+                cursor = transaction.connection.cursor()
+
+                query = table.select(
+                    table.id,
+                    where=(Coalesce(table.next_call, now) <= now)
+                    & ~table.id.in_(skip_task_ids)
+                    & (table.active == Literal(True)),
+                    order_by=[table.id.asc],
+                    limit=1)
+                if database.has_select_for():
+                    For = database.get_select_for_skip_locked()
+                    query.for_ = For('UPDATE')
+                cursor.execute(*query)
+                row = cursor.fetchone()
+                if not row:
+                    break
+                task_id, = row
+                if current_task_id is not None and current_task_id != task_id:
+                    # Get another task so reset the transaction setup
+                    count = 0
+                    current_task_id = None
+                    transaction_extras.clear()
+                    continue
+                task = cls(task_id)
 
-            for cron in crons:
                 def duration():
                     return (time.monotonic() - started) * 1000
                 started = time.monotonic()
-                name = '<Cron %s@%s %s>' % (cron.id, db_name, cron.method)
-                transaction_extras = {}
-                count = 0
-                while True:
-                    if count:
-                        time.sleep(0.02 * (retry - count))
-                    try:
-                        with processing(name), \
-                                transaction.new_transaction(
-                                    **transaction_extras) as cron_trans:
-                            try:
-                                cron.run_once()
-                                cron_trans.commit()
-                            except TransactionError as e:
-                                cron_trans.rollback()
-                                e.fix(transaction_extras)
-                                continue
-                            except backend.DatabaseOperationalError:
-                                if count < retry:
-                                    cron_trans.rollback()
-                                    count += 1
-                                    logger.debug("Retry: %i", count)
-                                    continue
-                                else:
-                                    raise
-                    except (UserError, UserWarning) as e:
-                        Error.report(cron, e)
+                name = '<Cron %s@%s %s>' % (task.id, db_name, task.method)
+                try:
+                    if not database.has_select_for():
+                        task.lock()
+                    with processing(name):
+                        task.run_once()
+                    task.next_call = task.compute_next_call()
+                    task.save()
+                    logger.info("%s in %i ms", name, duration())
+                except Exception as e:
+                    transaction.rollback()
+                    if isinstance(e, TransactionError):
+                        e.fix(transaction_extras)
+                        current_task_id = task_id
+                        continue
+                    if (isinstance(e, backend.DatabaseOperationalError)
+                            and count < retry):
+                        count += 1
+                        current_task_id = task_id
+                        logger.debug("Retry: %i", count)
+                        continue
+                    if isinstance(e, (UserError, UserWarning)):
+                        Error.report(task, e)
                         logger.info(
                             "%s failed after %i ms", name, duration())
-                    except Exception:
+                    else:
                         logger.exception(
                             "%s failed after %i ms", name, duration())
-                    cron.next_call = cron.compute_next_call(now)
-                    cron.save()
-                    break
-                logger.info("%s in %i ms", name, duration())
-        while transaction.tasks:
-            task_id = transaction.tasks.pop()
-            run_task(db_name, task_id)
+                    skip_task_ids.append(task_id)
+                current_task_id = None
+                count = 0
+                transaction_extras.clear()
+            while transaction.tasks:
+                task_id = transaction.tasks.pop()
+                run_task(db_name, task_id)
         logger.info('cron finished for "%s"', db_name)