Enable persistent agents

Cargo.toml

@@ -26,6 +26,10 @@ members = [
 ]
 resolver = "2"
 
+[workspace.lints.clippy]
+too_many_arguments = "allow"
+wrong_self_convention = "allow"
+
 [workspace.dependencies]
 ansi-to-html = { version = "0.2", features =  [ "lazy-init" ] }
 anyhow = "1"

agent/src/control/mod.rs

@@ -18,7 +18,15 @@ use protocol::{Decoder, FactoryInfo, Message, Payload};
 pub(crate) mod protocol;
 pub(crate) mod server;
 
-pub const SOCKET_PATH: &str = "/var/run/buildomat.sock";
+/// Default socket path.
+const DEFAULT_SOCKET_PATH: &str = "/var/run/buildomat.sock";
+
+/// Get the control socket path for client (bmat) connections.
+/// Uses BUILDOMAT_SOCKET env var if set (set by agent when running jobs),
+/// otherwise falls back to the default path.
+fn socket_path() -> String {
+    std::env::var("BUILDOMAT_SOCKET").unwrap_or_else(|_| DEFAULT_SOCKET_PATH.to_string())
+}
 
 struct Stuff {
     us: Option<UnixStream>,
@@ -28,7 +36,7 @@ struct Stuff {
 
 impl Stuff {
     async fn connect(&mut self) -> Result<()> {
-        self.us = Some(UnixStream::connect(SOCKET_PATH).await?);
+        self.us = Some(UnixStream::connect(socket_path()).await?);
         self.dec = Some(Decoder::new());
         Ok(())
     }
@@ -178,13 +186,13 @@ async fn cmd_address_list(mut l: Level<Stuff>) -> Result<()> {
                 let mut r = Row::default();
 
                 r.add_str("name", &addr.name);
-                r.add_str("cidr", &net.to_string());
-                r.add_str("first", &first.to_string());
-                r.add_str("last", &last.to_string());
+                r.add_str("cidr", net.to_string());
+                r.add_str("first", first.to_string());
+                r.add_str("last", last.to_string());
                 r.add_u64("count", addr.count.into());
                 r.add_str("family", "inet");
-                r.add_str("network", &net.network().to_string());
-                r.add_str("mask", &net.netmask().to_string());
+                r.add_str("network", net.network().to_string());
+                r.add_str("mask", net.netmask().to_string());
                 r.add_str("routed", if addr.routed { "yes" } else { "no" });
                 r.add_str("gateway", addr.gateway.as_deref().unwrap_or("-"));
 
@@ -578,3 +586,21 @@ async fn cmd_factory_private(mut l: Level<Stuff>) -> Result<()> {
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_socket_path_default() {
+        std::env::remove_var("BUILDOMAT_SOCKET");
+        assert_eq!(socket_path(), "/var/run/buildomat.sock");
+    }
+
+    #[test]
+    fn test_socket_path_from_env() {
+        std::env::set_var("BUILDOMAT_SOCKET", "/custom/path/buildomat.sock");
+        assert_eq!(socket_path(), "/custom/path/buildomat.sock");
+        std::env::remove_var("BUILDOMAT_SOCKET");
+    }
+}

agent/src/control/server.rs

@@ -15,10 +15,7 @@ use tokio::{
     },
 };
 
-use super::{
-    protocol::{Decoder, Message, Payload},
-    SOCKET_PATH,
-};
+use super::protocol::{Decoder, Message, Payload};
 
 #[derive(Debug)]
 pub struct Request {
@@ -45,20 +42,25 @@ impl Request {
     }
 }
 
-pub fn listen() -> Result<Receiver<Request>> {
+pub fn listen(socket_path: &std::path::Path) -> Result<Receiver<Request>> {
     /*
      * Create the UNIX socket that the control program will use to contact the
      * agent.
      */
-    std::fs::remove_file(SOCKET_PATH).ok();
-    let ul = UnixListener::bind(SOCKET_PATH)?;
+    // Create parent directory if it doesn't exist (needed for persistent mode)
+    if let Some(parent) = socket_path.parent() {
+        std::fs::create_dir_all(parent).ok();
+    }
+
+    std::fs::remove_file(socket_path).ok();
+    let ul = UnixListener::bind(socket_path)?;
 
     /*
      * Allow everyone to connect:
      */
-    let mut perm = std::fs::metadata(SOCKET_PATH)?.permissions();
+    let mut perm = std::fs::metadata(socket_path)?.permissions();
     perm.set_mode(0o777);
-    std::fs::set_permissions(SOCKET_PATH, perm)?;
+    std::fs::set_permissions(socket_path, perm)?;
 
     /*
      * Create channel to hand requests back to the main loop.

agent/src/exec.rs

@@ -23,10 +23,7 @@ fn spawn_reader<T>(
 where
     T: Read + Send + 'static,
 {
-    let stream = match stream {
-        Some(stream) => stream,
-        None => return None,
-    };
+    let stream = stream?;
 
     Some(std::thread::spawn(move || {
         let mut r = BufReader::new(stream);
@@ -297,7 +294,7 @@ fn run_common(
                  * process.
                  */
                 if ab.bgproc.is_none() {
-                    tx.blocking_send(ab.exit(&start, &end, std::i32::MAX))
+                    tx.blocking_send(ab.exit(&start, &end, i32::MAX))
                         .unwrap();
                 }
 
@@ -332,7 +329,7 @@ fn run_common(
                 let code = if let Some(code) = es.code() {
                     code
                 } else {
-                    std::i32::MAX
+                    i32::MAX
                 };
                 tx.blocking_send(ab.exit(&start, &end, code)).unwrap();
                 stdio_warning