Add 'Run in new cursor' option on job functions

It is forbidden to commit inside a job, because it releases the job lock
and can cause it to start again, while still being run, by the dead jobs
requeuer. For some use cases, it may actually be legitimate, or at least
be needed in the short term before actual updates in the code.

A new option on the job function, false by default, allow to run the job
in a new transaction, at the cost of an additional connection +
transaction overhead.

Related to #889

Author: guewen

queue_job/controllers/main.py

@@ -38,8 +38,9 @@ def _prevent_commit(cr):
     def forbidden_commit(*args, **kwargs):
         raise RuntimeError(
             "Commit is forbidden in queue jobs. "
-            "If the current job is a cron running as queue job, "
-            "modify it to run as a normal cron."
+            'You may want to enable the "Run in new cursor" 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."
         )
 
     original_commit = cr.commit

queue_job/job.py

@@ -8,6 +8,7 @@
 import sys
 import uuid
 import weakref
+from contextlib import contextmanager, nullcontext
 from datetime import datetime, timedelta
 from random import randint
 
@@ -406,10 +407,6 @@ def __init__(
         self.method_name = func.__name__
         self.recordset = recordset
 
-        self.env = env
-        self.job_model = self.env["queue.job"]
-        self.job_model_name = "queue.job"
-
         self.job_config = (
             self.env["queue.job.function"].sudo().job_config(self.job_function_name)
         )
@@ -487,7 +484,12 @@ def perform(self):
         """
         self.retry += 1
         try:
-            self.result = self.func(*tuple(self.args), **self.kwargs)
+            if self.job_config.run_in_new_cursor:
+                env_context_manager = self._with_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
@@ -507,6 +509,14 @@ def perform(self):
 
         return self.result
 
+    @contextmanager
+    def _with_temporary_env(self):
+        with self.env.registry.cursor() as new_cr:
+            env = self.recordset.env
+            self.recordset = self.recordset.with_env(env(cr=new_cr))
+            yield
+            self.recordset = self.recordset.with_env(env)
+
     def _get_common_dependent_jobs_query(self):
         return """
             UPDATE queue_job
@@ -665,6 +675,10 @@ def __hash__(self):
     def db_record(self):
         return self.db_records_from_uuids(self.env, [self.uuid])
 
+    @property
+    def env(self):
+        return self.recordset.env
+
     @property
     def func(self):
         recordset = self.recordset.with_context(job_uuid=self.uuid)

queue_job/models/queue_job_function.py

@@ -28,7 +28,8 @@ class QueueJobFunction(models.Model):
         "related_action_enable "
         "related_action_func_name "
         "related_action_kwargs "
-        "job_function_id ",
+        "job_function_id "
+        "run_in_new_cursor",
     )
 
     def _default_channel(self):
@@ -79,6 +80,13 @@ def _default_channel(self):
         "enable, func_name, kwargs.\n"
         "See the module description for details.",
     )
+    run_in_new_cursor = fields.Boolean(
+        string="Run in new cursor",
+        help="When enabled, the job will be executed in a new database cursor. "
+        "This is necessary when the job function needs to commit the transaction "
+        "during its execution, at the cost of an additional database connection "
+        "and transaction management overhead.",
+    )
 
     @api.depends("model_id.model", "method")
     def _compute_name(self):
@@ -149,6 +157,7 @@ def job_default_config(self):
             related_action_func_name=None,
             related_action_kwargs={},
             job_function_id=None,
+            run_in_new_cursor=False,
         )
 
     def _parse_retry_pattern(self):
@@ -184,6 +193,7 @@ def job_config(self, name):
             related_action_func_name=config.related_action.get("func_name"),
             related_action_kwargs=config.related_action.get("kwargs", {}),
             job_function_id=config.id,
+            run_in_new_cursor=config.run_in_new_cursor,
         )
 
     def _retry_pattern_format_error_message(self):

queue_job/tests/common.py

@@ -276,7 +276,7 @@ def _add_job(self, *args, **kwargs):
 
     def _prepare_context(self, job):
         # pylint: disable=context-overridden
-        job_model = job.job_model.with_context({})
+        job_model = job.env["queue.job"].with_context({})
         field_records = job_model._fields["records"]
         # Filter the context to simulate store/load of the job
         job.recordset = field_records.convert_to_write(job.recordset, job_model)

queue_job/tests/test_model_job_function.py

@@ -42,6 +42,7 @@ def test_function_job_config(self):
                     ' "func_name": "related_action_foo",'
                     ' "kwargs": {"b": 1}}'
                 ),
+                "run_in_new_cursor": True,
             }
         )
         self.assertEqual(
@@ -53,5 +54,6 @@ def test_function_job_config(self):
                 related_action_func_name="related_action_foo",
                 related_action_kwargs={"b": 1},
                 job_function_id=job_function.id,
+                run_in_new_cursor=True,
             ),
         )

queue_job/views/queue_job_function_views.xml

@@ -10,6 +10,7 @@
                     <field name="model_id" required="1" />
                     <field name="method" required="1" />
                     <field name="channel_id" />
+                    <field name="run_in_new_cursor" />
                     <field name="edit_retry_pattern" widget="ace" />
                     <field name="edit_related_action" widget="ace" />
                 </group>
@@ -24,6 +25,7 @@
             <list>
                 <field name="name" />
                 <field name="channel_id" />
+                <field name="run_in_new_cursor" />
             </list>
         </field>
     </record>