diff --git a/.gitignore b/.gitignore index cad92250..82e53780 100644 --- a/.gitignore +++ b/.gitignore @@ -23,7 +23,7 @@ proptest-regressions/ # Nix result result-* -/generated +generated # direnv .direnv/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e90ce46..60f8fb8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,11 +3,14 @@ ### Added - Gateway: per-relay `blockfrost_gateway_relay_healthy`, `blockfrost_gateway_relay_data_node_up`, and `blockfrost_gateway_relay_info` metrics in `GET /metrics` (and the same data points in `GET /stats`) +- Gateway: `blockfrost_gateway_healthy` metric in `GET /metrics`, mirroring the `healthy` field of `GET /` - New endpoints proxied to the data node: `/accounts/{stake_address}/utxos`, `/addresses/{address}`, and `/blocks/slot/{slot_number}` - `--max-response-body-bytes` to configure the maximum proxied response body size (default 10 MiB) ### Fixed +- Gateway: `GET /` no longer always reports `healthy: true` — it now reflects periodic health checks of PostgreSQL connectivity and the Blockfrost API +- Gateway: refuses to start when the initial health check fails (e.g. an invalid Blockfrost project token) - Raised the proxied body limit from 1 MiB to 10 MiB - Restored `local_address` bind for unspecified IPs like `0.0.0.0`, fixing an IPv6 regression that broke the IPv4-forcing behavior of `--server-address 0.0.0.0` - Inverted the metrics answer in the `--init` config prompt diff --git a/Cargo.lock b/Cargo.lock index 91d73f3e..188a918a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4855,6 +4855,7 @@ version = "0.3.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e" dependencies = [ + "chrono", "matchers", "nu-ansi-term", "once_cell", diff --git a/Cargo.toml b/Cargo.toml index d1e678f5..d256bc30 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -103,7 +103,11 @@ tower = { version = "0.5.3", features = ["limit"] } tower-http = { version = "0.6.8", features = ["normalize-path"] } tracing = "0.1.44" tracing-journald = "0.3.2" -tracing-subscriber = { version = "0.3.22", features = ["env-filter", "fmt"] } +tracing-subscriber = { version = "0.3.22", features = [ + "env-filter", + "fmt", + "chrono", +] } tungstenite = "0.28.0" twelf = { version = "0.15.0", features = ["clap", "toml"] } url = { version = "2", features = ["serde"] } diff --git a/crates/common/src/tracing.rs b/crates/common/src/tracing.rs index 0f7ccf35..68f1f32e 100644 --- a/crates/common/src/tracing.rs +++ b/crates/common/src/tracing.rs @@ -153,10 +153,14 @@ pub fn setup_tracing(log_level: Level, log_target_env: &str) { } fn setup_default(log_level: Level) { + // Human-readable local time instead of the default + let timer = tracing_subscriber::fmt::time::ChronoLocal::new("%Y-%m-%d %H:%M:%S%.3f".to_owned()); + tracing_subscriber::fmt() .with_max_level(log_level) .event_format( Format::default() + .with_timer(timer) .with_ansi(true) .with_level(true) .with_target(true) diff --git a/crates/gateway/CHANGELOG.md b/crates/gateway/CHANGELOG.md index ad83fac7..cfb851ff 100644 --- a/crates/gateway/CHANGELOG.md +++ b/crates/gateway/CHANGELOG.md @@ -3,6 +3,7 @@ ### Added - Per-relay `blockfrost_gateway_relay_healthy`, `blockfrost_gateway_relay_data_node_up`, and `blockfrost_gateway_relay_info` metrics in `GET /metrics` (and the same data points in `GET /stats`) +- `blockfrost_gateway_healthy` metric in `GET /metrics`, mirroring the `healthy` field of `GET /` - Prometheus metrics endpoint `GET /metrics` exposing per-relay stats (connection status, WebSocket RTT, connected-since timestamp, request/response counters) and PostgreSQL connection-pool gauges (max size, open, available, waiting) - Prometheus counter `blockfrost_gateway_http_requests_total` with `method`, `route`, and `status_code` labels for Gateway API requests - `blockfrost_gateway_build_info` metric exposing the Gateway version and git revision @@ -12,6 +13,8 @@ ### Fixed +- `GET /` no longer always reports `healthy: true` — it now reflects periodic health checks of PostgreSQL connectivity and the Blockfrost API +- The Gateway now refuses to start when the initial health check fails (e.g. an invalid Blockfrost project token) - The underlying Blockfrost API error (e.g. rate limiting) is now logged when the license NFT check fails during registration ### Removed diff --git a/crates/gateway/src/api/metrics.rs b/crates/gateway/src/api/metrics.rs index c4201faa..da27644c 100644 --- a/crates/gateway/src/api/metrics.rs +++ b/crates/gateway/src/api/metrics.rs @@ -1,4 +1,5 @@ use crate::db::{DB, PoolStatus}; +use crate::health_monitor::HealthMonitor; use crate::load_balancer::LoadBalancerState; use axum::{Extension, http::StatusCode, response::IntoResponse}; use metrics::{describe_counter, describe_gauge, gauge}; @@ -122,6 +123,7 @@ const RELAY_METRICS: &[RelayMetric] = &[ pub(crate) async fn render_prometheus( load_balancer: &LoadBalancerState, db_pool: &PoolStatus, + healthy: bool, ) -> Result { let now_chrono = chrono::Utc::now(); let now_instant = std::time::Instant::now(); @@ -160,6 +162,13 @@ pub(crate) async fn render_prometheus( let mut out = String::new(); + writeln!( + out, + "# HELP blockfrost_gateway_healthy Whether the Gateway is healthy, i.e. its periodic health checks (PostgreSQL connectivity, Blockfrost API) succeeded. Mirrors the `healthy` field of `GET /`." + )?; + writeln!(out, "# TYPE blockfrost_gateway_healthy gauge")?; + writeln!(out, "blockfrost_gateway_healthy {}", u8::from(healthy))?; + writeln!( out, "# HELP blockfrost_gateway_connected_relays Number of relays currently connected via WebSocket." @@ -231,11 +240,13 @@ pub(crate) async fn render_prometheus( pub async fn route( Extension(load_balancer): Extension, Extension(db): Extension, + Extension(health_monitor): Extension, Extension(prometheus_handle): Extension, ) -> Result { + let healthy = health_monitor.current_status().await.healthy; let mut body = prometheus_handle.render(); body.push_str( - &render_prometheus(&load_balancer, &db.pool_status()) + &render_prometheus(&load_balancer, &db.pool_status(), healthy) .await .map_err(|e| { error!("failed to render gateway metrics: {e}"); @@ -304,10 +315,12 @@ mod tests { }); lb.active_relays.lock().await.insert(uuid, relay); - let out = render_prometheus(&lb, &test_pool_status()) + let out = render_prometheus(&lb, &test_pool_status(), true) .await .expect("render metrics"); + assert!(out.contains("# TYPE blockfrost_gateway_healthy gauge")); + assert!(out.contains("\nblockfrost_gateway_healthy 1\n")); assert!(out.contains("# TYPE blockfrost_gateway_connected_relays gauge")); assert!(out.contains("\nblockfrost_gateway_connected_relays 1\n")); assert!(out.contains("# TYPE blockfrost_gateway_db_pool_max_size gauge")); @@ -348,7 +361,7 @@ mod tests { .await .insert(uuid, test_relay_state("Icebreaker2")); - let out = render_prometheus(&lb, &test_pool_status()) + let out = render_prometheus(&lb, &test_pool_status(), true) .await .expect("render metrics"); @@ -370,12 +383,21 @@ mod tests { #[tokio::test] async fn reports_zero_when_no_relays() { let lb = LoadBalancerState::new(None, test_key()); - let out = render_prometheus(&lb, &test_pool_status()) + let out = render_prometheus(&lb, &test_pool_status(), true) .await .expect("render metrics"); assert!(out.contains("\nblockfrost_gateway_connected_relays 0\n")); } + #[tokio::test] + async fn reports_unhealthy_gateway() { + let lb = LoadBalancerState::new(None, test_key()); + let out = render_prometheus(&lb, &test_pool_status(), false) + .await + .expect("render metrics"); + assert!(out.contains("\nblockfrost_gateway_healthy 0\n")); + } + #[tokio::test] async fn escapes_label_values() { let lb = LoadBalancerState::new(None, test_key()); @@ -385,7 +407,7 @@ mod tests { .await .insert(uuid, test_relay_state("we\"ird\\name")); - let out = render_prometheus(&lb, &test_pool_status()) + let out = render_prometheus(&lb, &test_pool_status(), true) .await .expect("render metrics"); assert!(out.contains("relay=\"we\\\"ird\\\\name\"")); diff --git a/crates/gateway/src/api/root.rs b/crates/gateway/src/api/root.rs index 679ceff0..c30a0631 100644 --- a/crates/gateway/src/api/root.rs +++ b/crates/gateway/src/api/root.rs @@ -1,5 +1,6 @@ use crate::config::Config; -use axum::{Extension, Json}; +use crate::health_monitor::HealthMonitor; +use axum::{Extension, Json, http::StatusCode, response::IntoResponse}; use serde::Serialize; #[derive(Serialize)] @@ -8,15 +9,102 @@ pub struct Response { pub version: String, pub healthy: bool, pub commit: &'static str, + pub errors: Vec<&'static str>, } -pub async fn route(Extension(config): Extension) -> Json { +pub async fn route( + Extension(config): Extension, + Extension(health_monitor): Extension, +) -> impl IntoResponse { + let status = health_monitor.current_status().await; + + let http_status = if status.healthy { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + }; + let response = Response { url: config.server.url.clone(), version: env!("CARGO_PKG_VERSION").to_string(), commit: env!("GIT_REVISION"), - healthy: true, + healthy: status.healthy, + errors: status.errors.iter().map(|e| e.code).collect(), }; - Json(response) + (http_status, Json(response)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::health_monitor::{HealthError, HealthStatus}; + use crate::types::Network; + + fn test_config() -> Config { + Config { + server: crate::config::Server { + address: "127.0.0.1:0".to_string(), + log_level: tracing::Level::INFO, + network: Network::Preview, + url: Some(url::Url::parse("https://gateway.example.com").unwrap()), + peer_urls: vec![], + peer_secret: [7; 32], + }, + database: crate::config::Db { + connection_string: "postgresql://user:pass@localhost/db".to_string(), + pool_max_size: std::num::NonZeroUsize::new(8).unwrap(), + }, + blockfrost: crate::config::Blockfrost { + project_id: "previewXXX".to_string(), + nft_asset: "asset".to_string(), + }, + hydra_platform: None, + hydra_bridge: None, + } + } + + async fn get_root(status: HealthStatus) -> (StatusCode, serde_json::Value) { + let monitor = HealthMonitor::new_static(status); + let response = route(Extension(test_config()), Extension(monitor)) + .await + .into_response(); + let http_status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("read body"); + (http_status, serde_json::from_slice(&body).expect("JSON")) + } + + #[tokio::test] + async fn healthy_returns_200() { + let (status, json) = get_root(HealthStatus { + healthy: true, + errors: vec![], + }) + .await; + + assert_eq!(status, StatusCode::OK); + assert_eq!(json["healthy"], serde_json::json!(true)); + assert_eq!(json["errors"], serde_json::json!([])); + } + + #[tokio::test] + async fn unhealthy_returns_503_with_error_codes_only() { + let detail = "database pool error: connection to internal-db-host refused"; + let (status, json) = get_root(HealthStatus { + healthy: false, + errors: vec![HealthError { + code: "database_unreachable", + detail: detail.to_string(), + }], + }) + .await; + + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(json["healthy"], serde_json::json!(false)); + assert_eq!(json["errors"], serde_json::json!(["database_unreachable"])); + // The detailed reason must not leak into the public response. + assert!(!json.to_string().contains("internal-db-host")); + } } diff --git a/crates/gateway/src/blockfrost.rs b/crates/gateway/src/blockfrost.rs index 75ea6316..548c959f 100644 --- a/crates/gateway/src/blockfrost.rs +++ b/crates/gateway/src/blockfrost.rs @@ -23,6 +23,35 @@ impl BlockfrostAPI { } } + /// Verifies Blockfrost API connectivity + pub async fn ping(&self, timeout: std::time::Duration) -> Result<(), String> { + if cfg!(feature = "dev_mock_db") { + return Ok(()); + } + + tokio::time::timeout(timeout, self.api.blocks_latest()) + .await + .map_err(|_| { + format!( + "Blockfrost API health check timed out after {}s", + timeout.as_secs() + ) + })? + .map(|_| ()) + .map_err(|e| { + // The SDK’s error messages span multiple lines; flatten them + // so that they log (and serialize) as a single line. + let flat = e + .to_string() + .lines() + .map(str::trim) + .filter(|l| !l.is_empty()) + .collect::>() + .join(", "); + format!("Blockfrost API error: {flat}") + }) + } + // Parse asset from the unit async fn parse_asset(&self, unit: &str) -> Result { if unit.len() < self.policy_id_size { diff --git a/crates/gateway/src/db.rs b/crates/gateway/src/db.rs index 87c81a98..fe3e6f5f 100644 --- a/crates/gateway/src/db.rs +++ b/crates/gateway/src/db.rs @@ -46,6 +46,36 @@ impl DB { Self { pool } } + /// Verifies database connectivity by running a trivial query through the connection pool + pub async fn ping(&self, timeout: std::time::Duration) -> Result<(), String> { + if cfg!(feature = "dev_mock_db") { + return Ok(()); + } + + let check = async { + let connection = self + .pool + .get() + .await + .map_err(|e| format!("database pool error: {e}"))?; + connection + .interact(|c| diesel::sql_query("SELECT 1").execute(c)) + .await + .map_err(|e| format!("database interaction error: {e}"))? + .map_err(|e| format!("database query error: {e}"))?; + Ok(()) + }; + + tokio::time::timeout(timeout, check) + .await + .unwrap_or_else(|_| { + Err(format!( + "database health check timed out after {}s (pool exhausted or database unreachable)", + timeout.as_secs() + )) + }) + } + pub fn pool_status(&self) -> PoolStatus { let status = self.pool.status(); PoolStatus { diff --git a/crates/gateway/src/health_monitor.rs b/crates/gateway/src/health_monitor.rs new file mode 100644 index 00000000..900aabdc --- /dev/null +++ b/crates/gateway/src/health_monitor.rs @@ -0,0 +1,129 @@ +use crate::blockfrost::BlockfrostAPI; +use crate::db::DB; +use std::sync::Arc; +use tokio::sync::{Mutex, Notify}; +use tokio::time::{self, Duration, Instant}; + +#[derive(Clone)] +pub struct HealthError { + /// Stable machine-readable code, safe to expose in `GET /`. + pub code: &'static str, + /// Detailed reason; only logged server-side, never exposed over HTTP. + pub detail: String, +} + +#[derive(Clone)] +pub struct HealthStatus { + pub healthy: bool, + pub errors: Vec, +} + +impl HealthStatus { + pub fn details(&self) -> String { + self.errors + .iter() + .map(|e| e.detail.as_str()) + .collect::>() + .join("; ") + } +} + +#[derive(Clone)] +pub struct HealthMonitor { + status: Arc>, +} + +impl HealthMonitor { + const DB_PING_TIMEOUT: Duration = Duration::from_secs(5); + const BLOCKFROST_PING_TIMEOUT: Duration = Duration::from_secs(5); + /// The Blockfrost check runs on its own, slower cadence to keep the + /// request-quota usage of the health checks negligible. + const BLOCKFROST_CHECK_INTERVAL: Duration = Duration::from_secs(60); + const HEALTHY_CHECK_INTERVAL: Duration = Duration::from_secs(10); + const UNHEALTHY_CHECK_INTERVAL: Duration = Duration::from_secs(2); + + pub async fn current_status(&self) -> HealthStatus { + self.status.lock().await.clone() + } + + pub fn new_static(status: HealthStatus) -> Self { + Self { + status: Arc::new(Mutex::new(status)), + } + } + + pub async fn spawn(db: DB, blockfrost_api: BlockfrostAPI) -> Self { + let self_ = Self::new_static(HealthStatus { + healthy: true, + errors: vec![], + }); + let status = self_.status.clone(); + + let first_check_done = Arc::new(Notify::new()); + let first_check_done_ = first_check_done.clone(); + + tokio::spawn(async move { + // `None` until the first check completes, so that we only log + // actual changes (startup failures are handled by `main`). + let mut previous_codes: Option> = None; + let mut blockfrost_error: Option = None; + let mut last_blockfrost_check: Option = None; + loop { + let db_error = db.ping(Self::DB_PING_TIMEOUT).await.err(); + + if last_blockfrost_check + .is_none_or(|at| at.elapsed() >= Self::BLOCKFROST_CHECK_INTERVAL) + { + blockfrost_error = blockfrost_api + .ping(Self::BLOCKFROST_PING_TIMEOUT) + .await + .err(); + last_blockfrost_check = Some(Instant::now()); + } + + let errors: Vec = [ + db_error.map(|detail| HealthError { + code: "database_unreachable", + detail, + }), + blockfrost_error.clone().map(|detail| HealthError { + code: "blockfrost_api_unreachable", + detail, + }), + ] + .into_iter() + .flatten() + .collect(); + let healthy = errors.is_empty(); + + let codes: Vec<&'static str> = errors.iter().map(|e| e.code).collect(); + let status_ = HealthStatus { healthy, errors }; + + if previous_codes.is_some() && previous_codes.as_ref() != Some(&codes) { + if healthy { + tracing::warn!("Gateway became healthy again."); + } else { + tracing::warn!("Gateway unhealthy: {}", status_.details()); + } + } + + previous_codes = Some(codes); + + *status.lock().await = status_; + + first_check_done_.notify_one(); + + let delay = if healthy { + Self::HEALTHY_CHECK_INTERVAL + } else { + Self::UNHEALTHY_CHECK_INTERVAL + }; + + time::sleep(delay).await; + } + }); + + first_check_done.notified().await; + self_ + } +} diff --git a/crates/gateway/src/lib.rs b/crates/gateway/src/lib.rs index a714555b..76b58ebc 100644 --- a/crates/gateway/src/lib.rs +++ b/crates/gateway/src/lib.rs @@ -3,6 +3,7 @@ pub mod blockfrost; pub mod config; pub mod db; pub mod errors; +pub mod health_monitor; pub mod hydra_server_bridge; pub mod hydra_server_platform; pub mod load_balancer; diff --git a/crates/gateway/src/main.rs b/crates/gateway/src/main.rs index a3d86fa7..58e581ef 100644 --- a/crates/gateway/src/main.rs +++ b/crates/gateway/src/main.rs @@ -7,8 +7,8 @@ use axum::{ }; use bf_common::tracing::setup_tracing; use blockfrost_gateway::{ - api, blockfrost, config, db, hydra_server_bridge, hydra_server_platform, load_balancer, - middlewares, rate_limit, sdk_bridge_ws, + api, blockfrost, config, db, health_monitor, hydra_server_bridge, hydra_server_platform, + load_balancer, middlewares, rate_limit, sdk_bridge_ws, }; use clap::Parser; use colored::Colorize; @@ -42,6 +42,16 @@ async fn main() -> Result<()> { ) .await; let blockfrost_api = blockfrost::BlockfrostAPI::new(&config.blockfrost.project_id); + let health_monitor = + health_monitor::HealthMonitor::spawn(pool.clone(), blockfrost_api.clone()).await; + + // Fail fast on startup problems + let initial_health = health_monitor.current_status().await; + if !initial_health.healthy { + tracing::error!("Refusing to start unhealthy: {}", initial_health.details()); + std::process::exit(1); + } + let hydras_manager = if let Some(hydra_platform_config) = &config.hydra_platform { Some( hydra_server_platform::HydrasManager::new( @@ -103,6 +113,7 @@ async fn main() -> Result<()> { .layer(Extension(load_balancer)) .layer(Extension(config.clone())) .layer(Extension(pool)) + .layer(Extension(health_monitor)) .layer(Extension(blockfrost_api)) .layer(Extension(register_rate_limiter)) .layer(Extension(prometheus_handle));