Reproducing a Race Condition Deterministically

An intermittent failure in production, a ticket that says "cannot reproduce", and a defensive lock added on suspicion: that is the usual life cycle of a race condition. The alternative is a test that fails on every single run while the bug is present, which is achievable far more often than people expect because Python's races nearly always have a window that a test can hold open rather than wait for.

Prerequisites

• Python 3.9+; threading and unittest.mock are standard library.
• pytest >= 8.0 plus pytest-timeout, so a mis-written coordination deadlocks into a failure rather than a hang.
• The concepts in testing threads and race conditions, particularly why the GIL does not make statements atomic.

Solution

Take a lazily-initialised client — the classic double-initialisation bug — and make it fail reliably.

import threading


class ServiceRegistry:
    def __init__(self, build_client):
        self._build_client = build_client
        self._client = None
        self.build_count = 0

    def get_client(self):
        if self._client is None:            # check
            client = self._build_client()   # ← the window: slow, and not atomic
            self.build_count += 1
            self._client = client           # act
        return self._client

import threading

import pytest


@pytest.mark.timeout(10)
def test_client_is_built_exactly_once():
    inside = threading.Event()              # first thread reached the window
    release = threading.Event()             # test says the first may continue

    def slow_build():
        inside.set()
        assert release.wait(timeout=5), "test never released the builder"
        return object()

    registry = ServiceRegistry(build_client=slow_build)
    results: list[object] = []

    first = threading.Thread(target=lambda: results.append(registry.get_client()))
    first.start()
    assert inside.wait(timeout=5), "first thread never entered the builder"

    # Second thread arrives while the first is provably still inside the window.
    second = threading.Thread(target=lambda: results.append(registry.get_client()))
    second.start()

    release.set()
    for thread in (first, second):
        thread.join(timeout=5)
        assert not thread.is_alive()

    assert registry.build_count == 1, "the client was built more than once"
    assert results[0] is results[1], "the two threads got different clients"

Why this works

The bug needs the second thread to evaluate self._client is None after the first has entered _build_client and before it assigns. Waiting for that by chance requires the window to be wide, which it is not — _build_client might take microseconds. The seam makes the window as wide as the test wants.

inside.set() tells the test the first thread has passed the check and is in the builder. release.wait() keeps it there. The test starts the second thread at that exact moment, so the ordering is a precondition of the test rather than a coincidence. Every run therefore produces the same interleaving, and the invariant assertion — build_count == 1 — fails every time until the lock is added.

Edge cases and failure modes

• No timeout on the waits. Every Event.wait and Thread.join needs one, or a mistake in the coordination produces a hang instead of a failure. The assertion on wait's return value converts a timed-out wait into a readable message.
• Assertions inside worker threads. An AssertionError in a thread is printed and discarded. Collect results in a list and assert in the main thread, or use a ThreadPoolExecutor whose result() re-raises.
• Exceptions before the event is set. If the builder raises before inside.set(), the test waits the full five seconds and then fails with a confusing message. Set the event first, then do the work that can fail.
• The seam left in production. A build_client parameter with a real default is fine. A module-level TESTING flag is not — the production path stops being the tested path.
• Fixing only the symptom. Adding a lock around the assignment but not the check leaves the bug intact. The lock must cover the whole check-then-act sequence, which the same test will confirm.

The fix, and confirming the test sees it

import threading


class ServiceRegistry:
    def __init__(self, build_client):
        self._build_client = build_client
        self._client = None
        self._lock = threading.Lock()
        self.build_count = 0

    def get_client(self):
        # Fast path: no lock once initialised.
        if self._client is None:
            with self._lock:
                # Re-check inside the lock: another thread may have built it
                # while this one was waiting for the lock.
                if self._client is None:
                    client = self._build_client()
                    self.build_count += 1
                    self._client = client
        return self._client

