Support liquidated vaults in metrics (#559)

* support liquidated vaults for metrics

* add method to fetch only active vaults ids from manager, only accept requests if not liquidated

* add metric for liquidated

* clippy fixes

* simplify liquidation status metric updater

* remove comment

* remove unused method

clients/vault/src/metrics.rs

@@ -9,8 +9,10 @@ use lazy_static::lazy_static;
 use primitives::{stellar, Asset, DecimalsLookup};
 use runtime::{
 	prometheus::{
-		gather, proto::MetricFamily, Encoder, Gauge, GaugeVec, IntCounter, IntGaugeVec, Opts,
-		Registry, TextEncoder,
+		core::{AtomicI64, GenericGauge},
+		gather,
+		proto::MetricFamily,
+		Encoder, Gauge, GaugeVec, IntCounter, IntGaugeVec, Opts, Registry, TextEncoder,
 	},
 	types::currency_id::CurrencyIdExt,
 	AggregateUpdatedEvent, CollateralBalancesPallet, CurrencyId, Error as RuntimeError, FixedU128,
@@ -83,6 +85,11 @@ lazy_static! {
 	pub static ref RESTART_COUNT: IntCounter =
 		IntCounter::new("restart_count", "Number of service restarts")
 			.expect("Failed to create prometheus metric");
+	pub static ref LIQUIDATED: IntGaugeVec = IntGaugeVec::new(
+		Opts::new("liquidated", "Boolean reporting if the vault is currently liquidated"),
+		&[CURRENCY_LABEL]
+	)
+	.expect("Failed to create prometheus metric");
 }
 const STELLAR_NATIVE_ASSET_TYPE: [u8; 6] = *b"native";
 
@@ -108,6 +115,7 @@ pub struct PerCurrencyMetrics {
 	asset_balance: XLMBalance,
 	issues: RequestCounter,
 	redeems: RequestCounter,
+	liquidated: GenericGauge<AtomicI64>,
 }
 
 #[async_trait]
@@ -117,8 +125,9 @@ pub trait VaultDataReader {
 
 #[async_trait]
 impl VaultDataReader for VaultIdManager {
+	// get_all_entries fetches from active and liquidated vaults
 	async fn get_entries(&self) -> Vec<VaultData> {
-		self.get_entries().await
+		self.get_all_entries().await
 	}
 }
 
@@ -231,6 +240,7 @@ impl PerCurrencyMetrics {
 				completed_count: REDEEMS.with(&request_type_label("completed")),
 				expired_count: REDEEMS.with(&request_type_label("expired")),
 			},
+			liquidated: LIQUIDATED.with(&labels),
 		}
 	}
 
@@ -242,6 +252,7 @@ impl PerCurrencyMetrics {
 			publish_locked_collateral(vault, parachain_rpc.clone()),
 			publish_required_collateral(vault, parachain_rpc.clone()),
 			publish_collateralization(vault, parachain_rpc.clone()),
+			update_liquidation_status(vault)
 		);
 	}
 }
@@ -259,6 +270,7 @@ pub fn register_custom_metrics() -> Result<(), RuntimeError> {
 	REGISTRY.register(Box::new(MEAN_POLL_DURATION.clone()))?;
 	REGISTRY.register(Box::new(MEAN_SCHEDULED_DURATION.clone()))?;
 	REGISTRY.register(Box::new(RESTART_COUNT.clone()))?;
+	REGISTRY.register(Box::new(LIQUIDATED.clone()))?;
 
 	Ok(())
 }
@@ -481,6 +493,11 @@ async fn publish_redeem_count<V: VaultDataReader>(
 	}
 }
 
+async fn update_liquidation_status(vault: &VaultData) {
+	let liquidated_flag: i64 = if vault.liquidated { 1 } else { 0 };
+	vault.metrics.liquidated.set(liquidated_flag);
+}
+
 pub async fn monitor_bridge_metrics(
 	parachain_rpc: SpacewalkParachain,
 	vault_id_manager: VaultIdManager,
@@ -499,6 +516,7 @@ pub async fn monitor_bridge_metrics(
 						.iter()
 						.filter(|vault| vault.vault_id.collateral_currency() == **currency_id)
 					{
+						let _ = update_liquidation_status(vault).await;
 						let _ = publish_locked_collateral(vault, parachain_rpc.clone()).await;
 						let _ = publish_required_collateral(vault, parachain_rpc.clone()).await;
 						publish_collateralization(vault, parachain_rpc.clone()).await;

clients/vault/src/system.rs

@@ -52,6 +52,7 @@ pub struct VaultData {
 	pub vault_id: VaultId,
 	pub stellar_wallet: ArcRwLock<StellarWallet>,
 	pub metrics: PerCurrencyMetrics,
+	pub liquidated: bool,
 }
 
 #[derive(Clone)]
@@ -88,19 +89,21 @@ impl VaultIdManager {
 						vault_id: key.clone(),
 						stellar_wallet: stellar_wallet.clone(),
 						metrics: PerCurrencyMetrics::dummy(),
+						liquidated: false,
 					},
 				)
 			})
 			.collect();
 		Self { vault_data: Arc::new(RwLock::new(vault_data)), spacewalk_parachain, stellar_wallet }
 	}
 
