From b0b895652359a38bf45937ecf63cc9285e90ccd9 Mon Sep 17 00:00:00 2001 From: Richard deMeester Date: Tue, 13 Oct 2026 11:45:43 +0200 Subject: [PATCH 1/2] [FIX] queue_job: apply max_retries to PostgreSQL concurrency errors The typical transaction serialization errors are retried, but the conversion from OperationalError to RetryableJobError happens in the runners, outside Job.perform() - and perform() is the only place max_retries is consulted. retry_postpone() simply writes a new eta and requeues, so a job that keeps colliding is retried for ever while its retry counter climbs past the limit and it never reaches the failed state. Seen in production on a long report-building job: it sat at try 10 of a maximum of 5, each attempt running for some eight minutes, until it was closed by hand. Convert the error inside Job.perform() instead, so it falls into the RetryableJobError handling that is already there and is subject to max_retries like any other retryable error. The duplicated handling in queue_job's controller and in queue_job_cron_jobrunner then comes out, and PG_RETRY moves beside DEFAULT_MAX_RETRIES. _enqueue_dependent_jobs keeps its own OperationalError handling: that is a separate retry loop around the dependency update, not the job itself. Tests: a serialization failure is retried and gives up at max_retries, and any other OperationalError is not retried at all. --- queue_job/controllers/main.py | 16 +++------- queue_job/job.py | 26 ++++++++++++---- queue_job_cron_jobrunner/models/queue_job.py | 26 +++++----------- test_queue_job/models/test_models.py | 20 ++++++++++++ test_queue_job/tests/test_job.py | 32 ++++++++++++++++++++ 5 files changed, 83 insertions(+), 37 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 706c02be79..74a251ed94 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -22,8 +22,6 @@ _logger = logging.getLogger(__name__) -PG_RETRY = 5 # seconds - DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5 @@ -159,16 +157,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 diff --git a/queue_job/job.py b/queue_job/job.py index 04081976c5..a051a3f6aa 100644 --- a/queue_job/job.py +++ b/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 to wait before retrying a serialization failure RETRY_INTERVAL = 10 * 60 # seconds _logger = logging.getLogger(__name__) @@ -511,12 +515,22 @@ 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 + raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err except RetryableJobError as err: if err.ignore_retry: self.retry -= 1 diff --git a/queue_job_cron_jobrunner/models/queue_job.py b/queue_job_cron_jobrunner/models/queue_job.py index 3213d28d0b..0f8feed37f 100644 --- a/queue_job_cron_jobrunner/models/queue_job.py +++ b/queue_job_cron_jobrunner/models/queue_job.py @@ -6,12 +6,7 @@ import traceback from io import StringIO -from psycopg2 import OperationalError - -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 import api, models from odoo.addons.queue_job.exception import FailedJobError, RetryableJobError from odoo.addons.queue_job.job import Job @@ -60,20 +55,13 @@ 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 diff --git a/test_queue_job/models/test_models.py b/test_queue_job/models/test_models.py index fc9e43edf8..b044f4a121 100644 --- a/test_queue_job/models/test_models.py +++ b/test_queue_job/models/test_models.py @@ -4,6 +4,8 @@ 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 @@ -11,6 +13,22 @@ 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 diff --git a/test_queue_job/tests/test_job.py b/test_queue_job/tests/test_job.py index fac1caf7aa..ac41b1b4e3 100644 --- a/test_queue_job/tests/test_job.py +++ b/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 @@ -109,6 +111,36 @@ 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_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) From d0c72914eae248d5ef50e1ea48e3e3736edab794 Mon Sep 17 00:00:00 2001 From: Richard deMeester Date: Tue, 27 Oct 2026 09:01:44 +0100 Subject: [PATCH 2/2] [IMP] queue_job: let a job's retry_pattern set the serialization retry delay Follow-up to the review on OCA/queue#1009. A job that keeps hitting serialization failures was waiting 5 seconds before each retry, ignoring its specific retry_pattern. Now, we pass no delay at all when the job function declares a retry_pattern, and the pattern will apply as it does for any other RetryableJobError. Jobs without a pattern are still passed PG_RETRY and still wait 5 seconds. The constant cannot simply be dropped for them: _get_retry_seconds() would then fall back to RETRY_INTERVAL - ten minutes - for a conflict that usually clears at once. A job function that declares a retry_pattern will change behaviour: a serialization failure now waits what the pattern says. That is the point of it; the job has said how it wants to be retried. The surprise is a pattern that does not cover the early tries - {3: 180} waits RETRY_INTERVAL on tries 1 and 2, so ten minutes where it used to be five seconds. PG_RETRY is also re-exported from queue_job.controllers.main, so that imports from its old home keep working. queue_job_cron_jobrunner had a hardcoded seconds=5, discarding the delay the error asks for, so a retry_pattern would be ignored under that runner; it now passes err.seconds like the controller does. Tests: with no retry_pattern a serialization failure still waits PG_RETRY, and with one it waits what the pattern says. --- queue_job/controllers/main.py | 5 ++++- queue_job/job.py | 9 +++++++-- queue_job_cron_jobrunner/models/queue_job.py | 3 ++- .../data/queue_job_function_data.xml | 8 ++++++++ test_queue_job/models/test_models.py | 3 +++ test_queue_job/tests/test_job.py | 20 +++++++++++++++++++ 6 files changed, 44 insertions(+), 4 deletions(-) diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 74a251ed94..b3a26e93f0 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -18,7 +18,10 @@ from ..delay import chain, group from ..exception import FailedJobError, RetryableJobError -from ..job import ENQUEUED, Job + +# 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 _logger = logging.getLogger(__name__) diff --git a/queue_job/job.py b/queue_job/job.py index a051a3f6aa..517a8e086c 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -60,7 +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 to wait before retrying a serialization failure +PG_RETRY = 5 # seconds, when the job function defines no retry_pattern RETRY_INTERVAL = 10 * 60 # seconds _logger = logging.getLogger(__name__) @@ -530,7 +530,12 @@ def perform(self): # limit, so it retries for ever. if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY: raise - raise RetryableJobError(err.pgerror, seconds=PG_RETRY) from err + # 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 diff --git a/queue_job_cron_jobrunner/models/queue_job.py b/queue_job_cron_jobrunner/models/queue_job.py index 0f8feed37f..533c14091f 100644 --- a/queue_job_cron_jobrunner/models/queue_job.py +++ b/queue_job_cron_jobrunner/models/queue_job.py @@ -7,6 +7,7 @@ from io import StringIO from odoo import api, models + from odoo.addons.queue_job.exception import FailedJobError, RetryableJobError from odoo.addons.queue_job.job import Job @@ -65,7 +66,7 @@ def _process(self, commit=False): 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) diff --git a/test_queue_job/data/queue_job_function_data.xml b/test_queue_job/data/queue_job_function_data.xml index c01903b0f3..0606504348 100644 --- a/test_queue_job/data/queue_job_function_data.xml +++ b/test_queue_job/data/queue_job_function_data.xml @@ -20,6 +20,14 @@ job_with_retry_pattern__no_zero + + + job_with_retry_pattern__pg_concurrency + +