diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index b31b7a02b7..67fa347e83 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -13,11 +13,7 @@ from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY from ..job import Job, ENQUEUED -from ..exception import (NoSuchJobError, - NotReadableJobError, - RetryableJobError, - FailedJobError, - NothingToDoJob) +from ..exception import (RetryableJobError, FailedJobError, NothingToDoJob) _logger = logging.getLogger(__name__) @@ -26,42 +22,17 @@ class RunJobController(http.Controller): - def _load_job(self, env, job_uuid): - """Reload a job from the backend""" - try: - job = Job.load(env, job_uuid) - except NoSuchJobError: - # just skip it - job = None - except NotReadableJobError: - _logger.exception('Could not read job: %s', job_uuid) - raise - return job - def _try_perform_job(self, env, job): """Try to perform the job.""" - - # if the job has been manually set to DONE or PENDING, - # or if something tries to run a job that is not enqueued - # before its execution, stop - if job.state != ENQUEUED: - _logger.warning('job %s is in state %s ' - 'instead of enqueued in /runjob', - job.uuid, job.state) - return - - # TODO: set_started should be done atomically with - # update queue_job set=state=started - # where state=enqueid and id= job.set_started() job.store() - http.request.env.cr.commit() - + env.cr.commit() _logger.debug('%s started', job) + job.perform() job.set_done() job.store() - http.request.env.cr.commit() + env.cr.commit() _logger.debug('%s done', job) @http.route('/queue_job/session', type='http', auth="none") @@ -87,10 +58,23 @@ def retry_postpone(job, message, seconds=None): job.store() env.cr.commit() - job = self._load_job(env, job_uuid) - if job is None: + # ensure the job to run is in the correct state and lock the record + env.cr.execute( + "SELECT state FROM queue_job " + "WHERE uuid=%s AND state=%s " + "FOR UPDATE", + (job_uuid, ENQUEUED) + ) + if not env.cr.fetchone(): + _logger.warn( + "was requested to run job %s, but it does not exist, " + "or is not in state %s", + job_uuid, ENQUEUED + ) return "" - env.cr.commit() + + job = Job.load(env, job_uuid) + assert job and job.state == ENQUEUED try: try: diff --git a/queue_job/exception.py b/queue_job/exception.py index a3e7db0066..2a6e5ac6a2 100644 --- a/queue_job/exception.py +++ b/queue_job/exception.py @@ -14,10 +14,6 @@ class NoSuchJobError(JobError): """The job does not exist.""" -class NotReadableJobError(JobError): - """The job cannot be read from the storage.""" - - class FailedJobError(JobError): """A job had an error having to be resolved.""" diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 4380316e2d..f0d7b9b863 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -224,8 +224,14 @@ def set_job_pending(): cr.execute( "UPDATE queue_job SET state=%s, " "date_enqueued=NULL, date_started=NULL " - "WHERE uuid=%s and state=%s", (PENDING, job_uuid, ENQUEUED) + "WHERE uuid=%s and state=%s " + "RETURNING uuid", (PENDING, job_uuid, ENQUEUED) ) + if cr.fetchone(): + _logger.warning( + "state of job %s was reset from %s to %s", + job_uuid, ENQUEUED, PENDING, + ) # TODO: better way to HTTP GET asynchronously (grequest, ...)? # if this was python3 I would be doing this with