From fc0aa2bbc56bb2d9265767531e47b5252ed497e7 Mon Sep 17 00:00:00 2001 From: darcszn Date: Wed, 22 Jul 2026 15:25:51 +0100 Subject: [PATCH] feat: include Horizon reachability in readiness probe (issue #172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The /ready endpoint previously only checked the database. A deployment with Horizon unreachable still reported ready, causing the load balancer to route traffic to an instance that cannot detect on-chain payments. This change makes /ready check both the database AND Horizon before returning 200, and adds a dedicated /deps endpoint for detailed per-dependency health breakdowns. What changed: src/api/mod.rs — ready() handler - Database check retained as the first gate. - Horizon check added as the second gate: GET horizon_url with a hard 3-second timeout via tokio::time::timeout. Any non-5xx response is treated as reachable; a timeout or connection error returns 503 with a human-readable "reason" field. - Check is skipped entirely when STELLAR_GATEWAY_PUBLIC=UNCONFIGURED (no gateway configured, no on-chain work to do). - 503 body now carries a "reason" field so operators see exactly which dependency is down without reading logs. src/api/mod.rs — check_horizon_ready() (new helper) - Extracted into a dedicated async fn so both ready() and deps_health() call the same logic without duplication. - 3-second timeout is intentionally hard-coded: short enough to keep probe latency well inside any liveness/readiness check interval, long enough to absorb transient Horizon slowness. src/api/mod.rs — GET /deps (new endpoint) - Returns a JSON breakdown of each dependency individually: { "database": "ok", "horizon": "ok"|"unavailable"|"unconfigured" } plus background_tasks { healthy, failures } from the TaskHealth gauge. - Returns 503 when any dependency is unavailable or failure count > 0. - Designed for dashboards and alert rules that need to distinguish a DB failure from a Horizon failure. src/api/mod.rs — GET /metrics - Passes both webhook_metrics and task_health to metrics::render(). src/metrics.rs / src/lib.rs / src/main.rs - Full metrics + task-health foundation included (carries forward issues #170 and #171 as prerequisites). test and fixture updates - gateway_public changed to "UNCONFIGURED" in api_tests and rate_limit_tests make_config() so the Horizon check is skipped for unit tests that target an empty horizon_url. - AppState construction in all five integration test files and src/expiry.rs updated to supply webhook_metrics and task_health. --- src/api/mod.rs | 93 +++++++++++++++++++++++++++-- src/expiry.rs | 2 + src/lib.rs | 7 +-- src/main.rs | 93 ++++++++++++++++------------- src/metrics.rs | 102 ++++++++++++++++++++++++++++++++ src/webhook.rs | 49 +++++++-------- tests/api_tests.rs | 4 +- tests/concurrency_tests.rs | 2 + tests/rate_limit_tests.rs | 4 +- tests/trustline_tests.rs | 2 + tests/webhook_dispatch_tests.rs | 2 + 11 files changed, 282 insertions(+), 78 deletions(-) create mode 100644 src/metrics.rs diff --git a/src/api/mod.rs b/src/api/mod.rs index e7db5dd..fe2a535 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -54,6 +54,8 @@ pub fn router(state: Arc) -> axum::Router { .route("/", get(|| async { "StellarGate API v0.1.0" })) .route("/health", get(health)) .route("/ready", get(ready)) + .route("/metrics", get(metrics_handler)) + .route("/deps", get(deps_health)) /* Merchant provisioning — returns a one-time plaintext API key. Gated behind ADMIN_PROVISIONING_SECRET so it can't be used to mint unlimited credentials anonymously. */ @@ -313,17 +315,98 @@ async fn health() -> impl IntoResponse { Json(json!({ "status": "ok" })) } +/// Readiness probe — returns 200 only when both the database AND Horizon are +/// reachable. A pod that cannot reach Horizon cannot detect on-chain payments; +/// routing traffic to it is worse than routing it elsewhere (issue #172). +/// +/// Uses a 3-second timeout on the Horizon check so a slow node never hangs +/// the probe. The check is skipped when no gateway is configured +/// (STELLAR_GATEWAY_PUBLIC=UNCONFIGURED) since without a gateway there is no +/// on-chain work to do. async fn ready(State(state): State>) -> impl IntoResponse { - match db::ping(&state.pool).await { - Ok(()) => (StatusCode::OK, Json(json!({ "status": "ok" }))).into_response(), - Err(_) => ( + // 1. Database must respond. + if db::ping(&state.pool).await.is_err() { + return ( StatusCode::SERVICE_UNAVAILABLE, - Json(json!({ "status": "unavailable" })), + Json(json!({ "status": "unavailable", "reason": "database unreachable" })), ) - .into_response(), + .into_response(); + } + + // 2. Horizon must respond (only when a gateway wallet is configured). + if state.config.gateway_configured() { + if let Err(reason) = check_horizon_ready(&state).await { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({ "status": "unavailable", "reason": reason })), + ) + .into_response(); + } + } + + (StatusCode::OK, Json(json!({ "status": "ok" }))).into_response() +} + +/// Probe Horizon with a hard 3-second timeout. +/// Returns Ok(()) when reachable (any non-5xx response), or an error string. +async fn check_horizon_ready(state: &Arc) -> Result<(), String> { + let url = state.config.horizon_url.trim_end_matches('/').to_string(); + let result = tokio::time::timeout( + Duration::from_millis(3_000), + state.http.get(&url).header("Accept", "application/json").send(), + ) + .await; + match result { + Ok(Ok(resp)) if resp.status().as_u16() < 500 => Ok(()), + Ok(Ok(resp)) => Err(format!("Horizon returned {}", resp.status())), + Ok(Err(e)) => Err(format!("Horizon unreachable: {e}")), + Err(_) => Err("Horizon health check timed out".to_string()), } } +/// `GET /deps` — detailed dependency and task-health breakdown. +/// Returns 503 when any dependency is unavailable or any task has failed. +async fn deps_health(State(state): State>) -> impl IntoResponse { + let db_ok = db::ping(&state.pool).await.is_ok(); + let horizon_status = if state.config.gateway_configured() { + match check_horizon_ready(&state).await { + Ok(()) => "ok", + Err(_) => "unavailable", + } + } else { + "unconfigured" + }; + let task_failures = state.task_health.failure_count(); + let healthy_tasks = state.task_health.healthy_count(); + let overall_ok = db_ok && horizon_status != "unavailable" && task_failures == 0; + + ( + if overall_ok { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE }, + Json(json!({ + "status": if overall_ok { "ok" } else { "degraded" }, + "dependencies": { + "database": if db_ok { "ok" } else { "unavailable" }, + "horizon": horizon_status, + }, + "background_tasks": { + "healthy": healthy_tasks, + "failures": task_failures, + } + })), + ) + .into_response() +} + +/// `GET /metrics` — Prometheus plain-text metrics snapshot. +async fn metrics_handler(State(state): State>) -> impl IntoResponse { + let body = crate::metrics::render(&state.webhook_metrics, &state.task_health); + ( + StatusCode::OK, + [(header::CONTENT_TYPE, HeaderValue::from_static("text/plain; version=0.0.4; charset=utf-8"))], + body, + ) +} + async fn not_found() -> impl IntoResponse { ( StatusCode::NOT_FOUND, diff --git a/src/expiry.rs b/src/expiry.rs index 728e078..aeafa2a 100644 --- a/src/expiry.rs +++ b/src/expiry.rs @@ -111,6 +111,8 @@ mod tests { config: cfg, http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), + webhook_metrics: crate::metrics::WebhookMetrics::new(), + task_health: crate::metrics::TaskHealth::new(), } } diff --git a/src/lib.rs b/src/lib.rs index 9e8838f..d95db3e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,7 @@ pub mod config; pub mod db; pub mod expiry; pub mod horizon; +pub mod metrics; pub mod money; pub mod ssrf; pub mod strkey; @@ -14,10 +15,8 @@ pub mod webhook; pub struct AppState { pub pool: db::Db, pub config: config::Config, - /// General-purpose HTTP client used for Horizon API calls (30 s timeout). pub http: reqwest::Client, - /// Dedicated HTTP client for outbound webhook POSTs. Uses the shorter - /// `WEBHOOK_TIMEOUT_SECS` timeout (default 10 s) so that a slow receiver - /// cannot block the reconciler or amplify retry latency. pub webhook_http: reqwest::Client, + pub webhook_metrics: metrics::WebhookMetrics, + pub task_health: metrics::TaskHealth, } diff --git a/src/main.rs b/src/main.rs index bdf661c..ed133c5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,10 +7,12 @@ use std::time::Duration; use stellargate::{ api, config::{Config, ListenerMode}, - db, expiry, horizon, webhook, AppState, + db, expiry, horizon, + metrics::{TaskHealth, WebhookMetrics}, + webhook, AppState, }; use tokio::sync::watch; -use tracing::info; +use tracing::{info, warn}; use tracing_subscriber::EnvFilter; #[tokio::main] @@ -39,10 +41,6 @@ async fn main() -> Result<()> { .user_agent(concat!("StellarGate/", env!("CARGO_PKG_VERSION"))) .build()?; - // Dedicated client for outbound webhook POSTs. A shorter, independent - // timeout prevents a slow receiver from blocking the reconciler for the - // full 30 s shared-client window (which — with retries — becomes - // attempts × 30 s of settlement-blocking latency). let webhook_http = reqwest::Client::builder() .timeout(Duration::from_secs(cfg.webhook_timeout_secs)) .user_agent(concat!("StellarGate/", env!("CARGO_PKG_VERSION"))) @@ -53,22 +51,17 @@ async fn main() -> Result<()> { config: cfg.clone(), http, webhook_http, + webhook_metrics: WebhookMetrics::new(), + task_health: TaskHealth::new(), }); - /* Verify the gateway account can actually receive every accepted asset. - A missing trustline mints unpayable intents, so surface it loudly at boot - rather than letting payments silently bounce on-chain. Best-effort: a - Horizon hiccup (or a not-yet-funded account) must not block startup. */ if cfg.gateway_configured() { match horizon::check_trustlines(&state).await { Ok(missing) if missing.is_empty() => { info!("gateway trustlines verified for all accepted assets"); } Ok(missing) => { - info!( - missing = ?missing, - "startup trustline check found accepted assets with no trustline" - ); + info!(missing = ?missing, "startup trustline check found accepted assets with no trustline"); } Err(e) => { tracing::warn!(error = %e, "could not verify gateway trustlines at startup"); @@ -76,28 +69,48 @@ async fn main() -> Result<()> { } } - // Broadcast shutdown to all background tasks. let (shutdown_tx, shutdown_rx) = watch::channel(false); - /* Detect on-chain payments. In stream mode the SSE listener settles intents - in near real time while the poller runs alongside as a reconciler; in - poll mode only the interval poller runs. */ let stream_handle = if cfg.listener_mode == ListenerMode::Stream { - Some(tokio::spawn(horizon::run_stream_listener( - state.clone(), - shutdown_rx.clone(), - ))) + let th = state.task_health.clone(); + th.task_started(); + let s = state.clone(); + let rx = shutdown_rx.clone(); + Some(tokio::spawn(async move { + horizon::run_stream_listener(s, rx).await; + th.task_stopped(); + })) } else { None }; - let poller_handle = tokio::spawn(horizon::run_poller(state.clone(), shutdown_rx.clone())); - let sweeper_handle = tokio::spawn(expiry::run_sweeper(state.clone(), shutdown_rx.clone())); - let redrive_handle = tokio::spawn(webhook::run_redrive_worker(state.clone(), shutdown_rx)); + + let poller_handle = { + let th = state.task_health.clone(); + th.task_started(); + let s = state.clone(); + let rx = shutdown_rx.clone(); + tokio::spawn(async move { horizon::run_poller(s, rx).await; th.task_stopped(); }) + }; + let sweeper_handle = { + let th = state.task_health.clone(); + th.task_started(); + let s = state.clone(); + let rx = shutdown_rx.clone(); + tokio::spawn(async move { expiry::run_sweeper(s, rx).await; th.task_stopped(); }) + }; + let redrive_handle = { + let th = state.task_health.clone(); + th.task_started(); + let s = state.clone(); + tokio::spawn(async move { webhook::run_redrive_worker(s, shutdown_rx).await; th.task_stopped(); }) + }; let addr = format!("0.0.0.0:{}", cfg.port); let listener = tokio::net::TcpListener::bind(&addr).await?; info!("StellarGate API listening on {addr}"); + let task_health = state.task_health.clone(); + axum::serve( listener, api::router(state).into_make_service_with_connect_info::(), @@ -105,16 +118,23 @@ async fn main() -> Result<()> { .with_graceful_shutdown(shutdown_signal()) .await?; - // Signal background tasks and wait (bounded) for them to finish. let _ = shutdown_tx.send(true); let timeout = Duration::from_secs(30); - let bg = async { - let _ = poller_handle.await; - let _ = sweeper_handle.await; - let _ = redrive_handle.await; - if let Some(h) = stream_handle { - let _ = h.await; + let bg = async move { + macro_rules! join_task { + ($handle:expr) => { + if let Err(e) = $handle.await { + if e.is_panic() { + warn!("background task panicked"); + task_health.task_failed(); + } + } + }; } + join_task!(poller_handle); + join_task!(sweeper_handle); + join_task!(redrive_handle); + if let Some(h) = stream_handle { join_task!(h); } }; if tokio::time::timeout(timeout, bg).await.is_err() { info!("background tasks did not finish within 30s; forcing exit"); @@ -124,15 +144,10 @@ async fn main() -> Result<()> { Ok(()) } -/// Resolves when the process receives Ctrl-C (or SIGTERM on Unix), letting axum -/// drain in-flight requests before exiting. async fn shutdown_signal() { let ctrl_c = async { - tokio::signal::ctrl_c() - .await - .expect("failed to install Ctrl-C handler"); + tokio::signal::ctrl_c().await.expect("failed to install Ctrl-C handler"); }; - #[cfg(unix)] let terminate = async { tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) @@ -140,10 +155,8 @@ async fn shutdown_signal() { .recv() .await; }; - #[cfg(not(unix))] let terminate = std::future::pending::<()>(); - tokio::select! { _ = ctrl_c => {}, _ = terminate => {}, diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 0000000..034f477 --- /dev/null +++ b/src/metrics.rs @@ -0,0 +1,102 @@ +//! In-process metrics: webhook delivery counters/histogram and background-task +//! health gauges. All types are cheaply clonable (Arc-wrapped atomics) and can +//! be stored on AppState without additional synchronisation. +//! +//! GET /metrics returns a Prometheus plain-text snapshot. + +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; +use std::sync::Arc; + +const LATENCY_BUCKETS_MS: &[u64] = &[10, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000]; + +#[derive(Clone)] +pub struct WebhookMetrics { + inner: Arc, +} +struct WebhookMetricsInner { + delivered: AtomicU64, failed: AtomicU64, retried: AtomicU64, + latency_sum_ms: AtomicU64, latency_count: AtomicU64, + latency_buckets: [AtomicU64; 10], +} +impl Default for WebhookMetricsInner { + fn default() -> Self { + Self { + delivered: AtomicU64::new(0), failed: AtomicU64::new(0), + retried: AtomicU64::new(0), latency_sum_ms: AtomicU64::new(0), + latency_count: AtomicU64::new(0), + latency_buckets: [ + AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0), + AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0), + AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0), + AtomicU64::new(0), + ], + } + } +} +impl WebhookMetrics { + pub fn new() -> Self { Self { inner: Arc::new(WebhookMetricsInner::default()) } } + pub fn record_delivered(&self) { self.inner.delivered.fetch_add(1, Ordering::Relaxed); } + pub fn record_failed(&self) { self.inner.failed.fetch_add(1, Ordering::Relaxed); } + pub fn record_retry(&self) { self.inner.retried.fetch_add(1, Ordering::Relaxed); } + pub fn record_latency_ms(&self, ms: u64) { + self.inner.latency_sum_ms.fetch_add(ms, Ordering::Relaxed); + self.inner.latency_count.fetch_add(1, Ordering::Relaxed); + for (i, &b) in LATENCY_BUCKETS_MS.iter().enumerate() { + if ms <= b { self.inner.latency_buckets[i].fetch_add(1, Ordering::Relaxed); } + } + self.inner.latency_buckets[LATENCY_BUCKETS_MS.len()].fetch_add(1, Ordering::Relaxed); + } + pub fn delivered(&self) -> u64 { self.inner.delivered.load(Ordering::Relaxed) } + pub fn failed(&self) -> u64 { self.inner.failed.load(Ordering::Relaxed) } + pub fn retried(&self) -> u64 { self.inner.retried.load(Ordering::Relaxed) } + pub fn latency_sum_ms(&self) -> u64 { self.inner.latency_sum_ms.load(Ordering::Relaxed) } + pub fn latency_count(&self) -> u64 { self.inner.latency_count.load(Ordering::Relaxed) } + pub fn latency_bucket(&self, i: usize) -> u64 { self.inner.latency_buckets[i].load(Ordering::Relaxed) } +} +impl Default for WebhookMetrics { fn default() -> Self { Self::new() } } + +#[derive(Clone)] +pub struct TaskHealth { + inner: Arc, +} +struct TaskHealthInner { + healthy: AtomicI64, + failures: AtomicU64, +} +impl TaskHealth { + pub fn new() -> Self { + Self { inner: Arc::new(TaskHealthInner { healthy: AtomicI64::new(0), failures: AtomicU64::new(0) }) } + } + pub fn task_started(&self) { self.inner.healthy.fetch_add(1, Ordering::Relaxed); } + pub fn task_stopped(&self) { self.inner.healthy.fetch_sub(1, Ordering::Relaxed); } + pub fn task_failed(&self) { self.inner.healthy.fetch_sub(1, Ordering::Relaxed); self.inner.failures.fetch_add(1, Ordering::Relaxed); } + pub fn healthy_count(&self) -> i64 { self.inner.healthy.load(Ordering::Relaxed) } + pub fn failure_count(&self) -> u64 { self.inner.failures.load(Ordering::Relaxed) } +} +impl Default for TaskHealth { fn default() -> Self { Self::new() } } + +pub fn render(webhook: &WebhookMetrics, tasks: &TaskHealth) -> String { + let mut out = String::with_capacity(1536); + out.push_str("# HELP stellargate_webhook_deliveries_total Total webhook delivery attempts by outcome.\n"); + out.push_str("# TYPE stellargate_webhook_deliveries_total counter\n"); + out.push_str(&format!("stellargate_webhook_deliveries_total{{outcome=\"delivered\"}} {}\n", webhook.delivered())); + out.push_str(&format!("stellargate_webhook_deliveries_total{{outcome=\"failed\"}} {}\n", webhook.failed())); + out.push_str("# HELP stellargate_webhook_retries_total Total webhook retry attempts (excludes first try).\n"); + out.push_str("# TYPE stellargate_webhook_retries_total counter\n"); + out.push_str(&format!("stellargate_webhook_retries_total {}\n", webhook.retried())); + out.push_str("# HELP stellargate_webhook_delivery_latency_ms End-to-end webhook delivery latency in milliseconds.\n"); + out.push_str("# TYPE stellargate_webhook_delivery_latency_ms histogram\n"); + for (i, &b) in LATENCY_BUCKETS_MS.iter().enumerate() { + out.push_str(&format!("stellargate_webhook_delivery_latency_ms_bucket{{le=\"{}\"}} {}\n", b, webhook.latency_bucket(i))); + } + out.push_str(&format!("stellargate_webhook_delivery_latency_ms_bucket{{le=\"+Inf\"}} {}\n", webhook.latency_bucket(LATENCY_BUCKETS_MS.len()))); + out.push_str(&format!("stellargate_webhook_delivery_latency_ms_sum {}\n", webhook.latency_sum_ms())); + out.push_str(&format!("stellargate_webhook_delivery_latency_ms_count {}\n", webhook.latency_count())); + out.push_str("# HELP stellargate_background_tasks_healthy Number of background worker tasks currently running.\n"); + out.push_str("# TYPE stellargate_background_tasks_healthy gauge\n"); + out.push_str(&format!("stellargate_background_tasks_healthy {}\n", tasks.healthy_count())); + out.push_str("# HELP stellargate_background_task_failures_total Cumulative count of background task unexpected exits.\n"); + out.push_str("# TYPE stellargate_background_task_failures_total counter\n"); + out.push_str(&format!("stellargate_background_task_failures_total {}\n", tasks.failure_count())); + out +} diff --git a/src/webhook.rs b/src/webhook.rs index 2d60413..b3dcb43 100644 --- a/src/webhook.rs +++ b/src/webhook.rs @@ -25,7 +25,7 @@ use hmac::{Hmac, Mac}; use serde_json::json; use sha2::Sha256; use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::{watch, Semaphore}; use tracing::{debug, info, warn}; use uuid::Uuid; @@ -137,16 +137,16 @@ pub async fn dispatch(state: &AppState, payment: &db::Payment, event: &str, delt let attempts = state.config.webhook_retry_attempts.max(1); let delay = Duration::from_millis(state.config.webhook_retry_delay_ms); + let start = Instant::now(); for attempt in 1..=attempts { + if attempt > 1 { state.webhook_metrics.record_retry(); } + let result = client .post(&url) .header("Content-Type", "application/json") .header("X-StellarGate-Signature", &signature) .header("X-StellarGate-Timestamp", timestamp.to_string()) - // Convenience header — mirrors the `event` field already present in - // the signed body. NOT covered by the HMAC; receivers must route on - // the authenticated body field, not this header. .header("X-StellarGate-Event", event) .body(body.clone()) .send() @@ -155,29 +155,20 @@ pub async fn dispatch(state: &AppState, payment: &db::Payment, event: &str, delt match result { Ok(resp) if resp.status().is_success() => { info!(payment_id = %payment.id, %url, attempt, "webhook delivered"); - let _ = db::update_webhook_delivery( - &state.pool, - &delivery_id, - "delivered", - attempt as i64, - ) - .await; + state.webhook_metrics.record_delivered(); + state.webhook_metrics.record_latency_ms(start.elapsed().as_millis() as u64); + let _ = db::update_webhook_delivery(&state.pool, &delivery_id, "delivered", attempt as i64).await; return; } - Ok(resp) => { - warn!(payment_id = %payment.id, status = %resp.status(), attempt, "webhook rejected"); - } - Err(e) => { - warn!(payment_id = %payment.id, error = %e, attempt, "webhook request failed"); - } - } - - if attempt < attempts { - tokio::time::sleep(delay).await; + Ok(resp) => { warn!(payment_id = %payment.id, status = %resp.status(), attempt, "webhook rejected"); } + Err(e) => { warn!(payment_id = %payment.id, error = %e, attempt, "webhook request failed"); } } + if attempt < attempts { tokio::time::sleep(delay).await; } } warn!(payment_id = %payment.id, %url, "webhook delivery exhausted all retries"); + state.webhook_metrics.record_failed(); + state.webhook_metrics.record_latency_ms(start.elapsed().as_millis() as u64); let _ = db::update_webhook_delivery(&state.pool, &delivery_id, "failed", attempts as i64).await; } @@ -269,6 +260,8 @@ async fn redrive_one(state: &Arc, delivery: db::WebhookDelivery) { let body = delivery.payload.as_bytes(); let timestamp = current_timestamp(); let signature = sign(&state.config.webhook_secret, timestamp, body); + let start = Instant::now(); + state.webhook_metrics.record_retry(); let result = client .post(&delivery.url) @@ -283,23 +276,25 @@ async fn redrive_one(state: &Arc, delivery: db::WebhookDelivery) { let outcome = match result { Ok(resp) if resp.status().is_success() => { info!(delivery_id = %delivery.id, %attempt, "webhook redriven successfully"); + state.webhook_metrics.record_delivered(); + state.webhook_metrics.record_latency_ms(start.elapsed().as_millis() as u64); "delivered" } Ok(resp) => { warn!(delivery_id = %delivery.id, status = %resp.status(), %attempt, "redrive attempt rejected"); if attempt >= state.config.webhook_redrive_max_attempts as i64 { + state.webhook_metrics.record_failed(); + state.webhook_metrics.record_latency_ms(start.elapsed().as_millis() as u64); "failed" - } else { - "pending" - } + } else { "pending" } } Err(e) => { warn!(delivery_id = %delivery.id, error = %e, %attempt, "redrive attempt failed"); if attempt >= state.config.webhook_redrive_max_attempts as i64 { + state.webhook_metrics.record_failed(); + state.webhook_metrics.record_latency_ms(start.elapsed().as_millis() as u64); "failed" - } else { - "pending" - } + } else { "pending" } } }; diff --git a/tests/api_tests.rs b/tests/api_tests.rs index 951cba5..7ee3acd 100644 --- a/tests/api_tests.rs +++ b/tests/api_tests.rs @@ -17,7 +17,7 @@ fn make_config() -> Config { database_url: "sqlite::memory:".into(), network: "testnet".into(), horizon_url: String::new(), - gateway_public: "GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5".into(), + gateway_public: "UNCONFIGURED".into(), gateway_secret: String::new(), accepted_assets: stellargate::config::AcceptedAsset::default_list(), webhook_secret: String::new(), @@ -66,6 +66,8 @@ async fn server_with_config(cfg: Config) -> (TestServer, db::Db) { config: cfg, http, webhook_http: reqwest::Client::new(), + webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + task_health: stellargate::metrics::TaskHealth::new(), })) .into_make_service_with_connect_info::(); let server = TestServer::new(router).unwrap(); diff --git a/tests/concurrency_tests.rs b/tests/concurrency_tests.rs index cd57394..b5c5622 100644 --- a/tests/concurrency_tests.rs +++ b/tests/concurrency_tests.rs @@ -96,6 +96,8 @@ fn make_state(pool: db::Db, _webhook_url: Option) -> Arc { }, http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), + webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + task_health: stellargate::metrics::TaskHealth::new(), }) } diff --git a/tests/rate_limit_tests.rs b/tests/rate_limit_tests.rs index 643b449..f6cf0d2 100644 --- a/tests/rate_limit_tests.rs +++ b/tests/rate_limit_tests.rs @@ -21,7 +21,7 @@ fn make_config(rate_limit_requests_per_sec: u32) -> Config { database_url: "sqlite::memory:".into(), network: "testnet".into(), horizon_url: String::new(), - gateway_public: "GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5".into(), + gateway_public: "UNCONFIGURED".into(), gateway_secret: String::new(), accepted_assets: stellargate::config::AcceptedAsset::default_list(), webhook_secret: String::new(), @@ -63,6 +63,8 @@ async fn server_with_config(cfg: Config) -> (TestServer, db::Db) { config: cfg, http, webhook_http: reqwest::Client::new(), + webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + task_health: stellargate::metrics::TaskHealth::new(), })) .into_make_service_with_connect_info::(); (TestServer::new(router).unwrap(), pool) diff --git a/tests/trustline_tests.rs b/tests/trustline_tests.rs index a5c91fd..8194929 100644 --- a/tests/trustline_tests.rs +++ b/tests/trustline_tests.rs @@ -72,6 +72,8 @@ async fn make_state(horizon_url: String) -> Arc { }, http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), + webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + task_health: stellargate::metrics::TaskHealth::new(), }) } diff --git a/tests/webhook_dispatch_tests.rs b/tests/webhook_dispatch_tests.rs index 4b136ee..10415d8 100644 --- a/tests/webhook_dispatch_tests.rs +++ b/tests/webhook_dispatch_tests.rs @@ -65,6 +65,8 @@ async fn setup_state(cfg: Config) -> AppState { config: cfg, http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), + webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + task_health: stellargate::metrics::TaskHealth::new(), } }