Transactions, locking and race conditions

The clubhouse yoga class at 6 am has 5 places, and at 6 pm the evening before, 20 residents press Book in the same second. The obvious code, read the class, check the places, save one fewer, confirmed all 20 of them on PostgreSQL 18, and no error appeared anywhere.

You will make groups of writes all or nothing with transaction.atomic(), reproduce that race and fix it with a conditional update and with a row lock, turn a table into a work queue with skip_locked, see why SQLite cannot test any of this, and send the confirmation email only once the booking is committed. The races ran on Django 6.1.2 with PostgreSQL 18.6, each on a throwaway server of its own; the other examples on SQLite 3.51.0.

atomic() makes a group of writes all or nothing

A booking is two writes: a row in Booking and one place fewer in Slot. If the second fails, the first must not stay. Django runs in autocommit mode, each query committed at once, unless a transactiontransaction: A group of database writes that is committed as a whole or rolled back as a whole. Until it commits, other connections do not see its writes; Django opens one with transaction.atomic(). is open, and transaction.atomic() opens one for its block:

"""transaction.atomic() on SQLite: a block that fails leaves nothing behind, an inner block is a savepoint that can
fail alone, and a rollback does not undo what Python holds. The SQL of each step is printed as it was sent.

Run it next to manage.py. seed.py makes db.sqlite3 there with every table and the yoga class, which has 5 places.
"""

import warnings

import seed  # noqa: F401  (starts Django, creates the tables, adds the rows)
from django.db import IntegrityError, connection, transaction
from django.db.models import F
from django.test.utils import CaptureQueriesContext

from amenities.models import Booking, Slot
from seed import compact

slot = Slot.objects.get(name='Yoga, 6 am')


def state(label):
    print(f'   {label}: bookings {Booking.objects.count()}, places_left {Slot.objects.get(pk=slot.pk).places_left}')


print('1. An exception leaves the block: both writes are rolled back')
try:
    with transaction.atomic():
        Booking.objects.create(slot=slot, resident='Asha K.')
        Slot.objects.filter(pk=slot.pk).update(places_left=F('places_left') - 1)
        raise RuntimeError('the payment failed')
except RuntimeError as error:
    print('   caught:', error)
state('after the rollback')

print('2. An inner atomic() is a savepoint: its failure is rolled back alone')
with CaptureQueriesContext(connection) as queries:
    with transaction.atomic():
        Booking.objects.create(slot=slot, resident='Imran Q.')
        try:
            with transaction.atomic():
                Booking.objects.create(slot=slot, resident='Imran Q.')  # the same resident again
        except IntegrityError as error:
            print('   the second booking was refused:', error)
        Slot.objects.filter(pk=slot.pk).update(places_left=F('places_left') - 1)
for query in queries:
    print('     ', compact(query['sql']))
state('after the commit')

print('3. A rollback does not change the objects Python holds')
slot.refresh_from_db()
try:
    with transaction.atomic():
        slot.places_left -= 1
        slot.save(update_fields=['places_left'])
        raise RuntimeError('the payment failed')
except RuntimeError:
    pass
print('   slot.places_left in Python:', slot.places_left, '| in the database:', Slot.objects.get(pk=slot.pk).places_left)

print('4. The savepoint functions of Django 6.1')
with transaction.atomic():
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter('always')
        transaction.savepoint()
    print('   transaction.savepoint():', caught[0].category.__name__, '-', caught[0].message)
    sid = transaction.savepoint_create()
    Booking.objects.create(slot=slot, resident='Meera P.')
    transaction.savepoint_rollback(sid)
state('savepoint_create(), then savepoint_rollback()')

1. An exception leaves the block: both writes are rolled back
   caught: the payment failed
   after the rollback: bookings 0, places_left 5
