How to Handle Errors and Exceptions in Bubus Event Handlers: A Complete Guide

Bubus automatically converts every exception into a structured EventResult with status "error", surfaces it through awaitable APIs, and cancels pending child events to prevent resource leaks.

Bubus is an asynchronous event-bus library developed by browser-use that guarantees FIFO processing and automatic child-event tracking. Understanding how to handle errors and exceptions in bubus event handlers is essential for building resilient applications that can gracefully manage failures without losing error context or leaving orphaned tasks.

How Bubus Captures and Stores Handler Errors

When a handler runs, Bubus wraps execution in a try-catch block that converts any raised exception into a structured result record. This ensures errors are never lost, even if the caller does not immediately await the event.

The Error Conversion Pipeline

In bubus/models.py, the EventResult.update() method automatically rewrites returned exception objects into proper error states:

# Bubus automatically rewrites a returned exception into an error result

if 'result' in kwargs and isinstance(kwargs['result'], BaseException):
    logger.warning(
        f'ℹ Event handler {self.handler_name} returned an exception object, '
        f'auto-converting to EventResult(result=None, status="error", error={kwargs["result"]})'
    )
    kwargs['error'] = kwargs['result']
    kwargs['status'] = 'error'
    kwargs['result'] = None

Source: EventResult.update() – lines 1000‑1007

Exception Handling in execute_handler()

When a handler raises during execution, EventBus.execute_handler() in bubus/service.py catches the exception, records it via event_result_update(), and re-raises for observability:

except Exception as e:
    monitor_task.cancel()
    event.event_result_update(handler=handler, eventbus=self, error=e)
    logger.error(
        f'❌ {self} Error in event handler {get_handler_name(handler)}({event}) -> '
        f'\n{red}{type(e).__name__}({e}){reset}\n{_log_filtered_traceback(e)}',
    )
    raise

Source: execute_handler error branch – lines 1194‑1198

Timeout Handling and Automatic Cancellation

Bubus enforces per-handler timeouts and automatically cleans up pending child events when a parent handler fails or exceeds its time limit. This prevents resource leaks and ensures that failures do not leave orphaned tasks running.

Configuring Handler Timeouts

When a handler exceeds its allocated time, Bubus raises a TimeoutError, updates the corresponding EventResult, and cancels any pending child events:

except TimeoutError as e:
    monitor_task.cancel()
    children = (
        f' and interrupted any processing of {len(event.event_children)} child events'
        if event.event_children else ''
    )
    handler_timeout_error = TimeoutError(
        f'Event handler {get_handler_name(handler)}#{handler_id[-4:]}({event}) '
        f'timed out after {event_result.timeout}s{children}'
    )
    event.event_result_update(handler=handler, eventbus=self, error=handler_timeout_error)
    event.event_cancel_pending_child_processing(handler_timeout_error)
    from bubus.logging import log_timeout_tree
    log_timeout_tree(event, event_result)
    raise handler_timeout_error from e

Source: execute_handler timeout branch – lines 1184‑1192

Child Event Cancellation on Failure

The event_cancel_pending_child_processing() method in bubus/models.py recursively propagates cancellation to all pending child events when a parent handler fails:

def event_cancel_pending_child_processing(self, error: BaseException) -> None:
    if not isinstance(error, asyncio.CancelledError):
        error = asyncio.CancelledError(
            f'Cancelled pending handler as a result of parent error {error}'
        )
    for child_event in self.event_children:
        for result in child_event.event_results.values():
            if result.status == 'pending':
                result.update(error=error)
        child_event.event_cancel_pending_child_processing(error)

Source: event_cancel_pending_child_processing – lines 742‑754

Retrieving Errors from Event Results

Bubus provides multiple patterns for accessing error details after dispatch, allowing you to either fail fast or inspect failures programmatically.

Awaiting Events and Results

The simplest pattern awaits the event directly, which raises the first error encountered:

# Dispatch an event and await its completion.

completed_event = await bus.dispatch(MyEvent())

# If any handler errored, awaiting the event will raise the first error.

BaseEvent.__await__ forwards to event.event_result(), which surfaces stored exceptions.

For granular control, retrieve results by handler ID:

result = await bus.dispatch(MyEvent()).event_result()

# Returns the first *truthy* result; raises if the handler errored.

Inspecting All Handler Results

To handle errors without raising, iterate over the full results mapping:

ev = await bus.dispatch(MyEvent())
for handler_id, res in ev.event_results.items():
    if res.status == 'error':
        print(f'❗ Handler {res.handler_name} failed →', res.error)
    else:
        print(f'✅ Handler {res.handler_name} returned →', res.result)

