Merge pull request #81 from Vovke/api-specs

feat: server health call implementation

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.2.3] - 2024-10-15
+
+### 🚀 Features
+
+- Server health call implementation
+
 ## [0.2.2] - 2024-10-10
 
 ### 🚀 Features

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "kalatori"
 authors = ["Alzymologist Oy <contact@zymologia.fi>"]
-version = "0.2.2"
+version = "0.2.3"
 edition = "2021"
 description = "A gateway daemon for Kalatori."
 license = "GPL-3.0-or-later"

src/chain/definitions.rs

@@ -11,7 +11,7 @@ use crate::{
         tracker::ChainWatcher,
     },
     definitions::{
-        api_v2::{OrderInfo, Timestamp},
+        api_v2::{OrderInfo, RpcInfo, Timestamp},
         Balance,
     },
     error::{ChainError, NotHex},
@@ -89,6 +89,7 @@ pub enum ChainRequest {
     WatchAccount(WatchAccount),
     Reap(WatchAccount),
     Shutdown(oneshot::Sender<()>),
+    GetConnectedRpcs(oneshot::Sender<Vec<RpcInfo>>),
 }
 
 #[derive(Debug)]

src/chain/mod.rs

@@ -1,7 +1,6 @@
 //! Everything related to actual interaction with blockchain
 
 use std::collections::HashMap;
