[18.0][IMP] queue_job: do not use separate cursor to write error result

queue_job/controllers/main.py

@@ -99,9 +99,6 @@ def _try_perform_job(cls, env, job):
         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()
@@ -147,13 +144,6 @@ def _enqueue_dependent_jobs(cls, env, 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)
@@ -168,26 +158,27 @@ def retry_postpone(job, message, seconds=None):
 
         except RetryableJobError as err:
             # delay the job later, requeue
-            retry_postpone(job, str(err), seconds=err.seconds)
+            job.env.clear()
+            job.postpone(result=str(err), seconds=err.seconds)
+            job.set_pending(reset_retry=False)
+            job.store()
             _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()
+            with StringIO() as buff:
+                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()
-                job.on_fail(vals)
-                buff.close()
+            vals = cls._get_failure_values(job, traceback_txt, orig_exception)
+            job.set_failed(**vals)
+            job.store()
+            job.on_fail(vals)
+            env.cr.commit()
             raise
 
         cls._enqueue_dependent_jobs(env, job)

queue_job/job.py

@@ -8,7 +8,7 @@
 import sys
 import uuid
 import weakref
-from contextlib import contextmanager, nullcontext
+from contextlib import contextmanager
 from datetime import datetime, timedelta
 from random import randint
 
@@ -518,7 +518,12 @@ def perform(self):
             if self.job_config.allow_commit:
                 env_context_manager = self.in_temporary_env()
             else:
-                env_context_manager = nullcontext()
+                # On failure, it rolls back the failed transaction to the savepoint.
+                # It frees locks and restores the connection to a usable state for
+                # failure handling (e.g. on_fail hook, using the same cursor)
+                # On success, the savepoint flushes pending writes, before 'set_done',
+                # so that the stored computed fields are part of the 'exec_time'.
+                env_context_manager = self.env.cr.savepoint()
             with env_context_manager:
                 self.result = self.func(*tuple(self.args), **self.kwargs)
         except RetryableJobError as err:

queue_job/tests/test_run_job_controller.py

@@ -1,4 +1,5 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+from datetime import datetime
 from unittest.mock import patch
 
 from odoo.tests.common import TransactionCase
@@ -30,6 +31,8 @@ def test_get_failure_values(self):
     def test_runjob_success(self):
         job = self.env["queue.job"].with_delay()._test_job()
         RunJobController._runjob(self.env, job)
+        self.assertEqual(job.retry, 1)
+        self.assertEqual(job.max_retries, 5)
         self.assertEqual(job.state, "done")
         self.assertEqual(job.db_record().state, "done")
 
@@ -41,22 +44,136 @@ def test_runjob_on_fail(self):
             patch(
                 "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail"
             ) as mocked_hook,
-            patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env,
+            patch.object(self.env.cr, "commit") as mocked_commit,
             self.assertRaises(JobError),
             mute_logger("odoo.addons.queue_job.controllers.main"),
         ):
-            mocked_temp_env.return_value.__enter__.return_value = self.env
             RunJobController._runjob(self.env, job)
         self.assertEqual(job.state, "failed")
