Skip to content

Commit ffb43ff

Browse files
mattdfmattdf
authored
Merge branch 'develop' into fix/remove-hardcoded-nvml-path
2 parents 908c94d + e57aa1d commit ffb43ff

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

73 files changed

+728
-931
lines changed

‎crates/discovery/src/api/server.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -85,12 +85,12 @@ pub async fn start_server(
8585
max_nodes_per_ip: u32,
8686
chain_sync_enabled: bool,
8787
) -> std::io::Result<()> {
88-
info!("Starting server at http://{}:{}", host, port);
88+
info!("Starting server at http://{host}:{port}");
8989

9090
let validators = match contracts.prime_network.get_validator_role().await {
9191
Ok(validators) => validators,
9292
Err(e) => {
93-
error!("❌ Failed to get validator role: {}", e);
93+
error!("❌ Failed to get validator role: {e}");
9494
std::process::exit(1);
9595
}
9696
};
@@ -108,15 +108,15 @@ pub async fn start_server(
108108
.with_redis(redis_store.client.clone())
109109
.await
110110
.map_err(|e| {
111-
std::io::Error::other(format!("Failed to initialize Redis connection pool: {}", e))
111+
std::io::Error::other(format!("Failed to initialize Redis connection pool: {e}"))
112112
})?,
113113
);
114114
let validate_signatures = Arc::new(
115115
ValidatorState::new(vec![])
116116
.with_redis(redis_store.client.clone())
117117
.await
118118
.map_err(|e| {
119-
std::io::Error::other(format!("Failed to initialize Redis connection pool: {}", e))
119+
std::io::Error::other(format!("Failed to initialize Redis connection pool: {e}"))
120120
})?
121121
.with_validator(move |_| true),
122122
);

‎crates/discovery/src/chainsync/sync.rs‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -66,8 +66,7 @@ impl ChainSync {
6666
.await
6767
.map_err(|e| {
6868
error!(
69-
"Error retrieving node info for provider {} and node {}: {}",
70-
provider_address, node_address, e
69+
"Error retrieving node info for provider {provider_address} and node {node_address}: {e}"
7170
);
7271
anyhow::anyhow!("Failed to retrieve node info")
7372
})?;
@@ -77,10 +76,7 @@ impl ChainSync {
7776
.get_provider(provider_address)
7877
.await
7978
.map_err(|e| {
80-
error!(
81-
"Error retrieving provider info for {}: {}",
82-
provider_address, e
83-
);
79+
error!("Error retrieving provider info for {provider_address}: {e}");
8480
anyhow::anyhow!("Failed to retrieve provider info")
8581
})?;
8682

@@ -151,7 +147,7 @@ impl ChainSync {
151147
match nodes {
152148
Ok(nodes) => {
153149
let total_nodes = nodes.len();
154-
info!("Syncing {} nodes", total_nodes);
150+
info!("Syncing {total_nodes} nodes");
155151

156152
// Process nodes in parallel with concurrency limit
157153
let results: Vec<Result<(), Error>> = stream::iter(nodes)
@@ -174,7 +170,7 @@ impl ChainSync {
174170
Ok(_) => success_count += 1,
175171
Err(e) => {
176172
failure_count += 1;
177-
warn!("Node sync failed: {}", e);
173+
warn!("Node sync failed: {e}");
178174
}
179175
}
180176
}
@@ -196,7 +192,7 @@ impl ChainSync {
196192
);
197193
}
198194
Err(e) => {
199-
error!("Error getting nodes from store: {}", e);
195+
error!("Error getting nodes from store: {e}");
200196
}
201197
}
202198
}

‎crates/discovery/src/location_enrichment.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ impl LocationEnrichmentService {
3838
interval.tick().await;
3939

4040
if let Err(e) = self.enrich_nodes_without_location().await {
41-
error!("Location enrichment cycle failed: {}", e);
41+
error!("Location enrichment cycle failed: {e}");
4242
}
4343
}
4444
}

‎crates/discovery/src/main.rs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,8 +36,7 @@ impl std::str::FromStr for ServiceMode {
3636
"processor" => Ok(ServiceMode::Processor),
3737
"full" => Ok(ServiceMode::Full),
3838
_ => Err(format!(
39-
"Invalid mode: {}. Use 'api', 'processor', or 'full'",
40-
s
39+
"Invalid mode: {s}. Use 'api', 'processor', or 'full'"
4140
)),
4241
}
4342
}
@@ -137,7 +136,7 @@ async fn main() -> Result<()> {
137136
info!("Starting location enrichment service");
138137
tokio::spawn(async move {
139138
if let Err(e) = location_enrichment.run(30).await {
140-
error!("Location enrichment service failed: {}", e);
139+
error!("Location enrichment service failed: {e}");
141140
}
142141
});
143142
}
@@ -155,7 +154,7 @@ async fn main() -> Result<()> {
155154
)
156155
.await
157156
{
158-
error!("❌ Failed to start server: {}", err);
157+
error!("❌ Failed to start server: {err}");
159158
}
160159

