Struct DistributedBus 

pub struct DistributedBus { /* private fields */ }

Clustered EventBus: a doorbell over the durable store (spec §3.2).

Implementations

impl DistributedBus

pub async fn connect(url: &str, store: Arc<dyn Store>) -> Result<Self>

Build a bus on its own pool. Each live subscription pins one connection for its LISTEN, so the pool is sized above sqlx’s default of 10 to allow a useful SSE fan-out before subscribe/publish start queuing on the pool; raise it further (or share a pool) for high-fan-out deployments.

pub fn new(pool: PgPool, store: Arc<dyn Store>) -> Self

Build a bus on an existing pool (shared-pool wiring — size the pool for the SSE fan-out, since each subscription pins a LISTEN connection).

pub async fn subscribe_all_ready(&self) -> Result<BusSubscription>

EventBus::subscribe_all, but the LISTEN is established before this returns rather than lazily on the stream’s first poll (#132).

The lazy form cannot tell a caller when it is safe to publish. A doorbell rung in the gap is simply not heard, and because the per-key high-water starts at 0, the miss is only repaired by the next doorbell for that key — which, for a stream that just went quiet, may never come. Ringing repeatedly and hoping is what the #132 test was doing, and it still starved to zero deliveries on a contended runner.

Connection and LISTEN failures surface here as an Err instead of silently ending the stream, which is the other thing the lazy form cannot express.

Trait Implementations

impl EventBus for DistributedBus

fn publish(&self, event: FleetEvent)

Fan an event out to current subscribers. Never blocks on slow/absent receivers (delivery is best-effort relative to the durable store).

fn subscribe(&self, stream_key: &str) -> BusSubscription

A live subscription to one stream’s events. The returned stream yields only events whose stream key equals stream_key; consumers dedup the initial-history/live overlap on seq.

fn subscribe_all(&self) -> BusSubscription

A live subscription to EVERY stream’s events, unfiltered. Needed by fleet-wide watchers (e.g. FleetManager::watch_changes) that can’t name a stream key in advance — a brand-new agent’s own id keys its AgentDiscovered event, so no one can pre-subscribe to it by key.