2. An inner atomic() is a savepoint: its failure is rolled back alone
   the second booking was refused: UNIQUE constraint failed: amenities_booking.slot_id, amenities_booking.resident
      BEGIN
      INSERT INTO booking (slot_id, resident) VALUES (1, 'Imran Q.') RETURNING booking.id
      SAVEPOINT s…_x1
      INSERT INTO booking (slot_id, resident) VALUES (1, 'Imran Q.') RETURNING booking.id
      ROLLBACK TO SAVEPOINT s…_x1
      RELEASE SAVEPOINT s…_x1
      UPDATE slot SET places_left = (slot.places_left - 1) WHERE slot.id = 1
      COMMIT
   after the commit: bookings 1, places_left 4
3. A rollback does not change the objects Python holds
   slot.places_left in Python: 3 | in the database: 4
4. The savepoint functions of Django 6.1
   transaction.savepoint(): RemovedInDjango2028Warning - savepoint() is deprecated. Use savepoint_create() instead.
   savepoint_create(), then savepoint_rollback(): bookings 1, places_left 4

In step 1 the payment failed inside the block, the exception left it, and both writes were rolled back: 0 bookings and 5 places. Step 2 nests one block inside another. The inner atomic() cannot start a second transaction, so it sets a savepointsavepoint: A marked point inside a transaction that the transaction can roll back to without undoing the work done before it. A nested transaction.atomic() block sets one., and when its INSERT broke the constraint one_booking_per_resident, only that part was undone. Step through the SQL it sent:

Catch the error around an inner atomic(), as step 2 does, never inside the block that keeps going: Django’s documentation warns that a caught database error leaves the transaction broken, and the next query raises TransactionManagementError. Step 3 shows what a rollback does not do: slot.places_left was 3 in Python and 4 in the database, because objects in memory are not rolled back. Reload them with refresh_from_db().

ATOMIC_REQUESTS = True in a database’s settings wraps every view in such a transaction. Django’s documentation calls the model simple but inefficient when traffic increases, and middleware and the rendering of template responses run outside it. Step 4 is new in 6.1: transaction.savepoint() raised a RemovedInDjango2028Warning, and its new name, savepoint_create(), worked as before.

Read, then write: 20 residents confirmed for 5 places

atomic() keeps one request’s writes together; it does not stop two requests from meeting. This program runs five versions of book() from 20 threads at once against PostgreSQL 18, through the throwaway server of pgserver.py:

"""Twenty residents press Book for the 6 am yoga class at the same moment, and it has 5 places. Five ways to write
book(), each run by 20 threads against PostgreSQL 18. A barrier makes every thread start at once; for the first two
ways, a second barrier holds every thread between its read and its write, the worst moment two requests can meet.

The class has a CheckConstraint, places_left >= 0, in amenities/models.py.
"""

import threading

import pgserver

pgserver.start()

import seed  # noqa: E402,F401  (starts Django, creates the tables, adds the rows)
from django.db import IntegrityError, connection, connections, transaction  # noqa: E402
from django.db.models import F  # noqa: E402

from amenities.models import Booking, Slot  # noqa: E402

THREADS = 20


def read_modify_write(slot_id, who, between):
    slot = Slot.objects.get(pk=slot_id)
    between.wait()  # every thread has read places_left before any of them writes
    if slot.places_left > 0:
        slot.places_left -= 1
        slot.save(update_fields=['places_left'])
        Booking.objects.create(slot_id=slot_id, resident=who)
        return 'confirmed'
    return 'full'


def read_modify_write_in_atomic(slot_id, who, between):
    with transaction.atomic():
        return read_modify_write(slot_id, who, between)


def f_without_condition(slot_id, who, between):
    with transaction.atomic():
        Slot.objects.filter(pk=slot_id).update(places_left=F('places_left') - 1)
        Booking.objects.create(slot_id=slot_id, resident=who)
    return 'confirmed'


def conditional_update(slot_id, who, between):
    with transaction.atomic():
        took = Slot.objects.filter(pk=slot_id, places_left__gt=0).update(places_left=F('places_left') - 1)
        if not took:
            return 'full'
        Booking.objects.create(slot_id=slot_id, resident=who)
    return 'confirmed'


