diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 706c02be79..b3a26e93f0 100644 --- a/queue_job/controllers/main.py +++ b/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 diff --git a/queue_job/job.py b/queue_job/job.py index 04081976c5..517a8e086c 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, 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 diff --git a/queue_job_cron_jobrunner/models/queue_job.py b/queue_job_cron_jobrunner/models/queue_job.py index 3213d28d0b..533c14091f 100644 --- a/queue_job_cron_jobrunner/models/queue_job.py +++ b/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) 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 + +