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

queue_job/controllers/main.py

@@ -147,13 +147,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 +161,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
 
@@ -513,7 +513,7 @@ def perform(self):
             if self.job_config.allow_commit:
                 env_context_manager = self.in_temporary_env()
             else:
-                env_context_manager = nullcontext()
+                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,93 @@ 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)
 
     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"),
         ):
-            mocked_temp_env.return_value.__enter__.return_value = self.env
             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"),
+        ):
+            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)