[IMP] queue_job: prevent commit during queue job execution

This would release the job lock, causing spurious restarts by the dead jobs requeuer.

Author: sbidoul

queue_job/controllers/main.py

@@ -6,6 +6,7 @@
 import random
 import time
 import traceback
+from contextlib import contextmanager
 from io import StringIO
 
 from psycopg2 import OperationalError, errorcodes
@@ -26,6 +27,29 @@
 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. "
+            "If the current job is a cron running as queue job, "
+            "modify it to run as a normal cron."
+        )
+
+    original_commit = cr.commit
+    cr.commit = forbidden_commit
+    try:
+        yield
+    finally:
+        cr.commit = original_commit
+
+
 class RunJobController(http.Controller):
     @classmethod
     def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None:
@@ -69,13 +93,16 @@ def _acquire_job(cls, env: api.Environment, job_uuid: str) -> Job | None:
     def _try_perform_job(cls, env, job):
         """Try to perform the job, mark it done and commit if successful."""
         _logger.debug("%s started", job)
-        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()
+        # 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()
         env.cr.commit()
         _logger.debug("%s done", job)
 
@@ -201,6 +228,7 @@ def create_test_job(
         size=1,
         failure_rate=0,
         job_duration=0,
+        commit_within_job=False,
     ):
         if not http.request.env.user.has_group("base.group_erp_manager"):
             raise Forbidden(http.request.env._("Access Denied"))
@@ -246,6 +274,7 @@ def create_test_job(
                 description=description,
                 failure_rate=failure_rate,
                 job_duration=job_duration,
+                commit_within_job=commit_within_job,
             )
 
         if size > 1:
@@ -257,6 +286,7 @@ def create_test_job(
                 description=description,
                 failure_rate=failure_rate,
                 job_duration=job_duration,
+                commit_within_job=commit_within_job,
             )
         return ""
 
@@ -269,6 +299,7 @@ def _create_single_test_job(
         size=1,
         failure_rate=0,
         job_duration=0,
+        commit_within_job=False,
     ):
         delayed = (
             http.request.env["queue.job"]
@@ -278,7 +309,11 @@ def _create_single_test_job(
                 channel=channel,
                 description=description,
             )
-            ._test_job(failure_rate=failure_rate, job_duration=job_duration)
+            ._test_job(
+                failure_rate=failure_rate,
+                job_duration=job_duration,
+                commit_within_job=commit_within_job,
+            )
         )
         return f"job uuid: {delayed.db_record().uuid}"
 
@@ -293,6 +328,7 @@ def _create_graph_test_jobs(
         description="Test job",
         failure_rate=0,
         job_duration=0,
+        commit_within_job=False,
     ):
         model = http.request.env["queue.job"]
         current_count = 0
@@ -315,7 +351,11 @@ def _create_graph_test_jobs(
                         max_retries=max_retries,
                         channel=channel,
                         description=f"{description} #{current_count}",
-                    )._test_job(failure_rate=failure_rate, job_duration=job_duration)
+                    )._test_job(
+                        failure_rate=failure_rate,
+                        job_duration=job_duration,
+                        commit_within_job=commit_within_job,
+                    )
                 )
 
             grouping = random.choice(possible_grouping_methods)

queue_job/models/queue_job.py

@@ -458,9 +458,11 @@ def related_action_open_record(self):
             )
         return action
 
-    def _test_job(self, failure_rate=0, job_duration=0):
+    def _test_job(self, failure_rate=0, job_duration=0, commit_within_job=False):
         _logger.info("Running test job.")
         if random.random() <= failure_rate:
             raise JobError("Job failed")
         if job_duration:
             time.sleep(job_duration)
+        if commit_within_job:
+            self.env.cr.commit()