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 7863dee..8147224 100644 --- a/chancy/worker.py +++ b/chancy/worker.py @@ -496,6 +496,12 @@ 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." )