Heisenbug Autopsy Using Deterministic Log Reordering In Async Python

I ran into a classic Heisenbug: the bug disappeared when I added logging. At first I assumed it was “just” timing, but the pattern was too consistent—my async code behaved differently depending on how often I printed status lines.

What finally nailed it for me was a deterministic trick: I forced all log writes through a single in-memory queue and replayed them after the fact in a fixed order, so I could analyze causality without perturbing scheduling.

This post documents exactly what I built and what I learned while debugging an async failure caused by race conditions.

The tiny async failure that “moved” when I logged

I had an async task group where each task:

1. waits on an event,
2. updates a shared dictionary,
3. sometimes triggers a failure when it expects a key to exist.

Here’s the simplified repro that matches the original behavior: without logging, the key is missing rarely; with logging, it fails more often.

import asyncio
import random


async def worker(name: str, ready: asyncio.Event, shared: dict, results: list[str]):
    await ready.wait()
    # Artificial jitter changes scheduling
    await asyncio.sleep(random.random() * 0.001)

    # Sometimes we "expect" a key created by another worker,
    # but the other worker might not have run yet.
    if name == "b":
        # This line is the crash site (KeyError) in the original case.
        results.append(f"b sees keys: {sorted(shared.keys())}")
        _ = shared["a"]  # may not exist

    # Everyone writes their own key after the read for b
    shared[name] = f"value-{name}"


async def run_once():
    ready = asyncio.Event()
    shared = {}
    results: list[str] = []

    workers = [
        asyncio.create_task(worker("a", ready, shared, results)),
        asyncio.create_task(worker("b", ready, shared, results)),
    ]

    # Let tasks start
    await asyncio.sleep(0)
    ready.set()

    await asyncio.gather(*workers)
    return results


async def main():
    # Repeat to reproduce
    for i in range(2000):
        try:
            await run_once()
        except KeyError as e:
            print("Failed on iteration", i, "with", repr(e))
            return
    print("No failure")


if __name__ == "__main__":
    asyncio.run(main())

On my machine, this fails quickly. Adding print() calls inside the tasks changes the scheduling enough that the race becomes more visible (and sometimes vanishes), which is the heart of a Heisenbug.

Why “print debugging” changes async behavior

In async Python, print() doesn’t just add information—it can change timing:

• console output is slow compared to normal execution,
• stdout flushing/locking can block the event loop briefly,
• added delays reorder task execution.

So instead of trying to observe the system “live”, I switched to recording in a way that minimally affects scheduling, then replaying deterministically.

The deterministic log reorderer: record first, analyze later

The idea: every time something important happens in a task, I record an event object with:

• a monotonic timestamp (time.monotonic_ns()),
• a task id (which task emitted the event),
• an event type and payload,
• and a sequence number to break timestamp ties.

Then I push these records into a queue with put_nowait(). That avoids blocking the event loop with I/O.

After the run, I sort the events into a deterministic order and print an analysis trace.

Minimal logger

import asyncio
import time
import random
from dataclasses import dataclass, field
from typing import Any


@dataclass(order=True)
class LogEvent:
    sort_index: tuple[int, int] = field(init=False, repr=False)
    t_ns: int
    seq: int
    task: str
    kind: str
    data: dict[str, Any]

    def __post_init__(self):
        # sort_index breaks ties deterministically using seq
        self.sort_index = (self.t_ns, self.seq)


class DeterministicLogger:
    def __init__(self):
        self._q: asyncio.Queue[LogEvent] = asyncio.Queue()
        self._seq = 0

    def record(self, task: str, kind: str, data: dict[str, Any]):
        # time.monotonic_ns is monotonic and suitable for ordering attempts
        t = time.monotonic_ns()
        self._seq += 1
        ev = LogEvent(t_ns=t, seq=self._seq, task=task, kind=kind, data=data)
        # put_nowait avoids blocking the event loop
        self._q.put_nowait(ev)

    async def drain(self) -> list[LogEvent]:
        events: list[LogEvent] = []
        while True:
            try:
                events.append(self._q.get_nowait())
            except asyncio.QueueEmpty:
                break
        return sorted(events, key=lambda e: e.sort_index)

Important details:

• put_nowait() keeps logging from becoming backpressure.
• timestamps + sequence number give deterministic ordering for the same run.
• later sorting means I can format output without affecting execution order.

Re-running the repro with deterministic trace

Now I’ll instrument the same failing scenario, but instead of printing inside the tasks, I record log events.

