diff --git a/docs/networking/gateway.md b/docs/networking/gateway.md index 2a89f8af..0a2dfd04 100644 --- a/docs/networking/gateway.md +++ b/docs/networking/gateway.md @@ -316,7 +316,7 @@ idempotent — skips the rename when content is unchanged. on subsequent poll ticks. -`agents.conf` uses atomic `.tmp` + `rename()` writes so a crashing +`agents.conf` goes out through a unique tmp file and an atomic `rename()`, so a crashing c0re process never leaves a partial or unparseable file behind. ### Per-agent static frontend split diff --git a/hive-c0re/src/agent_set.rs b/hive-c0re/src/agent_set.rs index e8cc44cc..bf066b9b 100644 --- a/hive-c0re/src/agent_set.rs +++ b/hive-c0re/src/agent_set.rs @@ -9,9 +9,12 @@ use std::path::Path; use anyhow::{Context, Result, bail}; use tokio::sync::Mutex; -/// Serialises [`refresh`]: `topology.json` is read, changed and written back, -/// so two concurrent refreshes would lose one of the changes. -static REFRESH: Mutex<()> = Mutex::const_new(()); +/// Serialises [`refresh`] and each tick of the 10-second poll in +/// `workers::agent_sockets`. `topology.json` is read, changed and written +/// back, so two concurrent refreshes would lose one of the changes; the poll +/// holds it across list-and-write so it cannot publish an older agent list +/// over one `refresh` just wrote. +pub(crate) static REFRESH: Mutex<()> = Mutex::const_new(()); /// Bring the host-side files in line with `agents`, the live agent set. /// diff --git a/hive-c0re/src/gateway_nginx.rs b/hive-c0re/src/gateway_nginx.rs index 3fb3dbe5..11ad4bf1 100644 --- a/hive-c0re/src/gateway_nginx.rs +++ b/hive-c0re/src/gateway_nginx.rs @@ -193,26 +193,30 @@ pub async fn write(names: &[String]) -> Result<()> { std::fs::create_dir_all(parent) .with_context(|| format!("create dir {}", parent.display()))?; } - let tmp = path.with_extension("conf.tmp"); - std::fs::write(&tmp, &body).with_context(|| format!("write tmp {}", tmp.display()))?; - // Mode set on the TMP file, before the rename, because a rename - // carries the source's mode and owner and discards the destination's. - // The tmpfiles rule declaring 0644 is `f` (create-if-absent), so it - // never re-applies: from the first republish on, the published mode is - // whatever this process's umask happened to produce. That is fine - // today and becomes a dead gateway the day anything tightens c0re's - // umask, since nginx reads this file as a different, unprivileged - // user. Making it explicit means the mode is a property of the - // publisher rather than of an unrelated unit's settings. - std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o644)) - .with_context(|| format!("chmod tmp {}", tmp.display()))?; - std::fs::rename(&tmp, &path).with_context(|| { - format!( - "rename {} -> {} (atomic publish)", - tmp.display(), - path.display() - ) - })?; + let tmp = agent_sockets::unique_tmp(&path); + let published = (|| -> Result<()> { + std::fs::write(&tmp, &body).with_context(|| format!("write tmp {}", tmp.display()))?; + // Mode set on the TMP file, before the rename, because a rename + // carries the source's mode and owner and discards the destination's. + // The tmpfiles rule declaring 0644 is `f` (create-if-absent), so it + // never re-applies: from the first republish on, the published mode is + // whatever this process's umask happened to produce. That is fine + // today and becomes a dead gateway the day anything tightens c0re's + // umask, since nginx reads this file as a different, unprivileged + // user. Making it explicit means the mode is a property of the + // publisher rather than of an unrelated unit's settings. + std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o644)) + .with_context(|| format!("chmod tmp {}", tmp.display()))?; + std::fs::rename(&tmp, &path).with_context(|| { + format!( + "rename {} -> {} (atomic publish)", + tmp.display(), + path.display() + ) + })?; + Ok(()) + })(); + agent_sockets::remove_tmp_on_err(&tmp, published)?; // Reset backoff so a fresh topology change gets an immediate attempt, // not a stale cooldown from a previous failure. LAST_FAILED_RELOAD.store(0, Ordering::Relaxed); diff --git a/hive-c0re/src/workers/agent_sockets.rs b/hive-c0re/src/workers/agent_sockets.rs index e8c2002f..98ab4ab0 100644 --- a/hive-c0re/src/workers/agent_sockets.rs +++ b/hive-c0re/src/workers/agent_sockets.rs @@ -1,5 +1,5 @@ //! `/var/lib/hyperhive/run/agent-sockets.json` writer. Atomic -//! `.tmp` + `rename()` write so the gateway's nginx worker never +//! unique-tmp + `rename()` write so the gateway's nginx worker never //! reads a partial file. Includes manager and sub-agents so the gateway //! can route `/agent//` for all containers with a bound unix //! socket. @@ -127,7 +127,7 @@ fn render(map: &BTreeMap) -> String { /// Atomically write the JSON for `names` to /// `/var/lib/hyperhive/run/agent-sockets.json`. Writes via a sibling -/// `.tmp` + rename so a crashing process never leaves a +/// unique tmp file + rename so a crashing process never leaves a /// partial file behind that the gateway worker would fail to parse. /// /// Idempotent — if the rendered content matches what's already on @@ -136,24 +136,51 @@ fn render(map: &BTreeMap) -> String { /// watchers) don't fire spurious reload events. pub fn write(names: &[String]) -> Result<()> { let map = build_map(names); - let body = render(&map); - let path = host_sockets_path(); - if std::fs::read_to_string(&path).ok().as_deref() == Some(&body) { + publish(&host_sockets_path(), &render(&map)) +} + +/// A tmp path next to `path` that no other call in this process shares, so +/// overlapping publishes never write the same tmp file. +pub(crate) fn unique_tmp(path: &Path) -> PathBuf { + static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let mut name = path.file_name().unwrap_or_default().to_os_string(); + name.push(format!(".{}.{n}.tmp", std::process::id())); + path.with_file_name(name) +} + +/// Publish `body` at `path` through a unique tmp file + rename, skipping the +/// write when the file already holds `body`. +fn publish(path: &Path, body: &str) -> Result<()> { + if std::fs::read_to_string(path).ok().as_deref() == Some(body) { return Ok(()); } if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; } - let tmp = path.with_extension("json.tmp"); - std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?; - std::fs::rename(&tmp, &path).with_context(|| { - format!( - "rename {} -> {} (atomic publish)", - tmp.display(), - path.display() - ) - })?; - Ok(()) + let tmp = unique_tmp(path); + let published = std::fs::write(&tmp, body) + .with_context(|| format!("write {}", tmp.display())) + .and_then(|()| { + std::fs::rename(&tmp, path).with_context(|| { + format!( + "rename {} -> {} (atomic publish)", + tmp.display(), + path.display() + ) + }) + }); + remove_tmp_on_err(&tmp, published) +} + +/// Pass `result` through, removing `tmp` first when it is an error. Unique tmp +/// names are never overwritten by a later attempt, so a failed publish must +/// not leave its file behind. +pub(crate) fn remove_tmp_on_err(tmp: &Path, result: Result) -> Result { + if result.is_err() { + let _ = std::fs::remove_file(tmp); + } + result } /// Spawn the marker poll task. Periodically re-runs `write` so the @@ -180,6 +207,10 @@ pub fn spawn_poll() { // window before the next agent restart. loop { interval.tick().await; + // Same lock as `agent_set::refresh`: the list and both writes + // happen as one step, so a refresh cannot be overwritten by an + // older list. + let guard = crate::agent_set::REFRESH.lock().await; match crate::lifecycle::agents_for_meta_listing().await { Ok(agents) => { let names: Vec = agents.into_iter().map(|a| a.name).collect(); @@ -194,6 +225,7 @@ pub fn spawn_poll() { if let Err(e) = crate::gateway_nginx::write(&names).await { tracing::debug!(error = ?e, "gateway_nginx poll write failed"); } + drop(guard); // Retry a pending nginx reload that failed on a // previous tick (no-op if no reload is pending). crate::gateway_nginx::reload_if_pending().await; @@ -210,6 +242,56 @@ pub fn spawn_poll() { mod tests { use super::*; + /// Concurrent publishes to one path never fail, and the file ends up + /// holding exactly one of the inputs. + #[test] + fn concurrent_publish_leaves_one_complete_body() { + let dir = std::env::temp_dir().join(format!("hive-sockets-test-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let path = dir.join("agent-sockets.json"); + let bodies: Vec = (0..4) + .map(|i| render(&build_map_with(&[format!("agent{i}")], |_| true)).repeat(200)) + .collect(); + let handles: Vec<_> = (0..8) + .map(|t| { + let (path, bodies) = (path.clone(), bodies.clone()); + std::thread::spawn(move || { + for i in 0..50 { + publish(&path, &bodies[(t + i) % bodies.len()]).unwrap(); + } + }) + }) + .collect(); + for h in handles { + h.join().unwrap(); + } + let got = std::fs::read_to_string(&path).unwrap(); + assert!(bodies.contains(&got)); + assert_eq!( + std::fs::read_dir(&dir).unwrap().count(), + 1, + "tmp files left" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A publish that fails at the rename leaves no tmp file behind. + #[test] + fn failed_publish_removes_its_tmp_file() { + let dir = std::env::temp_dir().join(format!("hive-sockets-fail-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let path = dir.join("agent-sockets.json"); + // A directory at the target makes the rename fail. + std::fs::create_dir_all(&path).unwrap(); + assert!(publish(&path, "{}").is_err()); + let names: Vec<_> = std::fs::read_dir(&dir) + .unwrap() + .map(|e| e.unwrap().file_name()) + .collect(); + assert_eq!(names, vec![std::ffi::OsString::from("agent-sockets.json")]); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn socket_path_for_uses_subdir_layout() { // Per-agent subdir + fixed socket filename — see module-level