def select_for_update(slot_id, who, between):
    with transaction.atomic():
        slot = Slot.objects.select_for_update().get(pk=slot_id)
        if slot.places_left == 0:
            return 'full'
        slot.places_left -= 1
        slot.save(update_fields=['places_left'])
        Booking.objects.create(slot_id=slot_id, resident=who)
    return 'confirmed'


def race(book):
    slot = Slot.objects.create(name=f'Yoga, 6 am ({book.__name__})', places_left=5)
    start, between = threading.Barrier(THREADS), threading.Barrier(THREADS)  # between: used by the first two
    told, errors = [], []

    def resident(n):
        start.wait()
        try:
            told.append(book(slot.pk, f'resident {n}', between))
        except IntegrityError as error:
            told.append(type(error).__name__)
            errors.append(str(error).splitlines()[0])
        finally:
            connections.close_all()  # each thread has its own connection to PostgreSQL

    threads = [threading.Thread(target=resident, args=(n,)) for n in range(1, THREADS + 1)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    slot.refresh_from_db()
    outcome = {answer: told.count(answer) for answer in sorted(set(told))}
    print(f'{book.__name__:<28} told {outcome}; places_left {slot.places_left};'
          f' bookings in the table {Booking.objects.filter(slot=slot).count()}')
    if errors:
        print('    every IntegrityError said:', errors[0] if len(set(errors)) == 1 else sorted(set(errors)))


version = connection.cursor().connection.info.server_version
print(f'PostgreSQL {version // 10000}.{version % 10000} | 20 residents, 5 places')
for book in (read_modify_write, read_modify_write_in_atomic, f_without_condition, conditional_update,
             select_for_update):
    race(book)

PostgreSQL 18.6 | 20 residents, 5 places
read_modify_write            told {'confirmed': 20}; places_left 4; bookings in the table 20
read_modify_write_in_atomic  told {'confirmed': 20}; places_left 4; bookings in the table 20
f_without_condition          told {'IntegrityError': 15, 'confirmed': 5}; places_left 0; bookings in the table 5
    every IntegrityError said: new row for relation "amenities_slot" violates check constraint "places_left_not_negative"
conditional_update           told {'confirmed': 5, 'full': 15}; places_left 0; bookings in the table 5
select_for_update            told {'confirmed': 5, 'full': 15}; places_left 0; bookings in the table 5

"""A throwaway PostgreSQL 18 server for one example: start() creates a database cluster in a folder of its own, starts
the server listening only on 127.0.0.1, an address no other computer can reach, and tells Django its port through
NIVAS_PG_PORT (config/settings.py reads it). The server is stopped and its folder deleted when the example ends, also
when it fails.

It needs PostgreSQL's server programs, initdb and pg_ctl, on PATH (or in the mise installation beside this Python),
and the psycopg package, which Django uses to talk to PostgreSQL.
"""

import atexit
import os
import shutil
import socket
import subprocess
import sys
import tempfile
from pathlib import Path


def _postgres_bin():
    on_path = shutil.which('pg_ctl')
    if on_path:
        return Path(on_path).parent
    installs = Path(sys.executable).resolve().parents[3]  # <mise>/installs/python/3.14.8/bin/python3
    found = sorted(installs.glob('postgres/18.*/bin/pg_ctl'))
    if not found:
        sys.exit('PostgreSQL 18 was not found: install it so that pg_ctl is on PATH')
    return found[-1].parent


def start():
    pg = _postgres_bin()
    work = Path(tempfile.mkdtemp(prefix='pg', dir='.')).resolve()
    data = work / 'data'
    subprocess.run([pg / 'initdb', '-D', data, '-U', 'postgres', '-A', 'trust', '-E', 'UTF8', '--no-locale'],
                   check=True, capture_output=True)
    with socket.socket() as probe:  # ask the operating system for a port that nothing uses
        probe.bind(('127.0.0.1', 0))
        port = probe.getsockname()[1]
    with open(data / 'postgresql.conf', 'a') as conf:
        conf.write(f"listen_addresses = '127.0.0.1'\nport = {port}\nunix_socket_directories = ''\n")
    subprocess.run([pg / 'pg_ctl', '-D', data, '-l', work / 'server.log', '-w', 'start'], check=True,
                   capture_output=True)

    def stop():
        subprocess.run([pg / 'pg_ctl', '-D', data, '-m', 'fast', '-w', 'stop'], capture_output=True)
        shutil.rmtree(work, ignore_errors=True)

    atexit.register(stop)
    os.environ['NIVAS_PG_PORT'] = str(port)

from django.db import models


class Slot(models.Model):
    """One session of an amenity, such as the clubhouse yoga class at 6 am, with the places still free."""

    name = models.CharField(max_length=60)
    places_left = models.IntegerField()

    class Meta:
        constraints = [
            models.CheckConstraint(condition=models.Q(places_left__gte=0), name='places_left_not_negative'),
        ]

    def __str__(self):
        return self.name


class Booking(models.Model):
    slot = models.ForeignKey(Slot, on_delete=models.CASCADE, related_name='bookings')
    resident = models.CharField(max_length=100)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=['slot', 'resident'], name='one_booking_per_resident'),
        ]

    def __str__(self):
        return f'{self.resident} in {self.slot}'


