diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 18e257f1c..15d728b0e 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -4,208 +4,25 @@ import logging import random -import time -import traceback -from contextlib import contextmanager -from io import StringIO -from psycopg2 import OperationalError, errorcodes from werkzeug.exceptions import BadRequest, Forbidden -from odoo import SUPERUSER_ID, _, api, http -from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY -from odoo.tools import config +from odoo import SUPERUSER_ID, _, http from ..delay import chain, group -from ..exception import FailedJobError, RetryableJobError -from ..job import ENQUEUED, Job -_logger = logging.getLogger(__name__) - -PG_RETRY = 5 # seconds - -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. " - '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/Upgrade-warning:-commits-inside-jobs" - ) +# unused imports are kept for backward compatibility +from ..executor import ( + DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE, # noqa: F401 + PG_RETRY, # noqa: F401 + JobExecutor, + _prevent_commit, # noqa: F401 +) - original_commit = cr.commit - cr.commit = forbidden_commit - try: - yield - finally: - cr.commit = original_commit +_logger = logging.getLogger(__name__) class RunJobController(http.Controller): - @classmethod - def _acquire_job(cls, env: api.Environment, job_uuid: str) -> 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 NO KEY 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() - if not job.lock(): - _logger.warning( - "was requested to run job %s, but it could not be locked", - job_uuid, - ) - return None - return 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) - # TODO refactor, the relation between env and job.env is not clear - assert env.cr is job.env.cr - with _prevent_commit(env.cr): - job.perform() - # Triggers any stored computed fields before calling 'set_done' - # so that will be part of the 'exec_time' - env.flush_all() - job.set_done() - job.store() - env.flush_all() - if not config["test_enable"]: - env.cr.commit() - _logger.debug("%s done", 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: - 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.error( - "%s, maximum number of tries reached to update dependencies", - errorcodes.lookup(err.pgcode), - ) - raise - wait_time = random.uniform(0.0, 2**tries) - tries += 1 - _logger.info( - "%s, retry %d/%d in %.04f sec...", - errorcodes.lookup(err.pgcode), - tries, - DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE, - wait_time, - ) - time.sleep(wait_time) - else: - break - _logger.debug("%s enqueue depends done", 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) - except OperationalError as err: - # Automatically retry the typical transaction serialization - # errors - if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY: - raise - - _logger.debug("%s OperationalError, postponed", job) - raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err - - except RetryableJobError as err: - # delay the job later, requeue - retry_postpone(job, str(err), seconds=err.seconds) - _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() - _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() - buff.close() - raise - - cls._enqueue_dependent_jobs(env, job) - - @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__"): - exception_name = orig_exception.__module__ + "." + exception_name - exc_message = ( - orig_exception.args[0] if orig_exception.args else str(orig_exception) - ) - return { - "exc_info": traceback_txt, - "exc_name": exception_name, - "exc_message": exc_message, - } - @http.route( "/queue_job/runjob", type="http", @@ -215,11 +32,13 @@ def _get_failure_values(cls, job, traceback_txt, orig_exception): ) 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) + # update_env (in contrast to a local request.env(user=...)) replaces + # the uid=None environment installed by auth="none" on the request + # itself. On Odoo >= 19 it additionally repoints + # transaction.default_env, which otherwise makes flushes recompute + # stored fields with uid=None (see OCA/queue issue #922) + http.request.update_env(user=SUPERUSER_ID) + JobExecutor(http.request.env, job_uuid).run() return "" # flake8: noqa: C901 diff --git a/queue_job/executor.py b/queue_job/executor.py new file mode 100644 index 000000000..39cebdcc0 --- /dev/null +++ b/queue_job/executor.py @@ -0,0 +1,215 @@ +# Copyright (c) 2015-2016 ACSONE SA/NV () +# Copyright 2013-2016 Camptocamp SA +# Copyright 2026 QoQa Services SA +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) + +import logging +import random +import time +import traceback +from contextlib import contextmanager +from io import StringIO + +from psycopg2 import OperationalError, errorcodes + +from odoo import api +from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY +from odoo.tools import config + +from .exception import FailedJobError, RetryableJobError +from .job import ENQUEUED, Job + +_logger = logging.getLogger(__name__) + +PG_RETRY = 5 # seconds + +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. " + '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/Upgrade-warning:-commits-inside-jobs" + ) + + original_commit = cr.commit + cr.commit = forbidden_commit + try: + yield + finally: + cr.commit = original_commit + + +class JobExecutor: + def __init__(self, env: api.Environment, job_uuid: str): + self.job_uuid = job_uuid + self.env = env + + def run(self): + job = self.acquire() + if not job: + return + self.run_job(job) + + def acquire(self) -> Job | None: + """Acquire the 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. + """ + self.env.cr.execute( + "SELECT uuid FROM queue_job WHERE uuid=%s AND state=%s " + "FOR NO KEY UPDATE SKIP LOCKED", + (self.job_uuid, ENQUEUED), + ) + if not self.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", + self.job_uuid, + ENQUEUED, + ) + return None + # TODO: lazy-load recordset, args, kwargs etc. + job = Job.load(self.env, self.job_uuid) + assert job and job.state == ENQUEUED + job.set_started() + job.store() + self.env.cr.commit() # pylint: disable=invalid-commit + if not job.lock(): + _logger.warning( + "was requested to run job %s, but it could not be locked", + self.job_uuid, + ) + return None + return job + + def run_job(self, job): + 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: + self.try_perform_job(job) + except OperationalError as err: + # Automatically retry the typical transaction serialization + # errors + if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY: + raise + + _logger.debug("%s OperationalError, postponed", job) + raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err + + except RetryableJobError as err: + # delay the job later, requeue + retry_postpone(job, str(err), seconds=err.seconds) + _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 + self.env.cr.rollback() + return + + except (FailedJobError, Exception) as orig_exception: + buff = StringIO() + traceback.print_exc(file=buff) + traceback_txt = buff.getvalue() + _logger.error(traceback_txt) + job.env.clear() + with job.in_temporary_env(): + vals = self._get_failure_values(traceback_txt, orig_exception) + job.set_failed(**vals) + job.store() + buff.close() + raise + + self._enqueue_dependent_jobs(job) + + def try_perform_job(self, job): + """Try to perform the job, mark it done and commit if successful.""" + _logger.debug("%s started", job) + # TODO: clarify which env has "control" over the job state and + # which env "executes" the job method + assert self.env.cr is job.env.cr + with _prevent_commit(self.env.cr): + job.perform() + # Triggers any stored computed fields before calling 'set_done' + # so that will be part of the 'exec_time' + job.env.flush_all() + job.set_done() + job.store() + job.env.flush_all() + if not config["test_enable"]: + self.env.cr.commit() # pylint: disable=invalid-commit + _logger.debug("%s done", job) + + def _enqueue_dependent_jobs(self, job): + """Set the dependent jobs of a done job to pending.""" + if not job.should_check_dependents(): + return + + _logger.debug("%s enqueue depends started", job) + tries = 0 + while True: + try: + 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.error( + "%s, maximum number of tries reached to update dependencies", + errorcodes.lookup(err.pgcode), + ) + raise + wait_time = random.uniform(0.0, 2**tries) + tries += 1 + _logger.info( + "%s, retry %d/%d in %.04f sec...", + errorcodes.lookup(err.pgcode), + tries, + DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE, + wait_time, + ) + time.sleep(wait_time) + else: + break + _logger.debug("%s enqueue depends done", job) + + @staticmethod + def _get_failure_values(traceback_txt, orig_exception): + """Collect relevant data from exception.""" + exception_name = orig_exception.__class__.__name__ + if hasattr(orig_exception, "__module__"): + exception_name = orig_exception.__module__ + "." + exception_name + exc_message = ( + orig_exception.args[0] if orig_exception.args else str(orig_exception) + ) + return { + "exc_info": traceback_txt, + "exc_name": exception_name, + "exc_message": exc_message, + } diff --git a/queue_job/tests/__init__.py b/queue_job/tests/__init__.py index 16bcdff96..b3f92e30e 100644 --- a/queue_job/tests/__init__.py +++ b/queue_job/tests/__init__.py @@ -1,10 +1,12 @@ -from . import test_run_rob_controller -from . import test_runner_channels -from . import test_runner_runner -from . import test_delayable -from . import test_delayable_split -from . import test_json_field -from . import test_model_job_channel -from . import test_model_job_function -from . import test_queue_job_protected_write -from . import test_wizards +from . import ( + test_delayable, + test_delayable_split, + test_executor, + test_json_field, + test_model_job_channel, + test_model_job_function, + test_queue_job_protected_write, + test_runner_channels, + test_runner_runner, + test_wizards, +) diff --git a/queue_job/tests/test_executor.py b/queue_job/tests/test_executor.py new file mode 100644 index 000000000..d28a6d775 --- /dev/null +++ b/queue_job/tests/test_executor.py @@ -0,0 +1,24 @@ +# Copyright 2026 QoQa Services SA (https://www.qoqa.ch) +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) + +from odoo.tests.common import TransactionCase + +from ..executor import JobExecutor +from ..job import Job + + +class TestJobExecutor(TransactionCase): + def test_run_success(self): + job = self.env["queue.job"].with_delay()._test_job() + JobExecutor(self.env, job.uuid).run() + self.assertEqual(job.state, "done") + self.assertEqual(job.db_record().state, "done") + + def test_get_failure_values(self): + method = self.env["res.users"].mapped + job = Job(method) + executor = JobExecutor(self.env, job.uuid) + result = executor._get_failure_values("info", Exception("zero", "one")) + self.assertEqual( + result, {"exc_info": "info", "exc_name": "Exception", "exc_message": "zero"} + )