GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dedup routes before submission, stagger out batches

Browse Source
This commit is contained in:
durch
2025-06-25 12:00:17 +02:00
parent 688660a7d5
commit 6fb3f23ccd
4 changed files with 40 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -6045,7 +6045,7 @@ dependencies = [
[[package]]
name = "nym-network-monitor"
version = "1.1.5"
version = "1.1.6"
dependencies = [
"anyhow",
"axum 0.7.9",
common/types/src/monitoring.rs
+1 -1
View File
@@ -11,7 +11,7 @@ static NETWORK_MONITORS: LazyLock<HashSet<String>> = LazyLock::new(|| {
nm
});
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, ToSchema)]
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, ToSchema, Eq, PartialEq, Hash)]
pub struct RouteResult {
pub layer1: u32,
pub layer2: u32,
nym-network-monitor/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-network-monitor"
version = "1.1.5"
version = "1.1.6"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
nym-network-monitor/src/accounting.rs
+37 -34
View File
@@ -546,42 +546,45 @@ pub async fn submit_metrics(database_url: Option<&String>) -> anyhow::Result<()>
.collect::<Result<Vec<_>, _>>()?;
let network_account = NetworkAccount::finalize()?;
let accounting_routes = network_account.accounting_routes;
info!("Submitting {} accounting routes", accounting_routes.len());
match accounting_routes
.chunks(10)
.map(|chunk| {
let route_results = chunk
.iter()
.map(|route| {
RouteResult::new(
route.mix_nodes.0,
route.mix_nodes.1,
route.mix_nodes.2,
route.gateway_node,
route.success,
)
})
.collect::<Vec<RouteResult>>();
let monitor_results = MonitorResults::Route(route_results);
let monitor_message = MonitorMessage::new(monitor_results, private_key);
client.post(&route_submit_url).json(&monitor_message).send()
let route_results = network_account.accounting_routes
.iter()
.map(|route| {
RouteResult::new(
route.mix_nodes.0,
route.mix_nodes.1,
route.mix_nodes.2,
route.gateway_node,
route.success,
)
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.collect::<HashSet<RouteResult>>()
.into_iter()
.collect::<Result<Vec<_>, _>>()
{
Ok(_) => info!(
"Successfully submitted accounting routes to {}",
nym_api_url
),
Err(e) => error!(
"Error submitting accounting routes to {}: {}",
nym_api_url, e
),
};
.collect::<Vec<RouteResult>>();
info!("Submitting {} accounting routes", route_results.len());
for batch in route_results.chunks(1000) {
match batch
.chunks(100)
.map(|chunk| {
let monitor_results = MonitorResults::Route(chunk.to_vec());
let monitor_message = MonitorMessage::new(monitor_results, private_key);
client.post(&route_submit_url).json(&monitor_message).send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()
{
Ok(_) => info!(
"Successfully submitted accounting routes to {}",
nym_api_url
),
Err(e) => error!(
"Error submitting accounting routes to {}: {}",
nym_api_url, e
),
};
}
}
}
} else {