[18.0][IMP] queue_job_cron: Behave like native cron

queue_job_cron/models/ir_cron.py

@@ -3,6 +3,7 @@
 import logging
 
 from odoo import api, fields, models
+from odoo.tools import SQL
 
 from odoo.addons.queue_job.job import identity_exact
 
@@ -14,16 +15,24 @@ class IrCron(models.Model):
 
     no_parallel_queue_job_run = fields.Boolean(
         help="Avoid parallel run. "
-        "If the cron job is already running, the new one will be skipped. "
-        "By default, odoo never runs the same cron job in parallel. This "
-        "option is therefore set to True by default when job is run as a "
-        "queue job.",
+        "If a job of this cron is already pending, the new one will be "
+        "skipped. If one is already running, the new one is postponed until "
+        "it ends. By default, odoo never runs the same cron job in parallel. "
+        "This option is therefore set to True by default when job is run as "
+        "a queue job.",
         default=True,
     )
 
     run_as_queue_job = fields.Boolean(
         help="Specify if this cron should be ran as a queue job"
     )
+    allow_commit = fields.Boolean(
+        help="Run this cron's server action in a dedicated database cursor "
+        "when executed as a queue job, allowing it to commit partial "
+        "progress during execution. Incurs the overhead of an extra "
+        "database connection. Only effective when 'Run as Queue Job' is "
+        "enabled.",
+    )
     channel_id = fields.Many2one(
         comodel_name="queue.job.channel",
         compute="_compute_run_as_queue_job",
@@ -42,8 +51,75 @@ def _compute_run_as_queue_job(self):
             else:
                 cron.channel_id = False
 
+    @api.onchange("run_as_queue_job")
+    def _onchange_run_as_queue_job(self):
+        if not self.run_as_queue_job:
+            self.allow_commit = False
+
     def _run_job_as_queue_job(self, server_action):
-        return server_action.run()
+        self.ensure_one()
+        if self.no_parallel_queue_job_run:
+            self._lock_running_cron()
+        if self.allow_commit:
+            with self.env.registry.cursor() as new_cr:
+                new_env = self.env(cr=new_cr)
+                res, deactivate, triggered = self.with_env(
+                    new_env
+                )._run_server_action_with_progress(server_action.with_env(new_env))
+        else:
+            res, deactivate, triggered = self._run_server_action_with_progress(
+                server_action
+            )
+        if deactivate:
+            # on the job cursor: the dedicated one would wait on our row lock
+            self.active = False
+        if triggered:
+            # the cron worker skipped the triggers committed while we held the
+            # row lock: wake it up once the job cursor commits
+            self.env.cr.postcommit.add(self._notifydb)
+        return res
+
+    def _lock_running_cron(self):
+        """Hold the cron row until the job ends, as the native cron worker
+        does while it runs a cron: the worker acquires crons with
+        FOR NO KEY UPDATE SKIP LOCKED, so it won't enqueue this cron again
+        (e.g. from a trigger committed mid-run with allow_commit) before this
+        job ends. When another run holds it, LockNotAvailable is turned into a
+        postponed retry by queue_job.
+        """
+        self.env.cr.execute(
+            SQL(
+                "SELECT id FROM ir_cron WHERE id = %s FOR NO KEY UPDATE NOWAIT",
+                self.id,
+            ),
+            log_exceptions=False,
+        )
+
+    def _run_server_action_with_progress(self, server_action):
+        """Run the server action with the progress API wired as the native
+        cron runner does, since _notify_progress is a no-op without a
+        progress record in context. Re-trigger the cron instead of looping
+        over batches.
+
+        :return: a triple ``(result, deactivate, triggered)``, ``result``
+            being the return value of the server action
+        """
+        cron, progress = self._add_progress()
+        res = server_action.with_env(cron.env).run()
+        if progress.done and progress.remaining:
+            cron._trigger()
+        triggered = bool(
+            self.env["ir.cron.trigger"]
+            .sudo()
+            .search_count(
+                [
+                    ("cron_id", "=", self.id),
+                    ("call_at", "<=", fields.Datetime.now()),
+                ],
+                limit=1,
+            )
+        )
+        return res, progress.deactivate and not progress.remaining, triggered
 
     def method_direct_trigger(self):
         for cron in self:

queue_job_cron/readme/USAGE.md

@@ -6,6 +6,8 @@ channel of its execution. To configure dedicated channels please refers
 to queue_job help:
 <https://github.com/OCA/queue/blob/12.0/queue_job/README.rst>
 
-Channels can be used to manage sequential jobs and prevent concurrency
-accesses. To do that you just have to define a channel per cron limited
-to 1 at time.
+Like odoo does for regular crons, the same cron is never run twice at the
+same time, unless "No Parallel Queue Job Run" is unchecked: the job holds
+the cron lock while it runs, and a second job of the same cron is postponed
+until the first one ends. Channels can still be used to limit how many
+different crons run at the same time.

queue_job_cron/tests/test_queue_job_cron.py

@@ -1,17 +1,46 @@
 # Copyright 2019 ACSONE SA/NV (<http://acsone.eu>)
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+from unittest import mock
+
+from psycopg2.errors import LockNotAvailable
+
+from odoo import api
+from odoo.sql_db import db_connect
+from odoo.tests import Form
 from odoo.tests.common import TransactionCase
 
+from odoo.addons.queue_job.controllers.main import _prevent_commit
+
 
 class TestQueueJobCron(TransactionCase):
-    def setUp(self):
-        super().setUp()
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.cron = cls.env["ir.cron"].create(
+            {
+                "name": "Test queue job cron run",
+                "model_id": cls.env.ref("base.model_res_partner").id,
+                "state": "code",
+                "code": "model",
+                "interval_number": 1,
+                "interval_type": "days",
+                "run_as_queue_job": True,
+            }
+        )
+
+    def _run(self, code):
+        self.cron.code = code
+        self.cron._run_job_as_queue_job(server_action=self.cron.ir_actions_server_id)
+
+    def _triggers(self):
+        return self.env["ir.cron.trigger"].search([("cron_id", "=", self.cron.id)])
 
     def test_queue_job_cron(self):
         QueueJob = self.env["queue.job"]
         default_channel = self.env.ref("queue_job_cron.channel_root_ir_cron")
         cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
         self.assertFalse(cron.run_as_queue_job)
+        self.assertFalse(cron.allow_commit)
 
         cron.method_direct_trigger()
         nb_jobs = QueueJob.search_count([("name", "=", cron.name)])
@@ -37,8 +66,7 @@ def test_queue_job_cron_depends(self):
 
     def test_queue_job_cron_run(self):
         cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
-        IrCron = self.env["ir.cron"]
-        IrCron._run_job_as_queue_job(server_action=cron.ir_actions_server_id)
+        cron._run_job_as_queue_job(server_action=cron.ir_actions_server_id)
 
     def test_queue_job_no_parallelism(self):
         cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
@@ -82,3 +110,94 @@ def test_queue_job_cron_callback(self):
         self.assertEqual(nb_partners_after_cron, nb_partners + 1)
         nb_jobs_after_cron = self.env["queue.job"].search_count([])
         self.assertEqual(nb_jobs_after_cron, nb_jobs + 1)
+
+    def test_queue_job_cron_allow_commit_reset_onchange(self):
+        """Unchecking 'Run as Queue Job' in the form clears allow_commit."""
+        cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
+        with Form(cron) as cron_form:
+            cron_form.run_as_queue_job = True
+            cron_form.allow_commit = True
+            cron_form.run_as_queue_job = False
+            self.assertFalse(cron_form.allow_commit)
+
+    def test_queue_job_cron_commit_forbidden_without_allow_commit(self):
+        """Without allow_commit, the job runs on self.cr - the same
+        cursor _try_perform_job protects with _prevent_commit for any
+        real job execution - so an explicit commit() inside the server
+        action must raise, exactly as it would for a normal job."""
+        partner_model = self.env.ref("base.model_res_partner")
+        action = self.env["ir.actions.server"].create(
+            {
+                "name": "Queue job cron forbidden commit",
+                "state": "code",
+                "model_id": partner_model.id,
+                "crud_model_id": partner_model.id,
+                "code": "env.cr.commit()",
+            }
+        )
+        cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
+        self.assertFalse(cron.allow_commit)
+        with _prevent_commit(self.cr):
+            with self.assertRaisesRegex(ValueError, "Commit is forbidden"):
+                cron._run_job_as_queue_job(server_action=action)
+
+    def test_queue_job_cron_allow_commit_bypasses_forbidden_commit(self):
+        """allow_commit runs on its own dedicated cursor, so it's
+        unaffected by _prevent_commit patching self.cr - this is the
+        whole reason the option exists."""
+        partner_model = self.env.ref("base.model_res_partner")
+        with self.registry.cursor() as setup_cr:
+            setup_env = api.Environment(setup_cr, self.env.uid, {})
+            action = setup_env["ir.actions.server"].create(
+                {
+                    "name": "Queue job cron allow_commit bypasses forbidden commit",
+                    "state": "code",
+                    "model_id": partner_model.id,
+                    "crud_model_id": partner_model.id,
+                    "code": "env.cr.commit()",
+                }
+            )
+            action_id = action.id
+        cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
+        cron.write({"allow_commit": True})
+        action = self.env["ir.actions.server"].browse(action_id)
+        with _prevent_commit(self.cr):
+            cron._run_job_as_queue_job(server_action=action)  # must not raise
+
+    def test_no_progress(self):
+        self._run("model")
+        self.assertTrue(self.cron.active)
+        self.assertFalse(self._triggers())
+
+    def test_progress_remaining_retriggers(self):
+        self._run("env['ir.cron']._notify_progress(done=1, remaining=1)")
+        self.assertTrue(self.cron.active)
+        self.assertTrue(self._triggers())
+
+    def test_progress_done_deactivates(self):
+        self._run(
+            "env['ir.cron']._notify_progress(done=1, remaining=0, deactivate=True)"
+        )
+        self.assertFalse(self.cron.active)
+        self.assertFalse(self._triggers())
+
+    def test_running_cron_is_locked(self):
+        cron = self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs")
+        action = self.cron.ir_actions_server_id
+        other_cr = db_connect(self.env.cr.dbname).cursor()
+        try:
+            other_cr.execute(
+                "SELECT id FROM ir_cron WHERE id = %s FOR NO KEY UPDATE", [cron.id]
+            )
+            with self.assertRaises(LockNotAvailable), self.env.cr.savepoint():
+                cron._run_job_as_queue_job(server_action=action)
+        finally:
+            other_cr.close()
+
+    def test_parallel_run_allowed(self):
+        self.cron.no_parallel_queue_job_run = False
+        with mock.patch.object(
+            type(self.cron), "_lock_running_cron", autospec=True
+        ) as lock:
+            self._run("model")
+        lock.assert_not_called()

queue_job_cron/views/ir_cron_view.xml

@@ -11,6 +11,7 @@
                     name="no_parallel_queue_job_run"
                     invisible="not run_as_queue_job"
                 />
+                <field name="allow_commit" invisible="not run_as_queue_job" />
                 <field
                     name="channel_id"
                     invisible="not run_as_queue_job"