fix nbspc

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dkl"
-version = "1.2.1"
+version = "1.0.0"
 edition = "2024"
 
 [profile.release]
@@ -23,14 +23,12 @@ eyre = "0.6.12"
 fastrand = "2.3.0"
 futures = "0.3.31"
 futures-util = "0.3.31"
-getrandom = "0.4.2"
 glob = "0.3.2"
 hex = "0.4.3"
 human-units = "0.5.3"
 log = "0.4.27"
 lz4 = "1.28.1"
-memchr = "2.8.0"
-nix = { version = "0.31.2", features = ["fs", "process", "signal", "user"] }
+nix = { version = "0.31.2", features = ["process", "signal", "user"] }
 openssl = "0.10.73"
 page_size = "0.6.0"
 reqwest = { version = "0.13.1", features = ["json", "stream", "native-tls", "socks"], default-features = false }
@@ -38,9 +36,9 @@ rpassword = "7.4.0"
 rust-argon2 = "3.0.0"
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
-serde_yaml = { version = "0.10.0", package = "serde_yaml_ng" }
+serde_yaml = "0.9.34"
 signal-hook = "0.4.4"
 tabled = "0.20.0"
 thiserror = "2.0.12"
-tokio = { version = "1.45.1", features = ["fs", "io-std", "macros", "process", "rt", "signal"] }
+tokio = { version = "1.45.1", features = ["fs", "io-std", "macros", "process", "rt"] }
 

Dockerfile

@@ -1,4 +1,4 @@
-from mcluseau/rust:1.95.0 as build
+from mcluseau/rust:1.94.0 as build
 
 workdir /app
 copy . .
@@ -10,6 +10,6 @@ run \
  && find target/release -maxdepth 1 -type f -executable -exec cp -v {} /dist/ +
 
 # ------------------------------------------------------------------------
-from alpine:3.23.4
+from alpine:3.23
 copy --from=build /dist/ /bin/
 

build-dist

@@ -1 +1 @@
-cargo install --locked --path . --root dist
+cargo install --path . --root dist

src/apply.rs

@@ -1,20 +1,12 @@
-use eyre::{Result, format_err};
+use eyre::Result;
 use log::info;
 use std::path::Path;
 use tokio::fs;
 
-use crate::File;
+use crate::base64_decode;
 