Using Filter Helpers

The event_results_filtered() method provides declarative error extraction:

# Get only successful results.

good = await ev.event_results_filtered()

# Get only errors without raising.

errors = await ev.event_results_filtered(
    include=lambda r: r.status == 'error',
    raise_if_any=False,
    raise_if_none=False,
)

Source: event_results_filtered implementation – lines 748‑784

Best Practices for Robust Error Handling

Implementing defensive patterns ensures your event-driven application remains stable under failure conditions.

Explicitly Catching Expected Errors

Wrap risky operations to convert domain exceptions into meaningful error states:

@bus.on('DataEvent')
async def handler(event: DataEvent):
    try:
        # May raise ValueError for bad payload

        processed = await process(event.payload)
    except ValueError as exc:
        # Convert to a controlled error result

        raise RuntimeError('Invalid payload') from exc   # Bubus will record this as error

    return processed

Returning Sentinels Instead of Raising

For non-fatal failures, return None or a sentinel object to keep the event flow alive:

@bus.on('OptionalEvent')
def handler(event):
    if not event.payload:
        return None   # Treated as “no result”, not an error.

    return compute(event.payload)

Defensive Timeout Wrapping

Set explicit timeouts for unreliable I/O to trigger Bubus’s automatic cleanup:

from asyncio import wait_for

@bus.on('SlowEvent')
async def handler(event):
    # 5‑second limit

    return await wait_for(do_slow_work(event), timeout=5.0)

If this timeout expires, Bubus converts it into a structured TimeoutError and cancels any pending child events.

Preventing Infinite Loops

Bubus includes loop detection to prevent forwarding cycles that could mask errors. In bubus/service.py, the _would_create_loop() method checks the event path before dispatching:

if hasattr(handler, '__self__') and isinstance(handler.__self__, EventBus) and handler.__name__ == 'dispatch':
    target_bus = handler.__self__
    if target_bus.name in event.event_path:
        logger.debug(
            f'⚠️ {self} handler {get_handler_name(handler)}… '
            f'skipped to prevent infinite forwarding loop with {target_bus.name}'
        )
        return True

Source: _would_create_loop() – lines 3330‑3338

When a loop is detected, the handler is silently skipped, preventing stack-overflow crashes and ensuring that errors in cyclic forwarding chains are surfaced rather than masked by infinite recursion.

Summary

• Bubus converts every exception into a structured EventResult with status "error" via EventResult.update() in bubus/models.py, ensuring no errors are lost.
• Timeouts trigger automatic cancellation: when a handler exceeds its limit, Bubus raises TimeoutError, updates the corresponding EventResult, and cancels any pending child events via event_cancel_pending_child_processing().
• Errors surface through awaitable APIs: awaiting an event or calling event_result() raises the first error, while event_results_filtered() allows inspection without raising.
• Loop detection prevents infinite forwarding cycles via EventBus._would_create_loop() in bubus/service.py, stopping stack-overflow crashes before they occur.
• Best practices include explicit exception catching, returning sentinels for non-fatal failures, and wrapping unreliable I/O with defensive timeouts to trigger Bubus’s built-in cleanup mechanisms.

Frequently Asked Questions

What happens when a Bubus event handler raises an exception?

When a handler raises an exception, Bubus catches it in EventBus.execute_handler() (located in bubus/service.py), converts it into an EventResult with status "error", stores the exception in the error field, and re-raises the exception for observability. The error is now accessible via event.event_results or by awaiting the event, which will raise the stored exception.

How does Bubus handle handler timeouts?

Bubus enforces per-handler timeouts in execute_handler(). If a handler exceeds its allocated time, Bubus raises a TimeoutError, updates the corresponding EventResult with the error status, and automatically calls event_cancel_pending_child_processing() to cancel any pending child events. This prevents resource leaks and ensures that failures in parent handlers do not leave orphaned tasks running.

Can I retrieve errors without raising an exception?

Yes. Instead of awaiting the event directly—which raises the first error encountered—you can inspect the event_results dictionary or use the event_results_filtered() method. For example, you can pass include=lambda r: r.status == 'error' along with raise_if_any=False to obtain a list of failed results without triggering an exception, allowing you to implement custom error recovery logic.

What is the best way to prevent infinite loops when forwarding events?

Bubus automatically prevents infinite forwarding loops via the _would_create_loop() method in bubus/service.py. Before invoking a handler that dispatches to another bus, Bubus checks if the target bus name already exists in the event's path. If a loop is detected, the handler is silently skipped, preventing stack-overflow crashes and ensuring that errors in cyclic forwarding chains are surfaced rather than masked by infinite recursion.

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