From 2461d800d8841cb73775b45e53f7277af567333c Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 17 Apr 2026 15:00:08 +0200 Subject: [PATCH 1/3] Sort pending updates by unique_key in _maintain_updates Enforce a deterministic lock ordering on chancy_jobs rows so that concurrent push_many_ex INSERT ... ON CONFLICT transactions cannot deadlock with the worker batch update. Sorting by (unique_key, id) guarantees both sides acquire row-level exclusive locks in the same order, eliminating the circular wait detected in CAP-1073. --- chancy/worker.py | 6 ++++++ 1 file changed, 6 insertions(+) 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." ) From ddf7dc4635dd1a51acb88fdc3f435c8b5cf2a020 Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 17 Apr 2026 15:01:54 +0200 Subject: [PATCH 2/3] Sort jobs by unique_key in push_many_ex to prevent worker deadlocks Jobs sharing a unique_key collide via ON CONFLICT DO UPDATE with rows being updated by the worker's _maintain_updates batch. Without a deterministic lock order, the two transactions can acquire row locks in opposite orders and deadlock (CAP-1073). Sorting the inserts by unique_key matches the ordering now applied worker-side, removing the cycle. References are placed back in the caller's original order so the public return contract is unchanged. --- chancy/app.py | 27 +++++++++++++++++++++------ 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/chancy/app.py b/chancy/app.py index 94e7f19..e417882 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: list[Reference | None] = [None] * len(jobs) + 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[original_index] = Reference(record["id"]) if self.notifications: for queue in set( @@ -710,14 +716,17 @@ 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: list[Reference | None] = [None] * len(jobs) + 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[original_index] = Reference(record["id"]) for queue in set(job.queue for job in jobs): self.sync_notify(cursor, "queue.pushed", {"q": queue}) @@ -1248,6 +1257,12 @@ def _declare_sql(self, upsert: bool): action=action, ) + @staticmethod + def _job_unique_key(job: Job | IsAJob[..., Any]) -> str: + if callable(job): + job = job.job + return job.unique_key or "" + @staticmethod def _get_job_params(job: Job | IsAJob[..., Any]) -> dict: """ From af323516d77c45b7669dc1307602d91b1ecf10be Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Mon, 27 Apr 2026 16:55:24 +0200 Subject: [PATCH 3/3] Address review feedback: align Job pattern and fix typing of references - Use isinstance(job, Job) in _job_unique_key for consistency with the pattern already used at the call site in push_many_ex. - Replace pre-allocated list[Reference | None] with a dict[int, Reference] intermediate so the function returns a clean list[Reference] without needing a cast or carrying Optionals through to mypy. --- chancy/app.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/chancy/app.py b/chancy/app.py index e417882..02e48bd 100644 --- a/chancy/app.py +++ b/chancy/app.py @@ -678,7 +678,7 @@ async def push_many_ex( # 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: list[Reference | None] = [None] * len(jobs) + references_by_index: dict[int, Reference] = {} for original_index, job in sorted( enumerate(jobs), key=lambda pair: self._job_unique_key(pair[1]) ): @@ -687,7 +687,7 @@ async def push_many_ex( self._get_job_params(job), ) record = await cursor.fetchone() - references[original_index] = Reference(record["id"]) + references_by_index[original_index] = Reference(record["id"]) if self.notifications: for queue in set( @@ -696,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] @@ -717,7 +717,7 @@ def sync_push_many_ex( :return: A list of references to the jobs in the queue. """ # Lock ordering: see push_many_ex for rationale (CAP-1073). - references: list[Reference | None] = [None] * len(jobs) + references_by_index: dict[int, Reference] = {} for original_index, job in sorted( enumerate(jobs), key=lambda pair: self._job_unique_key(pair[1]) ): @@ -726,12 +726,12 @@ def sync_push_many_ex( self._get_job_params(job), ) record = cursor.fetchone() - references[original_index] = 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: @@ -1259,9 +1259,8 @@ def _declare_sql(self, upsert: bool): @staticmethod def _job_unique_key(job: Job | IsAJob[..., Any]) -> str: - if callable(job): - job = job.job - return job.unique_key or "" + actual = job if isinstance(job, Job) else job.job + return actual.unique_key or "" @staticmethod def _get_job_params(job: Job | IsAJob[..., Any]) -> dict: