fix: add /health/all endpoint for per-bot health monitoring (#14)

buzzkillb · web-flow · commit 7bda8536a7f2 · 2026-04-08T10:38:22.000-07:00
* Simplify health/status responses and upgrade image crate to 0.25

* Fix image 0.25 API: use encoder.write_image instead of deprecated write_with_encoder

* Fix: SHANGHAI premium display, BTC rotation, HTML parsing fallback, semaphore in struct, health server bind failures

* Fix division by zero, add price validation, add uniqueness constraint on prices

* Use execute instead of prepare_cached for INSERT OR REPLACE

* Fix: use config.update_interval for rotation, add is_fallback flag to PriceData

* Fix unreachable code, use is_fallback for stale indicator

* Fix: division by zero guard, update fallback prices, add chart range guard

* Fix: semaphore rate limiting, integer underflow, unreachable panics, silent error swallowing, division by zero

* Security: safer regex, sanitize error messages for users, reduce DB pool size

* fix: add /health/all endpoint for per-bot health monitoring

- Added is_all_healthy() method that returns true only when ALL bots are healthy
- Added /health/all endpoint that returns 503 if any bot is unhealthy
- Updated Docker health check to use /health/all instead of /health
- This ensures container restarts when any bot (e.g. JLP) drops offline
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -21,7 +21,7 @@ services:
         max-size: "10m"
         max-file: "3"
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
+      test: ["CMD", "curl", "-f", "http://localhost:8080/health/all"]
       interval: 30s
       timeout: 10s
       retries: 3
diff --git a/src/bot.rs b/src/bot.rs
diff --git a/src/charting.rs b/src/charting.rs
@@ -173,10 +173,15 @@ pub fn generate_price_chart(
             .map(|(_, p)| *p)
             .fold(f64::NEG_INFINITY, f64::max);
 
-        // Add padding
-        let range_padding = (max_price - min_price) * 0.05;
-        let y_min = min_price - range_padding;
-        let y_max = max_price + range_padding;
+        // Ensure minimum range for Y axis
+        let price_range = (max_price - min_price).abs();
+        let (y_min, y_max) = if price_range < 0.0001 {
+            (min_price - 1.0, max_price + 1.0)
+        } else {
+            // Add padding
+            let range_padding = price_range * 0.05;
+            (min_price - range_padding, max_price + range_padding)
+        };
 
         let mut chart = ChartBuilder::on(&root)
             .caption(title, text_style)
diff --git a/src/database.rs b/src/database.rs
@@ -33,6 +33,7 @@ impl PriceDatabase {
                          timestamp INTEGER NOT NULL,
                          created_at TEXT DEFAULT CURRENT_TIMESTAMP
                      );
+                     CREATE UNIQUE INDEX IF NOT EXISTS idx_prices_crypto_timestamp_unique ON prices(crypto_name, timestamp);
                      CREATE INDEX IF NOT EXISTS idx_prices_crypto_timestamp ON prices(crypto_name, timestamp);
                      CREATE TABLE IF NOT EXISTS price_aggregates (
                          id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -53,7 +54,7 @@ impl PriceDatabase {
         });
 
         let pool = Pool::builder()
-            .max_size(10) // Allow up to 10 concurrent connections
+            .max_size(4) // SQLite performs better with fewer connections
             .build(manager)
             .map_err(|e| {
                 BotError::Database(rusqlite::Error::ToSqlConversionFailure(Box::new(e)))
@@ -83,11 +84,10 @@ impl PriceDatabase {
         let conn = self.get_connection()?;
         let current_time = get_current_timestamp()?;
 
-        let mut stmt = conn.prepare_cached(
-            "INSERT INTO prices (crypto_name, price, timestamp) VALUES (?, ?, ?)",
+        conn.execute(
+            "INSERT OR REPLACE INTO prices (crypto_name, price, timestamp) VALUES (?1, ?2, ?3)",
+            [crypto_name, &price.to_string(), &current_time.to_string()],
         )?;
-
-        stmt.execute([crypto_name, &price.to_string(), &current_time.to_string()])?;
         debug!("Saved {} price to database: ${}", crypto_name, price);
         Ok(())
     }
@@ -163,7 +163,16 @@ impl PriceDatabase {
         ];
 
         for (seconds, label) in periods {
-            let time_ago = current_time - seconds;
+            // Guard against underflow if clock goes backwards
+            let time_ago = if current_time >= seconds {
+                current_time - seconds
+            } else {
+                debug!(
+                    "Clock appears to have gone backwards, skipping {} price lookup",
+                    label
+                );
+                continue;
+            };
 
             // Try to get price from appropriate data source based on age
             let old_price = if seconds <= 24 * 3600 {
diff --git a/src/discord_api.rs b/src/discord_api.rs
@@ -15,30 +15,34 @@ const RATE_LIMIT_DELAY_MS: u64 = 2000; // 2 seconds between Discord API calls
 #[derive(Clone)]
 pub struct DiscordApi {
     http: Arc<Http>,
+    semaphore: Arc<Semaphore>,
 }
 
 impl DiscordApi {
     pub fn new(http: Arc<Http>) -> Self {
-        Self { http }
+        Self {
+            http,
+            semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_CALLS)),
+        }
     }
 
     /// Rate-limited Discord API call helper
+    #[allow(unused_variables)]
     async fn rate_limited_call<F, Fut, T>(&self, mut operation: F) -> Result<T, serenity::Error>
     where
         F: FnMut() -> Fut,
         Fut: std::future::Future<Output = Result<T, serenity::Error>>,
     {
-        static SEMAPHORE: Semaphore = Semaphore::const_new(MAX_CONCURRENT_CALLS);
-
-        let _permit = SEMAPHORE
+        let permit = self
+            .semaphore
             .acquire()
             .await
             .map_err(|_| serenity::Error::Other("Semaphore acquire error"))?;
 
         // Enforce minimum delay between calls
         sleep(Duration::from_millis(RATE_LIMIT_DELAY_MS)).await;
 
-        // Execute the operation with retry logic
+        // Execute the operation with retry logic - permit is held during all attempts
         for attempt in 1..=MAX_RETRIES {
             match operation().await {
                 Ok(result) => return Ok(result),
@@ -60,7 +64,7 @@ impl DiscordApi {
             }
         }
 
-        unreachable!()
+        Err(serenity::Error::Other("Rate limiting exhausted"))
     }
 
     /// Update bot nickname in a specific guild
diff --git a/src/errors.rs b/src/errors.rs
@@ -33,6 +33,24 @@ pub enum BotError {
     Parse(String),
 }
 
+impl BotError {
+    pub fn user_message(&self) -> &'static str {
+        match self {
+            BotError::PriceNotFound(_) => "Price data not available for this asset",
+            BotError::InvalidInput(_) => "Invalid input provided",
+            BotError::Discord(_) => "Discord API error - please try again",
+            BotError::EnvVar(_) => "Configuration error - please contact support",
+            // Internal errors - don't expose details to users
+            BotError::Database(_)
+            | BotError::Http(_)
+            | BotError::Json(_)
+            | BotError::SystemTime(_)
+            | BotError::Io(_)
+            | BotError::Parse(_) => "An internal error occurred. Please try again later.",
+        }
+    }
+}
+
 impl From<std::env::VarError> for BotError {
     fn from(err: std::env::VarError) -> Self {
         BotError::EnvVar(err.to_string())
@@ -51,4 +69,4 @@ impl From<std::num::ParseFloatError> for BotError {
     }
 }
 
-pub type BotResult<T> = Result<T, BotError>; 
+pub type BotResult<T> = Result<T, BotError>;
diff --git a/src/health.rs b/src/health.rs
@@ -208,6 +208,16 @@ impl HealthAggregator {
         false
     }
 
+    pub fn is_all_healthy(&self) -> bool {
+        if let Ok(bots) = self.bots.lock() {
+            if bots.is_empty() {
+                return true;
+            }
+            return bots.iter().all(|b| b.is_healthy());
+        }
+        false
+    }
+
     pub fn to_json(&self) -> serde_json::Value {
         let bots = match self.bots.lock() {
             Ok(bots) => bots,
diff --git a/src/health_server.rs b/src/health_server.rs
@@ -6,26 +6,28 @@ use tracing::{error, info};
 
 pub type SharedHealth = Arc<HealthAggregator>;
 
-pub async fn start_health_server(health: SharedHealth, port: u16) {
+pub async fn start_health_server(
+    health: SharedHealth,
+    port: u16,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let app = Router::new()
         .route("/health", get(health_check))
+        .route("/health/all", get(health_check_all))
         .route("/", get(health_check))
         .route("/test-discord", get(test_discord_connectivity))
         .with_state(health);
 
     let addr = format!("127.0.0.1:{}", port);
 
-    match TcpListener::bind(&addr).await {
-        Ok(listener) => {
-            info!("Health check server listening on {}", addr);
-            if let Err(e) = axum::serve(listener, app).await {
-                error!("Health server error: {}", e);
-            }
-        }
-        Err(e) => {
-            error!("Failed to bind health server to {}: {}", addr, e);
-        }
-    }
+    let listener = TcpListener::bind(&addr).await.map_err(|e| {
+        error!("Failed to bind health server to {}: {}", addr, e);
+        e
+    })?;
+
+    info!("Health check server listening on {}", addr);
+    axum::serve(listener, app).await?;
+
+    Ok(())
 }
 
 async fn health_check(
@@ -44,6 +46,19 @@ async fn health_check(
     }
 }
 
+async fn health_check_all(
+    State(health): State<SharedHealth>,
+) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
+    let is_all_healthy = health.is_all_healthy();
+    let status = health.to_json();
+
+    if is_all_healthy {
+        Ok(Json(status))
+    } else {
+        Err((StatusCode::SERVICE_UNAVAILABLE, Json(status)))
+    }
+}
+
 async fn test_discord_connectivity(
     State(_health): State<SharedHealth>,
 ) -> Result<Json<serde_json::Value>, StatusCode> {
diff --git a/src/main.rs b/src/main.rs
@@ -1,31 +1,31 @@
-mod errors;
+mod bot;
+mod charting;
 mod config;
-mod utils;
-mod health;
-mod health_server;
 mod database;
+#[cfg(test)]
+mod database_tests;
+mod db_cleanup;
 mod discord_api;
-mod bot;
+mod errors;
+mod health;
+mod health_server;
 mod price_service;
-mod db_cleanup;
-mod charting;
 mod shanghai_price_service;
-#[cfg(test)]
-mod database_tests;
+mod utils;
 
-use errors::BotResult;
+use bot::start_bot;
 use config::BotConfig;
 use database::PriceDatabase;
 use db_cleanup::DatabaseCleanup;
-use bot::start_bot;
-use health::{HealthState, HealthAggregator};
+use errors::BotResult;
+use health::{HealthAggregator, HealthState};
 use health_server::start_health_server;
 
 use dotenv::dotenv;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::sleep;
-use tracing::{info, error, warn};
+use tracing::{error, info, warn};
 
 const RECONNECT_DELAY_SECONDS: u64 = 30;
 
@@ -35,10 +35,10 @@ async fn main() -> BotResult<()> {
     tracing_subscriber::fmt()
         .with_env_filter("info,discord_bot=debug,discord_bot::database=info")
         .init();
-    
+
     info!("🚀 Starting RustyMcPriceface Unified Container...");
     dotenv().ok();
-    
+
     // Initialize shared database
     info!("📦 Initializing shared database...");
     let db = match PriceDatabase::new(config::DATABASE_PATH) {
@@ -110,7 +110,7 @@ async fn main() -> BotResult<()> {
         // Create health state for this bot and register with aggregator
         let health = Arc::new(HealthState::new(ticker.clone()));
         let health_clone = health.clone();
-        
+
         // Add to aggregator
         health_agg_clone.add_bot(health);
 
@@ -122,16 +122,26 @@ async fn main() -> BotResult<()> {
                 let emoji = utils::get_crypto_emoji(&ticker);
                 info!("{} Starting {} bot...", emoji, ticker);
 
-                match start_bot(bot_config.clone(), db_clone.clone(), health_clone.clone(), health_agg_clone.clone()).await {
+                match start_bot(
+                    bot_config.clone(),
+                    db_clone.clone(),
+                    health_clone.clone(),
+                    health_agg_clone.clone(),
+                )
+                .await
+                {
                     Ok(_) => {
                         error!("{} {} bot exited unexpectedly", emoji, ticker);
-                    },
+                    }
                     Err(e) => {
                         error!("{} {} bot crashed: {}", emoji, ticker, e);
                     }
                 }
 
-                error!("{} Restarting {} bot in {} seconds...", emoji, ticker, RECONNECT_DELAY_SECONDS);
+                error!(
+                    "{} Restarting {} bot in {} seconds...",
+                    emoji, ticker, RECONNECT_DELAY_SECONDS
+                );
                 sleep(Duration::from_secs(RECONNECT_DELAY_SECONDS)).await;
             }
         });
@@ -142,7 +152,10 @@ async fn main() -> BotResult<()> {
     info!("🏥 Starting health check server...");
     let health_for_server = health_aggregator.clone();
     tokio::spawn(async move {
-        start_health_server(health_for_server, 8080).await;
+        if let Err(e) = start_health_server(health_for_server, 8080).await {
+            error!("❌ Health server failed: {}", e);
+            panic!("Health server must start successfully for container health checks");
+        }
     });
 
     // Give health server time to start
@@ -158,6 +171,6 @@ async fn main() -> BotResult<()> {
     } else {
         warn!("⚠️ No bots to run. Exiting.");
     }
-    
+
     Ok(())
-}
+}
diff --git a/src/price_service.rs b/src/price_service.rs
