API Reference¶

Common Data Structures¶

Shared data structures for frontrun.

frontrun.common.any_async(fns)[source]¶

Return True if any element is a coroutine function.

Non-callables are ignored so callers can pass dicts of name -> value directly.

Parameters:: fns (Iterable[Any])
Return type:: bool

frontrun.common.check_invariant(invariant, state)[source]¶

Evaluate invariant on state, tolerating AssertionError.

Returns (failed, assertion_message). failed is True when the invariant returns a falsy value or raises AssertionError. When AssertionError was raised, its message is returned in the second slot so callers can fold it into their result’s explanation.

Parameters:

invariant (Callable[[Any], Any])
state (Any)

Return type:

tuple[bool, str | None]

exception frontrun.common.NondeterministicSQLError[source]¶

Bases: Exception

Raised when SQL INSERT statements are detected during exploration.

Autoincrement/SERIAL/IDENTITY columns assign IDs based on execution order, making test results non-deterministic across interleavings. Pre-allocate rows with explicit IDs in your test setup instead.

Pass warn_nondeterministic_sql=False to suppress this check if you understand the implications.

class frontrun.common.Step(execution_name, marker_name)[source]¶

Bases: object

Represents a single step in the execution schedule.

Parameters:

execution_name (str)
marker_name (str)

execution_name¶

The name of the execution unit (thread/task) that should execute this step

Type:: str

marker_name¶

The marker name that identifies this synchronization point

Type:: str

execution_name: str¶

marker_name: str¶

class frontrun.common.Schedule(steps)[source]¶

Bases: object

Defines the execution order for tasks at synchronization points.

A schedule is a linear sequence of steps that specify which task should execute which marker in order.

Initialize a schedule with a list of steps.

Parameters:: steps (list[Step]) – Ordered list of Step objects defining the execution sequence

__init__(steps)[source]¶

Initialize a schedule with a list of steps.

Parameters:: steps (list[Step]) – Ordered list of Step objects defining the execution sequence

class frontrun.common.InterleavingResult(property_holds, counterexample=None, num_explored=0, unique_interleavings=0, failures=<factory>, explanation=None, reproduction_attempts=0, reproduction_successes=0, sql_anomaly=None, races_detected=False, exhausted=None, failure_kind=None, inconclusive_reason=None)[source]¶

Bases: object

Result of exploring interleavings.

Returned by frontrun.explore().

Parameters:

property_holds (bool | None)
counterexample (list[int] | Schedule | None)
num_explored (int)
unique_interleavings (int)
failures (list[tuple[int, list[int]]])
explanation (str | None)
reproduction_attempts (int)
reproduction_successes (int)
sql_anomaly (SqlAnomaly | None)
races_detected (bool)
exhausted (bool | None)
failure_kind (str | None)
inconclusive_reason (str | None)

property_holds¶

Tri-state verdict. True is a pass certificate (at least one interleaving completed, every worker body ran, no coverage-degrading event) — it can only be produced by frontrun._certificate.certify_pass(). False means a failure was found and implies a counterexample/failure record exists. None means the exploration was inconclusive (no evidence either way — e.g. a budget expired before any interleaving completed); see inconclusive_reason. None is falsy, so if result.property_holds: stays fail-closed.

Type:: bool | None

counterexample¶

First schedule that violated the invariant (if any).

Type:: list[int] | Schedule | None

num_explored¶

How many interleavings were tested.

Type:: int

unique_interleavings¶

Number of distinct schedule orderings observed. Provides a lower bound on interleaving-space coverage. Relevant for random bytecode exploration; DPOR always explores distinct interleavings so this equals num_explored.

Type:: int

failures¶

All failing (execution_number, schedule) pairs. Populated by DPOR (thread and process execution); holds every failing execution when stop_on_first=False, otherwise at most the first.

Type:: list[tuple[int, list[int]]]

explanation¶

Human-readable explanation of the race condition, showing interleaved source lines and the conflict pattern. None if no race was found.

Type:: str | None

reproduction_attempts¶

Number of times the counterexample schedule was re-run to test reproducibility. 0 if no counterexample.

Type:: int

reproduction_successes¶

How many of those re-runs reproduced the invariant violation.

Type:: int

sql_anomaly¶

Classified SQL isolation anomaly (if any SQL I/O events were recorded). A SqlAnomaly instance, or None if the failure did not involve SQL.

Type:: SqlAnomaly | None

exhausted¶

Whether the search space was fully covered. Populated by execution="process" (from CrossProcessResult .exhausted); None means the mode that produced this result does not report it (thread/async execution currently leaves it unset). A preemption-bounded DPOR search (the default, preemption_bound=2) never claims True — full coverage requires preemption_bound=None.

Type:: bool | None

failure_kind¶

Structured category of the failure for execution="process" — one of "invariant", "worker_error", "deadlock", "timeout", "nondeterministic", "step_limit", "branch_limit". None when the invariant held or for thread/async execution (which encodes the failure in explanation only).

Type:: str | None

inconclusive_reason¶

Machine-readable cause (and remedy) when property_holds is None — e.g. “total_timeout=0.01s elapsed before any interleaving completed; increase total_timeout”. None for pass/fail verdicts.

Type:: str | None

property_holds: bool | None¶

counterexample: list[int] | Schedule | None = None¶

num_explored: int = 0¶

unique_interleavings: int = 0¶

failures: list[tuple[int, list[int]]]¶

explanation: str | None = None¶

reproduction_attempts: int = 0¶

reproduction_successes: int = 0¶

sql_anomaly: SqlAnomaly | None = None¶

races_detected: bool = False¶

exhausted: bool | None = None¶

failure_kind: str | None = None¶

inconclusive_reason: str | None = None¶

assert_holds(msg_prefix='', *, allow_inconclusive=False)[source]¶

Raise unless the exploration produced a pass certificate.

Prefer this over assert result.property_holds, result.explanation.

Parameters:

msg_prefix (str) – Optional string prepended to the message. Useful for identifying which assertion failed when multiple calls appear in one test.
allow_inconclusive (bool) – Opt into the weaker claim “no failure found”: do not raise when the result is inconclusive (property_holds=None). A genuine failure still raises.

Raises:

AssertionError – A counterexample was found (property_holds is False); the message carries the explanation.
InconclusiveExploration – The exploration was inconclusive (property_holds is None) and allow_inconclusive was not set; the message names cause and remedy.

Return type:

None

frontrun.common.compute_serializable_states(setup, thread_funcs, state_hash=None)[source]¶

Compute the set of valid serializable states.

Runs all N! sequential orderings of the thread functions and collects the hash of each resulting state. An interleaved execution is serializable if its final state hash is in this set.

Parameters:

setup (Callable[[], Any]) – Factory that creates fresh shared state.
thread_funcs (list[Callable[[Any], None]]) – Thread/task functions (each takes state as argument).
state_hash (Callable[[Any], Any] | None) – Hash function for state. If None, uses repr().

Returns:

Set of valid state hashes.

Return type:

set[Any]

async frontrun.common.compute_serializable_states_async(setup, task_funcs, state_hash=None)[source]¶

Async version of compute_serializable_states.

Runs all N! sequential orderings of async task functions.

Parameters:

setup (Callable[[], Any])
task_funcs (list[Callable[[Any], Any]])
state_hash (Callable[[Any], Any] | None)

Return type:

set[Any]

frontrun.common.resolve_serializable_hash_fn(serializable_invariant)[source]¶

Extract the state-hash function from a serializable_invariant parameter.

Returns the callable itself when it is a hash function, or None when the caller passed True (meaning “use the default repr”).

Parameters:: serializable_invariant (Callable[[Any], Any] | bool)
Return type:: Callable[[Any], Any] | None

frontrun.common.check_serializability_violation(state, serial_valid_states, hash_fn, execution_num)[source]¶

Check whether state violates serializability.

Returns an explanation string when the state hash is not in serial_valid_states, or None if it passes.

hash_fn should be the resolved state-hash function (use resolve_serializable_hash_fn() to convert the raw serializable_invariant parameter, falling back to repr when it returns None).

Parameters:

state (Any)
serial_valid_states (set[Any])
hash_fn (Callable[[Any], Any])
execution_num (int)

Return type:

str | None

Unified Exploration Entry Points¶

frontrun.explore(setup, workers, invariant, *, count=None, strategy='dpor', execution='thread', clock='real', clock_diagnostics=False, max_executions=None, preemption_bound=2, max_branches=100000, timeout_per_run=5.0, stop_on_first=True, detect_io=True, reuse_workers=False, deadlock_timeout=None, reproduce_on_failure=10, total_timeout=None, warn_nondeterministic_sql=True, lock_timeout=None, trace_packages=None, track_dunder_dict_accesses=False, search=None, patch_sleep=True, serializable_invariant=False, error_on_any_race=False, max_attempts=200, max_ops=None, seed=None, debug=False, detect_sql=False)[source]¶

