[FIX] queue_job: apply max_retries to PostgreSQL concurrency errors

queue_job/controllers/main.py

@@ -18,11 +18,12 @@
 
 from ..delay import chain, group
 from ..exception import FailedJobError, RetryableJobError
-from ..job import ENQUEUED, Job
 
-_logger = logging.getLogger(__name__)
+# PG_RETRY now lives in job.py, and is re-exported here so that existing
+# imports from this module keep working.
+from ..job import ENQUEUED, PG_RETRY, Job  # noqa: F401
 
-PG_RETRY = 5  # seconds
+_logger = logging.getLogger(__name__)
 
 DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5
 
@@ -159,16 +160,10 @@ def retry_postpone(job, message, seconds=None):
                 job.store()
 
         try:
-            try:
-                cls._try_perform_job(env, job)
-            except OperationalError as err:
-                # Automatically retry the typical transaction serialization
-                # errors
-                if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
-                    raise
-
-                _logger.debug("%s OperationalError, postponed", job)
-                raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err
+            # Job.perform() turns the retryable PostgreSQL concurrency errors
+            # into RetryableJobError itself, so that max_retries applies to
+            # them; there is nothing to convert here.
+            cls._try_perform_job(env, job)
 
         except RetryableJobError as err:
             # delay the job later, requeue

queue_job/job.py

@@ -12,8 +12,11 @@
 from datetime import datetime, timedelta
 from random import randint
 
+from psycopg2 import OperationalError
+
 import odoo
 from odoo.models import BaseModel
+from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY
 
 from .exception import FailedJobError, NoSuchJobError, RetryableJobError
 
@@ -57,6 +60,7 @@ def _rebind_to_cr(value, cr):
 
 DEFAULT_PRIORITY = 10  # used by the PriorityQueue to sort the jobs
 DEFAULT_MAX_RETRIES = 5
+PG_RETRY = 5  # seconds, when the job function defines no retry_pattern
 RETRY_INTERVAL = 10 * 60  # seconds
 
 _logger = logging.getLogger(__name__)
@@ -511,12 +515,27 @@ def perform(self):
         """
         self.retry += 1
         try:
-            if self.job_config.allow_commit:
-                env_context_manager = self.in_temporary_env()
-            else:
-                env_context_manager = nullcontext()
-            with env_context_manager:
-                self.result = self.func(*tuple(self.args), **self.kwargs)
+            try:
+                if self.job_config.allow_commit:
+                    env_context_manager = self.in_temporary_env()
+                else:
+                    env_context_manager = nullcontext()
+                with env_context_manager:
+                    self.result = self.func(*tuple(self.args), **self.kwargs)
+            except OperationalError as err:
+                # The typical transaction serialization errors are retryable,
+                # and converting them here rather than in the runners is what
+                # subjects them to max_retries below: a job raising them from
+                # outside perform() is postponed without ever consulting the
+                # limit, so it retries for ever.
+                if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
+                    raise
+                # A job function that declares a retry_pattern has said how it
+                # wants to be retried, so leave the delay to it.
+                raise RetryableJobError(
+                    err.pgerror,
+                    seconds=None if self.job_config.retry_pattern else PG_RETRY,
+                ) from err
         except RetryableJobError as err:
             if err.ignore_retry:
                 self.retry -= 1

queue_job_cron_jobrunner/models/queue_job.py

@@ -6,12 +6,8 @@
 import traceback
 from io import StringIO
 
-from psycopg2 import OperationalError
+from odoo import api, models
 
-from odoo import api, models, tools
-from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY
-
-from odoo.addons.queue_job.controllers.main import PG_RETRY
 from odoo.addons.queue_job.exception import FailedJobError, RetryableJobError
 from odoo.addons.queue_job.job import Job
 
@@ -60,24 +56,17 @@ def _process(self, commit=False):
 
         # Actual processing
         try:
-            try:
-                with self.env.cr.savepoint():
-                    job.perform()
-                    job.set_done()
-                    job.store()
-            except OperationalError as err:
-                # Automatically retry the typical transaction serialization errors
-                if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
-                    raise
-                message = tools.ustr(err.pgerror, errors="replace")
-                job.postpone(result=message, seconds=PG_RETRY)
-                job.set_pending(reset_retry=False)
+            # Job.perform() turns the retryable PostgreSQL concurrency errors
+            # into RetryableJobError itself, so that max_retries applies to
+            # them; they arrive below with everything else.
+            with self.env.cr.savepoint():
+                job.perform()
+                job.set_done()
                 job.store()
-                _logger.debug("%s OperationalError, postponed", job)
 
         except RetryableJobError as err:
             # delay the job later, requeue
-            job.postpone(result=str(err), seconds=5)
+            job.postpone(result=str(err), seconds=err.seconds)
             job.set_pending(reset_retry=False)
             job.store()
             _logger.debug("%s postponed", job)

test_queue_job/data/queue_job_function_data.xml

@@ -20,6 +20,14 @@
         <field name="method">job_with_retry_pattern__no_zero</field>
         <field name="retry_pattern" eval="{3: 180}" />
     </record>
+    <record
+        id="job_function_test_queue_job_job_with_retry_pattern__pg_concurrency"
+        model="queue.job.function"
+    >
+        <field name="model_id" ref="test_queue_job.model_test_queue_job" />
+        <field name="method">job_with_retry_pattern__pg_concurrency</field>
+        <field name="retry_pattern" eval="{1: 60}" />
+    </record>
     <record
         id="job_function_test_queue_job_job_commit_with_arg_records"
         model="queue.job.function"

test_queue_job/models/test_models.py

@@ -4,13 +4,31 @@
 import functools
 from datetime import datetime, timedelta
 
+from psycopg2 import OperationalError, errorcodes
+
 from odoo import api, fields, models
 
 from odoo.addons.queue_job.delay import chain
 from odoo.addons.queue_job.exception import RetryableJobError
 from odoo.addons.queue_job.job import identity_exact
 
 
+class ConcurrencyError(OperationalError):
+    """A serialization failure, as psycopg2 raises it.
+
+    psycopg2 exposes pgcode read-only and errors.lookup() leaves it unset, so
+    the code has to be carried on a subclass.
+    """
+
+    @property
+    def pgcode(self):
+        return errorcodes.SERIALIZATION_FAILURE
+
+    @property
+    def pgerror(self):
+        return "could not serialize access due to concurrent update"
+
+
 class QueueJob(models.Model):
     _inherit = "queue.job"
 
@@ -79,6 +97,8 @@ def testing_method(self, *args, **kwargs):
         """
         if kwargs.get("raise_retry"):
             raise RetryableJobError("Must be retried later")