-pub async fn files(files: &[File], root: &str, dry_run: bool) -> Result<()> {
-    for f in files {
-        if let Err(e) = file(f, root, dry_run).await {
-            return Err(format_err!("{}: {e}", f.path));
-        }
-    }
-    Ok(())
-}
-
-pub async fn file(file: &File, root: &str, dry_run: bool) -> Result<()> {
+pub async fn files(files: &[crate::File], root: &str, dry_run: bool) -> Result<()> {
+    for file in files {
         let path = chroot(root, &file.path);
         let path = Path::new(&path);
 
@@ -22,49 +14,60 @@ pub async fn file(file: &File, root: &str, dry_run: bool) -> Result<()> {
             fs::create_dir_all(parent).await?;
         }
 
-    let kind = file.kind();
-    let content = kind.content()?;
-
         use crate::FileKind as K;
-    match kind.as_ref() {
-        K::Skip => {
-            info!("{}: kind is skip", file.path);
-            return Ok(());
-        }
-        K::Content(_) | K::Content64(_) | K::Parts(_) => {
-            let content = content.expect("this file kind should have content");
+        match &file.kind {
+            K::Content(content) => {
                 if dry_run {
-                info!("would create {} ({} bytes)", file.path, content.len());
+                    info!(
+                        "would create {} ({} bytes from content)",
+                        file.path,
+                        content.len()
+                    );
                 } else {
-                fs::write(path, &content).await?;
+                    fs::write(path, content.as_bytes()).await?;
                 }
             }
-        K::Dir => {
+            K::Content64(content) => {
+                let content = base64_decode(content)?;
+                if dry_run {
+                    info!(
+                        "would create {} ({} bytes from content64)",
+                        file.path,
+                        content.len()
+                    );
+                } else {
+                    fs::write(path, content).await?
+                }
+            }
+            K::Dir(true) => {
                 if dry_run {
                     info!("would create {} (directory)", file.path);
                 } else {
                     fs::create_dir(path).await?;
                 }
             }
+            K::Dir(false) => {} // shouldn't happen, but semantic is to ignore
             K::Symlink(tgt) => {
                 if dry_run {
                     info!("would create {} (symlink to {})", file.path, tgt);
                 } else {
-                let _ = fs::remove_file(path).await; // we're ln --force
                     fs::symlink(tgt, path).await?;
                 }
             }
         }
 
         if dry_run {
-        return Ok(());
+            continue;
         }
 
-    if !file.is_symlink() {
-        set_perms(path, file.mode).await?;
+        match file.kind {
+            K::Symlink(_) => {}
+            _ => set_perms(path, file.mode).await?,
         }
 
         info!("created {}", file.path);
+    }
+
     Ok(())
 }
 

src/bin/dkl.rs

@@ -1,5 +1,5 @@
 use clap::{CommandFactory, Parser, Subcommand};
-use eyre::{Result, format_err};
+use eyre::{format_err, Result};
 use human_units::Duration;
 use log::{debug, error};
 use std::net::SocketAddr;
@@ -9,8 +9,6 @@ use tokio::fs;
 #[derive(Parser)]
 #[command()]
 struct Cli {
-    #[arg(long)]
-    log_to: Option<PathBuf>,
     #[command(subcommand)]
     command: Command,
 }
@@ -34,17 +32,17 @@ enum Command {
     Logger {
         /// Path where the logs are stored
         #[arg(long, short = 'p', default_value = "/var/log", env = "DKL_LOG_PATH")]
-        log_path: PathBuf,
+        log_path: String,
         /// Name of the log instead of the command's basename
         #[arg(long, short = 'n')]
-        log_name: Option<PathBuf>,
+        log_name: Option<String>,
         /// prefix log lines with time & stream
         #[arg(long)]
         with_prefix: bool,
         /// exec command in this cgroup
         #[arg(long)]
         cgroup: Option<String>,
-        command: PathBuf,
+        command: String,
         args: Vec<String>,
     },
     Log {
@@ -92,11 +90,6 @@ enum Command {
         #[command(subcommand)]
         cmd: CgCmd,
     },
-
-    Rc {
-        #[command(subcommand)]
-        cmd: RcCmd,
-    },
 }
 
 #[derive(Subcommand)]
@@ -111,56 +104,16 @@ enum CgCmd {
     },
 }
 
-#[derive(Subcommand)]
-enum RcCmd {
-    Run,
-    Ls,
-    Status,
-    ReloadConfig,
-    Start {
-        #[arg(add = completions(dkl::rc::complete))]
-        key: String,
-    },
-    Stop {
-        #[arg(add = completions(dkl::rc::complete))]
-        key: String,
-    },
-    Reload {
-        #[arg(add = completions(dkl::rc::complete))]
-        key: String,
-    },
-    Sig {
-        #[arg(add = completions(dkl::rc::complete))]
-        key: String,
-        signal: u32,
-    },
-    Ctl {
-        args: Vec<String>,
-    },
-}
-
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<()> {
     clap_complete::CompleteEnv::with_factory(Cli::command).complete();
 
     let cli = Cli::parse();
 
-    {
-        let mut builder = env_logger::builder();
-        builder.parse_filters("info").parse_default_env();
-
-        if let Some(log_to) = cli.log_to {
-            builder.target(env_logger::Target::Pipe(Box::new(
-                std::fs::OpenOptions::new()
-                    .create(true)
-                    .append(true)
-                    .open(log_to)
-                    .unwrap(),
-            )));
-        }
-
-        builder.init();
-    }
+    env_logger::builder()
+        .parse_filters("info")
+        .parse_default_env()
+        .init();
 
     use Command as C;
     match cli.command {
@@ -174,24 +127,23 @@ async fn main() -> Result<()> {
             apply_config(&config, &filters, &prefix, dry_run).await
         }
         C::Logger {
-            log_path,
-            log_name,
+            ref log_path,
+            ref log_name,
             with_prefix,
             cgroup,
             command,
             args,
         } => {
-            let log_name = log_name.unwrap_or_else(|| command.file_prefix().unwrap().into());
+            let command = command.as_str();
+            let log_name = log_name.as_deref().unwrap_or_else(|| basename(command));
 
-            let logger = dkl::logger::Logger {
+            dkl::logger::Logger {
                 log_path,
                 log_name,
                 with_prefix,
-                cgroup,
-            };
-
-            let cmd = logger.setup(command, &args).await?;
-            logger.exec(cmd).await
+            }
+            .run(cgroup, command, &args)
+            .await
         }
         C::Log {
             log_path,
@@ -235,20 +187,6 @@ async fn main() -> Result<()> {
                 cols,
             } => Ok(dkl::cgroup::ls(root, &exclude, cols.as_deref()).await?),
         },
-
-        C::Rc { cmd } => match cmd {
-            RcCmd::Run => Ok(dkl::rc::run().await?),
-            RcCmd::Ls => Ok(dkl::rc::ctl(["ls"]).await?),
-            RcCmd::Status => Ok(dkl::rc::ctl(["status"]).await?),
-            RcCmd::ReloadConfig => Ok(dkl::rc::ctl(["reload-config"]).await?),
-            RcCmd::Start { key } => Ok(dkl::rc::ctl(["start", &key]).await?),
-            RcCmd::Stop { key } => Ok(dkl::rc::ctl(["stop", &key]).await?),
-            RcCmd::Reload { key } => Ok(dkl::rc::ctl(["reload", &key]).await?),
-            RcCmd::Sig { key, signal } => {
-                Ok(dkl::rc::ctl(["sig", &key, &signal.to_string()]).await?)
-            }
-            RcCmd::Ctl { args } => Ok(dkl::rc::ctl(&args).await?),
-        },
     }
 }
 
@@ -360,6 +298,10 @@ fn parse_ts_arg(ts: Option<String>) -> Result<Option<dkl::logger::Timestamp>> {
     }
 }
 
+fn basename(path: &str) -> &str {
+    path.rsplit_once('/').map_or(path, |split| split.1)
+}
+
 fn parse_globs(filters: &[String]) -> Result<Vec<glob::Pattern>> {
     let mut errors = false;
     let filters = (filters.iter())
@@ -379,25 +321,3 @@ fn parse_globs(filters: &[String]) -> Result<Vec<glob::Pattern>> {
 
     Ok(filters)
 }
-
-use clap_complete::{ArgValueCandidates, CompletionCandidate};
-
-fn completions(f: impl AsyncFn() -> Vec<String> + Send + Sync + 'static) -> ArgValueCandidates {
-    let f = std::sync::Arc::new(f);
-
-    ArgValueCandidates::new(move || {
-        let f = f.clone();
-        std::thread::spawn(move || {
-            tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .unwrap()
-                .block_on(async move { f().await })
-        })
-        .join()
-        .into_iter()
-        .flatten()
-        .map(CompletionCandidate::new)
-        .collect::<Vec<_>>()
-    })
-}

src/bin/dls.rs

@@ -3,7 +3,6 @@ use clap::{CommandFactory, Parser, Subcommand};
 use eyre::format_err;
 use futures_util::Stream;
 use futures_util::StreamExt;
-use std::path::PathBuf;
 use std::time::{Duration, SystemTime};
 use tokio::fs;
 use tokio::io::{AsyncWrite, AsyncWriteExt};
@@ -22,8 +21,6 @@ struct Cli {
 
 #[derive(Subcommand)]
 enum Command {
-    #[command(subcommand)]
-    Config(Config),
     Clusters,
     Cluster {
         cluster: String,
@@ -43,16 +40,6 @@ enum Command {
     Hash {
         salt: String,
     },
-    Store {
-        store_path: PathBuf,
-        #[command(subcommand)]
-        op: StoreOp,
-    },
-}
-
-#[derive(Subcommand)]
-enum Config {
-    Upload { config_path: PathBuf },
 }
 
 #[derive(Subcommand)]
@@ -109,12 +96,6 @@ enum ClusterCommand {
     },
 }
 
-#[derive(Subcommand)]
-enum StoreOp {
-    Get { data_path: PathBuf },
-    Set { data_path: PathBuf, value: String },
-}
-
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> eyre::Result<()> {
     clap_complete::CompleteEnv::with_factory(Cli::command).complete();
@@ -250,29 +231,6 @@ async fn main() -> eyre::Result<()> {
             println!("hash (hex):    {}", hex::encode(&hash));
             println!("hash (base64): {}", dkl::base64_encode(&hash));
         }
-        C::Store { store_path, op } => {
-            let mut s = dls::store::Store::new(store_path);
-            s.unlock(&std::env::var("DLS_STORE_PW").unwrap()).await?;
-
-            match op {
-                StoreOp::Get { data_path } => {
-                    let mut data = std::io::Cursor::new(s.read(data_path).await?);
-                    tokio::io::copy(&mut data, &mut tokio::io::stdout()).await?;
-                }
-                StoreOp::Set { data_path, value } => s.write(data_path, value.as_bytes()).await?,
-            }
-        }
-        C::Config(cmd) => match cmd {
-            Config::Upload { config_path } => {
-                let cfg = fs::read(&config_path).await?;
-                let cfg: dls::Config = if config_path.ends_with(".yaml") {
-                    serde_yaml::from_slice(&cfg)?
-                } else {
-                    serde_json::from_slice(&cfg)?
-                };
-                dls().upload_config(&cfg).await?;
-            }
-        },
     };
 
     Ok(())

src/bootstrap.rs

@@ -250,10 +250,4 @@ pub struct Bootstrap {
     pub dev: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub seed: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub seed_ca: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub seed_servername: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub seed_proxy: Option<String>,
 }

src/cgroup.rs

@@ -64,8 +64,8 @@ pub async fn ls(
     }
 
     use tabled::settings::{
-        Alignment, Modify,
         object::{Column, Row},
+        Alignment, Modify,
     };
     let mut table = table.build();
     table.with(tabled::settings::Style::psql());
@@ -176,15 +176,6 @@ impl Cgroup {
         dir.push(name);
         Self::read(self.path.child(name), &dir).await
     }
-
-    pub async fn write_param(
-        &self,
-        name: impl AsRef<StdPath>,
-        value: impl AsRef<[u8]>,
-    ) -> fs::Result<()> {
-        let cg_dir = PathBuf::from(self.path.as_ref());
-        fs::write(cg_dir.join(name), value).await
-    }
 }
 
 impl PartialEq for Cgroup {

src/dls.rs

@@ -23,18 +23,6 @@ impl Client {
         }
     }
 
-    pub fn new_with_http_client(
-        base_url: String,
-        token: String,
-        http_client: reqwest::Client,
-    ) -> Self {
-        Self {
-            base_url,
-            token,
-            http_client,
-        }
-    }
-
     pub fn with_proxy(self, proxy: String) -> reqwest::Result<Self> {
         let proxy = reqwest::Proxy::all(proxy)?;
         Ok(Self {
@@ -77,15 +65,6 @@ impl Client {
         Ok(resp.bytes_stream())
     }
 
-    pub async fn upload_config(&self, config: &Config) -> Result<()> {
-        do_req(self.req(Method::POST, "configs")?.json(config), &self.token).await?;
-        Ok(())
-    }
-
-    pub async fn get_config(&self) -> Result<Config> {
-        self.get_json("config").await
-    }
-
     pub async fn get_json<T: serde::de::DeserializeOwned>(&self, path: impl Display) -> Result<T> {
         self.req_json(self.get(&path)?).await
     }
@@ -175,7 +154,7 @@ impl<'t> Host<'t> {
     pub async fn asset(
         &self,
         asset_name: &str,
-    ) -> Result<impl Stream<Item = reqwest::Result<Bytes>> + use<>> {
+    ) -> Result<impl Stream<Item = reqwest::Result<Bytes>>> {
         let req = self.dls.get(format!("hosts/{}/{asset_name}", self.name))?;
         let resp = do_req(req, &self.dls.token).await?;
         Ok(resp.bytes_stream())
@@ -183,23 +162,17 @@ impl<'t> Host<'t> {
 }
 
 #[derive(Default, serde::Deserialize, serde::Serialize)]
-#[serde(default, rename_all = "PascalCase")]
+#[serde(rename_all = "PascalCase")]
 pub struct Config {
-    #[serde(deserialize_with = "deserialize_null_as_default", alias = "clusters")]
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
     pub clusters: Vec<ClusterConfig>,
-    #[serde(deserialize_with = "deserialize_null_as_default", alias = "hosts")]
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
     pub hosts: Vec<HostConfig>,
-    #[serde(
-        deserialize_with = "deserialize_null_as_default",
-        alias = "hosttemplates"
-    )]
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
     pub host_templates: Vec<HostConfig>,
-    #[serde(default, rename = "SSLConfig", alias = "sslconfig")]
+    #[serde(default, rename = "SSLConfig")]
     pub ssl_config: String,
-    #[serde(
-        deserialize_with = "deserialize_null_as_default",
-        alias = "extracacerts"
-    )]
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
     pub extra_ca_certs: Map<String, String>,
 }
 
@@ -217,57 +190,40 @@ where
 #[derive(serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct ClusterConfig {
-    #[serde(alias = "name")]
     pub name: String,
-    #[serde(alias = "bootstrappods")]
     pub bootstrap_pods: String,
-    #[serde(alias = "addons")]
     pub addons: String,
 }
 
 #[derive(Default, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct HostConfig {
-    #[serde(alias = "name")]
     pub name: String,
-    #[serde(
-        default,
-        skip_serializing_if = "Option::is_none",
-        alias = "cluster_name"
-    )]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cluster_name: Option<String>,
 
-    #[serde(rename = "IPs", alias = "ips")]
+    #[serde(rename = "IPs")]
     pub ips: Vec<IpAddr>,
 
-    #[serde(default, skip_serializing_if = "Map::is_empty", alias = "labels")]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub labels: Map<String, String>,
-    #[serde(default, skip_serializing_if = "Map::is_empty", alias = "annotations")]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub annotations: Map<String, String>,
 
-    #[serde(
-        rename = "IPXE",
-        skip_serializing_if = "Option::is_none",
-        alias = "ipxe"
-    )]
+    #[serde(rename = "IPXE", skip_serializing_if = "Option::is_none")]
     pub ipxe: Option<String>,
 
-    #[serde(alias = "initrd")]
     pub initrd: String,
-    #[serde(alias = "kernel")]
     pub kernel: String,
-    #[serde(alias = "versions")]
     pub versions: Map<String, String>,
 
     /// initrd config template
-    #[serde(alias = "bootstrapconfig")]
     pub bootstrap_config: String,
     /// files to add to the final initrd config, with rendering
-    #[serde(default, skip_serializing_if = "Vec::is_empty", alias = "initrdfiles")]
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub initrd_files: Vec<crate::File>,
 
     /// system config template
-    #[serde(alias = "config")]
     pub config: String,
 }
 

src/dls/store.rs

@@ -1,13 +1,4 @@
-use openssl::symm::Mode;
-use std::borrow::Cow;
-use std::path::{Path, PathBuf};
-use tokio::{fs, io::AsyncWriteExt};
-
-pub type Salt = [u8; 16];
-pub type Key = [u8; 32];
-pub type Hash = Vec<u8>;
-
-pub fn hash_password(salt: &[u8], passphrase: &str) -> Result<Key> {
+pub fn hash_password(salt: &[u8], passphrase: &str) -> argon2::Result<[u8; 32]> {
     let hash = argon2::hash_raw(
         passphrase.as_bytes(),
         salt,
@@ -15,191 +6,12 @@ pub fn hash_password(salt: &[u8], passphrase: &str) -> Result<Key> {
             variant: argon2::Variant::Argon2id,
             hash_length: 32,
             time_cost: 1,
-            mem_cost: 64 << 10,
+            mem_cost: 65536,
             thread_mode: argon2::ThreadMode::Parallel,
             lanes: 4,
             ..Default::default()
         },
-    )
-    .map_err(Error::Hash)?;
+    )?;
 
-    Ok(hash.try_into().unwrap())
-}
-
-pub struct Store {
-    path: PathBuf,
-    key: Option<Key>,
-}
-
-impl Store {
-    pub fn new(path: impl Into<PathBuf>) -> Self {
-        Self {
-            path: path.into(),
-            key: None,
-        }
-    }
-
-    pub async fn unlock(&mut self, passphrase: &str) -> Result<()> {
-        let keys = self.read_keys().await?;
-
-        let salt = keys.salt;
-
-        let user_key = hash_password(&salt, passphrase)?;
-        let user_hash = openssl::sha::sha512(&user_key);
-
-        for nk in keys.keys {
-            if nk.hash != user_hash {
-                continue;
-            }
-
-            // build iv+data from salt
-            let mut enc_data = Vec::with_capacity(salt.len() + nk.enc_key.len());
-            enc_data.extend_from_slice(&salt);
-            enc_data.extend_from_slice(&nk.enc_key);
-
-            let key = crypt(&enc_data, &user_key, Mode::Decrypt)?;
-            let key = key.try_into().map_err(|_| Error::InvalidKey)?;
-
-            self.key = Some(key);
-            return Ok(());
-        }
-
-        Err(Error::KeyNotFound)
-    }
-
-    pub async fn read(&self, path: impl AsRef<Path>) -> Result<Vec<u8>> {
-        let enc_data = self.read_file(path.as_ref().with_extension("data")).await?;
-        self.decrypt(&enc_data)
-    }
-    pub async fn read_to_string(&self, path: impl AsRef<Path>) -> Result<String> {
-        let path = path.as_ref();
-        String::from_utf8(self.read(path).await?).map_err(|_| Error::InvalidUtf8(path.into()))
-    }
-
-    pub async fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> Result<()> {
-        let enc_data = self.encrypt(data)?;
-        safe_write(self.path.join(&path).with_extension("data"), &enc_data).await
-    }
-
-    async fn read_keys(&self) -> Result<Keys<'_>> {
-        let keys = self.read_file(".keys").await?;
-        let Some(keys) = keys.strip_prefix(b"{json}") else {
-            return Err(Error::InvalidKeys);
-        };
-
-        serde_json::from_slice(keys).map_err(Error::KeysParse)
-    }
-
-    async fn read_file(&self, subpath: impl AsRef<Path>) -> Result<Vec<u8>> {
-        let path = self.path.join(subpath);
-        fs::read(&path).await.map_err(|e| Error::Read(path, e))
-    }
-
-    fn encrypt(&self, src: &[u8]) -> Result<Vec<u8>> {
-        self.crypt(src, Mode::Encrypt)
-    }
-    fn decrypt(&self, src: &[u8]) -> Result<Vec<u8>> {
-        self.crypt(src, Mode::Decrypt)
-    }
-    fn crypt(&self, src: &[u8], mode: Mode) -> Result<Vec<u8>> {
-        let key = self.key.as_ref().ok_or(Error::NotUnlocked)?;
-        crypt(src, key, mode)
-    }
-}
-
-async fn safe_write(path: impl AsRef<Path>, contents: &[u8]) -> Result<()> {
-    let path = path.as_ref();
-    let tmp = path.with_added_extension("new");
-
-    let tmp_err = |e| Error::Write(tmp.clone(), e);
-
-    let mut file = fs::OpenOptions::new()
-        .create(true)
-        .truncate(true)
-        .write(true)
-        .open(&tmp)
-        .await
-        .map_err(tmp_err)?;
-
-    file.write_all(contents).await.map_err(tmp_err)?;
-    file.sync_all().await.map_err(tmp_err)?;
-    file.shutdown().await.map_err(tmp_err)?;
-
-    fs::rename(tmp, &path)
-        .await
-        .map_err(|e| Error::Write(path.into(), e))
-}
-
-fn crypt(mut src: &[u8], key: &Key, mode: Mode) -> Result<Vec<u8>> {
-    use openssl::symm::{Cipher, Crypter};
-
-    let mut iv = [0u8; 16];
-    let mut dst: Vec<u8>;
-    let crypt_dst: &mut [u8];
-
-    match mode {
-        Mode::Encrypt => {
-            getrandom::fill(&mut iv).unwrap();
-            dst = vec![0u8; iv.len() + src.len()];
-            dst[..iv.len()].copy_from_slice(&iv);
-            crypt_dst = &mut dst[iv.len()..];
-        }
-        Mode::Decrypt => {
-            iv = src[..iv.len()]
-                .try_into()
-                .map_err(|_| Error::DecryptInputToSmall)?;
-            src = &src[iv.len()..];
-            dst = vec![0u8; src.len()];
-            crypt_dst = &mut dst;
-        }
-    }
-
-    let cipher = Cipher::aes_256_cfb128();
-    Crypter::new(cipher, mode, key, Some(&iv))
-        .expect("Failed to init AES")
-        .update(src, crypt_dst)
-        .expect("AES CFB encrypt/decrypt failed");
-
-    Ok(dst)
-}
-
-pub type Result<T> = std::result::Result<T, Error>;
-
-#[derive(Debug, thiserror::Error)]
-pub enum Error {
-    #[error("hash error: {0}")]
-    Hash(argon2::Error),
-    #[error("read {0} failed: {1}")]
-    Read(PathBuf, std::io::Error),
-    #[error("write {0} failed: {1}")]
-    Write(PathBuf, std::io::Error),
-    #[error("read {0} failed: invalid UTF-8")]
-    InvalidUtf8(PathBuf),
-    #[error("invalid keys data")]
-    InvalidKeys,
-    #[error("invalid key")]
-    InvalidKey,
-    #[error("keys parse error: {0}")]
-    KeysParse(serde_json::Error),
-    #[error("key not found")]
-    KeyNotFound,
-    #[error("store not unlocked")]
-    NotUnlocked,
-    #[error("decrypt input too small")]
-    DecryptInputToSmall,
-}
-
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "PascalCase")]
-struct Keys<'t> {
-    salt: Salt,
-    keys: Vec<NamedKey<'t>>,
-}
-
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "PascalCase")]
-struct NamedKey<'t> {
-    name: Cow<'t, str>,
-    hash: Hash,
-    enc_key: Key,
+    unsafe { Ok(hash.try_into().unwrap_unchecked()) }
 }