Explore thread/task interleavings for concurrency bugs.

A unified entry point that dispatches to the appropriate underlying implementation based on worker type (sync vs async) and strategy.

Parameters:

setup (Callable[[], Any]) – Creates fresh shared state for each execution.
workers (Callable[[Any], Any] | list[Callable[[Any], Any]] | tuple[Callable[[Any], Any], ...]) – A callable (when count is provided) or a list/tuple of callables. Sync callables run as threads; async callables (coroutine functions) run as asyncio tasks.
invariant (Callable[[Any], bool]) – Predicate over shared state; must be True after all workers complete.
count (int | None) – When workers is a single callable, replicate it this many times. Must be positive. Cannot be used when workers is a list/tuple.
strategy (Literal['dpor', 'random']) – "dpor" (default) for systematic DPOR exploration, or "random" for random schedule sampling.
execution (Literal['thread', 'process']) – "thread" (default) runs workers as threads/async tasks in this process; "process" runs each worker in its own spawned Python process, coordinating over a socket. Process mode has the same setup / workers / invariant / count shape; workers and the setup() return value are serialised with dill (so closures and lambdas work, not just module-level functions), and setup() should return a handle to external SQL/Redis state (a DB path/URL, not a live connection). Supports strategy="dpor" with sync workers only and needs the process extra (pip install frontrun[process]). See Cross-Process Exploration.
reuse_workers (bool) – Process execution only. Spawn each worker process once and re-run it per interleaving instead of respawning (amortises spawn cost). Thread execution rejects an explicit reuse_workers=True with ValueError (there are no worker processes to reuse).
max_executions (int | None) – Safety limit on total executions (DPOR only).
preemption_bound (int | None) – Limit on preemptions per execution (DPOR only).
max_branches (int) – Maximum scheduling points per execution (DPOR only).
timeout_per_run (float) – Timeout for each individual run.
stop_on_first (bool) – Stop on first invariant violation (DPOR only).
detect_io (bool) – Detect socket/file I/O operations as resource accesses. For async DPOR, also activates Redis key-level patching. For the async random strategy the flag is narrower: it only gates SQL driver patching (detect_io=True implies detect_sql=True); socket/file/Redis detection is not available on that path. Note the difference from the standalone frontrun.explore_async_random(), whose detect_sql defaults to False — going through explore(strategy="random") with async workers patches SQL drivers by default because detect_io defaults to True here.
deadlock_timeout (float | None) – Seconds to wait before declaring a deadlock. Defaults to 5.0 for thread execution and 15.0 for process execution (spawning processes is slower), unless set explicitly.
reproduce_on_failure (int) – Replay counterexample this many times (not supported for async random).
total_timeout (float | None) – Maximum total exploration time in seconds.
warn_nondeterministic_sql (bool) – Raise on nondeterministic SQL INSERT (not supported for async random).
lock_timeout (int | None) – Auto-set PostgreSQL lock_timeout (milliseconds; DPOR only).
trace_packages (list[str] | None) – Package patterns to trace in addition to user code.
track_dunder_dict_accesses (bool) – Report obj.__dict__ accesses (sync DPOR only).
search (str | None) – Wakeup-tree traversal strategy (sync DPOR and process execution only).
patch_sleep (bool) – For clock="real", make time.sleep / asyncio.sleep cooperative zero-wall-time yields. For clock="virtual" or "explored", required: positive sleeps become scheduler-owned virtual deadlines and sleep(0) remains a yield.
serializable_invariant (Callable[[Any], Any] | bool) – Check serializability against sequential runs.
error_on_any_race (bool) – Treat unsynchronized races as failures (DPOR only; the random strategies reject True with their own ValueError rather than silently ignoring it).
clock (Literal['real', 'virtual', 'explored']) – "real" (default) leaves time untouched. "virtual" gives each execution a scheduler-owned virtual clock: explored code reads virtual time from time.time() / time.monotonic() / time.perf_counter(), sleeps become zero-wall-time virtual deadlines, timed lock acquires time out deterministically, and the clock autojumps to the earliest pending deadline when nothing is runnable. "explored" additionally makes the clock advance a schedulable choice, so timer firings are explored against other operations (“the retry fired between the read and the write”). Rule of thumb: use "virtual" to make timeout/TTL logic reachable deterministically at zero wall cost; add "explored" when the timing of a timer firing is itself the race you are hunting. Works with both strategies, sync and async. Requires patch_sleep=True; not supported with execution="process" (worker processes read real time) or serializable_invariant (the sequential baseline runs outside the scheduler). See Virtual clock: timeout, retry, and TTL races.
clock_diagnostics (bool) – When using a virtual clock, warn when traced worker frames hold references to real time.* functions captured before frontrun patched the time module. Diagnostics do not change scheduling behavior. Requires frame tracing: DPOR and sync random can emit diagnostics; async random accepts the option for API compatibility but cannot inspect frames.
max_attempts (int) – Random schedule samples to try (random strategy only).
max_ops (int | None) – Maximum schedule length per attempt (random strategy only).
seed (int | None) – RNG seed for reproducibility (random strategy only).
debug (bool) – Enable debug output (sync random only).
detect_sql (bool) – Patch async SQL drivers (async workers only; detect_io=True already implies it).

Returns:

InterleavingResult (sync) or a coroutine that resolves to one (async workers).

Raises:

ValueError – If count and a list of workers are both provided, count <= 0, strategy, execution or clock is unrecognised, or a non-real clock is combined with patch_sleep=False, serializable_invariant, or execution="process". Also raised for any explicitly-passed option the selected strategy/mode does not support, rather than silently ignoring it: e.g. seed= with strategy="dpor", preemption_bound= with strategy="random", reproduce_on_failure= with async strategy="random", detect_sql= with sync workers, or reuse_workers= with execution="thread". execution="process" additionally rejects async workers, strategy="random", and every option that requires the in-process scheduler (serializable_invariant, error_on_any_race, lock_timeout, trace_packages, track_dunder_dict_accesses, detect_sql, non-real clock / clock_diagnostics, and non-default detect_io, patch_sleep, timeout_per_run, reproduce_on_failure, warn_nondeterministic_sql, max_attempts, max_ops, seed, debug). Explicit-option detection is value-based: passing an option at its default value is indistinguishable from omitting it, and is accepted (a no-op either way).

Return type:

Any

frontrun.explore_random(setup, threads, invariant, max_attempts=200, max_ops=300, timeout_per_run=5.0, seed=None, debug=False, detect_io=True, deadlock_timeout=5.0, reproduce_on_failure=10, total_timeout=None, warn_nondeterministic_sql=True, trace_packages=None, patch_sleep=True, serializable_invariant=False, error_on_any_race=False, clock='real', clock_diagnostics=False)[source]¶

Search for interleavings that violate an invariant.

Note

When running under pytest, this function requires the frontrun CLI wrapper (frontrun pytest ...) or the --frontrun-patch-locks flag. Without it, the test is automatically skipped.

Generates random opcode-level schedules and tests whether the invariant holds under each one. If a violation is found, returns immediately with the counterexample schedule.

This is the bytecode-level analogue of property-based testing: instead of generating random inputs, we generate random interleavings and check that the result satisfies an invariant.

Parameters:

setup (Callable[[], T]) – Returns fresh shared state for each attempt.
threads (list[Callable[[T], None]]) – Callables that each receive the state as their argument.
invariant (Callable[[T], bool]) – Predicate on the state. Returns True if the property holds.
max_attempts (int) – How many random interleavings to try.
max_ops (int) – Maximum randomly sampled schedule-prefix length per attempt. If workers need more turns to finish, the scheduler extends the prefix deterministically and records those turns in any returned counterexample.
timeout_per_run (float) – Timeout for each individual run.
seed (int | None) – Optional RNG seed for reproducibility.
detect_io (bool) – Automatically detect socket/file I/O and treat them as scheduling points (default True).
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).
reproduce_on_failure (int) – When a counterexample is found, replay the same schedule this many times to measure reproducibility (default 10). Set to 0 to skip reproduction testing.
total_timeout (float | None) – Maximum total time in seconds for the entire exploration (default None = unlimited). When exceeded, returns results gathered so far.
warn_nondeterministic_sql (bool) – If True (default), raise NondeterministicSQLError when SQL INSERT statements are detected but lastrowid capture failed (e.g. psycopg2 without RETURNING). Set to False to suppress. When capture succeeds, INSERTs use stable indexical resource IDs automatically.
trace_packages (list[str] | None) – List of package name patterns (fnmatch syntax) to trace in addition to user code. By default, code in site-packages is skipped. Use this to include specific installed packages, e.g. ["django_*", "mylib.*"].
patch_sleep (bool) – If True (default), time.sleep yields to the scheduler instead of blocking. Required for clock != "real".
serializable_invariant (Callable[[T], Any] | bool) – Check serializability against sequential runs. Cannot be combined with a virtual clock.
error_on_any_race (bool) – Not supported here — requires the DPOR strategy.
clock (Literal['real', 'virtual', 'explored']) – "real" (default), "virtual" (autojump virtual clock: time reads are virtual, sleeps cost zero wall time and jump the clock when nothing else can run), or "explored" (schedule entries landing on a sleeping thread advance the clock, so the random sampler also explores early timer firings). See Virtual clock: timeout, retry, and TTL races.
clock_diagnostics (bool) – With a virtual clock, warn when traced worker frames hold captured real time.* clock-read functions.
debug (bool)

