eventbus

EventBus Package

Asynchronous, non-blocking event bus for observability and instrumentation events in the LLM proxy.

Purpose & Responsibilities

• Non-blocking Event Publishing: Events published asynchronously to avoid blocking request handling
• Fan-out Broadcasting: Multiple subscribers receive all published events (in-memory backend only)
• Backend Flexibility: In-memory (single process) or Redis Streams (distributed with guaranteed delivery)
• Graceful Degradation: Buffer overflow drops events rather than blocking (in-memory backend)
• At-least-once Delivery: Redis Streams backend with consumer groups, acknowledgment, and crash recovery

Architecture

flowchart TB
    subgraph Publishers
        P1[Proxy Instance 1]
        P2[Proxy Instance 2]
    end
    
    subgraph EventBus["Event Bus (In-Memory, Redis List, or Redis Streams)"]
        Buffer[Event Buffer]
    end
    
    subgraph Subscribers
        D[Dispatcher Service]
        M[Metrics Collector]
    end
    
    P1 -->|Publish| Buffer
    P2 -->|Publish| Buffer
    Buffer -->|Subscribe| D
    Buffer -->|Subscribe| M

Implementations

In-Memory EventBus

Best for single-process deployments and local development.

Key Functions: NewInMemoryEventBus(bufferSize), Publish(), Subscribe(), Stop(), Stats()

Redis Streams EventBus (Recommended for Production)

Best for distributed deployments requiring durable, guaranteed delivery with consumer groups.

Key Functions: NewRedisStreamsEventBus(), Publish(), Subscribe(), Stop(), Acknowledge(), StreamLength(), PendingCount()

Consumer Group Semantics

Redis Streams implements the competing consumers pattern:

1. Publishing: Events are added to stream via XADD
2. Consuming: Dispatchers read via XREADGROUP with consumer groups
3. Acknowledgment: Successfully processed messages are acknowledged with XACK
4. Recovery: Pending messages from crashed consumers are claimed via XCLAIM

sequenceDiagram
    participant Proxy
    participant Stream as Redis Stream
    participant D1 as Dispatcher 1
    participant D2 as Dispatcher 2
    
    Proxy->>Stream: XADD event
    D1->>Stream: XREADGROUP (blocks)
    Stream->>D1: Event (assigned to D1)
    D1->>D1: Process Event
    D1->>Stream: XACK (acknowledge)
    
    Note over D2: D2 claims stuck messages<br/>if D1 crashes
    D2->>Stream: XCLAIM pending msgs

Event Schema

The Event struct captures HTTP request/response data for observability:

Configuration Options

General Options

Redis Streams Options

Integration Flow

sequenceDiagram
    participant Client
    participant Proxy
    participant Middleware
    participant EventBus
    participant Dispatcher
    participant Backend
    
    Client->>Proxy: HTTP Request
    Proxy->>Middleware: Process Request
    Middleware->>EventBus: Publish(Event)
    EventBus-->>Middleware: (non-blocking)
    Middleware->>Proxy: Response
    Proxy->>Client: HTTP Response
    
    EventBus->>Dispatcher: Event (via Subscribe)
    Dispatcher->>Backend: SendEvents(batch)

Testing Guidance

• Unit Tests: Create a mock implementing EventBus interface to capture published events
• Integration Tests: Use NewInMemoryEventBus with small buffer size to test overflow behavior
• Redis Streams Tests: Use miniredis for integration testing with Redis Streams
• Existing Tests: See eventbus_test.go and redis_streams_test.go for comprehensive examples

Related Documentation

• Dispatcher Package - Event processing and backend delivery
• Instrumentation Guide - Complete observability documentation

Files