Skip to main content

AD-012: Distributed Event Bus

Summary​

Bowrain-server and bowrain-worker replicas coordinate through a shared event broker. The EventBus interface has one production backend — Redis Streams (ElastiCache on AWS; a stock redis container for local and self-hosted stacks) — opted into with BOWRAIN_EVENT_BACKEND=redis. Every SubscribeGroup component becomes a Redis Streams consumer group, so any replica can consume events with no leader election; Subscribe/SubscribeAll are fan-out (every replica sees every event), which is what the SSE/gRPC relays need.

Redis carries the bus because it is already a platform dependency (sessions, the sync-hash cache, agent pub/sub) and is managed on AWS as ElastiCache — the event bus needs no separate broker, so the compute node holds no queue state. Self-hosted stacks run the same pair the cloud does: one broker story everywhere, one fewer container to operate.

Context​

Bowrain's automation engine (AD-013), activity recorder, notification dispatcher, push-completion tracker, and audit logger all react to platform events. Running multiple server replicas — behind a load balancer, across zones, or in a worker pool — requires a broker that delivers each event to exactly one member of each consumer group and retains messages across replica restarts. An in-process channel bus cannot meet either requirement.

Decision​

EventBus interface​

type EventBus interface {
    Publish(ctx context.Context, event Event) error
    Subscribe(ctx context.Context, eventType EventType, h EventHandler) (Subscription, error)
    SubscribeAll(ctx context.Context, h EventHandler) (Subscription, error)
    Unsubscribe(sub Subscription) error
}

type EventHandler func(ctx context.Context, event Event) error

Two implementations:

ImplementationBackendPurpose
ChannelEventBusGo channels (in-process)Unit tests and single-process development
RedisEventBusRedis StreamsProduction (ElastiCache), local + self-hosted

RedisEventBus publishes with a single XADD (capped by MAXLEN so the stream stays bounded). SubscribeGroup reads via XREADGROUP on a Redis Streams consumer group — competing consumers, position preserved across restarts. Subscribe/SubscribeAll read the stream tail with XREAD independently, so every such subscriber sees every event (fan-out). The starting position is resolved synchronously at subscribe time, so an event published immediately afterward is not lost to a $-resolution race.

Runtime selection​

The server picks a backend from configuration:

if os.Getenv("BOWRAIN_EVENT_BACKEND") == "redis" && cfg.RedisURL != "" {
    bus = event.NewRedisEventBus(cfg.RedisURL, cfg.RedisPassword)
} else {
    bus = event.NewChannelEventBus()
}

A test binary uses the channel bus. Docker Compose and the AWS deployment set BOWRAIN_EVENT_BACKEND=redis and point BOWRAIN_REDIS_URL at the shared Redis/ElastiCache — the same instance the session store and sync-hash cache already use. The worker selects the backend the same way, so server and worker share one broker.

Event model​

type Event struct {
    ID          string
    Type        EventType
    Source      string             // connector ID, tool name, or "user"
    ProjectID   string
    CausationID string             // ID of the event that caused this one (loop prevention)
    Payload     any
    Timestamp   time.Time
}

Events serialize as JSON. Every event has a type drawn from a registered schema; subscribers filter on the Type field.

Registered event types include:

Event typeEmitter
content.changedEventEmittingStore on mutations
content.extractedConnector pull
translation.updatedBlock target updates
translation.reviewedReview decisions
connector.syncedConnector completion
flow.completed / flow.failedFlow executor (type defined; not yet emitted)
quality.gate.failedQuality gate evaluation
terminology.changedTerms store mutations
push.completedSync push commit
push.automations.completedPushCompletionTracker (AD-014)
project.updatedProject settings / locale additions
run.started / run.completedAutomationRunManager (AD-013)
agent.*Bravo agent (AD-016)

Consumer groups​

One Redis Streams consumer group per subscriber component:

Stream: bowrain:events
  ├── Group: automations         → AutomationEngine
  ├── Group: activity-recorder   → ActivityRecorder
  ├── Group: notifications       → NotificationDispatcher
  ├── Group: push-tracker        → PushCompletionTracker
  ├── Group: progress-tracker    → ProgressTracker
  ├── Group: audit-logger        → AuditLogger
  ├── Group: siem-exporter       → SIEMExporter
  ├── Group: graph-syncer        → GraphSyncer
  ├── Group: convergence-onpush  → on-push convergence trigger (push / new locale → run)
  └── Group: forge-delivery      → forge delivery tier (run completed → pull request)

Consumer groups are for state-advancing subscribers: a missed event would silently strand work (a push that never converges, a converged run that never reaches the forge), so the group's persisted position must survive replica restarts and deploy rollovers. Pure freshness subscribers — the SSE/gRPC change relays and the platform-config cache refresh — stay on fan-out Subscribe/SubscribeAll: every instance must react, and a missed event is healed by the next read or reconnect.

No leader election​

Any bowrain-server or bowrain-worker replica can consume from any subscription. The broker, not the application, owns delivery semantics. This eliminates custom coordination — there is no LeaderElector, no lease table, no IsLeader gating, no polling to discover work.

Test backend​

ChannelEventBus remains available for unit tests that exercise the event flow without external infrastructure. Integration tests against real behavior use Redis via Docker Compose.

Consequences​

  • Horizontal scaling is a deployment concern, not an application concern. Adding replicas does not require code changes.
  • Events published on any replica reach exactly one consumer per group.
  • Failover is automatic: unacked messages redeliver when a replica crashes or is drained.
  • Zero-delay event flow — no polling intervals, no 5-second sleep in trackers.
  • Local development stays fast because Redis is already part of the Docker Compose topology; no extra service is required.
  • Tests remain deterministic with the in-process bus.