[16.0][IMP] queue_job: retry concurrency errors wrapped in other exceptions

queue_job/controllers/main.py

@@ -18,7 +18,7 @@
 from odoo.tools import config
 
 from ..delay import chain, group
-from ..exception import FailedJobError, NothingToDoJob, RetryableJobError
+from ..exception import FailedJobError, JobError, NothingToDoJob, RetryableJobError
 from ..job import ENQUEUED, Job
 
 _logger = logging.getLogger(__name__)
@@ -28,6 +28,17 @@
 DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5
 
 
+def _concurrency_error(err):
+    while err is not None and not isinstance(err, JobError):
+        if (
+            isinstance(err, OperationalError)
+            and err.pgcode in PG_CONCURRENCY_ERRORS_TO_RETRY
+        ):
+            return err
+        err = err.__cause__ or err.__context__
+    return None
+
+
 @contextmanager
 def _prevent_commit(cr):
     """Context manager to prevent commits on a cursor.
@@ -153,15 +164,16 @@ def retry_postpone(job, message, seconds=None):
         try:
             try:
                 cls._try_perform_job(env, job)
-            except OperationalError as err:
+            except Exception as err:
                 # Automatically retry the typical transaction serialization
                 # errors
-                if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
+                cause = _concurrency_error(err)
+                if cause is None:
                     raise
 
                 _logger.debug("%s OperationalError, postponed", job)
                 raise RetryableJobError(
-                    tools.ustr(err.pgerror, errors="replace"), seconds=PG_RETRY
+                    tools.ustr(cause.pgerror, errors="replace"), seconds=PG_RETRY
                 ) from err
 
         except NothingToDoJob as err:

queue_job/tests/test_run_rob_controller.py

@@ -1,11 +1,18 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 
+from psycopg2 import OperationalError, errorcodes
+
 from odoo.tests.common import TransactionCase
 
-from ..controllers.main import RunJobController
+from ..controllers.main import RunJobController, _concurrency_error
+from ..exception import RetryableJobError
 from ..job import Job
 
 
+class SerializationFailure(OperationalError):
+    pgcode = errorcodes.SERIALIZATION_FAILURE
+
+
 class TestRunJobController(TransactionCase):
     def test_get_failure_values(self):
         method = self.env["res.users"].mapped
@@ -21,3 +28,17 @@ def test_runjob_success(self):
         RunJobController._runjob(self.env, job)
         self.assertEqual(job.state, "done")
         self.assertEqual(job.db_record().state, "done")
+
+    def test_concurrency_error_wrapped_in_another_error(self):
+        original = SerializationFailure("could not serialize access")
+        wrapped = ValueError("wrapped")
+        wrapped.__context__ = original
+        self.assertIs(_concurrency_error(wrapped), original)
+
+    def test_other_errors_are_not_concurrency_errors(self):
+        self.assertIsNone(_concurrency_error(ValueError("business error")))
+
+    def test_job_errors_are_not_concurrency_errors(self):
+        error = RetryableJobError("retry later")
+        error.__context__ = SerializationFailure("could not serialize access")
+        self.assertIsNone(_concurrency_error(error))