Returns:

InterleavingResult with the outcome. The unique_interleavings field reports how many distinct execution orderings were observed, providing a lower bound on exploration coverage.

Return type:

InterleavingResult

async frontrun.explore_async_random(setup, tasks, invariant, max_attempts=200, max_ops=100, timeout_per_run=5.0, seed=None, deadlock_timeout=5.0, detect_sql=False, trace_packages=None, patch_sleep=True, serializable_invariant=False, error_on_any_race=False, total_timeout=None, clock='real', clock_diagnostics=False)[source]¶

Search for async interleavings that violate an invariant.

Generates random await-point-level schedules and tests whether the invariant holds under each one. If a violation is found, returns immediately with the counterexample schedule.

This is the async analogue of property-based testing for concurrency: instead of generating random inputs, we generate random interleavings and check that the result satisfies an invariant.

Note: max_ops defaults to 100 (vs 300 for bytecode.py) because async code has far fewer interleaving points than threaded bytecode execution. Each await_point() call represents a much coarser-grained checkpoint.

Parameters:

setup (Callable[[], Any]) – Returns fresh shared state for each attempt.
tasks (list[Callable[[Any], Coroutine[Any, Any, None]]]) – Async callables that each receive the state as their argument.
invariant (Callable[[Any], bool]) – Predicate on the state. Returns True if the property holds.
max_attempts (int) – How many random interleavings to try.
max_ops (int) – Maximum schedule length per attempt.
timeout_per_run (float) – Timeout for each individual run.
seed (int | None) – Optional RNG seed for reproducibility.
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).
detect_sql (bool) – If True, patch async DBAPI drivers (aiosqlite, psycopg AsyncCursor, aiomysql, asyncpg) to intercept SQL and report table-level conflicts.
trace_packages (list[str] | None) – Accepted for API compatibility but not used. The async shuffler operates at await-point granularity and does not perform file-level tracing.
patch_sleep (bool) – If True (default), asyncio.sleep yields to the scheduler instead of waiting. Required for clock != "real".
serializable_invariant (Callable[[Any], Any] | bool) – Check serializability against sequential runs. Cannot be combined with a virtual clock.
error_on_any_race (bool) – Not supported here — requires the DPOR strategy.
clock (Literal['real', 'virtual', 'explored']) – "real" (default), "virtual" (autojump virtual clock: time reads are virtual, asyncio.sleep costs zero wall time), or "explored" (schedule entries landing on a sleeping task advance the clock, exploring early timer firings). Tasks that block on primitives the scheduler cannot see (e.g. a raw asyncio.Lock) are handled by a quiescence heuristic; prefer the DPOR strategy for lock-heavy async code. asyncio.wait_for, asyncio.timeout, and asyncio.timeout_at inside explored tasks use virtual deadlines. See Virtual clock: timeout, retry, and TTL races.
clock_diagnostics (bool) – Accepted for API consistency. Async random does not trace worker frames, so captured time.* references cannot be diagnosed and a RuntimeWarning is emitted.
total_timeout (float | None)

Returns:

InterleavingResult with the outcome. The unique_interleavings field reports how many distinct schedule orderings were observed.

Return type:

InterleavingResult

Virtual Clock¶

See Virtual clock: timeout, retry, and TTL races for a guide. frontrun.explore(..., clock="virtual") / clock="explored" control time as a scheduled quantity; there is no separate entry point.

frontrun.ClockMode is the public type alias Literal["real", "virtual", "explored"] used by the clock= parameter. frontrun.VIRTUAL_EPOCH is the deterministic starting timestamp used by each virtual-clock execution; application code normally should compare times within one execution rather than depend on its numeric value.

Cross-Process Exploration¶

See Cross-Process Exploration for a guide. frontrun.explore(..., execution="process") mirrors the thread/async interface; the functions below are the lower-level entry point.

frontrun.explore_processes(processes, *, setup, invariant, count=None, strategy='dpor', max_iterations=4096, max_steps_per_run=100000, max_executions=None, preemption_bound=2, max_branches=100000, total_timeout=None, stop_on_first=True, search=None, deadlock_timeout=15.0, reuse_workers=False)[source]¶

Explore interleavings of processes contending on shared external state.

processes is a mapping of label → Subprocess (preserved as result.worker_labels), a plain sequence, or a single Subprocess with count=N to replicate it (the mirror of explore(workers=fn, count=N)).

setup resets the external state before each interleaving and returns a handle to it (e.g. a DB URL / connection info). That handle is passed to invariant(state), which checks the state afterwards and returns a bool — matching explore(execution="process"). Both run in this (coordinator) process; invariant may ignore state and read the shared store directly.

strategy:

"dpor" (default) drives the Rust DPOR engine, pruning equivalent interleavings (partial-order reduction) and detecting cross-worker SELECT FOR UPDATE deadlocks. max_executions / preemption_bound / max_branches / total_timeout bound the search, search selects the wakeup-tree traversal order, and stop_on_first=False keeps exploring after a failure, accumulating every failing execution in CrossProcessResult.failures. exhausted=True (full coverage) requires preemption_bound=None; the default bound (2) truncates the search, so bounded runs report exhausted=False.
"exhaustive" enumerates every interleaving at external-access granularity, bounded by max_iterations and max_steps_per_run per execution. Useful as a reduction-free cross-check.

Each strategy rejects the other’s bounds when passed explicitly (a silently ignored option is a correctness footgun): max_iterations is exhaustive-only; the DPOR knobs above are DPOR-only. Explicit-option detection is value-based, so passing a knob at its default value is indistinguishable from omitting it.

Parameters:

processes (Mapping[str, Subprocess] | Sequence[Subprocess] | Subprocess)
setup (Callable[[], Any])
invariant (Callable[[Any], bool])
count (int | None)
strategy (Literal['dpor', 'exhaustive'])
max_iterations (int)
max_steps_per_run (int)
max_executions (int | None)
preemption_bound (int | None)
max_branches (int)
total_timeout (float | None)
stop_on_first (bool)
search (str | None)
deadlock_timeout (float)
reuse_workers (bool)

Return type:

CrossProcessResult

class frontrun.Subprocess(target, args=<factory>)[source]¶

A worker to spawn: a "module:callable" target and its positional args.

args are passed to the spawned process as JSON through the environment, so they must be JSON-serialisable and survive a JSON round-trip: a tuple arrives as a list and a dict with non-string keys comes back string-keyed. Pass plain scalars / lists / string-keyed dicts, or use frontrun.explore(execution="process") (which pickles) for richer args. The callable must be synchronous; async/awaitable targets are rejected. It runs in the child with frontrun’s SQL interception routed to the coordinator.

Parameters:

target (str)
args (tuple[Any, ...])

class frontrun.CrossProcessResult(ok, iterations, exhausted, failing_schedule=None, failure=None, failure_kind=None, accesses=None, worker_labels=<factory>, failures=<factory>, workers_executed=<factory>, truncation=None)[source]¶

Outcome of a cross-process exploration.

Parameters:

ok (bool)
iterations (int)
exhausted (bool)
failing_schedule (list[int] | None)
failure (str | None)
failure_kind (str | None)
accesses (list[tuple[int, str, str]] | None)
worker_labels (dict[int, str])
failures (list[tuple[int, list[int]]])
workers_executed (list[bool])
truncation (str | None)

Trace Markers¶

Frontrun: Deterministic thread interleaving using sys.settrace and comment markers.

This module provides a mechanism to control thread execution order by marking synchronization points in code with special comments and enforcing a predefined execution schedule using Python’s tracing facilities.

Example usage:

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