+        if kwargs.get("raise_pg_concurrency"):
+            raise ConcurrencyError()
         if kwargs.get("return_context"):
             return self.env.context
         return args, kwargs
@@ -105,6 +125,9 @@ def job_with_retry_pattern(self):
     def job_with_retry_pattern__no_zero(self):
         return
 
+    def job_with_retry_pattern__pg_concurrency(self):
+        raise ConcurrencyError()
+
     def mapped(self, func):
         return super().mapped(func)
 

test_queue_job/tests/test_job.py

@@ -5,6 +5,8 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
+from psycopg2 import OperationalError
+
 import odoo.tests.common as common
 from odoo import Command
 
@@ -22,6 +24,7 @@
     ENQUEUED,
     FAILED,
     PENDING,
+    PG_RETRY,
     RETRY_INTERVAL,
     STARTED,
     WAIT_DEPENDENCIES,
@@ -109,6 +112,55 @@ def test_retryable_error(self):
             test_job.perform()
         self.assertEqual(test_job.retry, 3)
 
+    def test_pg_concurrency_error_is_retryable_and_capped(self):
+        """A serialization failure is retried, and gives up at max_retries.
+
+        The runners used to convert these outside Job.perform(), which is the
+        only place max_retries is consulted, so such a job was postponed for
+        ever while its retry counter climbed past the limit.
+        """
+        test_job = Job(
+            self.method, kwargs={"raise_pg_concurrency": True}, max_retries=3
+        )
+        self.assertEqual(test_job.retry, 0)
+        with self.assertRaises(RetryableJobError):
+            test_job.perform()
+        self.assertEqual(test_job.retry, 1)
+        with self.assertRaises(RetryableJobError):
+            test_job.perform()
+        self.assertEqual(test_job.retry, 2)
+        with self.assertRaises(FailedJobError):
+            test_job.perform()
+        self.assertEqual(test_job.retry, 3)
+
+    def test_pg_concurrency_error_waits_pg_retry_without_a_pattern(self):
+        test_job = Job(
+            self.method, kwargs={"raise_pg_concurrency": True}, max_retries=3
+        )
+        with self.assertRaises(RetryableJobError) as caught:
+            test_job.perform()
+        seconds = test_job._get_retry_seconds(caught.exception.seconds)
+        self.assertEqual(seconds, PG_RETRY)
+
+    def test_pg_concurrency_error_follows_the_jobs_retry_pattern(self):
+        """A job function declaring a retry_pattern gets its own delay."""
+        method = self.env["test.queue.job"].job_with_retry_pattern__pg_concurrency
+        test_job = Job(method, max_retries=3)
+        with self.assertRaises(RetryableJobError) as caught:
+            test_job.perform()
+        self.assertIsNone(caught.exception.seconds)
+        seconds = test_job._get_retry_seconds(caught.exception.seconds)
+        self.assertEqual(seconds, 60)
+
+    def test_other_operational_error_is_not_retried(self):
+        error = OperationalError("undefined table")
+        test_job = Job(self.method, max_retries=3)
+        with mock.patch.object(
+            type(self.env["test.queue.job"]), "testing_method", side_effect=error
+        ):
+            with self.assertRaises(OperationalError):
+                test_job.perform()
+
     def test_infinite_retryable_error(self):
         test_job = Job(self.method, kwargs={"raise_retry": True}, max_retries=0)
         self.assertEqual(test_job.retry, 0)