diff --git a/node/rpc/src/admin/api.rs b/node/rpc/src/admin/api.rs index d26962c..2ea96c4 100644 --- a/node/rpc/src/admin/api.rs +++ b/node/rpc/src/admin/api.rs @@ -2,7 +2,7 @@ use crate::types::{LocationInfo, NetworkInfo, PeerInfo}; use jsonrpsee::core::RpcResult; use jsonrpsee::proc_macros::rpc; use std::collections::HashMap; -use sync::FileSyncInfo; +use sync::{FileSyncInfo, SyncServiceState}; #[rpc(server, client, namespace = "admin")] pub trait Rpc { @@ -27,6 +27,9 @@ pub trait Rpc { #[method(name = "terminateSync")] async fn terminate_sync(&self, tx_seq: u64) -> RpcResult; + #[method(name = "getSyncServiceState")] + async fn get_sync_service_state(&self) -> RpcResult; + #[method(name = "getSyncStatus")] async fn get_sync_status(&self, tx_seq: u64) -> RpcResult; diff --git a/node/rpc/src/admin/impl.rs b/node/rpc/src/admin/impl.rs index db6b0ff..cd62779 100644 --- a/node/rpc/src/admin/impl.rs +++ b/node/rpc/src/admin/impl.rs @@ -8,7 +8,7 @@ use network::{multiaddr::Protocol, Multiaddr}; use std::collections::HashMap; use std::net::IpAddr; use storage::config::all_shards_available; -use sync::{FileSyncInfo, SyncRequest, SyncResponse}; +use sync::{FileSyncInfo, SyncRequest, SyncResponse, SyncServiceState}; use task_executor::ShutdownReason; pub struct RpcServerImpl { @@ -119,6 +119,17 @@ impl RpcServer for RpcServerImpl { } } + async fn get_sync_service_state(&self) -> RpcResult { + info!("admin_getSyncServiceState()"); + + let response = self.ctx.request_sync(SyncRequest::SyncState).await?; + + match response { + SyncResponse::SyncState { state } => Ok(state), + _ => Err(error::internal_error("unexpected response type")), + } + } + #[tracing::instrument(skip(self), err)] async fn get_sync_status(&self, tx_seq: u64) -> RpcResult { info!("admin_getSyncStatus({tx_seq})"); diff --git a/node/sync/src/lib.rs b/node/sync/src/lib.rs index f42b0f6..7ffad90 100644 --- a/node/sync/src/lib.rs +++ b/node/sync/src/lib.rs @@ -7,9 +7,10 @@ mod controllers; mod service; pub mod test_util; +use auto_sync::{batcher_random::RandomBatcherState, batcher_serial::SerialBatcherState}; pub use controllers::FileSyncInfo; use duration_str::deserialize_duration; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; pub use service::{SyncMessage, SyncReceiver, SyncRequest, SyncResponse, SyncSender, SyncService}; use std::{ fmt::Debug, @@ -64,3 +65,11 @@ impl InstantWrapper { self.0.elapsed() } } + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncServiceState { + pub num_syncing: usize, + pub auto_sync_serial: Option, + pub auto_sync_random: Option, +} diff --git a/node/sync/src/service.rs b/node/sync/src/service.rs index dfd84e8..f38c827 100644 --- a/node/sync/src/service.rs +++ b/node/sync/src/service.rs @@ -4,7 +4,7 @@ use crate::controllers::{ FailureReason, FileSyncGoal, FileSyncInfo, SerialSyncController, SyncState, MAX_CHUNKS_TO_REQUEST, }; -use crate::Config; +use crate::{Config, SyncServiceState}; use anyhow::{bail, Result}; use file_location_cache::FileLocationCache; use libp2p::swarm::DialError; @@ -74,6 +74,7 @@ pub enum SyncMessage { #[derive(Debug)] pub enum SyncRequest { + SyncState, SyncStatus { tx_seq: u64, }, @@ -99,6 +100,7 @@ pub enum SyncRequest { #[derive(Debug)] pub enum SyncResponse { + SyncState { state: SyncServiceState }, SyncStatus { status: Option }, SyncFile { err: String }, FileSyncInfo { result: HashMap }, @@ -278,6 +280,23 @@ impl SyncService { sender: channel::ResponseSender, ) { match req { + SyncRequest::SyncState => { + let state = match &self.auto_sync_manager { + Some(manager) => SyncServiceState { + num_syncing: self.controllers.len(), + auto_sync_serial: Some(manager.serial.get_state().await), + auto_sync_random: manager.random.get_state().await.ok(), + }, + None => SyncServiceState { + num_syncing: self.controllers.len(), + auto_sync_serial: None, + auto_sync_random: None, + }, + }; + + let _ = sender.send(SyncResponse::SyncState { state }); + } + SyncRequest::SyncStatus { tx_seq } => { let status = self .controllers