class Job(models.Model):
    """A piece of work for a background worker, such as a dues reminder to send."""

    task = models.CharField(max_length=100)
    claimed_by = models.CharField(max_length=20, blank=True)

The first line is the version most code starts as, read_modify_write(): get the slot, check places_left in Python, save one fewer, create the booking. A barrier held all 20 threads between their read and their write, the worst moment for two requests to meet, so every one of them read 5. Step through two of them:

That is a lost updatelost update: Two requests read the same value, each writes a new value computed from what it read, and the second write erases the first. No error is raised, and the data is wrong.: each write overwrote the one before with a value computed from an old read. All 20 residents were told “confirmed”, 20 bookings were written, and places_left said 4 places remained. Without the barrier the requests meet only when their timing happens to overlap, which makes the bug rare in testing and likely on a busy morning. The CheckConstraint on Slot, places_left >= 0, saw nothing wrong: every value written, 4, was valid.

A conditional update or a row lock confirms exactly 5

The last three lines of the same run moved the check into the database.

f_without_condition() used F('places_left') - 1, so the database subtracted from the value in the row and no update was lost. But it had no condition: the sixth request would have stored -1. The CheckConstraint refused it, and the next 14, with an IntegrityError each. The constraint was the last line of defence: it kept the data right, while 15 residents saw an error instead of “full”.

conditional_update() put the condition into the same statement, UPDATE … SET places_left = places_left - 1 WHERE places_left > 0, and update() returned 1 for a place taken and 0 for full. Step through what PostgreSQL does when two requests send it for the last place:

PostgreSQL’s documentation describes the re-check: an UPDATE whose row was changed by a transaction that has not ended yet waits until that transaction is over, then tests its condition once more against the row as it now is. select_for_update() does the same work with a row lockrow lock: A lock a transaction holds on one row, for example after SELECT … FOR UPDATE, so that other transactions that want to change or lock that row wait until it commits or rolls back. taken by the read: inside atomic() it sends SELECT … FOR UPDATE, every other request that wants to lock or change the row waits, and Python can check whatever it needs before saving. Both confirmed exactly 5 and told 15 “full”. Answer three questions about your own write, and the table marks the ways that fit:

In short: a conditional update when one condition on the row decides; select_for_update() when Python must decide, with a transaction kept short because others wait for it; and a constraint underneath both, so that a bug elsewhere becomes an error instead of bad data.

Skipping locked rows turns a table into a work queue

Six dues reminders wait in the Job table, and several workers send them. Each must take jobs nobody else has taken, without queueing behind the others. Worker A locks the first two and keeps its transaction open; worker B then asks three ways:

"""Three workers take jobs from one table on PostgreSQL 18. Worker A locks the first two jobs and keeps its
transaction open; meanwhile worker B asks for two jobs in three ways: with nowait=True, with skip_locked=True, and,
after A has committed, as usual. Events make the order of the steps the same on every run.
"""

import threading

import pgserver

pgserver.start()

import seed  # noqa: E402,F401  (starts Django, creates the tables, adds the rows)
from django.db import DatabaseError, connections, transaction  # noqa: E402

from amenities.models import Job  # noqa: E402

a_holds_its_jobs, b_is_done = threading.Event(), threading.Event()


def claim(worker, **lock):
    """Claims the first two unclaimed jobs, in id order, and returns their tasks."""
    jobs = list(Job.objects.select_for_update(**lock).filter(claimed_by='').order_by('pk')[:2])
    for job in jobs:
        job.claimed_by = worker
        job.save(update_fields=['claimed_by'])
    return [job.task for job in jobs]


def worker_a():
    with transaction.atomic():
        print('A claims', claim('A'), 'and keeps its transaction open')
        a_holds_its_jobs.set()
        b_is_done.wait()  # A's two rows stay locked until B has tried
    print('A commits')
    connections.close_all()


def worker_b():
    a_holds_its_jobs.wait()
    try:
        with transaction.atomic():
            claim('B', nowait=True)
    except DatabaseError as error:
        print('B with nowait=True:', type(error).__name__, '-', str(error).strip())
    with transaction.atomic():
        print('B with skip_locked=True claims', claim('B', skip_locked=True), 'at once')
    b_is_done.set()
    connections.close_all()