def worker_function():: x = read_data() # frontrun: read write_data(x) # frontrun: write
schedule = Schedule([: Step(“thread1”, “read”), Step(“thread2”, “read”), Step(“thread1”, “write”), Step(“thread2”, “write”),

])

executor = TraceExecutor(schedule) executor.run({“thread1”: worker_function, “thread2”: worker_function}) ```

class frontrun.trace_markers.MarkerRegistry[source]¶

Bases: object

Tracks marker locations in source code for efficient lookup.

This class scans source files to find lines with frontrun markers and maintains a mapping from (filename, line_number) to marker names.

scan_frame(frame)[source]¶

Scan the source file for the given frame to find all markers.

Parameters:: frame (Any) – A Python frame object from the trace function
Return type:: None

get_marker(filename, lineno)[source]¶

Get the marker name for a specific file location.

Parameters:

filename (str) – The source file path
lineno (int) – The line number

Returns:

The marker name if found, None otherwise

Return type:

str | None

class frontrun.trace_markers.ThreadCoordinator(schedule, *, deadlock_timeout=5.0)[source]¶

Bases: object

Coordinates thread execution according to a schedule.

This class manages the synchronization between threads, ensuring that each thread executes markers in the order specified by the schedule.

Parameters:

schedule (Schedule)
deadlock_timeout (float)

wait_for_turn(execution_name, marker_name, *, _reacquire_execution_lock=False)[source]¶

Block until it’s this execution unit’s turn to execute this marker.

When _reacquire_execution_lock is True (used by the trace executors), _execution_lock is acquired while the condition lock is still held, before returning. This prevents other threads from racing ahead between being notified and the caller resuming execution. The caller must have already released _execution_lock before calling this method.

Parameters:

execution_name (str) – The name of the calling execution unit
marker_name (str) – The marker that was hit
_reacquire_execution_lock (bool)

report_error(error)[source]¶

Report an error and wake up all waiting threads.

Parameters:: error (Exception) – The exception that occurred

is_finished()[source]¶

Check if the schedule has completed or encountered an error.

Return type:: bool

class frontrun.trace_markers.TraceExecutor(schedule, *, deadlock_timeout=5.0)[source]¶

Bases: object

Facade over sync and async marker-based schedule execution.

Pass a dict mapping thread/task names to zero-argument callables (sync or async). run starts every worker and waits for them in one call:

executor = TraceExecutor(schedule)
executor.run({"t1": worker1, "t2": worker2})

Async usage accepts the async-task mapping the same way:

executor = TraceExecutor(schedule)
executor.run({"task1": coro1, "task2": coro2})

Parameters:

schedule (Schedule)
deadlock_timeout (float)

run(tasks, *, timeout=None)[source]¶

Run all named workers under schedule control and wait for completion.

Pass a dict mapping thread/task names to zero-argument callables. run starts every worker and joins them in a single call:

executor.run({"thread1": fn1, "thread2": fn2}, timeout=5.0)

Async coroutines in the dict are dispatched to the async executor.

Parameters:

tasks (dict[str, Callable[[...], Any]]) – Mapping of execution-unit name to a zero-argument callable.
timeout (float | None) – Total wait timeout in seconds.

Return type:

None

wait(timeout=None)[source]¶

Parameters:: timeout (float | None)
Return type:: None

reset()[source]¶

Return type:: None

property threads: list[Thread]¶

property thread_errors: dict[str, BaseException]¶

property coordinator: ThreadCoordinator¶

property marker_registry: MarkerRegistry¶

property task_errors: dict[str, Exception]¶

frontrun.trace_markers.all_marker_schedules(threads)[source]¶

Enumerate ALL valid interleavings of thread markers.

A valid interleaving places every thread’s markers in the schedule while preserving their relative order within each thread.

For N threads with marker counts k1, k2, …, kN, the total number of valid interleavings is the multinomial coefficient:

(k1 + k2 + ... + kN)! / (k1! * k2! * ... * kN!)

Parameters:: threads (dict[str, list[str]]) – Mapping from thread/execution names to their ordered list of marker names.
Returns:: A list of Schedule objects covering every valid interleaving.
Return type:: list[Schedule]

Example:

schedules = all_marker_schedules(
    threads={"t1": ["a", "b"], "t2": ["x", "y"]},
)
assert len(schedules) == 6  # C(4,2)

frontrun.trace_markers.explore_marker_interleavings(setup, threads, invariant, *, stop_on_first=True, deadlock_timeout=5.0, timeout=10.0)[source]¶

Explore all marker-level interleavings and check an invariant.

Generates every valid interleaving of the declared markers (preserving per-thread order), runs each one against real code via TraceExecutor, and checks the invariant after each execution.

This sits between manual trace markers (exact schedule, one interleaving) and bytecode exploration (random, enormous search space). For N threads with a few markers each, the search space is small enough to explore exhaustively — giving completeness guarantees at the marker granularity.

Parameters:

setup (Callable[[...], Any]) – Factory producing fresh shared state for each execution.
threads (dict[str, tuple[Callable[[...], None], list[str]]]) – Mapping from execution name to (target_fn, markers) where target_fn takes the setup result and markers is the ordered list of # frontrun: marker names that target_fn hits.
invariant (Callable[[...], bool]) – Predicate on the shared state; returns True if correct.
stop_on_first (bool) – Stop after finding the first invariant violation (default True).
deadlock_timeout (float) – Per-schedule deadlock detection timeout.
timeout (float | None) – Per-schedule join timeout.

Returns:

An InterleavingResult. The counterexample field is a Schedule (not a list of ints) when a violation is found.

Return type:

InterleavingResult

frontrun.trace_markers.frontrun(schedule, threads, thread_args=None, thread_kwargs=None, timeout=None, deadlock_timeout=5.0)[source]¶

Convenience function to run multiple threads with a schedule.

Parameters:

schedule (Schedule) – The Schedule defining execution order
threads (dict[str, Callable[[...], None]]) – Dictionary mapping execution unit names to their target functions
thread_args (dict[str, tuple[Any, ...]] | None) – Optional dictionary mapping execution unit names to argument tuples
thread_kwargs (dict[str, dict[str, Any]] | None) – Optional dictionary mapping execution unit names to keyword argument dicts
timeout (float | None) – Optional timeout for waiting
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).

Returns:

The TraceExecutor instance (useful for inspection)

Return type:

TraceExecutor

Example

```python frontrun(

schedule=Schedule([Step(“t1”, “marker1”), Step(“t2”, “marker1”)]), threads={“t1”: worker_func, “t2”: worker_func},

)¶

frontrun.trace_markers.marker_schedule_strategy(threads)[source]¶

Hypothesis strategy that generates valid marker-level Schedule objects.

A valid schedule interleaves each thread’s markers while preserving their relative order within each thread. This provides a much smaller search space than opcode-level exploration while still covering all meaningful interleavings at the marker granularity.

Parameters:

threads (dict[str, list[str]]) –

Mapping from thread/execution names to their ordered list of marker names. Example:

{"t1": ["read", "write"], "t2": ["read", "write"]}

Returns:

A Hypothesis strategy producing Schedule objects.

Return type:

Any

Example:

from hypothesis import given
from frontrun.trace_markers import marker_schedule_strategy

@given(schedule=marker_schedule_strategy(
    threads={"w1": ["read", "write"], "w2": ["read", "write"]},
))
def test_no_lost_update(schedule):
    ...

Marker Schedule Exploration¶

frontrun.trace_markers.marker_schedule_strategy(threads)[source]¶

Hypothesis strategy that generates valid marker-level Schedule objects.

Parameters:

threads (dict[str, list[str]]) –

Mapping from thread/execution names to their ordered list of marker names. Example:

{"t1": ["read", "write"], "t2": ["read", "write"]}

Returns:

A Hypothesis strategy producing Schedule objects.

Return type:

Any

Example:

from hypothesis import given
from frontrun.trace_markers import marker_schedule_strategy

@given(schedule=marker_schedule_strategy(
    threads={"w1": ["read", "write"], "w2": ["read", "write"]},
))
def test_no_lost_update(schedule):
    ...

frontrun.trace_markers.all_marker_schedules(threads)[source]¶

Enumerate ALL valid interleavings of thread markers.

A valid interleaving places every thread’s markers in the schedule while preserving their relative order within each thread.

For N threads with marker counts k1, k2, …, kN, the total number of valid interleavings is the multinomial coefficient:

(k1 + k2 + ... + kN)! / (k1! * k2! * ... * kN!)

Parameters:: threads (dict[str, list[str]]) – Mapping from thread/execution names to their ordered list of marker names.
Returns:: A list of Schedule objects covering every valid interleaving.
Return type:: list[Schedule]

Example:

schedules = all_marker_schedules(
    threads={"t1": ["a", "b"], "t2": ["x", "y"]},
)
assert len(schedules) == 6  # C(4,2)

frontrun.trace_markers.explore_marker_interleavings(setup, threads, invariant, *, stop_on_first=True, deadlock_timeout=5.0, timeout=10.0)[source]¶

Explore all marker-level interleavings and check an invariant.

Generates every valid interleaving of the declared markers (preserving per-thread order), runs each one against real code via TraceExecutor, and checks the invariant after each execution.

Parameters:

setup (Callable[[...], Any]) – Factory producing fresh shared state for each execution.
threads (dict[str, tuple[Callable[[...], None], list[str]]]) – Mapping from execution name to (target_fn, markers) where target_fn takes the setup result and markers is the ordered list of # frontrun: marker names that target_fn hits.
invariant (Callable[[...], bool]) – Predicate on the shared state; returns True if correct.
stop_on_first (bool) – Stop after finding the first invariant violation (default True).
deadlock_timeout (float) – Per-schedule deadlock detection timeout.
timeout (float | None) – Per-schedule join timeout.

Returns:

An InterleavingResult. The counterexample field is a Schedule (not a list of ints) when a violation is found.

Return type:

InterleavingResult

Async Trace Markers¶

Frontrun: Deterministic async task interleaving using comment-based markers.

This module provides a mechanism to control async task execution order by marking synchronization points in code with # frontrun: marker_name comments, matching the elegant syntax of the sync trace_markers module.

Key Insight: Thread-Based Execution with sys.settrace¶

This implementation mirrors the sync architecture exactly:

Each async task runs in its own thread via asyncio.run(task_fn())
sys.settrace fires on every line, including inside synchronous sub-coroutines
When a marker is detected, the trace function blocks the thread via ThreadCoordinator.wait_for_turn()
Execution resumes exactly where it paused

This solves the synchronous function body bug: when await self.get_balance() completes synchronously (no internal yields), the trace still fires on every line during that execution. After get_balance() returns, the trace fires for the next line (after the marker), detects the marker, and blocks before that line executes.

Marker Semantics¶

Marker Placement: Place markers to gate the operations you want to control.

Inline markers (marker on same line as operation):: current = self.balance # frontrun: read_balance
Separate-line markers (marker before operation):: # frontrun: read_balance current = self.balance

Both styles work identically: the marker gates execution of the line, ensuring it only executes after the scheduler approves this task at this marker.

Example usage:

async def worker_function():
    # frontrun: read_data
    x = await read_data()
    # frontrun: write_data
    await write_data(x)

schedule = Schedule([
    Step("task1", "read_data"),
    Step("task2", "read_data"),
    Step("task1", "write_data"),
    Step("task2", "write_data"),
])

executor = AsyncTraceExecutor(schedule)
executor.run({
    'task1': worker_function,
    'task2': worker_function,
})

Or using the convenience function:

async_frontrun(
    schedule=schedule,
    tasks={'task1': worker1, 'task2': worker2},
)

class frontrun.async_trace_markers.AsyncTraceExecutor(schedule, *, deadlock_timeout=5.0)[source]¶

Bases: object

Executes async tasks with interlaced execution according to a schedule.

This is the main interface for the async frontrun library. It uses comment-based markers (# frontrun: marker_name) to control task execution order.

Unlike the sync version which runs tasks in actual threads, this runs each async task in its own thread with its own event loop via asyncio.run().

Initialize the executor with a schedule.

Parameters:

schedule (Schedule) – The Schedule defining the execution order
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).

__init__(schedule, *, deadlock_timeout=5.0)[source]¶

Initialize the executor with a schedule.

Parameters:

schedule (Schedule) – The Schedule defining the execution order
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).

run(tasks, timeout=10.0)[source]¶

Run all tasks with controlled interleaving based on comment markers.

This is now a synchronous method that creates threads and waits for them.

Parameters:

tasks (dict[str, Callable[[], Coroutine[Any, Any, None]]]) – Dictionary mapping task names to their async functions
timeout (float) – Timeout in seconds for all tasks to complete

Raises:

TimeoutError – If tasks don’t complete within the timeout
Any exception that occurred in a task during execution –

Return type:

None

reset()[source]¶: Reset the executor for another run (for testing purposes).

frontrun.async_trace_markers.async_frontrun(schedule, tasks, task_args=None, task_kwargs=None, timeout=10.0, deadlock_timeout=5.0)[source]¶

Convenience function to run multiple async tasks with a schedule.

This is now a synchronous function (not async) that creates an executor and runs the tasks.

Tasks use # frontrun: marker_name comments to mark synchronization points. No need to pass marker functions to tasks - the executor automatically detects markers via sys.settrace.

Parameters:

schedule (Schedule) – The Schedule defining execution order
tasks (dict[str, Callable[[...], Coroutine[Any, Any, None]]]) – Dictionary mapping execution unit names to their async target functions
task_args (dict[str, tuple[Any, ...]] | None) – Optional dictionary mapping execution unit names to argument tuples
task_kwargs (dict[str, dict[str, Any]] | None) – Optional dictionary mapping execution unit names to keyword argument dicts
timeout (float) – Timeout in seconds for the entire execution
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).

Returns:

The AsyncTraceExecutor instance (useful for inspection)

Return type:

AsyncTraceExecutor

Example:

async def worker(account, amount):
    # frontrun: before_deposit
    await account.deposit(amount)

async_frontrun(
    schedule=Schedule([
        Step("t1", "before_deposit"),
        Step("t2", "before_deposit")
    ]),
    tasks={"t1": worker, "t2": worker},
    task_args={"t1": (account, 50), "t2": (account, 50)},
)

Async Bytecode Instrumentation¶

Await-point-level deterministic async concurrency testing.

Uses the shared InterleavedLoop abstraction to control which async task resumes at each await point, enabling fine-grained control over task interleaving.

This pairs naturally with property-based testing: rather than specifying exact schedules, generate random interleavings and check that invariants hold (or that bugs can be found).

The core insight: in async Python, context switches happen ONLY at await points. The event loop is single-threaded. By controlling which task resumes at each await point, we can explore the full space of possible interleavings — and there are far fewer of them than in threaded code.

Example — find a race condition with random schedule exploration:

>>> import asyncio
>>> import frontrun
>>>
>>> class Counter:
...     def __init__(self):
...         self.value = 0
...     async def increment(self):
...         temp = self.value
...         await asyncio.sleep(0)  # any natural await is a scheduling point
...         self.value = temp + 1
>>>
>>> result = asyncio.run(frontrun.explore_async_random(
...     setup=lambda: Counter(),
...     tasks=[lambda c: c.increment(), lambda c: c.increment()],
...     invariant=lambda c: c.value == 2,
... ))
>>> assert result.property_holds, result.explanation  # fails — lost update!

Any natural await in user code is a scheduling point. await_point() is still available as an explicit extra yield when a test wants to force an additional checkpoint.

class frontrun.async_shuffler.AwaitScheduler(schedule, num_tasks, *, deadlock_timeout=5.0, detect_sql=False, virtual_clock=None, clock_mode='real', max_ops=0, extension_seed=None)[source]¶

Bases: InterleavedLoop

Controls async task execution at await-point granularity.

The schedule is a list of task indices. Each entry means “let this task resume from its next await point.” When the sampled prefix is exhausted, it is extended deterministically so tasks remain controlled.

Built on the shared InterleavedLoop abstraction, using index-based scheduling as its policy.

Parameters:

schedule (list[int])
num_tasks (int)
deadlock_timeout (float)
detect_sql (bool)
virtual_clock (Any)
clock_mode (str)
max_ops (int)
extension_seed (int | None)

add_timeout_deadline(task_id, deadline, token)[source]¶

Parameters:

task_id (int)
deadline (float)
token (object)

Return type:

None

remove_timeout_deadline(task_id, token)[source]¶

Cancel a virtual timeout deadline (no-op default).

Parameters:

task_id (int)
token (object)

Return type:

None

park_timed_wait(task_id)[source]¶

Called by the virtual asyncio.wait_for wrapper for waits on bare futures/tasks: the parked task has no further scheduling points, so the schedule must skip its entries (see should_proceed) and the clock advance is what wakes it.

Parameters:: task_id (int)
Return type:: None

unpark_timed_wait(task_id)[source]¶

Unregister a task from a virtual timed park (no-op default).

Parameters:: task_id (int)
Return type:: None

async kick_stalled_schedule(task_id)[source]¶

Wake schedule progress after task_id parked itself in a timed wait.

Tasks waiting in pause() for the parked task’s schedule entries must re-check should_proceed (which now skips those entries), and if every live task is blocked only the clock can move.

