logos-blockchain/logos-blockchain-testing
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-testing.git synced 2026-03-31 16:23:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rename cfgsync client and server surface

Browse Source
This commit is contained in:
andrussal 2026-03-10 11:12:01 +01:00
parent 9ebf029f2a
commit ef1d7663c5
7 changed files with 89 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
cfgsync/adapter/src/lib.rs
View File

@ -2,7 +2,7 @@ use std::{collections::HashMap, error::Error, sync::Mutex};
use cfgsync_artifacts::ArtifactFile;
use cfgsync_core::{
CfgSyncErrorResponse, ConfigResolveResponse, NodeArtifactsPayload, NodeConfigSource,
CfgsyncErrorResponse, ConfigResolveResponse, NodeArtifactsPayload, NodeConfigSource,
NodeRegistration, RegisterNodeResponse,
};
use serde::{Deserialize, Serialize};
@ -259,7 +259,7 @@ where
let registration = match self.registration_for(&registration.identifier) {
Some(registration) => registration,
None => {
return ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(
return ConfigResolveResponse::Error(CfgsyncErrorResponse::not_ready(
&registration.identifier,
));
}
@ -269,12 +269,12 @@ where
let catalog = match self.materializer.materialize_snapshot(&registrations) {
Ok(Some(catalog)) => catalog,
Ok(None) => {
return ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(
return ConfigResolveResponse::Error(CfgsyncErrorResponse::not_ready(
&registration.identifier,
));
}
Err(error) => {
return ConfigResolveResponse::Error(CfgSyncErrorResponse::internal(format!(
return ConfigResolveResponse::Error(CfgsyncErrorResponse::internal(format!(
"failed to materialize config snapshot: {error}"
)));
}
@ -284,7 +284,7 @@ where
Some(config) => ConfigResolveResponse::Config(NodeArtifactsPayload::from_files(
config.files.clone(),
)),
None => ConfigResolveResponse::Error(CfgSyncErrorResponse::missing_config(
None => ConfigResolveResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
)),
}
@ -309,7 +309,7 @@ where
let registration = match self.registration_for(&registration.identifier) {
Some(registration) => registration,
None => {
return ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(
return ConfigResolveResponse::Error(CfgsyncErrorResponse::not_ready(
&registration.identifier,
));
}
@ -320,10 +320,10 @@ where
Ok(Some(artifacts)) => ConfigResolveResponse::Config(NodeArtifactsPayload::from_files(
artifacts.files().to_vec(),
)),
Ok(None) => ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(
Ok(None) => ConfigResolveResponse::Error(CfgsyncErrorResponse::not_ready(
&registration.identifier,
)),
Err(error) => ConfigResolveResponse::Error(CfgSyncErrorResponse::internal(format!(
Err(error) => ConfigResolveResponse::Error(CfgsyncErrorResponse::internal(format!(
"failed to materialize config for host {}: {error}",
registration.identifier
))),
@ -484,7 +484,7 @@ mod tests {
use cfgsync_artifacts::ArtifactFile;
use cfgsync_core::{
CfgSyncErrorCode, ConfigResolveResponse, NodeConfigSource, NodeRegistration,
CfgsyncErrorCode, ConfigResolveResponse, NodeConfigSource, NodeRegistration,
};
use super::{
@ -535,7 +535,7 @@ mod tests {
match provider.resolve(&registration) {
ConfigResolveResponse::Config(_) => panic!("expected not-ready error"),
ConfigResolveResponse::Error(error) => {
assert!(matches!(error.code, CfgSyncErrorCode::NotReady))
assert!(matches!(error.code, CfgsyncErrorCode::NotReady))
}
}
}
@ -579,7 +579,7 @@ mod tests {
match provider.resolve(&node_a) {
ConfigResolveResponse::Config(_) => panic!("expected not-ready error"),
ConfigResolveResponse::Error(error) => {
assert!(matches!(error.code, CfgSyncErrorCode::NotReady))
assert!(matches!(error.code, CfgsyncErrorCode::NotReady))
}
}

29
cfgsync/core/src/client.rs
View File

@ -1,7 +1,7 @@
use serde::Serialize;
use thiserror::Error;
use crate::repo::{CfgSyncErrorResponse, NodeArtifactsPayload, NodeRegistration};
use crate::repo::{CfgsyncErrorCode, CfgsyncErrorResponse, NodeArtifactsPayload, NodeRegistration};
/// cfgsync client-side request/response failures.
#[derive(Debug, Error)]
@ -12,7 +12,7 @@ pub enum ClientError {
Status {
status: reqwest::StatusCode,
message: String,
error: Option<CfgSyncErrorResponse>,
error: Option<CfgsyncErrorResponse>,
},
#[error("failed to parse cfgsync response: {0}")]
Decode(serde_json::Error),
@ -22,16 +22,17 @@ pub enum ClientError {
pub enum ConfigFetchStatus {
Ready,
NotReady,
Missing,
}
/// Reusable HTTP client for cfgsync server endpoints.
#[derive(Clone, Debug)]
pub struct CfgSyncClient {
pub struct CfgsyncClient {
base_url: String,
http: reqwest::Client,
}
impl CfgSyncClient {
impl CfgsyncClient {
#[must_use]
pub fn new(base_url: impl Into<String>) -> Self {
let mut base_url = base_url.into();
@ -80,10 +81,15 @@ impl CfgSyncClient {
status,
error: Some(error),
..
}) if status == reqwest::StatusCode::TOO_EARLY => {
let _ = error;
Ok(ConfigFetchStatus::NotReady)
}
}) => match error.code {
CfgsyncErrorCode::NotReady => Ok(ConfigFetchStatus::NotReady),
CfgsyncErrorCode::MissingConfig => Ok(ConfigFetchStatus::Missing),
CfgsyncErrorCode::Internal => Err(ClientError::Status {
status,
message: error.message.clone(),
error: Some(error),
}),
},
Err(error) => Err(error),
}
}
@ -100,7 +106,7 @@ impl CfgSyncClient {
let status = response.status();
let body = response.text().await?;
if !status.is_success() {
let error = serde_json::from_str::<CfgSyncErrorResponse>(&body).ok();
let error = serde_json::from_str::<CfgsyncErrorResponse>(&body).ok();
let message = error
.as_ref()
.map(|err| err.message.clone())
@ -126,7 +132,7 @@ impl CfgSyncClient {
let status = response.status();
let body = response.text().await?;
if !status.is_success() {
let error = serde_json::from_str::<CfgSyncErrorResponse>(&body).ok();
let error = serde_json::from_str::<CfgsyncErrorResponse>(&body).ok();
let message = error
.as_ref()
.map(|err| err.message.clone())
@ -149,3 +155,6 @@ impl CfgSyncClient {
}
}
}
#[doc(hidden)]
pub type CfgSyncClient = CfgsyncClient;

16
cfgsync/core/src/lib.rs
View File

@ -7,23 +7,27 @@ pub mod server;
#[doc(hidden)]
pub use bundle::{CfgSyncBundle, CfgSyncBundleNode};
pub use bundle::{NodeArtifactsBundle, NodeArtifactsBundleEntry};
pub use client::{CfgSyncClient, ClientError, ConfigFetchStatus};
#[doc(hidden)]
pub use client::CfgSyncClient;
pub use client::{CfgsyncClient, ClientError, ConfigFetchStatus};
pub use render::{
CfgsyncConfigOverrides, CfgsyncOutputPaths, RenderedCfgsync, apply_cfgsync_overrides,
apply_timeout_floor, ensure_bundle_path, load_cfgsync_template_yaml,
render_cfgsync_yaml_from_template, write_rendered_cfgsync,
};
pub use repo::{
BundleConfigSource, BundleConfigSourceError, CFGSYNC_SCHEMA_VERSION, CfgSyncErrorCode,
CfgSyncErrorResponse, ConfigResolveResponse, NodeArtifactFile, NodeArtifactsPayload,
BundleConfigSource, BundleConfigSourceError, CFGSYNC_SCHEMA_VERSION, CfgsyncErrorCode,
CfgsyncErrorResponse, ConfigResolveResponse, NodeArtifactFile, NodeArtifactsPayload,
NodeConfigSource, NodeRegistration, RegisterNodeResponse, RegistrationPayload,
StaticConfigSource,
};
#[doc(hidden)]
pub use repo::{
CfgSyncFile, CfgSyncPayload, ConfigProvider, ConfigRepo, FileConfigProvider,
FileConfigProviderError, RegistrationResponse, RepoResponse,
CfgSyncErrorCode, CfgSyncErrorResponse, CfgSyncFile, CfgSyncPayload, ConfigProvider,
ConfigRepo, FileConfigProvider, FileConfigProviderError, RegistrationResponse, RepoResponse,
};
#[doc(hidden)]
pub use server::CfgSyncState;
pub use server::{CfgsyncServerState, RunCfgsyncError, cfgsync_app, run_cfgsync};
pub use server::{CfgsyncServerState, RunCfgsyncError, build_cfgsync_router, serve_cfgsync};
#[doc(hidden)]
pub use server::{cfgsync_app, run_cfgsync};

30
cfgsync/core/src/repo.rs
View File

@ -159,7 +159,7 @@ impl NodeArtifactsPayload {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CfgSyncErrorCode {
pub enum CfgsyncErrorCode {
MissingConfig,
NotReady,
Internal,
@ -168,16 +168,16 @@ pub enum CfgSyncErrorCode {
/// Structured error body returned by cfgsync server.
#[derive(Debug, Clone, Serialize, Deserialize, Error)]
#[error("{code:?}: {message}")]
pub struct CfgSyncErrorResponse {
pub code: CfgSyncErrorCode,
pub struct CfgsyncErrorResponse {
pub code: CfgsyncErrorCode,
pub message: String,
}
impl CfgSyncErrorResponse {
impl CfgsyncErrorResponse {
#[must_use]
pub fn missing_config(identifier: &str) -> Self {
Self {
code: CfgSyncErrorCode::MissingConfig,
code: CfgsyncErrorCode::MissingConfig,
message: format!("missing config for host {identifier}"),
}
}
@ -185,7 +185,7 @@ impl CfgSyncErrorResponse {
#[must_use]
pub fn not_ready(identifier: &str) -> Self {
Self {
code: CfgSyncErrorCode::NotReady,
code: CfgsyncErrorCode::NotReady,
message: format!("config for host {identifier} is not ready"),
}
}
@ -193,7 +193,7 @@ impl CfgSyncErrorResponse {
#[must_use]
pub fn internal(message: impl Into<String>) -> Self {
Self {
code: CfgSyncErrorCode::Internal,
code: CfgsyncErrorCode::Internal,
message: message.into(),
}
}
@ -202,13 +202,13 @@ impl CfgSyncErrorResponse {
/// Resolution outcome for a requested node identifier.
pub enum ConfigResolveResponse {
Config(NodeArtifactsPayload),
Error(CfgSyncErrorResponse),
Error(CfgsyncErrorResponse),
}
/// Outcome for a node registration request.
pub enum RegisterNodeResponse {
Registered,
Error(CfgSyncErrorResponse),
Error(CfgsyncErrorResponse),
}
/// Source of cfgsync node payloads.
@ -235,7 +235,7 @@ impl NodeConfigSource for StaticConfigSource {
if self.configs.contains_key(&registration.identifier) {
RegisterNodeResponse::Registered
} else {
RegisterNodeResponse::Error(CfgSyncErrorResponse::missing_config(
RegisterNodeResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
))
}
@ -247,7 +247,7 @@ impl NodeConfigSource for StaticConfigSource {
.cloned()
.map_or_else(
|| {
ConfigResolveResponse::Error(CfgSyncErrorResponse::missing_config(
ConfigResolveResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
))
},
@ -372,7 +372,7 @@ mod tests {
)) {
ConfigResolveResponse::Config(_) => panic!("expected missing-config error"),
ConfigResolveResponse::Error(error) => {
assert!(matches!(error.code, CfgSyncErrorCode::MissingConfig));
assert!(matches!(error.code, CfgsyncErrorCode::MissingConfig));
assert!(error.message.contains("unknown-node"));
}
}
@ -515,3 +515,9 @@ pub type CfgSyncFile = NodeArtifactFile;
#[doc(hidden)]
pub type CfgSyncPayload = NodeArtifactsPayload;
#[doc(hidden)]
pub type CfgSyncErrorCode = CfgsyncErrorCode;
#[doc(hidden)]
pub type CfgSyncErrorResponse = CfgsyncErrorResponse;

37
cfgsync/core/src/server.rs
View File

@ -4,7 +4,7 @@ use axum::{Json, Router, extract::State, http::StatusCode, response::IntoRespons
use thiserror::Error;
use crate::repo::{
CfgSyncErrorCode, ConfigResolveResponse, NodeConfigSource, NodeRegistration,
CfgsyncErrorCode, ConfigResolveResponse, NodeConfigSource, NodeRegistration,
RegisterNodeResponse,
};
@ -75,15 +75,15 @@ fn resolve_node_config_response(
state.repo.resolve(registration)
}
fn error_status(code: &CfgSyncErrorCode) -> StatusCode {
fn error_status(code: &CfgsyncErrorCode) -> StatusCode {
match code {
CfgSyncErrorCode::MissingConfig => StatusCode::NOT_FOUND,
CfgSyncErrorCode::NotReady => StatusCode::TOO_EARLY,
CfgSyncErrorCode::Internal => StatusCode::INTERNAL_SERVER_ERROR,
CfgsyncErrorCode::MissingConfig => StatusCode::NOT_FOUND,
CfgsyncErrorCode::NotReady => StatusCode::TOO_EARLY,
CfgsyncErrorCode::Internal => StatusCode::INTERNAL_SERVER_ERROR,
}
}
pub fn cfgsync_app(state: CfgsyncServerState) -> Router {
pub fn build_cfgsync_router(state: CfgsyncServerState) -> Router {
Router::new()
.route("/register", post(register_node))
.route("/node", post(node_config))
@ -92,8 +92,8 @@ pub fn cfgsync_app(state: CfgsyncServerState) -> Router {
}
/// Runs cfgsync HTTP server on the provided port until shutdown/error.
pub async fn run_cfgsync(port: u16, state: CfgsyncServerState) -> Result<(), RunCfgsyncError> {
let app = cfgsync_app(state);
pub async fn serve_cfgsync(port: u16, state: CfgsyncServerState) -> Result<(), RunCfgsyncError> {
let app = build_cfgsync_router(state);
println!("Server running on http://0.0.0.0:{port}");
let bind_addr = format!("0.0.0.0:{port}");
@ -111,6 +111,11 @@ pub async fn run_cfgsync(port: u16, state: CfgsyncServerState) -> Result<(), Run
#[doc(hidden)]
pub type CfgSyncState = CfgsyncServerState;
#[doc(hidden)]
pub use build_cfgsync_router as cfgsync_app;
#[doc(hidden)]
pub use serve_cfgsync as run_cfgsync;
#[cfg(test)]
mod tests {
use std::{collections::HashMap, sync::Arc};
@ -119,7 +124,7 @@ mod tests {
use super::{CfgsyncServerState, NodeRegistration, node_config, register_node};
use crate::repo::{
CFGSYNC_SCHEMA_VERSION, CfgSyncErrorCode, CfgSyncErrorResponse, ConfigResolveResponse,
CFGSYNC_SCHEMA_VERSION, CfgsyncErrorCode, CfgsyncErrorResponse, ConfigResolveResponse,
NodeArtifactFile, NodeArtifactsPayload, NodeConfigSource, RegisterNodeResponse,
};
@ -132,7 +137,7 @@ mod tests {
if self.data.contains_key(&registration.identifier) {
RegisterNodeResponse::Registered
} else {
RegisterNodeResponse::Error(CfgSyncErrorResponse::missing_config(
RegisterNodeResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
))
}
@ -144,7 +149,7 @@ mod tests {
.cloned()
.map_or_else(
|| {
ConfigResolveResponse::Error(CfgSyncErrorResponse::missing_config(
ConfigResolveResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
))
},
@ -161,7 +166,7 @@ mod tests {
impl NodeConfigSource for RegistrationAwareProvider {
fn register(&self, registration: NodeRegistration) -> RegisterNodeResponse {
if !self.data.contains_key(&registration.identifier) {
return RegisterNodeResponse::Error(CfgSyncErrorResponse::missing_config(
return RegisterNodeResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
));
}
@ -182,7 +187,7 @@ mod tests {
.expect("test registration store should not be poisoned");
if !registrations.contains_key(&registration.identifier) {
return ConfigResolveResponse::Error(CfgSyncErrorResponse::not_ready(
return ConfigResolveResponse::Error(CfgsyncErrorResponse::not_ready(
&registration.identifier,
));
}
@ -192,7 +197,7 @@ mod tests {
.cloned()
.map_or_else(
|| {
ConfigResolveResponse::Error(CfgSyncErrorResponse::missing_config(
ConfigResolveResponse::Error(CfgsyncErrorResponse::missing_config(
&registration.identifier,
))
},
@ -248,9 +253,9 @@ mod tests {
#[test]
fn missing_config_error_uses_expected_code() {
let error = CfgSyncErrorResponse::missing_config("missing-node");
let error = CfgsyncErrorResponse::missing_config("missing-node");
assert!(matches!(error.code, CfgSyncErrorCode::MissingConfig));
assert!(matches!(error.code, CfgsyncErrorCode::MissingConfig));
}
#[tokio::test]

14
cfgsync/runtime/src/client.rs
View File

@ -6,7 +6,7 @@ use std::{
use anyhow::{Context as _, Result, bail};
use cfgsync_core::{
CFGSYNC_SCHEMA_VERSION, CfgSyncClient, NodeArtifactFile, NodeArtifactsPayload,
CFGSYNC_SCHEMA_VERSION, CfgsyncClient, NodeArtifactFile, NodeArtifactsPayload,
NodeRegistration, RegistrationPayload,
};
use thiserror::Error;
@ -26,7 +26,7 @@ async fn fetch_with_retry(
payload: &NodeRegistration,
server_addr: &str,
) -> Result<NodeArtifactsPayload> {
let client = CfgSyncClient::new(server_addr);
let client = CfgsyncClient::new(server_addr);
for attempt in 1..=FETCH_ATTEMPTS {
match fetch_once(&client, payload).await {
@ -47,7 +47,7 @@ async fn fetch_with_retry(
}
async fn fetch_once(
client: &CfgSyncClient,
client: &CfgsyncClient,
payload: &NodeRegistration,
) -> Result<NodeArtifactsPayload> {
let response = client.fetch_node_config(payload).await?;
@ -75,7 +75,7 @@ async fn pull_config_files(payload: NodeRegistration, server_addr: &str) -> Resu
}
async fn register_node(payload: &NodeRegistration, server_addr: &str) -> Result<()> {
let client = CfgSyncClient::new(server_addr);
let client = CfgsyncClient::new(server_addr);
for attempt in 1..=FETCH_ATTEMPTS {
match client.register_node(payload).await {
@ -188,7 +188,7 @@ mod tests {
use cfgsync_core::{
CfgsyncServerState, NodeArtifactsBundle, NodeArtifactsBundleEntry, NodeArtifactsPayload,
StaticConfigSource, run_cfgsync,
StaticConfigSource, serve_cfgsync,
};
use tempfile::tempdir;
@ -213,7 +213,9 @@ mod tests {
let port = allocate_test_port();
let address = format!("http://127.0.0.1:{port}");
let server = tokio::spawn(async move {
run_cfgsync(port, state).await.expect("run cfgsync server");
serve_cfgsync(port, state)
.await
.expect("run cfgsync server");
});
pull_config_files(

4
cfgsync/runtime/src/server.rs
View File

@ -3,7 +3,7 @@ use std::{fs, path::Path, sync::Arc};
use anyhow::Context as _;
use cfgsync_adapter::{NodeArtifacts, NodeArtifactsCatalog, RegistrationConfigProvider};
use cfgsync_core::{
BundleConfigSource, CfgsyncServerState, NodeArtifactsBundle, NodeConfigSource, run_cfgsync,
BundleConfigSource, CfgsyncServerState, NodeArtifactsBundle, NodeConfigSource, serve_cfgsync,
};
use serde::Deserialize;
use thiserror::Error;
@ -159,7 +159,7 @@ pub async fn run_cfgsync_server(config_path: &Path) -> anyhow::Result<()> {
let bundle_path = resolve_bundle_path(config_path, &config.bundle_path);
let state = build_server_state(&config, &bundle_path)?;
run_cfgsync(config.port, state).await?;
serve_cfgsync(config.port, state).await?;
Ok(())
}