Quick Start

Finding a race automatically

The front door is frontrun.explore(): hand it a way to build shared state, some workers, and an invariant, and DPOR systematically explores every meaningfully different thread interleaving until one breaks the invariant or all of them are proven safe. No annotations needed:

import frontrun

class Counter:
    def __init__(self):
        self.value = 0

    def increment(self):
        temp = self.value
        self.value = temp + 1

def test_increment_is_atomic():
    result = frontrun.explore(
        setup=Counter,
        workers=Counter.increment,
        count=2,
        invariant=lambda c: c.value == 2,
    )
    result.assert_holds()

Run it through the frontrun CLI wrapper — plain pytest skips frontrun.explore() tests (see Installation):

frontrun pytest test_counter.py

The failure output names the conflicting accesses, shows the interleaved source-line trace, and replays the counterexample schedule to confirm it reproduces deterministically. See DPOR in Practice for the practical guide, and How Frontrun Works for the other exploration strategies (random bytecode exploration, marker schedule exploration).

The rest of this guide covers trace markers — the fully manual approach, for when you already know the interleaving you’re after and want to pin it down as a deterministic regression test. If your workers are separate OS processes contending on a shared database or Redis rather than threads, see Cross-Process Exploration. If the race you’re hunting involves timeouts, retries, or TTL expiry, see Virtual clock: timeout, retry, and TTL races (clock="virtual" / clock="explored").

Triggering a Race Condition with Trace Markers

from frontrun.common import Schedule, Step
from frontrun.trace_markers import TraceExecutor

class Counter:
    def __init__(self):
        self.value = 0

    def increment(self):
        temp = self.value  # frontrun: read_value
        temp += 1
        self.value = temp  # frontrun: write_value

def test_counter_lost_update():
    counter = Counter()

    # Both threads read before either writes, causing a lost update
    schedule = Schedule([
        Step("thread1", "read_value"),    # T1 reads 0
        Step("thread2", "read_value"),    # T2 reads 0 (both see same value!)
        Step("thread1", "write_value"),   # T1 writes 1
        Step("thread2", "write_value"),   # T2 writes 1 (overwrites T1's update!)
    ])

    executor = TraceExecutor(schedule)
    executor.run({
        "thread1": counter.increment,
        "thread2": counter.increment,
    }, timeout=5.0)

    assert counter.value == 1  # One increment lost

How Trace Markers Work

Trace markers are comments of the form # frontrun: <name> that tell Frontrun where synchronization points are. A marker gates the code that follows it: when a thread reaches a marker, it pauses until the scheduler grants it a turn. Only then does the gated code execute.

Under the hood, each thread runs with a sys.settrace callback that fires on every source line. The callback checks whether the line contains a # frontrun: comment (via a MarkerRegistry that caches marker locations per file). When a marker is hit, the thread blocks on a threading.Condition until the scheduler advances to that step.

Two placement styles are supported:

1. Inline with code (marker on the same line as the operation it gates):

   def increment(self):
       temp = self.value  # frontrun: read_value
       temp += 1
       self.value = temp  # frontrun: write_value
   

   Here read_value gates the read of self.value, and write_value gates the write.

2. On a separate line before the operation:

   def increment(self):
       # frontrun: read_value
       temp = self.value
       temp += 1
       # frontrun: write_value
       self.value = temp
   

   The semantics are the same: the marker gates the next executable line.

Name markers after the operation they gate (read_value, write_balance, acquire_lock, etc.) rather than using temporal prefixes like before_ or after_.

Creating Schedules

A schedule defines the execution order of marked synchronization points:

from frontrun.common import Schedule, Step

schedule = Schedule([
    Step("thread1", "marker_name_1"),
    Step("thread1", "marker_name_2"),
    Step("thread2", "marker_name_1"),
    Step("thread2", "marker_name_2"),
])

Each Step specifies the thread/task name and the marker name to execute at that step.

Running with Controlled Interleaving

Pass a {name: callable} dict to run() to start all threads and wait for them in a single call:

from frontrun.trace_markers import TraceExecutor

executor = TraceExecutor(schedule)
executor.run({
    "thread1": task_function_1,
    "thread2": task_function_2,
}, timeout=5.0)

This is the preferred form and matches the async API exactly.

Async Support

Async trace markers use the same comment-based syntax. Each async task runs in its own thread (via asyncio.run), with sys.settrace controlling interleaving between tasks.

A marker gates the next await expression. When a task reaches a marker, it pauses until the scheduler grants it a turn; only then does the gated await execute. Between two markers the task runs without interruption from other scheduled tasks.

from frontrun import TraceExecutor
from frontrun.common import Schedule, Step

class AsyncCounter:
    def __init__(self):
        self.value = 0

    async def get_value(self):
        return self.value

    async def set_value(self, new_value):
        self.value = new_value

    async def increment(self):
        # frontrun: read_value
        temp = await self.get_value()
        # frontrun: write_value
        await self.set_value(temp + 1)

def test_async_counter_lost_update():
    counter = AsyncCounter()

    # Both tasks read before either writes --- triggers the lost update
    schedule = Schedule([
        Step("task1", "read_value"),
        Step("task2", "read_value"),
        Step("task1", "write_value"),
        Step("task2", "write_value"),
    ])

    executor = TraceExecutor(schedule)
    executor.run({
        "task1": counter.increment,
        "task2": counter.increment,
    })

    # Both tasks read 0, then both write 1 --- one increment is lost
    assert counter.value == 1

def test_async_counter_serialized():
    counter = AsyncCounter()

    # Serialized: task1 completes before task2 starts
    schedule = Schedule([
        Step("task1", "read_value"),
        Step("task1", "write_value"),
        Step("task2", "read_value"),
        Step("task2", "write_value"),
    ])

    executor = TraceExecutor(schedule)
    executor.run({
        "task1": counter.increment,
        "task2": counter.increment,
    })

    assert counter.value == 2  # No lost update

Prefer assert_holds() over manual asserts

Instead of writing assert result.property_holds, result.explanation after every exploration call, use the convenience helper InterleavingResult.assert_holds():

result = frontrun.explore(setup=setup, workers=[thread1, thread2], invariant=invariant)
result.assert_holds()  # raises AssertionError with explanation on failure

An optional msg_prefix is prepended to the explanation, which is handy when multiple assertions appear in one test:

result.assert_holds(msg_prefix="counter race: ")