Health

health hands out a HealthProbe that reports whether a started service is running, stopping, stopped or failed, and how many of its subscriptions consume.

The probe is a cheap clone, so each task that reports health holds its own. Its state is a HealthState: Running, ShuttingDown, Stopped or Failed with the fail-fast diagnostic. The probe outlives shutdown and keeps reporting the final state. stopping resolves when the service tears itself down after a fail-fast failure, so a server beside it can stop with it, as the HTTP page shows.

subscriptions returns a SubscriptionCount: the subscriptions that consume and the subscriptions the service started. A subscription counts once, whatever its number of workers. The count changes only when a subscription starts or stops, so a delivery pays nothing for it.

Example

examples/http_outbox.rs

/// The liveness endpoint: the probe outlives `shutdown`, so this route reports the terminal
/// state for as long as requests still reach it. Here that window is the graceful drain: a
/// fail-fast also resolves the `stopping()` future below and stops the HTTP server. A host
/// that should keep serving after the messaging side dies would omit that select arm.
///
/// This service is unhealthy once a subscription stops consuming, even though the rest of it
/// still runs: the orders it no longer reads would wait unanswered.
async fn healthz(State(health): State<HealthProbe>) -> (StatusCode, String) {
    let count = health.subscriptions();
    match health.state() {
        HealthState::Running if count.active == count.started => {
            (StatusCode::OK, "running".to_owned())
        }
        HealthState::Running => (
            StatusCode::SERVICE_UNAVAILABLE,
            format!(
                "{} of {} subscriptions consuming",
                count.active, count.started
            ),
        ),
        state => (StatusCode::SERVICE_UNAVAILABLE, format!("{state:?}")),
    }
}

A subscription that stops

A subscription's stream can end while the service runs: the server cancels a consumer when its queue is deleted, or the broker gives up reconnecting. By default the subscription logs the end and stops, and the rest of the service runs on. The probe still reports Running and counts one subscription fewer, and the service decides whether that is healthy. on_subscription_end with SubscriptionEnd::FailFast makes the end a fail-fast failure instead. The service then shuts down gracefully, run returns an error that names the subscription, and the probe reports Failed. An end during shutdown is part of the shutdown and is never reported.

examples/app_running.rs

/// The service exists to answer orders, so a subscription whose stream ends while the
/// service runs stops the service.
fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_subscription_end(SubscriptionEnd::FailFast)
        .with_broker(broker, |b| {
            b.include(issue_receipt).out_reply(Publish);
        })
}