161160
tokio::signal::ctrl_c().await?;
@@ -175,7 +174,7 @@ async fn main() -> Result<()> {
175174
)
176175
.await
177176
{
178-
error!("❌ Failed to start server: {}", err);
177+
error!("❌ Failed to start server: {err}");
179178
}
180179
}
181180
}

‎crates/discovery/src/store/node_store.rs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ impl NodeStore {
2121
}
2222

2323
pub async fn get_node(&self, address: String) -> Result<Option<DiscoveryNode>, Error> {
24-
let key = format!("node:{}", address);
24+
let key = format!("node:{address}");
2525
let mut con = self.get_connection().await?;
2626
let node: Option<String> = con.get(&key).await?;
2727
let node = match node {
@@ -39,7 +39,7 @@ impl NodeStore {
3939
return Ok(None);
4040
}
4141

42-
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{}", id)).collect();
42+
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{id}")).collect();
4343
let serialized_nodes: Vec<String> =
4444
redis::pipe().get(&node_keys).query_async(&mut con).await?;
4545

@@ -60,7 +60,7 @@ impl NodeStore {
6060
return Ok(0);
6161
}
6262

63-
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{}", id)).collect();
63+
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{id}")).collect();
6464

6565
let mut count = 0;
6666
for key in node_keys {
@@ -77,7 +77,7 @@ impl NodeStore {
7777

7878
pub async fn register_node(&self, node: Node) -> Result<(), Error> {
7979
let address = node.id.clone();
80-
let key = format!("node:{}", address);
80+
let key = format!("node:{address}");
8181

8282
let mut con = self.get_connection().await?;
8383

@@ -104,7 +104,7 @@ impl NodeStore {
104104
pub async fn update_node(&self, node: DiscoveryNode) -> Result<(), Error> {
105105
let mut con = self.get_connection().await?;
106106
let address = node.id.clone();
107-
let key = format!("node:{}", address);
107+
let key = format!("node:{address}");
108108
let serialized_node = serde_json::to_string(&node)?;
109109

110110
let _: () = redis::pipe()
@@ -125,7 +125,7 @@ impl NodeStore {
125125
return Ok(Vec::new());
126126
}
127127

128-
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{}", id)).collect();
128+
let node_keys: Vec<String> = node_ids.iter().map(|id| format!("node:{id}")).collect();
129129

130130
let mut pipe = redis::pipe();
131131
for key in &node_keys {
@@ -138,7 +138,7 @@ impl NodeStore {
138138
let serialized_nodes = match serialized_nodes {
139139
Ok(nodes) => nodes,
140140
Err(e) => {
141-
error!("Error querying nodes from Redis: {}", e);
141+
error!("Error querying nodes from Redis: {e}");
142142
return Err(e.into());
143143
}
144144
};
@@ -159,7 +159,7 @@ impl NodeStore {
159159

160160
pub async fn get_node_by_id(&self, node_id: &str) -> Result<Option<DiscoveryNode>, Error> {
161161
let mut con = self.get_connection().await?;
162-
let key = format!("node:{}", node_id);
162+
let key = format!("node:{node_id}");
163163

164164
let serialized_node: Option<String> = con.get(&key).await?;
165165

‎crates/discovery/src/store/redis.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,15 +22,15 @@ impl RedisStore {
2222
pub fn new(redis_url: &str) -> Self {
2323
match Client::open(redis_url) {
2424
Ok(client) => {
25-
info!("Successfully connected to Redis at {}", redis_url);
25+
info!("Successfully connected to Redis at {redis_url}");
2626
Self {
2727
client,
2828
#[cfg(test)]
2929
server: Arc::new(RedisServer::new()),
3030
}
3131
}
3232
Err(e) => {
33-
panic!("Redis connection error: {}", e);
33+
panic!("Redis connection error: {e}");
3434
}
3535
}
3636
}

‎crates/orchestrator/src/api/routes/groups.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,7 @@ async fn fetch_node_logs_p2p(
226226
{
227227
Ok(node) => node,
228228
Err(e) => {
229-
error!("Failed to get node {}: {}", node_address, e);
229+
error!("Failed to get node {node_address}: {e}");
230230
return json!({
231231
"success": false,
232232
"error": format!("Failed to get node: {}", e)
@@ -244,7 +244,7 @@ async fn fetch_node_logs_p2p(
244244
match (&node.worker_p2p_id, &node.worker_p2p_addresses) {
245245
(Some(p2p_id), Some(p2p_addrs)) if !p2p_addrs.is_empty() => (p2p_id, p2p_addrs),
246246
_ => {
247-
error!("Node {} does not have P2P information", node_address);
247+
error!("Node {node_address} does not have P2P information");
248248
return json!({
249249
"success": false,
250250
"error": "Node does not have P2P information",
@@ -268,15 +268,15 @@ async fn fetch_node_logs_p2p(
268268
})
269269
}
270270
Ok(Err(e)) => {
271-
error!("P2P request failed for node {}: {}", node_address, e);
271+
error!("P2P request failed for node {node_address}: {e}");
272272
json!({
273273
"success": false,
274274
"error": format!("P2P request failed: {}", e),
275275
"status": node.status.to_string()
276276
})
277277
}
278278
Err(_) => {
279-
error!("P2P request timed out for node {}", node_address);
279+
error!("P2P request timed out for node {node_address}");
280280
json!({
281281
"success": false,
282282
"error": "P2P request timed out",
@@ -286,7 +286,7 @@ async fn fetch_node_logs_p2p(
286286
}
287287
}
288288
None => {
289-
error!("Node {} not found in orchestrator", node_address);
289+
error!("Node {node_address} not found in orchestrator");
290290
json!({
291291
"success": false,
292292
"error": "Node not found in orchestrator",

‎crates/orchestrator/src/api/routes/heartbeat.rs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ async fn heartbeat(
6565
)
6666
.await
6767
{
68-
error!("Error updating node task: {}", e);
68+
error!("Error updating node task: {e}");
6969
}
7070

7171
// Record task state metric if task information is available
@@ -82,7 +82,7 @@ async fn heartbeat(
8282
.update_node_p2p_id(&node_address, p2p_id)
8383
.await
8484
{
85-
error!("Error updating node p2p id: {}", e);
85+
error!("Error updating node p2p id: {e}");
8686
}
8787
}
8888

@@ -92,7 +92,7 @@ async fn heartbeat(
9292
.beat(&heartbeat)
9393
.await
9494
{
95-
error!("Heartbeat Error: {}", e);
95+
error!("Heartbeat Error: {e}");
9696
}
9797
if let Some(metrics) = heartbeat.metrics.clone() {
9898
// Get current metric keys for this node efficiently using HKEYS
@@ -104,7 +104,7 @@ async fn heartbeat(
104104
{
105105
Ok(keys) => keys,
106106
Err(e) => {
107-
error!("Error getting metric keys for node: {}", e);
107+
error!("Error getting metric keys for node: {e}");
108108
Vec::new()
109109
}
110110
};
@@ -137,7 +137,7 @@ async fn heartbeat(
137137
.delete_metric(task_id, label, &node_address.to_string())
138138
.await
139139
{
140-
error!("Error deleting metric: {}", e);
140+
error!("Error deleting metric: {e}");
141141
}
142142
}
143143
}
@@ -149,7 +149,7 @@ async fn heartbeat(
149149
.store_metrics(Some(metrics.clone()), node_address)
150150
.await
151151
{
152-
error!("Error storing metrics: {}", e);
152+
error!("Error storing metrics: {e}");
153153
}
154154
}
155155

‎crates/orchestrator/src/api/routes/metrics.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ async fn get_metrics(app_state: Data<AppState>) -> HttpResponse {
3838
{
3939
Ok(metrics) => metrics,
4040
Err(e) => {
41-
error!("Error getting aggregate metrics for all tasks: {}", e);
41+
error!("Error getting aggregate metrics for all tasks: {e}");
4242
Default::default()
4343
}
4444
};
@@ -63,7 +63,7 @@ async fn get_all_metrics(app_state: Data<AppState>) -> HttpResponse {
6363
{
6464
Ok(metrics) => metrics,
6565
Err(e) => {
66-
error!("Error getting all metrics: {}", e);
66+
error!("Error getting all metrics: {e}");
6767
Default::default()
6868
}
6969
};
@@ -112,7 +112,7 @@ async fn create_metric(
112112
.store_manual_metrics(metric.label.clone(), metric.value)
113113
.await
114114
{
115-
error!("Error storing manual metric: {}", e);
115+
error!("Error storing manual metric: {e}");
116116
}
117117
HttpResponse::Ok().json(json!({"success": true}))
118118
}
@@ -143,7 +143,7 @@ async fn delete_metric(
143143
{
144144
Ok(success) => success,
145145
Err(e) => {
146-
error!("Error deleting metric: {}", e);
146+
error!("Error deleting metric: {e}");
147147
false
148148
}
149149
};

0 commit comments

Comments
 (0)