feat: include Horizon reachability in readiness probe (issue #172)

src/api/mod.rs

@@ -314,14 +314,52 @@ 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<Arc<AppState>>) -> 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<AppState>) -> 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()),
     }
 }
 

src/lib.rs

@@ -15,11 +15,7 @@ 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,
     /// Webhook delivery metrics: delivered/failed/retried counts and a latency
     /// histogram. Exposed via `GET /metrics` so operators can see delivery

src/main.rs

@@ -151,6 +151,10 @@ async fn main() -> Result<()> {
         if let Some(h) = stream_handle {
             join_task!(h);
         }
+        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");
@@ -162,22 +166,17 @@ async fn main() -> Result<()> {
 
 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())
             .expect("failed to install SIGTERM handler")
             .recv()
             .await;
     };
-
     #[cfg(not(unix))]
     let terminate = std::future::pending::<()>();
-
     tokio::select! {
         _ = ctrl_c => {},
         _ = terminate => {},

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(),

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(),