From bd82a709b92e9a1abe0c7b042ec2a1d5f853d75c Mon Sep 17 00:00:00 2001 From: Florent Xicluna Date: Tue, 22 Sep 2026 00:34:35 +0200 Subject: [PATCH 01/43] oca-port: blacklist PR(s) 715, 726, 743, 782, 775, 804, 854, 938 for queue_job --- .oca/oca-port/blacklist/queue_job.json | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 .oca/oca-port/blacklist/queue_job.json diff --git a/.oca/oca-port/blacklist/queue_job.json b/.oca/oca-port/blacklist/queue_job.json new file mode 100644 index 0000000000..3fd748577f --- /dev/null +++ b/.oca/oca-port/blacklist/queue_job.json @@ -0,0 +1,12 @@ +{ + "pull_requests": { + "OCA/queue#715": "Already in 14.0", + "OCA/queue#726": "Not applicable", + "OCA/queue#743": "Already in 14.0", + "OCA/queue#782": "Already in 14.0", + "OCA/queue#775": "Already in 14.0", + "OCA/queue#804": "Not applicable", + "OCA/queue#854": "(auto) Nothing to port from PR #854", + "OCA/queue#938": "Not applicable" + } +} From a6dec6ae690205c5066808ff568c2ddd3326def4 Mon Sep 17 00:00:00 2001 From: Vincent Hatakeyama Date: Fri, 24 Jan 2025 11:32:02 +0100 Subject: [PATCH 02/43] [FIX] queue_job: indicate that run_job need a read/write connection Without this fix, if db_replica_host is set, Odoo might pass a readonly database cursor and the FOR UPDATE in the method would fail. --- queue_job/controllers/main.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index ddec4d95ca..db27aaf46f 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -73,7 +73,13 @@ def _enqueue_dependent_jobs(self, env, job): else: break - @http.route("/queue_job/runjob", type="http", auth="none", save_session=False) + @http.route( + "/queue_job/runjob", + type="http", + auth="none", + save_session=False, + readonly=False, + ) def runjob(self, db, job_uuid, **kw): http.request.session.db = db env = http.request.env(user=odoo.SUPERUSER_ID) From 751f2bd5d8f66ce3c42904e6414c79650e2c8e10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Tue, 2 Jul 2024 13:50:11 +0200 Subject: [PATCH 03/43] [FIX] queue_job: close connection to databases without job queue Without this, we leak connections to Databases that don't have queue_job installed. --- queue_job/jobrunner/runner.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index d4dcabceb2..b070a7c4e6 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -433,6 +433,8 @@ def initialize_databases(self): for job_data in cr: self.channel_manager.notify(db_name, *job_data) _logger.info("queue job runner ready for db %s", db_name) + else: + db.close() def run_jobs(self): now = _odoo_now() From 083476567b8cf985ec04db567f748ebd0422d310 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Tue, 2 Jul 2024 13:45:08 +0200 Subject: [PATCH 04/43] [FIX] queue_job: handle exceptions in Database constructor Without this we risk connection leaks in case of exceptions in the constructor. --- queue_job/jobrunner/runner.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index b070a7c4e6..f6686e75fc 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -264,10 +264,14 @@ def __init__(self, db_name): self.db_name = db_name connection_info = _connection_info_for(db_name) self.conn = psycopg2.connect(**connection_info) - self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) - self.has_queue_job = self._has_queue_job() - if self.has_queue_job: - self._initialize() + try: + self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) + self.has_queue_job = self._has_queue_job() + if self.has_queue_job: + self._initialize() + except BaseException: + self.close() + raise def close(self): # pylint: disable=except-pass From df3b2bb520b63bae5ee62a3c1855c0ced65c0895 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Tue, 2 Jul 2024 13:06:14 +0200 Subject: [PATCH 05/43] [IMP] queue_job: HA job runner using session level advisory lock --- queue_job/jobrunner/runner.py | 26 ++++++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index f6686e75fc..adba5f8a7e 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -159,12 +159,17 @@ SELECT_TIMEOUT = 60 ERROR_RECOVERY_DELAY = 5 +PG_ADVISORY_LOCK_ID = 2293787760715711918 _logger = logging.getLogger(__name__) select = selectors.DefaultSelector +class MasterElectionLost(Exception): + pass + + # Unfortunately, it is not possible to extend the Odoo # server command line arguments, so we resort to environment variables # to configure the runner (channels mostly). @@ -268,6 +273,7 @@ def __init__(self, db_name): self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) self.has_queue_job = self._has_queue_job() if self.has_queue_job: + self._acquire_master_lock() self._initialize() except BaseException: self.close() @@ -284,6 +290,14 @@ def close(self): pass self.conn = None + def _acquire_master_lock(self): + """Acquire the master runner lock or raise MasterElectionLost""" + with closing(self.conn.cursor()) as cr: + cr.execute("SELECT pg_try_advisory_lock(%s)", (PG_ADVISORY_LOCK_ID,)) + if not cr.fetchone()[0]: + msg = f"could not acquire master runner lock on {self.db_name}" + raise MasterElectionLost(msg) + def _has_queue_job(self): with closing(self.conn.cursor()) as cr: cr.execute( @@ -416,7 +430,7 @@ def get_db_names(self): db_names = config["db_name"].split(",") else: db_names = odoo.service.db.list_dbs(True) - return db_names + return sorted(db_names) def close_databases(self, remove_jobs=True): for db_name, db in self.db_by_name.items(): @@ -525,7 +539,7 @@ def run(self): while not self._stop: # outer loop does exception recovery try: - _logger.info("initializing database connections") + _logger.debug("initializing database connections") # TODO: how to detect new databases or databases # on which queue_job is installed after server start? self.initialize_databases() @@ -540,6 +554,14 @@ def run(self): except InterruptedError: # Interrupted system call, i.e. KeyboardInterrupt during select self.stop() + except MasterElectionLost as e: + _logger.debug( + "master election lost: %s, sleeping %ds and retrying", + e, + ERROR_RECOVERY_DELAY, + ) + self.close_databases() + time.sleep(ERROR_RECOVERY_DELAY) except Exception: _logger.exception( "exception: sleeping %ds and retrying", ERROR_RECOVERY_DELAY From f689d5af42bfd9f4e542ebf453bf3a78963d6059 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Tue, 2 Jul 2024 20:35:11 +0200 Subject: [PATCH 06/43] [IMP] queue_job: make sorting more explicit --- queue_job/jobrunner/runner.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index adba5f8a7e..4d27a0e1cc 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -430,7 +430,7 @@ def get_db_names(self): db_names = config["db_name"].split(",") else: db_names = odoo.service.db.list_dbs(True) - return sorted(db_names) + return db_names def close_databases(self, remove_jobs=True): for db_name, db in self.db_by_name.items(): @@ -443,7 +443,8 @@ def close_databases(self, remove_jobs=True): self.db_by_name = {} def initialize_databases(self): - for db_name in self.get_db_names(): + for db_name in sorted(self.get_db_names()): + # sorting is important to avoid deadlocks in acquiring the master lock db = Database(db_name) if db.has_queue_job: self.db_by_name[db_name] = db From 53fa96d41b5a86075f43491ab0c0f118611d72de Mon Sep 17 00:00:00 2001 From: Zina Rasoamanana Date: Fri, 6 Dec 2024 11:32:24 +0100 Subject: [PATCH 07/43] [IMP] queue_job: remove cron garbage collector and automatically requeue jobs in timeout [IMP] queue_job: increment 'retry' when re-queuing job that have been killed --- queue_job/__manifest__.py | 2 +- queue_job/controllers/main.py | 2 + queue_job/data/queue_data.xml | 9 -- queue_job/job.py | 56 +++++++ queue_job/jobrunner/runner.py | 141 +++++++++++++----- .../migrations/14.0.3.16.0/pre-migration.py | 11 ++ queue_job/models/__init__.py | 1 + queue_job/models/queue_job.py | 50 ------- queue_job/models/queue_job_lock.py | 16 ++ queue_job/security/ir.model.access.csv | 1 + test_queue_job/__manifest__.py | 1 + test_queue_job/data/queue_job_test_job.xml | 18 +++ test_queue_job/models/test_models.py | 31 ++++ test_queue_job/tests/__init__.py | 1 + test_queue_job/tests/test_autovacuum.py | 20 ++- test_queue_job/tests/test_requeue_dead_job.py | 101 +++++++++++++ 16 files changed, 355 insertions(+), 106 deletions(-) create mode 100644 queue_job/migrations/14.0.3.16.0/pre-migration.py create mode 100644 queue_job/models/queue_job_lock.py create mode 100644 test_queue_job/data/queue_job_test_job.xml create mode 100644 test_queue_job/tests/test_requeue_dead_job.py diff --git a/queue_job/__manifest__.py b/queue_job/__manifest__.py index e25e4742e9..698e2fc8d3 100644 --- a/queue_job/__manifest__.py +++ b/queue_job/__manifest__.py @@ -2,7 +2,7 @@ { "name": "Job Queue", - "version": "14.0.3.15.1", + "version": "14.0.3.16.0", "author": "Camptocamp,ACSONE SA/NV,Odoo Community Association (OCA)", "website": "https://github.com/OCA/queue", "license": "LGPL-3", diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index db27aaf46f..7f4030376c 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -32,6 +32,8 @@ def _try_perform_job(self, env, job): job.set_started() job.store() env.cr.commit() + job.lock() + _logger.debug("%s started", job) job.perform() diff --git a/queue_job/data/queue_data.xml b/queue_job/data/queue_data.xml index ca5a747746..a2680cc475 100644 --- a/queue_job/data/queue_data.xml +++ b/queue_job/data/queue_data.xml @@ -1,15 +1,6 @@ - - Jobs Garbage Collector - 5 - minutes - -1 - - code - model.requeue_stuck_jobs() - Job failed diff --git a/queue_job/job.py b/queue_job/job.py index e486d1f001..100acb53b5 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -238,6 +238,61 @@ 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): + """ + 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): + """ + Lock row of job that is being performed + + If a job cannot be locked, + it means that the job wasn't started, + a RetryableJobError is thrown. + """ + self.env.cr.execute( + """ + SELECT + * + FROM + queue_job_lock + WHERE + queue_job_id in ( + SELECT + id + FROM + queue_job + WHERE + uuid = %s + AND state='started' + ) + FOR UPDATE; + """, + [self.uuid], + ) + + # 1 job should be locked + if 1 != len(self.env.cr.fetchall()): + raise RetryableJobError( + f"Trying to lock job that wasn't started, uuid: {self.uuid}" + ) + @classmethod def _load_from_db_record(cls, job_db_record): stored = job_db_record @@ -830,6 +885,7 @@ 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 4d27a0e1cc..fa0379197d 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -114,22 +114,6 @@ * After creating a new database or installing queue_job on an existing database, Odoo must be restarted for the runner to detect it. -* When Odoo shuts down normally, it waits for running jobs to finish. - However, when the Odoo server crashes or is otherwise force-stopped, - running jobs are interrupted while the runner has no chance to know - they have been aborted. In such situations, jobs may remain in - ``started`` or ``enqueued`` state after the Odoo server is halted. - Since the runner has no way to know if they are actually running or - not, and does not know for sure if it is safe to restart the jobs, - it does not attempt to restart them automatically. Such stale jobs - therefore fill the running queue and prevent other jobs to start. - You must therefore requeue them manually, either from the Jobs view, - or by running the following SQL statement *before starting Odoo*: - -.. code-block:: sql - - update queue_job set state='pending' where state in ('started', 'enqueued') - .. rubric:: Footnotes .. [1] From a security standpoint, it is safe to have an anonymous HTTP @@ -155,7 +139,7 @@ from odoo.tools import config from . import queue_job_config -from .channels import ENQUEUED, NOT_DONE, PENDING, ChannelManager +from .channels import ENQUEUED, NOT_DONE, ChannelManager SELECT_TIMEOUT = 60 ERROR_RECOVERY_DELAY = 5 @@ -212,28 +196,6 @@ def _connection_info_for(db_name): def _async_http_get(scheme, host, port, user, password, db_name, job_uuid): - # Method to set failed job (due to timeout, etc) as pending, - # to avoid keeping it as enqueued. - def set_job_pending(): - connection_info = _connection_info_for(db_name) - conn = psycopg2.connect(**connection_info) - conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) - with closing(conn.cursor()) as cr: - cr.execute( - "UPDATE queue_job SET state=%s, " - "date_enqueued=NULL, date_started=NULL " - "WHERE uuid=%s and state=%s " - "RETURNING uuid", - (PENDING, job_uuid, ENQUEUED), - ) - if cr.fetchone(): - _logger.warning( - "state of job %s was reset from %s to %s", - job_uuid, - ENQUEUED, - PENDING, - ) - # TODO: better way to HTTP GET asynchronously (grequest, ...)? # if this was python3 I would be doing this with # asyncio, aiohttp and aiopg @@ -241,6 +203,7 @@ def urlopen(): url = "{}://{}:{}/queue_job/runjob?db={}&job_uuid={}".format( scheme, host, port, db_name, job_uuid ) + # pylint: disable=except-pass try: auth = None if user: @@ -254,10 +217,10 @@ def urlopen(): # for codes between 500 and 600 response.raise_for_status() except requests.Timeout: - set_job_pending() + # A timeout is a normal behaviour, it shouldn't be logged as an exception + pass except Exception: _logger.exception("exception in GET %s", url) - set_job_pending() thread = threading.Thread(target=urlopen) thread.daemon = True @@ -361,6 +324,96 @@ def set_job_enqueued(self, uuid): (ENQUEUED, 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 + id in ( + SELECT + queue_job_id + FROM + queue_job_lock + WHERE + queue_job_id in ( + SELECT + id + FROM + queue_job + WHERE + state IN ('enqueued','started') + AND date_enqueued < + (now() AT TIME ZONE 'utc' - INTERVAL '10 sec') + ) + FOR UPDATE SKIP LOCKED + ) + RETURNING uuid + """ + + def requeue_dead_jobs(self): + """ + Set started and enqueued jobs but not locked to pending + + A job is locked when it's being executed + When a job is killed, it releases the lock + + If the number of retries exceeds the number of max retries, + the job is set as 'failed' with the error 'JobFoundDead'. + + Adding a buffer on 'date_enqueued' to check + that it has been enqueued for more than 10sec. + This prevents from requeuing jobs before they are actually started. + + When Odoo shuts down normally, it waits for running jobs to finish. + However, when the Odoo server crashes or is otherwise force-stopped, + running jobs are interrupted while the runner has no chance to know + they have been aborted. + """ + + with closing(self.conn.cursor()) as cr: + query = self._query_requeue_dead_jobs() + + cr.execute(query) + + for (uuid,) in cr.fetchall(): + _logger.warning("Re-queued dead job with uuid: %s", uuid) + class QueueJobRunner: def __init__( @@ -455,6 +508,11 @@ def initialize_databases(self): else: db.close() + def requeue_dead_jobs(self): + for db in self.db_by_name.values(): + if db.has_queue_job: + db.requeue_dead_jobs() + def run_jobs(self): now = _odoo_now() for job in self.channel_manager.get_jobs_to_run(now): @@ -547,6 +605,7 @@ def run(self): _logger.info("database connections ready") # inner loop does the normal processing while not self._stop: + self.requeue_dead_jobs() self.process_notifications() self.run_jobs() self.wait_notification() diff --git a/queue_job/migrations/14.0.3.16.0/pre-migration.py b/queue_job/migrations/14.0.3.16.0/pre-migration.py new file mode 100644 index 0000000000..931c336866 --- /dev/null +++ b/queue_job/migrations/14.0.3.16.0/pre-migration.py @@ -0,0 +1,11 @@ +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +from openupgradelib import openupgrade + + +@openupgrade.migrate() +def migrate(env, version): + # Remove cron garbage collector + openupgrade.delete_records_safely_by_xml_id( + env, + ["queue_job.ir_cron_queue_job_garbage_collector"], + ) diff --git a/queue_job/models/__init__.py b/queue_job/models/__init__.py index 4744e7ab46..6265dfe9cb 100644 --- a/queue_job/models/__init__.py +++ b/queue_job/models/__init__.py @@ -3,3 +3,4 @@ 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.py b/queue_job/models/queue_job.py index 5c69655181..d96b6255f7 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -6,7 +6,6 @@ from datetime import datetime, timedelta from odoo import _, api, exceptions, fields, models -from odoo.osv import expression from odoo.tools import config, html_escape, index_exists from odoo.addons.base_sparse_field.models.fields import Serialized @@ -412,55 +411,6 @@ def autovacuum(self): break return True - def requeue_stuck_jobs(self, enqueued_delta=5, started_delta=0): - """Fix jobs that are in a bad states - - :param in_queue_delta: lookup time in minutes for jobs - that are in enqueued state - - :param started_delta: lookup time in minutes for jobs - that are in enqueued state, - 0 means that it is not checked - """ - self._get_stuck_jobs_to_requeue( - enqueued_delta=enqueued_delta, started_delta=started_delta - ).requeue() - return True - - def _get_stuck_jobs_domain(self, queue_dl, started_dl): - domain = [] - now = fields.datetime.now() - if queue_dl: - queue_dl = now - timedelta(minutes=queue_dl) - domain.append( - [ - "&", - ("date_enqueued", "<=", fields.Datetime.to_string(queue_dl)), - ("state", "=", "enqueued"), - ] - ) - if started_dl: - started_dl = now - timedelta(minutes=started_dl) - domain.append( - [ - "&", - ("date_started", "<=", fields.Datetime.to_string(started_dl)), - ("state", "=", "started"), - ] - ) - if not domain: - raise exceptions.ValidationError( - _("If both parameters are 0, ALL jobs will be requeued!") - ) - return expression.OR(domain) - - def _get_stuck_jobs_to_requeue(self, enqueued_delta, started_delta): - job_model = self.env["queue.job"] - stuck_jobs = job_model.search( - self._get_stuck_jobs_domain(enqueued_delta, started_delta) - ) - return stuck_jobs - def related_action_open_record(self): """Open a form view with the record(s) of the job. diff --git a/queue_job/models/queue_job_lock.py b/queue_job/models/queue_job_lock.py new file mode 100644 index 0000000000..b01c7f3a91 --- /dev/null +++ b/queue_job/models/queue_job_lock.py @@ -0,0 +1,16 @@ +# 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 634daf8ede..4def7dc38a 100644 --- a/queue_job/security/ir.model.access.csv +++ b/queue_job/security/ir.model.access.csv @@ -1,5 +1,6 @@ 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,1,1 +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/__manifest__.py b/test_queue_job/__manifest__.py index 85a7861365..994138d8e2 100644 --- a/test_queue_job/__manifest__.py +++ b/test_queue_job/__manifest__.py @@ -13,6 +13,7 @@ "data/queue_job_channel_data.xml", "data/queue_job_function_data.xml", "security/ir.model.access.csv", + "data/queue_job_test_job.xml", ], "installable": True, } diff --git a/test_queue_job/data/queue_job_test_job.xml b/test_queue_job/data/queue_job_test_job.xml new file mode 100644 index 0000000000..8a28ab70a0 --- /dev/null +++ b/test_queue_job/data/queue_job_test_job.xml @@ -0,0 +1,18 @@ + + + + + + diff --git a/test_queue_job/models/test_models.py b/test_queue_job/models/test_models.py index 03fa792137..2e85be8057 100644 --- a/test_queue_job/models/test_models.py +++ b/test_queue_job/models/test_models.py @@ -1,6 +1,8 @@ # Copyright 2016 Camptocamp SA # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +from datetime import datetime, timedelta + from odoo import api, fields, models from odoo.addons.queue_job.delay import chain @@ -29,6 +31,35 @@ def testing_related__url(self, **kwargs): "url": kwargs["url"].format(subject=subject), } + @api.model + def _create_test_started_job(self, uuid): + """Create started jobs to be used within tests""" + self.env["queue.job"].with_context( + _job_edit_sentinel=self.env["queue.job"].EDIT_SENTINEL, + ).create( + { + "uuid": uuid, + "state": "started", + "model_name": "queue.job", + "method_name": "write", + } + ) + + @api.model + def _create_test_enqueued_job(self, uuid): + """Create enqueued jobs to be used within tests""" + self.env["queue.job"].with_context( + _job_edit_sentinel=self.env["queue.job"].EDIT_SENTINEL, + ).create( + { + "uuid": uuid, + "state": "enqueued", + "model_name": "queue.job", + "method_name": "write", + "date_enqueued": datetime.now() - timedelta(minutes=1), + } + ) + class ModelTestQueueJob(models.Model): diff --git a/test_queue_job/tests/__init__.py b/test_queue_job/tests/__init__.py index 0405022ce0..62347148e5 100644 --- a/test_queue_job/tests/__init__.py +++ b/test_queue_job/tests/__init__.py @@ -7,3 +7,4 @@ from . import test_job_function from . import test_related_actions from . import test_delay_mocks +from . import test_requeue_dead_job diff --git a/test_queue_job/tests/test_autovacuum.py b/test_queue_job/tests/test_autovacuum.py index 851af421bc..422a527082 100644 --- a/test_queue_job/tests/test_autovacuum.py +++ b/test_queue_job/tests/test_autovacuum.py @@ -27,12 +27,16 @@ def test_autovacuum(self): date_done = datetime.now() - timedelta(days=29) stored.write({"date_done": date_done}) self.env["queue.job"].autovacuum() - self.assertEqual(len(self.env["queue.job"].search([])), 1) + self.assertEqual( + len(self.env["queue.job"].search([("channel", "!=", False)])), 1 + ) date_done = datetime.now() - timedelta(days=31) stored.write({"date_done": date_done}) self.env["queue.job"].autovacuum() - self.assertEqual(len(self.env["queue.job"].search([])), 0) + self.assertEqual( + len(self.env["queue.job"].search([("channel", "!=", False)])), 0 + ) def test_autovacuum_multi_channel(self): root_channel = self.env.ref("queue_job.channel_root") @@ -47,11 +51,17 @@ def test_autovacuum_multi_channel(self): {"channel": channel_60days.complete_name, "date_done": date_done} ) - self.assertEqual(len(self.env["queue.job"].search([])), 2) + self.assertEqual( + len(self.env["queue.job"].search([("channel", "!=", False)])), 2 + ) self.env["queue.job"].autovacuum() - self.assertEqual(len(self.env["queue.job"].search([])), 1) + self.assertEqual( + len(self.env["queue.job"].search([("channel", "!=", False)])), 1 + ) date_done = datetime.now() - timedelta(days=61) job_60days.write({"date_done": date_done}) self.env["queue.job"].autovacuum() - self.assertEqual(len(self.env["queue.job"].search([])), 0) + self.assertEqual( + len(self.env["queue.job"].search([("channel", "!=", False)])), 0 + ) diff --git a/test_queue_job/tests/test_requeue_dead_job.py b/test_queue_job/tests/test_requeue_dead_job.py new file mode 100644 index 0000000000..a6328fed76 --- /dev/null +++ b/test_queue_job/tests/test_requeue_dead_job.py @@ -0,0 +1,101 @@ +# 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 + +from odoo.addons.queue_job.job import Job +from odoo.addons.queue_job.jobrunner.runner import Database + +from .common import JobCommonCase + + +@tagged("post_install", "-at_install") +class TestRequeueDeadJob(JobCommonCase): + def _get_demo_job(self, uuid): + # job created during load of demo data + job = self.env["queue.job"].search( + [ + ("uuid", "=", uuid), + ], + limit=1, + ) + + self.assertTrue( + job, + f"Demo data queue job {uuid} should be loaded in order" + " to make this tests work", + ) + + return job + + 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 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) + job_obj = Job.load(self.env, queue_job.uuid) + + job_obj.set_started() + self.assertEqual(job_obj.state, "started") + + locks = self.get_locks(job_obj.uuid) + + self.assertEqual(1, len(locks)) + + 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) + + # Row should be locked + self.assertEqual(0, len(locks)) + + def test_requeue_dead_jobs(self): + queue_job = self._get_demo_job("test_enqueued_job") + job_obj = Job.load(self.env, queue_job.uuid) + + job_obj.set_enqueued() + job_obj.set_started() + job_obj.date_enqueued = datetime.now() - timedelta(minutes=1) + job_obj.store() + + # requeue dead jobs using current cursor + query = Database(self.env.cr.dbname)._query_requeue_dead_jobs() + self.env.cr.execute(query) + + uuids_requeued = self.env.cr.fetchall() + self.assertTrue(queue_job.uuid in j[0] for j in uuids_requeued) From ab8bf57370b55818dfc3e187dfd5371248dd6013 Mon Sep 17 00:00:00 2001 From: Benjamin Willig Date: Wed, 17 Apr 2024 18:01:51 +0200 Subject: [PATCH 08/43] [FIX] queue_job: fix exception msg handling 'name' attributes for odoo's exception has been deprecated and produces a warning message. This commit fixes the behavior --- queue_job/controllers/main.py | 4 +++- queue_job/tests/__init__.py | 1 + queue_job/tests/test_run_job_controller.py | 17 +++++++++++++++++ 3 files changed, 21 insertions(+), 1 deletion(-) create mode 100644 queue_job/tests/test_run_job_controller.py diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 7f4030376c..1d7a830835 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -173,7 +173,9 @@ def _get_failure_values(self, job, traceback_txt, orig_exception): exception_name = orig_exception.__class__.__name__ if hasattr(orig_exception, "__module__"): exception_name = orig_exception.__module__ + "." + exception_name - exc_message = getattr(orig_exception, "name", str(orig_exception)) + exc_message = ( + orig_exception.args[0] if orig_exception.args else str(orig_exception) + ) return { "exc_info": traceback_txt, "exc_name": exception_name, diff --git a/queue_job/tests/__init__.py b/queue_job/tests/__init__.py index db53ac3a60..51fd9816ba 100644 --- a/queue_job/tests/__init__.py +++ b/queue_job/tests/__init__.py @@ -1,3 +1,4 @@ +from . import test_run_job_controller from . import test_runner_channels from . import test_runner_runner from . import test_delayable diff --git a/queue_job/tests/test_run_job_controller.py b/queue_job/tests/test_run_job_controller.py new file mode 100644 index 0000000000..bb63bc82ec --- /dev/null +++ b/queue_job/tests/test_run_job_controller.py @@ -0,0 +1,17 @@ +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo.tests.common import TransactionCase + +from ..controllers.main import RunJobController +from ..job import Job + + +class TestRunJobController(TransactionCase): + def test_get_failure_values(self): + method = self.env["res.users"].mapped + job = Job(method) + ctrl = RunJobController() + rslt = ctrl._get_failure_values(job, "info", Exception("zero", "one")) + self.assertEqual( + rslt, {"exc_info": "info", "exc_name": "Exception", "exc_message": "zero"} + ) From 7a0ac6dfa88a93b476212d2478d75c498e6a87cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Honor=C3=A9?= Date: Tue, 12 Aug 2025 14:20:30 +0200 Subject: [PATCH 09/43] [IMP] queue_job: avoid deprecation warning about datetime utcnow() datetime.datetime.utcnow() is now deprecated and should be replaced by datetime.datetime.now() (optional TZ parameter). As the original _odoo_now() doesn't contain the Timezone, the parameter datetime.UTC is not added into this improvement --- queue_job/jobrunner/runner.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index fa0379197d..7e072790fd 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -123,7 +123,6 @@ of running Odoo is obviously not for production purposes. """ -import datetime import logging import os import selectors @@ -170,15 +169,10 @@ def _channels(): ) -def _datetime_to_epoch(dt): +def _odoo_now(): # important: this must return the same as postgresql # EXTRACT(EPOCH FROM TIMESTAMP dt) - return (dt - datetime.datetime(1970, 1, 1)).total_seconds() - - -def _odoo_now(): - dt = datetime.datetime.utcnow() - return _datetime_to_epoch(dt) + return time.time() def _connection_info_for(db_name): From 29be61a3d90b60641b052443144dd9a6eb1290ee Mon Sep 17 00:00:00 2001 From: Florent Xicluna Date: Mon, 2 Jun 2025 14:51:45 +0200 Subject: [PATCH 10/43] [IMP] queue_job: prevent invalid change of job status --- queue_job/models/queue_job.py | 17 ++++-- queue_job/tests/test_wizards.py | 57 ++++++++++++++++++++ queue_job/wizards/queue_jobs_to_cancelled.py | 6 +-- queue_job/wizards/queue_jobs_to_done.py | 2 + queue_job/wizards/queue_requeue_job.py | 1 + 5 files changed, 76 insertions(+), 7 deletions(-) diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index d96b6255f7..64f4e5ae42 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -16,6 +16,7 @@ from ..job import ( CANCELLED, DONE, + ENQUEUED, FAILED, PENDING, STARTED, @@ -326,18 +327,26 @@ def _change_job_state(self, state, result=None): raise ValueError("State not supported: %s" % state) def button_done(self): + # If job was set to STARTED or CANCELLED, do not set it to DONE + states_from = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, FAILED) result = _("Manually set to done by %s") % self.env.user.name - self._change_job_state(DONE, result=result) + records = self.filtered(lambda job_: job_.state in states_from) + records._change_job_state(DONE, result=result) return True def button_cancelled(self): + # If job was set to DONE or WAIT_DEPENDENCIES, do not cancel it + states_from = (PENDING, ENQUEUED, FAILED) result = _("Cancelled by %s") % self.env.user.name - self._change_job_state(CANCELLED, result=result) + records = self.filtered(lambda job_: job_.state in states_from) + records._change_job_state(CANCELLED, result=result) return True def requeue(self): - jobs_to_requeue = self.filtered(lambda job_: job_.state != WAIT_DEPENDENCIES) - jobs_to_requeue._change_job_state(PENDING) + # If job is already in queue or started, do not requeue it + states_from = (FAILED, DONE, CANCELLED) + records = self.filtered(lambda job_: job_.state in states_from) + records._change_job_state(PENDING) return True def _message_post_on_failure(self): diff --git a/queue_job/tests/test_wizards.py b/queue_job/tests/test_wizards.py index 2ac162d313..7738836d2f 100644 --- a/queue_job/tests/test_wizards.py +++ b/queue_job/tests/test_wizards.py @@ -46,3 +46,60 @@ def test_03_done(self): wizard = self._wizard("queue.jobs.to.done") wizard.set_done() self.assertEqual(self.job.state, "done") + + def test_04_requeue_forbidden(self): + wizard = self._wizard("queue.requeue.job") + + # State WAIT_DEPENDENCIES is not requeued + self.job.state = "wait_dependencies" + wizard.requeue() + self.assertEqual(self.job.state, "wait_dependencies") + + # State PENDING, ENQUEUED or STARTED are ignored too + for test_state in ("pending", "enqueued", "started"): + self.job.state = test_state + wizard.requeue() + self.assertEqual(self.job.state, test_state) + + # States CANCELLED, DONE or FAILED will change status + self.job.state = "cancelled" + wizard.requeue() + self.assertEqual(self.job.state, "pending") + + def test_05_cancel_forbidden(self): + wizard = self._wizard("queue.jobs.to.cancelled") + + # State WAIT_DEPENDENCIES is not cancelled + self.job.state = "wait_dependencies" + wizard.set_cancelled() + self.assertEqual(self.job.state, "wait_dependencies") + + # State DONE is not cancelled + self.job.state = "done" + wizard.set_cancelled() + self.assertEqual(self.job.state, "done") + + # State PENDING, ENQUEUED or FAILED will be cancelled + for test_state in ("pending", "enqueued"): + self.job.state = test_state + wizard.set_cancelled() + self.assertEqual(self.job.state, "cancelled") + + def test_06_done_forbidden(self): + wizard = self._wizard("queue.jobs.to.done") + + # State STARTED is not set DONE manually + self.job.state = "started" + wizard.set_done() + self.assertEqual(self.job.state, "started") + + # State CANCELLED is not cancelled + self.job.state = "cancelled" + wizard.set_done() + self.assertEqual(self.job.state, "cancelled") + + # State WAIT_DEPENDENCIES, PENDING, ENQUEUED or FAILED will be set to DONE + for test_state in ("wait_dependencies", "pending", "enqueued"): + self.job.state = test_state + wizard.set_done() + self.assertEqual(self.job.state, "done") diff --git a/queue_job/wizards/queue_jobs_to_cancelled.py b/queue_job/wizards/queue_jobs_to_cancelled.py index 9e73374ebd..bb9f831576 100644 --- a/queue_job/wizards/queue_jobs_to_cancelled.py +++ b/queue_job/wizards/queue_jobs_to_cancelled.py @@ -10,8 +10,8 @@ class SetJobsToCancelled(models.TransientModel): _description = "Cancel all selected jobs" def set_cancelled(self): - jobs = self.job_ids.filtered( - lambda x: x.state in ("pending", "failed", "enqueued") - ) + # Only jobs with state PENDING, FAILED, ENQUEUED + # will change to CANCELLED + jobs = self.job_ids jobs.button_cancelled() return {"type": "ir.actions.act_window_close"} diff --git a/queue_job/wizards/queue_jobs_to_done.py b/queue_job/wizards/queue_jobs_to_done.py index ff1366ffed..caf8129213 100644 --- a/queue_job/wizards/queue_jobs_to_done.py +++ b/queue_job/wizards/queue_jobs_to_done.py @@ -10,6 +10,8 @@ class SetJobsToDone(models.TransientModel): _description = "Set all selected jobs to done" def set_done(self): + # Only jobs with state WAIT_DEPENDENCIES, PENDING, ENQUEUED or FAILED + # will change to DONE jobs = self.job_ids jobs.button_done() return {"type": "ir.actions.act_window_close"} diff --git a/queue_job/wizards/queue_requeue_job.py b/queue_job/wizards/queue_requeue_job.py index 67d2ffcbdc..a88256300f 100644 --- a/queue_job/wizards/queue_requeue_job.py +++ b/queue_job/wizards/queue_requeue_job.py @@ -20,6 +20,7 @@ def _default_job_ids(self): ) def requeue(self): + # Only jobs with state FAILED, DONE or CANCELLED will change to PENDING jobs = self.job_ids jobs.requeue() return {"type": "ir.actions.act_window_close"} From 7ea337125d630c1447c1028ecf0715810404a931 Mon Sep 17 00:00:00 2001 From: hoangtrann Date: Sat, 22 Nov 2025 06:05:08 +0700 Subject: [PATCH 11/43] [IMP] queue_job: requeue orphaned jobs --- queue_job/jobrunner/runner.py | 43 +++++++++++-------- test_queue_job/tests/test_requeue_dead_job.py | 16 +++++++ 2 files changed, 42 insertions(+), 17 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 7e072790fd..e190c03883 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -359,23 +359,26 @@ def _query_requeue_dead_jobs(self): ELSE exc_info END) WHERE - id in ( - SELECT - queue_job_id - FROM - queue_job_lock - WHERE - queue_job_id in ( - SELECT - id - FROM - queue_job - WHERE - state IN ('enqueued','started') - AND date_enqueued < - (now() AT TIME ZONE 'utc' - INTERVAL '10 sec') - ) - FOR UPDATE SKIP LOCKED + 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 UPDATE SKIP LOCKED + ) + OR NOT EXISTS ( + SELECT + 1 + FROM + queue_job_lock + WHERE + queue_job_lock.queue_job_id = queue_job.id + ) ) RETURNING uuid """ @@ -398,6 +401,12 @@ def requeue_dead_jobs(self): However, when the Odoo server crashes or is otherwise force-stopped, running jobs are interrupted while the runner has no chance to know they have been aborted. + + This also handles orphaned jobs (enqueued but never started, no lock). + This edge case occurs when the runner marks a job as 'enqueued' + but the HTTP request to start the job never reaches the Odoo server + (e.g., due to server shutdown/crash between setting enqueued and + the controller receiving the request). """ with closing(self.conn.cursor()) as cr: diff --git a/test_queue_job/tests/test_requeue_dead_job.py b/test_queue_job/tests/test_requeue_dead_job.py index a6328fed76..58890adf24 100644 --- a/test_queue_job/tests/test_requeue_dead_job.py +++ b/test_queue_job/tests/test_requeue_dead_job.py @@ -99,3 +99,19 @@ def test_requeue_dead_jobs(self): uuids_requeued = self.env.cr.fetchall() self.assertTrue(queue_job.uuid in j[0] for j in uuids_requeued) + + def test_requeue_orphaned_jobs(self): + queue_job = self._get_demo_job("test_enqueued_job") + job_obj = Job.load(self.env, queue_job.uuid) + + # Only enqueued job, don't set it to started to simulate the scenario + # that system shutdown before job is starting + job_obj.set_enqueued() + job_obj.date_enqueued = datetime.now() - timedelta(minutes=1) + job_obj.store() + + # job is now picked up by the requeue query (which includes orphaned jobs) + query = Database(self.env.cr.dbname)._query_requeue_dead_jobs() + self.env.cr.execute(query) + uuids_requeued = self.env.cr.fetchall() + self.assertTrue(queue_job.uuid in j[0] for j in uuids_requeued) From ec8693f6e504ea34ba65fd25cc73657de9c70fad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Fri, 2 Jan 2026 15:00:27 +0100 Subject: [PATCH 12/43] [FIX] queue_job: fix retry mechanisme for job dependencies When a SerializationFailure occurs when updating the state of dependent jobs, the cursor is not usable anymore so the retry failed with `current transaction is aborted`. A savepoint fixes that. --- queue_job/controllers/main.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 1d7a830835..4970114181 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -50,14 +50,15 @@ def _enqueue_dependent_jobs(self, env, job): tries = 0 while True: try: - job.enqueue_waiting() + with job.env.cr.savepoint(): + job.enqueue_waiting() except OperationalError as err: # Automatically retry the typical transaction serialization # errors if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY: raise if tries >= DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE: - _logger.info( + _logger.error( "%s, maximum number of tries reached to update dependencies", errorcodes.lookup(err.pgcode), ) From 24266ab7556004ee27827d0b678fe1a54170046f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Fri, 2 Jan 2026 19:36:25 +0100 Subject: [PATCH 13/43] [FIX] queue_job: set exec_time readonly --- queue_job/models/queue_job.py | 1 + 1 file changed, 1 insertion(+) diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 64f4e5ae42..f5c81bbec6 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -104,6 +104,7 @@ class QueueJob(models.Model): date_done = fields.Datetime(readonly=True) exec_time = fields.Float( string="Execution Time (avg)", + readonly=True, group_operator="avg", help="Time required to execute this job in seconds. Average when grouped.", ) From 944b9219136fc63614ca3c89c01acb74cee80e87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Sat, 3 Jan 2026 11:34:18 +0100 Subject: [PATCH 14/43] [IMP] queue_job: use state constant in lock function --- queue_job/job.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/queue_job/job.py b/queue_job/job.py index 100acb53b5..bb5b8e2cf3 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -280,11 +280,11 @@ def lock(self): queue_job WHERE uuid = %s - AND state='started' + AND state = %s ) FOR UPDATE; """, - [self.uuid], + [self.uuid, STARTED], ) # 1 job should be locked From 0f2c5c1356999af8bb1e1d06738b871e09303f18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Fri, 2 Jan 2026 12:18:38 +0100 Subject: [PATCH 15/43] [IMP] queue_job: refactor job acquisition In this commit we cleanly separate the job acquisition (i.e. verifying the job is in the exepected state, marking it started and locking it) from job execution. We also avoid trying to start the job if it is already locked by using SKIP LOCKED and exiting early. Indeed in such situations the job is likely already being handled by another worker so there is no point trying to start it, so we exit early and let it be handled either by the other worker or the dead job requeuer. --- queue_job/controllers/main.py | 58 +++++++++++++------ queue_job/job.py | 19 +++--- test_queue_job/tests/__init__.py | 1 + test_queue_job/tests/common.py | 10 ++++ test_queue_job/tests/test_acquire_job.py | 49 ++++++++++++++++ test_queue_job/tests/test_requeue_dead_job.py | 17 ------ 6 files changed, 106 insertions(+), 48 deletions(-) create mode 100644 test_queue_job/tests/test_acquire_job.py diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 4970114181..c365425e51 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -7,6 +7,7 @@ import time import traceback from io import StringIO +from typing import Union from psycopg2 import OperationalError, errorcodes from werkzeug.exceptions import BadRequest, Forbidden @@ -27,15 +28,47 @@ class RunJobController(http.Controller): - def _try_perform_job(self, env, job): - """Try to perform the job.""" + @classmethod + def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[Job, None]: + """Acquire a job for execution. + + - make sure it is in ENQUEUED state + - 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. + """ + env.cr.execute( + "SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s " + "FOR UPDATE SKIP LOCKED", + (job_uuid, ENQUEUED), + ) + if not env.cr.fetchone(): + _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", + job_uuid, + ENQUEUED, + ) + return None + job = Job.load(env, job_uuid) + assert job and job.state == ENQUEUED job.set_started() job.store() env.cr.commit() - job.lock() + if not job.lock(): + _logger.warning( + "was requested to run job %s, but it could not be locked", + job_uuid, + ) + return None + return job + def _try_perform_job(self, env, job): + """Try to perform the job, mark it done and commit if successful.""" _logger.debug("%s started", job) - job.perform() # Triggers any stored computed fields before calling 'set_done' # so that will be part of the 'exec_time' @@ -97,23 +130,10 @@ def retry_postpone(job, message, seconds=None): job.store() new_cr.commit() - # ensure the job to run is in the correct state and lock the record - env.cr.execute( - "SELECT state FROM queue_job WHERE uuid=%s AND state=%s FOR UPDATE", - (job_uuid, ENQUEUED), - ) - if not env.cr.fetchone(): - _logger.warning( - "was requested to run job %s, but it does not exist, " - "or is not in state %s", - job_uuid, - ENQUEUED, - ) + job = self._acquire_job(env, job_uuid) + if not job: return "" - job = Job.load(env, job_uuid) - assert job and job.state == ENQUEUED - try: try: self._try_perform_job(env, job) diff --git a/queue_job/job.py b/queue_job/job.py index bb5b8e2cf3..7ce08f6266 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -238,7 +238,7 @@ 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): + def add_lock_record(self) -> None: """ Create row in db to be locked while the job is being performed. """ @@ -258,13 +258,11 @@ def add_lock_record(self): [self.uuid], ) - def lock(self): - """ - Lock row of job that is being performed + def lock(self) -> bool: + """Lock row of job that is being performed. - If a job cannot be locked, - it means that the job wasn't started, - a RetryableJobError is thrown. + 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. """ self.env.cr.execute( """ @@ -282,16 +280,13 @@ def lock(self): uuid = %s AND state = %s ) - FOR UPDATE; + FOR UPDATE SKIP LOCKED; """, [self.uuid, STARTED], ) # 1 job should be locked - if 1 != len(self.env.cr.fetchall()): - raise RetryableJobError( - f"Trying to lock job that wasn't started, uuid: {self.uuid}" - ) + return bool(self.env.cr.fetchall()) @classmethod def _load_from_db_record(cls, job_db_record): diff --git a/test_queue_job/tests/__init__.py b/test_queue_job/tests/__init__.py index 62347148e5..0cfacebdf3 100644 --- a/test_queue_job/tests/__init__.py +++ b/test_queue_job/tests/__init__.py @@ -1,3 +1,4 @@ +from . import test_acquire_job from . import test_autovacuum from . import test_delayable from . import test_dependencies diff --git a/test_queue_job/tests/common.py b/test_queue_job/tests/common.py index 63ca8cca48..e33ffcef5e 100644 --- a/test_queue_job/tests/common.py +++ b/test_queue_job/tests/common.py @@ -19,3 +19,13 @@ def _create_job(self): stored = Job.db_record_from_uuid(self.env, test_job.uuid) self.assertEqual(len(stored), 1) return stored + + def _get_demo_job(self, uuid): + # job created during load of demo data + job = self.env["queue.job"].search([("uuid", "=", uuid)], limit=1) + self.assertTrue( + job, + f"Demo data queue job {uuid!r} should be loaded in order " + "to make this test work", + ) + return job diff --git a/test_queue_job/tests/test_acquire_job.py b/test_queue_job/tests/test_acquire_job.py new file mode 100644 index 0000000000..6e036ae747 --- /dev/null +++ b/test_queue_job/tests/test_acquire_job.py @@ -0,0 +1,49 @@ +# Copyright 2026 ACSONE SA/NV +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +import logging +from unittest import mock + +from odoo.tests import tagged + +from odoo.addons.queue_job.controllers.main import RunJobController + +from .common import JobCommonCase + + +@tagged("post_install", "-at_install") +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", + ) + with mock.patch.object( + self.env.cr, "commit", mock.Mock(side_effect=self.env["base"].flush) + ) as mock_commit: + job = RunJobController._acquire_job(self.env, job_uuid="test_enqueued_job") + mock_commit.assert_called_once() + self.assertIsNotNone(job) + 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", + ) + + def test_acquire_started_job(self): + with mock.patch.object( + self.env.cr, "commit", mock.Mock(side_effect=self.env["base"].flush) + ) as mock_commit: + with self.assertLogs(level=logging.WARNING) as logs: + job = RunJobController._acquire_job(self.env, "test_started_job") + mock_commit.assert_not_called() + self.assertIsNone(job) + self.assertIn( + "was requested to run job test_started_job, but it does not exist", + logs.output[0], + ) diff --git a/test_queue_job/tests/test_requeue_dead_job.py b/test_queue_job/tests/test_requeue_dead_job.py index 58890adf24..510276be63 100644 --- a/test_queue_job/tests/test_requeue_dead_job.py +++ b/test_queue_job/tests/test_requeue_dead_job.py @@ -13,23 +13,6 @@ @tagged("post_install", "-at_install") class TestRequeueDeadJob(JobCommonCase): - def _get_demo_job(self, uuid): - # job created during load of demo data - job = self.env["queue.job"].search( - [ - ("uuid", "=", uuid), - ], - limit=1, - ) - - self.assertTrue( - job, - f"Demo data queue job {uuid} should be loaded in order" - " to make this tests work", - ) - - return job - def get_locks(self, uuid, cr=None): """ Retrieve lock rows From 25060d23b58b9661749ec0194893560e118331a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Sat, 3 Jan 2026 11:26:02 +0100 Subject: [PATCH 16/43] [IMP] queue_job: refactor runjob Extract the logic to run one job out of the /queue_job/runjob route. Towards making this logic reusable in other job executors. --- queue_job/controllers/main.py | 36 +++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index c365425e51..ea768fb088 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -109,17 +109,7 @@ def _enqueue_dependent_jobs(self, env, job): else: break - @http.route( - "/queue_job/runjob", - type="http", - auth="none", - save_session=False, - readonly=False, - ) - def runjob(self, db, job_uuid, **kw): - http.request.session.db = db - env = http.request.env(user=odoo.SUPERUSER_ID) - + def _runjob(self, env: api.Environment, job: Job) -> None: def retry_postpone(job, message, seconds=None): job.env.clear() with odoo.api.Environment.manage(): @@ -130,10 +120,6 @@ def retry_postpone(job, message, seconds=None): job.store() new_cr.commit() - job = self._acquire_job(env, job_uuid) - if not job: - return "" - try: try: self._try_perform_job(env, job) @@ -165,7 +151,6 @@ def retry_postpone(job, message, seconds=None): # 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() @@ -187,8 +172,6 @@ def retry_postpone(job, message, seconds=None): self._enqueue_dependent_jobs(env, job) _logger.debug("%s enqueue depends done", job) - return "" - def _get_failure_values(self, job, traceback_txt, orig_exception): """Collect relevant data from exception.""" exception_name = orig_exception.__class__.__name__ @@ -203,6 +186,23 @@ def _get_failure_values(self, job, traceback_txt, orig_exception): "exc_message": exc_message, } + @http.route( + "/queue_job/runjob", + type="http", + auth="none", + save_session=False, + readonly=False, + ) + def runjob(self, db, job_uuid, **kw): + http.request.session.db = db + env = http.request.env(user=SUPERUSER_ID) + job = self._acquire_job(env, job_uuid) + if not job: + return "" + self._runjob(env, job) + return "" + + # flake8: noqa: C901 @http.route("/queue_job/create_test_job", type="http", auth="user") def create_test_job( self, From 25a15a7e4086e2aa4c7b9bc49711fc9cb82a59b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Sat, 3 Jan 2026 11:39:13 +0100 Subject: [PATCH 17/43] [IMP] queue_job: convert job execution logic to class method Towards making this logic reusable. --- queue_job/controllers/main.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index ea768fb088..9884d38571 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -66,7 +66,8 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[Job, None]: return None return job - def _try_perform_job(self, env, job): + @classmethod + def _try_perform_job(cls, env, job): """Try to perform the job, mark it done and commit if successful.""" _logger.debug("%s started", job) job.perform() @@ -79,7 +80,8 @@ def _try_perform_job(self, env, job): env.cr.commit() _logger.debug("%s done", job) - def _enqueue_dependent_jobs(self, env, job): + @classmethod + def _enqueue_dependent_jobs(cls, env, job): tries = 0 while True: try: @@ -109,7 +111,8 @@ def _enqueue_dependent_jobs(self, env, job): else: break - def _runjob(self, env: api.Environment, job: Job) -> None: + @classmethod + def _runjob(cls, env: api.Environment, job: Job) -> None: def retry_postpone(job, message, seconds=None): job.env.clear() with odoo.api.Environment.manage(): @@ -122,7 +125,7 @@ def retry_postpone(job, message, seconds=None): try: try: - self._try_perform_job(env, job) + cls._try_perform_job(env, job) except OperationalError as err: # Automatically retry the typical transaction serialization # errors @@ -161,7 +164,7 @@ def retry_postpone(job, message, seconds=None): with odoo.api.Environment.manage(): with odoo.registry(job.env.cr.dbname).cursor() as new_cr: job.env = job.env(cr=new_cr) - vals = self._get_failure_values(job, traceback_txt, orig_exception) + vals = cls._get_failure_values(job, traceback_txt, orig_exception) job.set_failed(**vals) job.store() new_cr.commit() @@ -169,10 +172,11 @@ def retry_postpone(job, message, seconds=None): raise _logger.debug("%s enqueue depends started", job) - self._enqueue_dependent_jobs(env, job) + cls._enqueue_dependent_jobs(env, job) _logger.debug("%s enqueue depends done", job) - def _get_failure_values(self, job, traceback_txt, orig_exception): + @classmethod + def _get_failure_values(cls, job, traceback_txt, orig_exception): """Collect relevant data from exception.""" exception_name = orig_exception.__class__.__name__ if hasattr(orig_exception, "__module__"): From 9dac8abd4566448036065616fc371d3d2fbd52ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Sun, 4 Jan 2026 13:38:48 +0100 Subject: [PATCH 18/43] [IMP] queue_job: take weaker locks Since we are not going to delete records nor modify foreign keys, we can take a weaker lock. --- queue_job/controllers/main.py | 2 +- queue_job/job.py | 2 +- queue_job/jobrunner/runner.py | 2 +- test_queue_job/tests/test_requeue_dead_job.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 9884d38571..5f4fca95d5 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -42,7 +42,7 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[Job, None]: """ env.cr.execute( "SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s " - "FOR UPDATE SKIP LOCKED", + "FOR NO KEY UPDATE SKIP LOCKED", (job_uuid, ENQUEUED), ) if not env.cr.fetchone(): diff --git a/queue_job/job.py b/queue_job/job.py index 7ce08f6266..7071eb7183 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -280,7 +280,7 @@ def lock(self) -> bool: uuid = %s AND state = %s ) - FOR UPDATE SKIP LOCKED; + FOR NO KEY UPDATE SKIP LOCKED; """, [self.uuid, STARTED], ) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index e190c03883..bf09bfc863 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -369,7 +369,7 @@ def _query_requeue_dead_jobs(self): queue_job_lock WHERE queue_job_lock.queue_job_id = queue_job.id - FOR UPDATE SKIP LOCKED + FOR NO KEY UPDATE SKIP LOCKED ) OR NOT EXISTS ( SELECT diff --git a/test_queue_job/tests/test_requeue_dead_job.py b/test_queue_job/tests/test_requeue_dead_job.py index 510276be63..a267c43c87 100644 --- a/test_queue_job/tests/test_requeue_dead_job.py +++ b/test_queue_job/tests/test_requeue_dead_job.py @@ -35,7 +35,7 @@ def get_locks(self, uuid, cr=None): WHERE uuid = %s ) - FOR UPDATE SKIP LOCKED + FOR NO KEY UPDATE SKIP LOCKED """, [uuid], ) From bdb284c3564afa4ac16b36b880711f9112280c33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Bidoul?= Date: Wed, 21 Jan 2026 13:19:50 +0100 Subject: [PATCH 19/43] [IMP] queue_job: prevent commit during queue job execution This would release the job lock, causing spurious restarts by the dead jobs requeuer. --- queue_job/controllers/main.py | 56 +++++++++++++++++++++++++++++------ queue_job/models/queue_job.py | 4 ++- 2 files changed, 50 insertions(+), 10 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 5f4fca95d5..ebf60ed874 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -6,6 +6,7 @@ import random import time import traceback +from contextlib import contextmanager from io import StringIO from typing import Union @@ -27,6 +28,29 @@ DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5 +@contextmanager +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 + it to be started again by the dead jobs requeuer. + """ + + def forbidden_commit(*args, **kwargs): + raise RuntimeError( + "Commit is forbidden in queue jobs. " + "If the current job is a cron running as queue job, " + "modify it to run as a normal cron." + ) + + original_commit = cr.commit + cr.commit = forbidden_commit + try: + yield + finally: + cr.commit = original_commit + + class RunJobController(http.Controller): @classmethod def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[Job, None]: @@ -70,13 +94,16 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[Job, None]: def _try_perform_job(cls, env, job): """Try to perform the job, mark it done and commit if successful.""" _logger.debug("%s started", job) - job.perform() - # Triggers any stored computed fields before calling 'set_done' - # so that will be part of the 'exec_time' - env["base"].flush() - job.set_done() - job.store() - env["base"].flush() + # 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["base"].flush() + job.set_done() + job.store() + env["base"].flush() env.cr.commit() _logger.debug("%s done", job) @@ -216,6 +243,7 @@ def create_test_job( description="Test job", size=1, failure_rate=0, + commit_within_job=False, ): """Create test jobs @@ -264,6 +292,7 @@ def create_test_job( channel=channel, description=description, failure_rate=failure_rate, + commit_within_job=commit_within_job, ) if size > 1: @@ -274,6 +303,7 @@ def create_test_job( channel=channel, description=description, failure_rate=failure_rate, + commit_within_job=commit_within_job, ) return "" @@ -285,6 +315,7 @@ def _create_single_test_job( description="Test job", size=1, failure_rate=0, + commit_within_job=False, ): delayed = ( http.request.env["queue.job"] @@ -294,7 +325,10 @@ def _create_single_test_job( channel=channel, description=description, ) - ._test_job(failure_rate=failure_rate) + ._test_job( + failure_rate=failure_rate, + commit_within_job=commit_within_job, + ) ) return "job uuid: %s" % (delayed.db_record().uuid,) @@ -309,6 +343,7 @@ def _create_graph_test_jobs( channel=None, description="Test job", failure_rate=0, + commit_within_job=False, ): model = http.request.env["queue.job"] current_count = 0 @@ -331,7 +366,10 @@ def _create_graph_test_jobs( max_retries=max_retries, channel=channel, description="%s #%d" % (description, current_count), - )._test_job(failure_rate=failure_rate) + )._test_job( + failure_rate=failure_rate, + commit_within_job=commit_within_job, + ) ) grouping = random.choice(possible_grouping_methods) diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index f5c81bbec6..8f4730d080 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -454,7 +454,9 @@ def related_action_open_record(self): ) return action - def _test_job(self, failure_rate=0): + def _test_job(self, failure_rate=0, commit_within_job=False): _logger.info("Running test job.") if random.random() <= failure_rate: raise JobError("Job failed") + if commit_within_job: + self.env.cr.commit() # pylint: disable=invalid-commit From 2f7d8885caf2644bf655ba1903b95fdd3cca3d34 Mon Sep 17 00:00:00 2001 From: Alexandre Fayolle Date: Wed, 18 Feb 2026 22:16:33 +0100 Subject: [PATCH 20/43] [FIX] mock.patch(auto_spec=True) does not exist There is no parameter auto_spec to mock.patch. This is likely a typo. Running the tests with Python 3.12 give a RuntimeError ``` File "/usr/local/lib/python3.12/unittest/mock.py", line 1311, in __init__ _check_spec_arg_typos(kwargs) File "/usr/local/lib/python3.12/unittest/mock.py", line 1287, in _check_spec_arg_typos raise RuntimeError( RuntimeError: 'auto_spec' might be a typo. ``` --- queue_job/tests/common.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/queue_job/tests/common.py b/queue_job/tests/common.py index 76c03218ba..40d952b665 100644 --- a/queue_job/tests/common.py +++ b/queue_job/tests/common.py @@ -95,7 +95,7 @@ def button_that_uses_delayable_chain(self): with mock.patch( "odoo.addons.queue_job.delay.Job", name="Job Class", - auto_spec=True, + unsafe=True, ) as job_cls_mock: with JobsTrap(job_cls_mock) as trap: yield trap From b16344173633e43b4279786af8a59f5fe7ed3658 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Thu, 19 Feb 2026 16:31:17 +0100 Subject: [PATCH 21/43] Skip check of dependencies when a done job has no dependents Every time a job is done, even if it is not part of a graph, it runs a query to look for dependents to enqueue. Storing the dependent uuids in the "dependencies" field was on purpose to know that we have no further jobs in the graph and that we can skip the check entirely and have no overhead in this case. It looks like an oversight, we can add the missing condition. --- queue_job/controllers/main.py | 7 +++++-- queue_job/job.py | 3 +++ test_queue_job/tests/test_dependencies.py | 13 +++++++++++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index ebf60ed874..94fc4159a6 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -109,6 +109,10 @@ def _try_perform_job(cls, env, job): @classmethod def _enqueue_dependent_jobs(cls, env, job): + if not job.should_check_dependents(): + return + + _logger.debug("%s enqueue depends started", job) tries = 0 while True: try: @@ -137,6 +141,7 @@ def _enqueue_dependent_jobs(cls, env, job): time.sleep(wait_time) else: break + _logger.debug("%s enqueue depends done", job) @classmethod def _runjob(cls, env: api.Environment, job: Job) -> None: @@ -198,9 +203,7 @@ def retry_postpone(job, message, seconds=None): buff.close() raise - _logger.debug("%s enqueue depends started", job) cls._enqueue_dependent_jobs(env, job) - _logger.debug("%s enqueue depends done", job) @classmethod def _get_failure_values(cls, job, traceback_txt, orig_exception): diff --git a/queue_job/job.py b/queue_job/job.py index 7071eb7183..f1cd9256e6 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -622,6 +622,9 @@ def _get_common_dependent_jobs_query(self): AND state = %s; """ + def should_check_dependents(self): + return any(self.__reverse_depends_on_uuids) + def enqueue_waiting(self): sql = self._get_common_dependent_jobs_query() self.env.cr.execute(sql, (PENDING, self.uuid, DONE, WAIT_DEPENDENCIES)) diff --git a/test_queue_job/tests/test_dependencies.py b/test_queue_job/tests/test_dependencies.py index 09a96e3864..c6ac50509b 100644 --- a/test_queue_job/tests/test_dependencies.py +++ b/test_queue_job/tests/test_dependencies.py @@ -287,3 +287,16 @@ def test_depends_graph_uuid_group(self): self.assertTrue(jobs[0].graph_uuid) self.assertTrue(jobs[1].graph_uuid) self.assertEqual(jobs[0].graph_uuid, jobs[1].graph_uuid) + + def test_should_check_dependents(self): + job_root = Job(self.method) + job_a = Job(self.method) + job_a.add_depends({job_root}) + + DelayableGraph._ensure_same_graph_uuid([job_root, job_a]) + + job_root.store() + job_a.store() + + self.assertTrue(job_root.should_check_dependents()) + self.assertFalse(job_a.should_check_dependents()) From e7695d5f281b4e214a570d88625e2b1ac6208bcb Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Thu, 19 Feb 2026 17:02:42 +0100 Subject: [PATCH 22/43] Fix dependents error after retryable job error --- queue_job/controllers/main.py | 1 + 1 file changed, 1 insertion(+) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 94fc4159a6..d9448c0e92 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -186,6 +186,7 @@ def retry_postpone(job, message, seconds=None): # 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() From 7fc4bf818fd89c96a3bd52568978955b4546e824 Mon Sep 17 00:00:00 2001 From: Andrii9090-tecnativa Date: Wed, 4 Mar 2026 11:45:50 +0100 Subject: [PATCH 23/43] [FIX] queue_job: Fix TestJson In this case, when a module adds a value in context, the tests fail --- queue_job/tests/test_json_field.py | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/queue_job/tests/test_json_field.py b/queue_job/tests/test_json_field.py index 802911c9eb..7d2533df93 100644 --- a/queue_job/tests/test_json_field.py +++ b/queue_job/tests/test_json_field.py @@ -20,16 +20,20 @@ def test_encoder_recordset(self): partner = self.env(user=demo_user, context=context).ref("base.main_partner") value = partner value_json = json.dumps(value, cls=JobEncoder) + expected_context = context.copy() expected = { "uid": demo_user.id, "_type": "odoo_recordset", "model": "res.partner", "ids": [partner.id], "su": False, - # no allowed context by default, must be changed in 16.0 - "context": {}, } - self.assertEqual(json.loads(value_json), expected) + result_dict = json.loads(value_json) + result_context = result_dict.pop("context") + self.assertEqual(result_dict, expected) + # context is tested separately as the order/amount of keys is not guaranteed + for key in result_context: + self.assertEqual(result_context[key], expected_context[key]) def test_encoder_recordset_list(self): demo_user = self.env.ref("base.user_demo") @@ -50,7 +54,20 @@ def test_encoder_recordset_list(self): "context": {}, }, ] - self.assertEqual(json.loads(value_json), expected) + result_dict = json.loads(value_json) + for result_value, expected_value in zip(result_dict, expected): + if isinstance(expected_value, dict): + for key in result_value: + if key == "context": + for context_key in result_value["context"]: + self.assertEqual( + result_value["context"][context_key], + expected_value["context"][context_key], + ) + else: + self.assertEqual(result_value[key], expected_value[key]) + else: + self.assertEqual(result_value, expected_value) def test_decoder_recordset(self): demo_user = self.env.ref("base.user_demo") From ebefc01f541b144e0ce339623d2e531dcffae4ee Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Thu, 19 Feb 2026 15:55:48 +0100 Subject: [PATCH 24/43] Add 'Allow Commit' option on job functions It is forbidden to commit inside a job, because it releases the job lock and can cause it to start again, while still being run, by the dead jobs requeuer. For some use cases, it may actually be legitimate, or at least be needed in the short term before actual updates in the code. A new option on the job function, false by default, allow to run the job in a new transaction, at the cost of an additional connection + transaction overhead. Related to #889 --- queue_job/controllers/main.py | 5 +-- queue_job/job.py | 35 +++++++++++++++++--- queue_job/models/queue_job_function.py | 10 +++++- queue_job/tests/common.py | 2 +- queue_job/tests/test_model_job_function.py | 2 ++ queue_job/views/queue_job_function_views.xml | 2 ++ 6 files changed, 47 insertions(+), 9 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index d9448c0e92..28d8eda95c 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -39,8 +39,9 @@ def _prevent_commit(cr): def forbidden_commit(*args, **kwargs): raise RuntimeError( "Commit is forbidden in queue jobs. " - "If the current job is a cron running as queue job, " - "modify it to run as a normal cron." + 'You may want to enable the "Allow Commit" option on the Job ' + "Function. Alternatively, if the current job is a cron running as " + "queue job, you can modify it to run as a normal cron." ) original_commit = cr.commit diff --git a/queue_job/job.py b/queue_job/job.py index f1cd9256e6..64d3de0baa 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -16,6 +16,16 @@ from .exception import FailedJobError, NoSuchJobError, RetryableJobError +try: + from contextlib import contextmanager, nullcontext +except ImportError: # Python 3.6 + from contextlib import contextmanager + + @contextmanager + def nullcontext(): + yield + + WAIT_DEPENDENCIES = "wait_dependencies" PENDING = "pending" ENQUEUED = "enqueued" @@ -491,10 +501,6 @@ def __init__( self.method_name = func.__name__ self.recordset = recordset - self.env = env - self.job_model = self.env["queue.job"] - self.job_model_name = "queue.job" - self.job_config = ( self.env["queue.job.function"].sudo().job_config(self.job_function_name) ) @@ -572,7 +578,12 @@ def perform(self): """ self.retry += 1 try: - self.result = self.func(*tuple(self.args), **self.kwargs) + if self.job_config.allow_commit: + env_context_manager = self._with_temporary_env() + else: + env_context_manager = nullcontext() + with env_context_manager: + self.result = self.func(*tuple(self.args), **self.kwargs) except RetryableJobError as err: if err.ignore_retry: self.retry -= 1 @@ -592,6 +603,16 @@ def perform(self): return self.result + @contextmanager + def _with_temporary_env(self): + with self.env.registry.cursor() as new_cr: + env = self.recordset.env + self.recordset = self.recordset.with_env(env(cr=new_cr)) + try: + yield + finally: + self.recordset = self.recordset.with_env(env) + def _get_common_dependent_jobs_query(self): return """ UPDATE queue_job @@ -763,6 +784,10 @@ def __lt__(self, other): def db_record(self): return self.db_records_from_uuids(self.env, [self.uuid]) + @property + def env(self): + return self.recordset.env + @property def func(self): recordset = self.recordset.with_context(job_uuid=self.uuid) diff --git a/queue_job/models/queue_job_function.py b/queue_job/models/queue_job_function.py index d839708dcf..034e499fb8 100644 --- a/queue_job/models/queue_job_function.py +++ b/queue_job/models/queue_job_function.py @@ -28,7 +28,8 @@ class QueueJobFunction(models.Model): "related_action_enable " "related_action_func_name " "related_action_kwargs " - "job_function_id ", + "job_function_id " + "allow_commit", ) def _default_channel(self): @@ -79,6 +80,11 @@ def _default_channel(self): "enable, func_name, kwargs.\n" "See the module description for details.", ) + allow_commit = fields.Boolean( + help="Allows the job to commit transactions during execution. " + "Under the hood, this executes the job in a new database cursor, " + "which incurs a slight overhead.", + ) @api.depends("model_id.model", "method") def _compute_name(self): @@ -143,6 +149,7 @@ def job_default_config(self): related_action_func_name=None, related_action_kwargs={}, job_function_id=None, + allow_commit=False, ) def _parse_retry_pattern(self): @@ -178,6 +185,7 @@ def job_config(self, name): related_action_func_name=config.related_action.get("func_name"), related_action_kwargs=config.related_action.get("kwargs", {}), job_function_id=config.id, + allow_commit=config.allow_commit, ) def _retry_pattern_format_error_message(self): diff --git a/queue_job/tests/common.py b/queue_job/tests/common.py index 40d952b665..2287fb3f45 100644 --- a/queue_job/tests/common.py +++ b/queue_job/tests/common.py @@ -277,7 +277,7 @@ def _add_job(self, *args, **kwargs): def _prepare_context(self, job): # pylint: disable=context-overridden - job_model = job.job_model.with_context({}) + job_model = job.env["queue.job"].with_context({}) field_records = job_model._fields["records"] # Filter the context to simulate store/load of the job job.recordset = field_records.convert_to_write(job.recordset, job_model) diff --git a/queue_job/tests/test_model_job_function.py b/queue_job/tests/test_model_job_function.py index e6ddf3fcc3..3fe67b20ba 100644 --- a/queue_job/tests/test_model_job_function.py +++ b/queue_job/tests/test_model_job_function.py @@ -42,6 +42,7 @@ def test_function_job_config(self): ' "func_name": "related_action_foo",' ' "kwargs": {"b": 1}}' ), + "allow_commit": True, } ) self.assertEqual( @@ -53,5 +54,6 @@ def test_function_job_config(self): related_action_func_name="related_action_foo", related_action_kwargs={"b": 1}, job_function_id=job_function.id, + allow_commit=True, ), ) diff --git a/queue_job/views/queue_job_function_views.xml b/queue_job/views/queue_job_function_views.xml index a6e2ce402c..6c208b2b67 100644 --- a/queue_job/views/queue_job_function_views.xml +++ b/queue_job/views/queue_job_function_views.xml @@ -11,6 +11,7 @@ + @@ -25,6 +26,7 @@ + From 2ddadb66bb19e454c96221af8ed0a2b42cfce363 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Fri, 20 Feb 2026 15:16:52 +0100 Subject: [PATCH 25/43] Improve documentation on allow commit --- queue_job/controllers/main.py | 3 ++- queue_job/models/queue_job_function.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 28d8eda95c..f25497da2a 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -41,7 +41,8 @@ def forbidden_commit(*args, **kwargs): "Commit is forbidden in queue jobs. " 'You may want to enable the "Allow Commit" option on the Job ' "Function. Alternatively, if the current job is a cron running as " - "queue job, you can modify it to run as a normal cron." + "queue job, you can modify it to run as a normal cron. More details on: " + "https://github.com/OCA/queue/wiki/%5BDRAFT%5D-Upgrade-warning:-commits-inside-jobs" ) original_commit = cr.commit diff --git a/queue_job/models/queue_job_function.py b/queue_job/models/queue_job_function.py index 034e499fb8..39726738a8 100644 --- a/queue_job/models/queue_job_function.py +++ b/queue_job/models/queue_job_function.py @@ -83,7 +83,8 @@ def _default_channel(self): allow_commit = fields.Boolean( help="Allows the job to commit transactions during execution. " "Under the hood, this executes the job in a new database cursor, " - "which incurs a slight overhead.", + "which incurs an overhead as it requires an extra connection to " + "the database. " ) @api.depends("model_id.model", "method") From 154049bb7b15c6d501e4b1196cb08daa39ce52af Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Tue, 24 Feb 2026 11:35:38 +0100 Subject: [PATCH 26/43] Fix missing job.env setter As the controller changes env on Job instances. --- queue_job/controllers/main.py | 4 +++- queue_job/job.py | 11 +++++++---- queue_job/tests/test_run_job_controller.py | 6 ++++++ 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index f25497da2a..8f19094c28 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -16,6 +16,7 @@ import odoo from odoo import _, http, tools from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY +from odoo.tools import config from ..delay import chain, group from ..exception import FailedJobError, NothingToDoJob, RetryableJobError @@ -106,7 +107,8 @@ def _try_perform_job(cls, env, job): job.set_done() job.store() env["base"].flush() - env.cr.commit() + if not config["test_enable"]: + env.cr.commit() _logger.debug("%s done", job) @classmethod diff --git a/queue_job/job.py b/queue_job/job.py index 64d3de0baa..a550587b73 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -497,7 +497,6 @@ def __init__( raise TypeError("Job accepts only methods of Models") recordset = func.__self__ - env = recordset.env self.method_name = func.__name__ self.recordset = recordset @@ -550,10 +549,10 @@ def __init__( self.exc_message = None self.exc_info = None - if "company_id" in env.context: - company_id = env.context["company_id"] + if "company_id" in self.env.context: + company_id = self.env.context["company_id"] else: - company_id = env.company.id + company_id = self.env.company.id self.company_id = company_id self._eta = None self.eta = eta @@ -788,6 +787,10 @@ def db_record(self): def env(self): return self.recordset.env + @env.setter + def env(self, env): + self.recordset = self.recordset.with_env(env) + @property def func(self): recordset = self.recordset.with_context(job_uuid=self.uuid) diff --git a/queue_job/tests/test_run_job_controller.py b/queue_job/tests/test_run_job_controller.py index bb63bc82ec..1a15f4363a 100644 --- a/queue_job/tests/test_run_job_controller.py +++ b/queue_job/tests/test_run_job_controller.py @@ -15,3 +15,9 @@ def test_get_failure_values(self): self.assertEqual( rslt, {"exc_info": "info", "exc_name": "Exception", "exc_message": "zero"} ) + + def test_runjob_success(self): + job = self.env["queue.job"].with_delay()._test_job() + RunJobController._runjob(self.env, job) + self.assertEqual(job.state, "done") + self.assertEqual(job.db_record().state, "done") From 69d37e58c04ba85395a580a1251fd61523041f8a Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Tue, 24 Feb 2026 16:00:20 +0100 Subject: [PATCH 27/43] Add failure retry in queue job test job --- queue_job/controllers/main.py | 13 +++++++++++++ queue_job/models/queue_job.py | 18 +++++++++++++++--- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 8f19094c28..e83c85f03d 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -252,6 +252,7 @@ def create_test_job( size=1, failure_rate=0, commit_within_job=False, + failure_retry_seconds=0, ): """Create test jobs @@ -293,6 +294,12 @@ def create_test_job( except ValueError: max_retries = None + if failure_retry_seconds is not None: + try: + failure_retry_seconds = int(failure_retry_seconds) + except ValueError: + failure_retry_seconds = 0 + if size == 1: return self._create_single_test_job( priority=priority, @@ -301,6 +308,7 @@ def create_test_job( description=description, failure_rate=failure_rate, commit_within_job=commit_within_job, + failure_retry_seconds=failure_retry_seconds, ) if size > 1: @@ -312,6 +320,7 @@ def create_test_job( description=description, failure_rate=failure_rate, commit_within_job=commit_within_job, + failure_retry_seconds=failure_retry_seconds, ) return "" @@ -324,6 +333,7 @@ def _create_single_test_job( size=1, failure_rate=0, commit_within_job=False, + failure_retry_seconds=0, ): delayed = ( http.request.env["queue.job"] @@ -336,6 +346,7 @@ def _create_single_test_job( ._test_job( failure_rate=failure_rate, commit_within_job=commit_within_job, + failure_retry_seconds=failure_retry_seconds, ) ) return "job uuid: %s" % (delayed.db_record().uuid,) @@ -352,6 +363,7 @@ def _create_graph_test_jobs( description="Test job", failure_rate=0, commit_within_job=False, + failure_retry_seconds=0, ): model = http.request.env["queue.job"] current_count = 0 @@ -377,6 +389,7 @@ def _create_graph_test_jobs( )._test_job( failure_rate=failure_rate, commit_within_job=commit_within_job, + failure_retry_seconds=failure_retry_seconds, ) ) diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 8f4730d080..17d760dac4 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -11,7 +11,7 @@ from odoo.addons.base_sparse_field.models.fields import Serialized from ..delay import Graph -from ..exception import JobError +from ..exception import JobError, RetryableJobError from ..fields import JobSerialized from ..job import ( CANCELLED, @@ -454,9 +454,21 @@ def related_action_open_record(self): ) return action - def _test_job(self, failure_rate=0, commit_within_job=False): + def _test_job( + self, + failure_rate=0, + commit_within_job=False, + failure_retry_seconds=0, + ): _logger.info("Running test job.") if random.random() <= failure_rate: - raise JobError("Job failed") + if failure_retry_seconds: + raise RetryableJobError( + f"Retryable job failed, will be retried in " + f"{failure_retry_seconds} seconds", + seconds=failure_retry_seconds, + ) + else: + raise JobError("Job failed") if commit_within_job: self.env.cr.commit() # pylint: disable=invalid-commit From 5b65e792fbcc1942da84a30a206e996245cb75ae Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Tue, 24 Feb 2026 16:00:50 +0100 Subject: [PATCH 28/43] Simplify job env management --- queue_job/controllers/main.py | 27 ++++++++++----------------- queue_job/job.py | 14 +++++++------- 2 files changed, 17 insertions(+), 24 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index e83c85f03d..78c8997ee5 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -13,8 +13,7 @@ from psycopg2 import OperationalError, errorcodes from werkzeug.exceptions import BadRequest, Forbidden -import odoo -from odoo import _, http, tools +from odoo import SUPERUSER_ID, _, api, http, tools from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY from odoo.tools import config @@ -151,13 +150,10 @@ def _enqueue_dependent_jobs(cls, env, job): def _runjob(cls, env: api.Environment, job: Job) -> None: def retry_postpone(job, message, seconds=None): job.env.clear() - with odoo.api.Environment.manage(): - with odoo.registry(job.env.cr.dbname).cursor() as new_cr: - job.env = job.env(cr=new_cr) - job.postpone(result=message, seconds=seconds) - job.set_pending(reset_retry=False) - job.store() - new_cr.commit() + with job.in_temporary_env(): + job.postpone(result=message, seconds=seconds) + job.set_pending(reset_retry=False) + job.store() try: try: @@ -198,14 +194,11 @@ def retry_postpone(job, message, seconds=None): traceback_txt = buff.getvalue() _logger.error(traceback_txt) job.env.clear() - with odoo.api.Environment.manage(): - with odoo.registry(job.env.cr.dbname).cursor() as new_cr: - job.env = job.env(cr=new_cr) - vals = cls._get_failure_values(job, traceback_txt, orig_exception) - job.set_failed(**vals) - job.store() - new_cr.commit() - buff.close() + with job.in_temporary_env(): + vals = cls._get_failure_values(job, traceback_txt, orig_exception) + job.set_failed(**vals) + job.store() + buff.close() raise cls._enqueue_dependent_jobs(env, job) diff --git a/queue_job/job.py b/queue_job/job.py index a550587b73..6beae415d1 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -578,7 +578,7 @@ def perform(self): self.retry += 1 try: if self.job_config.allow_commit: - env_context_manager = self._with_temporary_env() + env_context_manager = self.in_temporary_env() else: env_context_manager = nullcontext() with env_context_manager: @@ -603,14 +603,14 @@ def perform(self): return self.result @contextmanager - def _with_temporary_env(self): + def in_temporary_env(self): with self.env.registry.cursor() as new_cr: - env = self.recordset.env - self.recordset = self.recordset.with_env(env(cr=new_cr)) + env = self.env + self._env = env(cr=new_cr) try: yield finally: - self.recordset = self.recordset.with_env(env) + self._env = env def _get_common_dependent_jobs_query(self): return """ @@ -788,7 +788,7 @@ def env(self): return self.recordset.env @env.setter - def env(self, env): + def _env(self, env): self.recordset = self.recordset.with_env(env) @property @@ -855,7 +855,7 @@ def model_name(self): @property def user_id(self): - return self.recordset.env.uid + return self.env.uid @property def eta(self): From 78ecac0c471757c0c76139be7b555f220aa4a4e3 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Fri, 27 Feb 2026 08:56:23 +0100 Subject: [PATCH 29/43] Update upgrade warning link --- queue_job/controllers/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 78c8997ee5..ff8c46b72a 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -42,7 +42,7 @@ def forbidden_commit(*args, **kwargs): 'You may want to enable the "Allow Commit" option on the Job ' "Function. Alternatively, if the current job is a cron running as " "queue job, you can modify it to run as a normal cron. More details on: " - "https://github.com/OCA/queue/wiki/%5BDRAFT%5D-Upgrade-warning:-commits-inside-jobs" + "https://github.com/OCA/queue/wiki/Upgrade-warning:-commits-inside-jobs" ) original_commit = cr.commit From 700ec6105eb9e0481a4523d599c2970eefbd40ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Todorovich?= Date: Fri, 6 Jun 2025 11:37:01 -0300 Subject: [PATCH 30/43] [FIX] queue_job: cancel job waiting dependencies --- queue_job/models/queue_job.py | 4 ++-- queue_job/tests/test_wizards.py | 11 +++-------- queue_job/views/queue_job_views.xml | 2 +- queue_job/wizards/queue_jobs_to_cancelled.py | 2 +- 4 files changed, 7 insertions(+), 12 deletions(-) diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 17d760dac4..d6b2303a5a 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -336,8 +336,8 @@ def button_done(self): return True def button_cancelled(self): - # If job was set to DONE or WAIT_DEPENDENCIES, do not cancel it - states_from = (PENDING, ENQUEUED, FAILED) + # If job was set to DONE do not cancel it + states_from = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, FAILED) result = _("Cancelled by %s") % self.env.user.name records = self.filtered(lambda job_: job_.state in states_from) records._change_job_state(CANCELLED, result=result) diff --git a/queue_job/tests/test_wizards.py b/queue_job/tests/test_wizards.py index 7738836d2f..54718f658a 100644 --- a/queue_job/tests/test_wizards.py +++ b/queue_job/tests/test_wizards.py @@ -69,18 +69,13 @@ def test_04_requeue_forbidden(self): def test_05_cancel_forbidden(self): wizard = self._wizard("queue.jobs.to.cancelled") - # State WAIT_DEPENDENCIES is not cancelled - self.job.state = "wait_dependencies" - wizard.set_cancelled() - self.assertEqual(self.job.state, "wait_dependencies") - # State DONE is not cancelled self.job.state = "done" wizard.set_cancelled() self.assertEqual(self.job.state, "done") - # State PENDING, ENQUEUED or FAILED will be cancelled - for test_state in ("pending", "enqueued"): + # State PENDING, ENQUEUED, WAIT_DEPENDENCIES or FAILED will be cancelled + for test_state in ("pending", "enqueued", "wait_dependencies", "failed"): self.job.state = test_state wizard.set_cancelled() self.assertEqual(self.job.state, "cancelled") @@ -99,7 +94,7 @@ def test_06_done_forbidden(self): self.assertEqual(self.job.state, "cancelled") # State WAIT_DEPENDENCIES, PENDING, ENQUEUED or FAILED will be set to DONE - for test_state in ("wait_dependencies", "pending", "enqueued"): + for test_state in ("wait_dependencies", "pending", "enqueued", "failed"): self.job.state = test_state wizard.set_done() self.assertEqual(self.job.state, "done") diff --git a/queue_job/views/queue_job_views.xml b/queue_job/views/queue_job_views.xml index c0388c7012..abd52dc20b 100644 --- a/queue_job/views/queue_job_views.xml +++ b/queue_job/views/queue_job_views.xml @@ -25,7 +25,7 @@ />