CAP-1073: lock ordering on unique_key to prevent worker/push deadlocks

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:
         """

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."
             )