diff --git a/queue_job/README.rst b/queue_job/README.rst index 6d5101bed4..2d65273b73 100644 --- a/queue_job/README.rst +++ b/queue_job/README.rst @@ -278,8 +278,8 @@ one is done: ) -Enqueing Job Options --------------------- +Enqueuing Job Options +--------------------- * priority: default is 10, the closest it is to 0, the faster it will be executed diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index c4f7028a74..d8399f0a36 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -19,7 +19,7 @@ from ..delay import chain, group from ..exception import FailedJobError, NothingToDoJob, RetryableJobError -from ..job import ENQUEUED, Job +from ..job import ENQUEUED, STARTED, Job _logger = logging.getLogger(__name__) @@ -32,7 +32,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. """ @@ -62,16 +62,12 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[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", @@ -79,12 +75,10 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[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, @@ -94,15 +88,12 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Union[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 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() @@ -148,13 +139,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) @@ -180,26 +164,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 86f43981a0..4bf68acfb2 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -8,6 +8,7 @@ import sys import uuid import weakref +from contextlib import contextmanager from datetime import datetime, timedelta from functools import total_ordering from random import randint @@ -17,15 +18,6 @@ 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 - def _rebind_to_cr(value, cr): """Rebind any BaseModel inside ``value`` to the cursor ``cr``. @@ -247,17 +239,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: - raise NoSuchJobError( - "Job %s does no longer exist in the storage." % job_uuid - ) - return cls._load_from_db_record(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.") @classmethod def load_many(cls, env, job_uuids): @@ -268,55 +259,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): @@ -606,7 +565,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: @@ -955,7 +919,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 bf09bfc863..5717c7b104 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -128,7 +128,7 @@ import selectors import threading import time -from contextlib import closing, contextmanager +from contextlib import contextmanager import psycopg2 import requests @@ -249,14 +249,14 @@ def close(self): def _acquire_master_lock(self): """Acquire the master runner lock or raise MasterElectionLost""" - with closing(self.conn.cursor()) as cr: + with 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: + with self.conn.cursor() as cr: cr.execute( "SELECT 1 FROM pg_tables WHERE tablename=%s", ("ir_module_module",) ) @@ -285,7 +285,7 @@ def _has_queue_job(self): return True def _initialize(self): - with closing(self.conn.cursor()) as cr: + with self.conn.cursor() as cr: cr.execute("LISTEN queue_job") @contextmanager @@ -299,17 +299,17 @@ def select_jobs(self, where, args): "priority, EXTRACT(EPOCH FROM eta), state " "FROM queue_job WHERE %s" % (where,) ) - with closing(self.conn.cursor("select_jobs", withhold=True)) as cr: + with self.conn.cursor("select_jobs", withhold=True) as cr: cr.execute(query, args) yield cr def keep_alive(self): query = "SELECT 1" - with closing(self.conn.cursor()) as cr: + with self.conn.cursor() as cr: cr.execute(query) def set_job_enqueued(self, uuid): - with closing(self.conn.cursor()) as cr: + with self.conn.cursor() as cr: cr.execute( "UPDATE queue_job SET state=%s, " "date_enqueued=date_trunc('seconds', " @@ -319,69 +319,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): """ @@ -409,7 +372,7 @@ def requeue_dead_jobs(self): the controller receiving the request). """ - with closing(self.conn.cursor()) as cr: + with self.conn.cursor() as cr: query = self._query_requeue_dead_jobs() cr.execute(query) 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_function.py b/queue_job/models/queue_job_function.py index af9b1f6e5b..f139303dbc 100644 --- a/queue_job/models/queue_job_function.py +++ b/queue_job/models/queue_job_function.py @@ -89,7 +89,9 @@ def _default_channel(self): help="Allows the job to commit transactions during execution. " "Under the hood, this executes the job in a new database cursor, " "which incurs an overhead as it requires an extra connection to " - "the database. " + "the database.\n" + "Important: the new cursor cannot alter the queue.job record, " + "because the runner keeps an exclusive lock on it." ) @api.depends("model_id.model", "method") 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/readme/USAGE.rst b/queue_job/readme/USAGE.rst index b7614d2ab4..d535653a67 100644 --- a/queue_job/readme/USAGE.rst +++ b/queue_job/readme/USAGE.rst @@ -137,8 +137,8 @@ one is done: ) -Enqueing Job Options --------------------- +Enqueuing Job Options +--------------------- * priority: default is 10, the closest it is to 0, the faster it will be executed 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/queue_job/static/description/index.html b/queue_job/static/description/index.html index 4ca6fdb01d..45167586ed 100644 --- a/queue_job/static/description/index.html +++ b/queue_job/static/description/index.html @@ -422,7 +422,7 @@

Job Queue

  • Usage