Parameters:: task_id (int)
Return type:: None

async sleep_until(task_id, deadline=None, *, duration=None)[source]¶

Block task_id until the virtual clock reaches deadline.

With duration= the deadline is computed under _condition after the fairness yield: the yield spans a full loop pass, during which another task’s step can advance the clock — a caller-side now() read would then register an already-stale deadline.

Parameters:

task_id (int)
deadline (float | None)
duration (float | None)

Return type:

None

should_proceed(task_id, marker=None)[source]¶

Return True if this task should resume now.

Called while holding the condition lock. Must not await.

Parameters:

task_id (Any) – Identity of the calling task (str, int, etc.)
marker (Any) – Optional context from the yield point (e.g. a marker name, an (operation, phase) tuple, or None).

Return type:

bool

on_proceed(task_id, marker=None)[source]¶

Update scheduling state after a task is allowed to proceed.

Called while holding the condition lock, immediately after should_proceed returned True. Must not await.

Parameters:

task_id (Any) – Identity of the task that is proceeding.
marker (Any) – Same marker value passed to should_proceed.

Return type:

None

report_and_wait(_frame, _thread_id)[source]¶

No-op scheduling hook called by async SQL interception.

Scheduling already happens at the patched cursor’s own await points, so this just lets the SQL call proceed.

Parameters:

_frame (Any)
_thread_id (int)

Return type:

bool

acquire_row_locks(_thread_id, _resource_ids)[source]¶

No-op: random exploration does not arbitrate SQL row locks.

Parameters:

_thread_id (int)
_resource_ids (list[str])

Return type:

None

release_row_locks(_thread_id, _resources=None)[source]¶

No-op: random exploration does not arbitrate SQL row locks.

Parameters:

_thread_id (int)
_resources (object)

Return type:

None

async pause(task_id, marker=None)[source]¶

Yield point: block until the scheduling policy says to proceed.

Tasks call this at every point where a context switch could happen. The call blocks (yields to the event loop) until should_proceed() returns True for this task, then calls on_proceed() and returns.

Uses all-tasks-waiting detection: if every non-done task is blocked in pause() and none can proceed, deadlock is detected instantly.

Parameters:

task_id (Any) – Identity of the calling task.
marker (Any) – Optional scheduling context.

Return type:

None

property had_error: bool¶: Check if an error occurred during execution.

class frontrun.async_shuffler.AsyncShuffler(scheduler)[source]¶

Bases: object

Run concurrent async functions with await-point-level interleaving control.

Creates asyncio tasks for each function and delegates to the AwaitScheduler (an InterleavedLoop subclass) for execution and context setup.

Parameters:: scheduler (AwaitScheduler)

async run(funcs, args=None, kwargs=None, timeout=10.0, *, detect_external_deadlock=True)[source]¶

Run async functions concurrently with controlled interleaving.

Parameters:

funcs (list[Callable[[...], Coroutine[Any, Any, None]]]) – One async callable per task.
args (list[tuple[Any, ...]] | None) – Per-task positional args.
kwargs (list[dict[str, Any]] | None) – Per-task keyword args.
timeout (float) – Max wait time for all tasks.
detect_external_deadlock (bool)

frontrun.async_shuffler.controlled_interleaving(schedule, num_tasks=2)[source]¶

Context manager for running async code under a specific interleaving.

Parameters:

schedule (list[int]) – List of task indices controlling await-point execution order.
num_tasks (int) – Number of tasks.

Yields:

AsyncShuffler runner.

Return type:

AsyncGenerator[AsyncShuffler, None]

Example

>>> async with controlled_interleaving([0, 1, 0, 1], num_tasks=2) as runner:
...     await runner.run([coro1, coro2])

async frontrun.async_shuffler.run_with_schedule(schedule, setup, tasks, timeout=5.0, deadlock_timeout=5.0, detect_sql=False)[source]¶

Run one async interleaving and return the state object.

Parameters:

schedule (list[int]) – Await-point-level schedule (list of task indices).
setup (Callable[[], Any]) – Returns fresh shared state.
tasks (list[Callable[[Any], Coroutine[Any, Any, None]]]) – Async callables that each receive the state as their argument.
timeout (float) – Max seconds.
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).
detect_sql (bool) – If True, patch async DBAPI drivers (aiosqlite, psycopg AsyncCursor, aiomysql, asyncpg) to intercept SQL and report table-level conflicts.

Returns:

The state object after execution.

Return type:

Any

async frontrun.async_shuffler.explore_async_random(setup, tasks, invariant, max_attempts=200, max_ops=100, timeout_per_run=5.0, seed=None, deadlock_timeout=5.0, detect_sql=False, trace_packages=None, patch_sleep=True, serializable_invariant=False, error_on_any_race=False, total_timeout=None, clock='real', clock_diagnostics=False)[source]¶

Search for async interleavings that violate an invariant.

Generates random await-point-level schedules and tests whether the invariant holds under each one. If a violation is found, returns immediately with the counterexample schedule.

This is the async analogue of property-based testing for concurrency: instead of generating random inputs, we generate random interleavings and check that the result satisfies an invariant.

Parameters:

setup (Callable[[], Any]) – Returns fresh shared state for each attempt.
tasks (list[Callable[[Any], Coroutine[Any, Any, None]]]) – Async callables that each receive the state as their argument.
invariant (Callable[[Any], bool]) – Predicate on the state. Returns True if the property holds.
max_attempts (int) – How many random interleavings to try.
max_ops (int) – Maximum schedule length per attempt.
timeout_per_run (float) – Timeout for each individual run.
seed (int | None) – Optional RNG seed for reproducibility.
deadlock_timeout (float) – Seconds to wait before declaring a deadlock (default 5.0). Increase for code that legitimately blocks in C extensions (NumPy, database queries, network I/O).
detect_sql (bool) – If True, patch async DBAPI drivers (aiosqlite, psycopg AsyncCursor, aiomysql, asyncpg) to intercept SQL and report table-level conflicts.
trace_packages (list[str] | None) – Accepted for API compatibility but not used. The async shuffler operates at await-point granularity and does not perform file-level tracing.
patch_sleep (bool) – If True (default), asyncio.sleep yields to the scheduler instead of waiting. Required for clock != "real".
serializable_invariant (Callable[[Any], Any] | bool) – Check serializability against sequential runs. Cannot be combined with a virtual clock.
error_on_any_race (bool) – Not supported here — requires the DPOR strategy.
clock (Literal['real', 'virtual', 'explored']) – "real" (default), "virtual" (autojump virtual clock: time reads are virtual, asyncio.sleep costs zero wall time), or "explored" (schedule entries landing on a sleeping task advance the clock, exploring early timer firings). Tasks that block on primitives the scheduler cannot see (e.g. a raw asyncio.Lock) are handled by a quiescence heuristic; prefer the DPOR strategy for lock-heavy async code. asyncio.wait_for, asyncio.timeout, and asyncio.timeout_at inside explored tasks use virtual deadlines. See Virtual clock: timeout, retry, and TTL races.
clock_diagnostics (bool) – Accepted for API consistency. Async random does not trace worker frames, so captured time.* references cannot be diagnosed and a RuntimeWarning is emitted.
total_timeout (float | None)

Returns:

InterleavingResult with the outcome. The unique_interleavings field reports how many distinct schedule orderings were observed.

Return type:

InterleavingResult

frontrun.async_shuffler.schedule_strategy(num_tasks, max_ops=100)[source]¶

Hypothesis strategy for generating fair await-point schedules.

Generates schedules as a sequence of rounds, where each round is a random permutation of all task indices. This guarantees every task gets exactly the same number of scheduling slots, preventing starvation.

For use with hypothesis @given decorator in your own tests:

>>> from hypothesis import given
>>> from frontrun.async_shuffler import schedule_strategy, run_with_schedule
>>> import asyncio
>>>
>>> @given(schedule=schedule_strategy(2))
... def test_my_invariant(schedule):
...     state = asyncio.run(run_with_schedule(schedule, setup, tasks))
...     assert state.value == expected

Note: max_ops defaults to 100 (vs 300 for bytecode.py) because async code has far fewer interleaving points. Each schedule entry corresponds to one await_point() call, not one bytecode opcode.

Parameters:

num_tasks (int)
max_ops (int)

Return type:

Any

Trace Formatting¶

Trace recording, filtering, and formatting for comprehensible race condition errors.

When frontrun finds a race condition, the raw counterexample is a list of thread indices — one per bytecode instruction. This module transforms that into a human-readable “story” of which source lines executed in which order.

