queue_job: extract controller's runjob as JobExecutor

Author: guewen

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