From d1a9cd10b29d591defa319ec6a173516554ed685 Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 3 Apr 2026 11:11:53 +0200 Subject: [PATCH 1/6] Add deadlock retry in _maintain_updates to prevent worker crashes Catch DeadlockDetected specifically and retry once before re-raising. This prevents the worker from crashing on transient PostgreSQL deadlocks (e.g. concurrent batch UPDATE vs INSERT ON CONFLICT) while logging the error for observability via Sentry/Datadog. Ref: CAP-1073 Co-Authored-By: Claude Opus 4.6 --- chancy/worker.py | 83 ++++++++++++++++++++++++++++++++---------------- 1 file changed, 56 insertions(+), 27 deletions(-) diff --git a/chancy/worker.py b/chancy/worker.py index 5634250..941aaa6 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 @@ -469,33 +470,15 @@ async def _maintain_notifications(self): j = json.loads(notification.payload) await self.hub.emit(j.pop("t"), j) - async def _maintain_updates(self): + async def _flush_updates(self, pending_updates: list[QueuedJob]): """ - Process updates to job instances. + Flush pending job updates to the database. - We maintain a queue of updates to job instances, and process them in - batches to significantly reduce the number of overall transactions that - need to be made, at the cost of potentially losing some updates if the - worker is stopped unexpectedly. The frequency of these updates can be - controlled by setting the `send_outgoing_interval` attribute on the - worker. + On deadlock, retries once before re-raising. This prevents the + worker from crashing on transient deadlocks while still surfacing + the error for observability. """ - while True: - if self.outgoing.empty(): - await asyncio.sleep(self.send_outgoing_interval) - continue - - pending_updates = [] - while len(pending_updates) < 1000: - try: - pending_updates.append(self.outgoing.get_nowait()) - except asyncio.QueueEmpty: - break - - self.chancy.log.debug( - f"Processing {len(pending_updates)} outgoing updates." - ) - + for attempt in range(2): async with self.chancy.pool.connection() as conn: async with conn.cursor(row_factory=dict_row) as cursor: async with conn.transaction(): @@ -537,16 +520,62 @@ async def _maintain_updates(self): for update in pending_updates ], ) + return + except DeadlockDetected: + if attempt == 0: + self.chancy.log.error( + "Deadlock detected while updating" + f" {len(pending_updates)} jobs," + " retrying once." + ) + 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: - # 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." + "Failed to apply updates to job" + " instances." ) for update in pending_updates: await self.outgoing.put(update) raise + async def _maintain_updates(self): + """ + Process updates to job instances. + + We maintain a queue of updates to job instances, and process them in + batches to significantly reduce the number of overall transactions that + need to be made, at the cost of potentially losing some updates if the + worker is stopped unexpectedly. The frequency of these updates can be + controlled by setting the `send_outgoing_interval` attribute on the + worker. + """ + while True: + if self.outgoing.empty(): + await asyncio.sleep(self.send_outgoing_interval) + continue + + pending_updates = [] + while len(pending_updates) < 1000: + try: + pending_updates.append(self.outgoing.get_nowait()) + except asyncio.QueueEmpty: + break + + self.chancy.log.debug( + f"Processing {len(pending_updates)} outgoing updates." + ) + + await self._flush_updates(pending_updates) + for update in pending_updates: for plugin in self.chancy.plugins.values(): await plugin.on_job_updated(job=update, worker=self) From 5f87831eeed1c4dd800d9b65b37de58e80abf7ef Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 3 Apr 2026 11:27:13 +0200 Subject: [PATCH 2/6] Fix deadlock retry: move try/except outside transaction context manager The previous implementation caught DeadlockDetected inside the `conn.transaction()` block and used `continue`, which caused the context manager to attempt a COMMIT on an aborted transaction. Move the try/except outside so the transaction rolls back properly. Also removes the unnecessary _flush_updates extraction and adds a 100ms backoff between retry attempts. Ref: CAP-1073 Co-Authored-By: Claude Opus 4.6 --- chancy/worker.py | 154 +++++++++++++++++++++++------------------------ 1 file changed, 76 insertions(+), 78 deletions(-) diff --git a/chancy/worker.py b/chancy/worker.py index 941aaa6..b95f8ea 100644 --- a/chancy/worker.py +++ b/chancy/worker.py @@ -470,83 +470,6 @@ async def _maintain_notifications(self): j = json.loads(notification.payload) await self.hub.emit(j.pop("t"), j) - async def _flush_updates(self, pending_updates: list[QueuedJob]): - """ - Flush pending job updates to the database. - - On deadlock, retries once before re-raising. This prevents the - worker from crashing on transient deadlocks while still surfacing - the error for observability. - """ - for attempt in range(2): - 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 - ], - ) - return - except DeadlockDetected: - if attempt == 0: - self.chancy.log.error( - "Deadlock detected while updating" - f" {len(pending_updates)} jobs," - " retrying once." - ) - 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 - async def _maintain_updates(self): """ Process updates to job instances. @@ -574,7 +497,82 @@ async def _maintain_updates(self): f"Processing {len(pending_updates)} outgoing updates." ) - await self._flush_updates(pending_updates) + for attempt in range(2): + 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 < 1: + 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(): From cb5e70aa0ce6c3c6ee4c25a83d29e7def3d619e7 Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 3 Apr 2026 11:42:38 +0200 Subject: [PATCH 3/6] Extract DEADLOCK_MAX_RETRIES class constant for configurable retry count Single source of truth for the deadlock retry limit instead of scattered magic numbers. Ref: CAP-1073 Co-Authored-By: Claude Opus 4.6 --- chancy/worker.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/chancy/worker.py b/chancy/worker.py index b95f8ea..7863dee 100644 --- a/chancy/worker.py +++ b/chancy/worker.py @@ -125,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, @@ -497,12 +500,10 @@ async def _maintain_updates(self): f"Processing {len(pending_updates)} outgoing updates." ) - for attempt in range(2): + 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.cursor(row_factory=dict_row) as cursor: async with conn.transaction(): await cursor.executemany( sql.SQL( @@ -549,7 +550,7 @@ async def _maintain_updates(self): ) break except DeadlockDetected: - if attempt < 1: + if attempt < self.DEADLOCK_MAX_RETRIES: self.chancy.log.error( "Deadlock detected while updating" f" {len(pending_updates)} jobs," From 2461d800d8841cb73775b45e53f7277af567333c Mon Sep 17 00:00:00 2001 From: audecasteigts Date: Fri, 17 Apr 2026 15:00:08 +0200 Subject: [PATCH 4/6] 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 5/6] 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 6/6] 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: