From 84e68d573c841a0b81c8a329868526b70f8b3707 Mon Sep 17 00:00:00 2001 From: Benoit Aimont Date: Fri, 14 Aug 2026 11:50:10 +0200 Subject: [PATCH 1/3] [IMP] queue_job_cron - allow commit within a cron that runs as queue job --- queue_job_cron/models/ir_cron.py | 17 +++++- queue_job_cron/tests/test_queue_job_cron.py | 58 +++++++++++++++++++++ queue_job_cron/views/ir_cron_view.xml | 1 + 3 files changed, 75 insertions(+), 1 deletion(-) diff --git a/queue_job_cron/models/ir_cron.py b/queue_job_cron/models/ir_cron.py index 7fc99cef5b..0a78fe744d 100644 --- a/queue_job_cron/models/ir_cron.py +++ b/queue_job_cron/models/ir_cron.py @@ -24,6 +24,13 @@ class IrCron(models.Model): 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 +49,16 @@ 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() + if not self.allow_commit: + return server_action.run() + with self.env.registry.cursor() as new_cr: + return server_action.with_env(self.env(cr=new_cr)).run() def method_direct_trigger(self): for cron in self: diff --git a/queue_job_cron/tests/test_queue_job_cron.py b/queue_job_cron/tests/test_queue_job_cron.py index 8457ef404d..d6c97cafd7 100644 --- a/queue_job_cron/tests/test_queue_job_cron.py +++ b/queue_job_cron/tests/test_queue_job_cron.py @@ -1,7 +1,11 @@ # Copyright 2019 ACSONE SA/NV () # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html). +from odoo import api +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): @@ -12,6 +16,7 @@ def test_queue_job_cron(self): 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)]) @@ -82,3 +87,56 @@ 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 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" /> + Date: Tue, 3 Nov 2026 11:34:53 +0100 Subject: [PATCH 2/3] [IMP] queue_job_cron: never run the same cron twice in parallel Unlike the native cron worker, which holds the cron row lock for the whole run, a cron run as a queue job is only locked while it is enqueued, and no_parallel_queue_job_run only skips a job when an identical one is pending. A run lasting longer than the cron interval, or a trigger committed mid-run with allow_commit, started a second job of the same cron in parallel. The job now holds the cron row lock while it runs. A concurrent job of the same cron gets LockNotAvailable, which queue_job turns into a postponed retry. --- queue_job_cron/models/ir_cron.py | 48 +++++++++++++++++--- queue_job_cron/readme/USAGE.md | 8 ++-- queue_job_cron/tests/test_queue_job_cron.py | 49 +++++++++++++++++++-- 3 files changed, 93 insertions(+), 12 deletions(-) diff --git a/queue_job_cron/models/ir_cron.py b/queue_job_cron/models/ir_cron.py index 0a78fe744d..f38f3f00e7 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,10 +15,11 @@ 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, ) @@ -55,10 +57,46 @@ def _onchange_run_as_queue_job(self): self.allow_commit = False def _run_job_as_queue_job(self, server_action): + self.ensure_one() + if self.no_parallel_queue_job_run: + self._lock_running_cron() if not self.allow_commit: return server_action.run() with self.env.registry.cursor() as new_cr: - return server_action.with_env(self.env(cr=new_cr)).run() + new_env = self.env(cr=new_cr) + res = server_action.with_env(new_env).run() + triggered = bool( + new_env["ir.cron.trigger"] + .sudo() + .search_count( + [ + ("cron_id", "=", self.id), + ("call_at", "<=", fields.Datetime.now()), + ], + limit=1, + ) + ) + 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 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 d6c97cafd7..48a9177e78 100644 --- a/queue_job_cron/tests/test_queue_job_cron.py +++ b/queue_job_cron/tests/test_queue_job_cron.py @@ -1,6 +1,11 @@ # 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 @@ -8,8 +13,24 @@ 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 test_queue_job_cron(self): QueueJob = self.env["queue.job"] @@ -42,8 +63,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") @@ -140,3 +160,24 @@ def test_queue_job_cron_allow_commit_bypasses_forbidden_commit(self): 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_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() From 996f775449359c036b2f778200f3e68d2450692b Mon Sep 17 00:00:00 2001 From: Quentin Groulard Date: Tue, 3 Nov 2026 11:35:06 +0100 Subject: [PATCH 3/3] [IMP] queue_job_cron: support the cron progress API _notify_progress is a no-op without the progress record the native cron runner puts in context, so a batched cron run as a queue job processed a single batch per run and could not deactivate itself. The server action now runs with a progress record: the cron is triggered again while records remain, and deactivated when it asks so. --- queue_job_cron/models/ir_cron.py | 53 +++++++++++++++------ queue_job_cron/tests/test_queue_job_cron.py | 20 ++++++++ 2 files changed, 58 insertions(+), 15 deletions(-) diff --git a/queue_job_cron/models/ir_cron.py b/queue_job_cron/models/ir_cron.py index f38f3f00e7..b6e86df4c1 100644 --- a/queue_job_cron/models/ir_cron.py +++ b/queue_job_cron/models/ir_cron.py @@ -60,22 +60,19 @@ def _run_job_as_queue_job(self, server_action): self.ensure_one() if self.no_parallel_queue_job_run: self._lock_running_cron() - if not self.allow_commit: - return server_action.run() - with self.env.registry.cursor() as new_cr: - new_env = self.env(cr=new_cr) - res = server_action.with_env(new_env).run() - triggered = bool( - new_env["ir.cron.trigger"] - .sudo() - .search_count( - [ - ("cron_id", "=", self.id), - ("call_at", "<=", fields.Datetime.now()), - ], - limit=1, - ) + 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 @@ -98,6 +95,32 @@ def _lock_running_cron(self): 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: if not cron.run_as_queue_job: diff --git a/queue_job_cron/tests/test_queue_job_cron.py b/queue_job_cron/tests/test_queue_job_cron.py index 48a9177e78..23da5c2d9b 100644 --- a/queue_job_cron/tests/test_queue_job_cron.py +++ b/queue_job_cron/tests/test_queue_job_cron.py @@ -32,6 +32,9 @@ 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") @@ -161,6 +164,23 @@ def test_queue_job_cron_allow_commit_bypasses_forbidden_commit(self): 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