diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 02e05a6a96..3bb49ff511 100644 --- a/queue_job/controllers/main.py +++ b/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: diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_rob_controller.py index 1a15f4363a..249060e14d 100644 --- a/queue_job/tests/test_run_rob_controller.py +++ b/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))