diff --git a/queue_job_cron/models/ir_cron.py b/queue_job_cron/models/ir_cron.py index 7fc99cef5b..b6e86df4c1 100644 --- a/queue_job_cron/models/ir_cron.py +++ b/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: diff --git a/queue_job_cron/readme/USAGE.md b/queue_job_cron/readme/USAGE.md index ef02ba3ba3..17a16c1646 100644 --- a/queue_job_cron/readme/USAGE.md +++ b/queue_job_cron/readme/USAGE.md @@ -6,6 +6,8 @@ channel of its execution. To configure dedicated channels please refers to queue_job help: -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. diff --git a/queue_job_cron/tests/test_queue_job_cron.py b/queue_job_cron/tests/test_queue_job_cron.py index 8457ef404d..23da5c2d9b 100644 --- a/queue_job_cron/tests/test_queue_job_cron.py +++ b/queue_job_cron/tests/test_queue_job_cron.py @@ -1,17 +1,46 @@ # Copyright 2019 ACSONE SA/NV () # 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() diff --git a/queue_job_cron/views/ir_cron_view.xml b/queue_job_cron/views/ir_cron_view.xml index 135b71402a..3e373d0e4f 100644 --- a/queue_job_cron/views/ir_cron_view.xml +++ b/queue_job_cron/views/ir_cron_view.xml @@ -11,6 +11,7 @@ name="no_parallel_queue_job_run" invisible="not run_as_queue_job" /> +