diff --git a/chancy/app.py b/chancy/app.py index 94e7f19..02e48bd 100644 --- a/chancy/app.py +++ b/chancy/app.py @@ -674,14 +674,20 @@ async def push_many_ex( :param jobs: The jobs to push onto the queue. :return: A list of references to the jobs in the queue. """ - references = [] - for job in jobs: + # Lock ordering: push jobs sorted by unique_key so concurrent worker + # UPDATE transactions acquire row locks in the same order and cannot + # deadlock (see CAP-1073). References are returned in the caller's + # original order to preserve the public API contract. + references_by_index: dict[int, Reference] = {} + for original_index, job in sorted( + enumerate(jobs), key=lambda pair: self._job_unique_key(pair[1]) + ): await cursor.execute( self._push_job_sql(), self._get_job_params(job), ) record = await cursor.fetchone() - references.append(Reference(record["id"])) + references_by_index[original_index] = Reference(record["id"]) if self.notifications: for queue in set( @@ -690,7 +696,7 @@ async def push_many_ex( ): await self.notify(cursor, "queue.pushed", {"q": queue}) - return references + return [references_by_index[i] for i in range(len(jobs))] def sync_push_many_ex( self, cursor: Cursor, jobs: list[Job] @@ -710,19 +716,22 @@ def sync_push_many_ex( :param jobs: The jobs to push onto the queue. :return: A list of references to the jobs in the queue. """ - references = [] - for job in jobs: + # Lock ordering: see push_many_ex for rationale (CAP-1073). + references_by_index: dict[int, Reference] = {} + for original_index, job in sorted( + enumerate(jobs), key=lambda pair: self._job_unique_key(pair[1]) + ): cursor.execute( self._push_job_sql(), self._get_job_params(job), ) record = cursor.fetchone() - references.append(Reference(record["id"])) + references_by_index[original_index] = Reference(record["id"]) for queue in set(job.queue for job in jobs): self.sync_notify(cursor, "queue.pushed", {"q": queue}) - return references + return [references_by_index[i] for i in range(len(jobs))] @_ensure_pool_is_open async def get_job(self, ref: Reference) -> QueuedJob | None: @@ -1248,6 +1257,11 @@ def _declare_sql(self, upsert: bool): action=action, ) + @staticmethod + def _job_unique_key(job: Job | IsAJob[..., Any]) -> str: + actual = job if isinstance(job, Job) else job.job + return actual.unique_key or "" + @staticmethod def _get_job_params(job: Job | IsAJob[..., Any]) -> dict: """ diff --git a/chancy/worker.py b/chancy/worker.py index 5634250..8147224 100644 --- a/chancy/worker.py +++ b/chancy/worker.py @@ -14,6 +14,7 @@ from psycopg import sql from psycopg import AsyncConnection +from psycopg.errors import DeadlockDetected from psycopg.rows import dict_row from psycopg.types.json import Json @@ -124,6 +125,9 @@ class Worker: :param register_signal_handlers: Whether to register signal handlers. """ + #: Maximum number of retries on deadlock before re-raising. + DEADLOCK_MAX_RETRIES = 1 + def __init__( self, chancy: Chancy, @@ -492,60 +496,90 @@ async def _maintain_updates(self): except asyncio.QueueEmpty: break + # Lock ordering: sort by unique_key so concurrent push_many_ex + # transactions acquire row locks in the same order and cannot + # deadlock. Jobs without a unique_key cannot collide via + # ON CONFLICT so grouping them first is safe. + pending_updates.sort(key=lambda u: (u.unique_key or "", u.id)) + self.chancy.log.debug( f"Processing {len(pending_updates)} outgoing updates." ) - async with self.chancy.pool.connection() as conn: - async with conn.cursor(row_factory=dict_row) as cursor: - async with conn.transaction(): - try: - await cursor.executemany( - sql.SQL( - """ - UPDATE - {jobs} - SET - state = %(state)s, - started_at = %(started_at)s, - completed_at = %(completed_at)s, - scheduled_at = %(scheduled_at)s, - attempts = %(attempts)s, - errors = %(errors)s, - meta = %(meta)s, - max_attempts = %(max_attempts)s - WHERE - id = %(id)s - """ - ).format( - jobs=sql.Identifier( - f"{self.chancy.prefix}jobs" - ) - ), - [ - { - "id": update.id, - "state": update.state.value, - "started_at": update.started_at, - "completed_at": update.completed_at, - "scheduled_at": update.scheduled_at, - "attempts": update.attempts, - "errors": Json(update.errors), - "meta": Json(update.meta), - "max_attempts": update.max_attempts, - } - for update in pending_updates - ], - ) - except Exception: - # If we were unable to apply the updates, we should - # re-queue them for the next poll. - self.chancy.log.exception( - "Failed to apply updates to job instances." - ) - for update in pending_updates: - await self.outgoing.put(update) - raise + for attempt in range(self.DEADLOCK_MAX_RETRIES + 1): + try: + async with self.chancy.pool.connection() as conn: + async with conn.cursor(row_factory=dict_row) as cursor: + async with conn.transaction(): + await cursor.executemany( + sql.SQL( + """ + UPDATE + {jobs} + SET + state = %(state)s, + started_at = %(started_at)s, + completed_at = %(completed_at)s, + scheduled_at = %(scheduled_at)s, + attempts = %(attempts)s, + errors = %(errors)s, + meta = %(meta)s, + max_attempts = %(max_attempts)s + WHERE + id = %(id)s + """ + ).format( + jobs=sql.Identifier( + f"{self.chancy.prefix}jobs" + ) + ), + [ + { + "id": update.id, + "state": update.state.value, + "started_at": update.started_at, + "completed_at": ( + update.completed_at + ), + "scheduled_at": ( + update.scheduled_at + ), + "attempts": update.attempts, + "errors": Json(update.errors), + "meta": Json(update.meta), + "max_attempts": ( + update.max_attempts + ), + } + for update in pending_updates + ], + ) + break + except DeadlockDetected: + if attempt < self.DEADLOCK_MAX_RETRIES: + self.chancy.log.error( + "Deadlock detected while updating" + f" {len(pending_updates)} jobs," + " retrying once." + ) + await asyncio.sleep(0.1) + continue + + self.chancy.log.exception( + "Deadlock detected while updating" + f" {len(pending_updates)} jobs," + " retry exhausted." + ) + for update in pending_updates: + await self.outgoing.put(update) + raise + except Exception: + self.chancy.log.exception( + "Failed to apply updates to job instances." + ) + for update in pending_updates: + await self.outgoing.put(update) + raise for update in pending_updates: for plugin in self.chancy.plugins.values():