diff --git a/procrastinate/manager.py b/procrastinate/manager.py index d83e7a0df..1927884a5 100644 --- a/procrastinate/manager.py +++ b/procrastinate/manager.py @@ -148,10 +148,16 @@ async def fetch_job(self, queues: Iterable[str] | None) -> jobs.Job | None: : None if no suitable job was found. The job otherwise. """ - - row = await self.connector.execute_query_one_async( - query=sql.queries["fetch_job"], queues=queues - ) + try: + row = await self.connector.execute_query_one_async( + query=sql.queries["fetch_job"], queues=queues + ) + except exceptions.UniqueViolation as _exc: + # getting a job with lock lead to conflicts + # get any doable job instead + row = await self.connector.execute_query_one_async( + query=sql.queries["fetch_job_without_lock"], queues=queues + ) # fetch_tasks will always return a row, but is there's no relevant # value, it will all be None diff --git a/procrastinate/sql/migrations/03.00.00_02_pre_new_fetch_job_function.sql b/procrastinate/sql/migrations/03.00.00_02_pre_new_fetch_job_function.sql new file mode 100644 index 000000000..a8b55c5f6 --- /dev/null +++ b/procrastinate/sql/migrations/03.00.00_02_pre_new_fetch_job_function.sql @@ -0,0 +1,77 @@ +-- new fetch job. Only checks for doing. On update conflict return NULL +CREATE OR REPLACE FUNCTION procrastinate_fetch_job_v1(target_queue_names character varying[]) + RETURNS procrastinate.procrastinate_jobs + LANGUAGE plpgsql +AS $function$ +DECLARE + found_jobs procrastinate_jobs; + retry_count INT := 0; +BEGIN + LOOP + BEGIN + WITH candidate AS ( + SELECT jobs.* + FROM procrastinate_jobs AS jobs + WHERE + (jobs.lock IS NULL OR + NOT EXISTS ( -- reject the job if its lock has current jobs + SELECT 1 + FROM procrastinate_jobs AS jobs_with_locks + WHERE + jobs.lock IS NOT NULL + AND jobs_with_locks.lock = jobs.lock + AND jobs_with_locks.status = 'doing' + LIMIT 1 + )) + AND jobs.status = 'todo' + AND (target_queue_names IS NULL OR jobs.queue_name = ANY(target_queue_names)) + AND (jobs.scheduled_at IS NULL OR jobs.scheduled_at <= now()) + ORDER BY jobs.priority DESC, jobs.id ASC LIMIT 1 + FOR UPDATE OF jobs SKIP LOCKED + ) + UPDATE procrastinate_jobs + SET status = 'doing' + FROM candidate + WHERE procrastinate_jobs.id = candidate.id + RETURNING procrastinate_jobs.* INTO found_jobs; + + RETURN found_jobs; + EXCEPTION + WHEN unique_violation THEN + PERFORM pg_sleep(random() * 0.01); -- fuzzy retry with upto 10ms sleep + retry_count := retry_count + 1; + IF retry_count >= 3 THEN + RAISE; --reraise + END IF; + END; + END LOOP; +END; +$function$ +; + +CREATE OR REPLACE FUNCTION procrastinate_fetch_job_without_lock_v1(target_queue_names character varying[]) + RETURNS procrastinate.procrastinate_jobs + LANGUAGE plpgsql +AS $$ +DECLARE + found_jobs procrastinate_jobs; +BEGIN + WITH candidate AS ( + SELECT jobs.* + FROM procrastinate_jobs AS jobs + WHERE + jobs.lock IS NULL + AND jobs.status = 'todo' + AND (target_queue_names IS NULL OR jobs.queue_name = ANY(target_queue_names)) + AND (jobs.scheduled_at IS NULL OR jobs.scheduled_at <= now()) + ORDER BY jobs.priority DESC, jobs.id ASC LIMIT 1 + FOR UPDATE OF jobs SKIP LOCKED + ) + UPDATE procrastinate_jobs + SET status = 'doing' + FROM candidate + WHERE procrastinate_jobs.id = candidate.id + RETURNING procrastinate_jobs.* INTO found_jobs; + RETURN found_jobs; +END; +$$; diff --git a/procrastinate/sql/queries.sql b/procrastinate/sql/queries.sql index ddb1b92c2..9c35d0826 100644 --- a/procrastinate/sql/queries.sql +++ b/procrastinate/sql/queries.sql @@ -17,6 +17,11 @@ SELECT procrastinate_defer_periodic_job_v1(%(queue)s, %(lock)s, %(queueing_lock) SELECT id, status, task_name, priority, lock, queueing_lock, args, scheduled_at, queue_name, attempts FROM procrastinate_fetch_job_v1(%(queues)s::varchar[]); +-- fetch_job_without_lock -- +-- Get the first awaiting job +SELECT id, status, task_name, priority, lock, queueing_lock, args, scheduled_at, queue_name, attempts + FROM procrastinate_fetch_job_without_lock_v1(%(queues)s); + -- select_stalled_jobs -- -- Get running jobs that started more than a given time ago SELECT job.id, status, task_name, priority, lock, queueing_lock, args, scheduled_at, queue_name, attempts, max(event.at) started_at diff --git a/procrastinate/sql/schema.sql b/procrastinate/sql/schema.sql index 3160f1d0d..11eb90f34 100644 --- a/procrastinate/sql/schema.sql +++ b/procrastinate/sql/schema.sql @@ -164,34 +164,74 @@ CREATE FUNCTION procrastinate_fetch_job_v1( LANGUAGE plpgsql AS $$ DECLARE - found_jobs procrastinate_jobs; + found_jobs procrastinate_jobs; + retry_count INT := 0; BEGIN - WITH candidate AS ( - SELECT jobs.* - FROM procrastinate_jobs AS jobs - WHERE - -- reject the job if its lock has earlier jobs - NOT EXISTS ( - SELECT 1 - FROM procrastinate_jobs AS earlier_jobs + LOOP + BEGIN + WITH candidate AS ( + SELECT jobs.* + FROM procrastinate_jobs AS jobs + WHERE + (jobs.lock IS NULL OR + NOT EXISTS ( -- reject the job if its lock has current jobs + SELECT 1 + FROM procrastinate_jobs AS jobs_with_locks WHERE jobs.lock IS NOT NULL - AND earlier_jobs.lock = jobs.lock - AND earlier_jobs.status IN ('todo', 'doing') - AND earlier_jobs.id < jobs.id) - AND jobs.status = 'todo' - AND (target_queue_names IS NULL OR jobs.queue_name = ANY( target_queue_names )) - AND (jobs.scheduled_at IS NULL OR jobs.scheduled_at <= now()) - ORDER BY jobs.priority DESC, jobs.id ASC LIMIT 1 - FOR UPDATE OF jobs SKIP LOCKED + AND jobs_with_locks.lock = jobs.lock + AND jobs_with_locks.status in ('doing', 'aborting') + LIMIT 1 + )) + AND jobs.status = 'todo' + AND (target_queue_names IS NULL OR jobs.queue_name = ANY(target_queue_names)) + AND (jobs.scheduled_at IS NULL OR jobs.scheduled_at <= now()) + ORDER BY jobs.priority DESC, jobs.id ASC LIMIT 1 + FOR UPDATE OF jobs SKIP LOCKED + ) + UPDATE procrastinate_jobs + SET status = 'doing' + FROM candidate + WHERE procrastinate_jobs.id = candidate.id + RETURNING procrastinate_jobs.* INTO found_jobs; + + RETURN found_jobs; + EXCEPTION + WHEN unique_violation THEN + PERFORM pg_sleep(random() * 0.01); -- fuzzy retry with upto 10ms sleep + retry_count := retry_count + 1; + IF retry_count >= 3 THEN + RAISE; --reraise + END IF; + END; + END LOOP; +END; +$$; + +CREATE FUNCTION procrastinate_fetch_job_without_lock_v1(target_queue_names character varying[]) + RETURNS procrastinate.procrastinate_jobs + LANGUAGE plpgsql +AS $$ +DECLARE + found_jobs procrastinate_jobs; +BEGIN + WITH candidate AS ( + SELECT jobs.* + FROM procrastinate_jobs AS jobs + WHERE + jobs.lock IS NULL + AND jobs.status = 'todo' + AND (target_queue_names IS NULL OR jobs.queue_name = ANY(target_queue_names)) + AND (jobs.scheduled_at IS NULL OR jobs.scheduled_at <= now()) + ORDER BY jobs.priority DESC, jobs.id ASC LIMIT 1 + FOR UPDATE OF jobs SKIP LOCKED ) UPDATE procrastinate_jobs - SET status = 'doing' - FROM candidate - WHERE procrastinate_jobs.id = candidate.id - RETURNING procrastinate_jobs.* INTO found_jobs; - - RETURN found_jobs; + SET status = 'doing' + FROM candidate + WHERE procrastinate_jobs.id = candidate.id + RETURNING procrastinate_jobs.* INTO found_jobs; + RETURN found_jobs; END; $$;