From 90156f85268fad9b9c0794dfecc3c963cbb42ac6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Todorovich?= Date: Tue, 29 Sep 2026 14:23:11 -0300 Subject: [PATCH 1/3] [FIX] queue_job: deadlock when the on fail hook writes the job's records The failure is handled from a temporary cursor while the failed job's transaction still holds its locks: an on fail hook writing on the same records waited forever. Perform the job in a savepoint so that its locks are released when it fails. --- queue_job/controllers/main.py | 12 ++++-- queue_job/tests/test_run_job_controller.py | 47 ++++++++++++++++++++++ 2 files changed, 55 insertions(+), 4 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 91cda8ca3b..87b6bc04d9 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -98,10 +98,14 @@ 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): - job.perform() - # Triggers any stored computed fields before calling 'set_done' - # so that will be part of the 'exec_time' - env.flush_all() + # 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.set_done() job.store() env.flush_all() diff --git a/queue_job/tests/test_run_job_controller.py b/queue_job/tests/test_run_job_controller.py index 9982a887ff..2c60c5b0a3 100644 --- a/queue_job/tests/test_run_job_controller.py +++ b/queue_job/tests/test_run_job_controller.py @@ -1,4 +1,5 @@ # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +from contextlib import closing from unittest.mock import patch from odoo.tests.common import TransactionCase @@ -50,6 +51,52 @@ def test_runjob_on_fail(self): self.assertEqual(job.state, "failed") 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. + """ + function = self.env.ref("queue_job.job_function_queue_job__test_job") + function.on_fail_method = "_test_on_fail" + # Use a committed record, so that the other cursor can see it + partner = self.env.ref("base.partner_admin") + job = self.env["queue.job"].with_delay()._test_job() + + # The job writes on the partner (locking it), then fails + def failing_job(): + partner.name = "Written by 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. + 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], + ) + + with ( + self.assertRaises(JobError), + patch( + "odoo.addons.queue_job.models.queue_job.QueueJob._test_job", + side_effect=failing_job, + ), + patch( + "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_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 ( From dc1398158dd1cc53618895ab15c1fd9e8dbdb424 Mon Sep 17 00:00:00 2001 From: Florent Xicluna Date: Wed, 21 Oct 2026 09:46:07 +0200 Subject: [PATCH 2/3] [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_job_controller.py | 106 +++++++++++++++++---- 3 files changed, 108 insertions(+), 46 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 87b6bc04d9..fc3e888c89 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 736484a81a..d0ca30e78a 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 @@ -518,7 +518,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_job_controller.py b/queue_job/tests/test_run_job_controller.py index 2c60c5b0a3..6d7b44208d 100644 --- a/queue_job/tests/test_run_job_controller.py +++ b/queue_job/tests/test_run_job_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") @@ -42,22 +44,20 @@ def test_runjob_on_fail(self): 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_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("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.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), + 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, "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) From 302fde0bccff1876767fa77b7ae5c9281b7aee0b Mon Sep 17 00:00:00 2001 From: Florent Xicluna Date: Wed, 23 Sep 2026 07:24:23 +0200 Subject: [PATCH 3/3] [IMP] queue_job: use queue_job row lock instead of separate queue_job_lock table --- queue_job/controllers/main.py | 24 ++--- queue_job/job.py | 66 ++++---------- queue_job/jobrunner/runner.py | 89 ++++++------------- queue_job/models/__init__.py | 1 - queue_job/models/queue_job_lock.py | 16 ---- queue_job/security/ir.model.access.csv | 1 - test_queue_job/tests/common.py | 10 +++ test_queue_job/tests/test_acquire_job.py | 12 +-- test_queue_job/tests/test_requeue_dead_job.py | 42 +-------- 9 files changed, 69 insertions(+), 192 deletions(-) delete mode 100644 queue_job/models/queue_job_lock.py diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index fc3e888c89..48bcc0d53a 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -18,7 +18,7 @@ from ..delay import chain, group from ..exception import FailedJobError, RetryableJobError -from ..job import ENQUEUED, Job +from ..job import ENQUEUED, STARTED, Job _logger = logging.getLogger(__name__) @@ -31,7 +31,7 @@ def _prevent_commit(cr): """Context manager to prevent commits on a cursor. - Commiting while the job is not finished would release the job lock, causing + Committing while the job is not finished would release the job lock, causing it to be started again by the dead jobs requeuer. """ @@ -61,16 +61,12 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None: - mark it as STARTED and commit the state change - acquire the job lock - If successful, return the Job instance, otherwise return None. This - function may fail to acquire the job is not in the expected state or is - already locked by another worker. + If successful, return the Job instance, otherwise return None. + This function may fail to acquire the job, if not in the expected state + or if locked by another worker. """ - env.cr.execute( - "SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s " - "FOR NO KEY UPDATE SKIP LOCKED", - (job_uuid, ENQUEUED), - ) - if not env.cr.fetchone(): + job = Job.load(env, job_uuid, raise_if_not_found=False) + if not job or not job.lock(ENQUEUED): _logger.warning( "was requested to run job %s, but it does not exist, " "or is not in state %s, or is being handled by another worker", @@ -78,12 +74,10 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None: ENQUEUED, ) return None - job = Job.load(env, job_uuid) - assert job and job.state == ENQUEUED job.set_started() job.store() env.cr.commit() - if not job.lock(): + if not job.lock(STARTED): _logger.warning( "was requested to run job %s, but it could not be locked", job_uuid, @@ -93,7 +87,7 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None: @classmethod def _try_perform_job(cls, env, job): - """Try to perform the job, mark it done and commit if successful.""" + """Try to perform the job, mark it DONE and commit if successful.""" _logger.debug("%s started", job) # TODO refactor, the relation between env and job.env is not clear assert env.cr is job.env.cr diff --git a/queue_job/job.py b/queue_job/job.py index d0ca30e78a..40a4566e8a 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -224,15 +224,16 @@ class Job: """ @classmethod - def load(cls, env, job_uuid): + def load(cls, env, job_uuid, raise_if_not_found=True): """Read a single job from the Database Raise an error if the job is not found. """ stored = cls.db_records_from_uuids(env, [job_uuid]) - if not stored: + if stored: + return cls._load_from_db_record(stored) + if raise_if_not_found: raise NoSuchJobError(f"Job {job_uuid} does no longer exist in the storage.") - return cls._load_from_db_record(stored) @classmethod def load_many(cls, env, job_uuids): @@ -243,55 +244,23 @@ def load_many(cls, env, job_uuids): recordset = cls.db_records_from_uuids(env, job_uuids) return {cls._load_from_db_record(record) for record in recordset} - def add_lock_record(self) -> None: - """ - Create row in db to be locked while the job is being performed. - """ - self.env.cr.execute( - """ - INSERT INTO - queue_job_lock (id, queue_job_id) - SELECT - id, id - FROM - queue_job - WHERE - uuid = %s - ON CONFLICT(id) - DO NOTHING; - """, - [self.uuid], - ) - - def lock(self) -> bool: - """Lock row of job that is being performed. + def lock(self, state) -> bool: + """Lock job that is being performed. Return False if a job cannot be locked: it means that the job is not in - STARTED state or is already locked by another worker. + expected state or is already locked by another worker. + Lock is released at the commit or rollback of the transaction. """ - self.env.cr.execute( - """ - SELECT - * - FROM - queue_job_lock - WHERE - queue_job_id in ( - SELECT - id - FROM - queue_job - WHERE - uuid = %s - AND state = %s - ) - FOR NO KEY UPDATE SKIP LOCKED; - """, - [self.uuid, STARTED], + lock_query = ( + "SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s" + " FOR NO KEY UPDATE SKIP LOCKED;" ) - - # 1 job should be locked - return bool(self.env.cr.fetchall()) + self.env.cr.execute(lock_query, [self.uuid, state]) + if not self.env.cr.fetchone(): + _logger.debug("Lock NOT acquired on %s Job %s", state, self.uuid) + return False + _logger.debug("Lock acquired on %s Job %s", state, self.uuid) + return True @classmethod def _load_from_db_record(cls, job_db_record): @@ -862,7 +831,6 @@ def set_started(self): self.state = STARTED self.date_started = datetime.now() self.worker_pid = os.getpid() - self.add_lock_record() def set_done(self, result=None): self.state = DONE diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 5cbd52ef48..2d9ce8bbb6 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -354,69 +354,32 @@ def set_job_enqueued(self, uuid): ) def _query_requeue_dead_jobs(self): - return """ - UPDATE - queue_job - SET - state=( - CASE - WHEN - max_retries IS NOT NULL AND - max_retries != 0 AND -- infinite retries if max_retries is 0 - retry IS NOT NULL AND - retry>max_retries - THEN 'failed' - ELSE 'pending' - END), - retry=( - CASE - WHEN state='started' - THEN COALESCE(retry,0)+1 ELSE retry - END), - exc_name=( - CASE - WHEN - max_retries IS NOT NULL AND - max_retries != 0 AND -- infinite retries if max_retries is 0 - retry IS NOT NULL AND - retry>max_retries - THEN 'JobFoundDead' - ELSE exc_name - END), - exc_info=( - CASE - WHEN - max_retries IS NOT NULL AND - max_retries != 0 AND -- infinite retries if max_retries is 0 - retry IS NOT NULL AND - retry>max_retries - THEN 'Job found dead after too many retries' - ELSE exc_info - END) - WHERE - state IN ('enqueued','started') - AND date_enqueued < (now() AT TIME ZONE 'utc' - INTERVAL '10 sec') - AND ( - id in ( - SELECT - queue_job_id - FROM - queue_job_lock - WHERE - queue_job_lock.queue_job_id = queue_job.id - FOR NO KEY UPDATE SKIP LOCKED - ) - OR NOT EXISTS ( - SELECT - 1 - FROM - queue_job_lock - WHERE - queue_job_lock.queue_job_id = queue_job.id - ) - ) - RETURNING uuid - """ + return """\ + WITH dead_job AS ( + SELECT id, + (max_retries IS NOT NULL AND + max_retries != 0 AND -- infinite retries if max_retries is 0 + retry IS NOT NULL AND + retry > max_retries) "stop_retry", + (CASE WHEN state='started' + THEN COALESCE(retry, 0) + 1 ELSE retry END) "retry" + FROM queue_job + WHERE state IN ('enqueued', 'started') + AND date_enqueued < now() AT TIME ZONE 'utc' - INTERVAL '10 sec' + FOR NO KEY UPDATE SKIP LOCKED) + + UPDATE queue_job + SET retry = dead_job.retry, + state = CASE WHEN stop_retry THEN 'failed' ELSE 'pending' END, + exc_name = CASE WHEN stop_retry THEN 'JobFoundDead' ELSE exc_name END, + exc_info = + CASE WHEN stop_retry + THEN 'Job found dead after too many retries' + ELSE exc_info END + FROM dead_job + WHERE queue_job.id = dead_job.id + + RETURNING uuid;""" def requeue_dead_jobs(self): """ diff --git a/queue_job/models/__init__.py b/queue_job/models/__init__.py index 6265dfe9cb..4744e7ab46 100644 --- a/queue_job/models/__init__.py +++ b/queue_job/models/__init__.py @@ -3,4 +3,3 @@ from . import queue_job from . import queue_job_channel from . import queue_job_function -from . import queue_job_lock diff --git a/queue_job/models/queue_job_lock.py b/queue_job/models/queue_job_lock.py deleted file mode 100644 index b01c7f3a91..0000000000 --- a/queue_job/models/queue_job_lock.py +++ /dev/null @@ -1,16 +0,0 @@ -# Copyright 2025 ACSONE SA/NV -# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). - -from odoo import fields, models - - -class QueueJobLock(models.Model): - _name = "queue.job.lock" - _description = "Queue Job Lock" - - queue_job_id = fields.Many2one( - comodel_name="queue.job", - required=True, - ondelete="cascade", - index=True, - ) diff --git a/queue_job/security/ir.model.access.csv b/queue_job/security/ir.model.access.csv index 9a451d6a78..f9511a615e 100644 --- a/queue_job/security/ir.model.access.csv +++ b/queue_job/security/ir.model.access.csv @@ -1,6 +1,5 @@ id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink access_queue_job_manager,queue job manager,queue_job.model_queue_job,queue_job.group_queue_job_manager,1,1,0,0 -access_queue_job_lock_manager,queue job lock manager,queue_job.model_queue_job_lock,queue_job.group_queue_job_manager,1,0,0,0 access_queue_job_function_manager,queue job functions manager,queue_job.model_queue_job_function,queue_job.group_queue_job_manager,1,1,1,1 access_queue_job_channel_manager,queue job channel manager,queue_job.model_queue_job_channel,queue_job.group_queue_job_manager,1,1,1,1 access_queue_requeue_job,queue requeue job manager,queue_job.model_queue_requeue_job,queue_job.group_queue_job_manager,1,1,1,1 diff --git a/test_queue_job/tests/common.py b/test_queue_job/tests/common.py index d3173a2198..f7f19073dc 100644 --- a/test_queue_job/tests/common.py +++ b/test_queue_job/tests/common.py @@ -30,3 +30,13 @@ def _get_demo_job(self, uuid): "to make this test work", ) return job + + def is_job_locked(self, job, cr=None): + lock_query = ( + "SELECT 1 FROM queue_job WHERE uuid = %s FOR NO KEY UPDATE SKIP LOCKED" + ) + with self.env.registry.cursor() as cr: + cr.execute(lock_query, [job.uuid]) + if not cr.fetchone(): + return True + return False diff --git a/test_queue_job/tests/test_acquire_job.py b/test_queue_job/tests/test_acquire_job.py index 3f0c92a2be..cb8bf8b841 100644 --- a/test_queue_job/tests/test_acquire_job.py +++ b/test_queue_job/tests/test_acquire_job.py @@ -15,10 +15,8 @@ class TestRequeueDeadJob(JobCommonCase): def test_acquire_enqueued_job(self): job_record = self._get_demo_job(uuid="test_enqueued_job") self.assertFalse( - self.env["queue.job.lock"].search( - [("queue_job_id", "=", job_record.id)], - ), - "A job lock record should not exist at this point", + self.is_job_locked(job_record), + "A job lock should not exist at this point", ) with mock.patch.object( self.env.cr, "commit", mock.Mock(side_effect=self.env.flush_all) @@ -29,10 +27,8 @@ def test_acquire_enqueued_job(self): self.assertEqual(job.uuid, "test_enqueued_job") self.assertEqual(job.state, "started") self.assertTrue( - self.env["queue.job.lock"].search( - [("queue_job_id", "=", job_record.id)] - ), - "A job lock record should exist at this point", + self.is_job_locked(job_record), + "A job lock should exist at this point", ) def test_acquire_started_job(self): diff --git a/test_queue_job/tests/test_requeue_dead_job.py b/test_queue_job/tests/test_requeue_dead_job.py index a267c43c87..faec906bf0 100644 --- a/test_queue_job/tests/test_requeue_dead_job.py +++ b/test_queue_job/tests/test_requeue_dead_job.py @@ -1,6 +1,5 @@ # Copyright 2025 ACSONE SA/NV # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). -from contextlib import closing from datetime import datetime, timedelta from odoo.tests import tagged @@ -13,35 +12,6 @@ @tagged("post_install", "-at_install") class TestRequeueDeadJob(JobCommonCase): - def get_locks(self, uuid, cr=None): - """ - Retrieve lock rows - """ - if cr is None: - cr = self.env.cr - - cr.execute( - """ - SELECT - queue_job_id - FROM - queue_job_lock - WHERE - queue_job_id IN ( - SELECT - id - FROM - queue_job - WHERE - uuid = %s - ) - FOR NO KEY UPDATE SKIP LOCKED - """, - [uuid], - ) - - return cr.fetchall() - def test_add_lock_record(self): queue_job = self._get_demo_job("test_started_job") self.assertEqual(len(queue_job), 1) @@ -50,22 +20,16 @@ def test_add_lock_record(self): job_obj.set_started() self.assertEqual(job_obj.state, "started") - locks = self.get_locks(job_obj.uuid) - - self.assertEqual(1, len(locks)) + self.assertFalse(self.is_job_locked(job_obj)) def test_lock(self): queue_job = self._get_demo_job("test_started_job") job_obj = Job.load(self.env, queue_job.uuid) job_obj.set_started() - job_obj.lock() - - with closing(self.env.registry.cursor()) as new_cr: - locks = self.get_locks(job_obj.uuid, new_cr) + job_obj.lock("started") - # Row should be locked - self.assertEqual(0, len(locks)) + self.assertTrue(self.is_job_locked(job_obj)) def test_requeue_dead_jobs(self): queue_job = self._get_demo_job("test_enqueued_job")