The pipeline: 1. Record a TraceEvent at each opcode during the failing run. 2. Filter to events that touch shared state (LOAD_ATTR/STORE_ATTR, etc.) 3. Deduplicate consecutive events from the same thread on the same source line. 4. Classify the conflict pattern (lost update, order violation, etc.) 5. Format as an interleaved source-line trace.

class frontrun._trace_format.TraceEvent(step_index, thread_id, filename, lineno, function_name, opcode, access_type=None, attr_name=None, obj_type_name=None, call_chain=None, detail=None)[source]¶

Bases: object

A single recorded event from the trace.

Parameters:

step_index (int)
thread_id (int)
filename (str)
lineno (int)
function_name (str)
opcode (str)
access_type (str | None)
attr_name (str | None)
obj_type_name (str | None)
call_chain (list[str] | None)
detail (str | None)

step_index: int¶

thread_id: int¶

filename: str¶

lineno: int¶

function_name: str¶

opcode: str¶

access_type: str | None¶

attr_name: str | None¶

obj_type_name: str | None¶

call_chain: list[str] | None¶

detail: str | None¶

class frontrun._trace_format.SourceLineEvent(thread_id, filename, lineno, function_name, source_line, access_type=None, attr_name=None, obj_type_name=None, call_chain=None, detail=None)[source]¶

Bases: object

A deduplicated, source-level event for display.

Parameters:

thread_id (int)
filename (str)
lineno (int)
function_name (str)
source_line (str)
access_type (str | None)
attr_name (str | None)
obj_type_name (str | None)
call_chain (list[str] | None)
detail (str | None)

thread_id: int¶

filename: str¶

lineno: int¶

function_name: str¶

source_line: str¶

access_type: str | None¶

attr_name: str | None¶

obj_type_name: str | None¶

call_chain: list[str] | None¶

detail: str | None¶

frontrun._trace_format.qualified_name(frame)[source]¶

Get a qualified function name from a frame (e.g. DB.dict).

Parameters:: frame (Any)
Return type:: str

frontrun._trace_format.build_call_chain(frame, *, filter_fn, max_depth=3)[source]¶

Walk user-code frames from frame upward, returning qualified names.

filter_fn(filename) -> bool selects which frames are user code (typically frontrun._tracing.should_trace_file()). Returns None when the chain would be empty.

Parameters:

frame (Any)
filter_fn (Any)
max_depth (int)

Return type:

list[str] | None

class frontrun._trace_format.TraceRecorder(*, enabled=True)[source]¶

Bases: object

Accumulates TraceEvent objects during a single execution.

Thread-safe: multiple threads call record() concurrently, each holding the scheduler lock (so ordering is deterministic).

Parameters:: enabled (bool)

events: list[TraceEvent]¶

enabled¶

record(thread_id, frame, opcode=None, access_type=None, attr_name=None, obj=None, obj_type_name=None, call_chain=None)[source]¶

Record one trace event from a frame object.

Parameters:

thread_id (int)
frame (Any)
opcode (str | None)
access_type (str | None)
attr_name (str | None)
obj (Any)
obj_type_name (str | None)
call_chain (list[str] | None)

Return type:

None

record_io(thread_id, resource_id, kind, *, call_chain=None, detail=None)[source]¶

Record an I/O event that has no Python frame (e.g. C-level socket I/O).

Parameters:

thread_id (int)
resource_id (str)
kind (str)
call_chain (list[str] | None)
detail (str | None)

Return type:

None

record_from_opcode(thread_id, frame)[source]¶

Record an event using the frame’s current instruction.

Used by the bytecode explorer, which doesn’t do shadow-stack analysis. We inspect the instruction to extract access info.

Parameters:

thread_id (int)
frame (Any)

Return type:

None

frontrun._trace_format.filter_to_shared_accesses(events)[source]¶

Keep only events that access shared mutable state.

Parameters:: events (list[TraceEvent])
Return type:: list[TraceEvent]

frontrun._trace_format.deduplicate_to_source_lines(events)[source]¶

Collapse consecutive events from the same thread+line into one SourceLineEvent.

When multiple opcodes on the same source line produce events (e.g., LOAD_ATTR then STORE_ATTR for self.value += 1), merge them into a single entry with a combined access_type — but only when they access the same (obj_type, attr_name) key. Events with different keys on the same line get separate entries so that filtering can distinguish them later (e.g. an attribute read vs an I/O event).

Parameters:: events (list[TraceEvent])
Return type:: list[SourceLineEvent]

class frontrun._trace_format.ConflictInfo(pattern, summary, attr_name=None)[source]¶

Bases: object

Description of the conflict pattern found in the trace.

Parameters:

pattern (str)
summary (str)
attr_name (str | None)

pattern: str¶

summary: str¶

attr_name: str | None = None¶

frontrun._trace_format.classify_conflict(events)[source]¶

Examine a filtered, deduplicated trace and classify the conflict type.

Looks for classic patterns: - Lost update: R_a R_b W_a W_b (or R_a R_b W_b W_a) - Write-write: W_a W_b on same attribute without intervening sync

Parameters:: events (list[SourceLineEvent])
Return type:: ConflictInfo

class frontrun._trace_format.CollapsedRun(count, thread_id)[source]¶

Bases: object

Placeholder for a collapsed sequence of events from one thread.

Parameters:

count (int)
thread_id (int)

count: int¶

thread_id: int¶

frontrun._trace_format.condense_trace(lines, *, max_lines=30)[source]¶

Condense a trace to show only the essential interleaving.

Strategy: 1. Always filter to events involved in cross-thread data conflicts

(same attribute accessed by 2+ threads with at least one write). After filtering, re-merge consecutive same-line events that were previously separated by now-removed entries.

If still too long, collapse single-thread runs (keep first/last).
Cap at max_lines.

Returns a mixed list of SourceLineEvent and CollapsedRun placeholders for the formatter to render.

Parameters:

lines (list[SourceLineEvent])
max_lines (int)

Return type:

list[SourceLineEvent | CollapsedRun]

frontrun._trace_format.format_trace(events, *, num_threads, thread_names=None, num_explored=0, invariant_desc=None, show_opcodes=False, reproduction_attempts=0, reproduction_successes=0, max_lines=30)[source]¶

Format a trace as a human-readable interleaved source-line display.

Parameters:

events (list[TraceEvent]) – Raw trace events from a TraceRecorder.
num_threads (int) – Total number of threads.
thread_names (list[str] | None) – Optional display names for threads.
num_explored (int) – Number of interleavings explored before finding the bug.
invariant_desc (str | None) – Description of the violated invariant.
show_opcodes (bool) – If True, include opcode-level detail for each line.
reproduction_attempts (int) – How many times the schedule was replayed.
reproduction_successes (int) – How many replays reproduced the failure.
max_lines (int) – Maximum trace lines before condensation (default 30).

Returns:

Multi-line string suitable for printing or attaching to test output.

Return type:

str

Async Scheduler Utilities¶

Async event loop abstraction for deterministic task interleaving.

This module provides InterleavedLoop, the shared foundation for all async frontrun POCs. It wraps asyncio’s cooperative scheduling to give deterministic control over which task resumes at each yield point.

In async Python, the event loop decides which ready task to resume after each await point. InterleavedLoop intercepts this decision, using a pluggable scheduling policy to control the execution order.

Key insight: async code is single-threaded and cooperative. Context switches happen ONLY at await points. InterleavedLoop exploits this by gating each yield point through an asyncio.Condition — tasks wait until the scheduling policy says it’s their turn.

Both async approaches build on this abstraction: - async_trace_markers (comment annotations): marker-based scheduling - async_shuffler (property-based): index-based scheduling

Each POC subclasses InterleavedLoop and implements two methods: - should_proceed(task_id, marker): return True when a task should resume - on_proceed(task_id, marker): update internal scheduling state

Example — a simple round-robin scheduler:

>>> class RoundRobinLoop(InterleavedLoop):
...     def __init__(self, order):
...         super().__init__()
...         self._order = order
...         self._step = 0
...
...     def should_proceed(self, task_id, marker=None):
...         if self._step >= len(self._order):
...             return True
...         return self._order[self._step] == task_id
...
...     def on_proceed(self, task_id, marker=None):
...         self._step += 1

class frontrun.async_scheduler.InterleavedLoop(*, deadlock_timeout=5.0)[source]¶

Bases: object

Wrapped event loop for deterministic async task interleaving.

This class controls which async task resumes at each yield point. Tasks call await loop.pause(task_id) at points where a context switch could happen, and the loop’s scheduling policy decides whether the task should proceed or wait.

Subclasses must implement:: should_proceed(task_id, marker): Is it this task’s turn? on_proceed(task_id, marker): Update state after a task proceeds.
The base class provides:: pause(): Yield point that gates on the scheduling policy run_all(): Run tasks with controlled interleaving Error propagation, timeout handling, and done-task tracking

