serialise the agent-sockets and gateway publishes and give each its own tmp file

Closes #6094

docs/networking/gateway.md

@@ -316,7 +316,7 @@ idempotent — skips the rename when content is unchanged.
 on subsequent poll ticks.
 <!-- vale write-good.Passive = YES -->
 
-`agents.conf` uses atomic `<path>.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

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,
+/// Serialises [`refresh`] and each tick of the 10-second poll in
-/// so two concurrent refreshes would lose one of the changes.
+/// `workers::agent_sockets`. `topology.json` is read, changed and written
-static REFRESH: Mutex<()> = Mutex::const_new(());
+/// 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.
 ///

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");
+    let tmp = agent_sockets::unique_tmp(&path);
-    std::fs::write(&tmp, &body).with_context(|| format!("write tmp {}", tmp.display()))?;
+    let published = (|| -> Result<()> {
-    // Mode set on the TMP file, before the rename, because a rename
+        std::fs::write(&tmp, &body).with_context(|| format!("write tmp {}", tmp.display()))?;
-    // carries the source's mode and owner and discards the destination's.
+        // Mode set on the TMP file, before the rename, because a rename
-    // The tmpfiles rule declaring 0644 is `f` (create-if-absent), so it
+        // carries the source's mode and owner and discards the destination's.
-    // never re-applies: from the first republish on, the published mode is
+        // The tmpfiles rule declaring 0644 is `f` (create-if-absent), so it
-    // whatever this process's umask happened to produce. That is fine
+        // never re-applies: from the first republish on, the published mode is
-    // today and becomes a dead gateway the day anything tightens c0re's
+        // whatever this process's umask happened to produce. That is fine
-    // umask, since nginx reads this file as a different, unprivileged
+        // today and becomes a dead gateway the day anything tightens c0re's
-    // user. Making it explicit means the mode is a property of the
+        // umask, since nginx reads this file as a different, unprivileged
-    // publisher rather than of an unrelated unit's settings.
+        // user. Making it explicit means the mode is a property of the
-    std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o644))
+        // publisher rather than of an unrelated unit's settings.
-        .with_context(|| format!("chmod tmp {}", tmp.display()))?;
+        std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o644))
-    std::fs::rename(&tmp, &path).with_context(|| {
+            .with_context(|| format!("chmod tmp {}", tmp.display()))?;
-        format!(
+        std::fs::rename(&tmp, &path).with_context(|| {
-            "rename {} -> {} (atomic publish)",
+            format!(
-            tmp.display(),
+                "rename {} -> {} (atomic publish)",
-            path.display()
+                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);

hive-c0re/src/workers/agent_sockets.rs

@@ -1,5 +1,5 @@
 //! `/var/lib/hyperhive/run/agent-sockets.json` writer. Atomic
-//! `<path>.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/<name>/` for all containers with a bound unix
 //! socket.
@@ -127,7 +127,7 @@ fn render(map: &BTreeMap<String, PathBuf>) -> String {
 
 /// Atomically write the JSON for `names` to
 /// `/var/lib/hyperhive/run/agent-sockets.json`. Writes via a sibling
-/// `<path>.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, PathBuf>) -> String {
 /// watchers) don't fire spurious reload events.
 pub fn write(names: &[String]) -> Result<()> {
     let map = build_map(names);
-    let body = render(&map);
+    publish(&host_sockets_path(), &render(&map))
-    let path = host_sockets_path();
+}
-    if std::fs::read_to_string(&path).ok().as_deref() == Some(&body) {
+
+/// 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");
+    let tmp = unique_tmp(path);
-    std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
+    let published = std::fs::write(&tmp, body)
-    std::fs::rename(&tmp, &path).with_context(|| {
+        .with_context(|| format!("write {}", tmp.display()))
-        format!(
+        .and_then(|()| {
-            "rename {} -> {} (atomic publish)",
+            std::fs::rename(&tmp, path).with_context(|| {
-            tmp.display(),
+                format!(
-            path.display()
+                    "rename {} -> {} (atomic publish)",
-        )
+                    tmp.display(),
-    })?;
+                    path.display()
-    Ok(())
+                )
+            })
+        });
+    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<T>(tmp: &Path, result: Result<T>) -> Result<T> {
+    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<String> = 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<String> = (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