# How FIFO Event Processing Works in bubus: Queue Architecture and Event Flow

> Discover how bubus ensures FIFO event processing using CleanShutdownQueue and a single run loop. Learn about its queue architecture and event flow for reliable order.

- Repository: [Browser Use/bubus](https://github.com/browser-use/bubus)
- Tags: deep-dive
- Published: 2026-02-26

---

**bubus guarantees strict first-in-first-out (FIFO) event ordering by wrapping `asyncio.Queue` in `CleanShutdownQueue` and consuming events sequentially in a single run loop, isolating dispatch order from handler execution time.**

The bubus library implements deterministic FIFO event processing through a specialized queue architecture that ensures events are handled in the exact order they are dispatched. This design prevents race conditions and reordering even when event handlers have varying execution times. Understanding the underlying mechanics in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) reveals how the library maintains ordering guarantees in asynchronous applications.

## Core Architecture for FIFO Guarantees

### CleanShutdownQueue – The Ordered Buffer

The foundation of bubus FIFO event processing resides in `CleanShutdownQueue`, defined in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) (lines **56‑71**). This thin wrapper inherits directly from `asyncio.Queue` and preserves the standard FIFO semantics while adding graceful shutdown capabilities. Because it never reorders items, events remain in strict enqueue order regardless of system load.

### EventBus.dispatch() – Atomic Enqueue

When an event enters the system, the `EventBus.dispatch()` method (lines **62‑68** in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) validates the event, appends the current bus name to `event.event_path`, and immediately places the event on the queue. The implementation uses `self.event_queue.put_nowait(event)`, which appends the event to the **tail** of the underlying `asyncio.Queue`, preserving the order of arrival.

### The Single-Consumer Run Loop

A dedicated background task ensures FIFO consumption. The `_run_loop` method (lines **64‑71**) continuously executes `await self.step()`, which internally calls `_get_next_event`. This method performs `await self.event_queue.get()` (lines **90‑99**), retrieving the **oldest** pending item from the queue head. By utilizing a single consumer pattern, bubus eliminates any possibility of newer events bypassing older ones during processing.

## Event Flow from Dispatch to Handler

The complete FIFO pipeline follows five distinct stages:

