feat!: gather telemetry

Cargo.toml

@@ -1,13 +1,11 @@
 [package]
 name = "iroha_explorer"
 description = "Backend API of Iroha 2 Explorer"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2021"
 
 [dependencies]
-# FIXME: `iroha` must expose `Status` struct
-iroha_telemetry = { git = "https://github.com/hyperledger/iroha.git", rev = "22d84e7ba64df81c63b2006ced467f6be5752833" }
-iroha = { git = "https://github.com/hyperledger/iroha.git", rev = "22d84e7ba64df81c63b2006ced467f6be5752833" }
+iroha = { git = "https://github.com/hyperledger-iroha/iroha.git", rev = "3f039a3f51fdb16207a481dd2bedb8d17df86186" }
 
 tracing = "0.1.32"
 tracing-subscriber = { version = "0.3.10", features = ["env-filter"] }
@@ -29,12 +27,17 @@ sqlx = { version = "0.8.0", features = ["sqlite", "runtime-tokio", "chrono"] }
 eyre = "0.6.12"
 parity-scale-codec = "3.7.4"
 base64 = "0.22.1"
+futures-util = "0.3.30"
+async-stream = "0.3.6"
+derive_more = { version = "2.0.1", features = ["display", "from_str"] }
+circular-buffer = "1.1.0"
+reqwest = { version = "0.12.15", features = ["json"] }
+backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 
 [dev-dependencies]
 insta = { version = "1.42.2", features = ["json", "csv"] }
 expect-test = "1.5.0"
-iroha_crypto = { git = "https://github.com/hyperledger/iroha.git", rev = "22d84e7ba64df81c63b2006ced467f6be5752833", features = ["rand"] }
+iroha_crypto = { git = "https://github.com/hyperledger-iroha/iroha.git", rev = "3f039a3f51fdb16207a481dd2bedb8d17df86186", features = ["rand"] }
 tokio = { version = "1", features = ["process"] }
-reqwest = { version = "0.12.15", features = ["json"] }
 
 

src/database_update.rs

@@ -1,23 +1,26 @@
 use crate::iroha_client_wrap::ClientWrap;
 use crate::repo::{scan_iroha, Repo};
+use crate::telemetry::blockchain::State;
+use crate::telemetry::Telemetry;
 use eyre::WrapErr;
 use sqlx::sqlite::SqliteConnectOptions;
 use sqlx::ConnectOptions;
 use std::time::Duration;
-use tokio::task::spawn_blocking;
 use tokio::time::sleep;
 
 pub struct DatabaseUpdateLoop {
     repo: Repo,
     client: ClientWrap,
-    last_update_block: u64,
+    telemetry: Telemetry,
+    last_update_block: u32,
 }
 
 impl DatabaseUpdateLoop {
-    pub fn new(repo: Repo, client: ClientWrap) -> Self {
+    pub fn new(repo: Repo, client: ClientWrap, telemetry: Telemetry) -> Self {
         Self {
             repo,
             client,
+            telemetry,
             last_update_block: 0,
         }
     }
@@ -39,13 +42,13 @@ impl DatabaseUpdateLoop {
     }
 
     async fn attempt(&mut self) -> eyre::Result<()> {
-        let client_clone = self.client.clone();
-        let status = spawn_blocking(move || client_clone.get_status())
-            .await?
-            .wrap_err("Failed to fetch Iroha status")?;
+        let Some(metrics) = self.telemetry.single_peer(&self.client.torii_url()).await? else {
+            tracing::warn!("Skipping database update - peer metrics are not available");
+            return Ok(());
+        };
 
-        if status.blocks == self.last_update_block {
-            tracing::debug!("No new blocks - skipping update");
+        if metrics.block == self.last_update_block {
+            tracing::debug!("Skipping database update - no blocks difference");
             return Ok(());
         }
 
@@ -59,9 +62,18 @@ impl DatabaseUpdateLoop {
             .wrap_err("Failed to scan Iroha")?;
         self.repo.swap(conn).await;
 
-        self.last_update_block = status.blocks;
+        self.last_update_block = metrics.block;
         tracing::info!(%self.last_update_block, "Updated the database");
 
+        if let Err(err) = self.update_telemetry().await {
+            tracing::error!(?err, "Failed to update blockchain state in telemetry")
+        }
+
         Ok(())
     }
+
+    async fn update_telemetry(&self) -> eyre::Result<()> {
+        let state = State::scan(&self.repo).await?;
+        self.telemetry.update_blockchain_state(state).await
+    }
 }