threads = [threading.Thread(target=worker_a), threading.Thread(target=worker_b)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
with transaction.atomic():
    print('C claims', claim('C'))
print('who has which job:', list(Job.objects.order_by('pk').values_list('task', 'claimed_by')))

A claims ['dues reminder 1', 'dues reminder 2'] and keeps its transaction open
B with nowait=True: OperationalError - could not obtain lock on row in relation "amenities_job"
B with skip_locked=True claims ['dues reminder 3', 'dues reminder 4'] at once
A commits
C claims ['dues reminder 5', 'dues reminder 6']
who has which job: [('dues reminder 1', 'A'), ('dues reminder 2', 'A'), ('dues reminder 3', 'B'), ('dues reminder 4', 'B'), ('dues reminder 5', 'C'), ('dues reminder 6', 'C')]

With nowait=True, B’s query met A’s locked rows and failed at once: OperationalError: could not obtain lock on row in relation "amenities_job". With skip_locked=True, it passed over the locked rows, FOR UPDATE SKIP LOCKED, and took jobs 3 and 4 while A still held 1 and 2. Without either option it would have waited until A committed. After both had committed, C took the last two.

Django’s documentation adds the details: nowait and skip_locked cannot be combined, and a database that has FOR UPDATE but not one of the options, such as MariaDB for of, raises NotSupportedError; SQLite, which has no FOR UPDATE at all, leaves the clause out, as the next section shows. A claimed job should still be marked in its row, as claimed_by is here, so that it stays taken after the transaction ends.

SQLite has no row locks, so test locking on PostgreSQL

Every lock in the last two sections came from PostgreSQL. This program sends the same select_for_update() to PostgreSQL 18 and to SQLite, inside a transaction and outside one:

"""The same select_for_update() on PostgreSQL 18 and on SQLite: the SQL each one sends, inside a transaction and
outside one. Row locks are a PostgreSQL feature that SQLite does not have, so a test on SQLite cannot show them.

The settings name two databases when pgserver.py has started a server: PostgreSQL as 'default', SQLite as 'sqlite'.
"""

import pgserver

pgserver.start()

import seed  # noqa: E402,F401  (starts Django, creates the tables in both databases, adds the rows)
from django.db import connections, transaction  # noqa: E402
from django.db.transaction import TransactionManagementError  # noqa: E402
from django.test.utils import CaptureQueriesContext  # noqa: E402

from amenities.models import Job, Slot  # noqa: E402
from seed import compact  # noqa: E402

for alias in ('default', 'sqlite'):
    connection = connections[alias]
    print(f'{connection.display_name}: has_select_for_update {connection.features.has_select_for_update},'
          f' skip_locked {connection.features.has_select_for_update_skip_locked}')
    with CaptureQueriesContext(connection) as queries:
        with transaction.atomic(using=alias):
            Slot.objects.using(alias).select_for_update().get(name='Yoga, 6 am')
            list(Job.objects.using(alias).select_for_update(skip_locked=True).order_by('pk')[:2])
    for query in queries:
        print('   ', compact(query['sql']))
    try:
        list(Slot.objects.using(alias).select_for_update())
        print('    outside a transaction: no error, and no lock')
    except TransactionManagementError as error:
        print('    outside a transaction:', type(error).__name__, '-', error)

PostgreSQL: has_select_for_update True, skip_locked True
    BEGIN
    SELECT slot.id, slot.name, slot.places_left FROM slot WHERE slot.name = 'Yoga, 6 am' LIMIT 21 FOR UPDATE
    SELECT job.id, job.task, job.claimed_by FROM job ORDER BY job.id ASC LIMIT 2 FOR UPDATE SKIP LOCKED
    COMMIT
    outside a transaction: TransactionManagementError - select_for_update cannot be used outside of a transaction.
SQLite: has_select_for_update False, skip_locked False
    BEGIN
    SELECT slot.id, slot.name, slot.places_left FROM slot WHERE slot.name = 'Yoga, 6 am' LIMIT 21
    SELECT job.id, job.task, job.claimed_by FROM job ORDER BY job.id ASC LIMIT 2
    COMMIT
    outside a transaction: no error, and no lock

On PostgreSQL the reads ended in FOR UPDATE and FOR UPDATE SKIP LOCKED, and outside a transaction Django refused with TransactionManagementError, because a lock would end with the statement. On SQLite, has_select_for_update is False: Django left the clause out, and outside a transaction there was no error and no lock. Django’s documentation puts it plainly: on such backends select_for_update() has no effect.

So a locking bug can pass every test on SQLite. Django’s own test case hides one more: TestCase wraps each test in a transaction, so code that forgot atomic() around select_for_update() passes there too, and the documentation tells you to test it with TransactionTestCase. A test that shows a race needs real concurrency: several threads or processes, each committing, against the database production uses, and a barrier so that they meet every time, as race.py does.

Side effects wait for the commit: no email for a failed booking

A confirmation email is a write too, to someone’s inbox, and no rollback can take it back. This program books four residents, two of whose payments fail, with the email sent inside the block or with transaction.on_commit():

"""The booking confirmation email, sent two ways: inside the transaction, and with transaction.on_commit(), which
waits until the booking is committed. Then two callbacks, one of which fails, with robust=True.

Run it next to manage.py. seed.py makes db.sqlite3 there with every table and the yoga class. The settings send mail
to django.core.mail.outbox, a list in memory, instead of a mail server.
"""

import logging

import seed  # noqa: F401  (starts Django, creates the tables, adds the rows)
from django.core import mail
from django.db import transaction

from amenities.models import Booking, Slot

slot = Slot.objects.get(name='Yoga, 6 am')


def send_confirmation(resident):
    mail.send_mail(f'Booked: {slot.name}', f'{resident}, your place is confirmed.', 'office@nivas.example',
                   [f'{resident.split()[0].lower()}@residents.example'])


def book(resident, *, email_inside, fail):
    try:
        with transaction.atomic():
            Booking.objects.create(slot=slot, resident=resident)
            if email_inside:
                send_confirmation(resident)
            else:
                transaction.on_commit(lambda: send_confirmation(resident))
            if fail:
                raise RuntimeError('the payment failed')
    except RuntimeError:
        pass
    booked = Booking.objects.filter(resident=resident).exists()
    print(f'{resident:<9} email {"inside the block" if email_inside else "on_commit()":<16} payment'
          f' {"failed" if fail else "ok":<7} booked {booked!s:<5} emails sent so far {len(mail.outbox)}')


book('Asha K.', email_inside=True, fail=False)
book('Imran Q.', email_inside=True, fail=True)
book('Meera P.', email_inside=False, fail=False)
book('Rohan M.', email_inside=False, fail=True)
print('to:', [message.to[0] for message in mail.outbox])


def broken():
    raise ConnectionError('the SMS gateway is down')


class Show(logging.Handler):
    """Prints what Django logs about a failed callback, without the traceback the log also holds."""

    def emit(self, record):
        print('django.db.backends.base logged:', record.getMessage())


logging.getLogger('django.db.backends.base').addHandler(Show())
logging.getLogger('django.db.backends.base').propagate = False


with transaction.atomic():
    transaction.on_commit(broken, robust=True)
    transaction.on_commit(lambda: print('the next callback still ran after the failure'))

Asha K.   email inside the block payment ok      booked True  emails sent so far 1
Imran Q.  email inside the block payment failed  booked False emails sent so far 2
Meera P.  email on_commit()      payment ok      booked True  emails sent so far 3
Rohan M.  email on_commit()      payment failed  booked False emails sent so far 3
to: ['asha@residents.example', 'imran@residents.example', 'meera@residents.example']
django.db.backends.base logged: Error calling broken in on_commit() during transaction (the SMS gateway is down).
the next callback still ran after the failure

Imran Q.’s payment failed after his email had gone: the booking was rolled back, and he had a confirmation for a place he does not have. With on_commit(), Meera P.’s email left after her booking was committed, and Rohan M.’s was thrown away with his rollback. Django’s documentation gives the rules: the callback runs after the open transaction commits, is discarded if it rolls back, runs at once when no transaction is open, and in a savepoint waits for the outer commit.

The last lines show the other side. A callback runs after the commit, so its failure cannot undo the booking: robust=True logged “Error calling broken in on_commit() during transaction (the SMS gateway is down).” and let the next callback run. A booking whose email failed is committed without its email, which is why important effects go to a queue or a task that a worker retries. TestCase never commits, so its captureOnCommitCallbacks() lets a test run or inspect the callbacks.

Practise: replay the race and claim the jobs

Two pieces of this lesson rebuilt in plain Python. First replay the booking race step by step, so that the lost update appears exactly when the reads come before the writes; then claim jobs the way SKIP LOCKED and NOWAIT decide. Each step has tests you can run here and a hint:

def replay(schedule, places, strategy):
    """(sorted names told confirmed, places left) after the steps of the requests ran in the order of schedule."""
    confirmed = []
    for name in schedule:
        if name in confirmed:
            continue  # a request's second step
        if places > 0:
            places -= 1
            confirmed.append(name)
    return sorted(confirmed), places

from race import replay


def test_one_after_the_other():
    """read-modify-write is right when one request finishes before the next starts"""
    assert replay(['A', 'A', 'B', 'B'], 1, 'read-modify-write') == (['A'], 0)


def test_lost_update():
    """both read 1, both write 0: two confirmations for one place"""
    assert replay(['A', 'B', 'A', 'B'], 1, 'read-modify-write') == (['A', 'B'], 0)


def test_the_lesson_race():
    """20 reads before any write, 5 places: all 20 confirmed and 4 places still showing, as race.py recorded"""
    names = [f'r{n:02}' for n in range(1, 21)]
    confirmed, left = replay(names + names, 5, 'read-modify-write')
    assert len(confirmed) == 20
    assert left == 4


def test_conditional_never_oversells():
    """the conditional update confirms exactly as many as there are places"""
    names = [f'r{n:02}' for n in range(1, 21)]
    confirmed, left = replay(names, 5, 'conditional')
    assert confirmed == names[:5]
    assert left == 0


def test_conditional_in_any_order():
    """the order of the requests changes who gets a place, never how many"""
    assert replay(['C', 'A', 'B'], 2, 'conditional') == (['A', 'C'], 0)


def test_full_from_the_start():
    """no places: nobody is confirmed, whatever the strategy"""
    assert replay(['A', 'B', 'A', 'B'], 0, 'read-modify-write') == ([], 0)
    assert replay(['A', 'B'], 0, 'conditional') == ([], 0)

class LockNotAvailable(Exception):
    """Another transaction holds the lock on a row the query must lock, as PostgreSQL reports for NOWAIT."""


def claim(jobs, locked, batch, skip_locked):
    """The ids of the jobs this worker claims, in id order."""
    free = [job['id'] for job in jobs if job['claimed_by'] == '']
    return free[:batch]

from jobs import LockNotAvailable, claim

JOBS = [{'id': n, 'claimed_by': ''} for n in range(1, 7)]


def test_skip_locked_passes_over():
    """A holds jobs 1 and 2, so B gets 3 and 4, as work_queue.py recorded"""
    assert claim(JOBS, {1, 2}, 2, skip_locked=True) == [3, 4]


def test_nowait_refuses():
    """without SKIP LOCKED the first two jobs are 1 and 2, which are locked: NOWAIT raises"""
    try:
        claim(JOBS, {1, 2}, 2, skip_locked=False)
    except LockNotAvailable:
        return
    raise AssertionError('claim() should have raised LockNotAvailable')


def test_nowait_when_nothing_is_locked():
    """with no lock in the way NOWAIT claims like any query"""
    assert claim(JOBS, set(), 2, skip_locked=False) == [1, 2]


def test_claimed_jobs_are_never_taken():
    """jobs already claimed are not offered again"""
    jobs = [{'id': n, 'claimed_by': 'A' if n <= 4 else ''} for n in range(1, 7)]
    assert claim(jobs, set(), 2, skip_locked=True) == [5, 6]


def test_fewer_than_a_batch():
    """near the end of the queue a worker gets what is left"""
    jobs = [{'id': n, 'claimed_by': 'C' if n != 6 else ''} for n in range(1, 7)]
    assert claim(jobs, set(), 2, skip_locked=True) == [6]


def test_id_order():
    """jobs come out in id order, whatever the order of the list"""
    assert claim(list(reversed(JOBS)), {3}, 3, skip_locked=True) == [1, 2, 4]

Then the quiz on the six ideas of this lesson, from atomic() to on_commit(); it starts with the idea where your answers have been weakest:

Interview questions

Warm-up (fresher): What does transaction.atomic() do, and what happens when atomic blocks are nested?

atomic() runs its block in a database transaction: if the block ends normally, its writes are committed together; if an exception leaves it, they are all rolled back. Nested inside another block, an atomic() creates a savepoint instead of a new transaction, so if the inner block fails, only its own writes are rolled back, and the outer block can catch the error and continue. Nothing is committed until the outermost block ends. A rollback does not reset the objects in Python, which must be reloaded.

Two design questions of this module start from this lesson: the booking rush itself, and an email for a booking that was never made.

Key takeaways

• atomic() commits a block as a whole or rolls it back as a whole; a nested atomic() is a savepoint, and a rollback never resets objects in Python.
• Reading a value, changing it in Python and saving it loses updates when requests meet: 20 residents were confirmed for 5 places, and no error appeared.
• A conditional update with F() or a row lock from select_for_update() confirmed exactly 5, and a constraint caught the version that forgot the condition.
• select_for_update(skip_locked=True) lets workers share a queue without waiting, and nowait=True turns a wait into an error.
• SQLite has no row locks and TestCase hides a missing transaction, so locking is tested on PostgreSQL with real threads or processes.
• transaction.on_commit() sends emails and tasks only after the data is committed, and its failures need a retry.