Watch
0
0
Fork
You've already forked hyperhive
0

hive-c0re/swarm-*: drop clippy allows and dedupe unix_now + KV open_or_create

No behaviour change. Casts replaced by try_from / Duration math; schedules.rs
reuses agent_renewal::unix_now; the five swarm-queue-client KV open-or-create
copies share one private kv::open_or_create.
This commit is contained in:
atlas 2026-10-09 09:48:03 +02:00 • committed by mara
commit b874d21cda
11 changed files with 74 additions and 133 deletions

View file

@ -222,16 +222,8 @@ pub async fn gather() -> Vec<ContainerResource> {
let t0: Vec<Option<u64>> = candidates.iter().map(|(_, d)| read_usage_usec(d)).collect(); let t0: Vec<Option<u64>> = candidates.iter().map(|(_, d)| read_usage_usec(d)).collect();
sleep(CPU_SAMPLE).await; sleep(CPU_SAMPLE).await;
#[allow( let nproc = f64::from(u32::try_from(host_nproc()).unwrap_or(u32::MAX));
clippy::cast_precision_loss, let interval_secs = CPU_SAMPLE.as_secs_f64();
reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost"
)]
let nproc = host_nproc() as f64;
#[allow(
clippy::cast_precision_loss,
reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost"
)]
let interval_usec = CPU_SAMPLE.as_micros() as f64;
// Snapshot the slow disk cache once (cheap clone of a small map) so // Snapshot the slow disk cache once (cheap clone of a small map) so
// the per-agent loop below is a plain map lookup — no `du` on this // the per-agent loop below is a plain map lookup — no `du` on this
@ -246,12 +238,8 @@ pub async fn gather() -> Vec<ContainerResource> {
let usage_now = read_usage_usec(dir); let usage_now = read_usage_usec(dir);
let cpu_pct = match (t0[i], usage_now) { let cpu_pct = match (t0[i], usage_now) {
(Some(a), Some(b)) => { (Some(a), Some(b)) => {
#[allow( let delta_secs = Duration::from_micros(b.saturating_sub(a)).as_secs_f64();
clippy::cast_precision_loss, (delta_secs / (interval_secs * nproc)) * 100.0
reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost"
)]
let delta = b.saturating_sub(a) as f64;
(delta / (interval_usec * nproc)) * 100.0
} }
_ => 0.0, _ => 0.0,
}; };

View file

@ -236,13 +236,8 @@ struct AgentAgg {
skills: HashMap<String, u64>, skills: HashMap<String, u64>,
} }
#[allow(
clippy::cast_sign_loss,
clippy::cast_possible_truncation,
reason = "v.max(0) clamps to a non-negative value first, so the i64->u64 cast is exact: no sign loss, and no truncation since u64 covers all non-negative i64"
)]
fn u64_from_i64(v: i64) -> u64 { fn u64_from_i64(v: i64) -> u64 {
v.max(0) as u64 u64::try_from(v).unwrap_or(0)
} }
/// Aggregate one agent's turn-stats over `[from, now]`. Errors bubble /// Aggregate one agent's turn-stats over `[from, now]`. Errors bubble

View file

@ -154,8 +154,7 @@ fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc<Mutex<Vec<Con
if let Ok(g) = snap.lock() { if let Ok(g) = snap.lock() {
for c in g.iter() { for c in g.iter() {
if let Some(usec) = c.cpu_time_usec { if let Some(usec) = c.cpu_time_usec {
#[allow(clippy::cast_precision_loss)] obs.observe(Duration::from_micros(usec).as_secs_f64(), &attrs(c));
obs.observe(usec as f64 / 1_000_000.0, &attrs(c));
} }
} }
} }

View file

@ -30,6 +30,7 @@ use serde::{Deserialize, Deserializer, Serialize};
use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule}; use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule};
use utoipa::ToSchema; use utoipa::ToSchema;
use crate::agent_renewal::unix_now;
use crate::wanted::WantedWriter; use crate::wanted::WantedWriter;
use crate::{AppState, error_problem}; use crate::{AppState, error_problem};
@ -723,12 +724,6 @@ pub(crate) async fn remove_schedule(
Ok(StatusCode::NO_CONTENT) Ok(StatusCode::NO_CONTENT)
} }
fn unix_now() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;

View file

@ -62,27 +62,13 @@ pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, Error> { ) -> Result<async_nats::jetstream::kv::Store, Error> {
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await { crate::kv::open_or_create(
Ok(store) => Ok(store), &js,
Err(e) => { BUCKET,
tracing::info!( "agent-icon",
bucket = BUCKET, "Current icon published by each agent".to_owned(),
reason = %e, )
"agent-icon bucket not available, creating it" .await
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: BUCKET.to_owned(),
description: "Current icon published by each agent".to_owned(),
history: 1,
..Default::default()
})
.await
.map_err(|source| Error::CreateBucket {
bucket: BUCKET.to_owned(),
source,
})
}
}
} }
#[cfg(test)] #[cfg(test)]