src/dynlay.rs

@@ -60,7 +60,7 @@ impl<'t> Dynlay<'t> {
 
         // mount layer
         info!("mounting layer");
-        sudo("mount", &[lay_path, &mount_path_str]).await?;
+        sudo("mount", &["-t", "squashfs", lay_path, &mount_path_str]).await?;
 
         let mut paths = spawn_walk_dir(mount_path.clone());
         while let Some(result) = paths.recv().await {

src/lib.rs

@@ -1,5 +1,3 @@
-use std::borrow::Cow;
-
 pub mod apply;
 pub mod bootstrap;
 pub mod cgroup;
@@ -11,7 +9,7 @@ pub mod logger;
 pub mod proxy;
 pub mod rc;
 
-#[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Config {
     pub layers: Vec<String>,
     pub root_user: RootUser,
@@ -25,14 +23,14 @@ pub struct Config {
     pub users: Vec<User>,
 }
 
-#[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct RootUser {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub password_hash: Option<String>,
     pub authorized_keys: Vec<String>,
 }
 
-#[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Mount {
     pub dev: String,
     pub path: String,
@@ -42,14 +40,14 @@ pub struct Mount {
     pub options: Option<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct Group {
     pub name: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub gid: Option<u32>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct User {
     pub name: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -58,45 +56,22 @@ pub struct User {
     pub gid: Option<u32>,
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 pub struct File {
     pub path: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub mode: Option<u32>,
-    pub kind: Option<FileKind>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub content: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub content64: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub symlink: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub parts: Option<Vec<FilePart>>,
-    #[serde(default, skip_serializing_if = "is_false")]
-    pub dir: bool,
+    #[serde(flatten)]
+    pub kind: FileKind,
 }
 
-fn is_false(b: &bool) -> bool {
-    !b
-}
-
-#[derive(Default, Debug, Clone, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum FileKind {
-    #[default]
-    Skip,
     Content(String),
     Content64(String),
-    Parts(Vec<FilePart>),
     Symlink(String),
-    Dir,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
-#[serde(rename_all = "lowercase")]
-pub enum FilePart {
-    Content(String),
-    Content64(String),
+    Dir(bool),
 }
 
 // ------------------------------------------------------------------------
@@ -111,71 +86,11 @@ impl Config {
 }
 
 pub fn base64_decode(s: &str) -> Result<Vec<u8>, base64::DecodeError> {
-    use base64::{prelude::BASE64_STANDARD_NO_PAD as B64, Engine as _};
+    use base64::{Engine as _, prelude::BASE64_STANDARD_NO_PAD as B64};
     B64.decode(s.trim_end_matches('='))
 }
 
 pub fn base64_encode(b: &[u8]) -> String {
-    use base64::{prelude::BASE64_STANDARD as B64, Engine as _};
+    use base64::{Engine as _, prelude::BASE64_STANDARD as B64};
     B64.encode(b)
 }
-
-impl<'t> File {
-    pub fn kind(&'t self) -> Cow<'t, FileKind> {
-        self.kind.as_ref().map(Cow::Borrowed).unwrap_or_else(|| {
-            use FileKind::*;
-            Cow::Owned(if let Some(ref s) = self.content {
-                Content(s.clone())
-            } else if let Some(ref s) = self.content64 {
-                Content64(s.clone())
-            } else if let Some(ref s) = self.symlink {
-                Symlink(s.clone())
-            } else if let Some(ref p) = self.parts {
-                Parts(p.clone())
-            } else if self.dir {
-                Dir
-            } else {
-                Skip
-            })
-        })
-    }
-
-    pub fn is_symlink(&self) -> bool {
-        if let Some(ref kind) = self.kind {
-            matches!(kind, FileKind::Symlink(_))
-        } else {
-            self.symlink.is_some()
-        }
-    }
-}
-
-impl FileKind {
-    pub fn content<'t>(&'t self) -> Result<Option<Cow<'t, [u8]>>, base64::DecodeError> {
-        use FileKind::*;
-        Ok(match self {
-            Content(content) => Some(Cow::Borrowed(content.as_bytes())),
-            Content64(content) => {
-                let content = base64_decode(content)?;
-                Some(Cow::Owned(content))
-            }
-            Parts(parts) => {
-                let mut assembly = Vec::new();
-                for part in parts {
-                    assembly.extend(part.content()?.into_iter());
-                }
-                Some(Cow::Owned(assembly))
-            }
-            _ => None,
-        })
-    }
-}
-
-impl FilePart {
-    pub fn content(&self) -> Result<Cow<'_, [u8]>, base64::DecodeError> {
-        use FilePart::*;
-        Ok(match self {
-            Content(content) => Cow::Borrowed(content.as_bytes()),
-            Content64(content) => Cow::Owned(base64_decode(content)?),
-        })
-    }
-}

src/logger.rs

@@ -1,16 +1,15 @@
 use async_compression::tokio::write::{ZstdDecoder, ZstdEncoder};
 use chrono::{DurationRound, TimeDelta, Utc};
-use eyre::{Result, format_err};
+use eyre::{format_err, Result};
 use log::{debug, error, warn};
-use std::ffi::OsStr;
 use std::path::{Path, PathBuf};
 use std::process::Stdio;
 use tokio::{
     fs::File,
     io::{self, AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, BufWriter},
-    process::{Child, Command},
+    process,
     sync::mpsc,
-    time::{Duration, sleep},
+    time::{sleep, Duration},
 };
 
 use crate::{cgroup, fs};
@@ -23,92 +22,82 @@ const TRUNC_DELTA: TimeDelta = TimeDelta::hours(1);
 const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
 const WRITE_RETRY_DELAY: Duration = Duration::from_secs(1);
 
-pub struct Logger {
-    pub log_path: PathBuf,
-    pub log_name: PathBuf,
+pub struct Logger<'t> {
+    pub log_path: &'t str,
+    pub log_name: &'t str,
     pub with_prefix: bool,
-    pub cgroup: Option<String>,
 }
 
-impl Logger {
-    pub async fn setup<I, S>(&self, command: impl AsRef<OsStr>, args: I) -> fs::Result<Command>
-    where
-        I: IntoIterator<Item = S>,
-        S: AsRef<OsStr>,
-    {
+impl<'t> Logger<'t> {
+    pub async fn run(&self, cgroup: Option<String>, command: &str, args: &[String]) -> Result<()> {
         // make sure we can at least open the log before starting the command
-        let archives_path = &self.log_path.join("archives");
-        fs::create_dir_all(archives_path).await?;
+        let archives_path = &format!("{path}/archives", path = self.log_path);
+        (fs::create_dir_all(archives_path).await)
+            .map_err(|e| format_err!("failed to create archives dir: {e}"))?;
+        let archives_read_dir = (fs::read_dir(archives_path).await)
+            .map_err(|e| format_err!("failed to list archives: {e}"))?;
 
-        let archives_read_dir = fs::read_dir(archives_path).await?;
-
-        let prev_stamp = trunc_ts(Utc::now());
+        let mut prev_stamp = trunc_ts(Utc::now());
+        let mut current_log = BufWriter::new(self.open_log(prev_stamp).await?);
 
         tokio::spawn(compress_archives(
             archives_read_dir,
-            self.log_name.clone(),
+            self.log_name.to_string(),
             prev_stamp.format(TS_FORMAT).to_string(),
         ));
 
-        // create the command
-        let mut cmd = Command::new(command);
+        // start the command
+        let mut cmd = process::Command::new(command);
+        cmd.args(args).stdout(Stdio::piped()).stderr(Stdio::piped());
+        if let Some(cgroup) = cgroup.as_deref() {
+            let mut cg_path = PathBuf::from(cgroup::ROOT);
 
-        cmd.args(args);
+            let mut parts = cgroup.split('/').chain([self.log_name].into_iter());
+            let mut part = parts.next().unwrap(); // 1 element guaranteed
 
-        if let Some(cgroup) = self.cgroup.as_deref() {
-            let cg_path = PathBuf::from(cgroup::ROOT)
-                .join(cgroup)
-                .join(&self.log_name);
+            loop {
+                cg_path.push(part);
 
-            fs::create_dir_all(&cg_path).await?;
+                use std::io::ErrorKind as K;
+                match tokio::fs::create_dir(&cg_path).await {
+                    Ok(_) => debug!("created dir {}", cg_path.display()),
+                    Err(e) if e.kind() == K::AlreadyExists => {
+                        debug!("existing dir {}", cg_path.display())
+                    }
+                    Err(e) => Err(e)?,
+                }
+
+                let Some(next_part) = parts.next() else {
+                    break;
+                };
+
+                let control_file = &cg_path.join("cgroup.subtree_control");
+                debug!("control file {}", control_file.display());
+                tokio::fs::write(control_file, b"+cpu +memory +pids +io").await?;
+
+                part = next_part;
+            }
 
             let procs_file = cg_path.join("cgroup.procs");
             debug!("procs file {}", procs_file.display());
-
-            unsafe { cmd.pre_exec(move || std::fs::write(&procs_file, b"0")) };
+            fs::write(&procs_file, b"0").await?;
         }
 
-        Ok(cmd)
-    }
+        let mut child = cmd.spawn().map_err(|e| format_err!("exec failed: {e}"))?;
 
-    pub fn spawn(self, mut cmd: Command) -> std::io::Result<Child> {
-        // setup outputs for capture
-        cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
-
-        // spawn
-        let mut child = cmd.spawn()?;
-
-        // capture outputs
-        let (tx, rx) = mpsc::channel(8);
+        let (tx, mut rx) = mpsc::unbounded_channel();
 
         tokio::spawn(copy("stdout", child.stdout.take().unwrap(), tx.clone()));
         tokio::spawn(copy("stderr", child.stderr.take().unwrap(), tx));
 
-        // log outputs
-        tokio::spawn(self.log_stream(rx));
-
-        Ok(child)
-    }
-
-    // TODO: Result<!> when stable
-    pub async fn exec(self, cmd: Command) -> Result<()> {
-        let mut child = self.spawn(cmd)?;
-
         // forward signals
         if let Some(child_pid) = child.id() {
             forward_signals_to(child_pid as i32);
         }
 
-        let status = child.wait().await?;
-        std::process::exit(status.code().unwrap_or(-1));
-    }
-
-    async fn log_stream(self, mut rx: mpsc::Receiver<LogItem>) {
+        // handle output
         let mut flush_ticker = tokio::time::interval(FLUSH_INTERVAL);
 
-        let mut prev_stamp = trunc_ts(Utc::now());
-        let mut current_log = BufWriter::new(self.eventually_open_log(prev_stamp).await);
-
         loop {
             tokio::select!(
                 r = rx.recv() => {
@@ -128,11 +117,15 @@ impl Logger {
             );
         }
 
+        let status = child.wait().await?;
+
         // finalize
-        while let Err(e) = current_log.shutdown().await {
-            error!("final log shutdown failed: {e}");
+        while let Err(e) = current_log.flush().await {
+            error!("final log flush failed: {e}");
             sleep(WRITE_RETRY_DELAY).await;
         }
+
+        std::process::exit(status.code().unwrap_or(-1));
     }
 
     async fn log_item(
@@ -159,24 +152,9 @@ impl Logger {
         }
 
         out.write_all(&log.line).await?;
-        if log.line.last() != Some(&b'\n') {
-            out.write("↵\n".as_bytes()).await?;
-        }
         Ok(())
     }
 
-    async fn eventually_open_log(&self, ts: Timestamp) -> File {
-        loop {
-            match self.open_log(ts).await {
-                Ok(log) => break log,
-                Err(e) => {
-                    error!("open log failed: {e}");
-                    sleep(WRITE_RETRY_DELAY).await;
-                }
-            }
-        }
-    }
-
     async fn open_log(&self, ts: Timestamp) -> Result<File> {
         let log_file = &self.archive_path(ts);
 
@@ -187,9 +165,8 @@ impl Logger {
             .open(log_file)
             .await?;
 
-        let link_src = &self
-            .log_path
-            .join(&self.log_name)
+        let link_src = &PathBuf::from(self.log_path)
+            .join(self.log_name)
             .with_added_extension("log");
         let link_tgt = &self.archive_rel_path(ts);
 
@@ -206,21 +183,23 @@ impl Logger {
     }
 
     fn archive_path(&self, ts: Timestamp) -> PathBuf {
-        self.log_path.join(self.archive_rel_path(ts))
+        PathBuf::from(self.log_path).join(self.archive_rel_path(ts))
     }
     fn archive_rel_path(&self, ts: Timestamp) -> PathBuf {
         PathBuf::from("archives").join(self.archive_file(ts))
     }
-    fn archive_file(&self, ts: Timestamp) -> PathBuf {
-        self.log_name
-            .with_added_extension(ts.format(TS_FORMAT).to_string())
-            .with_added_extension("log")
+    fn archive_file(&self, ts: Timestamp) -> String {
+        format!(
+            "{name}.{ts}.log",
+            name = self.log_name,
+            ts = ts.format(TS_FORMAT),
+        )
     }
 }
 
 fn forward_signals_to(pid: i32) {
     use nix::{
-        sys::signal::{Signal, kill},
+        sys::signal::{kill, Signal},
         unistd::Pid,
     };
     use signal_hook::{consts::*, low_level::register};
@@ -252,66 +231,32 @@ struct LogItem {
     line: Vec<u8>,
 }
 
-async fn copy(stream_name: &'static str, out: impl AsyncRead + Unpin, tx: mpsc::Sender<LogItem>) {
+async fn copy(
+    stream_name: &'static str,
+    out: impl AsyncRead + Unpin,
+    tx: mpsc::UnboundedSender<LogItem>,
+) {
+    let mut out = BufReader::new(out);
     let buf_size = page_size::get();
-    let mut out = BufReader::with_capacity(buf_size, out);
+    loop {
         let mut line = Vec::with_capacity(buf_size);
+        if let Err(e) = out.read_until(b'\n', &mut line).await {
+            warn!("read {stream_name} failed: {e}");
+            return;
+        }
+        if line.is_empty() {
+            break;
+        }
 
-    macro_rules! send_line {
-        () => {
         let log = LogItem {
             stream_name,
             ts: chrono::Utc::now(),
-                line: line.clone(),
+            line,
         };
-            if let Err(e) = tx.send(log).await {
+        if let Err(e) = tx.send(log) {
             warn!("send line failed: {e}");
             return;
         }
-            line.clear();
-        };
-    }
-
-    loop {
-        let Ok(buf) = (out.fill_buf())
-            .await
-            .inspect_err(|e| warn!("read {stream_name} failed: {e}"))
-        else {
-            break;
-        };
-
-        if buf.is_empty() {
-            break;
-        }
-
-        let remaining = buf_size - line.len();
-
-        if let Some(pos) = memchr::memchr(b'\n', buf) {
-            let len = pos + 1;
-            let mut buf = &buf[..len];
-
-            if len > remaining {
-                line.extend_from_slice(&buf[..remaining]);
-                send_line!();
-                buf = &buf[remaining..];
-            }
-
-            line.extend_from_slice(buf);
-            out.consume(len);
-            send_line!();
-        } else if buf.len() >= remaining {
-            line.extend_from_slice(&buf[..remaining]);
-            out.consume(remaining);
-            send_line!();
-        } else {
-            line.extend_from_slice(buf);
-            let len = buf.len();
-            out.consume(len);
-        }
-    }
-
-    if !line.is_empty() {
-        send_line!();
     }
 }
 
@@ -320,12 +265,7 @@ pub fn trunc_ts(ts: Timestamp) -> Timestamp {
         .expect("duration_trunc failed")
 }
 
-async fn compress_archives(
-    mut read_dir: fs::ReadDir,
-    log_name: impl AsRef<Path>,
-    exclude_ts: String,
-) {
-    let log_name = log_name.as_ref();
+async fn compress_archives(mut read_dir: fs::ReadDir, log_name: String, exclude_ts: String) {
     loop {
         let Ok(Some(entry)) =
             (read_dir.next_entry().await).inspect_err(|e| error!("archive dir read failed: {e}"))

src/rc.rs

@@ -1,495 +1 @@
-use eyre::format_err;
-use log::{error, info, warn};
-use nix::sys::signal::Signal;
-use std::collections::{BTreeMap as Map, BTreeSet as Set};
-use std::path::PathBuf;
-use std::sync::LazyLock;
-use tokio::{
-    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, copy},
-    net::{UnixListener, UnixStream},
-    sync::{RwLock, mpsc, watch},
-};
 
-use crate::{cgroup, fs};
-
-mod runner;
-
-use runner::{Child, State};
-
-const CFG_PATH: &str = "/etc/direktil/rc.yaml";
-const SOCK_PATH: &str = "/run/dkl-rc/ctl.sock"; // Path::new when stable
-
-#[derive(Default, serde::Serialize, serde::Deserialize)]
-pub struct Config {
-    #[serde(default, skip_serializing_if = "Map::is_empty")]
-    pub cgroups: Map<String, CgroupConfig>,
-}
-
-#[derive(serde::Serialize, serde::Deserialize)]
-pub struct CgroupConfig {
-    pub controllers: String,
-    #[serde(default, skip_serializing_if = "Map::is_empty")]
-    pub settings: Map<String, String>,
-    #[serde(default, skip_serializing_if = "Map::is_empty")]
-    pub services: Map<String, Service>,
-}
-
-pub type Service = Vec<String>;
-
-static MANAGER: LazyLock<RwLock<Manager>> = LazyLock::new(|| RwLock::new(Manager::default()));
-
-type Result<T> = std::result::Result<T, Error>;
-
-#[derive(Debug, thiserror::Error)]
-enum Error {
-    #[error("invalid command: {0:?}")]
-    InvalidCommand(String),
-    #[error("config read failed: {0}")]
-    ConfigRead(fs::Error),
-    #[error("config parse failed: {0}")]
-    ConfigParse(serde_yaml::Error),
-    #[error("cgroup setup failed: {0}")]
-    CgroupSetup(fs::Error),
-    #[error("invalid key (cgroup/service)")]
-    InvalidKey,
-    #[error("unknown cgroup: {0:?}")]
-    UnknownCgroup(String),
-    #[error("unknown service: {0:?}")]
-    UnknownService(String),
-    #[error("invalid signal: {0:?}")]
-    InvalidSignal(String),
-    #[error("process exited")]
-    ProcessExited,
-    #[error("nothing running under {0:?}")]
-    NotRunning(String),
-    #[error("kill failed: {0:?}")]
-    KillFailed(nix::Error),
-    #[error("service runner is dead")]
-    RunnerDead,
-}
-
-pub async fn run() -> eyre::Result<()> {
-    info!("starting");
-
-    tokio::spawn(wait_terminate());
-
-    let _ = reload_config().await;
-    tokio::spawn(wait_reload());
-
-    if let Some(sock_dir) = PathBuf::from(SOCK_PATH).parent() {
-        let _ = tokio::fs::DirBuilder::new()
-            .mode(0o700)
-            .create(sock_dir)
-            .await;
-    }
-
-    let _ = tokio::fs::remove_file(SOCK_PATH).await;
-    let listener = UnixListener::bind(SOCK_PATH)?;
-
-    loop {
-        let Ok((conn, _)) = listener.accept().await else {
-            warn!("listener closed");
-            break;
-        };
-
-        tokio::spawn(async move { handle(conn).await });
-    }
-
-    cleanup().await;
-    Ok(())
-}
-
-async fn cleanup() {
-    let _ = tokio::fs::remove_file(SOCK_PATH).await;
-}
-
-pub async fn ctl<I, S>(args: I) -> eyre::Result<()>
-where
-    I: IntoIterator<Item = S>,
-    S: Into<String>,
-{
-    let args: Vec<_> = args.into_iter().map(|s| s.into()).collect();
-    let args = format!("{}\n", args.join(" "));
-
-    match ctl_exec(args.as_bytes()).await {
-        Ok(mut rd) => {
-            copy(&mut rd, &mut tokio::io::stdout()).await?;
-            std::process::exit(0);
-        }
-        Err(e) => {
-            eprintln!("{e}");
-            std::process::exit(1);
-        }
-    }
-}
-
-async fn ctl_exec(request: &[u8]) -> eyre::Result<BufReader<UnixStream>> {
-    let mut conn = UnixStream::connect(SOCK_PATH)
-        .await
-        .map_err(|e| format_err!("{SOCK_PATH}: {e}"))?;
-
-    conn.write_all(request).await?;
-
-    let mut rd = BufReader::with_capacity(64, conn);
-
-    let mut code = String::new();
-    rd.read_line(&mut code).await?;
-    let code: i32 = code.trim_ascii().parse()?;
-
-    if code != 0 {
-        let mut err = String::new();
-        rd.read_to_string(&mut err).await?;
-        return Err(format_err!("{}", err.trim_ascii_end()));
-    }
-
-    Ok(rd)
-}
-
-async fn handle(mut conn: UnixStream) {
-    let (rd, mut wr) = conn.split();
-    let mut rd = BufReader::with_capacity(64, rd).lines();
-
-    let Ok(Some(line)) = rd.next_line().await else {
-        return;
-    };
-
-    let mut line = line.split_ascii_whitespace();
-
-    macro_rules! next {
-        () => {{
-            match line.next() {
-                Some(v) => v,
-                None => return,
-            }
-        }};
-    }
-
-    let r = match next!() {
-        "ls" => Ok(Some(ls().await)),
-        "status" => Ok(Some(status().await)),
-        "reload-config" => reload_config().await.map(|_| None),
-        "start" => start(next!()).await.map(|_| None),
-        "stop" => stop(next!()).await.map(|_| None),
-        "reload" => reload(next!()).await.map(|_| None),
-        "sig" => sig(next!(), next!()).await.map(|_| None),
-        cmd => Err(Error::InvalidCommand(cmd.into())),
-    };
-
-    let _ = match r {
-        Ok(None) => wr.write_all(b"0\n").await,
-        Ok(Some(s)) => wr.write_all(format!("0\n{s}\n").as_bytes()).await,
-        Err(e) => wr.write_all(format!("1\n{e}\n").as_bytes()).await,
-    };
-
-    let _ = wr.shutdown().await;
-}
-
-async fn wait_terminate() {
-    use tokio::signal::unix::{SignalKind, signal};
-    let Ok(mut sig) = signal(SignalKind::terminate())
-        .inspect_err(|e| error!("failed to listen to SIGTERM (will be ignored): {e}"))
-    else {
-        return;
-    };
-
-    sig.recv().await;
-
-    info!("SIGTERM received, terminating");
-    MANAGER.write().await.terminate().await;
-
-    cleanup().await;
-    log::logger().flush();
-    std::process::exit(0);
-}
-
-async fn wait_reload() {
-    use tokio::signal::unix::{SignalKind, signal};
-    let Ok(mut sig) = signal(SignalKind::hangup())
-        .inspect_err(|e| error!("failed to listen to SIGHUP (will be ignored): {e}"))
-    else {
-        return;
-    };
-
-    loop {
-        sig.recv().await;
-        let _ = reload_config().await;
-    }
-}
-
-async fn reload_config() -> Result<()> {
-    let cfg = (fs::read(CFG_PATH).await)
-        .map_err(Error::ConfigRead)
-        .inspect_err(|e| error!("{e}"))?;
-
-    let cfg = serde_yaml::from_slice::<Config>(&cfg)
-        .map_err(Error::ConfigParse)
-        .inspect_err(|e| error!("{CFG_PATH}: {e}"))?;
-
-    info!("applying new config");
-    let r = MANAGER.write().await.apply_config(cfg).await;
-    match &r {
-        Ok(_) => info!("applied new config"),
-        Err(e) => info!("failed to apply new config: {e}"),
-    }
-    r
-}
-
-async fn ls() -> String {
-    let mut keys = String::new();
-    for (i, k) in MANAGER.read().await.runners.keys().enumerate() {
-        if i != 0 {
-            keys.push('\n');
-        }
-        keys.push_str(k);
-    }
-    keys
-}
-
-async fn status() -> String {
-    let status = MANAGER.read().await.status();
-
-    let mut table = tabled::builder::Builder::new();
-    table.push_record(["cgroup", "service", "PID", "state", "msg"]);
-
-    for (cg_svc, child) in status {
-        let (cg, svc) = cg_svc.split_once('/').unwrap();
-        let pid = child.pid.map(|p| p.to_string());
-        table.push_record([
-            cg,
-            svc,
-            pid.as_deref().unwrap_or("◌"),
-            &format!("{:?}", child.state),
-            child.msg.as_deref().unwrap_or("◌"),
-        ]);
-    }
-
-    (table.build())
-        .with(tabled::settings::Style::psql())
-        .to_string()
-}
-
-async fn start(key: &str) -> Result<()> {
-    MANAGER.write().await.start(key).await
-}
-
-async fn stop(key: &str) -> Result<()> {
-    MANAGER.write().await.stop(key).await
-}
-
-async fn reload(key: &str) -> Result<()> {
-    MANAGER.read().await.reload(key).await
-}
-
-async fn sig(key: &str, sig: &str) -> Result<()> {
-    let sig: Signal = sig.parse().map_err(|_| Error::InvalidSignal(sig.into()))?;
-    signal(key, sig).await
-}
-
-async fn child_for(key: &str) -> Result<Child> {
-    MANAGER.read().await.child_for(key)
-}
-
-async fn signal(key: &str, sig: Signal) -> Result<()> {
-    child_for(key).await?.kill(sig)
-}
-
-fn child_key(cg: &str, svc: &str) -> String {
-    [cg, svc].join("/")
-}
-fn split_key(key: &str) -> Result<(&str, &str)> {
-    key.split_once('/').ok_or(Error::InvalidKey)
-}
-
-#[derive(Default)]
-struct Manager {
-    cfg: Config,
-    procs: Map<String, watch::Receiver<Child>>,
-    runners: Map<String, mpsc::Sender<runner::Cmd>>,
-}
-
-impl Manager {
-    fn status(&self) -> Vec<(String, Child)> {
-        (self.procs.iter())
-            .map(|(n, c)| (n.clone(), c.borrow().clone()))
-            .collect()
-    }
-
-    fn child_for(&self, key: &str) -> Result<Child> {
-        (self.procs.get(key))
-            .map(|c| c.borrow().clone())
-            .ok_or_else(|| Error::NotRunning(key.into()))
-    }
-
-    async fn apply_config(&mut self, new_cfg: Config) -> Result<()> {
-        // create and configure cgroups
-        for (name, cg) in &new_cfg.cgroups {
-            let cg_path = PathBuf::from(cgroup::ROOT).join(name);
-            fs::create_dir_all(&cg_path)
-                .await
-                .map_err(Error::CgroupSetup)?;
-
-            fs::write(
-                cg_path.join("cgroup.subtree_control"),
-                cg.controllers.as_bytes(),
-            )
-            .await
-            .map_err(Error::CgroupSetup)?;
-
-            for (setting, value) in &cg.settings {
-                fs::write(cg_path.join(setting), value.as_bytes())
-                    .await
-                    .map_err(Error::CgroupSetup)?;
-            }
-        }
-
-        let new_svcs: Set<_> = new_cfg.service_keys().collect();
-
-        // stop removed services
-        let to_stop = Map::from_iter(self.runners.extract_if(.., |k, _| !new_svcs.contains(k)));
-        let mut stopped = Set::new();
-        for (key, runner_cmd) in to_stop {
-            if runner_cmd.send(runner::Cmd::Stop).await.is_err() {
-                // runner already dead
-                continue;
-            }
-            stopped.insert(key);
-        }
-
-        // start added services
-        for (key, cg, svc, service) in new_cfg.services() {
-            if self.runners.contains_key(&key) {
-                continue;
-            };
-
-            let cmd = self.spawn_runner(key, cg, svc, service.clone());
-            if let Err(e) = cmd.send(runner::Cmd::Start).await {
-                error!("runner instantly died: {e}");
-            }
-        }
-
-        // wait & cleanup stopped
-        for key in stopped {
-            let Some(mut child_rx) = self.procs.remove(&key) else {
-                continue;
-            };
-            let _ = child_rx
-                .wait_for(|c| matches!(c.state, State::Finalized))
-                .await;
-        }
-
-        self.cfg = new_cfg;
-        Ok(())
-    }
-
-    async fn terminate(&mut self) {
-        self.runners.clear();
-
-        for child in self.procs.values_mut() {
-            let _ = child
-                .wait_for(|c| matches!(c.state, State::Finalized))
-                .await;
-        }
-
-        self.procs.clear();
-    }
-
-    fn runner(&mut self, key: &str) -> Result<mpsc::Sender<runner::Cmd>> {
-        if let Some(c) = self.runners.get(key) {
-            return Ok(c.clone());
-        }
-
-        let (cg, svc) = split_key(key)?;
-        let service = self.cfg.service(key)?;
-
-        Ok(self.spawn_runner(key.into(), cg, svc, service.clone()))
-    }
-
-    fn spawn_runner(
-        &mut self,
-        key: String,
-        cg: &str,
-        svc: &str,
-        service: Service,
-    ) -> mpsc::Sender<runner::Cmd> {
-        let (runner, child_rx, cmds_tx) = runner::new(cg, svc, service);
-
-        tokio::spawn(runner.run());
-
-        self.procs.insert(key.clone(), child_rx);
-        self.runners.insert(key, cmds_tx.clone());
-
-        cmds_tx
-    }
-
-    async fn cmd(&mut self, key: &str, cmd: runner::Cmd) -> Result<()> {
-        if self.runner(key)?.send(cmd).await.is_err() {
-            // runner died
-            self.runners.remove(key);
-            return Err(Error::RunnerDead);
-        }
-        Ok(())
-    }
-
-    async fn start(&mut self, key: &str) -> Result<()> {
-        self.cmd(key, runner::Cmd::Start).await
-    }
-
-    async fn stop(&mut self, key: &str) -> Result<()> {
-        self.cmd(key, runner::Cmd::Stop).await
-    }
-
-    async fn reload(&self, key: &str) -> Result<()> {
-        let proc = (self.procs.get(key)) //
-            .ok_or_else(|| Error::UnknownService(key.into()))?;
-        proc.borrow().reload()
-    }
-}
-
-impl Config {
-    fn cgroup(&self, cg: &str) -> Result<&CgroupConfig> {
-        self.cgroups
-            .get(cg)
-            .ok_or_else(|| Error::UnknownCgroup(cg.into()))
-    }
-
-    fn service(&self, key: &str) -> Result<&Service> {
-        let (cg, svc) = split_key(key)?;
-        self.cgroup(cg)?.service(svc)
-    }
-
-    fn service_keys(&self) -> impl Iterator<Item = String> {
-        (self.cgroups.iter())
-            .map(|(cg_name, cg)| cg.services.keys().map(move |n| child_key(cg_name, n)))
-            .flatten()
-    }
-
-    fn services(&self) -> impl Iterator<Item = (String, &String, &String, &Service)> {
-        (self.cgroups.iter())
-            .map(|(cg_name, cg)| {
-                cg.services
-                    .iter()
-                    .map(move |(n, service)| (child_key(cg_name, n), cg_name, n, service))
-            })
-            .flatten()
-    }
-}
-
-impl CgroupConfig {
-    fn service(&self, svc: &str) -> Result<&Vec<String>> {
-        self.services
-            .get(svc)
-            .ok_or_else(|| Error::UnknownService(svc.into()))
-    }
-}
-
-pub async fn complete() -> Vec<String> {
-    let mut r = vec![];
-    let Ok(rd) = ctl_exec(b"ls\n").await else {
-        return r;
-    };
-    let mut rd = rd.lines();
-    while let Some(line) = rd.next_line().await.ok().flatten() {
-        r.push(line);
-    }
-    r
-}

src/rc/runner.rs

@@ -1,263 +0,0 @@
-use log::{error, warn};
-use nix::{
-    sys::signal::{Signal, kill},
-    unistd::Pid,
-};
-use std::num::NonZero;
-use tokio::{
-    process, select,
-    sync::{mpsc, watch},
-    time::{Duration, Instant, sleep, sleep_until},
-};
-
-use super::{Error, Result, Service};
-use crate::logger::Logger;
-
-const LOG_PATH: &str = "/var/log";
-
-const TERM_DELAY: Duration = Duration::from_secs(30);
-const KILL_DELAY: Duration = Duration::from_secs(10);
-const RESTART_DELAY: Duration = Duration::from_secs(8);
-
-#[derive(Debug, Clone, Copy)]
-pub enum Cmd {
-    Start,
-    Stop,
-}
-
-#[derive(Default, Clone, Copy, Debug)]
-pub enum State {
-    #[default]
-    NeverStarted,
-    Starting,
-    Running,
-    Crashed,
-    Stopping,
-    Stopped,
-    Finalized,
-}
-
-pub fn new(
-    cg: impl Into<String>,
-    svc: impl Into<String>,
-    service: Service,
-) -> (Runner, watch::Receiver<Child>, mpsc::Sender<Cmd>) {
-    let (manager, child_rx) = ProcessManager::new(service);
-    let (cmds_tx, cmds_rx) = mpsc::channel(1);
-    let r = Runner {
-        cg: cg.into(),
-        svc: svc.into(),
-        cmds_rx,
-        manager,
-    };
-    (r, child_rx, cmds_tx)
-}
-
-pub struct Runner {
-    cg: String,
-    svc: String,
-    cmds_rx: mpsc::Receiver<Cmd>,
-    manager: ProcessManager,
-}
-
-impl Runner {
-    pub async fn run(mut self) {
-        self.manager.update(State::NeverStarted);
-
-        loop {
-            let cmd = select! {
-                cmd = self.manager.manage() => {
-                    cmd
-                }
-                cmd = self.cmds_rx.recv() => {
-                    let Some(cmd) = cmd else {
-                        break; // command side dropped
-                    };
-                    Some(cmd)
-                }
-            };
-
-            if let Some(cmd) = cmd {
-                self.process_cmd(cmd).await;
-            }
-        }
-
-        self.process_cmd(Cmd::Stop).await;
-        self.manager.update(State::Finalized);
-    }
-
-    async fn process_cmd(&mut self, cmd: Cmd) {
-        let cg = &self.cg;
-        let svc = &self.svc;
-
-        match cmd {
-            Cmd::Start => {
-                self.manager.start(cg, svc).await;
-            }
-            Cmd::Stop => {
-                self.manager.stop().await;
-            }
-        }
-    }
-}
-
-struct ProcessManager {
-    service: Service,
-    child_tx: watch::Sender<Child>,
-    process: Option<process::Child>,
-    restart_deadline: Option<Instant>,
-}
-
-impl ProcessManager {
-    fn new(service: Service) -> (Self, watch::Receiver<Child>) {
-        let (child_tx, child_rx) = watch::channel(Child::default());
-        let pm = Self {
-            service,
-            child_tx,
-            process: None,
-            restart_deadline: None,
-        };
-        (pm, child_rx)
-    }
-
-    /// runs a management iteration (ie: waiting for the child or a restart deadline).
-    async fn manage(&mut self) -> Option<Cmd> {
-        if let Some(process) = self.process.as_mut() {
-            let msg = match process.wait().await {
-                Ok(status) => status.to_string(),
-                Err(e) => e.to_string(),
-            };
-            self.crashed(msg);
-            self.process = None;
-            self.restart_deadline = Some(Instant::now() + RESTART_DELAY);
-            None
-        } else if let Some(deadline) = self.restart_deadline {
-            sleep_until(deadline).await;
-            Some(Cmd::Start)
-        } else {
-            std::future::pending().await
-        }
-    }
-
-    async fn start(&mut self, cg: &str, svc: &str) {
-        if self.process.is_some() {
-            return;
-        }
-
-        self.update(State::Starting);
-
-        let logger = Logger {
-            log_path: LOG_PATH.into(),
-            log_name: svc.into(),
-            with_prefix: false,
-            cgroup: Some(cg.into()),
-        };
-
-        let mut args = self.service.iter();
-        let Some(cmd) = args.next() else {
-            error!("{cg}/{svc}: empty command");
-            return;
-        };
-
-        let Ok(cmd) = (logger.setup(cmd, args).await)
-            .inspect_err(|e| self.crashed(format!("setup failed: {e}")))
-        else {
-            return;
-        };
-
-        let Ok(child) = logger
-            .spawn(cmd)
-            .inspect_err(|e| self.crashed(format!("exec failed: {e}")))
-        else {
-            return;
-        };
-
-        self.process = Some(child);
-        self.restart_deadline = None;
-        self.update(State::Running);
-    }
-
-    async fn stop(&mut self) {
-        self.restart_deadline = None;
-
-        let Some(mut process) = self.process.take() else {
-            return;
-        };
-        let Some(pid) = process.id() else {
-            let _ = process.wait().await; // already dead, reap it
-            self.update(State::Stopped);
-            return;
-        };
-
-        let pid = pid as i32;
-        self.update_full(pid, State::Stopping, None);
-        let pid = Pid::from_raw(pid);
-
-        let _ = kill(pid, Signal::SIGTERM).inspect_err(|e| error!("kill -TERM {pid} failed: {e}"));
-
-        select! {
-            _ = process.wait() => {
-                self.update(State::Stopped);
-                return
-            },
-            _ = sleep(TERM_DELAY) => {
-                warn!("process {pid} did not exit during the grace period, killing");
-                let _ = process.kill().await.inspect_err(|e| error!("kill -KILL {pid} failed: {e}"));
-            }
-        }
-
-        select! {
-            _ = process.wait() => {
-                self.update(State::Stopped);
-                return
-            },
-            _ = sleep(KILL_DELAY) => {
-                error!("process {pid} still alive after SIGKILL");
-            }
-        }
-    }
-
-    fn process_pid(&self) -> i32 {
-        (self.process.as_ref())
-            .and_then(|c| Some(c.id()? as i32))
-            .unwrap_or(0)
-    }
-
-    fn update(&self, state: State) {
-        let pid = self.process_pid();
-        self.update_full(pid, state, None);
-    }
-
-    fn update_full(&self, pid: i32, state: State, msg: Option<String>) {
-        self.child_tx.send_replace(Child {
-            pid: NonZero::new(pid),
-            state,
-            msg,
-        });
-    }
-
-    fn crashed(&self, msg: String) {
-        let pid = self.process_pid();
-        self.update_full(pid, State::Crashed, Some(msg));
-    }
-}
-
-#[derive(Clone, Default)]
-pub struct Child {
-    pub pid: Option<NonZero<i32>>,
-    pub state: State,
-    pub msg: Option<String>,
-}
-
-impl Child {
-    pub fn reload(&self) -> Result<()> {
-        self.kill(Signal::SIGHUP)
-    }
-
-    pub fn kill(&self, sig: Signal) -> Result<()> {
-        let Some(pid) = self.pid else {
-            return Err(Error::ProcessExited);
-        };
-        kill(Pid::from_raw(pid.into()), sig).map_err(|e| Error::KillFailed(e))
-    }
-}