How to Handle Concurrent Event Dispatch Without Race Conditions in bubus

bubus prevents race conditions during concurrent event dispatch by using a single global re-entrant lock combined with a custom asyncio queue that serializes event processing while allowing parallel handler execution.

When building event-driven architectures with browser-use/bubus, handling concurrent event dispatch without race conditions is critical for system stability. The library implements a sophisticated concurrency model centered in bubus/service.py that ensures thread-safe event processing even when multiple coroutines call dispatch() simultaneously.

Core Concurrency Architecture

The Global Re-entrant Lock

At the heart of bubus concurrency safety is a global re-entrant lock implemented via the ReentrantLock class (lines 47-84 in bubus/service.py). This lock is acquired at the start of every event processing cycle within the _run_loop method (lines 45-52), specifically during the step() → process_event() chain.

The lock is re-entrant across different asyncio tasks, meaning a handler can safely dispatch new events without deadlocking itself. This is crucial for complex event chains where handlers need to trigger subsequent events.

The Custom Asyncio Queue

bubus uses a CleanShutdownQueue (lines 56-84 in bubus/service.py) instead of a standard asyncio.Queue. This custom queue overrides put() and get() to raise a dedicated QueueShutDown exception during graceful shutdown, allowing the run-loop to exit cleanly without "queue closed" warnings while ensuring all pending events are processed before shutdown completes.

Race Condition Prevention Mechanisms

Serializing Event Processing

The _run_loop task repeatedly calls step(), which acquires the global lock before processing any event. While the lock is held, process_event() gathers all applicable handlers and creates pending EventResult objects. This serialization ensures that event.results is never written to by multiple handlers simultaneously, even when parallel_handlers=True allows handlers to execute concurrently.

Preventing Infinite Forwarding Loops

When handlers forward events to other buses, the _would_create_loop() method (lines 332-348 in bubus/service.py) checks the current event's event_path attribute. If the target bus already appears in the path, the forward is skipped, preventing cycles like A→B→A that could cause race conditions or infinite recursion.

Blocking Recursive Handler Dispatches

The _handler_dispatched_ancestor() method (lines 889-901 in bubus/service.py) walks the event ancestry chain and aborts if recursion depth exceeds two levels. If a handler attempts to recursively dispatch the same event type that triggered it, bubus raises a clear RuntimeError, preventing stack overflow and state corruption.

Deadlock Detection for Long-Running Handlers

Each handler execution spawns a deadlock_monitor task (lines 1100-1116 in bubus/service.py) that logs a warning after 15 seconds. This helps identify handlers that block the global lock for too long, allowing developers to refactor blocking code before it causes system-wide latency.

Context Tracking for Child Events

Context variables (_current_event_context, inside_handler_context, _current_handler_id_context) defined in bubus/service.py (lines 37-45) track the parent event and handler ID. When handlers create child events, these contexts ensure proper parent-child linking and automatic cleanup when the parent completes, preventing orphaned events and memory leaks.

Practical Code Examples

Concurrent Event Dispatch

import asyncio
from bubus import EventBus, BaseEvent

class PingEvent(BaseEvent):
    event_type = "Ping"

bus = EventBus(parallel_handlers=True)          # handlers run concurrently

@bus.on("*")
async def logger(event: BaseEvent):
    print(f"🪵 Received {event}")

@bus.on(PingEvent)
async def handle_ping(event: PingEvent):
    await asyncio.sleep(0.1)                    # simulate work

    return "pong"

async def fire():
    # fire 10 events at once – they are queued safely

    tasks = [bus.dispatch(PingEvent()) for _ in range(10)]
    results = await asyncio.gather(*tasks)      # each result is a completed PingEvent

    print([r.event_result() for r in results]) # ["pong", …]

asyncio.run(fire())

Why it’s race-free: all dispatch() calls share the same CleanShutdownQueue; the run-loop processes them one-by-one while holding the global lock, so event.event_results is never written concurrently.

