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

queue_job/controllers/main.py

@@ -98,14 +98,7 @@ def _try_perform_job(cls, env, job):
         # TODO refactor, the relation between env and job.env is not clear
         assert env.cr is job.env.cr
         with _prevent_commit(env.cr):
-            # On failure, rolling back the savepoint releases the row locks
-            # the job took on the records it updated: the failure is handled
-            # from another cursor, which may write on the same records (e.g.
-            # the on fail hook).
-            # On success, leaving the savepoint flushes, before 'set_done', so
-            # that the stored computed fields are part of the 'exec_time'.
-            with env.cr.savepoint():
-                job.perform()
+            job.perform()
             job.set_done()
             job.store()
             env.flush_all()
@@ -151,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)
@@ -172,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
 
@@ -514,7 +514,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_rob_controller.py

@@ -1,5 +1,5 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
-from contextlib import closing
+from datetime import datetime
 from unittest.mock import patch
 
 from odoo.tests.common import TransactionCase
@@ -31,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")
 
@@ -39,25 +41,23 @@ def test_runjob_on_fail(self):
         function.on_fail_method = "_test_on_fail"
         job = self.env["queue.job"].with_delay()._test_job(failure_rate=1)
         with (
-            self.assertRaises(JobError),
             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_hook.call_count, 1)
+        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 runs on another cursor, while the failed job's transaction
-        is still open. Without rolling back the job's changes first, the hook
-        waits for the job's lock, which is only released after the hook.
+        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"
@@ -71,17 +71,17 @@ def failing_job():
             partner.flush_recordset()
             raise JobError("Job failed")
 
-        # The hook writes on the same partner from another cursor. Its changes
-        # are rolled back on close, and the lock timeout avoids waiting forever.
+        # 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):
-            with closing(self.registry.cursor()) as other_cr:
-                other_cr.execute("SET LOCAL lock_timeout = '2s'")
-                other_cr.execute(
-                    "UPDATE res_partner SET name = 'Written by hook' WHERE id = %s",
-                    [partner.id],
-                )
+            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",
@@ -91,19 +91,89 @@ def on_fail(**kwargs):
                 "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail",
                 side_effect=on_fail,
             ) as mocked_hook,
-            patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env,
         ):
-            mocked_temp_env.return_value.__enter__.return_value = self.env
             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.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),
-            patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env,
             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)