From 7774fed6088ab72855461232d103d2b4e3e697fd Mon Sep 17 00:00:00 2001 From: Florent Xicluna Date: Wed, 21 Oct 2026 09:46:07 +0200 Subject: [PATCH] [IMP] queue_job: do not use separate cursor to write error result --- queue_job/controllers/main.py | 39 +++---- queue_job/job.py | 9 +- queue_job/tests/test_run_rob_controller.py | 112 +++++++++++++++++---- 3 files changed, 111 insertions(+), 49 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 706c02be7..12af2ca23 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -98,14 +98,7 @@ def _try_perform_job(cls, env, job): # TODO refactor, the relation between env and job.env is not clear assert env.cr is job.env.cr with _prevent_commit(env.cr): - # On failure, rolling back the savepoint releases the row locks - # the job took on the records it updated: the failure is handled - # from another cursor, which may write on the same records (e.g. - # the on fail hook). - # On success, leaving the savepoint flushes, before 'set_done', so - # that the stored computed fields are part of the 'exec_time'. - with env.cr.savepoint(): - job.perform() + job.perform() job.set_done() job.store() env.flush_all() @@ -151,13 +144,6 @@ def _enqueue_dependent_jobs(cls, env, job): @classmethod def _runjob(cls, env: api.Environment, job: Job) -> None: - def retry_postpone(job, message, seconds=None): - job.env.clear() - with job.in_temporary_env(): - job.postpone(result=message, seconds=seconds) - job.set_pending(reset_retry=False) - job.store() - try: try: cls._try_perform_job(env, job) @@ -172,26 +158,27 @@ def retry_postpone(job, message, seconds=None): except RetryableJobError as err: # delay the job later, requeue - retry_postpone(job, str(err), seconds=err.seconds) + job.env.clear() + job.postpone(result=str(err), seconds=err.seconds) + job.set_pending(reset_retry=False) + job.store() _logger.debug("%s postponed", job) # Do not trigger the error up because we don't want an exception # traceback in the logs we should have the traceback when all # retries are exhausted - env.cr.rollback() return except (FailedJobError, Exception) as orig_exception: - buff = StringIO() - traceback.print_exc(file=buff) - traceback_txt = buff.getvalue() + with StringIO() as buff: + traceback.print_exc(file=buff) + traceback_txt = buff.getvalue() _logger.error(traceback_txt) job.env.clear() - with job.in_temporary_env(): - vals = cls._get_failure_values(job, traceback_txt, orig_exception) - job.set_failed(**vals) - job.store() - job.on_fail(vals) - buff.close() + vals = cls._get_failure_values(job, traceback_txt, orig_exception) + job.set_failed(**vals) + job.store() + job.on_fail(vals) + env.cr.commit() raise cls._enqueue_dependent_jobs(env, job) diff --git a/queue_job/job.py b/queue_job/job.py index 254f5ce85..a3c0813f8 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -8,7 +8,7 @@ import sys import uuid import weakref -from contextlib import contextmanager, nullcontext +from contextlib import contextmanager from datetime import datetime, timedelta from random import randint @@ -514,7 +514,12 @@ def perform(self): if self.job_config.allow_commit: env_context_manager = self.in_temporary_env() else: - env_context_manager = nullcontext() + # On failure, it rolls back the failed transaction to the savepoint. + # It frees locks and restores the connection to a usable state for + # failure handling (e.g. on_fail hook, using the same cursor) + # On success, the savepoint flushes pending writes, before 'set_done', + # so that the stored computed fields are part of the 'exec_time'. + env_context_manager = self.env.cr.savepoint() with env_context_manager: self.result = self.func(*tuple(self.args), **self.kwargs) except RetryableJobError as err: diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_rob_controller.py index 4090c811e..6d7b44208 100644 --- a/queue_job/tests/test_run_rob_controller.py +++ b/queue_job/tests/test_run_rob_controller.py @@ -1,5 +1,5 @@ # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). -from contextlib import closing +from datetime import datetime from unittest.mock import patch from odoo.tests.common import TransactionCase @@ -31,6 +31,8 @@ def test_get_failure_values(self): def test_runjob_success(self): job = self.env["queue.job"].with_delay()._test_job() RunJobController._runjob(self.env, job) + self.assertEqual(job.retry, 1) + self.assertEqual(job.max_retries, 5) self.assertEqual(job.state, "done") self.assertEqual(job.db_record().state, "done") @@ -39,25 +41,23 @@ def test_runjob_on_fail(self): function.on_fail_method = "_test_on_fail" job = self.env["queue.job"].with_delay()._test_job(failure_rate=1) with ( - self.assertRaises(JobError), patch( "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail" ) as mocked_hook, - patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env, + patch.object(self.env.cr, "commit") as mocked_commit, + self.assertRaises(JobError), mute_logger("odoo.addons.queue_job.controllers.main"), ): - mocked_temp_env.return_value.__enter__.return_value = self.env RunJobController._runjob(self.env, job) - self.assertEqual(job.state, "failed") - self.assertEqual(mocked_hook.call_count, 1) + self.assertEqual(job.state, "failed") + self.assertEqual(mocked_commit.call_count, 1) + self.assertEqual(mocked_hook.call_count, 1) @mute_logger("odoo.addons.queue_job.controllers.main") def test_runjob_on_fail_no_deadlock(self): """The on fail hook can write on the records the failed job wrote. - The hook runs on another cursor, while the failed job's transaction - is still open. Without rolling back the job's changes first, the hook - waits for the job's lock, which is only released after the hook. + The hook used to run on another cursor, it's not the case anymore. """ function = self.env.ref("queue_job.job_function_queue_job__test_job") function.on_fail_method = "_test_on_fail" @@ -71,17 +71,17 @@ def failing_job(): partner.flush_recordset() raise JobError("Job failed") - # The hook writes on the same partner from another cursor. Its changes - # are rolled back on close, and the lock timeout avoids waiting forever. + # The hook writes on the same partner. Its changes are rolled back on close, + # and the lock timeout avoids waiting forever. def on_fail(**kwargs): - with closing(self.registry.cursor()) as other_cr: - other_cr.execute("SET LOCAL lock_timeout = '2s'") - other_cr.execute( - "UPDATE res_partner SET name = 'Written by hook' WHERE id = %s", - [partner.id], - ) + self.env.cr.execute("SET LOCAL lock_timeout = '2s'") + self.env.cr.execute( + "UPDATE res_partner SET name = 'Written by hook' WHERE id = %s", + [partner.id], + ) with ( + patch.object(self.env.cr, "commit") as mocked_commit, self.assertRaises(JobError), patch( "odoo.addons.queue_job.models.queue_job.QueueJob._test_job", @@ -91,19 +91,89 @@ def on_fail(**kwargs): "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail", side_effect=on_fail, ) as mocked_hook, - patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env, ): - mocked_temp_env.return_value.__enter__.return_value = self.env RunJobController._runjob(self.env, job) + self.assertEqual(mocked_commit.call_count, 1) self.assertEqual(mocked_hook.call_count, 1) def test_runjob_on_fail_not_configured(self): job = self.env["queue.job"].with_delay()._test_job(failure_rate=1) with ( + patch.object(self.env.cr, "commit") as mocked_commit, + self.assertRaises(JobError), + mute_logger("odoo.addons.queue_job.controllers.main"), + ): + RunJobController._runjob(self.env, job) + self.assertEqual(job.retry, 1) + self.assertEqual(job.eta, None) + self.assertEqual(job.state, "failed") + self.assertEqual(mocked_commit.call_count, 1) + + def test_runjob_retry_then_fail(self): + retry_s = 8 + job = ( + self.env["queue.job"] + .with_delay() + ._test_job(failure_rate=1, failure_retry_seconds=retry_s) + ) + # Try once + with mute_logger("odoo.addons.queue_job.controllers.main"): + RunJobController._runjob(self.env, job) + eta_seconds = (job.eta - datetime.now()).total_seconds() + self.assertEqual(job.state, "pending") + self.assertEqual(job.retry, 1) + self.assertEqual(job.max_retries, 5) + self.assertAlmostEqual(eta_seconds, retry_s, places=0) + + # Retry 3 more times + with mute_logger("odoo.addons.queue_job.controllers.main"): + RunJobController._runjob(self.env, job) + RunJobController._runjob(self.env, job) + RunJobController._runjob(self.env, job) + eta_seconds = (job.eta - datetime.now()).total_seconds() + self.assertEqual(job.retry, 4) + self.assertEqual(job.state, "pending") + self.assertEqual(job.db_record().state, "pending") + self.assertAlmostEqual(eta_seconds, retry_s, places=0) + + # And the 5th time, it fails + with ( + patch.object(self.env.cr, "commit") as mocked_commit, self.assertRaises(JobError), - patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env, mute_logger("odoo.addons.queue_job.controllers.main"), ): - mocked_temp_env.return_value.__enter__.return_value = self.env RunJobController._runjob(self.env, job) + eta_seconds = (job.eta - datetime.now()).total_seconds() + self.assertEqual(job.retry, 5) self.assertEqual(job.state, "failed") + self.assertAlmostEqual(eta_seconds, retry_s, places=0) + self.assertEqual(mocked_commit.call_count, 1) + + def test_runjob_retry_then_succeed(self): + retry_s = 8 + job = ( + self.env["queue.job"] + .with_delay() + ._test_job(failure_rate=1, failure_retry_seconds=retry_s) + ) + # Try 4 times + with mute_logger("odoo.addons.queue_job.controllers.main"): + RunJobController._runjob(self.env, job) + RunJobController._runjob(self.env, job) + RunJobController._runjob(self.env, job) + RunJobController._runjob(self.env, job) + eta_seconds = (job.eta - datetime.now()).total_seconds() + self.assertEqual(job.retry, 4) + self.assertEqual(job.max_retries, 5) + self.assertEqual(job.state, "pending") + self.assertAlmostEqual(eta_seconds, retry_s, places=0) + + # Then it succeeds, at last + job.kwargs = {**job.kwargs, "failure_rate": 0} + with mute_logger("odoo.addons.queue_job.controllers.main"): + RunJobController._runjob(self.env, job) + eta_seconds = (job.eta - datetime.now()).total_seconds() + self.assertEqual(job.retry, 5) + self.assertEqual(job.state, "done") + self.assertEqual(job.db_record().state, "done") + self.assertAlmostEqual(eta_seconds, retry_s, places=0)