[IMP] queue_job: run jobs in cron workers

Author: sbidoul

queue_job/__manifest__.py

@@ -20,6 +20,7 @@
         "wizards/queue_requeue_job_views.xml",
         "views/queue_job_menus.xml",
         "data/queue_data.xml",
+        "data/queue_job_executor_cron.xml",
         "data/queue_job_function_data.xml",
     ],
     "assets": {

queue_job/controllers/main.py

@@ -161,6 +161,7 @@ def retry_postpone(job, message, seconds=None):
 
         _logger.debug("%s enqueue depends started", job)
         cls._enqueue_dependent_jobs(env, job)
+        env.cr.commit()
         _logger.debug("%s enqueue depends done", job)
 
     @classmethod
@@ -185,13 +186,26 @@ def _get_failure_values(cls, job, traceback_txt, orig_exception):
         save_session=False,
         readonly=False,
     )
-    def runjob(self, db, job_uuid, **kw):
+    def runjob(self, db: str, job_uuid: str | None, **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)
+        run_as = env["ir.config_parameter"].get_param("queue_job.run_as")
+        if run_as == "cron":
+            crons = env["ir.cron"].search(
+                env["queue.job.executor"]._executor_cron_domain()
+            )
+            assert crons, "No queue_job executor cron found"
+            for cron in crons:
+                # TODO Awaking all of them is a bit wasteful although not very
+                # costly. Ideally we should awaken only one that is not already
+                # running.
+                cron._trigger()
+        else:
+            # Run in this http worker
+            job = self._acquire_job(env, job_uuid)
+            if not job:
+                return ""
+            self._runjob(env, job)
         return ""
 
     # flake8: noqa: C901

queue_job/data/queue_job_executor_cron.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo>
+    <record id="queue_job_executor_cron" model="ir.cron">
+        <field name="name">Queue Job Executor</field>
+        <field name="model_id" ref="model_queue_job_executor" />
+        <field name="state">code</field>
+        <field name="code">model._execute_ready_jobs()</field>
+        <field name='interval_number'>1</field>
+        <field name='interval_type'>hours</field>
+    </record>
+</odoo>

queue_job/models/__init__.py

@@ -2,5 +2,6 @@
 from . import ir_model_fields
 from . import queue_job
 from . import queue_job_channel
+from . import queue_job_executor
 from . import queue_job_function
 from . import queue_job_lock

queue_job/models/queue_job_executor.py

@@ -0,0 +1,60 @@
+# Copyright (c) 2026 ACSONE SA/NV (<http://acsone.eu>)
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
+
+import logging
+
+from odoo import api, models
+
+from ..controllers.main import RunJobController
+from ..job import Job
+
+_logger = logging.getLogger(__name__)
+
+
+class QueueJobExecutor(models.AbstractModel):
+    _name = "queue.job.executor"
+    _description = "Queue Job Executor"
+
+    @api.model
+    def _executor_cron_domain(self) -> list:
+        model_id = self.env["ir.model"]._get("queue.job.executor").id
+        return [
+            ("model_id", "=", model_id),
+            ("state", "=", "code"),
+            ("code", "=", "model._execute_ready_jobs()"),
+        ]
+
+    @api.model
+    def _ensure_executor_crons(self, capacity: int) -> None:
+        """Since Odoo cron can't run cron jobs in parallel, we create several.
+
+        `capacity` should be equal to the root channel capacity. If it's more,
+        it's wasteful. If it's less, job will stay in ENQUEUED state longer than
+        needed and loop back to PENDING due to the dead jobs requeuer.
+        """
+        if capacity < 1:
+            return
+        ref_cron = self.env.ref("queue_job.queue_job_executor_cron")
+        ref_cron.active = True
+        # remove clones
+        self.env["ir.cron"].with_context(active_test=False).search(
+            self._executor_cron_domain() + [("id", "!=", ref_cron.id)]
+        ).unlink()
+        # re-create desired number of clones
+        for _i in range(1, capacity):
+            ref_cron.copy()
+
+    @api.model
+    def _enable_executor_cron(self, capacity: int) -> None:
+        self._ensure_executor_crons(capacity)
+        self.env["ir.config_parameter"].set_param("queue_job.run_as", "cron")
+
+    @api.model
+    def _execute_job(self, job: Job) -> None:
+        RunJobController._runjob(self.env, job)
+
+    @api.model
+    def _execute_ready_jobs(self) -> None:
+        while job := RunJobController._acquire_job(self.env):
+            _logger.debug("executor cron running queue job %s", job.uuid)
+            self._execute_job(job)