+        self.assertEqual(mocked_commit.call_count, 1)
+        self.assertEqual(mocked_hook.call_count, 1)
+
+    @mute_logger("odoo.addons.queue_job.controllers.main")
+    def test_runjob_on_fail_no_deadlock(self):
+        """The on fail hook can write on the records the failed job wrote.
+
+        The hook used to run on another cursor, it's not the case anymore.
+        """
+        function = self.env.ref("queue_job.job_function_queue_job__test_job")
+        function.on_fail_method = "_test_on_fail"
+        # Use a committed record, so that the other cursor can see it
+        partner = self.env.ref("base.partner_admin")
+        job = self.env["queue.job"].with_delay()._test_job()
+
+        # The job writes on the partner (locking it), then fails
+        def failing_job():
+            partner.name = "Written by job"
+            partner.flush_recordset()
+            raise JobError("Job failed")
+
+        # The hook writes on the same partner. Its changes are rolled back on close,
+        # and the lock timeout avoids waiting forever.
+        def on_fail(**kwargs):
+            self.env.cr.execute("SET LOCAL lock_timeout = '2s'")
+            self.env.cr.execute(
+                "UPDATE res_partner SET name = 'Written by hook' WHERE id = %s",
+                [partner.id],
+            )
+
+        with (
+            patch.object(self.env.cr, "commit") as mocked_commit,
+            self.assertRaises(JobError),
+            patch(
+                "odoo.addons.queue_job.models.queue_job.QueueJob._test_job",
+                side_effect=failing_job,
+            ),
+            patch(
+                "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail",
+                side_effect=on_fail,
+            ) as mocked_hook,
+        ):
+            RunJobController._runjob(self.env, job)
+        self.assertEqual(mocked_commit.call_count, 1)
         self.assertEqual(mocked_hook.call_count, 1)
 
     def test_runjob_on_fail_not_configured(self):
         job = self.env["queue.job"].with_delay()._test_job(failure_rate=1)
         with (
-            patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env,
+            patch.object(self.env.cr, "commit") as mocked_commit,
+            self.assertRaises(JobError),
+            mute_logger("odoo.addons.queue_job.controllers.main"),
+        ):
+            RunJobController._runjob(self.env, job)
+        self.assertEqual(job.retry, 1)
+        self.assertEqual(job.eta, None)
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(mocked_commit.call_count, 1)
+
+    def test_runjob_retry_then_fail(self):
+        retry_s = 8
+        job = (
+            self.env["queue.job"]
+            .with_delay()
+            ._test_job(failure_rate=1, failure_retry_seconds=retry_s)
+        )
+        # Try once
+        with mute_logger("odoo.addons.queue_job.controllers.main"):
+            RunJobController._runjob(self.env, job)
+        eta_seconds = (job.eta - datetime.now()).total_seconds()
+        self.assertEqual(job.state, "pending")
+        self.assertEqual(job.retry, 1)
+        self.assertEqual(job.max_retries, 5)
+        self.assertAlmostEqual(eta_seconds, retry_s, places=0)
+
+        # Retry 3 more times
+        with mute_logger("odoo.addons.queue_job.controllers.main"):
+            RunJobController._runjob(self.env, job)
+            RunJobController._runjob(self.env, job)
+            RunJobController._runjob(self.env, job)
+        eta_seconds = (job.eta - datetime.now()).total_seconds()
+        self.assertEqual(job.retry, 4)
+        self.assertEqual(job.state, "pending")
+        self.assertEqual(job.db_record().state, "pending")
+        self.assertAlmostEqual(eta_seconds, retry_s, places=0)
+
+        # And the 5th time, it fails
+        with (
+            patch.object(self.env.cr, "commit") as mocked_commit,
             self.assertRaises(JobError),
             mute_logger("odoo.addons.queue_job.controllers.main"),
         ):
-            mocked_temp_env.return_value.__enter__.return_value = self.env
             RunJobController._runjob(self.env, job)
+        eta_seconds = (job.eta - datetime.now()).total_seconds()
+        self.assertEqual(job.retry, 5)
         self.assertEqual(job.state, "failed")
+        self.assertAlmostEqual(eta_seconds, retry_s, places=0)
+        self.assertEqual(mocked_commit.call_count, 1)
+
+    def test_runjob_retry_then_succeed(self):
+        retry_s = 8
+        job = (
+            self.env["queue.job"]
+            .with_delay()
+            ._test_job(failure_rate=1, failure_retry_seconds=retry_s)
+        )
+        # Try 4 times
+        with mute_logger("odoo.addons.queue_job.controllers.main"):
+            RunJobController._runjob(self.env, job)
+            RunJobController._runjob(self.env, job)
+            RunJobController._runjob(self.env, job)
+            RunJobController._runjob(self.env, job)
+        eta_seconds = (job.eta - datetime.now()).total_seconds()
+        self.assertEqual(job.retry, 4)
+        self.assertEqual(job.max_retries, 5)
+        self.assertEqual(job.state, "pending")
+        self.assertAlmostEqual(eta_seconds, retry_s, places=0)
+
+        # Then it succeeds, at last
+        job.kwargs = {**job.kwargs, "failure_rate": 0}
+        with mute_logger("odoo.addons.queue_job.controllers.main"):
+            RunJobController._runjob(self.env, job)
+        eta_seconds = (job.eta - datetime.now()).total_seconds()
+        self.assertEqual(job.retry, 5)
+        self.assertEqual(job.state, "done")
+        self.assertEqual(job.db_record().state, "done")
+        self.assertAlmostEqual(eta_seconds, retry_s, places=0)