fix(#999): resolve all clippy warnings across the workspace
All crates now pass `cargo clippy --workspace -- -D warnings` cleanly.
Fixes span six crates (hive-sh4re, hive-ag3nt, hive-c0re, hive-forge,
hive-priv, hive-matrix-mcp was already clean):
- doc_markdown: wrap snake_case, type names, constants in backticks
- collapsible_if / collapsible_match: fold nested ifs into let-chains
- duration_suboptimal_units: Duration::from_secs(N) → from_mins/from_hours
- implicit_hasher: allow on HashMap-param fns where generalization is risky
- items_after_statements: hoist use to function tops
- map(f).unwrap_or(x) → map_or(x, f); map(f).unwrap_or_else(g) → map_or_else
- is_ok_and / is_none_or in place of map().unwrap_or(bool)
- needless_continue: {} instead of continue in loop match arms
- match_same_arms: Ok(None) | Err(_) merged
- format_push_str: write!() instead of push_str(&format!())
- while let replaces loop { let Some(..) = x else { break } }
- struct_excessive_bools / dead_code: allow on purpose-built structs
- too_many_lines / too_many_arguments: allow where refactor not worth it
- unused_async: remove async from poll_once in bash_runner
- needless_borrow: fix &repo deref in hive-forge comments verb
- cast_possible_truncation: allow u64→usize in fetch_tail
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
9efe9ca175
commit
5c5ca38fe8
32 changed files with 121 additions and 127 deletions
|
|
@ -229,7 +229,7 @@ async fn run_loop(socket: PathBuf) {
|
|||
let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
|
||||
|
||||
loop {
|
||||
poll_once(&socket, &claimed).await;
|
||||
poll_once(&socket, &claimed);
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
|
|
@ -258,7 +258,7 @@ async fn mark_interrupted(socket: &Path) {
|
|||
}
|
||||
}
|
||||
|
||||
async fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
|
||||
fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
|
||||
let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return };
|
||||
for entry in rd.flatten() {
|
||||
let path = entry.path();
|
||||
|
|
|
|||
|
|
@ -117,7 +117,7 @@ async fn main() -> Result<()> {
|
|||
/// Surface a `SYSTEM_SENDER` message in the live event bus + tracing
|
||||
/// log. Both agents and the manager receive `QuestionAnswered`,
|
||||
/// `ContainerCrash`, reparent notifications, and friends; the parse
|
||||
/// + log path is identical. Quiet no-op when `from` isn't
|
||||
/// and log path is identical. Quiet no-op when `from` isn't
|
||||
/// `SYSTEM_SENDER`.
|
||||
fn log_system_event(bus: &Bus, from: &str, body: &str) {
|
||||
if from != SYSTEM_SENDER {
|
||||
|
|
@ -165,7 +165,7 @@ fn consume_continue_sentinel() -> bool {
|
|||
|
||||
/// What a `Recv` long-poll returned. Decoupled from the per-role
|
||||
/// Response enum so `serve_loop` can pattern-match without seeing
|
||||
/// either AgentResponse or ManagerResponse directly.
|
||||
/// either `AgentResponse` or `ManagerResponse` directly.
|
||||
enum RecvOutcome {
|
||||
/// Long-poll returned at least one message; first one is detached.
|
||||
Message(hive_sh4re::DeliveredMessage),
|
||||
|
|
@ -212,7 +212,7 @@ trait Surface {
|
|||
fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;
|
||||
|
||||
/// Fire a `Wake { from: "self", body: "continue" }` at our own
|
||||
/// inbox — the request_next_turn sentinel pickup.
|
||||
/// inbox — the `request_next_turn` sentinel pickup.
|
||||
fn self_wake(socket: &Path) -> impl Future<Output = ()>;
|
||||
|
||||
/// Long-poll the broker for the next message. Wraps the
|
||||
|
|
|
|||
|
|
@ -80,12 +80,12 @@ fn harness_json_path() -> PathBuf {
|
|||
|
||||
fn read_harness_state() -> (bool, bool) {
|
||||
// Try the new consolidated file first.
|
||||
if let Ok(raw) = std::fs::read_to_string(harness_json_path()) {
|
||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) {
|
||||
let rate_limited = v.get("rate_limited").and_then(|x| x.as_bool()).unwrap_or(false);
|
||||
let needs_login = v.get("needs_login").and_then(|x| x.as_bool()).unwrap_or(false);
|
||||
return (rate_limited, needs_login);
|
||||
}
|
||||
if let Ok(raw) = std::fs::read_to_string(harness_json_path())
|
||||
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw)
|
||||
{
|
||||
let rate_limited = v.get("rate_limited").and_then(serde_json::Value::as_bool).unwrap_or(false);
|
||||
let needs_login = v.get("needs_login").and_then(serde_json::Value::as_bool).unwrap_or(false);
|
||||
return (rate_limited, needs_login);
|
||||
}
|
||||
// Fall back to legacy sentinel files written by older harness builds.
|
||||
let state_dir = crate::paths::state_dir();
|
||||
|
|
|
|||
|
|
@ -88,8 +88,7 @@ pub async fn run(socket: PathBuf) {
|
|||
// HIVE_FORGE_KEEP_SUBSCRIPTIONS=1 disables auto-unsubscribe for agents
|
||||
// that intentionally consume the full repo notification firehose (e.g. triage).
|
||||
let keep_subscriptions = std::env::var("HIVE_FORGE_KEEP_SUBSCRIPTIONS")
|
||||
.map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
|
||||
.unwrap_or(false);
|
||||
.is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"));
|
||||
|
||||
// Optional reason drop-list — comma-separated Forgejo `reason`
|
||||
// values to silently mark-read instead of deliver. See
|
||||
|
|
@ -159,7 +158,7 @@ fn notif_type_label(t: &str) -> &str {
|
|||
/// inside the forge-notify wrapper, so a leading `## title` line
|
||||
/// doesn't blow into an h2 in the dashboard render. See
|
||||
/// `docs/forge.md::Body excerpt + truncation + heading escape` for
|
||||
/// the strict-ATX-vs-`#tag` rationale and the split_inclusive
|
||||
/// the strict-ATX-vs-`#tag` rationale and the `split_inclusive`
|
||||
/// trailing-newline contract.
|
||||
fn escape_md_headings(body: &str) -> String {
|
||||
let mut out = String::with_capacity(body.len());
|
||||
|
|
|
|||
|
|
@ -448,20 +448,20 @@ fn format_bash_status(id: &str) -> String {
|
|||
let age = crate::serve_common::now_unix() - t;
|
||||
let _ = write!(out, ", running for {age}s");
|
||||
}
|
||||
if let Some(t) = task.completed_at {
|
||||
if let Some(s) = task.started_at {
|
||||
let _ = write!(out, ", took {}s", t - s);
|
||||
}
|
||||
if let Some(t) = task.completed_at
|
||||
&& let Some(s) = task.started_at
|
||||
{
|
||||
let _ = write!(out, ", took {}s", t - s);
|
||||
}
|
||||
if let Some(ref stdout) = task.stdout_tail {
|
||||
if !stdout.trim().is_empty() {
|
||||
let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim());
|
||||
}
|
||||
if let Some(ref stdout) = task.stdout_tail
|
||||
&& !stdout.trim().is_empty()
|
||||
{
|
||||
let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim());
|
||||
}
|
||||
if let Some(ref stderr) = task.stderr_tail {
|
||||
if !stderr.trim().is_empty() {
|
||||
let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim());
|
||||
}
|
||||
if let Some(ref stderr) = task.stderr_tail
|
||||
&& !stderr.trim().is_empty()
|
||||
{
|
||||
let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim());
|
||||
}
|
||||
out
|
||||
}
|
||||
|
|
@ -1913,14 +1913,14 @@ pub enum Flavor {
|
|||
}
|
||||
|
||||
/// Env var written by the meta renderer with a comma-separated list of
|
||||
/// `hive_sh4re::ToolGroup` snake_case names (e.g. `"messaging,inbox,meta"`).
|
||||
/// `hive_sh4re::ToolGroup` `snake_case` names (e.g. `"messaging,inbox,meta"`).
|
||||
/// When present, the harness expands the groups into per-tool allow entries
|
||||
/// instead of using the hardcoded flavor default. See `docs/conventions.md::Tool groups`.
|
||||
const TOOL_GROUPS_ENV: &str = "HIVE_TOOL_GROUPS";
|
||||
|
||||
/// `HIVE_CAPABILITIES` env var injected by `meta::render_flake` when the
|
||||
/// operator grants capabilities to this agent. Comma-separated
|
||||
/// `hive_sh4re::Capability` snake_case names. Absent = no extra capabilities.
|
||||
/// `hive_sh4re::Capability` `snake_case` names. Absent = no extra capabilities.
|
||||
const CAPABILITIES_ENV: &str = "HIVE_CAPABILITIES";
|
||||
|
||||
/// Returns the MCP tool names (without `mcp__hyperhive__` prefix) that are
|
||||
|
|
@ -1976,14 +1976,12 @@ fn effective_tool_groups(flavor: Flavor) -> Vec<hive_sh4re::ToolGroup> {
|
|||
for token in raw.split(',') {
|
||||
let t = token.trim().to_ascii_lowercase();
|
||||
// Parse via serde_json (the canonical deserialization path).
|
||||
match serde_json::from_value::<hive_sh4re::ToolGroup>(
|
||||
if let Ok(g) = serde_json::from_value::<hive_sh4re::ToolGroup>(
|
||||
serde_json::Value::String(t.clone()),
|
||||
) {
|
||||
Ok(g) => groups.push(g),
|
||||
Err(_) => tracing::warn!(
|
||||
token = %t,
|
||||
"{TOOL_GROUPS_ENV}: unknown tool group, skipping"
|
||||
),
|
||||
groups.push(g);
|
||||
} else {
|
||||
tracing::warn!(token = %t, "{TOOL_GROUPS_ENV}: unknown tool group, skipping");
|
||||
}
|
||||
}
|
||||
if groups.is_empty() {
|
||||
|
|
|
|||
|
|
@ -423,8 +423,7 @@ fn maybe_auto_reset(bus: &Bus) {
|
|||
// Compute idle seconds using the same clock as now_unix (unix epoch, i64).
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
.map_or(0, |d| d.as_secs());
|
||||
let idle_secs = now.saturating_sub(u64::try_from(last_ended).unwrap_or(0));
|
||||
let ttl = cache_ttl_secs();
|
||||
if idle_secs < ttl {
|
||||
|
|
@ -482,7 +481,7 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
|
|||
/// snapshot advances (mtime OR file-count change), avoiding the
|
||||
/// infinite-401 loop a bare-existence check would produce when stale
|
||||
/// credentials are already on disk. Mtime-snapshot resumption rationale
|
||||
/// + DirSnapshot two-axis design: see
|
||||
/// and `DirSnapshot` two-axis design: see
|
||||
/// [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md).
|
||||
///
|
||||
/// # Panics
|
||||
|
|
|
|||
|
|
@ -173,6 +173,7 @@ pub async fn serve(
|
|||
/// Marker-gating + the gateway-side consumer: see
|
||||
/// [`docs/gateway.md::Per-agent unix-socket upstream`](../../../docs/gateway.md).
|
||||
fn bind_unix(path: &Path) -> Result<tokio::net::UnixListener> {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent)
|
||||
.with_context(|| format!("create socket parent dir {}", parent.display()))?;
|
||||
|
|
@ -183,7 +184,6 @@ fn bind_unix(path: &Path) -> Result<tokio::net::UnixListener> {
|
|||
let _ = std::fs::remove_file(path);
|
||||
let listener = tokio::net::UnixListener::bind(path)
|
||||
.with_context(|| format!("bind unix socket at {}", path.display()))?;
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
|
||||
.with_context(|| format!("set perms on {}", path.display()))?;
|
||||
// Best-effort ready marker: failed write isn't fatal (the harness
|
||||
|
|
@ -319,11 +319,10 @@ async fn relay_ws_vnc(socket: axum::extract::ws::WebSocket, vnc_port: u16) {
|
|||
let ws_to_tcp = tokio::spawn(async move {
|
||||
while let Some(Ok(msg)) = futures_util::StreamExt::next(&mut ws_rx).await {
|
||||
match msg {
|
||||
Message::Binary(data) => {
|
||||
if tcp_tx.write_all(&data).await.is_err() {
|
||||
Message::Binary(data)
|
||||
if tcp_tx.write_all(&data).await.is_err() => {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Message::Close(_) => break,
|
||||
_ => {} // ping/pong/text: ignore
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ pub struct AgentSocket {
|
|||
}
|
||||
|
||||
pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result<AgentSocket> {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
let agent = agent.to_owned();
|
||||
if let Some(parent) = socket_path.parent() {
|
||||
std::fs::create_dir_all(parent)
|
||||
|
|
@ -36,7 +37,6 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result
|
|||
// perms (0755) lock it out. 0666 lets the agent user connect;
|
||||
// the bind source dir is per-agent on host so blast radius is
|
||||
// unchanged.
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
std::fs::set_permissions(socket_path, std::fs::Permissions::from_mode(0o666))
|
||||
.with_context(|| format!("chmod agent socket {}", socket_path.display()))?;
|
||||
tracing::info!(%agent, socket = %socket_path.display(), "agent socket listening");
|
||||
|
|
@ -95,7 +95,7 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Re
|
|||
/// cheap "is there anything pending?" check without blocking the
|
||||
/// turn for 30 seconds. To actually park, the caller passes a
|
||||
/// positive `wait_seconds`.
|
||||
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180);
|
||||
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_mins(3);
|
||||
|
||||
/// Server-side hard cap on `Recv.max`. Bounds the size of a single
|
||||
/// round-trip so a confused caller can't drain the entire inbox in
|
||||
|
|
@ -357,6 +357,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
|
|||
///
|
||||
/// The manager is not exempt - grant `read_host_journal` in
|
||||
/// `meta/capabilities.json` to enable it for any agent including the manager.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn dispatch_host_journal(
|
||||
agent: &str,
|
||||
unit: &Option<String>,
|
||||
|
|
@ -619,8 +620,7 @@ fn prepare_remind_storage(
|
|||
fn auto_reminder_path(agent: &str) -> String {
|
||||
let ts_ns = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos())
|
||||
.unwrap_or(0);
|
||||
.map_or(0, |d| d.as_nanos());
|
||||
format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -209,16 +209,16 @@ pub async fn ensure_manager(coord: &Arc<Coordinator>) -> Result<()> {
|
|||
/// Sort `names` in-place so parents precede their children in the topology.
|
||||
/// Uses BFS from root agents (depth 0). Agents absent from `topo` sort last,
|
||||
/// alphabetically within their tier. Stable within each depth tier.
|
||||
pub fn topology_sort(names: &mut Vec<String>, topo: &std::collections::BTreeMap<String, Option<String>>) {
|
||||
pub fn topology_sort(names: &mut [String], topo: &std::collections::BTreeMap<String, Option<String>>) {
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
// Build depth map using owned clones so the borrow on `names` is released
|
||||
// before the sort_by mutable borrow.
|
||||
let name_set: Vec<String> = names.clone();
|
||||
let name_set: Vec<String> = names.to_vec();
|
||||
let mut depth: HashMap<String, usize> = HashMap::new();
|
||||
let mut queue: VecDeque<String> = VecDeque::new();
|
||||
// Seed roots: entries with no parent, or names not present in topo at all.
|
||||
for name in &name_set {
|
||||
if topo.get(name).map_or(true, |p| p.is_none()) {
|
||||
if topo.get(name).is_none_or(Option::is_none) {
|
||||
depth.insert(name.clone(), 0);
|
||||
queue.push_back(name.clone());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|||
|
||||
use crate::coordinator::Coordinator;
|
||||
|
||||
const VACUUM_INTERVAL: Duration = Duration::from_secs(3600);
|
||||
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
|
||||
/// Keep completed task files for 48 hours before sweeping them.
|
||||
const KEEP_SECS: i64 = 48 * 3600;
|
||||
|
||||
|
|
@ -97,7 +97,7 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool {
|
|||
if !TERMINAL_STATUSES.contains(&status) {
|
||||
return false;
|
||||
}
|
||||
let completed_at = v.get("completed_at").and_then(|t| t.as_i64()).unwrap_or(i64::MAX);
|
||||
let completed_at = v.get("completed_at").and_then(serde_json::Value::as_i64).unwrap_or(i64::MAX);
|
||||
completed_at < cutoff
|
||||
}
|
||||
|
||||
|
|
@ -106,10 +106,10 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool {
|
|||
fn delete_trio(dir: &Path, stem: &str) {
|
||||
for ext in ["json", "out", "err"] {
|
||||
let path = dir.join(format!("{stem}.{ext}"));
|
||||
if path.exists() {
|
||||
if let Err(e) = std::fs::remove_file(&path) {
|
||||
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
|
||||
}
|
||||
if path.exists()
|
||||
&& let Err(e) = std::fs::remove_file(&path)
|
||||
{
|
||||
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -410,7 +410,7 @@ pub fn spawn_vacuum(coord: &Arc<crate::coordinator::Coordinator>) {
|
|||
use std::time::Duration;
|
||||
let logs = coord.build_logs.clone();
|
||||
let mut shutdown = coord.shutdown_rx();
|
||||
let interval = Duration::from_secs(3_600);
|
||||
let interval = Duration::from_hours(1);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match logs.vacuum() {
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
//! and `tool-groups.json`.
|
||||
//!
|
||||
//! Format: a JSON object mapping agent name to an array of
|
||||
//! `hive_sh4re::Capability` snake_case strings:
|
||||
//! `hive_sh4re::Capability` `snake_case` strings:
|
||||
//!
|
||||
//! ```json
|
||||
//! {
|
||||
|
|
|
|||
|
|
@ -217,12 +217,12 @@ fn read_dashboard_links(name: &str) -> Vec<DashboardLink> {
|
|||
/// don't lose state during the transition window.
|
||||
fn read_harness_flags(name: &str) -> (bool, bool) {
|
||||
let dir = Coordinator::agent_notes_dir(name);
|
||||
if let Ok(raw) = std::fs::read_to_string(dir.join("hyperhive-harness.json")) {
|
||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) {
|
||||
let rl = v.get("rate_limited").and_then(|x| x.as_bool()).unwrap_or(false);
|
||||
let nl = v.get("needs_login").and_then(|x| x.as_bool()).unwrap_or(false);
|
||||
return (rl, nl);
|
||||
}
|
||||
if let Ok(raw) = std::fs::read_to_string(dir.join("hyperhive-harness.json"))
|
||||
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw)
|
||||
{
|
||||
let rl = v.get("rate_limited").and_then(serde_json::Value::as_bool).unwrap_or(false);
|
||||
let nl = v.get("needs_login").and_then(serde_json::Value::as_bool).unwrap_or(false);
|
||||
return (rl, nl);
|
||||
}
|
||||
// Legacy fallback: presence of individual sentinel files.
|
||||
let rate_limited = dir.join("hyperhive-rate-limited").exists();
|
||||
|
|
|
|||
|
|
@ -159,6 +159,7 @@ fn try_bind(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
|
|||
sock.listen(1024)
|
||||
}
|
||||
|
||||
#[allow(clippy::struct_excessive_bools)]
|
||||
#[derive(Serialize)]
|
||||
struct StateSnapshot {
|
||||
/// Broker seq at the moment this snapshot was assembled. Clients
|
||||
|
|
@ -1833,7 +1834,7 @@ async fn get_build_log_stream(
|
|||
match notify_rx.recv().await {
|
||||
// Notification for a different build — ignore and wait
|
||||
// for the next one.
|
||||
Ok(notif_id) if notif_id != id => continue,
|
||||
Ok(notif_id) if notif_id != id => {}
|
||||
Ok(_) => {
|
||||
match logs.get_progress(id, stdout_cursor, stderr_cursor) {
|
||||
Ok(Some(prog)) => {
|
||||
|
|
@ -1845,20 +1846,17 @@ async fn get_build_log_stream(
|
|||
stderr_append: prog.stderr_append,
|
||||
status: prog.status,
|
||||
done,
|
||||
}) {
|
||||
if tx.send(Ok(Event::default().data(json))).await.is_err() {
|
||||
return; // browser disconnected
|
||||
}
|
||||
}) && tx.send(Ok(Event::default().data(json))).await.is_err() {
|
||||
return; // browser disconnected
|
||||
}
|
||||
if done {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Ok(None) => return, // vacuum reaped the row
|
||||
Err(_) => return,
|
||||
Ok(None) | Err(_) => return, // vacuum reaped row / channel closed
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ use rusqlite::{Connection, Result, params};
|
|||
|
||||
use crate::coordinator::Coordinator;
|
||||
|
||||
const VACUUM_INTERVAL: Duration = Duration::from_secs(3600);
|
||||
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
|
||||
const KEEP_SECS: i64 = 7 * 24 * 3600;
|
||||
|
||||
/// Background loop: sweep every existing agent state dir hourly, run
|
||||
|
|
|
|||
|
|
@ -161,8 +161,7 @@ pub fn duplicate_groups(raw: &str) -> Result<Vec<DuplicateGroup>> {
|
|||
pub async fn check_lock_in_sync(repo: &Path, tag: &str, approval_id: i64) -> Result<()> {
|
||||
let suffix = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos())
|
||||
.unwrap_or(0);
|
||||
.map_or(0, |d| d.as_nanos());
|
||||
let tmp_dir = std::env::temp_dir().join(format!("hive-flake-check-{approval_id}-{suffix}"));
|
||||
|
||||
// Detached worktree at the proposal tag — gives us a clean, mutable
|
||||
|
|
|
|||
|
|
@ -276,8 +276,7 @@ async fn mint_token(name: &str, scopes: &str) -> Result<String> {
|
|||
"{TOKEN_NAME_PREFIX}-{}",
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0)
|
||||
.map_or(0, |d| d.as_secs())
|
||||
);
|
||||
let stdout = forge_admin(&[
|
||||
"user",
|
||||
|
|
|
|||
|
|
@ -274,7 +274,7 @@ fn gateway_systemctl(args: &[&str]) -> bool {
|
|||
/// Synchronise the gateway nginx unit with the current agents.conf:
|
||||
///
|
||||
/// - **active**: send `nginx -s reload` (SIGHUP to master, zero-downtime
|
||||
/// worker replacement). Keeps RELOAD_PENDING set on failure so the
|
||||
/// worker replacement). Keeps `RELOAD_PENDING` set on failure so the
|
||||
/// next poll tick retries.
|
||||
/// - **failed / start-limit-hit**: run `systemctl reset-failed nginx`
|
||||
/// then `systemctl start nginx`. This is the self-healing path: a
|
||||
|
|
|
|||
|
|
@ -181,7 +181,7 @@ async fn port_collision(self_name: &str) -> Option<String> {
|
|||
None
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[allow(clippy::too_many_arguments, clippy::implicit_hasher)]
|
||||
pub async fn spawn(
|
||||
name: &str,
|
||||
hyperhive_flake: &str,
|
||||
|
|
@ -324,8 +324,7 @@ pub async fn is_running(name: &str) -> bool {
|
|||
.args(["is-active", "--quiet", &unit])
|
||||
.status()
|
||||
.await
|
||||
.map(|s| s.success())
|
||||
.unwrap_or(false)
|
||||
.is_ok_and(|s| s.success())
|
||||
}
|
||||
|
||||
/// Fully tear down a sub-agent's container: stop + remove via `nixos-container
|
||||
|
|
@ -346,7 +345,7 @@ pub async fn destroy(name: &str) -> Result<()> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[allow(clippy::too_many_arguments, clippy::implicit_hasher)]
|
||||
pub async fn rebuild(
|
||||
name: &str,
|
||||
hyperhive_flake: &str,
|
||||
|
|
@ -1004,7 +1003,7 @@ fn set_resource_limits(container: &str) -> Result<()> {
|
|||
std::fs::create_dir_all(&dir).with_context(|| format!("create {dir}"))?;
|
||||
let path = format!("{dir}/hyperhive-limits.conf");
|
||||
let content =
|
||||
format!("[Service]\nMemoryMax={DEFAULT_MEMORY_MAX}\nCPUQuota={DEFAULT_CPU_QUOTA}\n",);
|
||||
format!("[Service]\nMemoryMax={DEFAULT_MEMORY_MAX}\nCPUQuota={DEFAULT_CPU_QUOTA}\n");
|
||||
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
|
||||
tracing::info!(
|
||||
%path,
|
||||
|
|
@ -1081,6 +1080,7 @@ fn bind_child_agent_dirs(child: &str, binds: &mut String) {
|
|||
let _ = write!(binds, " --bind={config_dir}:/agents/{child}/config");
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_lines)]
|
||||
fn set_nspawn_flags(
|
||||
container: &str,
|
||||
runtime_dir: &Path,
|
||||
|
|
@ -1265,8 +1265,7 @@ async fn run(args: &[&str]) -> Result<()> {
|
|||
let agent = args
|
||||
.get(1)
|
||||
.copied()
|
||||
.map(|c| c.strip_prefix(AGENT_PREFIX).unwrap_or(c).to_string())
|
||||
.unwrap_or_else(|| "<unknown>".to_string());
|
||||
.map_or_else(|| "<unknown>".to_string(), |c| c.strip_prefix(AGENT_PREFIX).unwrap_or(c).to_string());
|
||||
|
||||
let logs = crate::build_logs::global();
|
||||
let log_id = logs.as_ref().and_then(|h| {
|
||||
|
|
|
|||
|
|
@ -227,7 +227,7 @@ async fn cmd_serve(
|
|||
let vacuum_coord = coord.clone();
|
||||
let mut vacuum_shutdown = coord.shutdown_rx();
|
||||
tokio::spawn(async move {
|
||||
let interval = std::time::Duration::from_secs(3600);
|
||||
let interval = std::time::Duration::from_hours(1);
|
||||
let keep_secs: i64 = 30 * 24 * 3600;
|
||||
loop {
|
||||
match vacuum_coord.broker.vacuum_delivered(keep_secs) {
|
||||
|
|
|
|||
Some files were not shown because too many files have changed in this diff Show more
Loading…
Reference in a new issue