-
 use substrate_crypto_light::common::AccountId32;
 use tokio::{
     sync::{mpsc, oneshot},
@@ -23,6 +22,7 @@ pub mod rpc;
 pub mod tracker;
 pub mod utils;
 
+use crate::definitions::api_v2::{Health, RpcInfo, ServerHealth};
 use definitions::{ChainRequest, ChainTrackerRequest, WatchAccount};
 use tracker::start_chain_watch;
 
@@ -35,7 +35,7 @@ const SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(120000);
 /// RPC server handle
 #[derive(Clone, Debug)]
 pub struct ChainManager {
-    pub tx: tokio::sync::mpsc::Sender<ChainRequest>,
+    pub tx: mpsc::Sender<ChainRequest>,
 }
 
 impl ChainManager {
@@ -54,6 +54,9 @@ impl ChainManager {
 
         let mut currency_map = HashMap::new();
 
+        // Create a channel for receiving RPC status updates
+        let (rpc_update_tx, mut rpc_update_rx) = mpsc::channel(1024);
+
         // start network monitors
         for c in chain {
             if c.endpoints.is_empty() {
@@ -82,61 +85,86 @@ impl ChainManager {
                 signer.interface(),
                 task_tracker.clone(),
                 cancellation_token.clone(),
+                rpc_update_tx.clone(),
             );
         }
 
         task_tracker
             .clone()
             .spawn("Blockchain connections manager", async move {
+                let mut rpc_statuses: HashMap<(String, String), Health> = HashMap::new();
+
 
                 // start requests engine
-                while let Some(request) = rx.recv().await {
-                    match request {
-                        ChainRequest::WatchAccount(request) => {
-                            if let Some(chain) = currency_map.get(&request.currency) {
-                                if let Some(receiver) = watch_chain.get(chain) {
-                                    let _unused = receiver
-                                        .send(ChainTrackerRequest::WatchAccount(request))
-                                        .await;
-                                } else {
-                                    let _unused = request
-                                        .res
-                                        .send(Err(ChainError::InvalidChain(chain.to_string())));
+                loop {
+                    tokio::select! {
+                        Some(request) = rx.recv() => {
+                            match request {
+                                ChainRequest::WatchAccount(request) => {
+                                    if let Some(chain) = currency_map.get(&request.currency) {
+                                        if let Some(receiver) = watch_chain.get(chain) {
+                                            let _unused = receiver
+                                                .send(ChainTrackerRequest::WatchAccount(request))
+                                                .await;
+                                        } else {
+                                            let _unused = request
+                                                .res
+                                                .send(Err(ChainError::InvalidChain(chain.to_string())));
+                                        }
+                                    } else {
+                                        let _unused = request
+                                            .res
+                                            .send(Err(ChainError::InvalidCurrency(request.currency)));
+                                    }
                                 }
-                            } else {
-                                let _unused = request
-                                    .res
-                                    .send(Err(ChainError::InvalidCurrency(request.currency)));
-                            }
-                        }
-                        ChainRequest::Reap(request) => {
-                            if let Some(chain) = currency_map.get(&request.currency) {
-                                if let Some(receiver) = watch_chain.get(chain) {
-                                    let _unused =
-                                        receiver.send(ChainTrackerRequest::Reap(request)).await;
-                                } else {
-                                    let _unused = request
-                                        .res
-                                        .send(Err(ChainError::InvalidChain(chain.to_string())));
+                                ChainRequest::Reap(request) => {
+                                    if let Some(chain) = currency_map.get(&request.currency) {
+                                        if let Some(receiver) = watch_chain.get(chain) {
+                                            let _unused =
+                                                receiver.send(ChainTrackerRequest::Reap(request)).await;
+                                        } else {
+                                            let _unused = request
+                                                .res
+                                                .send(Err(ChainError::InvalidChain(chain.to_string())));
+                                        }
+                                    } else {
+                                        let _unused = request
+                                            .res
+                                            .send(Err(ChainError::InvalidCurrency(request.currency)));
+                                    }
                                 }
-                            } else {
-                                let _unused = request
-                                    .res
-                                    .send(Err(ChainError::InvalidCurrency(request.currency)));
-                            }
-                        }
-                        ChainRequest::Shutdown(res) => {
-                            for (name, chain) in watch_chain.drain() {
-                                let (tx, rx) = oneshot::channel();
-                                if chain.send(ChainTrackerRequest::Shutdown(tx)).await.is_ok() {
-                                    if timeout(SHUTDOWN_TIMEOUT, rx).await.is_err() {
-                                        tracing::error!("Chain monitor for {name} took too much time to wind down, probably it was frozen. Discarding it.");
-                                    };
+                                ChainRequest::Shutdown(res) => {
+                                    for (name, chain) in watch_chain.drain() {
+                                        let (tx, rx) = oneshot::channel();
+                                        if chain.send(ChainTrackerRequest::Shutdown(tx)).await.is_ok() {
+                                            if timeout(SHUTDOWN_TIMEOUT, rx).await.is_err() {
+                                                tracing::error!("Chain monitor for {name} took too much time to wind down, probably it was frozen. Discarding it.");
+                                            };
+                                        }
+                                    }
+                                    let _ = res.send(());
+                                    break;
+                                }
+                                ChainRequest::GetConnectedRpcs(res_tx) => {
+                                    // Collect the RpcInfo from rpc_statuses
+                                    let connected_rpcs: Vec<RpcInfo> = rpc_statuses.iter().map(|((chain_name, rpc_url), status)| {
+                                        RpcInfo {
+                                            chain_name: chain_name.clone(),
+                                            rpc_url: rpc_url.clone(),
+                                            status: *status,
+                                        }
+                                    }).collect();
+                                    let _ = res_tx.send(connected_rpcs);
                                 }
                             }
-                            let _ = res.send(());
-                            break;
                         }
+                        Some(rpc_update) = rpc_update_rx.recv() => {
+                            rpc_statuses.insert(
+                                (rpc_update.chain_name.clone(), rpc_update.rpc_url.clone()),
+                                rpc_update.status,
+                            );
+                        }
+                        else => break,
                     }
                 }
 
@@ -162,6 +190,15 @@ impl ChainManager {
         rx.await.map_err(|_| ChainError::MessageDropped)?
     }
 
+    pub async fn get_connected_rpcs(&self) -> Result<Vec<RpcInfo>, Error> {
+        let (res_tx, res_rx) = oneshot::channel();
+        self.tx
+            .send(ChainRequest::GetConnectedRpcs(res_tx))
+            .await
+            .map_err(|_| Error::Fatal)?;
+        res_rx.await.map_err(|_| Error::Fatal)
+    }
+
     pub async fn reap(
         &self,
         id: String,

src/chain/tracker.rs

@@ -1,17 +1,17 @@
 //! A tracker that follows individual chain
 
-use std::{collections::HashMap, time::SystemTime};
-
 use frame_metadata::v15::RuntimeMetadataV15;
 use jsonrpsee::ws_client::{WsClient, WsClientBuilder};
 use serde_json::Value;
+use std::{collections::HashMap, time::SystemTime};
 use substrate_parser::{AsMetadata, ShortSpecs};
 use tokio::{
     sync::mpsc,
     time::{timeout, Duration},
 };
 use tokio_util::sync::CancellationToken;
 
+use crate::definitions::api_v2::{Health, RpcInfo};
 use crate::{
     chain::{
         definitions::{BlockHash, ChainTrackerRequest, Invoice},
@@ -38,6 +38,7 @@ pub fn start_chain_watch(
     signer: Signer,
     task_tracker: TaskTracker,
     cancellation_token: CancellationToken,
+    rpc_update_tx: mpsc::Sender<RpcInfo>,
 ) {
     task_tracker
         .clone()
@@ -51,9 +52,30 @@ pub fn start_chain_watch(
                 if shutdown || cancellation_token.is_cancelled() {
                     break;
                 }
+
+                let _ = rpc_update_tx.send(RpcInfo {
+                    chain_name: chain.name.clone(),
+                    rpc_url: endpoint.clone(),
+                    status: Health::Degraded,
+                }).await;
+
                 if let Ok(client) = WsClientBuilder::default().build(endpoint).await {
+                    let _ = rpc_update_tx.send(RpcInfo {
+                        chain_name: chain.name.clone(),
+                        rpc_url: endpoint.clone(),
+                        status: Health::Ok,
+                    }).await;
+
                     // prepare chain
-                    let watcher = match ChainWatcher::prepare_chain(&client, chain.clone(), &mut watched_accounts, endpoint, chain_tx.clone(), state.interface(), task_tracker.clone())
+                    let watcher = match ChainWatcher::prepare_chain(
+                        &client,
+                        chain.clone(),
+                        &mut watched_accounts,
+                        endpoint,
+                        chain_tx.clone(),
+                        state.interface(),
+                        task_tracker.clone(),
+                    )
                         .await
                     {
                         Ok(a) => a,
@@ -165,6 +187,12 @@ pub fn start_chain_watch(
                             }
                         }
                     }
+                } else {
+                    let _ = rpc_update_tx.send(RpcInfo {
+                        chain_name: chain.name.clone(),
+                        rpc_url: endpoint.clone(),
+                        status: Health::Critical,
+                    }).await;
                 }
             }
             Ok(format!("Chain {} monitor shut down", chain.name).into())

src/definitions.rs

@@ -176,6 +176,7 @@ pub mod api_v2 {
     pub enum WithdrawalStatus {
         Waiting,
         Failed,
+        Forced,
         Completed,
     }
 
@@ -185,24 +186,23 @@ pub mod api_v2 {
         pub supported_currencies: HashMap<std::string::String, CurrencyProperties>,
     }
 
-    #[allow(dead_code)] // TODO: Use this for health response?
     #[derive(Debug, Serialize)]
-    struct ServerHealth {
-        server_info: ServerInfo,
-        connected_rpcs: Vec<RpcInfo>,
-        status: Health,
+    pub struct ServerHealth {
+        pub server_info: ServerInfo,
+        pub connected_rpcs: Vec<RpcInfo>,
+        pub status: Health,
     }
 
-    #[derive(Debug, Serialize)]
-    struct RpcInfo {
-        rpc_url: String,
-        chain_name: String,
-        status: Health,
+    #[derive(Debug, Serialize, Clone)]
+    pub struct RpcInfo {
+        pub rpc_url: String,
+        pub chain_name: String,
+        pub status: Health,
     }
 
-    #[derive(Debug, Serialize)]
+    #[derive(Debug, Serialize, Clone, PartialEq, Copy)]
     #[serde(rename_all = "lowercase")]
-    enum Health {
+    pub enum Health {
         Ok,
         Degraded,
         Critical,

src/handlers/health.rs

@@ -1,4 +1,4 @@
-use crate::definitions::api_v2::ServerStatus;
+use crate::definitions::api_v2::{ServerHealth, ServerStatus};
 use crate::state::State;
 use axum::{extract::State as ExtractState, http::StatusCode, Json};
 
@@ -18,12 +18,18 @@ pub async fn status(
 }
 
 pub async fn health(
-    ExtractState(_state): ExtractState<State>,
+    ExtractState(state): ExtractState<State>,
 ) -> (
     [(axum::http::header::HeaderName, &'static str); 1],
-    Json<ServerStatus>,
+    Json<ServerHealth>,
 ) {
-    todo!();
+    match state.server_health().await {
+        Ok(status) => (
+            [(axum::http::header::CACHE_CONTROL, "no-store")],
+            Json(status),
+        ),
+        Err(_) => panic!("db connection is down, state is lost"),
+    }
 }
 
 pub async fn audit(ExtractState(_state): ExtractState<State>) -> StatusCode {