Lattice

EventBus Internals

Deep dive: subscriber queues, slow-consumer drop, MQTT bridge.

EventBus Internals (v2026-08-02T101544Z-ab2bbdebf5) 2026-08-01 | Deep dive: subscriber queues, slow-consumer drop, MQTT bridge. 1 - EventBus Initialization EventBus.__init__ daemon/src/event_bus.py:631 SUBSCRIBER_MAX_DEPTH _load_lvc event_bus_lvc.db kv/events/{event_type} CREATE TABLE IF NOT EXISTS lvc subscriber que… 2 - Event Publishing EventBus.publish daemon/src/event_bus.py:publish kv/events/{event_type} _SUBSCRIBER_EVICT_OVERFLOWS MQTT client client.publish(topic, message) NATS module nats_bus.publish_event(event_type, event) event data MQTT topic 3 - Event Subscription EventBus.subscribe daemon/src/event_bus.py:subscribe asyncio.Queue(maxsize=_SUBSCRIBER_MAX_DEPTH) SSE subscriber GET /api/events filtered events subscriber reg… 1. Initialize EventBus with max depth and LVC… 2. Publish event to subscribers, MQTT, and NATS 3. Subscribe new SSE client with filters 4. Handle overflow by dropping old events EventBus publish payload { "e": "holon:queued", "ts": <unix_float>, "id": "<uuid8>", "holon_id": "...", "to": "flynn" } KVEvent JSON structure { "ts": "2026-05-12T16:30:00+00:00", "instance": "instance-a", "service": "daemon", "event_type": "startup", "severity":… EventBus handles both internal and external event publishing/subscribing. MQTT and NATS are optional transports, controlled by mount methods. Legend actor = initiates work · process = code path · store = state on disk · bus = durable queue · dashed = separate process cyan = request flow · pink = state read/write · dashed green = pull / return path · red = refusal

Provenance

This drawing is generated, not drawn. It is rebuilt from the source files below, so when they change the picture changes — a diagram here cannot quietly describe a system that no longer works this way.

Owner
Lattice
Slug
event-bus-internals
Rendered
2026-08-02T101544Z
From commit
ab2bbdebf5
Watches
5 paths
  • daemon/src/event_bus.py
  • daemon/src/event_emitter.py
  • daemon/src/event_types.py
  • daemon/src/bus_aggregator.py
  • daemon/src/bus_stream_api.py