-	async fn add_vault_id(&self, vault_id: VaultId) -> Result<(), Error> {
+	async fn add_vault_id(&self, vault_id: VaultId, is_liquidated: bool) -> Result<(), Error> {
 		let metrics = PerCurrencyMetrics::new(&vault_id);
 		let data = VaultData {
 			vault_id: vault_id.clone(),
 			stellar_wallet: self.stellar_wallet.clone(),
 			metrics,
+			liquidated: is_liquidated,
 		};
 		PerCurrencyMetrics::initialize_values(self.spacewalk_parachain.clone(), &data).await;
 
@@ -119,11 +122,14 @@ impl VaultIdManager {
 		{
 			// check if vault is registered
 			match self.spacewalk_parachain.get_vault(&vault_id).await {
-				Ok(_) => self.add_vault_id(vault_id.clone()).await?,
-				Err(RuntimeError::VaultLiquidated) => tracing::error!(
-					"[{}] Vault is liquidated -- not going to process events for this vault.",
-					vault_id.pretty_print()
-				),
+				Ok(_) => self.add_vault_id(vault_id.clone(), false).await?,
+				Err(RuntimeError::VaultLiquidated) => {
+					self.add_vault_id(vault_id.clone(), true).await?;
+					tracing::error!(
+						"[{}] Vault is liquidated -- not going to process events for this vault.",
+						vault_id.pretty_print()
+					);
+				},
 				Err(e) => return Err(e.into()),
 			}
 		}
@@ -138,7 +144,7 @@ impl VaultIdManager {
 					let vault_id = event.vault_id;
 					if self.spacewalk_parachain.is_this_vault(&vault_id) {
 						tracing::info!("New vault registered: {}", vault_id.pretty_print());
-						let _ = self.add_vault_id(vault_id).await;
+						let _ = self.add_vault_id(vault_id, false).await;
 					}
 				},
 				|err| tracing::error!("Error (RegisterVaultEvent): {}", err.to_string()),
@@ -150,11 +156,32 @@ impl VaultIdManager {
 		self.vault_data.read().await.get(vault_id).map(|x| x.stellar_wallet.clone())
 	}
 
+	pub async fn get_active_vault(&self, vault_id: &VaultId) -> Option<VaultData> {
+		let vault = self.vault_data.read().await.get(vault_id)?.clone();
+		// Filter liquidated
+		if vault.liquidated {
+			return None;
+		}
+		return Some(vault);
+	}
+
 	pub async fn get_vault(&self, vault_id: &VaultId) -> Option<VaultData> {
 		self.vault_data.read().await.get(vault_id).cloned()
 	}
 
+	// Get all ACTIVE vaults
 	pub async fn get_entries(&self) -> Vec<VaultData> {
+		self.vault_data
+			.read()
+			.await
+			.iter()
+			.filter(|(_, value)| !value.liquidated)
+			.map(|(_, value)| value.clone())
+			.collect()
+	}
+
+	// Get all vaults including liquidated ones.
+	pub async fn get_all_entries(&self) -> Vec<VaultData> {
 		self.vault_data.read().await.iter().map(|(_, value)| value.clone()).collect()
 	}
 
@@ -163,6 +190,7 @@ impl VaultIdManager {
 			.read()
 			.await
 			.iter()
+			.filter(|(_, value)| !value.liquidated)
 			.map(|(vault_id, _)| vault_id.clone())
 			.collect()
 	}
@@ -174,6 +202,7 @@ impl VaultIdManager {
 			.read()
 			.await
 			.iter()
+			.filter(|(_, value)| !value.liquidated)
 			.map(|(vault_id, data)| (vault_id.clone(), data.stellar_wallet.clone()))
 			.collect()
 	}

pallets/stellar-relay/src/validation.rs

@@ -92,7 +92,7 @@ fn is_node_id_exist<T: Config>(
 			envelope.statement.slot_index,
 			envelope.statement.node_id
 		);
-		return None
+		return None;
 	}
 
 	Some(node_id)
@@ -174,7 +174,7 @@ pub fn validate_envelopes<'a, T: Config>(
 	for envelope in envelopes {
 		let Some(node_id) = is_node_id_exist::<T>(envelope, validators) else {
 			// ignore this envelope; continue to the next ones
-			continue
+			continue;
 		};
 
 		// Check if all envelopes are using the same slot index