1. **Event creation** – Developers create a subclass of `BaseEvent` (defined in [`bubus/models.py`](https://github.com/browser-use/bubus/blob/main/bubus/models.py)) containing the specific payload.
2. **Dispatch** – Calling `eventbus.dispatch(event)` enqueues the item via `put_nowait`, ensuring it lands at the tail of the `CleanShutdownQueue`.
3. **Background activation** – Upon the first dispatch, `_start()` spawns the asynchronous `_run_loop` task if not already running.
4. **Ordered dequeue** – `_get_next_event` blocks on `await self.event_queue.get()`, returning events in the exact order they were inserted.
5. **Protected processing** – The system acquires a global re-entrant lock via `_get_global_lock()`, then executes `process_event` to run all registered handlers. Even with `parallel_handlers=True`, the **event itself** completes before the queue advances to the next entry.

## Validation Through Testing

The repository includes rigorous FIFO verification in [`tests/test_eventbus.py`](https://github.com/browser-use/bubus/blob/main/tests/test_eventbus.py) within the `TestFIFOOrdering` class. This test deliberately introduces asymmetric handler delays—sleeping 0.05 seconds for even-numbered events and 0.01 seconds for odd-numbered events—to prove that processing order depends solely on enqueue sequence.

```python
class TestFIFOOrdering:
    """Test FIFO event processing"""
    async def test_fifo_with_varying_handler_delays(self, eventbus):
        processed_order = []
        handler_start_times = []

        async def handler(event: UserActionEvent) -> int:
            order = event.metadata.get('order', -1)
            handler_start_times.append((order, asyncio.get_event_loop().time()))
            if order % 2 == 0:
                await asyncio.sleep(0.05)   # even events are slower

            else:
                await asyncio.sleep(0.01)   # odd events are fast

            processed_order.append(order)
            return order

        eventbus.on('UserActionEvent', handler)

        # Emit 20 events rapidly

        for i in range(20):
            eventbus.dispatch(UserActionEvent(action=f'test_{i}', user_id='u1',
                                             metadata={'order': i}))

        await eventbus.wait_until_idle()

        # FIFO order must be preserved

        assert processed_order == list(range(20))
        # Handler start times must be monotonic

        for i in range(1, len(handler_start_times)):
            assert handler_start_times[i][1] >= handler_start_times[i - 1][1]

```

The assertions succeed only because the internal queue and run loop preserve strict FIFO semantics, regardless of individual handler execution speed.

## Implementing FIFO Event Processing in Your Application

### Basic FIFO Usage

The following example demonstrates strict ordering with sequential handlers:

```python
from bubus import EventBus, BaseEvent
import asyncio

class ClickEvent(BaseEvent):
    action: str
    user_id: str

async def logger(event: ClickEvent):
    print(f"🔔 {event.action} by {event.user_id}")

async def main():
    bus = EventBus()
    bus.on('ClickEvent', logger)

    # Dispatch several events quickly

    for i in range(5):
        bus.dispatch(ClickEvent(action=f'click_{i}', user_id='alice'))

    # Wait until all queued events are processed

    await bus.wait_until_idle()

asyncio.run(main())

```

Events print in the exact sequence `click_0` through `click_4` because `CleanShutdownQueue` maintains insertion order.

### Monitoring Queue Depth

For debugging or backpressure management, inspect the queue size directly:

```python
print("Queue size before:", bus.event_queue.qsize())
await bus.wait_until_idle()
print("Queue size after :", bus.event_queue.qsize())

```

### Parallel Handlers (Order Preserved)

Even when enabling concurrent handler execution, event ordering remains guaranteed:

```python
bus = EventBus(parallel_handlers=True)   # handlers for the SAME event may run concurrently

```

With `parallel_handlers=True`, handlers for a single event execute simultaneously, but the bus still dequeues the next event only after the current event completes, preserving FIFO flow.

## Summary

- **CleanShutdownQueue** in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) (lines **56‑71**) provides the underlying FIFO buffer by extending `asyncio.Queue` without reordering logic.
- **EventBus.dispatch()** (lines **62‑68**) enqueues events atomically using `put_nowait`, appending them to the queue tail.
- **Single-consumer architecture** via `_run_loop` and `_get_next_event` (lines **64‑71**, **90‑99**) pulls events from the queue head in strict insertion order.
- **Execution isolation** ensures that handler duration never affects the dequeue sequence, as validated by the `TestFIFOOrdering` suite in [`tests/test_eventbus.py`](https://github.com/browser-use/bubus/blob/main/tests/test_eventbus.py).

## Frequently Asked Questions

### Does bubus guarantee FIFO order when handlers run concurrently?

**Yes.** Even with `parallel_handlers=True`, bubus maintains strict FIFO event processing. The library launches handlers for the current event in parallel, but the `_run_loop` will not fetch the next event from `CleanShutdownQueue` until all handlers for the current event have been invoked. The queue itself remains sequentially processed.

### How does CleanShutdownQueue differ from a standard asyncio.Queue?

**CleanShutdownQueue** adds graceful shutdown signaling to the standard `asyncio.Queue` while preserving identical FIFO semantics. Located in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) (lines **56‑71**), it ensures that events are never reordered during system shutdown, allowing pending items to process in their original sequence before the bus terminates.

### Can event order be affected by slow handlers?

**No.** The test suite in [`tests/test_eventbus.py`](https://github.com/browser-use/bubus/blob/main/tests/test_eventbus.py) specifically validates this scenario by assigning longer `asyncio.sleep` durations to even-numbered events. Because the `_run_loop` waits for `process_event` to complete before calling `_get_next_event` again, slow handlers only delay subsequent events—they never allow later events to bypass earlier ones in the processing sequence.

### How do I verify FIFO processing in my own tests?

Use the `wait_until_idle()` method to block until the queue empties, then assert on the processing order. Create events with sequence metadata, dispatch them rapidly, and verify that the processed array matches the dispatch array exactly. The `bubus` test suite provides the `TestFIFOOrdering` class as a reference implementation for this validation pattern.