feat(worker): health/readiness endpoints + awa health CLI probe

CHANGELOG.md

@@ -4,6 +4,11 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Features
+
+- **Worker health & readiness endpoints ([#368](https://github.com/hardbyte/awa/issues/368)).** Opt-in HTTP listener in the worker runtime (`AWA_HEALTH_ADDR` or `ClientBuilder::health_addr`): `GET /healthz` (process liveness, stays `200` through graceful drain) and `GET /readyz` (`503` + JSON reasons when Postgres is unreachable, the schema is older than the binary, a claim loop stalls, heartbeat/maintenance die, or shutdown starts). No new dependencies in `awa-worker`. Kubernetes probe examples in `docs/deployment.md`.
+- **`awa health` CLI probe.** Cluster-level readiness from the database alone (reachable, schema migrated for this binary, storage state, fleet heartbeat counts) with `--json` output and exit codes for probe-less environments.
+
 ## [0.6.0] — 2026-07-04
 
 The 0.6 line makes the append-only **queue-storage engine** the default and the supported substrate for high-throughput, low-bloat operation on managed Postgres. The [#169](https://github.com/hardbyte/awa/issues/169) pinned-MVCC dead-tuple gate **passed** on `main`: after the rc line removed the last dominant hot-row update path (the `queue_claim_heads` ready-segment routing cache), a 60-minute idle-in-transaction soak holds bounded queue depth through the pinned hour and drains fully in recovery. See "Benchmark evidence" below for the caveats — awa is not immune to long readers, and this release does not claim otherwise.

awa-cli/src/health.rs

@@ -0,0 +1,134 @@
+//! `awa health` — cluster-level readiness probe over the database (#368).
+//!
+//! For environments without HTTP probing (cron drivers, systemd
+//! `ExecStartPre`, CI smoke checks): answers "can a worker built from this
+//! binary do useful work against this database, and is the fleet
+//! heartbeating?" from the schema alone. The per-process equivalents are the
+//! worker's `/healthz` / `/readyz` endpoints (`AWA_HEALTH_ADDR`).
+//!
+//! Exit codes: 0 = ready, 1 = not ready (unreachable database, missing or
+//! stale schema). Fleet numbers are informational — an idle cluster with no
+//! workers is still "ready" for producers and migrations.
+
+use serde::Serialize;
+use sqlx::PgPool;
+
+/// The `--json` output. Field names are a documented CLI surface.
+#[derive(Debug, Serialize)]
+pub struct HealthReport {
+    pub ready: bool,
+    pub postgres_connected: bool,
+    pub schema_version: Option<i32>,
+    pub expected_schema_version: i32,
+    pub schema_compatible: bool,
+    pub storage_state: Option<String>,
+    pub active_engine: Option<String>,
+    /// Runtimes whose heartbeat was seen inside the 90s liveness window.
+    pub live_runtimes: i64,
+    /// Live runtimes currently reporting `healthy = false`.
+    pub unhealthy_runtimes: i64,
+}
+
+/// The report shape for a database that could not be reached at all.
+pub fn unreachable_report() -> HealthReport {
+    HealthReport {
+        ready: false,
+        postgres_connected: false,
+        schema_version: None,
+        expected_schema_version: awa_model::migrations::CURRENT_VERSION,
+        schema_compatible: false,
+        storage_state: None,
+        active_engine: None,
+        live_runtimes: 0,
+        unhealthy_runtimes: 0,
+    }
+}
+
+pub async fn probe(pool: &PgPool) -> HealthReport {
+    let postgres_connected = sqlx::query("SELECT 1").execute(pool).await.is_ok();
+
+    let schema_version = if postgres_connected {
+        awa_model::migrations::current_version(pool).await.ok()
+    } else {
+        None
+    };
+    let expected_schema_version = awa_model::migrations::CURRENT_VERSION;
+    // A schema newer than the binary is the supported rolling-deploy skew;
+    // an older schema means `awa migrate` has not run.
+    let schema_compatible =
+        schema_version.is_some_and(|version| version >= expected_schema_version);
+
+    let (storage_state, active_engine) = if schema_compatible {
+        match awa_model::storage::status(pool).await {
+            Ok(status) => (Some(status.state), Some(status.active_engine)),
+            Err(_) => (None, None),
+        }
+    } else {
+        (None, None)
+    };
+
+    let (live_runtimes, unhealthy_runtimes) = if schema_compatible {
+        sqlx::query_as::<_, (i64, i64)>(
+            r#"
+            SELECT
+                count(*)::bigint,
+                count(*) FILTER (WHERE NOT healthy)::bigint
+            FROM awa.runtime_instances
+            WHERE last_seen_at + make_interval(secs => 90) >= now()
+            "#,
+        )
+        .fetch_one(pool)
+        .await
+        .unwrap_or((0, 0))
+    } else {
+        (0, 0)
+    };
+
+    HealthReport {
+        ready: postgres_connected && schema_compatible,
+        postgres_connected,
+        schema_version,
+        expected_schema_version,
+        schema_compatible,
+        storage_state,
+        active_engine,
+        live_runtimes,
+        unhealthy_runtimes,
+    }
+}
+
+pub fn render_human(report: &HealthReport) -> String {
+    let mut lines = Vec::new();
+    lines.push(format!(
+        "ready:              {}",
+        if report.ready { "yes" } else { "NO" }
+    ));
+    lines.push(format!(
+        "postgres:           {}",
+        if report.postgres_connected {
+            "connected"
+        } else {
+            "UNREACHABLE"
+        }
+    ));
+    lines.push(format!(
+        "schema version:     {} (binary expects >= {}){}",
+        report
+            .schema_version
+            .map_or_else(|| "none".to_string(), |v| v.to_string()),
+        report.expected_schema_version,
+        if report.schema_compatible {
+            ""
+        } else {
+            " — run `awa migrate`"
+        }
+    ));
+    if let (Some(state), Some(engine)) = (&report.storage_state, &report.active_engine) {
+        lines.push(format!("storage:            {state} ({engine})"));
+    }
+    lines.push(format!(
+        "live runtimes:      {} ({} unhealthy)",
+        report.live_runtimes, report.unhealthy_runtimes
+    ));
+    lines.join("\n")
+}

awa-cli/src/main.rs

@@ -4,6 +4,7 @@ use chrono::{DateTime, Utc};
 use clap::{Args, Parser, Subcommand};
 use sqlx::postgres::PgPoolOptions;
 
+mod health;
 mod storage_wait;
 
 #[derive(Parser)]
@@ -119,6 +120,15 @@ enum Commands {
         #[command(subcommand)]
         command: CallbackCommands,
     },
+    /// Cluster readiness probe: database reachable, schema migrated, fleet heartbeats
+    Health {
+        /// Emit the report as JSON
+        #[arg(long)]
+        json: bool,
+        /// Seconds to wait for the database connection
+        #[arg(long, default_value = "5")]
+        connect_timeout: u64,
+    },
 }
 
 #[derive(Subcommand)]
@@ -654,6 +664,32 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             axum::serve(listener, app).await?;
         }
 
+        Commands::Health {
+            json,
+            connect_timeout,
+        } => {
+            let db_url = require_pool(&cli.database_url)?;
+            let pool_result = PgPoolOptions::new()
+                .max_connections(1)
+                .acquire_timeout(Duration::from_secs(connect_timeout))
+                .connect(&db_url)
+                .await;
+
+            let report = match pool_result {
+                Ok(pool) => health::probe(&pool).await,
+                Err(_) => health::unreachable_report(),
+            };
+
+            if json {
+                println!("{}", serde_json::to_string_pretty(&report)?);
+            } else {
+                println!("{}", health::render_human(&report));
+            }
+            if !report.ready {
+                std::process::exit(1);
+            }
+        }
+
         // Most remaining CLI commands are single-shot (one query, then exit)
         // so a small pool is sufficient. `storage prepare-queue-storage-schema`
         // is the exception: it acquires an advisory-lock connection and then
@@ -668,7 +704,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 .await?;
 
             match command {
-                Commands::Migrate { .. } | Commands::Serve { .. } | Commands::Callbacks { .. } => {
+                Commands::Migrate { .. }
+                | Commands::Serve { .. }
+                | Commands::Callbacks { .. }
+                | Commands::Health { .. } => {
                     unreachable!()
                 }
 

awa-cli/tests/health_cli_test.rs

@@ -0,0 +1,77 @@
+//! `awa health` CLI probe tests (#368).
+//!
+//! Read-only against the standard test database (plus one unreachable-URL
+//! case), so no schema isolation is needed.
+//!
+//! Set DATABASE_URL=postgres://postgres:test@localhost:15432/awa_test
+
+use assert_cmd::Command;
+use sqlx::postgres::PgPoolOptions;
+
+fn database_url() -> String {
+    std::env::var("DATABASE_URL")
+        .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string())
+}
+
+async fn ensure_migrated() {
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&database_url())
+        .await
+        .expect("Failed to connect to test database — is Postgres running?");
+    awa_model::migrations::run(&pool)
+        .await
+        .expect("migrations should apply");
+}
+
+#[tokio::test]
+async fn test_health_ready_against_migrated_database() {
+    ensure_migrated().await;
+
+    let assert = tokio::task::spawn_blocking(move || {
+        let mut command = Command::cargo_bin("awa").expect("awa binary should build");
+        command
+            .arg("--database-url")
+            .arg(database_url())
+            .arg("health")
+            .arg("--json")
+            .assert()
+    })
+    .await
+    .expect("command task should join");
+
+    let output = assert.success().get_output().stdout.clone();
+    let report: serde_json::Value =
+        serde_json::from_slice(&output).expect("health --json should emit JSON");
+    assert_eq!(report["ready"], true);
+    assert_eq!(report["postgres_connected"], true);
+    assert_eq!(report["schema_compatible"], true);
+    assert_eq!(
+        report["expected_schema_version"],
+        awa_model::migrations::CURRENT_VERSION
+    );
+    assert!(report["storage_state"].is_string());
+}
+
+#[tokio::test]
+async fn test_health_unreachable_database_exits_nonzero_with_report() {
+    let assert = tokio::task::spawn_blocking(move || {
+        let mut command = Command::cargo_bin("awa").expect("awa binary should build");
+        command
+            .arg("--database-url")
+            .arg("postgres://postgres:wrong@127.0.0.1:9/awa_nowhere")
+            .arg("health")
+            .arg("--json")
+            .arg("--connect-timeout")
+            .arg("2")
+            .assert()
+    })
+    .await
+    .expect("command task should join");
+
+    let output = assert.failure().code(1).get_output().stdout.clone();
+    let report: serde_json::Value =
+        serde_json::from_slice(&output).expect("health --json should emit JSON even when down");
+    assert_eq!(report["ready"], false);
+    assert_eq!(report["postgres_connected"], false);
+}