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

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 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 (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) 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) 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 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.

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:

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:

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:

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 (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.

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 (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 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.

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