Safe Bus Forwarding

bus_a = EventBus(name="A")
bus_b = EventBus(name="B")

@bus_a.on("*")
async def forward_to_b(event):
    # safe forwarding – the path check in _would_create_loop() prevents A→B→A cycles

    await bus_b.dispatch(event)

@bus_b.on("*")
async def log_b(event):
    print(f"B got {event}")

await bus_a.dispatch(PingEvent())

If bus_b later tried to forward the same event back to bus_a, _would_create_loop() would detect bus_a already in event_path and skip the handler, avoiding an infinite loop.

Waiting for Idle State

await bus.wait_until_idle(timeout=5)   # blocks until the queue is empty and all handlers finished

The idle waiter yields to the event loop (await asyncio.sleep(0)) so any child events created during a handler are processed before the function returns, guaranteeing a clean, race-free state.

Key Implementation Files

File	What it Provides (relevant sections)
bubus/service.py	Core EventBus implementation, CleanShutdownQueue, ReentrantLock, global lock handling, event dispatch, loop, race-prevention helpers (_would_create_loop, _handler_dispatched_ancestor).
bubus/models.py	BaseEvent and EventResult definitions, context-variable tracking, completion signalling, child-event hierarchy.
bubus/helpers.py	Utility helpers (logging, pretty-printing) used by the bus – useful for debugging concurrent scenarios.

These files together provide the complete concurrency safety model in bubus.

Summary

• Global serialization: A single re-entrant lock in _run_loop ensures only one event is processed at a time, preventing concurrent writes to shared state.
• Safe forwarding: The _would_create_loop() check prevents infinite forwarding cycles between buses that could cause deadlocks.
• Recursion guards: _handler_dispatched_ancestor() blocks recursive event dispatches that would otherwise corrupt the event stack.
• Graceful shutdown: CleanShutdownQueue allows the bus to finish processing queued events before shutting down, preventing dropped events.
• Parallel execution: While event processing is serialized, handlers can run in parallel when parallel_handlers=True, with results safely merged under the global lock.

Frequently Asked Questions

How does bubus prevent race conditions when multiple tasks dispatch events simultaneously?

bubus serializes all event processing through a global re-entrant lock acquired in the _run_loop method of bubus/service.py. While multiple coroutines can call dispatch() concurrently, adding events to the CleanShutdownQueue, the run-loop processes them one-by-one under the lock. This ensures that event.results and other shared state are never modified by multiple handlers simultaneously, even when parallel_handlers=True allows concurrent handler execution.

What prevents infinite loops when forwarding events between multiple buses?

The _would_create_loop() method in bubus/service.py (lines 332-348) tracks the event_path attribute of each event. Before forwarding an event to another bus, it checks if the target bus name already exists in the path. If a cycle is detected (e.g., Bus A → Bus B → Bus A), the forward is skipped, preventing infinite recursion and the race conditions that would occur from unbounded event chaining.

How does bubus handle graceful shutdown while events are still processing?

bubus uses a CleanShutdownQueue class (lines 56-84 in bubus/service.py) that overrides standard asyncio.Queue methods. When shutdown is initiated, the queue raises a dedicated QueueShutDown exception instead of the standard "queue closed" warning, allowing the _run_loop to exit cleanly. The bus waits for all currently executing handlers to complete and for child events to finish before shutting down, ensuring no events are dropped or left in an inconsistent state.

Can handlers run in parallel, and how is thread safety maintained?

Yes, handlers can run in parallel when the EventBus is initialized with parallel_handlers=True. However, thread safety is maintained because the global lock in _run_loop serializes the dispatch and collection phases—gathering handlers and merging results—while only the handler execution runs concurrently. The event_completed_signal is set only when all handlers and child events finish, and context variables track parent-child relationships, ensuring that parallel execution never corrupts shared event state.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}