diff --git a/CHANGELOG.md b/CHANGELOG.md index a8314626..a6370ca3 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/awa-cli/src/health.rs b/awa-cli/src/health.rs new file mode 100644 index 00000000..34f446b2 --- /dev/null +++ b/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, + pub expected_schema_version: i32, + pub schema_compatible: bool, + pub storage_state: Option, + pub active_engine: Option, + /// 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") +} diff --git a/awa-cli/src/main.rs b/awa-cli/src/main.rs index cccca7d4..c57d5d5c 100644 --- a/awa-cli/src/main.rs +++ b/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> { 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> { .await?; match command { - Commands::Migrate { .. } | Commands::Serve { .. } | Commands::Callbacks { .. } => { + Commands::Migrate { .. } + | Commands::Serve { .. } + | Commands::Callbacks { .. } + | Commands::Health { .. } => { unreachable!() } diff --git a/awa-cli/tests/health_cli_test.rs b/awa-cli/tests/health_cli_test.rs new file mode 100644 index 00000000..7ea3a76e --- /dev/null +++ b/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); +} diff --git a/awa-worker/src/client.rs b/awa-worker/src/client.rs index 6975c4d8..ce99f32b 100644 --- a/awa-worker/src/client.rs +++ b/awa-worker/src/client.rs @@ -4,6 +4,7 @@ use crate::dispatcher::{ }; use crate::events::{BoxedUntypedEventHandler, JobEvent, UntypedJobEvent}; use crate::executor::{BoxedWorker, DlqPolicy, JobError, JobExecutor, JobResult, Worker}; +use crate::health_server::{self, HealthProbe}; use crate::heartbeat::HeartbeatService; use crate::maintenance::{MaintenanceService, RetentionPolicy}; use crate::runtime::{InFlightMap, InFlightRegistry}; @@ -21,8 +22,9 @@ use serde::de::DeserializeOwned; use sqlx::PgPool; use std::any::{Any, TypeId}; use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::Duration; use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinSet; @@ -57,6 +59,8 @@ pub enum BuildError { InvalidTerminalCountRollupInterval, #[error("invalid queue storage config: {0}")] InvalidQueueStorage(String), + #[error("invalid AWA_HEALTH_ADDR '{0}': expected a socket address like 0.0.0.0:8321")] + InvalidHealthAddr(String), } /// Health check result. @@ -163,6 +167,7 @@ pub struct ClientBuilder { storage: RuntimeStorage, transition_role: TransitionWorkerRole, storage_error: Option, + health_addr: Option, } impl ClientBuilder { @@ -220,6 +225,7 @@ impl ClientBuilder { storage, transition_role: TransitionWorkerRole::Auto, storage_error, + health_addr: None, } } @@ -941,6 +947,16 @@ impl ClientBuilder { } /// Build the client. + /// Serve `GET /healthz` and `GET /readyz` on this address while the + /// runtime is running (#368). Defaults to off; the `AWA_HEALTH_ADDR` + /// environment variable is honoured when this is not set. Port 0 binds + /// an ephemeral port — read it back with + /// [`Client::health_listener_addr`] after `start()`. + pub fn health_addr(mut self, addr: SocketAddr) -> Self { + self.health_addr = Some(addr); + self + } + pub fn build(self) -> Result { if self.queues.is_empty() { return Err(BuildError::NoQueuesConfigured); @@ -1054,6 +1070,18 @@ impl ClientBuilder { ); let dlq_policy = DlqPolicy::new(self.dlq_enabled_by_default, self.dlq_overrides); + let health_addr = match self.health_addr { + Some(addr) => Some(addr), + None => match std::env::var("AWA_HEALTH_ADDR") { + Ok(raw) if !raw.trim().is_empty() => Some( + raw.trim() + .parse::() + .map_err(|_| BuildError::InvalidHealthAddr(raw))?, + ), + _ => None, + }, + }; + Ok(Client { pool: self.pool, queues: self.queues, @@ -1107,6 +1135,8 @@ impl ClientBuilder { runtime_hostname: std::env::var("HOSTNAME").ok(), runtime_pid: std::process::id() as i32, runtime_version: env!("CARGO_PKG_VERSION"), + health_addr, + health_bound_addr: Arc::new(OnceLock::new()), }) } } @@ -1207,6 +1237,11 @@ pub struct Client { runtime_hostname: Option, runtime_pid: i32, runtime_version: &'static str, + /// Optional health/readiness listener address (#368). None = disabled. + health_addr: Option, + /// The address the health listener actually bound (set during `start()`; + /// differs from `health_addr` when port 0 requested an ephemeral port). + health_bound_addr: Arc>, } #[derive(Clone)] @@ -1574,6 +1609,34 @@ impl Client { service_handles.push(handle); } + // Health/readiness listener (#368). Bind failures abort startup — + // a misconfigured probe address must not fail silently. Uses + // service_cancel so /readyz keeps answering (503, shutting_down) + // through the graceful drain. + if let Some(addr) = self.health_addr { + let listener = tokio::net::TcpListener::bind(addr).await.map_err(|err| { + awa_model::AwaError::Validation(format!( + "failed to bind health listener on {addr}: {err}" + )) + })?; + let bound = listener.local_addr().map_err(|err| { + awa_model::AwaError::Validation(format!( + "failed to resolve health listener address: {err}" + )) + })?; + let _ = self.health_bound_addr.set(bound); + let probe = HealthProbe { + pool: self.pool.clone(), + dispatcher_alive: self.dispatcher_alive.clone(), + heartbeat_alive: self.heartbeat_alive.clone(), + maintenance_alive: self.maintenance_alive.clone(), + leader: self.leader.clone(), + dispatch_cancel: self.dispatch_cancel.clone(), + }; + let cancel = self.service_cancel.clone(); + service_handles.push(tokio::spawn(health_server::serve(listener, probe, cancel))); + } + // Start heartbeat service (uses service_cancel — stays alive during drain) let heartbeat = HeartbeatService::new( self.pool.clone(), @@ -2025,6 +2088,13 @@ impl Client { } /// Health check. + /// The address the health/readiness listener bound during `start()`. + /// `None` when the listener is disabled or `start()` has not run yet. + /// Useful with port 0 (ephemeral) configurations. + pub fn health_listener_addr(&self) -> Option { + self.health_bound_addr.get().copied() + } + pub async fn health_check(&self) -> HealthCheck { let postgres_connected = sqlx::query("SELECT 1").execute(&self.pool).await.is_ok(); let poll_loop_alive = self diff --git a/awa-worker/src/health_server.rs b/awa-worker/src/health_server.rs new file mode 100644 index 00000000..a9390aea --- /dev/null +++ b/awa-worker/src/health_server.rs @@ -0,0 +1,202 @@ +//! Opt-in health/readiness listener for the worker runtime (#368). +//! +//! Serves exactly two endpoints for infrastructure probes: +//! +//! - `GET /healthz` — process liveness. Returns `200` for as long as the +//! listener task is running, including during graceful drain: a draining +//! worker must not be killed by a liveness probe. +//! - `GET /readyz` — readiness. Returns `200` only while the runtime can do +//! useful work: database reachable, schema at least the binary's expected +//! version, every queue's claim loop ticking, heartbeat and maintenance +//! services alive, and not shutting down. Otherwise `503` with a JSON body +//! naming every failing check. +//! +//! Deliberately hand-rolled over `tokio::net::TcpListener`: two GET routes +//! do not justify pulling an HTTP stack into `awa-worker` (the admin UI keeps +//! its own axum server in `awa-ui`). Requests larger than a probe would ever +//! send are rejected, and every response closes the connection. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use serde::Serialize; +use sqlx::PgPool; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, info, warn}; + +/// Probe requests are tiny; anything beyond this is not a health check. +const MAX_REQUEST_BYTES: usize = 4096; +/// Bound the whole request/response exchange so a stuck peer cannot pin the +/// listener's accept loop resources. +const CONNECTION_DEADLINE: Duration = Duration::from_secs(5); + +/// Shared handles the readiness computation reads. Cloned from the runtime's +/// own liveness flags — the listener observes the same state the runtime +/// snapshot reporter publishes to `awa.runtime_instances`. +#[derive(Clone)] +pub(crate) struct HealthProbe { + pub(crate) pool: PgPool, + pub(crate) dispatcher_alive: Arc>>, + pub(crate) heartbeat_alive: Arc, + pub(crate) maintenance_alive: Arc, + pub(crate) leader: Arc, + pub(crate) dispatch_cancel: CancellationToken, +} + +/// The `/readyz` response body. Field names are part of the documented probe +/// surface (see `docs/deployment.md`). +#[derive(Debug, Serialize)] +pub(crate) struct ReadyzReport { + pub(crate) ready: bool, + pub(crate) postgres_connected: bool, + pub(crate) schema_version: Option, + pub(crate) expected_schema_version: i32, + pub(crate) schema_compatible: bool, + pub(crate) poll_loop_alive: bool, + pub(crate) heartbeat_alive: bool, + pub(crate) maintenance_alive: bool, + pub(crate) shutting_down: bool, + pub(crate) leader: bool, +} + +impl HealthProbe { + pub(crate) async fn readyz(&self) -> ReadyzReport { + let postgres_connected = sqlx::query("SELECT 1").execute(&self.pool).await.is_ok(); + + // A newer schema than the binary expects is the supported rolling- + // deploy skew (older binary against newer schema); an older schema + // means `awa migrate` has not run and the runtime cannot work. + let schema_version = if postgres_connected { + awa_model::migrations::current_version(&self.pool) + .await + .ok() + } else { + None + }; + let expected_schema_version = awa_model::migrations::CURRENT_VERSION; + let schema_compatible = + schema_version.is_some_and(|version| version >= expected_schema_version); + + let poll_loop_alive = self + .dispatcher_alive + .values() + .all(|alive| alive.load(Ordering::SeqCst)); + let heartbeat_alive = self.heartbeat_alive.load(Ordering::SeqCst); + let maintenance_alive = self.maintenance_alive.load(Ordering::SeqCst); + let shutting_down = self.dispatch_cancel.is_cancelled(); + + ReadyzReport { + ready: postgres_connected + && schema_compatible + && poll_loop_alive + && heartbeat_alive + && maintenance_alive + && !shutting_down, + postgres_connected, + schema_version, + expected_schema_version, + schema_compatible, + poll_loop_alive, + heartbeat_alive, + maintenance_alive, + shutting_down, + leader: self.leader.load(Ordering::SeqCst), + } + } +} + +/// Accept loop. Runs until `cancel` fires; each connection is handled on its +/// own task with a hard deadline. +pub(crate) async fn serve(listener: TcpListener, probe: HealthProbe, cancel: CancellationToken) { + let addr = listener.local_addr().ok(); + info!(?addr, "Health listener started (/healthz, /readyz)"); + loop { + tokio::select! { + () = cancel.cancelled() => { + info!(?addr, "Health listener stopped"); + return; + } + accepted = listener.accept() => { + match accepted { + Ok((stream, peer)) => { + let probe = probe.clone(); + tokio::spawn(async move { + let outcome = tokio::time::timeout( + CONNECTION_DEADLINE, + handle_connection(stream, probe), + ) + .await; + if let Ok(Err(err)) = outcome { + debug!(%peer, error = %err, "Health probe connection error"); + } + }); + } + Err(err) => { + warn!(error = %err, "Health listener accept failed"); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + } + } + } +} + +async fn handle_connection(mut stream: TcpStream, probe: HealthProbe) -> std::io::Result<()> { + let mut buf = Vec::with_capacity(512); + let mut chunk = [0u8; 512]; + let request_line = loop { + let read = stream.read(&mut chunk).await?; + if read == 0 { + return Ok(()); + } + buf.extend_from_slice(&chunk[..read]); + if let Some(end) = buf.iter().position(|&b| b == b'\n') { + break String::from_utf8_lossy(&buf[..end]).trim_end().to_string(); + } + if buf.len() > MAX_REQUEST_BYTES { + return write_response(&mut stream, 431, "{\"error\":\"request too large\"}").await; + } + }; + + let mut parts = request_line.split_whitespace(); + let method = parts.next().unwrap_or_default(); + let path = parts.next().unwrap_or_default(); + let path = path.split('?').next().unwrap_or_default(); + + if method != "GET" { + return write_response(&mut stream, 405, "{\"error\":\"method not allowed\"}").await; + } + + match path { + "/healthz" => write_response(&mut stream, 200, "{\"status\":\"ok\"}").await, + "/readyz" => { + let report = probe.readyz().await; + let status = if report.ready { 200 } else { 503 }; + let body = + serde_json::to_string(&report).unwrap_or_else(|_| "{\"ready\":false}".to_string()); + write_response(&mut stream, status, &body).await + } + _ => write_response(&mut stream, 404, "{\"error\":\"not found\"}").await, + } +} + +async fn write_response(stream: &mut TcpStream, status: u16, body: &str) -> std::io::Result<()> { + let reason = match status { + 200 => "OK", + 404 => "Not Found", + 405 => "Method Not Allowed", + 431 => "Request Header Fields Too Large", + 503 => "Service Unavailable", + _ => "Unknown", + }; + let response = format!( + "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len(), + ); + stream.write_all(response.as_bytes()).await?; + stream.shutdown().await +} diff --git a/awa-worker/src/lib.rs b/awa-worker/src/lib.rs index 4002bc96..d5618987 100644 --- a/awa-worker/src/lib.rs +++ b/awa-worker/src/lib.rs @@ -6,6 +6,7 @@ pub mod dispatcher; mod enqueue_specs; pub mod events; pub mod executor; +mod health_server; pub mod heartbeat; #[cfg(feature = "http-worker")] pub mod http_worker; diff --git a/awa/tests/health_endpoint_test.rs b/awa/tests/health_endpoint_test.rs new file mode 100644 index 00000000..d31064a5 --- /dev/null +++ b/awa/tests/health_endpoint_test.rs @@ -0,0 +1,376 @@ +//! Worker health/readiness listener tests (#368). +//! +//! Uses a dedicated `awa_health_test` database: these tests delete +//! `schema_version` rows and refuse connections to fault-inject, which must +//! not race the shared integration database. For the same reason the tests +//! serialize against each other — a process-local mutex plus a Postgres +//! advisory lock (so process-per-test runners stay safe too). +//! +//! Set DATABASE_URL=postgres://postgres:test@localhost:15432/awa_test + +use awa::{Client, JobArgs, JobResult, QueueConfig}; +use awa_model::{insert_with, migrations, InsertOpts}; +use serde::{Deserialize, Serialize}; +use sqlx::postgres::{PgConnection, PgPoolOptions}; +use sqlx::{Connection, PgPool}; +use std::net::SocketAddr; +use std::sync::OnceLock; +use std::time::Duration; +use tokio::sync::Mutex; + +static TEST_MUTEX: OnceLock> = OnceLock::new(); +const HEALTH_TEST_LOCK_KEY: i64 = 0x6177_6168_6c74_6831; // "awahlth1" + +fn test_mutex() -> &'static Mutex<()> { + TEST_MUTEX.get_or_init(|| Mutex::new(())) +} + +struct HealthTestGuard { + _local: tokio::sync::MutexGuard<'static, ()>, + _conn: PgConnection, +} + +async fn acquire_health_guard() -> HealthTestGuard { + let local = test_mutex().lock().await; + ensure_health_database().await; + let mut conn = PgConnection::connect(&health_database_url()) + .await + .expect("Failed to open health test lock connection"); + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(HEALTH_TEST_LOCK_KEY) + .execute(&mut conn) + .await + .expect("Failed to acquire health test advisory lock"); + HealthTestGuard { + _local: local, + _conn: conn, + } +} + +fn database_url() -> String { + std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()) +} + +fn replace_database_name(url: &str, db_name: &str) -> String { + let (base, query) = match url.split_once('?') { + Some((base, query)) => (base, Some(query)), + None => (url, None), + }; + let (prefix, _old_db) = base + .rsplit_once('/') + .expect("DATABASE_URL must include a database name"); + let mut out = format!("{prefix}/{db_name}"); + if let Some(query) = query { + out.push('?'); + out.push_str(query); + } + out +} + +fn health_database_url() -> String { + replace_database_name(&database_url(), "awa_health_test") +} + +async fn ensure_health_database() { + let admin_url = replace_database_name(&database_url(), "postgres"); + let mut admin = PgConnection::connect(&admin_url) + .await + .expect("Failed to connect to admin database"); + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = 'awa_health_test')", + ) + .fetch_one(&mut admin) + .await + .expect("Failed to check health database existence"); + if !exists { + // Tolerate the duplicate-database race: another test process can + // create it between the existence check and this statement. + if let Err(err) = sqlx::raw_sql("CREATE DATABASE awa_health_test") + .execute(&mut admin) + .await + { + let duplicate = matches!( + &err, + sqlx::Error::Database(db_err) if db_err.code().as_deref() == Some("42P04") + ); + assert!(duplicate, "Failed to create health test database: {err}"); + } + } +} + +async fn setup() -> PgPool { + let pool = PgPoolOptions::new() + .max_connections(10) + .acquire_timeout(Duration::from_secs(5)) + .connect(&health_database_url()) + .await + .expect("Failed to connect to health test database — is Postgres running?"); + migrations::run(&pool).await.expect("Failed to migrate"); + pool +} + +#[derive(Debug, Serialize, Deserialize, JobArgs)] +struct SlowJob { + pub sleep_ms: u64, +} + +fn ephemeral_addr() -> SocketAddr { + "127.0.0.1:0".parse().unwrap() +} + +async fn build_and_start(pool: &PgPool, queue: &str) -> Client { + let client = Client::builder(pool.clone()) + .queue(queue, QueueConfig::default()) + .register::(|args: SlowJob, _ctx| async move { + tokio::time::sleep(Duration::from_millis(args.sleep_ms)).await; + Ok(JobResult::Completed) + }) + .health_addr(ephemeral_addr()) + .build() + .expect("client should build"); + client.start().await.expect("client should start"); + client +} + +async fn get(addr: SocketAddr, path: &str) -> (u16, serde_json::Value) { + let response = reqwest::get(format!("http://{addr}{path}")) + .await + .expect("health request should connect"); + let status = response.status().as_u16(); + let body: serde_json::Value = response.json().await.expect("health body should be JSON"); + (status, body) +} + +/// Poll /readyz until it reports the expected `ready` value (dispatchers and +/// services flip their liveness flags asynchronously after start()). +async fn wait_for_ready(addr: SocketAddr, expected: bool) -> serde_json::Value { + let deadline = tokio::time::Instant::now() + Duration::from_secs(15); + loop { + let (status, body) = get(addr, "/readyz").await; + let ready = body["ready"].as_bool().unwrap_or(false); + if ready == expected { + assert_eq!(status, if expected { 200 } else { 503 }); + return body; + } + if tokio::time::Instant::now() > deadline { + panic!("timed out waiting for ready={expected}; last body: {body}"); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +#[tokio::test] +async fn test_healthz_and_readyz_report_running_worker() { + let _guard = acquire_health_guard().await; + let pool = setup().await; + let client = build_and_start(&pool, "health_baseline").await; + let addr = client + .health_listener_addr() + .expect("listener address should be bound after start()"); + + let body = wait_for_ready(addr, true).await; + assert_eq!(body["postgres_connected"], true); + assert_eq!(body["schema_compatible"], true); + assert_eq!(body["poll_loop_alive"], true); + assert_eq!(body["heartbeat_alive"], true); + assert_eq!(body["maintenance_alive"], true); + assert_eq!(body["shutting_down"], false); + assert_eq!(body["expected_schema_version"], migrations::CURRENT_VERSION); + + let (status, body) = get(addr, "/healthz").await; + assert_eq!(status, 200); + assert_eq!(body["status"], "ok"); + + // Unknown path and non-GET are rejected, not silently 200. + let (status, _) = get(addr, "/nope").await; + assert_eq!(status, 404); + let response = reqwest::Client::new() + .post(format!("http://{addr}/readyz")) + .send() + .await + .expect("POST should connect"); + assert_eq!(response.status().as_u16(), 405); + + client.shutdown(Duration::from_secs(10)).await; +} + +#[tokio::test] +async fn test_readyz_flips_503_on_schema_mismatch() { + let _guard = acquire_health_guard().await; + let pool = setup().await; + let client = build_and_start(&pool, "health_schema").await; + let addr = client.health_listener_addr().expect("listener address"); + wait_for_ready(addr, true).await; + + // Simulate a binary that is newer than the schema: remove the newest + // schema_version row so the DB reports an older version. + sqlx::query("DELETE FROM awa.schema_version WHERE version = $1") + .bind(migrations::CURRENT_VERSION) + .execute(&pool) + .await + .unwrap(); + + let body = wait_for_ready(addr, false).await; + assert_eq!(body["schema_compatible"], false); + assert_eq!(body["postgres_connected"], true); + + sqlx::query("INSERT INTO awa.schema_version (version, description) VALUES ($1, 'restored by health test')") + .bind(migrations::CURRENT_VERSION) + .execute(&pool) + .await + .unwrap(); + wait_for_ready(addr, true).await; + + client.shutdown(Duration::from_secs(10)).await; +} + +#[tokio::test] +async fn test_readyz_reports_shutting_down_during_drain() { + let _guard = acquire_health_guard().await; + let pool = setup().await; + let queue = "health_drain"; + sqlx::query("DELETE FROM awa.jobs WHERE queue = $1") + .bind(queue) + .execute(&pool) + .await + .unwrap(); + + // The worker runs in-process, so the handler can signal the moment its + // job starts executing — the compat view is not a reliable way to + // observe `running` under queue storage. + let started = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let started_flag = started.clone(); + let client = Client::builder(pool.clone()) + .queue(queue, QueueConfig::default()) + .register::(move |args: SlowJob, _ctx| { + let started = started_flag.clone(); + async move { + started.store(true, std::sync::atomic::Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(args.sleep_ms)).await; + Ok(JobResult::Completed) + } + }) + .health_addr(ephemeral_addr()) + .build() + .expect("client should build"); + client.start().await.expect("client should start"); + let addr = client.health_listener_addr().expect("listener address"); + wait_for_ready(addr, true).await; + + // Park a slow job so the drain phase stays open long enough to probe. + insert_with( + &pool, + &SlowJob { sleep_ms: 4000 }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await + .unwrap(); + + // Wait until the job is executing before starting the shutdown. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while !started.load(std::sync::atomic::Ordering::SeqCst) { + if tokio::time::Instant::now() > deadline { + panic!("slow job was never claimed"); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + let shutdown = { + let client = std::sync::Arc::new(client); + let handle = client.clone(); + tokio::spawn(async move { handle.shutdown(Duration::from_secs(15)).await }); + client + }; + + // While draining, readiness must report shutting_down and 503 — + // infrastructure should stop routing to this instance, but the + // liveness endpoint must keep answering 200 so it is not killed. + let body = wait_for_ready(addr, false).await; + assert_eq!(body["shutting_down"], true); + let (status, _) = get(addr, "/healthz").await; + assert_eq!(status, 200); + + // After the drain completes the listener goes away with the runtime. + let deadline = tokio::time::Instant::now() + Duration::from_secs(15); + loop { + if reqwest::get(format!("http://{addr}/healthz")) + .await + .is_err() + { + break; + } + if tokio::time::Instant::now() > deadline { + panic!("health listener still answering after shutdown"); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + drop(shutdown); +} + +#[tokio::test] +async fn test_readyz_flips_503_when_database_unreachable() { + let _guard = acquire_health_guard().await; + let pool = setup().await; + let client = build_and_start(&pool, "health_db_down").await; + let addr = client.health_listener_addr().expect("listener address"); + wait_for_ready(addr, true).await; + + // Container-free stand-in for a database outage: refuse new connections + // to the test database and kill every existing backend. Pooled + // connections die, reconnects are refused, so the probe's acquire fails. + // (Closing the pool instead would deadlock — the runtime's LISTEN + // connection never checks back in.) + let admin_url = replace_database_name(&database_url(), "postgres"); + let mut admin = PgConnection::connect(&admin_url) + .await + .expect("admin connection for outage injection"); + sqlx::raw_sql("ALTER DATABASE awa_health_test WITH ALLOW_CONNECTIONS false") + .execute(&mut admin) + .await + .expect("connection refusal should apply"); + sqlx::raw_sql( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity \ + WHERE datname = 'awa_health_test' AND pid <> pg_backend_pid()", + ) + .execute(&mut admin) + .await + .expect("backend termination should apply"); + + let body = wait_for_ready(addr, false).await; + assert_eq!(body["postgres_connected"], false); + + // Restore the database for the remaining tests, then let the runtime + // recover before shutting down cleanly. + sqlx::raw_sql("ALTER DATABASE awa_health_test WITH ALLOW_CONNECTIONS true") + .execute(&mut admin) + .await + .expect("connection restore should apply"); + wait_for_ready(addr, true).await; + client.shutdown(Duration::from_secs(10)).await; +} + +#[tokio::test] +async fn test_invalid_health_addr_env_is_a_build_error() { + // Env mutation is process-global; this is the only test that touches + // AWA_HEALTH_ADDR, and no other test reads it (they configure the + // builder explicitly). + let _guard = acquire_health_guard().await; + let pool = setup().await; + std::env::set_var("AWA_HEALTH_ADDR", "not-an-address"); + let result = Client::builder(pool.clone()) + .queue("health_env", QueueConfig::default()) + .register::(|_args: SlowJob, _ctx| async move { Ok(JobResult::Completed) }) + .build(); + std::env::remove_var("AWA_HEALTH_ADDR"); + let err = result + .err() + .expect("invalid AWA_HEALTH_ADDR must not build"); + assert!( + err.to_string().contains("AWA_HEALTH_ADDR"), + "unexpected error: {err}" + ); +} diff --git a/docs/configuration.md b/docs/configuration.md index 0c239af0..2ff74f10 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -610,6 +610,20 @@ For very high-throughput no-op queues, size `max_workers` for durable completion The terminal write path is already narrow for running/waiting jobs: successful receipt completions use compact `receipt_completion_batches_*` rows, and other retained terminal facts use `done_entries_*` without duplicating immutable ready body fields when the ready row is still retained. Both surfaces hydrate through `terminal_jobs`, so the default completion settings stay conservative without giving up durable terminal history. +## Worker health listener + +Set `AWA_HEALTH_ADDR` (or `ClientBuilder::health_addr`) to serve `GET /healthz` and +`GET /readyz` from the worker runtime for infrastructure probes: + +```bash +AWA_HEALTH_ADDR=0.0.0.0:8321 +``` + +Port `0` binds an ephemeral port (read it back with `Client::health_listener_addr()`). +Unset means no listener. Endpoint semantics, response fields, and Kubernetes probe +examples live in [deployment.md](deployment.md#health-checks); `awa health` covers +probe-less environments from the CLI. + ## Dead Letter Queue The DLQ is the **separate, durable hold-table for jobs that exhausted retries or hit a non-retryable terminal failure**. Without it, terminal failures live in ordinary queue-storage `done_entries_*` partitions and are reclaimed when queue-ring prune can safely truncate their segment. With DLQ enabled, those rows land in `dlq_entries`, are visible to the admin UI / API as a discrete backlog, and can be retried or purged by an operator. diff --git a/docs/deployment.md b/docs/deployment.md index 99bf4b87..3778caa7 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -167,25 +167,78 @@ then `40` seconds is a reasonable Kubernetes grace period. ## Health Checks -Awa provides a runtime health API in-process: +Workers ship an opt-in HTTP health listener ([#368](https://github.com/hardbyte/awa/issues/368)). +Enable it with the `AWA_HEALTH_ADDR` environment variable or the builder: + +```rust +let client = Client::builder(pool) + .queue("email", QueueConfig::default()) + .health_addr("0.0.0.0:8321".parse()?) // or AWA_HEALTH_ADDR=0.0.0.0:8321 + .build()?; +``` + +Two endpoints: + +- `GET /healthz` — process liveness. `200` for as long as the runtime is up, + **including during graceful drain** — a draining worker must not be killed + by its liveness probe. +- `GET /readyz` — readiness. `200` only when the worker can do useful work: + Postgres reachable, schema at least the binary's expected version, every + queue's claim loop ticking, heartbeat + maintenance services alive, and not + shutting down. Otherwise `503` with a JSON body naming each failing check + (`postgres_connected`, `schema_version`, `expected_schema_version`, + `schema_compatible`, `poll_loop_alive`, `heartbeat_alive`, + `maintenance_alive`, `shutting_down`, `leader`). + +Kubernetes probes against the listener: + +```yaml +spec: + template: + spec: + containers: + - name: worker + env: + - name: AWA_HEALTH_ADDR + value: "0.0.0.0:8321" + ports: + - name: health + containerPort: 8321 + livenessProbe: + httpGet: { path: /healthz, port: health } + periodSeconds: 10 + readinessProbe: + httpGet: { path: /readyz, port: health } + periodSeconds: 5 +``` + +The listener is deliberately minimal (no TLS, no auth): bind it to the pod IP +or loopback and keep it off public ingress. + +The in-process API remains available without the listener: - Rust: `client.health_check().await` - Python: `await client.health_check()` -It does not provide worker HTTP endpoints by itself. Your application should expose `/healthz` and `/readyz` if your platform expects HTTP probes. +### `awa health` — probe-less environments -Awa's current `health_check().healthy` result requires: +For cron drivers, systemd `ExecStartPre`, or CI smoke checks, `awa health` +answers the cluster-level question from the database alone — reachable, +schema migrated for this binary, and fleet heartbeat counts: -- Postgres reachable -- all configured dispatchers alive -- heartbeat loop alive -- maintenance loop alive -- not currently shutting down - -Recommended worker probes: +```console +$ awa health --database-url "$DATABASE_URL" +ready: yes +postgres: connected +schema version: 39 (binary expects >= 39) +storage: active (queue_storage) +live runtimes: 3 (0 unhealthy) +``` -- liveness: same conditions as `health_check().healthy` -- readiness: not shutting down +`--json` emits the same report as a machine-readable object. Exit code `0` +means ready; `1` means unreachable or unmigrated. Fleet numbers are +informational — an idle cluster with zero workers is still "ready" for +producers and migrations. `awa serve` is separate; its HTTP server is for the web UI/admin API, not for worker liveness.