Parameters:: deadlock_timeout (float)

virtual_clock: Any = None¶: Active virtual clock, or None in real-time mode.

async kick_stalled_schedule(task_id)[source]¶

Hand the turn onward after a task engine-blocked itself (no-op default).

Parameters:: task_id (int)
Return type:: None

async wait_until_scheduled_after_block(task_id, reason)[source]¶

Wait for a physically-woken task to be scheduled again (no-op default).

Parameters:

task_id (int)
reason (str)

Return type:

None

report_task_sync(task_id, event_type, sync_id)[source]¶

Report a happens-before sync edge to the engine (no-op default).

Parameters:

task_id (int)
event_type (str)
sync_id (int)

Return type:

None

report_task_access(task_id, object_id, kind)[source]¶

Report a memory / resource access to the engine (no-op default).

Parameters:

task_id (int)
object_id (int)
kind (str)

Return type:

None

on_task_suspended(task_id)[source]¶

Mark an ordinary natural await as physically suspended.

Parameters:: task_id (int)
Return type:: None

on_task_resumed(task_id)[source]¶

Mark a naturally suspended task runnable before its next pause.

Parameters:: task_id (int)
Return type:: None

add_timeout_deadline(task_id, deadline, token)[source]¶

Parameters:

task_id (int)
deadline (float)
token (object)

Return type:

None

remove_timeout_deadline(task_id, token)[source]¶

Cancel a virtual timeout deadline (no-op default).

Parameters:

task_id (int)
token (object)

Return type:

None

park_timed_wait(task_id)[source]¶

Register a task parked in a virtual timed wait with no engine bookkeeping (e.g. asyncio.wait_for on a bare future under the random strategy); no-op default for engine-backed schedulers.

Parameters:: task_id (int)
Return type:: None

unpark_timed_wait(task_id)[source]¶

Unregister a task from a virtual timed park (no-op default).

Parameters:: task_id (int)
Return type:: None

should_proceed(task_id, marker=None)[source]¶

Return True if this task should resume now.

Called while holding the condition lock. Must not await.

Parameters:

task_id (Any) – Identity of the calling task (str, int, etc.)
marker (Any) – Optional context from the yield point (e.g. a marker name, an (operation, phase) tuple, or None).

Return type:

bool

on_proceed(task_id, marker=None)[source]¶

Update scheduling state after a task is allowed to proceed.

Called while holding the condition lock, immediately after should_proceed returned True. Must not await.

Parameters:

task_id (Any) – Identity of the task that is proceeding.
marker (Any) – Same marker value passed to should_proceed.

Return type:

None

async pause(task_id, marker=None)[source]¶

Yield point: block until the scheduling policy says to proceed.

Tasks call this at every point where a context switch could happen. The call blocks (yields to the event loop) until should_proceed() returns True for this task, then calls on_proceed() and returns.

Uses all-tasks-waiting detection: if every non-done task is blocked in pause() and none can proceed, deadlock is detected instantly.

Parameters:

task_id (Any) – Identity of the calling task.
marker (Any) – Optional scheduling context.

Return type:

None

async run_all(task_funcs, timeout=10.0, *, detect_external_deadlock=False)[source]¶

Run tasks with controlled interleaving.

Parameters:

task_funcs (dict[Any, Callable[[...], Awaitable[None]]] | list[Callable[[...], Awaitable[None]]]) – Either a dict {task_id: async_callable} or a list of async callables (which get integer task_ids 0, 1, 2, …).
timeout (float) – Maximum total time to wait for all tasks.
detect_external_deadlock (bool) – Also detect deadlocks where every unfinished task is blocked on an unmanaged awaitable (e.g. a stock asyncio.Lock) rather than inside pause(). Such deadlocks are invisible to the pause-path detection; with this flag, a full deadlock_timeout window with zero scheduler progress records a deadlock in self._error before timing out, so callers can tell it from a slow-but-correct run.

Return type:

None

property had_error: bool¶: True if an error was reported during execution.

exception frontrun.async_scheduler.SchedulerTimeoutError[source]¶

Bases: TimeoutError

Timeout raised by frontrun’s scheduler machinery, not user code.

async frontrun.async_scheduler.frontrun_wait_for(awaitable, timeout)[source]¶

asyncio.wait_for whose timeout timer is tagged as frontrun-internal.

Parameters:

awaitable (Coroutine[Any, Any, _T] | Awaitable[_T])
timeout (float)

Return type:

ORM Helpers (contrib)¶

Django helpers for DPOR integration testing (sync and async).

Use django_dpor for both sync and async code:

Sync (threads):

result = django_dpor(
    setup=_State,
    threads=[thread_a, thread_b],
    invariant=_invariant,
)

Async (tasks):

result = await django_dpor(
    setup=_State,
    tasks=[task_a, task_b],
    invariant=_invariant,
)

async frontrun.contrib.django.async_django_dpor(setup, tasks, invariant, *, db_alias='default', lock_timeout=None, detect_sql=True, trace_packages=None, **kwargs)[source]¶

Run async DPOR exploration with per-task Django async connection management.

Each task closes the shared Django connection and opens a fresh one.

By default, trace_packages is set to DJANGO_TRACE_PACKAGES so that code inside django_* apps and django.contrib.sites submodules is traced. Pass an explicit list (or []) to override.

Parameters:

setup (Callable[[], T]) – Called once per execution to create fresh shared state.
tasks (list[Callable[[T], Coroutine[Any, Any, None]]]) – List of async callables, each receiving the shared state.
invariant (Callable[[T], bool]) – Predicate over shared state after all tasks complete.
db_alias (str) – Django database alias to use (default "default").
lock_timeout (int | None) – If set, execute SET lock_timeout = <N>ms on each task’s connection.
detect_sql (bool) – Passed through to the async DPOR engine (default True).
trace_packages (list[str] | None) – Package name patterns (fnmatch syntax) to trace. Defaults to DJANGO_TRACE_PACKAGES. Pass [] to disable extra tracing beyond user code.
**kwargs (Any) – Forwarded verbatim to frontrun.explore().

Return type:

InterleavingResult

frontrun.contrib.django.django_dpor(*args, **kwargs)[source]¶

Run DPOR with per-thread/task Django connection management.

Use threads=[...] for sync code or tasks=[...] for async code. The async form returns an awaitable.

Parameters:

args (Any)
kwargs (Any)

Return type:

Any

SQLAlchemy helpers for DPOR integration testing (sync and async).

Use sqlalchemy_dpor for both sync and async code:

Sync (threads):

result = sqlalchemy_dpor(
    engine=engine,
    setup=_State,
    threads=[thread_a, thread_b],
    invariant=_invariant,
)

Async (tasks):

result = await sqlalchemy_dpor(
    engine=async_engine,
    setup=_State,
    tasks=[task_a, task_b],
    invariant=_invariant,
)

Inside sync threads, use get_connection(). Inside async tasks, use get_async_connection().

async frontrun.contrib.sqlalchemy.async_sqlalchemy_dpor(engine, setup, tasks, invariant, *, lock_timeout=None, detect_sql=True, **kwargs)[source]¶

Run async DPOR exploration with per-task async SQLAlchemy connection management.

Parameters:

engine (Any) – An async SQLAlchemy AsyncEngine instance.
setup (Callable[[], T]) – Called once per execution to create fresh shared state.
tasks (list[Callable[[T], Coroutine[Any, Any, None]]]) – List of async callables, each receiving the shared state. Each task gets its own async connection. Use get_async_connection() inside the task to access it.
invariant (Callable[[T], bool]) – Predicate over shared state after all tasks complete.
lock_timeout (int | None) – If set, execute SET lock_timeout = <N>ms on each task’s connection.
detect_sql (bool) – Passed through to the async DPOR engine (default True).
**kwargs (Any) – Forwarded verbatim to frontrun.explore().

Return type:

InterleavingResult

frontrun.contrib.sqlalchemy.get_async_connection()[source]¶

Return the per-task async SQLAlchemy connection set by async_sqlalchemy_dpor.

Return type:: Any

frontrun.contrib.sqlalchemy.get_connection()[source]¶

Return the per-thread SQLAlchemy connection set by sqlalchemy_dpor.

Return type:: Any

frontrun.contrib.sqlalchemy.sqlalchemy_dpor(*args, **kwargs)[source]¶

Run DPOR with per-thread/task SQLAlchemy connection management.

Use threads=[...] for sync code or tasks=[...] for async code. The async form returns an awaitable.

Parameters:

args (Any)
kwargs (Any)

Return type:

Any