import asyncio
import random


async def worker(name: str, ready: asyncio.Event, shared: dict, logger: DeterministicLogger):
    await ready.wait()
    await asyncio.sleep(random.random() * 0.001)

    logger.record(name, "before_read", {"shared_keys": list(shared.keys())})

    if name == "b":
        # Crash site: we want to see whether "a" exists here.
        _ = shared["a"]

    logger.record(name, "after_write", {"will_write": name})
    shared[name] = f"value-{name}"
    logger.record(name, "write_done", {"written": name})


async def run_once_with_trace():
    ready = asyncio.Event()
    shared: dict[str, str] = {}
    logger = DeterministicLogger()

    tasks = [
        asyncio.create_task(worker("a", ready, shared, logger)),
        asyncio.create_task(worker("b", ready, shared, logger)),
    ]

    await asyncio.sleep(0)
    ready.set()

    try:
        await asyncio.gather(*tasks)
        return None, await logger.drain(), shared
    except KeyError as e:
        return e, await logger.drain(), shared


async def main():
    for i in range(2000):
        err, events, shared = await run_once_with_trace()
        if err is not None:
            print("Failed on iteration", i, "with", repr(err))
            print("\nDeterministic trace:")
            for ev in events:
                print(f"{ev.t_ns} seq={ev.seq} [{ev.task}] {ev.kind} {ev.data}")
            print("\nFinal shared:", shared)
            return
    print("No failure")


if __name__ == "__main__":
    asyncio.run(main())

What the trace tells you

On a failing iteration, the trace typically looks like this (shape matters more than exact numbers):

• b before_read sees keys: []
• a after_write hasn’t happened yet
• then b attempts shared["a"] and crashes
• later a write_done occurs (after the crash site)

This proves the bug is not “random” in the sense of mystery—it’s deterministic in the sense of ordering: b reads before a writes after the event is set.

Fixing the root cause: enforce ordering explicitly

Once I had the trace, the fix was straightforward: if b needs a to be written first, b must wait for that condition.

In async code, that usually means an asyncio.Event per dependency.

Ordered dependency fix

import asyncio
import random


async def worker_a(ready: asyncio.Event, a_written: asyncio.Event, shared: dict, logger: DeterministicLogger):
    await ready.wait()
    await asyncio.sleep(random.random() * 0.001)

    logger.record("a", "write_start", {})
    shared["a"] = "value-a"
    a_written.set()
    logger.record("a", "write_done", {})


async def worker_b(ready: asyncio.Event, a_written: asyncio.Event, shared: dict, logger: DeterministicLogger):
    await ready.wait()
    await asyncio.sleep(random.random() * 0.001)

    logger.record("b", "waiting_for_a", {})
    await a_written.wait()

    logger.record("b", "before_read", {"shared_keys": list(shared.keys())})
    _ = shared["a"]  # now guaranteed


async def run_once_fixed():
    ready = asyncio.Event()
    a_written = asyncio.Event()
    shared = {}
    logger = DeterministicLogger()

    tasks = [
        asyncio.create_task(worker_a(ready, a_written, shared, logger)),
        asyncio.create_task(worker_b(ready, a_written, shared, logger)),
    ]

    await asyncio.sleep(0)
    ready.set()

    await asyncio.gather(*tasks)
    return await logger.drain(), shared


async def main():
    for i in range(2000):
        events, shared = await run_once_fixed()
        if i == 0:
            print("Example trace (first successful run):")
            for ev in events:
                print(f"{ev.t_ns} seq={ev.seq} [{ev.task}] {ev.kind} {ev.data}")
            print("Final shared:", shared)
    print("Completed without KeyError")


if __name__ == "__main__":
    asyncio.run(main())

This removes the race by making the dependency explicit: b waits until a has written.

What I learned (and what I now do by default)

The key lesson from this “deterministic log reorderer” experiment is that observability can perturb concurrency. Traditional live logging (especially print) can change task scheduling enough to hide the bug you’re hunting.

By recording log events into an async queue and replaying them after the run in a deterministic order, I turned a Heisenbug into a traceable ordering problem:

• The failure happened because task b read shared["a"] before task a wrote it.
• Fixing the bug meant enforcing the dependency with an asyncio.Event, not “hoping” timing would line up.

In short: when debugging async code, I now default to non-blocking event recording + deterministic replay for root-cause analysis, and I treat ordering bugs as correctness bugs—not just timing glitches.