Skip to content

Branch with upgrades #90

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
AudeCstg wants to merge 8 commits into
base: main
Choose a base branch
from
Draft

Branch with upgrades #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d1a9cd1
Add deadlock retry in _maintain_updates to prevent worker crashes
AudeCstg Apr 3, 2026
5f87831
Fix deadlock retry: move try/except outside transaction context manager
AudeCstg Apr 3, 2026
cb5e70a
Extract DEADLOCK_MAX_RETRIES class constant for configurable retry count
AudeCstg Apr 3, 2026
a75c3bb
Merge pull request #1 from ouihelp/ac/cap-1073-deadlock-retry
AudeCstg Apr 13, 2026
2461d80
Sort pending updates by unique_key in _maintain_updates
AudeCstg Apr 17, 2026
ddf7dc4
Sort jobs by unique_key in push_many_ex to prevent worker deadlocks
AudeCstg Apr 17, 2026
af32351
Address review feedback: align Job pattern and fix typing of references
AudeCstg Apr 27, 2026
7acaac1
Merge pull request #2 from ouihelp/ac/cap-1073-lock-ordering
AudeCstg Apr 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 22 additions & 8 deletions chancy/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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]
Expand All @@ -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:
Expand Down Expand Up @@ -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:
"""
Expand Down
134 changes: 84 additions & 50 deletions chancy/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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():
Expand Down