View file

@ -97,28 +97,13 @@ pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, crate::Error> { ) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await { crate::kv::open_or_create(
Ok(store) => Ok(store), &js,
Err(e) => { BUCKET,
tracing::info!( "agent-schedule",
bucket = BUCKET, "Scheduled prompts the swarm controller declares for each agent".to_owned(),
reason = %e, )
"agent-schedule bucket not available, creating it" .await
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: BUCKET.to_owned(),
description: "Scheduled prompts the swarm controller declares for each agent"
.to_owned(),
history: 1,
..Default::default()
})
.await
.map_err(|source| crate::Error::CreateBucket {
bucket: BUCKET.to_owned(),
source,
})
}
}
} }
/// What reading one agent's key found. /// What reading one agent's key found.

View file

@ -109,27 +109,13 @@ pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, Error> { ) -> Result<async_nats::jetstream::kv::Store, Error> {
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await { crate::kv::open_or_create(
Ok(store) => Ok(store), &js,
Err(e) => { BUCKET,
tracing::info!( "agent-status",
bucket = BUCKET, "Last status snapshot offered by each agent".to_owned(),
reason = %e, )
"agent-status bucket not available, creating it" .await
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: BUCKET.to_owned(),
description: "Last status snapshot offered by each agent".to_owned(),
history: 1,
..Default::default()
})
.await
.map_err(|source| Error::CreateBucket {
bucket: BUCKET.to_owned(),
source,
})
}
}
} }
#[cfg(test)] #[cfg(test)]

View file

@ -0,0 +1,34 @@
//! The open-or-create sequence every KV bucket module shares.
use crate::Error;
/// Open `bucket`, creating it with `history: 1` and `description` when it
/// does not exist yet. `label` names the bucket in the creation log line.
pub(crate) async fn open_or_create(
js: &async_nats::jetstream::Context,
bucket: &str,
label: &str,
description: String,
) -> Result<async_nats::jetstream::kv::Store, Error> {
match js.get_key_value(bucket).await {
Ok(store) => Ok(store),
Err(e) => {
tracing::info!(
bucket,
reason = %e,
"{label} bucket not available, creating it"
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: bucket.to_owned(),
description,
history: 1,
..Default::default()
})
.await
.map_err(|source| Error::CreateBucket {
bucket: bucket.to_owned(),
source,
})
}
}
}

View file

@ -201,6 +201,9 @@ pub mod agent_icon;
/// writing needs `kv` and reading needs `agent-schedules`. /// writing needs `kv` and reading needs `agent-schedules`.
pub mod agent_schedules; pub mod agent_schedules;
#[cfg(feature = "kv")]
mod kv;
/// The subject the swarm controller publishes on when the hive-wide knowledge /// The subject the swarm controller publishes on when the hive-wide knowledge
/// repository has changed. One writer, many readers — every hive subscribes. /// repository has changed. One writer, many readers — every hive subscribes.
/// ///

View file

@ -51,25 +51,11 @@ pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, Error> { ) -> Result<async_nats::jetstream::kv::Store, Error> {
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await { crate::kv::open_or_create(
Ok(store) => Ok(store), &js,
Err(e) => { BUCKET,
tracing::info!( "status",
bucket = BUCKET, "Last status snapshot offered by each hive".to_owned(),
reason = %e, )
"status bucket not available, creating it" .await
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: BUCKET.to_owned(),
description: "Last status snapshot offered by each hive".to_owned(),
history: 1,
..Default::default()
})
.await
.map_err(|source| Error::CreateBucket {
bucket: BUCKET.to_owned(),
source,
})
}
}
} }

View file

@ -125,24 +125,8 @@ pub async fn open_or_create(
) -> Result<async_nats::jetstream::kv::Store, Error> { ) -> Result<async_nats::jetstream::kv::Store, Error> {
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
let bucket = bucket(hive); let bucket = bucket(hive);
match js.get_key_value(&bucket).await { let description = format!("Agent set the swarm controller declares for hive {hive}");
Ok(store) => Ok(store), crate::kv::open_or_create(&js, &bucket, "wanted-state", description).await
Err(e) => {
tracing::info!(
%bucket,
reason = %e,
"wanted-state bucket not available, creating it"
);
js.create_key_value(async_nats::jetstream::kv::Config {
bucket: bucket.clone(),
description: format!("Agent set the swarm controller declares for hive {hive}"),
history: 1,
..Default::default()
})
.await
.map_err(|source| Error::CreateBucket { bucket, source })
}
}
} }
/// Open the wanted-state bucket for reading, or `None` when it does not exist. /// Open the wanted-state bucket for reading, or `None` when it does not exist.