The inner re-check is the part people omit. Without it, both threads block on the lock, the first builds, and the second — already past the outer check — builds again the moment it acquires the lock. The test above catches exactly that mistake, which is why it is worth running against the partial fix as well as against no fix at all.

pytest tests/test_registry.py::test_client_is_built_exactly_once --count=10 -q

..........                                                    [100%]
10 passed in 0.14s

Ten passes with the full fix, ten failures without it, and ten failures with the lock but no re-check. That three-way check is what makes the test trustworthy rather than merely green.

Generalising the pattern

The registry above is one instance of a shape that recurs constantly, and recognising the shape is what makes the next reproduction quick rather than exploratory.

Every check-then-act race has the same three parts: a predicate read from shared state, an action that changes that state, and a gap between them that is not atomic. Naming those three for a given bug tells you immediately where the seam goes — inside the gap — and what to assert — the invariant the predicate was protecting.

The gap is not always visible as a line of code. It can be a property access that triggers a lazy import, a __getattr__ that queries a registry, or a logging call that formats an expensive repr — anything that yields control. That is why the seam-based approach beats reasoning: rather than proving the gap exists, the test simply holds it open and observes what another thread sees. Where the gap turns out not to exist, the test passes immediately and costs nothing; where it does, the failure names the invariant directly.

Reading the table the other way is also useful during review: any code matching column two without a lock spanning columns one and three is a race, whether or not anyone has seen it fail yet.

When no seam is available

Sometimes the window sits inside code you cannot change — a third-party client, a C extension wrapper, a function whose signature is fixed by a framework. Two techniques cover most of those cases.

Patch a collaborator rather than the function. The window is usually around a call to something slower: a database query, an HTTP request, a file read. Patching that collaborator with a slow version widens the window without touching the function under test.

from unittest.mock import patch


def test_double_write_without_modifying_the_service():
    entered, release = threading.Event(), threading.Event()
    original = repository.fetch

    def slow_fetch(*args, **kwargs):
        entered.set()
        release.wait(timeout=5)
        return original(*args, **kwargs)

    # The seam is the collaborator, which is already injectable via patching.
    with patch.object(repository, "fetch", slow_fetch):
        ...

Lower the switch interval and repeat. When there is no collaborator to patch either, sys.setswitchinterval(1e-6) makes CPython preempt threads far more aggressively, which widens every window in the process. Combined with a few thousand iterations it converts a one-in-a-million race into a one-in-ten, which is enough for a regression test if it is marked and kept out of the fast suite.

Whichever seam is used, restore the setting afterwards. sys.setswitchinterval is process-global, so leaving it at a microsecond for the rest of the session slows every subsequent test measurably and can itself introduce flakiness elsewhere — a fixture with a finally that restores the original value is the minimum discipline.

Frequently Asked Questions

Is adding a hook to production code just for a test acceptable? A single injected callable with a no-op default is acceptable and often clarifying — it names the point where concurrency matters. What is not acceptable is test-only branching inside the function, such as an if TESTING check, because that means production takes a different path from the one under test.

What if I cannot find the window? Widen the search by lowering sys.setswitchinterval so the interpreter preempts more aggressively, and run the operation from many threads with an invariant assertion. That turns a rare failure into a frequent one, which is enough to locate the window; then convert the frequent failure into a deterministic one with a seam.

Should the deterministic test replace the stress test? It should replace it in the fast suite. Keep a stress loop in a nightly job for the races nobody has thought of yet, but the known bug deserves a test that fails every time, not one that fails one run in fifty.

Related

• Testing Threads & Race Conditions — the barrier technique for when every thread runs the same code.
• Testing Thread Safety with Barriers and Events — the primitive-by-primitive patterns behind this coordination.
• Dumping Stacks on Deadlock with faulthandler — what to do when the fix introduces a lock-ordering problem.
• Patching Class Attributes with patch.object — the alternative seam when injection is not available.

← Back to Testing Threads & Race Conditions