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/delay.py b/queue_job/delay.py index 0ba54e48a..8d608505a 100644 --- a/queue_job/delay.py +++ b/queue_job/delay.py @@ -7,7 +7,7 @@ import uuid from collections import defaultdict, deque -from .job import Job +from .job import PENDING, WAIT_DEPENDENCIES, Job, JobSpec, JobStore from .utils import must_run_without_delay _logger = logging.getLogger(__name__) @@ -224,55 +224,68 @@ def _has_to_execute_directly(self, vertices): return False @staticmethod - def _ensure_same_graph_uuid(jobs): + def _ensure_same_graph_uuid(vertices): """Set the same graph uuid on all jobs of the same graph""" - jobs_count = len(jobs) - if jobs_count == 0: + count = len(vertices) + if count == 0: raise ValueError("Expecting jobs") - elif jobs_count == 1: - if jobs[0].graph_uuid: + elif count == 1: + if vertices[0]._graph_uuid: raise ValueError( - f"Job {jobs[0]} is a single job, it should not have a graph uuid" + f"Delayable {vertices[0]} is a single job, " + "it should not have a graph uuid" ) else: - graph_uuids = {job.graph_uuid for job in jobs if job.graph_uuid} + graph_uuids = { + vertex._graph_uuid for vertex in vertices if vertex._graph_uuid + } if len(graph_uuids) > 1: raise ValueError("Jobs cannot have dependencies between several graphs") elif len(graph_uuids) == 1: graph_uuid = graph_uuids.pop() else: graph_uuid = str(uuid.uuid4()) - for job in jobs: - job.graph_uuid = graph_uuid + for vertex in vertices: + vertex._graph_uuid = graph_uuid def delay(self): """Build the whole graph, creates jobs and delay them""" graph = self._connect_graphs() - vertices = graph.vertices() + vertices = list(graph.vertices()) for vertex in vertices: - vertex._build_job() + if vertex._job_uuid is None: + vertex._job_uuid = str(uuid.uuid4()) - self._ensure_same_graph_uuid([vertex._generated_job for vertex in vertices]) + self._ensure_same_graph_uuid(vertices) if self._has_to_execute_directly(vertices): self._execute_graph_direct(graph) return + depends = defaultdict(set) + reverse = defaultdict(set) for vertex, neighbour in graph.edges(): - neighbour._generated_job.add_depends({vertex._generated_job}) + # neighbour depends on vertex + depends[neighbour].add(vertex._job_uuid) + reverse[vertex].add(neighbour._job_uuid) # If all the jobs of the graph have another job with the same identity, # we do not create them. Maybe we should check that the found jobs are # part of the same graph, but not sure it's really required... # Also, maybe we want to check only the root jobs. - existing_mapping = {} + identity_keys = {} for vertex in vertices: if not vertex.identity_key: continue - generated_job = vertex._generated_job - existing = generated_job.job_record_with_same_identity_key() + identity_keys[vertex] = JobSpec.from_call( + vertex._job_method, vertex._job_args, vertex._job_kwargs + ).compute_identity_key(vertex.identity_key) + existing_mapping = {} + for vertex, key in identity_keys.items(): + # TODO: bulk search + existing = JobStore(vertex.recordset.env).record_with_same_identity_key(key) if not existing: # at least one does not exist yet, we'll delay the whole graph existing_mapping.clear() @@ -283,12 +296,41 @@ def delay(self): # can retrieve the existing job in "_generated_job". # existing_mapping contains something only if *all* the job with an # identity have an existing one. - for vertex, existing in existing_mapping.items(): - vertex._generated_job = existing + if existing_mapping: + for vertex, existing in existing_mapping.items(): + vertex._generated_job = Job(existing) return for vertex in vertices: - vertex._generated_job.store() + existing_job = vertex._generated_job + if isinstance(existing_job, Job): + # TODO: test this + # the delayable was already delayed on its own and is now + # attached to a graph + existing_job.graph_uuid = vertex._graph_uuid + existing_job._depends_on_uuids.update(depends[vertex]) + existing_job._reverse_depends_on_uuids.update(reverse[vertex]) + if depends[vertex]: + existing_job.state = WAIT_DEPENDENCIES + JobStore(vertex.recordset.env).save(existing_job) + continue + # the enqueue env/transaction is the one of the delayed recordset + vertex._generated_job = JobStore(vertex.recordset.env).enqueue( + vertex._job_method, + vertex._job_args, + vertex._job_kwargs, + priority=vertex.priority, + max_retries=vertex.max_retries, + eta=vertex.eta, + description=vertex.description, + channel=vertex.channel, + identity_key=vertex.identity_key, + job_uuid=vertex._job_uuid, + graph_uuid=vertex._graph_uuid, + depends_on_uuids=depends[vertex], + reverse_depends_on_uuids=reverse[vertex], + state=WAIT_DEPENDENCIES if depends[vertex] else PENDING, + ) def _execute_graph_direct(self, graph): for delayable in graph.topological_sort(): @@ -445,6 +487,8 @@ class Delayable: "_job_method", "_job_args", "_job_kwargs", + "_job_uuid", + "_graph_uuid", "_generated_job", ) @@ -473,6 +517,8 @@ def __init__( self._job_method = None self._job_args = () self._job_kwargs = {} + self._job_uuid = None + self._graph_uuid = None self._generated_job = None @@ -569,22 +615,6 @@ def split(self, size, chain=False): return (DelayableChain if chain else DelayableGroup)(*delayables) - def _build_job(self): - if self._generated_job: - return self._generated_job - self._generated_job = Job( - self._job_method, - args=self._job_args, - kwargs=self._job_kwargs, - priority=self.priority, - max_retries=self.max_retries, - eta=self.eta, - description=self.description, - channel=self.channel, - identity_key=self.identity_key, - ) - return self._generated_job - def _store_args(self, *args, **kwargs): self._job_args = args self._job_kwargs = kwargs @@ -603,7 +633,8 @@ def __getattr__(self, name): def _execute_direct(self): assert self._generated_job - self._generated_job.perform() + # TODO + # self._generated_job.perform() class DelayableRecordset: diff --git a/queue_job/executor.py b/queue_job/executor.py new file mode 100644 index 000000000..5eac7ac77 --- /dev/null +++ b/queue_job/executor.py @@ -0,0 +1,337 @@ +# 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 threading +import time +import traceback +from collections.abc import Sequence +from contextlib import contextmanager +from io import StringIO + +from psycopg2 import OperationalError, errorcodes + +from odoo import api, sql_db +from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY + +from .exception import RetryableJobError +from .job import ENQUEUED, Job, JobStore + +_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 TransactionPolicy: + """Defines how the executor behaves regarding database transactions + + The actual implementation differences are on commits between the production runtime + and the tests. + + All commits from the queue job internals should go through this. + """ + + def commit(self, cr: sql_db.BaseCursor) -> None: + raise NotImplementedError + + +class SingleTransactionPolicy(TransactionPolicy): + """Production policy: commits on the control transaction are real.""" + + def commit(self, cr: sql_db.BaseCursor) -> None: + cr.commit() # pylint: disable=invalid-commit + + +class TestTransactionPolicy(TransactionPolicy): + "Test policy: never commits" + + def commit(self, cr: sql_db.BaseCursor) -> None: + cr.flush() + + +class JobTransaction: + """Owns the environment and storage behavior. + + It is implemented for 2 types of transactions: + + * Control transaction: handle most of the state changes of the jobs + * Execution transaction: for the actual execution of the jobs + + Noteworthy: the state changes such as set a job to started or failed + are always done by the control transaction, the change to 'done' is + handled by the execution transaction, as it should be done in the + same transaction in which the job was executed to be transactional. + """ + + def __init__(self, env: api.Environment, store: JobStore): + self.env = env + self.store = store + + +class ControlTransaction(JobTransaction): + """Responsible for taking the lock on a job, setting started, failed, + retries, updating dependent jobs, ... + + The only transaction that can commit. + """ + + def __init__( + self, env: api.Environment, store: JobStore, policy: TransactionPolicy + ): + super().__init__(env, store) + self.policy = policy + + def commit(self) -> None: + self.policy.commit(self.env.cr) + + +class ExecutionTransaction(JobTransaction): + """Responsible for the execution of a job + + No commit allowed + """ + + +class JobExecutor: + job_store_class = JobStore + + def __init__(self, env: api.Environment, job_uuid: str): + self.job_uuid = job_uuid + self.control = ControlTransaction( + env, self.job_store_class(env), self.select_policy() + ) + + def select_policy(self) -> TransactionPolicy: + """Return the :class:`TransactionPolicy` for this executor.""" + if getattr(threading.current_thread(), "testing", False): + return TestTransactionPolicy() + return SingleTransactionPolicy() + + def _execution_transaction(self, env: api.Environment) -> "ExecutionTransaction": + return ExecutionTransaction(env, self.job_store_class(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.control.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.control.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 + job = Job.load(self.control.env, self.job_uuid) + assert job and job.state == ENQUEUED + job.set_started() + self.control.store.add_lock_record(job) + self.control.store.save_state(job, expected_states=(ENQUEUED,)) + self.control.commit() + if not self.control.store.lock(job): + _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): + initial_state = job.state + 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 + self._record_retry(job, err, (initial_state,)) + _logger.debug("%s postponed", job) + return + + except Exception as orig_exception: + self._record_failure(job, orig_exception, (initial_state,)) + 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) + if self._use_isolated_transaction(job): + self._perform_isolated_transaction(job) + else: + self._perform_single_transaction(job) + _logger.debug("%s done", job) + + def _use_isolated_transaction(self, job: Job) -> bool: + """Whether the job function runs in its own transaction.""" + return bool(job.job_config.allow_commit) + + def _perform_single_transaction(self, job: Job) -> None: + # default mode: the execution transaction deliberately aliases + # the control transaction, commits are forbidden while the job + # function runs + execution = self._execution_transaction(self.control.env) + initial_state = job.state + + # the savepoint protects the control transaction from the + # execution + savepoint = self.control.env.cr.savepoint(flush=False) + try: + with _prevent_commit(execution.env.cr): + job.perform(execution.env) + # Triggers any stored computed fields before calling 'set_done' + # so that will be part of the 'exec_time' + execution.env.flush_all() + job.set_done() + # when job happens without failure, the done state change is + # done by the execution env, so it is transactional + execution.store.save_state(job, (initial_state,)) + execution.env.flush_all() + except Exception: + # Rollback to the savepoint so we can register the failure state + # in the same transaction + savepoint.close(rollback=True) + self.control.env.clear() + raise + savepoint.close(rollback=False) + + self.control.commit() + + def _perform_isolated_transaction(self, job: Job) -> None: + # TODO: implement the "isolated" (allowing commits) execution mode + raise NotImplementedError + + def _record_retry( + self, + job: Job, + err: RetryableJobError, + expected_states: Sequence[str], + ) -> None: + """Postpone the job for a later retry""" + job.set_postpone(result=str(err), seconds=err.seconds) + self.control.store.save_state(job, expected_states) + self.control.commit() + + def _record_failure( + self, + job: Job, + orig_exception: Exception, + expected_states: Sequence[str], + ) -> None: + """Record the failure of the job with the exception details.""" + buff = StringIO() + traceback.print_exc(file=buff) + traceback_txt = buff.getvalue() + _logger.error(traceback_txt) + job.set_failed(**self._get_failure_values(traceback_txt, orig_exception)) + self.control.store.save_state(job, expected_states) + buff.close() + self.control.commit() + + 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 self.control.env.cr.savepoint(): + self.control.store.enqueue_waiting(job) + 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/job.py b/queue_job/job.py index 032bdb933..10419d558 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -7,10 +7,10 @@ import os import sys import uuid -import weakref -from contextlib import contextmanager, nullcontext +import warnings +from collections.abc import Callable, Sequence from datetime import datetime, timedelta -from random import randint +from typing import Any, NamedTuple import odoo @@ -91,6 +91,117 @@ def identity_exact_hasher(job_): return hasher +class JobSpec(NamedTuple): + """Method call being enqueued to a Job (read-only)""" + + model_name: str + method_name: str + recordset: odoo.models.BaseModel + args: tuple + kwargs: dict + uuid: str | None = None + state: str = PENDING # initial state can be pending or wait_dependencies + + @classmethod + def from_call(cls, func, args=None, kwargs=None) -> "JobSpec": + """Build the :class:`JobSpec` of a method call""" + if not _is_model_method(func): + raise TypeError("Job accepts only methods of Models") + if args is None: + args = () + if isinstance(args, list): + args = tuple(args) + assert isinstance(args, tuple), f"{args}: args are not a tuple" + if kwargs is None: + kwargs = {} + assert isinstance(kwargs, dict), f"{kwargs}: kwargs are not a dict" + recordset = func.__self__ + return JobSpec(recordset._name, func.__name__, recordset, args, kwargs) + + def compute_identity_key( + self, + identity_key: "str | Callable | None" = None, + ) -> str | None: + """Compute the identity key of a method call being enqueued. + + A callable identity key is invoked with the :class:`JobSpec` + of the call. + """ + if identity_key is None or isinstance(identity_key, str): + return identity_key + return identity_key(self) + + +# see what to do of this +def _run_job_function(func, args, kwargs, current_try, max_retries=None) -> Any: + """Run a job function with the retry limits logic.""" + current_try += 1 + try: + result = func(*tuple(args), **kwargs) + except RetryableJobError as err: + if err.ignore_retry: + current_try -= 1 + raise + elif not max_retries: # infinite retries + raise + elif current_try >= max_retries: + type_, value, _traceback = sys.exc_info() + # change the exception type but keep the original + # traceback and message: + # http://blog.ianbicking.org/2007/09/12/re-raising-exceptions/ + new_exc = FailedJobError( + "Max. retries (%d) reached: %s" % (max_retries, value or type_) + ) + raise new_exc from err + raise + + return result, current_try + + +def _normalize_eta(value) -> datetime | None: + """Normalize an eta given as datetime, timedelta or seconds.""" + if not value: + return None + if isinstance(value, timedelta): + return datetime.now() + value + if isinstance(value, int): + return datetime.now() + timedelta(seconds=value) + return value + + +class JobPayload: + """What a job executes: the recordset and the arguments. + + Built: + + * eagerly at enqueue time with the enqueue environment + * on demand at loading/execution time with the execution environment + + """ + + def __init__(self, recordset: odoo.models.BaseModel, args=None, kwargs=None): + if args is None: + args = () + if isinstance(args, list): + args = tuple(args) + assert isinstance(args, tuple), f"{args}: args are not a tuple" + if kwargs is None: + kwargs = {} + assert isinstance(kwargs, dict), f"{kwargs}: kwargs are not a dict" + self.recordset = recordset + self.args = args + self.kwargs = kwargs + + @classmethod + def from_record(cls, db_record: odoo.models.BaseModel) -> "JobPayload": + """Deserialize the payload columns of a job record.""" + return cls( + db_record.records, + tuple(db_record.args or ()), + dict(db_record.kwargs or {}), + ) + + class Job: """A Job is a task to execute. It is the in-memory representation of a job. @@ -202,473 +313,122 @@ class Job: """ + def __new__(cls, db_record=None, *args, **kwargs): + # Catch calls to Job(func, ...) used before API changes + if db_record is not None and not isinstance(db_record, odoo.models.BaseModel): + # TODO: backward compatibility with a "shim"? + # here db_record is a func, not a db_record + return JobCompat(db_record, *args, **kwargs) + return super().__new__(cls) + @classmethod - def load(cls, env, job_uuid): + def load(cls, env: odoo.api.Environment, job_uuid: str) -> "Job": """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(f"Job {job_uuid} does no longer exist in the storage.") - return cls._load_from_db_record(stored) + raise NoSuchJobError( + f"Job {job_uuid} does no longer exist in the database." + ) + return cls(db_record=stored) @classmethod - def load_many(cls, env, job_uuids): + def load_many( + cls, env: odoo.api.Environment, job_uuids: Sequence[str] + ) -> set["Job"]: """Read jobs in batch from the Database Jobs not found are ignored. """ 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. - - 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( - """ - 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], - ) - - # 1 job should be locked - return bool(self.env.cr.fetchall()) - - @classmethod - def _load_from_db_record(cls, job_db_record): - stored = job_db_record - - args = stored.args - kwargs = stored.kwargs - method_name = stored.method_name - - recordset = stored.records - method = getattr(recordset, method_name) - - eta = None - if stored.eta: - eta = stored.eta - - job_ = cls( - method, - args=args, - kwargs=kwargs, - priority=stored.priority, - eta=eta, - job_uuid=stored.uuid, - description=stored.name, - channel=stored.channel, - identity_key=stored.identity_key, - ) - - if stored.date_created: - job_.date_created = stored.date_created - - if stored.date_enqueued: - job_.date_enqueued = stored.date_enqueued - - if stored.date_started: - job_.date_started = stored.date_started - - if stored.date_done: - job_.date_done = stored.date_done - - if stored.date_cancelled: - job_.date_cancelled = stored.date_cancelled - - job_.state = stored.state - job_.graph_uuid = stored.graph_uuid if stored.graph_uuid else None - job_.result = stored.result if stored.result else None - job_.exc_info = stored.exc_info if stored.exc_info else None - job_.retry = stored.retry - job_.max_retries = stored.max_retries - if stored.company_id: - job_.company_id = stored.company_id.id - job_.identity_key = stored.identity_key - job_.worker_pid = stored.worker_pid - - job_.__depends_on_uuids.update(stored.dependencies.get("depends_on", [])) - job_.__reverse_depends_on_uuids.update( - stored.dependencies.get("reverse_depends_on", []) - ) - return job_ - - def job_record_with_same_identity_key(self): - """Check if a job to be executed with the same key exists.""" - existing = ( - self.env["queue.job"] - .sudo() - .search( - [ - ("identity_key", "=", self.identity_key), - ("state", "in", [WAIT_DEPENDENCIES, PENDING, ENQUEUED]), - ], - limit=1, - ) - ) - return existing + return {cls(db_record=record) for record in recordset} @staticmethod - def db_records_from_uuids(env, job_uuids): + def db_records_from_uuids( + env: odoo.api.Environment, job_uuids: Sequence[str] + ) -> odoo.models.BaseModel: model = env["queue.job"].sudo() record = model.search([("uuid", "in", tuple(job_uuids))]) return record.with_env(env).sudo() - def __init__( - self, - func, - args=None, - kwargs=None, - priority=None, - eta=None, - job_uuid=None, - max_retries=None, - description=None, - channel=None, - identity_key=None, - ): - """Create a Job - - :param func: function to execute - :type func: function - :param args: arguments for func - :type args: tuple - :param kwargs: keyworkd arguments for func - :type kwargs: dict - :param priority: priority of the job, - the smaller is the higher priority - :type priority: int - :param eta: the job can be executed only after this datetime - (or now + timedelta) - :type eta: datetime or timedelta - :param job_uuid: UUID of the job - :param max_retries: maximum number of retries before giving up and set - the job state to 'failed'. A value of 0 means infinite retries. - :param description: human description of the job. If None, description - is computed from the function doc or name - :param channel: The complete channel name to use to process the job. - :param identity_key: A hash to uniquely identify a job, or a function - that returns this hash (the function takes the job - as argument) - """ - if args is None: - args = () - if isinstance(args, list): - args = tuple(args) - assert isinstance(args, tuple), f"{args}: args are not a tuple" - if kwargs is None: - kwargs = {} - - assert isinstance(kwargs, dict), f"{kwargs}: kwargs are not a dict" + def db_record(self, env: odoo.api.Environment) -> odoo.models.BaseModel: + """The ``queue.job`` record of the job, using env.""" + return self.db_records_from_uuids(env, [self.uuid]) - if not _is_model_method(func): - raise TypeError("Job accepts only methods of Models") + def __init__(self, db_record: odoo.models.BaseModel = None): + """Build the job from its ``queue.job`` record""" + if db_record is None: + raise TypeError("TODO: explain how jobs are created") - recordset = func.__self__ - self.method_name = func.__name__ - self.recordset = recordset + stored = db_record + self._stored_id = stored.id + self.method_name = stored.method_name + self.model_name = stored.model_name + self.user_id = stored.user_id.id if stored.user_id else None - self.job_config = ( - self.env["queue.job.function"].sudo().job_config(self.job_function_name) + func_model = stored.env["queue.job.function"].sudo() + self.job_function_name = func_model.job_function_name( + stored.model_name, stored.method_name ) - - self.state = PENDING - - self.retry = 0 - if max_retries is None: - self.max_retries = DEFAULT_MAX_RETRIES - else: - self.max_retries = max_retries - - self._uuid = job_uuid - self.graph_uuid = None - - self.args = args - self.kwargs = kwargs - - self.__depends_on_uuids = set() - self.__reverse_depends_on_uuids = set() - self._depends_on = set() - self._reverse_depends_on = weakref.WeakSet() - - self.priority = priority - if self.priority is None: - self.priority = DEFAULT_PRIORITY - - self.date_created = datetime.now() - self._description = description - - if isinstance(identity_key, str): - self._identity_key = identity_key - self._identity_key_func = None - else: - # we'll compute the key on the fly when called - # from the function - self._identity_key = None - self._identity_key_func = identity_key - - self.date_enqueued = None - self.date_started = None - self.date_done = None - self.date_cancelled = None - - self.result = None - self.exc_name = None - self.exc_message = None - self.exc_info = None - - if "company_id" in self.env.context: - company_id = self.env.context["company_id"] - else: - company_id = self.env.company.id - self.company_id = company_id + self.job_config = func_model.job_config(self.job_function_name) + + self.state = stored.state + self.retry = stored.retry + self.max_retries = stored.max_retries + self.uuid = stored.uuid + self.graph_uuid = stored.graph_uuid if stored.graph_uuid else None + self.priority = stored.priority + self.date_created = stored.date_created or None + self.date_enqueued = stored.date_enqueued or None + self.date_started = stored.date_started or None + self.date_done = stored.date_done or None + self.date_cancelled = stored.date_cancelled or None + self.description = stored.name + self.identity_key = stored.identity_key or None + self.result = stored.result if stored.result else None + self.exc_name = stored.exc_name if stored.exc_name else None + self.exc_message = stored.exc_message if stored.exc_message else None + self.exc_info = stored.exc_info if stored.exc_info else None + self.company_id = stored.company_id.id if stored.company_id else None self._eta = None - self.eta = eta - self.channel = channel - self.worker_pid = None + if stored.eta: + self.eta = stored.eta + self._channel = stored.channel + self.worker_pid = stored.worker_pid - def add_depends(self, jobs): - if self in jobs: - raise ValueError("job cannot depend on itself") - self.__depends_on_uuids |= {j.uuid for j in jobs} - self._depends_on.update(jobs) - for parent in jobs: - parent.__reverse_depends_on_uuids.add(self.uuid) - parent._reverse_depends_on.add(self) - if any(j.state != DONE for j in jobs): - self.state = WAIT_DEPENDENCIES + self._depends_on_uuids = set(stored.dependencies.get("depends_on", [])) + self._reverse_depends_on_uuids = set( + stored.dependencies.get("reverse_depends_on", []) + ) - def perform(self): - """Execute the job. + def perform(self, env: odoo.api.Environment) -> Any: + """Execute the job on ``env`` The job is executed with the user which has initiated it. - """ - self.retry += 1 - try: - if self.job_config.allow_commit: - env_context_manager = self.in_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 - raise - elif not self.max_retries: # infinite retries - raise - elif self.retry >= self.max_retries: - type_, value, traceback = sys.exc_info() - # change the exception type but keep the original - # traceback and message: - # http://blog.ianbicking.org/2007/09/12/re-raising-exceptions/ - new_exc = FailedJobError( - "Max. retries (%d) reached: %s" % (self.max_retries, value or type_) - ) - raise new_exc from err - raise - return self.result + The payload (recordset, args, kwargs) is retrieved from ``env``, + and the job function runs on it, with the user which has initiated the + job. - @contextmanager - def in_temporary_env(self): - with self.env.registry.cursor() as new_cr: - env = self.env - self._env = env(cr=new_cr) - try: - yield - finally: - self._env = env - - def _get_common_dependent_jobs_query(self): - return """ - UPDATE queue_job - SET state = %s - FROM ( - SELECT child.id, array_agg(parent.state) as parent_states - FROM queue_job job - JOIN LATERAL - json_array_elements_text( - job.dependencies::json->'reverse_depends_on' - ) child_deps ON true - JOIN queue_job child - ON child.graph_uuid = job.graph_uuid - AND child.uuid = child_deps - JOIN LATERAL - json_array_elements_text( - child.dependencies::json->'depends_on' - ) parent_deps ON true - JOIN queue_job parent - ON parent.graph_uuid = job.graph_uuid - AND parent.uuid = parent_deps - WHERE job.uuid = %s - GROUP BY child.id - ) jobs - WHERE - queue_job.id = jobs.id - AND %s = ALL(jobs.parent_states) - AND state = %s; + The transaction of ``env`` is the responsibility of the caller: ``JobExecutor`` + passes its execution transaction. """ + payload = self.payload(env) + recordset = payload.recordset.with_context(job_uuid=self.uuid) + func = getattr(recordset, self.method_name) + self.result, self.retry = _run_job_function( + func, payload.args, payload.kwargs, self.retry, self.max_retries + ) 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)) - self.env["queue.job"].invalidate_model(["state"]) - - def cancel_dependent_jobs(self): - sql = self._get_common_dependent_jobs_query() - self.env.cr.execute(sql, (CANCELLED, self.uuid, CANCELLED, WAIT_DEPENDENCIES)) - self.env["queue.job"].invalidate_model(["state"]) + return any(self._reverse_depends_on_uuids) - def store(self): - """Store the Job""" - job_model = self.env["queue.job"] - # The sentinel is used to prevent edition sensitive fields (such as - # method_name) from RPC methods. - edit_sentinel = job_model.EDIT_SENTINEL - - db_record = self.db_record() - if db_record: - db_record.with_context(_job_edit_sentinel=edit_sentinel).write( - self._store_values() - ) - else: - job_model.with_context(_job_edit_sentinel=edit_sentinel).sudo().create( - self._store_values(create=True) - ) - - def _store_values(self, create=False): - vals = { - "state": self.state, - "priority": self.priority, - "retry": self.retry, - "max_retries": self.max_retries, - "exc_name": self.exc_name, - "exc_message": self.exc_message, - "exc_info": self.exc_info, - "company_id": self.company_id, - "result": str(self.result) if self.result else False, - "date_enqueued": False, - "date_started": False, - "date_done": False, - "exec_time": False, - "date_cancelled": False, - "eta": False, - "identity_key": False, - "worker_pid": self.worker_pid, - "graph_uuid": self.graph_uuid, - } - - if self.date_enqueued: - vals["date_enqueued"] = self.date_enqueued - if self.date_started: - vals["date_started"] = self.date_started - if self.date_done: - vals["date_done"] = self.date_done - if self.exec_time: - vals["exec_time"] = self.exec_time - if self.date_cancelled: - vals["date_cancelled"] = self.date_cancelled - if self.eta: - vals["eta"] = self.eta - if self.identity_key: - vals["identity_key"] = self.identity_key - - dependencies = { - "depends_on": [parent.uuid for parent in self.depends_on], - "reverse_depends_on": [ - children.uuid for children in self.reverse_depends_on - ], - } - vals["dependencies"] = dependencies - - if create: - vals.update( - { - "user_id": self.env.uid, - "channel": self.channel, - # The following values must never be modified after the - # creation of the job - "uuid": self.uuid, - "name": self.description, - "func_string": self.func_string, - "date_created": self.date_created, - "model_name": self.recordset._name, - "method_name": self.method_name, - "job_function_id": self.job_config.job_function_id, - "channel_method_name": self.job_function_name, - "records": self.recordset, - "args": self.args, - "kwargs": self.kwargs, - } - ) - - vals_from_model = self._store_values_from_model() - # Sanitize values: make sure you cannot screw core values - vals_from_model = {k: v for k, v in vals_from_model.items() if k not in vals} - vals.update(vals_from_model) - return vals - - def _store_values_from_model(self): - vals = {} - value_handlers_candidates = ( - "_job_store_values_for_" + self.method_name, - "_job_store_values", - ) - for candidate in value_handlers_candidates: - handler = getattr(self.recordset, candidate, None) - if handler is not None: - vals = handler(self) - return vals - - @property - def func_string(self): - model = repr(self.recordset) - args = [repr(arg) for arg in self.args] - kwargs = [f"{key}={val!r}" for key, val in self.kwargs.items()] - all_args = ", ".join(args + kwargs) - return f"{model}.{self.method_name}({all_args})" + def payload(self, env: odoo.api.Environment) -> JobPayload: + """The payload of the job in the ``env`` environment""" + record = env["queue.job"].sudo().browse(self._stored_id) + return JobPayload.from_record(record) def __eq__(self, other): return self.uuid == other.uuid @@ -676,82 +436,23 @@ def __eq__(self, other): def __hash__(self): return self.uuid.__hash__() - def db_record(self): - return self.db_records_from_uuids(self.env, [self.uuid]) - - @property - 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) - return getattr(recordset, self.method_name) - - @property - def job_function_name(self): - func_model = self.env["queue.job.function"].sudo() - return func_model.job_function_name(self.recordset._name, self.method_name) - - @property - def identity_key(self): - if self._identity_key is None: - if self._identity_key_func: - self._identity_key = self._identity_key_func(self) - return self._identity_key - - @identity_key.setter - def identity_key(self, value): - if isinstance(value, str): - self._identity_key = value - self._identity_key_func = None - else: - # we'll compute the key on the fly when called - # from the function - self._identity_key = None - self._identity_key_func = value - - @property - def depends_on(self): - if not self._depends_on: - self._depends_on = Job.load_many(self.env, self.__depends_on_uuids) - return self._depends_on - @property - def reverse_depends_on(self): - if not self._reverse_depends_on: - self._reverse_depends_on = Job.load_many( - self.env, self.__reverse_depends_on_uuids - ) - return set(self._reverse_depends_on) + def depends_on_uuids(self) -> set[str]: + """UUIDs of the jobs this job depends on (no database access).""" + return set(self._depends_on_uuids) @property - def description(self): - if self._description: - return self._description - elif self.func.__doc__: - return self.func.__doc__.splitlines()[0].strip() - else: - return f"{self.model_name}.{self.func.__name__}" + def reverse_depends_on_uuids(self) -> set[str]: + """UUIDs of the jobs depending on this job (no database access).""" + return set(self._reverse_depends_on_uuids) - @property - def uuid(self): - """Job ID, this is an UUID""" - if self._uuid is None: - self._uuid = str(uuid.uuid4()) - return self._uuid + def depends_on(self, env: odoo.api.Environment) -> "set[Job]": + """The jobs this job depends on, loaded in ``env``, never kept.""" + return Job.load_many(env, self._depends_on_uuids) - @property - def model_name(self): - return self.recordset._name - - @property - def user_id(self): - return self.env.uid + def reverse_depends_on(self, env: odoo.api.Environment) -> "set[Job]": + """The jobs depending on this job (see :meth:`depends_on`).""" + return Job.load_many(env, self._reverse_depends_on_uuids) @property def eta(self): @@ -759,14 +460,7 @@ def eta(self): @eta.setter def eta(self, value): - if not value: - self._eta = None - elif isinstance(value, timedelta): - self._eta = datetime.now() + value - elif isinstance(value, int): - self._eta = datetime.now() + timedelta(seconds=value) - else: - self._eta = value + self._eta = _normalize_eta(value) @property def channel(self): @@ -782,11 +476,12 @@ def exec_time(self): return (self.date_done - self.date_started).total_seconds() return None - def set_pending(self, result=None, reset_retry=True): - if any(j.state != DONE for j in self.depends_on): - self.state = WAIT_DEPENDENCIES - else: - self.state = PENDING + def set_pending( + self, + result: Any = None, + reset_retry: bool = True, + ) -> None: + self.state = PENDING self.date_enqueued = None self.date_started = None self.date_done = None @@ -807,7 +502,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 @@ -823,62 +517,497 @@ def set_cancelled(self, result=None): if result is not None: self.result = result - def set_failed(self, **kw): + def set_failed( + self, + exc_name: str | None = None, + exc_message: str | None = None, + exc_info: str | None = None, + **kwargs, + ) -> None: self.state = FAILED - for k, v in kw.items(): - if v is not None: - setattr(self, k, v) - def __repr__(self): - return "" % (self.uuid, self.priority) - - def _get_retry_seconds(self, seconds=None): - retry_pattern = self.job_config.retry_pattern - if not seconds and retry_pattern: - # ordered from higher to lower count of retries - patt = sorted(retry_pattern.items(), key=lambda t: t[0]) - seconds = RETRY_INTERVAL - for retry_count, postpone_seconds in patt: - if self.retry >= retry_count: - seconds = postpone_seconds - else: - break - elif not seconds: - seconds = RETRY_INTERVAL - if isinstance(seconds, (list | tuple)): - seconds = randint(seconds[0], seconds[1]) - return seconds - - def postpone(self, result=None, seconds=None): - """Postpone the job + if exc_name is not None: + self.exc_name = exc_name + if exc_message is not None: + self.exc_message = exc_message + if exc_info is not None: + self.exc_info = exc_info + if kwargs: + # TODO: deprecate? why do we allow this? + for key, value in kwargs.items(): + if value is not None: + setattr(self, key, value) + + def set_postpone( + self, + result: Any = None, + seconds: int | None = None, + ) -> None: + """Postpone the job for a later retry Write an estimated time arrival to n seconds later than now. Used when an retryable exception want to retry a job later. """ - eta_seconds = self._get_retry_seconds(seconds) - self.eta = timedelta(seconds=eta_seconds) + self.eta = timedelta(seconds=self._get_retry_seconds(seconds)) self.exc_name = None + self.exc_message = None self.exc_info = None - if result is not None: - self.result = result + self.set_pending(result=result, reset_retry=False) + + def postpone(self, result=None, seconds=None): + # TODO: deprecation warning + return self.set_postpone(result=result, seconds=seconds) + + def __repr__(self): + return f" int: + return self.job_config.retry_seconds(self.retry, seconds=seconds) + + def _no_env_compat(self, name: str, replacement: str): + warnings.warn( + f"{name} is deprecated and will be removed in the next major " + f"version, use {replacement} instead", + DeprecationWarning, + stacklevel=3, + ) + raise RuntimeError( + f"Job has no environment. {name} is deprecated, " + f"use {replacement} instead" + ) + + def store(self): + """Deprecated: use JobStore(env).enqueue(...) or JobStore(env).save(job)""" + self._no_env_compat( + "store", "JobStore(env).enqueue(...) or JobStore(env).save(job)" + ) + + def add_lock_record(self) -> None: + """Deprecated: use ``JobStore(env).add_lock_record(job)``.""" + self._no_env_compat("Job.add_lock_record", "JobStore(env).add_lock_record(job)") + + def lock(self) -> bool: + """Deprecated: use ``JobStore(env).lock(job)``.""" + self._no_env_compat("Job.lock", "JobStore(env).lock(job)") + + def enqueue_waiting(self) -> None: + """Deprecated: use ``JobStore(env).enqueue_waiting(job)``.""" + self._no_env_compat("Job.enqueue_waiting", "JobStore(env).enqueue_waiting(job)") + + def cancel_dependent_jobs(self) -> None: + """Deprecated: use ``JobStore(env).cancel_dependent_jobs(job)``.""" + self._no_env_compat( + "Job.cancel_dependent_jobs", "JobStore(env).cancel_dependent_jobs(job)" + ) + + def _store_values(self, create: bool = False) -> dict: + """Deprecated: use ``JobStore(env)._store_values(job)``.""" + self._no_env_compat("Job._store_values", "JobStore(env)._store_values(job)") + + +class JobCompat: + """Deprecated compatibility shim""" + + # TODO + def __init__( + self, + func, + args=None, + kwargs=None, + priority=None, + eta=None, + job_uuid=None, + max_retries=None, + description=None, + channel=None, + identity_key=None, + ): + job_spec = JobSpec.from_call(func, args, kwargs) + self.func = func + self.recordset = job_spec.recordset + self.args = job_spec.args + self.kwargs = job_spec.kwargs + self.model_name = job_spec.model_name + self.method_name = job_spec.method_name + self.priority = priority if priority is not None else DEFAULT_PRIORITY + self._eta = _normalize_eta(eta) + self.max_retries = DEFAULT_MAX_RETRIES if max_retries is None else max_retries + self.description = description + self.channel = channel + self.identity_key = identity_key + self._uuid = job_uuid + func_model = self.recordset.env["queue.job.function"].sudo() + self.job_config = func_model.job_config( + func_model.job_function_name(self.model_name, self.method_name) + ) + self.state = PENDING + self.retry = 0 + self.result = None + self.exc_name = None + self.exc_message = None + self.exc_info = None + self.date_created = datetime.now() + self.date_enqueued = None + self.date_started = None + self.date_done = None + self.date_cancelled = None + self.worker_pid = None + self._stored_job = None + + +class JobStore: + """Persistence of jobs + + The only writer of ``queue.job`` for any creation or updates. - def related_action(self): - record = self.db_record() - if not self.job_config.related_action_enable: - return None - - funcname = self.job_config.related_action_func_name - if not funcname: - funcname = record._default_related_action - if not isinstance(funcname, str): - raise ValueError( - "related_action must be the name of the " - "method on queue.job as string" + Records are stored with the environment / transaction given + at construction. + """ + + def __init__(self, env: odoo.api.Environment): + self.env = env + + def enqueue( + self, + func, + args: Sequence | None = None, + kwargs: dict | None = None, + *, + priority: int | None = None, + eta=None, + max_retries: int | None = None, + description: str | None = None, + channel: str | None = None, + identity_key: "str | Callable | None" = None, + job_uuid: str | None = None, + graph_uuid: str | None = None, + depends_on_uuids: "Sequence[str] | set[str] | None" = None, + reverse_depends_on_uuids: "Sequence[str] | set[str] | None" = None, + state: str = PENDING, + ) -> "Job | JobCompat": + """Create the ``queue_job`` record of a method call, return its Job. + + The place where jobs are born. The delay() calls end up here. + """ + job_spec = JobSpec.from_call(func, args, kwargs) + recordset = job_spec.recordset + env = recordset.env + + func_model = env["queue.job.function"].sudo() + job_function_name = func_model.job_function_name( + job_spec.model_name, job_spec.method_name + ) + job_config = func_model.job_config(job_function_name) + + if not isinstance(identity_key, str | None): + identity_key = identity_key(job_spec) + + if description is None: + description = ( + func.__doc__ and func.__doc__.splitlines()[0].strip() + ) or f"{job_spec.model_name}.{job_spec.method_name}" + + str_args = [repr(arg) for arg in job_spec.args] + str_args += [f"{key}={val!r}" for key, val in job_spec.kwargs.items()] + func_string = f"{recordset!r}.{job_spec.method_name}({', '.join(str_args)})" + + if "company_id" in env.context: + company_id = env.context["company_id"] + else: + company_id = env.company.id + + vals = { + "uuid": job_uuid or str(uuid.uuid4()), + "graph_uuid": graph_uuid, + "state": state, + "priority": DEFAULT_PRIORITY if priority is None else priority, + "retry": 0, + "max_retries": ( + DEFAULT_MAX_RETRIES if max_retries is None else max_retries + ), + "user_id": env.uid, + "company_id": company_id, + "channel": channel or job_config.channel, + "name": description, + "func_string": func_string, + "date_created": datetime.now(), + "model_name": job_spec.model_name, + "method_name": job_spec.method_name, + "job_function_id": job_config.job_function_id, + "channel_method_name": job_function_name, + "records": recordset, + "args": job_spec.args, + "kwargs": job_spec.kwargs, + "eta": _normalize_eta(eta) or False, + "identity_key": identity_key or False, + "dependencies": { + "depends_on": sorted(depends_on_uuids or ()), + "reverse_depends_on": sorted(reverse_depends_on_uuids or ()), + }, + } + + hook_job_spec = job_spec._replace(uuid=vals["uuid"], state=state) + # TODO: breaking as the hooks do not have a full job but a job spec + vals_from_model = self._values_from_model(hook_job_spec) + # Sanitize values: make sure you cannot screw core values + vals.update({k: v for k, v in vals_from_model.items() if k not in vals}) + + job_model = self.env["queue.job"].sudo() + record = job_model.with_context( + _job_edit_sentinel=job_model.EDIT_SENTINEL + ).create(vals) + return Job(record) + + def save(self, job: "Job") -> None: + """Update the database record of ``job``""" + self._maybe_transition_wait_dependencies(job) + + job_model = self.env["queue.job"].sudo() + + db_record = job_model.search([("uuid", "=", job.uuid)], limit=1) + if not db_record: + _logger.warning( + "job %s not saved: its record does no longer exist", job.uuid ) - action = getattr(record, funcname) - action_kwargs = self.job_config.related_action_kwargs - return action(**action_kwargs) + return + self._write_record(db_record, job, self._all_values(job)) + + def save_state(self, job: "Job", expected_states: Sequence[str]) -> bool: + """Persist the current state of ``job`` if the job matches the expected states + + The guard takes a row lock making the check and the write atomic: a + competing transition waits on the lock and then sees the new state. + + What the expected_states guards protects, considering odoo uses + REPEATABLE READ isolation: The execution of a job crosses transaction + boundaries, with several commits. The in-memory state of a job may + be outdated compared to the actual db state. + + Return True when the row was written, False when the job was not + in one of ``expected_states`` anymore. + """ + self._maybe_transition_wait_dependencies(job) + + # the raw locking SELECT bypasses the ORM: flush any pending + # state writes first + self.env["queue.job"].flush_model() + self.env.cr.execute( + "SELECT id FROM queue_job WHERE uuid = %s AND state IN %s " + "FOR NO KEY UPDATE", + (job.uuid, tuple(expected_states)), + ) + row = self.env.cr.fetchone() + if not row: + _logger.warning( + "state '%s' not saved for job %s: the job is not in " + "state(s) %s anymore (transitioned concurrently or deleted)", + job.state, + job.uuid, + ", ".join(expected_states), + ) + return False + record = self.env["queue.job"].sudo().browse(row[0]) + self._write_record(record, job, self._state_values(job)) + return True + + def _maybe_transition_wait_dependencies(self, job: "Job") -> None: + """Transition to wait dependencies unless of pending if still waiting + + A pending job with unfinished parents waits for them. + """ + if job.state != PENDING or not job._depends_on_uuids: + return + unfinished = ( + self.env["queue.job"] + .sudo() + .search_count( + [ + ("uuid", "in", job._depends_on_uuids), + ("state", "!=", DONE), + ], + limit=1, + ) + ) + if unfinished: + job.state = WAIT_DEPENDENCIES + + def _write_record( + self, db_record: odoo.models.BaseModel, job: "Job", vals: dict + ) -> None: + """Write the state values of ``job`` on its record.""" + vals_from_model = self._values_from_model(job) + # Sanitize values: make sure you cannot screw core values + vals.update({k: v for k, v in vals_from_model.items() if k not in vals}) + # The sentinel is used to prevent edition sensitive fields (such as + # method_name) from RPC methods. + db_record.with_context(_job_edit_sentinel=db_record.EDIT_SENTINEL).write(vals) + + def _state_values(self, job: "Job") -> dict: + """The fields saved by save_state. + + These are only the fields mutated by the various ``set_done``, ``set_failed``, + ... methods. + """ + return { + "state": job.state, + "retry": job.retry, + "result": str(job.result) if job.result else False, + "exc_name": job.exc_name, + "exc_message": job.exc_message, + "exc_info": job.exc_info, + "date_enqueued": job.date_enqueued or False, + "date_started": job.date_started or False, + "date_done": job.date_done or False, + "exec_time": job.exec_time or False, + "date_cancelled": job.date_cancelled or False, + "eta": job.eta or False, + "worker_pid": job.worker_pid, + } + + def _all_values(self, job: "Job") -> dict: + """The whole mutable data of the job.""" + vals = self._state_values(job) + vals.update( + { + "priority": job.priority, + "max_retries": job.max_retries, + "company_id": job.company_id, + "identity_key": job.identity_key or False, + "graph_uuid": job.graph_uuid, + "dependencies": { + "depends_on": sorted(job.depends_on_uuids), + "reverse_depends_on": sorted(job.reverse_depends_on_uuids), + }, + } + ) + return vals + + def _values_from_model(self, job: "Job | JobSpec") -> dict: + """Additional values contributed by the model's hooks.""" + vals = {} + if job.model_name not in self.env: + # the model of the job is gone (uninstalled module): state + # control must keep working, and there is no hook to call + return vals + model = self.env[job.model_name] + value_handlers_candidates = ( + "_job_store_values_for_" + job.method_name, + "_job_store_values", + ) + for candidate in value_handlers_candidates: + handler = getattr(model, candidate, None) + if handler is not None: + vals = handler(job) + return vals + + def add_lock_record(self, job: "Job") -> None: + """Create the row 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; + """, + [job.uuid], + ) + + def lock(self, job: "Job") -> bool: + """Lock the row of the job being performed. + + Return False when the job cannot be locked: it is not in STARTED + state or it is already locked by another worker. + """ + 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; + """, + [job.uuid, STARTED], + ) + + # 1 job should be locked + return bool(self.env.cr.fetchall()) + + def record_with_same_identity_key(self, identity_key: str) -> odoo.models.BaseModel: + """A pending job record with this identity key, if any.""" + return ( + self.env["queue.job"] + .sudo() + .search( + [ + ("identity_key", "=", identity_key), + ("state", "in", [WAIT_DEPENDENCIES, PENDING, ENQUEUED]), + ], + limit=1, + ) + ) + + _DEPENDENT_JOBS_QUERY = """ + UPDATE queue_job + SET state = %s + FROM ( + SELECT child.id, array_agg(parent.state) as parent_states + FROM queue_job job + JOIN LATERAL + json_array_elements_text( + job.dependencies::json->'reverse_depends_on' + ) child_deps ON true + JOIN queue_job child + ON child.graph_uuid = job.graph_uuid + AND child.uuid = child_deps + JOIN LATERAL + json_array_elements_text( + child.dependencies::json->'depends_on' + ) parent_deps ON true + JOIN queue_job parent + ON parent.graph_uuid = job.graph_uuid + AND parent.uuid = parent_deps + WHERE job.uuid = %s + GROUP BY child.id + ) jobs + WHERE + queue_job.id = jobs.id + AND %s = ALL(jobs.parent_states) + AND state = %s; + """ + + def enqueue_waiting(self, job: "Job") -> None: + """Set to pending the dependent jobs whose parents are all done.""" + self.env["queue.job"].flush_model() + self.env.cr.execute( + self._DEPENDENT_JOBS_QUERY, (PENDING, job.uuid, DONE, WAIT_DEPENDENCIES) + ) + self.env["queue.job"].invalidate_model(["state"]) + + def cancel_dependent_jobs(self, job: "Job") -> None: + """Cancel the dependent jobs whose parents are all cancelled.""" + self.env["queue.job"].flush_model() + self.env.cr.execute( + self._DEPENDENT_JOBS_QUERY, + (CANCELLED, job.uuid, CANCELLED, WAIT_DEPENDENCIES), + ) + self.env["queue.job"].invalidate_model(["state"]) def _is_model_method(func): diff --git a/queue_job/models/queue_job.py b/queue_job/models/queue_job.py index 011edf09d..fb4200157 100644 --- a/queue_job/models/queue_job.py +++ b/queue_job/models/queue_job.py @@ -24,6 +24,7 @@ STATES, WAIT_DEPENDENCIES, Job, + JobStore, ) _logger = logging.getLogger(__name__) @@ -279,8 +280,11 @@ def write(self, vals): def open_related_action(self): """Open the related action associated to the job""" self.ensure_one() - job = Job.load(self.env, self.uuid) - action = job.related_action() + func_model = self.env["queue.job.function"].sudo() + job_config = func_model.job_config( + func_model.job_function_name(self.model_name, self.method_name) + ) + action = job_config.related_action(self) if action is None: raise exceptions.UserError(_("No action available for this job")) return action @@ -308,26 +312,27 @@ def _change_job_state(self, state, result=None): Changing the state of the Job will automatically change some fields (date, result, ...). """ + store = JobStore(self.env) for record in self: job_ = Job.load(record.env, record.uuid) + state_when_loaded = job_.state if state == DONE: job_.set_done(result=result) - job_.store() - record.env["queue.job"].flush_model() - job_.enqueue_waiting() + store.save_state(job_, (state_when_loaded,)) + store.enqueue_waiting(job_) elif state == PENDING: job_.set_pending(result=result) - job_.store() + store.save_state(job_, (state_when_loaded,)) elif state == CANCELLED: job_.set_cancelled(result=result) - job_.store() - record.env["queue.job"].flush_model() - job_.cancel_dependent_jobs() + store.save_state(job_, (state_when_loaded,)) + store.cancel_dependent_jobs(job_) else: raise ValueError(f"State not supported: {state}") def button_done(self): # If job was set to STARTED or CANCELLED, do not set it to DONE + # TODO: harmonize transitions with a state machine states_from = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, FAILED) result = _("Manually set to done by {}").format(self.env.user.name) records = self.filtered(lambda job_: job_.state in states_from) diff --git a/queue_job/models/queue_job_function.py b/queue_job/models/queue_job_function.py index edf90c9ab..f8740304f 100644 --- a/queue_job/models/queue_job_function.py +++ b/queue_job/models/queue_job_function.py @@ -5,10 +5,13 @@ import logging import re from collections import namedtuple +from random import randint +import odoo from odoo import _, api, exceptions, fields, models, tools from ..fields import JobSerialized +from ..job import RETRY_INTERVAL _logger = logging.getLogger(__name__) @@ -16,12 +19,8 @@ regex_job_function_name = re.compile(r"^<([0-9a-z_\.]+)>\.([0-9a-zA-Z_]+)$") -class QueueJobFunction(models.Model): - _name = "queue.job.function" - _description = "Job Functions" - _log_access = False - - JobConfig = namedtuple( +class JobConfig( + namedtuple( "JobConfig", "channel " "retry_pattern " @@ -31,6 +30,48 @@ class QueueJobFunction(models.Model): "job_function_id " "allow_commit", ) +): + """Configuration of a job function, cached once per registry.""" + + def retry_seconds(self, retry: int, seconds: int | None = None) -> int: + """Seconds to postpone the given retry, from the retry pattern.""" + retry_pattern = self.retry_pattern + if not seconds and retry_pattern: + # ordered from higher to lower count of retries + patt = sorted(retry_pattern.items(), key=lambda t: t[0]) + seconds = RETRY_INTERVAL + for retry_count, postpone_seconds in patt: + if retry >= retry_count: + seconds = postpone_seconds + else: + break + if not seconds: + seconds = RETRY_INTERVAL + if isinstance(seconds, (list | tuple)): + seconds = randint(seconds[0], seconds[1]) + return seconds + + def related_action(self, job_record: odoo.models.BaseModel): + """Execute the configured related action on the job's record.""" + if not self.related_action_enable: + return None + + funcname = self.related_action_func_name or job_record._default_related_action + if not isinstance(funcname, str): + raise ValueError( + "related_action must be the name of the " + "method on queue.job as string" + ) + action = getattr(job_record, funcname) + return action(**self.related_action_kwargs) + + +class QueueJobFunction(models.Model): + _name = "queue.job.function" + _description = "Job Functions" + _log_access = False + + JobConfig = JobConfig def _default_channel(self): return self.env.ref("queue_job.channel_root") diff --git a/queue_job/tests/__init__.py b/queue_job/tests/__init__.py index 16bcdff96..a5a559f1d 100644 --- a/queue_job/tests/__init__.py +++ b/queue_job/tests/__init__.py @@ -1,10 +1,15 @@ -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_job_payload, + test_job_store, + test_json_field, + test_model_job_channel, + test_model_job_function, + test_queue_job_protected_write, + test_run_job_controller, + 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..3063b7d23 --- /dev/null +++ b/queue_job/tests/test_executor.py @@ -0,0 +1,107 @@ +# Copyright 2026 QoQa Services SA (https://www.qoqa.ch) +# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) + +from unittest import mock + +from odoo.tests.common import TransactionCase +from odoo.tools import mute_logger + +from ..exception import JobError, RetryableJobError +from ..executor import JobExecutor, TestTransactionPolicy +from ..job import DONE, FAILED, PENDING, STARTED, Job, JobStore + + +class TestJobExecutor(TransactionCase): + def _create_enqueued_job(self, **kwargs): + job = self.env["queue.job"].with_delay()._test_job(**kwargs) + job.set_enqueued() + JobStore(self.env).save(job) + # acquire() reads the state with raw SQL: flush the ORM writes. + # In production the state is written and committed by the jobrunner + # on its own connection. + self.env.flush_all() + return job + + def test_select_policy_in_tests(self): + executor = JobExecutor(self.env, "uuid") + self.assertIsInstance(executor.control.policy, TestTransactionPolicy) + + def test_acquire_marks_started_and_locks(self): + job = self._create_enqueued_job() + executor = JobExecutor(self.env, job.uuid) + acquired = executor.acquire() + self.assertEqual(acquired.uuid, job.uuid) + self.assertEqual(acquired.state, STARTED) + self.assertEqual(acquired.db_record(self.env).state, STARTED) + + def test_run_success(self): + job = self._create_enqueued_job() + executor = JobExecutor(self.env, job.uuid) + executor.run() + self.assertEqual(job.db_record(self.env).state, DONE) + + @mute_logger("odoo.addons.queue_job.executor") + def test_acquire_missing_job(self): + executor = JobExecutor(self.env, "not-existing-uuid") + self.assertIsNone(executor.acquire()) + + @mute_logger("odoo.addons.queue_job.executor") + def test_acquire_wrong_state(self): + """A job that is not enqueued cannot be acquired""" + job = self.env["queue.job"].with_delay()._test_job() + JobStore(self.env).save(job) # pending + executor = JobExecutor(self.env, job.uuid) + self.assertIsNone(executor.acquire()) + self.assertEqual(job.db_record(self.env).state, PENDING) + + @mute_logger("odoo.addons.queue_job.executor") + def test_run_job_failure(self): + job = self._create_enqueued_job(failure_rate=1) + executor = JobExecutor(self.env, job.uuid) + + acquired = executor.acquire() + # try/except instead of assertRaises: Odoo's assertRaises wraps the + # block in a savepoint, which would also roll back the recorded + # failure state + raised = None + try: + executor.run_job(acquired) + except JobError as err: + raised = err + + self.assertIsNotNone(raised) + + record = job.db_record(self.env) + + self.assertEqual(record.state, FAILED) + self.assertTrue(record.exc_name) + self.assertTrue( + any( + "Something bad happened" in (message.body or "") + for message in record.message_ids + ) + ) + + def test_run_job_retryable(self): + job = self._create_enqueued_job() + executor = JobExecutor(self.env, job.uuid) + acquired = executor.acquire() + + with mock.patch.object( + Job, "perform", side_effect=RetryableJobError("try later", seconds=60) + ): + executor.run_job(acquired) + + record = job.db_record(self.env) + + self.assertEqual(record.state, PENDING) + self.assertTrue(record.eta) + + def test_get_failure_values(self): + job = self._create_enqueued_job() + + 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"} + ) diff --git a/queue_job/tests/test_job_payload.py b/queue_job/tests/test_job_payload.py new file mode 100644 index 000000000..7bab09cbf --- /dev/null +++ b/queue_job/tests/test_job_payload.py @@ -0,0 +1,32 @@ +# 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 ..job import Job, JobStore + + +class TestJobPayload(TransactionCase): + def test_payload_binds_recordset_and_args(self): + partner1 = self.env["res.partner"].create({"name": "one"}) + partner2 = self.env["res.partner"].create({"name": "two"}) + job = partner1.with_delay().concat(partner2) + + JobStore(self.env).save(job) + self.env.flush_all() + + other_env = self.env(context=dict(self.env.context, _other_env=True)) + + loaded = Job.load(other_env, job.uuid) + payload = loaded.payload(other_env) + + self.assertEqual(payload.recordset, partner1) + self.assertEqual(payload.args[0], partner2) + + for record in (payload.recordset, *payload.args): + self.assertIs(record.env.cr, other_env.cr) + self.assertIs(record.env.transaction, other_env.transaction) + + # the same job binds the payload to a different env on another call + again = loaded.payload(self.env) + self.assertIs(again.recordset.env.transaction, self.env.transaction) diff --git a/queue_job/tests/test_job_store.py b/queue_job/tests/test_job_store.py new file mode 100644 index 000000000..e5be19559 --- /dev/null +++ b/queue_job/tests/test_job_store.py @@ -0,0 +1,55 @@ +# 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 ..job import DONE, PENDING, STARTED, JobStore + + +class TestJobStore(TransactionCase): + def _new_job(self): + job = self.env["queue.job"].with_delay()._test_job() + JobStore(self.env).save(job) + self.env.flush_all() + return job + + def test_save_expected_state(self): + job = self._new_job() # pending + job.set_enqueued() + job.set_started() + job.set_done() + + saved = JobStore(self.env).save_state(job, expected_states=(PENDING,)) + self.assertTrue(saved) + record = job.db_record(self.env) + self.assertEqual(record.state, DONE) + + def test_save_guards_state_unexpected(self): + job = self._new_job() # pending + job.set_enqueued() + job.set_started() + job.set_done() + + # when the expected state is started but it is actually done, + # it should be skipped and reported + with self.assertLogs("odoo.addons.queue_job.job", "WARNING") as watcher: + saved = JobStore(self.env).save_state(job, (STARTED,)) + + self.assertFalse(saved) + + self.assertIn("not saved", watcher.output[0]) + + record = job.db_record(self.env) + self.assertEqual(record.state, PENDING) + + def test_save_deleted_job(self): + job = self._new_job() # pending + job.db_record(self.env).unlink() + self.env.flush_all() + + job.set_enqueued() + + with self.assertLogs("odoo.addons.queue_job.job", "WARNING"): + saved = JobStore(self.env).save_state(job, (PENDING,)) + self.assertFalse(saved) + self.assertFalse(job.db_record(self.env)) diff --git a/queue_job/tests/test_queue_job_protected_write.py b/queue_job/tests/test_queue_job_protected_write.py index eadb16ab9..0896e95bc 100644 --- a/queue_job/tests/test_queue_job_protected_write.py +++ b/queue_job/tests/test_queue_job_protected_write.py @@ -32,12 +32,12 @@ def test_create_error(self): class TestJobWriteProtected(common.TransactionCase): def test_write_protected_field_error(self): job_ = self.env["res.partner"].with_delay().create({"name": "test"}) - db_job = job_.db_record() + db_job = job_.db_record(self.env) with self.assertRaises(exceptions.AccessError): db_job.method_name = "unlink" def test_write_allow_no_protected_field_error(self): job_ = self.env["res.partner"].with_delay().create({"name": "test"}) - db_job = job_.db_record() + db_job = job_.db_record(self.env) db_job.priority = 30 self.assertEqual(db_job.priority, 30) 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 000000000..63e5f648f --- /dev/null +++ b/queue_job/tests/test_run_job_controller.py @@ -0,0 +1,49 @@ +# 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 HttpCase, tagged +from odoo.tools import mute_logger + +from ..job import DONE, ENQUEUED, JobStore + + +@tagged("-at_install", "post_install") +class TestRunJobController(HttpCase): + """End-to-end tests of the ``/queue_job/runjob`` route""" + + def _create_enqueued_job(self, **kwargs): + job = self.env["queue.job"].with_delay()._test_job(**kwargs) + job.set_enqueued() + JobStore(self.env).save(job) + return job + + def _runjob(self, job_uuid): + response = self.url_open( + f"/queue_job/runjob?db={self.env.cr.dbname}&job_uuid={job_uuid}" + ) + response.raise_for_status() + return response + + def test_runjob(self): + job = self._create_enqueued_job() + self._runjob(job.uuid) + record = job.db_record(self.env) + # record.invalidate_recordset() + self.assertEqual(record.state, DONE) + + @mute_logger("odoo.addons.queue_job.executor") + def test_runjob_unknown_job(self): + response = self._runjob("not-existing-uuid") + self.assertEqual(response.text, "") + + @mute_logger("odoo.addons.queue_job.executor") + def test_runjob_not_enqueued_job_skip(self): + job = self.env["queue.job"].with_delay()._test_job() + JobStore(self.env).save(job) # pending, not enqueued + + self._runjob(job.uuid) + + record = job.db_record(self.env) + record.invalidate_recordset() + self.assertNotEqual(record.state, DONE) + self.assertNotEqual(record.state, ENQUEUED) diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_rob_controller.py deleted file mode 100644 index 1a15f4363..000000000 --- a/queue_job/tests/test_run_rob_controller.py +++ /dev/null @@ -1,23 +0,0 @@ -# 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"} - ) - - 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")