Wire a bus into a controller¶

bus, err := messaging.New(backend, messaging.Settings{Subscriptions: specs})
if err != nil {
    // A degraded construction still returns a usable bus. See the guide on it.
    return err
}

controller.Register("messaging",
    controls.WithStart(bus.Start),
    controls.WithStopErr(bus.Stop),
    controls.WithReadiness(bus.Readiness),
)

controller.RegisterHealthCheck(bus.HealthCheck("messaging-subscriptions"))

Register the bus, never the subscriptions¶

One registration, whatever the subscription count. That is deliberate, and it is the difference between a process that survives one broken consumer and one that does not.

A registered service whose readiness probe fails sets OverallHealthy: false for the whole process, so a process that cannot reach its bus reports itself out of rotation. That coupling is wanted at the bus: a service that cannot talk to messaging generally cannot do its job.

It is emphatically not wanted per subscription. If each subscription were a registered service, one broken handler would take the process out of rotation and the other subscriptions with it. Instead each subscription is a child of a controls.Supervisor the bus owns, and a failed child never makes the supervisor unready, at any proportion.

Two names, not one¶

controller.Register("messaging", ...)
controller.RegisterHealthCheck(bus.HealthCheck("messaging-subscriptions"))

The check name must differ from the service name. A Controller's contract is that a check name is unique across both services and health checks, but only the health-check map is enforced, so a collision is silently accepted and the report carries two entries with the same name.

What each signal tells you¶

st, ok := bus.Stats("orders")
if !ok {
    // no subscription of that name was ever declared
}

st.Pending      // waiting in the current source's queue, a gauge of now
st.TotalLost()  // everything shed at the bound, whatever the reason
st.Errors       // handlers that returned an error: the contract working
st.Panics       // handlers that panicked: a bug

ConsumerStats is one struct rather than four separate calls, because a discard, an error and a panic are three different questions and a caller reading one usually wants to read all three together. Pending and TotalLost() are a pair, and the point of the first is to see the second coming.

Three counters, three meanings, and no two of them share a number. A handler error is the contract working, a shed is policy working, and a panic is a bug. Conflating any two lets one hide inside another, and the reason each is worth watching is that it normally means one thing.

Pass a *slog.Logger in Settings to see the detail behind them: a handler error and a terminal subscription failure are both logged with the subscription name and the event identity, because "a handler failed" without saying on which event is not actionable.

The health check says something failed, not which¶

HealthCheck reports DEGRADED with a count. If you need to know which subscription died, and in a multi-tenant service you do, that comes from bus.Stats(name).Panics per subscription, or from the supervisor's own failure reporting. Plan the operator-facing surface around that rather than around the probe.