gk-replay: add "/workers" to dump the full list

standalone/replay/Cargo.toml

@@ -8,6 +8,7 @@ phala-mq = { path = "../../crates/phala-mq" }
 phala-types = { path = "../../crates/phala-types" }
 phala-trie-storage = { path = "../../crates/phala-trie-storage" }
 phactory = { path = "../../crates/phactory" }
+phactory-api = { path = "../../crates/phactory/api" }
 pherry = { path = "../pherry" }
 sp-core = { path = "../../substrate/primitives/core", default-features = false }
 

standalone/replay/src/replay_gk/data_persist.rs

@@ -1,12 +1,11 @@
 use super::EventRecord;
+use anyhow::Result;
 use chrono::TimeZone as _;
+use phactory::gk;
 use sqlx::types::Decimal;
 use sqlx::{postgres::PgPoolOptions, Row};
 use std::time::Duration;
 use tokio::sync::mpsc;
-use anyhow::Result;
-use phactory::gk;
-
 
 pub(super) async fn run_persist(mut rx: mpsc::Receiver<EventRecord>, uri: &str) {
     log::info!("Connecting to {}", uri);
@@ -56,10 +55,7 @@ pub(super) async fn run_persist(mut rx: mpsc::Receiver<EventRecord>, uri: &str)
                             Ok(last_sequence) => {
                                 log::info!("last_sequence={}", last_sequence);
                                 if last_sequence
-                                    >= records
-                                        .last()
-                                        .expect("records can not be empty")
-                                        .sequence
+                                    >= records.last().expect("records can not be empty").sequence
                                 {
                                     log::info!("Insert succeeded, let's move on");
                                     break;
@@ -84,10 +80,7 @@ pub(super) async fn run_persist(mut rx: mpsc::Receiver<EventRecord>, uri: &str)
     }
 }
 
-async fn insert_records(
-    pool: &sqlx::Pool<sqlx::Postgres>,
-    records: &[EventRecord],
-) -> Result<()> {
+async fn insert_records(pool: &sqlx::Pool<sqlx::Postgres>, records: &[EventRecord]) -> Result<()> {
     // Current version of sqlx does not support bulk insertion, so we have to do it manually.
     let mut sequences = vec![];
     let mut pubkeys = vec![];
@@ -147,10 +140,9 @@ fn cvt_fp(v: gk::FixedPoint) -> Decimal {
 }
 
 async fn get_last_sequence(pool: &sqlx::Pool<sqlx::Postgres>) -> Result<i64> {
-    let latest_row = sqlx::query(
-        "SELECT sequence FROM worker_finance_events ORDER BY sequence DESC LIMIT 1",
-    )
-    .fetch_optional(pool)
-    .await?;
+    let latest_row =
+        sqlx::query("SELECT sequence FROM worker_finance_events ORDER BY sequence DESC LIMIT 1")
+            .fetch_optional(pool)
+            .await?;
     Ok(latest_row.map_or(0, |row| row.get(0)))
 }

standalone/replay/src/replay_gk/httpserver.rs

@@ -2,24 +2,44 @@ use std::str::FromStr;
 
 use super::*;
 use actix_web::{get, web, App, HttpResponse, HttpServer};
+use phactory_api::prpc as pb;
 use subxt::sp_runtime::AccountId32;
 
 struct AppState {
     factory: Arc<Mutex<ReplayFactory>>,
 }
 
 #[get("/meminfo")]
-async fn meminfo(
-    data: web::Data<AppState>,
-) -> HttpResponse {
+async fn meminfo(data: web::Data<AppState>) -> HttpResponse {
     let factory = data.factory.lock().await;
-    let size = factory.storage.pairs(&[]).iter().map(|(k, v)| k.len() + v.len()).sum::<usize>();
+    let size = factory
+        .storage
+        .pairs(&[])
+        .iter()
+        .map(|(k, v)| k.len() + v.len())
+        .sum::<usize>();
     log::info!("Storage size: {}", size);
     HttpResponse::Ok().json(serde_json::json!({
             "storage_size": size,
     }))
 }
 
+fn serialize_worker_state(state: &pb::WorkerState) -> serde_json::Value {
+    serde_json::json!({
+        "benchmarking": state.bench_state.is_some(),
+        "mining": state.mining_state.is_some(),
+        "unresponsive": state.unresponsive,
+        "last_heartbeat_for_block": state.last_heartbeat_for_block,
+        "last_heartbeat_at_block": state.last_heartbeat_at_block,
+        "waiting_heartbeats": state.waiting_heartbeats,
+        "v": state.tokenomic_info.as_ref().map(|info| info.v.clone()),
+        "v_init": state.tokenomic_info.as_ref().map(|info| info.v_init.clone()),
+        "p_instant": state.tokenomic_info.as_ref().map(|info| info.p_instant.clone()),
+        "p_init": state.tokenomic_info.as_ref().map(|info| info.p_bench.clone()),
+        "tokenomic_info": format!("{:#?}", state.tokenomic_info),
+    })
+}
+
 #[get("/worker-state/{pubkey}")]
 async fn get_worker_state(
     web::Path(pubkey): web::Path<String>,
@@ -43,28 +63,41 @@ async fn get_worker_state(
         Some(state) => HttpResponse::Ok().json(serde_json::json!({
             "current_block": factory.current_block,
             "total_share": total_share.to_string(),
-            "benchmarking": state.bench_state.is_some(),
-            "mining": state.mining_state.is_some(),
-            "unresponsive": state.unresponsive,
-            "last_heartbeat_for_block": state.last_heartbeat_for_block,
-            "last_heartbeat_at_block": state.last_heartbeat_at_block,
-            "waiting_heartbeats": state.waiting_heartbeats,
-            "v": state.tokenomic_info.as_ref().map(|info| info.v.clone()),
-            "v_init": state.tokenomic_info.as_ref().map(|info| info.v_init.clone()),
-            "p_instant": state.tokenomic_info.as_ref().map(|info| info.p_instant.clone()),
-            "p_init": state.tokenomic_info.as_ref().map(|info| info.p_bench.clone()),
-            "tokenomic_info": format!("{:#?}", state.tokenomic_info),
+            "worker": serialize_worker_state(&state),
         })),
     }
 }
 
+#[get("/workers")]
+async fn dump_workers(data: web::Data<AppState>) -> HttpResponse {
+    let factory = data.factory.lock().await;
+
+    let total_share = factory.gk.sum_share();
+    let workers = factory.gk.dump_workers_state();
+    let workers: std::collections::BTreeMap<_, _> = workers
+        .iter()
+        .map(|(k, v)| {
+            (
+                "0x".to_string() + &hex::encode(&k),
+                serialize_worker_state(&v),
+            )
+        })
+        .collect();
+    HttpResponse::Ok().json(serde_json::json!({
+        "current_block": factory.current_block,
+        "total_share": total_share.to_string(),
+        "workers": workers
+    }))
+}
+
 pub async fn serve(bind_addr: String, factory: Arc<Mutex<ReplayFactory>>) {
     HttpServer::new(move || {
         let factory = factory.clone();
         App::new()
             .data(AppState { factory })
             .service(get_worker_state)
             .service(meminfo)
+            .service(dump_workers)
     })
     .disable_signals()
     .bind(&bind_addr)