Handling serialization failures with retry logic

Catch OperationalError, check the driver sqlstate for 40001 or 40P01, and replay the entire unit of work from a fresh session — retrying only the failed statement replays a decision made from data the database has already told you is stale. This is the retry loop that SERIALIZABLE requires, under transaction isolation and commit strategies.

Quick Answer

Before — a retry that replays the statement:

for _ in range(3):
    try:
        await session.execute(update_stmt)     # same values as the failed attempt
        await session.commit()
        break
    except OperationalError:
        await session.rollback()               # same inputs, same conflict

After — a retry that replays the unit of work:

from __future__ import annotations

import asyncio
import random
from collections.abc import Awaitable, Callable
from typing import TypeVar

from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

T = TypeVar("T")
RETRYABLE = {"40001", "40P01"}          # serialization_failure, deadlock_detected


def _sqlstate(exc: BaseException) -> str | None:
    orig = getattr(exc, "orig", None)
    return getattr(orig, "sqlstate", None) or getattr(
        getattr(orig, "diag", None), "sqlstate", None
    )


async def with_retry(
    factory: async_sessionmaker[AsyncSession],
    work: Callable[[AsyncSession], Awaitable[T]],
    attempts: int = 5,
    isolation_level: str = "SERIALIZABLE",
) -> T:
    """Replay the whole unit of work, from a fresh session, on a retryable abort."""
    for attempt in range(attempts):
        try:
            async with factory() as session:
                await session.connection(
                    execution_options={"isolation_level": isolation_level}
                )
                async with session.begin():
                    return await work(session)      # re-reads and recomputes
        except DBAPIError as exc:
            if _sqlstate(exc) not in RETRYABLE or attempt == attempts - 1:
                raise
            # Exponential backoff with jitter, so concurrent retriers separate.
            await asyncio.sleep(random.uniform(0, 0.05 * 2 ** attempt))
    raise AssertionError("unreachable")

work receives the session and must do its own reads. That is the whole design: the callable is the unit of work, so replaying it replays the reads as well as the writes.

Execution Context & Async Workflow Integration

A fresh session per attempt is not fastidiousness. A reused session serves the previously loaded instance from its identity map, so the "re-read" returns the same stale object and the retry fails identically, forever — which presents as a coroutine that never returns rather than as an error.

The work callable must therefore be written so that replaying it is meaningful:

# Async — a unit of work that is safe to replay
async def transfer(session: AsyncSession, source_id: int,
                   target_id: int, amount: decimal.Decimal) -> None:
    # Every read happens inside the transaction being retried.
    source = await session.get(Account, source_id, with_for_update=True)
    target = await session.get(Account, target_id, with_for_update=True)
    if source is None or target is None:
        raise NoSuchAccount
    if source.balance < amount:
        raise InsufficientFunds(source_id)      # not retryable — raised straight out
    source.balance -= amount
    target.balance += amount


await with_retry(Session, lambda s: transfer(s, 1, 2, decimal.Decimal("25.00")))

Two properties make this replayable. The reads are inside the callable, so each attempt sees current state. And the domain exceptions propagate rather than being caught by the retry loop, because InsufficientFunds is an answer rather than a failure — the loop only retries on the two SQLSTATEs it recognises.

The dangerous case is a unit of work with side effects outside the database. A retry replays everything, so a callable that charges a card or sends an e-mail does it once per attempt:

# Wrong — the customer is charged once per retry
async def checkout(session: AsyncSession, order_id: int) -> None:
    order = await session.get(Order, order_id)
    await payments.charge(order.total)          # external, not transactional
    order.status = "paid"

The fix is to move the external call outside the retried transaction, or to make it idempotent with a key derived from the order rather than from the attempt. Deciding which is a design question, and it has to be answered before a retry loop is put around anything with an external effect.

Resolving Warnings, Errors & Common Mistakes

Advanced: Choosing the Isolation Level, and Measuring the Cost

SERIALIZABLE gives the strongest guarantee and produces the most aborts. REPEATABLE READ produces fewer and permits write-skew anomalies. READ COMMITTED produces almost none and permits the most. The right choice depends on how much your correctness depends on invariants across rows.

The practical route is to scope the strong level to the transactions that need it rather than to the whole application:

# One unit of work at SERIALIZABLE, everything else at the engine default
await with_retry(Session, book_seat, isolation_level="SERIALIZABLE")
await with_retry(Session, refresh_dashboard, isolation_level="READ COMMITTED")

Then measure. Without numbers, the choice between SERIALIZABLE with retries and a pessimistic lock is a preference; with them it is a decision.

# Counting aborts and attempts is what turns the choice into a measurement
from collections import Counter

retry_stats: Counter[str] = Counter()


async def with_retry_counted(factory, work, **kwargs):
    for attempt in range(kwargs.get("attempts", 5)):
        try:
            result = await with_retry(factory, work, attempts=1, **kwargs)
            retry_stats[f"succeeded_on_attempt_{attempt + 1}"] += 1
            return result
        except DBAPIError as exc:
            if _sqlstate(exc) in RETRYABLE:
                retry_stats["retried"] += 1
                continue
            raise
    retry_stats["exhausted"] += 1
    raise

Three readings follow. An abort rate under a per cent or so means SERIALIZABLE is nearly free and the retry loop rarely runs. Above a few per cent, the retries are doing real work and a pessimistic SELECT ... FOR UPDATE may be cheaper, since waiting once beats replaying five times. And any non-trivial exhausted count means contention has outgrown optimistic control entirely.

The metric also protects against a subtler regression: a change that widens a transaction — an extra query, a slower computation — raises the abort rate without changing any of the isolation configuration. Watching the rate makes that visible as a graph rather than as an intermittent error nobody can reproduce. The instruments for it are in instrumenting and observing async queries.

Keeping Retried Work Idempotent

A retry loop is only as safe as the work it replays, and the audit is short enough to do once per callable.

Database writes are safe by construction. The failed attempt was aborted, so nothing it wrote exists. This is the whole reason retrying works.

Reads are safe and are the point. Each attempt re-reads, which is what makes the recomputation correct.

In-memory mutation outside the session is not safe. A callable that appends to a list, increments a counter or mutates a cache does it once per attempt. Keep the callable pure with respect to everything except the session.

External calls are not safe unless they are idempotent under a stable key. A payment charged with an idempotency key derived from the order identifier is safe; one keyed on a fresh UUID per attempt is not.

Logging is technically not idempotent and is almost always fine — but a log line saying "order placed" emitted from inside the retried block will appear once per attempt, and someone will eventually count those lines as a business metric.

# The shape that keeps the retried region minimal
async def place_order(order_id: int) -> None:
    # Outside: external effects, with their own idempotency
    payment = await payments.charge_idempotent(key=f"order-{order_id}")

    # Inside: only database work, safe to replay
    async def work(session: AsyncSession) -> None:
        order = await session.get(Order, order_id, with_for_update=True)
        order.payment_reference = payment.reference
        order.status = "paid"

    await with_retry(Session, work)
    logger.info("order placed", extra={"order_id": order_id})   # once

Keeping the retried region to database work only is worth more than any refinement of the backoff curve: it makes the question "is this safe to retry?" answerable by looking at one short function.

Frequently Asked Questions

Related

• Transaction Isolation and Commit Strategies — The parent guide: what each isolation level guarantees and what it costs.
• Setting transaction isolation level per session — Scoping the strong level to the transactions that need it.
• Implementing optimistic locking with version counters — The application-level route to the same guarantee, with the same retry requirement.
• Instrumenting and observing async queries — Measuring the abort rate that decides between optimistic and pessimistic control.