Compare commits

...

11 commits

Author SHA1 Message Date
Qyriad
b0fc0debc9 add init command 2026-04-01 14:51:28 +02:00
Qyriad
264700c9fd nix: add tests as a "package" in the flake 2026-03-31 19:50:11 +02:00
Qyriad
adaa020029 oh right this was supposed to be http 2026-03-31 19:50:11 +02:00
Qyriad
1ca5aa2e97 packaging: don't block main crate build on its docs 2026-03-27 13:14:59 +01:00
Qyriad
ef1a6054ee fix flakey tests 2026-03-27 12:57:09 +01:00
Qyriad
513627a677 remove unused code 2026-03-27 12:57:09 +01:00
Qyriad
111bf2b490 tweak logging 2026-03-27 12:57:09 +01:00
Qyriad
7cfb07241c daemon: allow TCP sockets as well as unix 2026-03-27 12:54:28 +01:00
Qyriad
d57f605d42 migrate distccd tests to also test daemon 2026-03-27 12:54:17 +01:00
Qyriad
da0500d1f9 attach pid to child stdio and pidfd types 2026-03-27 12:54:03 +01:00
Qyriad
c60a145587 remove unused code 2026-03-27 12:53:29 +01:00
16 changed files with 1437 additions and 1103 deletions

788
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -22,6 +22,7 @@ regex-full = ["dep:regex"]
regex-lite = ["dep:regex-lite"] regex-lite = ["dep:regex-lite"]
[dependencies] [dependencies]
axum = { version = "0.8.8", features = ["macros"] }
bimap = "0.6.3" bimap = "0.6.3"
bitflags = { version = "2.11.0", features = ["std"] } bitflags = { version = "2.11.0", features = ["std"] }
bstr = "1.12.1" bstr = "1.12.1"
@ -29,14 +30,17 @@ circular-buffer = "1.2.0"
clap = { version = "4.5.54", features = ["color", "derive"] } clap = { version = "4.5.54", features = ["color", "derive"] }
command-error = "0.8.0" command-error = "0.8.0"
const-str = "1.1.0" const-str = "1.1.0"
const_format = { version = "0.2.35", features = ["fmt"] }
displaydoc = "0.2.5" displaydoc = "0.2.5"
fs-err = "3.2.2" fs-err = "3.2.2"
humantime = "2.3.0" humantime = "2.3.0"
iddqd = "0.3.17" iddqd = "0.3.17"
indoc = "2.0.7"
itertools = "0.14.0" itertools = "0.14.0"
libc = { version = "0.2.180", features = ["extra_traits"] } libc = { version = "0.2.180", features = ["extra_traits"] }
#macro_rules_attribute = { version = "0.2.2", features = ["better-docs", "verbose-expansions"] } #macro_rules_attribute = { version = "0.2.2", features = ["better-docs", "verbose-expansions"] }
mio = { version = "1.1.1", features = ["os-ext", "os-poll", "net"] } mio = { version = "1.1.1", features = ["os-ext", "os-poll", "net"] }
parking_lot = { version = "0.12.5", features = ["arc_lock", "deadlock_detection", "hardware-lock-elision", "serde"] }
regex = { version = "1.12.3", optional = true } regex = { version = "1.12.3", optional = true }
regex-lite = { version = "0.1.9", optional = true } regex-lite = { version = "0.1.9", optional = true }
rustix = { version = "1.1.4", features = ["event", "fs", "net", "process", "termios"] } rustix = { version = "1.1.4", features = ["event", "fs", "net", "process", "termios"] }
@ -44,9 +48,12 @@ serde = { version = "1.0.228", features = ["derive", "rc"] }
serde_json = "1.0.149" serde_json = "1.0.149"
sync-fd = "0.1.0" sync-fd = "0.1.0"
tap = "1.0.1" tap = "1.0.1"
tokio = { version = "1.50.0", features = ["full", "mio", "tracing"] }
tracing = { version = "0.1.44", features = ["attributes"] } tracing = { version = "0.1.44", features = ["attributes"] }
tracing-human-layer = "0.2.1" tracing-human-layer = "0.2.1"
tracing-subscriber = { version = "0.3.22", default-features = false, features = ["std", "env-filter", "fmt", "ansi", "registry", "parking_lot"] } tracing-subscriber = { version = "0.3.22", default-features = false, features = ["std", "env-filter", "fmt", "ansi", "registry", "parking_lot"] }
utoipa = { version = "5.4.0", features = ["axum_extras", "config", "debug", "indexmap", "preserve_order", "preserve_path_order", "repr", "time", "url"] }
utoipa-axum = { version = "0.2.0", features = ["debug"] }
which = "8.0.2" which = "8.0.2"
[profile.dev] [profile.dev]

View file

@ -46,6 +46,7 @@
packages = extraVersions // { packages = extraVersions // {
default = dynix; default = dynix;
inherit dynix; inherit dynix;
tests = self.packages.${system}.dynix.allTests;
}; };
devShells = extraDevShells // { devShells = extraDevShells // {

View file

@ -67,7 +67,9 @@ in {
inherit (self) strictDeps __structuredAttrs; inherit (self) strictDeps __structuredAttrs;
inherit (self) doCheck doInstallCheck; inherit (self) doCheck doInstallCheck;
outputs = [ "out" "doc" ]; outputs = [ "out" ];
# "Fake" doc output, since it's actually built as a separate derivation.
passthru.doc = self.dynixCrateDocs;
src = lib.fileset.toSource { src = lib.fileset.toSource {
root = ./.; root = ./.;
@ -82,12 +84,6 @@ in {
lockFile = ./Cargo.lock; lockFile = ./Cargo.lock;
}; };
postInstall = ''
cargo doc --document-private-items
mkdir -p "$doc"
cp -r ./target/doc/* "$doc/"
'';
nativeBuildInputs = rustHooks.asList ++ [ nativeBuildInputs = rustHooks.asList ++ [
cargo cargo
]; ];
@ -103,6 +99,37 @@ in {
}; };
}; };
# Run `cargo doc` as a separate derivation, so it can be run in parallel while the main
# `cargo build` of `dynixCommand`.
# This does effectively run `cargo check` twice, but that's by far the fastest part of the build,
# so it's fine imo.
dynixCrateDocs = stdenv.mkDerivation {
pname = "${self.pname}-crate-docs";
inherit (self) version;
inherit (self) strictDeps __structuredAttrs;
inherit (self.dynixCommand) src cargoDeps nativeBuildInputs;
phases = [ "unpackPhase" "patchPhase" "cargoDocPhase" "installPhase" ];
cargoDocPhase = ''
runHook preCargoDoc
cargo doc --document-private-items
runHook postCargoDoc
'';
installPhase = ''
runHook preInstall
mkdir -p "$out"
cp -r ./target/doc/* "$out"
runHook postInstall
'';
meta = {
description = "Crate Rustdoc for Dynix";
inherit (self.meta) license;
};
};
dynixModules = stdenv.mkDerivation { dynixModules = stdenv.mkDerivation {
pname = "${self.pname}-modules"; pname = "${self.pname}-modules";
inherit (self) version; inherit (self) version;

View file

@ -3,7 +3,9 @@
// SPDX-License-Identifier: EUPL-1.1 // SPDX-License-Identifier: EUPL-1.1
use std::{ use std::{
env, env, iter,
net::SocketAddr,
ops::Deref,
sync::{Arc, LazyLock}, sync::{Arc, LazyLock},
}; };
@ -25,25 +27,28 @@ pub struct AppendCmd {
#[derive(Debug, Clone, PartialEq, clap::Parser)] #[derive(Debug, Clone, PartialEq, clap::Parser)]
#[command(long_about = None)] #[command(long_about = None)]
pub struct DaemonCmd { pub struct DaemonCmd {
/// Read from stdin instead of a Unix socket. /// Specify the bind address.
#[arg(long)] #[arg(long, default_value = "0.0.0.0:42420")]
pub stdin: bool, pub bind: SocketAddr,
}
/// Manually specify the full alternative path to the server socket. #[derive(Debug, Clone, PartialEq, clap::Parser)]
/// #[command(long_about = None)]
/// If not specified and `--stdin` is not specified, defaults to $XDG_RUNTIME_DIR/dynix.sock pub struct InitCmd {
/// Overwrite existing files.
#[arg(long)] #[arg(long)]
#[arg(conflicts_with = "stdin")] pub force: bool,
pub socket: Option<PathBuf>,
} }
#[derive(Debug, Clone, PartialEq, clap::Subcommand)] #[derive(Debug, Clone, PartialEq, clap::Subcommand)]
pub enum Subcommand { pub enum Subcommand {
Append(AppendCmd), Append(AppendCmd),
Init(InitCmd),
Daemon(DaemonCmd), Daemon(DaemonCmd),
OpenApiDocs,
} }
static DEFAULT_PATH: LazyLock<Box<OsStr>> = LazyLock::new(|| { pub static DEFAULT_PATH: LazyLock<Arc<Path>> = LazyLock::new(|| {
// This has to be in a let binding to keep the storage around. // This has to be in a let binding to keep the storage around.
let nixos_config = env::var_os("NIXOS_CONFIG"); let nixos_config = env::var_os("NIXOS_CONFIG");
let nixos_config = nixos_config let nixos_config = nixos_config
@ -51,7 +56,7 @@ static DEFAULT_PATH: LazyLock<Box<OsStr>> = LazyLock::new(|| {
.map(Path::new) .map(Path::new)
.unwrap_or(Path::new("/etc/nixos/configuration.nix")); .unwrap_or(Path::new("/etc/nixos/configuration.nix"));
nixos_config let boxed = nixos_config
.parent() .parent()
.unwrap_or_else(|| { .unwrap_or_else(|| {
error!( error!(
@ -60,11 +65,48 @@ static DEFAULT_PATH: LazyLock<Box<OsStr>> = LazyLock::new(|| {
); );
Path::new("/etc/nixos") Path::new("/etc/nixos")
}) })
.join("dynamic.nix") .join("dynamic.nix");
.into_os_string()
.into_boxed_os_str() Arc::from(boxed)
}); });
#[repr(transparent)]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct DynamicDotNix(pub Arc<Path>);
impl Default for DynamicDotNix {
fn default() -> Self {
DynamicDotNix(Arc::clone(&*DEFAULT_PATH))
}
}
impl From<&str> for DynamicDotNix {
fn from(s: &str) -> DynamicDotNix {
let path = Path::new(s);
let path: PathBuf = if path.is_relative() && !path.starts_with("./") {
iter::once(OsStr::new("./")).chain(path.iter()).collect()
} else {
path.to_path_buf()
};
DynamicDotNix(Arc::from(path))
}
}
impl Display for DynamicDotNix {
fn fmt(&self, f: &mut Formatter) -> FmtResult {
write!(f, "{}", self.0.display())?;
Ok(())
}
}
impl Deref for DynamicDotNix {
type Target = Arc<Path>;
fn deref(&self) -> &Arc<Path> {
&self.0
}
}
#[derive(Debug, Clone, PartialEq, clap::Parser)] #[derive(Debug, Clone, PartialEq, clap::Parser)]
#[command(version, about, author)] #[command(version, about, author)]
#[command(arg_required_else_help(true), args_override_self(true))] #[command(arg_required_else_help(true), args_override_self(true))]
@ -75,9 +117,11 @@ pub struct Args {
/// The .nix file with dynamic overrides to modify. /// The .nix file with dynamic overrides to modify.
/// [default: $(dirname ${NIXOS_CONFIG-/etc/nixos/configuration.nix})/dynamic.nix] /// [default: $(dirname ${NIXOS_CONFIG-/etc/nixos/configuration.nix})/dynamic.nix]
#[arg(long, global(true), default_value = &**DEFAULT_PATH)] #[arg(long, global(true))]
#[arg(default_value_t)]
#[arg(hide_default_value(true))] #[arg(hide_default_value(true))]
pub file: Arc<OsStr>, //#[arg(value_parser = clap::value_parser!(PathBuf))]
pub file: DynamicDotNix,
#[command(subcommand)] #[command(subcommand)]
pub subcommand: Subcommand, pub subcommand: Subcommand,

View file

@ -1,68 +1,42 @@
use std::{ use std::{
env, io, net::SocketAddr,
os::fd::{AsFd, BorrowedFd, IntoRawFd, OwnedFd, RawFd}, process::{Output, Stdio},
process::{Command, Stdio}, sync::LazyLock,
sync::{
Arc, LazyLock,
atomic::{AtomicUsize, Ordering},
},
time::Duration,
}; };
use iddqd::{BiHashMap, IdOrdMap}; use axum::{
Json, Router,
use mio::{Events, Interest, Poll, Token, event::Event, net::UnixListener, unix::SourceFd}; extract::State,
http::{self, StatusCode, header::HeaderMap},
use rustix::{ routing::post,
buffer::spare_capacity,
net::SocketFlags,
process::{Pid, PidfdFlags, Uid, WaitId, WaitIdOptions},
}; };
use tokio::{net::TcpListener, process::Command};
//use utoipa::{OpenApi as _, ToSchema, openapi::OpenApi};
//use utoipa_axum::router::{OpenApiRouter, UtoipaMethodRouterExt};
mod rustix { use serde::{Deserialize, Serialize};
pub use rustix::process::{getuid, pidfd_open, waitid};
pub use rustix::*;
}
//mod rustix_prelude { use crate::{SourceFile, prelude::*};
// pub use rustix::process::{getuid, pidfd_open, waitid};
//}
use serde_json::StreamDeserializer;
use crate::prelude::*;
pub mod api; pub mod api;
use api::DaemonCmd; use api::{ConvenientAttrPath, NixLiteral};
use crate::daemon_tokfd::{FdInfo, FdKind}; //pub static UID: LazyLock<Uid> = LazyLock::new(rustix::process::getuid);
use crate::{OwnedFdWithFlags, TokenFd}; //pub static USER_SOCKET_DIR: LazyLock<&'static Path> = LazyLock::new(|| {
// let dir: Box<Path> = env::var_os("XDG_RUNTIME_DIR")
// .map(PathBuf::from)
// .unwrap_or_else(|| ["/", "run", "user", &UID.to_string()].into_iter().collect())
// .into_boxed_path();
//
// Box::leak(dir)
//});
pub static UID: LazyLock<Uid> = LazyLock::new(rustix::process::getuid); //pub static TMPDIR: LazyLock<&'static Path> = LazyLock::new(|| {
// let dir: Box<Path> = env::temp_dir().into_boxed_path();
pub static USER_SOCKET_DIR: LazyLock<&'static Path> = LazyLock::new(|| { //
let dir: Box<Path> = env::var_os("XDG_RUNTIME_DIR") // Box::leak(dir)
.map(PathBuf::from) //});
.unwrap_or_else(|| ["/", "run", "user", &UID.to_string()].into_iter().collect())
.into_boxed_path();
Box::leak(dir)
});
pub static TMPDIR: LazyLock<&'static Path> = LazyLock::new(|| {
let dir: Box<Path> = env::temp_dir().into_boxed_path();
Box::leak(dir)
});
pub static NIXOS_REBUILD: LazyLock<&'static Path> = LazyLock::new(|| {
which::which("nixos-rebuild")
.inspect_err(|e| error!("couldn't find `nixos-rebuild` in PATH: {e}"))
.map(PathBuf::into_boxed_path)
.map(|boxed| &*Box::leak(boxed))
.unwrap_or(Path::new("/run/current-system/sw/bin/nixos-rebuild"))
});
pub static NIX: LazyLock<&'static Path> = LazyLock::new(|| { pub static NIX: LazyLock<&'static Path> = LazyLock::new(|| {
which::which("nix") which::which("nix")
@ -72,402 +46,154 @@ pub static NIX: LazyLock<&'static Path> = LazyLock::new(|| {
.unwrap_or(Path::new("/run/current-system/sw/bin/nix")) .unwrap_or(Path::new("/run/current-system/sw/bin/nix"))
}); });
const TIMEOUT_NEVER: Option<Duration> = None; pub async fn run(config: Config) {
let addr = config.addr.clone();
let router = Router::new()
.route("/set", post(ep_set_post))
// `.with_state()` has to be last for the type inference to work.
.with_state(config);
//let (router, api): (Router, OpenApi) = OpenApiRouter::with_openapi(ApiDoc::openapi())
// .routes(utoipa_axum::routes!(ep_set_post))
// // `.with_state()` has to be last for the type inference works.
// .with_state(config)
// .split_for_parts();
static NEXT_TOKEN_NUMBER: AtomicUsize = AtomicUsize::new(1); let listener = TcpListener::bind(addr).await.unwrap();
fn next_token() -> Token {
let tok = NEXT_TOKEN_NUMBER.fetch_add(1, Ordering::SeqCst);
// If the increment wrapped to 0, then we just increment it again. axum::serve(listener, router).await.unwrap();
if tok == 0 {
warn!("File descriptor token wrapped. That's... a lot.");
return next_token();
}
Token(tok)
} }
trait EventExt { #[derive(Debug, Clone)]
type Display; pub struct Config {
pub config_file: SourceFile,
fn display(&self) -> Self::Display; pub addr: SocketAddr,
pub token: Option<String>,
}
#[derive(Debug, Clone, PartialEq, PartialOrd)]
#[derive(Deserialize, Serialize)]
//#[derive(ToSchema)]
pub struct SetParams {
pub name: ConvenientAttrPath,
pub value: NixLiteral,
} }
#[derive(Copy)]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct EventDisplay { #[derive(Deserialize, Serialize)]
token: Token, //#[derive(ToSchema)]
error: bool, pub struct SetResponse {
writable: bool, /// Will be 0 if everything is okay.
write_closed: bool,
readable: bool,
read_closed: bool,
}
impl EventExt for Event {
type Display = EventDisplay;
fn display(&self) -> Self::Display {
EventDisplay {
token: self.token(),
error: self.is_error(),
writable: self.is_writable(),
write_closed: self.is_write_closed(),
readable: self.is_readable(),
read_closed: self.is_read_closed(),
}
}
}
impl Display for EventDisplay {
fn fmt(&self, f: &mut Formatter) -> FmtResult {
todo!()
}
}
#[derive(Debug)]
pub struct Daemon {
config_path: Arc<Path>,
fd: OwnedFdWithFlags,
path: Option<Box<Path>>,
poller: Poll,
fd_info: IdOrdMap<FdInfo>,
// Bijective mapping of [`mio::Token`]s to [`RawFd`]s.
tokfd: BiHashMap<TokenFd>,
}
/// `tokfd` handling.
impl Daemon {
fn fd_error_pop(&mut self, fd: RawFd) -> Option<IoError> {
let mut info = self.fd_info.get_mut(&fd).unwrap_or_else(|| {
if let Ok(name) = FdInfo::guess_name(fd) {
panic!(
"tried to pop error for unknown fd {fd} ({})",
name.to_string_lossy(),
);
}
panic!("tried to pop error for unknown fd {fd}")
});
info.error_buffer.pop_front().tap_some(|e| {
trace!("Popping error for {}: {e}", info.display());
})
}
fn fd_error_push(&mut self, fd: RawFd, error: IoError) -> Result<(), IoError> {
let mut info = self
.fd_info
.get_mut(&fd)
.unwrap_or_else(|| panic!("tried to push error for unknown fd {fd}"));
trace!("Pushing error for {}: {}", info.display(), error);
info.error_buffer.try_push_back(error)
}
fn main_fd_info(&self) -> &FdInfo {
self.fd_info.get(&self.fd.as_raw_fd()).unwrap_or_else(|| {
unreachable!(
"Main daemon fd {:?} was not registered with fd_info",
self.fd,
)
})
}
fn register(&mut self, fd: RawFd, kind: FdKind) -> Token {
let token = next_token();
debug!(
"Registering new {} FdInfo for {fd} with token {token:?}",
kind.name_str(),
);
self.fd_info.insert_unique(FdInfo::new(fd, kind)).unwrap();
self.tokfd
.insert_unique(TokenFd { token, fd })
.unwrap_or_else(|e| todo!("{e}"));
let mut source = SourceFd(&fd);
self.poller
.registry()
.register(&mut source, token, Interest::READABLE)
.unwrap_or_else(|e| unreachable!("registering {fd:?} with poller failed: {e}"));
token
}
#[expect(dead_code)]
fn register_with_name<S>(&mut self, fd: RawFd, kind: FdKind, name: Box<OsStr>) -> Token {
let token = next_token();
debug!(
"Registering new {} FdInfo for {fd} ({}) with token {token:?}",
name.to_string_lossy(),
kind.name_str(),
);
self.fd_info
.insert_unique(FdInfo::new_with_name(fd, kind, name))
.unwrap();
self.tokfd
.insert_unique(TokenFd { token, fd })
.unwrap_or_else(|e| todo!("{e}"));
let mut source = SourceFd(&fd);
self.poller
.registry()
.register(&mut source, token, Interest::READABLE)
.unwrap_or_else(|e| unreachable!("registering {fd:?} with poller failed: {e}"));
token
}
fn deregister(&mut self, fd: RawFd) {
let info = self
.fd_info
.remove(&fd)
.unwrap_or_else(|| unreachable!("tried to unregister unknown fd {fd}"));
debug!("Unregistering fd {}; calling close()", info.display());
unsafe { rustix::io::close(fd) };
let res = unsafe { libc::fcntl(fd, libc::F_GETFD, 0) };
debug_assert_eq!(res, -1);
debug_assert_eq!(
IoError::last_os_error().raw_os_error(),
Some(Errno::BADF.raw_os_error()),
);
self.tokfd.remove2(&fd).unwrap_or_else(|| todo!());
}
fn fd_for_token(&self, token: Token) -> Option<RawFd> {
self.tokfd
.get1(&token)
.map(|TokenFd { fd, .. }| fd)
.copied()
}
/// Not currently used, but here for completeness.
#[expect(dead_code)]
fn token_for_fd(&self, fd: RawFd) -> Option<Token> {
self.tokfd
.get2(&fd)
.map(|TokenFd { token, .. }| token)
.copied()
}
}
impl Daemon {
pub fn new(
config_path: Arc<Path>,
fd: OwnedFd,
kind: FdKind,
name: Option<Box<OsStr>>,
) -> Self {
let mut fd_info: IdOrdMap<FdInfo> = Default::default();
// Supposedly, the only possible ways this can fail are EMFILE, ENFILE, and ENOMEM.
// If any of those are the case, we're screwed anyway.
let poller = Poll::new().unwrap_or_else(|e| panic!("can't create new mio::Poll: {e}"));
// Make sure we register the poller in `fd_info`, so we can keep track of its errors.
fd_info
.insert_unique(FdInfo::new(poller.as_raw_fd(), FdKind::Poller))
.unwrap_or_else(|e| unreachable!("{e}"));
let fd = OwnedFdWithFlags::new_with_fallback(fd);
fd_info
.insert_unique(FdInfo::new(fd.as_raw_fd(), kind))
.unwrap_or_else(|e| unreachable!("{e}"));
debug!("opened daemon to {:?} file descriptor {fd:?}", name);
let path = name
.as_ref()
.map(PathBuf::from)
.map(PathBuf::into_boxed_path);
Self {
config_path,
fd,
path,
poller,
fd_info,
tokfd: Default::default(),
}
}
pub fn from_unix_socket_path(config_path: Arc<Path>, path: &Path) -> Result<Self, IoError> {
// We unconditionally unlink() `path` before binding, but completely ignore the result.
let _ = rustix::fs::unlink(path);
let listener = UnixListener::bind(path)
.tap_err(|e| error!("failed to bind AF_UNIX socket at {}: {e}", path.display()))?;
let listener_owned_fd = OwnedFd::from(listener);
// FIXME: should we KEEP_ALIVE?
rustix::net::sockopt::set_socket_keepalive(&listener_owned_fd, true).unwrap();
let path: Box<OsStr> = path.to_path_buf().into_boxed_path().into_boxed_os_str();
Ok(Self::new(
config_path,
listener_owned_fd,
FdKind::Socket,
Some(path),
))
}
pub fn open_default_socket(config_path: Arc<Path>) -> Result<Self, IoError> {
use IoErrorKind::*;
let preferred = USER_SOCKET_DIR.join("dynix.sock").into_boxed_path();
let constructed = match Self::from_unix_socket_path(config_path.clone(), &preferred) {
Ok(v) => v,
Err(e) if e.kind() == Unsupported => {
//
return Err(e);
},
Err(e) => {
warn!(
"failed binding AF_UNIX socket at {}: {e}; trying elsewhere",
preferred.display(),
);
let fallback = TMPDIR.join("dynix.sock").into_boxed_path();
Self::from_unix_socket_path(config_path, &fallback).tap_err(|e| {
error!(
"failed binding AF_UNIX socket at {}: {e}",
fallback.display(),
)
})?
},
};
Ok(constructed)
}
/// This panics if stdin cannot be opened.
/// ///
/// If you want to handle that error, use [`Daemon::from_raw_parts()`]. /// Will be -1 for an error with no code.
pub fn from_stdin(config_path: Arc<Path>) -> Self { pub status: i64,
let stdin = io::stdin(); pub msg: Option<String>,
let fd = stdin
.as_fd()
.try_clone_to_owned()
.expect("dynix daemon could not open stdin; try a Unix socket?");
Self::new(config_path, fd, FdKind::File, None)
}
//pub unsafe fn from_raw_parts(fd: OwnedFd) -> Self {
// Self {
// fd: OwnedFdWithFlags::new_with_fallback(fd),
// }
//}
pub fn fd(&self) -> BorrowedFd<'_> {
self.fd.as_fd()
}
} }
const DAEMON: Token = Token(0);
/// Private helpers. #[axum::debug_handler]
impl Daemon { //#[utoipa::path(
fn proxy_stdio(&mut self, fd: &BorrowedFd) -> Result<(), IoError> { // post,
let info = self.fd_info.get(&fd.as_raw_fd()).unwrap(); // path = "/set",
let label = match info.kind { // responses(
FdKind::ChildStdout => "stdout", // (status = 200, description = "Request was valid", body = SetResponse)
FdKind::ChildStderr => "stderr", // ),
other => unreachable!("child stdio cannot have kind {other:?}"), //)]
async fn ep_set_post(
State(config): State<Config>,
headers: HeaderMap,
Json(SetParams { name, value }): Json<SetParams>,
) -> Result<Json<SetResponse>, StatusCode> {
debug!("POST /set with name={name:?}, value={value:?}");
if let Some(token) = &config.token {
let Some(auth) = headers.get(http::header::AUTHORIZATION) else {
// FIXME: technically RFC9110 requires us to respond with a
// `WWW-Authenticate` header.
error!("token specified in config but not provided in request");
return Err(StatusCode::UNAUTHORIZED);
}; };
// FIXME: don't use a new allocation every time. // No need to go through UTF-8 decoding here.
let mut buffer: Vec<u8> = Vec::with_capacity(1024); if auth.as_bytes() != token.as_bytes() {
// FIXME: handle line buffering correctly. error!("token provided in request does not match configured token");
loop { return Err(StatusCode::UNAUTHORIZED);
let count = rustix::io::read(fd, spare_capacity(&mut buffer))
.inspect_err(|e| error!("read() on child stdio fd {fd:?} failed: {e}"))?;
if count == 0 {
break;
}
for line in buffer.lines() {
info!("[child {label}]: {}", line.as_bstr())
} }
} }
Ok(()) let file = config.config_file.clone();
}
fn read_cmd(&mut self, fd: &BorrowedFd) -> Result<(), IoError> { let prio = crate::get_where(file.clone());
// FIXME: don't use a new allocation every time. let new_prio = prio - 1;
let mut cmd_buffer: Vec<u8> = Vec::with_capacity(1024);
let _count = rustix::io::read(fd, spare_capacity(&mut cmd_buffer))
.tap_err(|e| error!("read() on daemon fd {fd:?} failed: {e}"))?;
// The buffer might have existing data from the last read.
let deserializer = serde_json::Deserializer::from_slice(&cmd_buffer);
let stream: StreamDeserializer<_, DaemonCmd> = deserializer.into_iter();
for cmd in stream {
let cmd = match cmd {
Ok(cmd) => cmd,
Err(e) if e.is_eof() => {
warn!("Got EOF before a valid command");
debug!("command buffer was: {:?}", cmd_buffer.as_bstr());
return Ok(());
},
Err(e) => {
warn!("error deserializing command: {e}");
debug!("command buffer was: {:?}", cmd_buffer.as_bstr());
// Don't propagate the error unless we have too many.
self.fd_error_push(fd.as_raw_fd(), e.into()).tap_err(|e| {
error!("Accumulated too many errors for daemon fd {fd:?}: {e}")
})?;
return Ok(());
},
};
debug!("got cmd: {cmd:?}");
let _ = rustix::io::write(fd, b"");
info!("dispatching command");
self.dispatch_cmd(cmd).unwrap_or_else(|e| todo!("{e}"));
}
Ok(())
}
fn dispatch_cmd(&mut self, cmd: DaemonCmd) -> Result<(), IoError> {
// Write the new file...
let (name, value) = match cmd {
DaemonCmd::Append { name, value } => (name, value),
};
let source_file = crate::open_source_file(self.config_path.clone())?;
let pri = crate::get_where(source_file.clone()).unwrap_or_else(|e| todo!("{e}"));
let new_pri = pri - 1;
// Get next priority line.
let opt_name = name.to_nix_decl(); let opt_name = name.to_nix_decl();
let new_line = crate::get_next_prio_line( let opt_val = value.to_nix_source();
source_file.clone(), let new_line = crate::get_next_prio_line(file.clone(), &opt_name, new_prio, &opt_val);
&opt_name,
new_pri,
&value.to_nix_source(),
)
.unwrap_or_else(|e| panic!("someone is holding a reference to source.lines(): {e}"));
crate::write_next_prio(source_file, new_line).unwrap_or_else(|e| todo!("{e}")); match crate::write_next_prio(file.clone(), new_line) {
Ok(()) => (),
Err(e) => {
error!("Couldn't write next generation to {}: {e}", file.display());
let status = e.raw_os_error().map(i64::from).unwrap_or(-1);
return Ok(Json(SetResponse {
status,
msg: Some(format!("{e}")),
}));
},
};
// Rebuild and switch. let child_status = match nix_run_apply(&config).await {
// FIXME: allow passing additional args. Ok(v) => v,
//let child = Command::new(*NIXOS_REBUILD) Err(e) => {
// .arg("switch") let status = e.raw_os_error().map(i64::from).unwrap_or(-1);
// .arg("--log-format") return Ok(Json(SetResponse {
// .arg("raw-with-logs") status,
// .arg("--no-reexec") msg: Some(format!("{e}")),
// .arg("-v") }));
// .stdout(Stdio::piped()) },
// .stderr(Stdio::piped()) };
// .spawn()
// .inspect_err(|e| { let Output {
// error!("failed to spawn `nixos-rebuild` command: {e}"); status,
// })?; stdout,
stderr,
} = child_status;
if status.code() != Some(0) {
error!(
"Child `nix run` process returned non-zero code {:?}",
status.code(),
);
error!("Child stdout: {}", stdout.as_bstr());
error!("Child stderr: {}", stderr.as_bstr());
}
let status = status.code().map(i64::from).unwrap_or(-1);
let msg = format!(
"Stdout: {}\nStderr: {}\n",
stdout.as_bstr(),
stderr.as_bstr()
);
Ok(Json(SetResponse {
status,
msg: Some(msg),
}))
}
async fn nix_run_apply(config: &Config) -> Result<Output, IoError> {
let configuration_nix = config
.config_file
.path()
.parent()
.unwrap()
.join("configuration.nix");
let configuration_nix = configuration_nix
.to_str()
.expect("specified NixOS config file is not a UTF-8 path");
let expr = format!(
"(import <nixpkgs/nixos> {{ configuration = {}; }})\
.config.dynamicism.applyDynamicConfiguration {{ baseConfiguration = {}; }}",
configuration_nix, configuration_nix,
);
let expr = "(import <nixpkgs/nixos> { }).config.dynamicism.applyDynamicConfiguration { }";
let child = Command::new(*NIX) let child = Command::new(*NIX)
.arg("run") .arg("run")
.arg("--show-trace") .arg("--show-trace")
@ -478,269 +204,28 @@ impl Daemon {
.arg(expr) .arg(expr)
.stdout(Stdio::piped()) .stdout(Stdio::piped())
.stderr(Stdio::piped()) .stderr(Stdio::piped())
.tap(|cmd| {
if tracing::enabled!(Level::DEBUG) {
let args = cmd
.as_std()
.get_args()
.map(OsStr::to_string_lossy)
.join(" ");
debug!("Spawning command: `nix {args}`");
}
})
.spawn() .spawn()
.inspect_err(|e| error!("failed to spawn `nix run` command: {e}"))?; .inspect_err(|e| error!("error spawning command: {e}"))?;
debug!("Spanwed child process {}", child.id()); let output = child.wait_with_output().await.inspect_err(|e| {
error!("couldn't wait for spawned child process: {e}");
let pid = Pid::from_child(&child);
let stdout = child.stdout.unwrap_or_else(|| {
unreachable!("`child` is given `.stdout(Stdio::piped())`");
});
let stderr = child.stderr.unwrap_or_else(|| {
unreachable!("`child` is given `.stderr(Stdio::piped())`");
});
let _token = self.register(stdout.into_raw_fd(), FdKind::ChildStdout);
let _token = self.register(stderr.into_raw_fd(), FdKind::ChildStderr);
match rustix::process::pidfd_open(pid, PidfdFlags::NONBLOCK) {
Ok(pidfd) => {
debug!("Opened pidfd {pidfd:?}, for process {pid}");
self.register(pidfd.into_raw_fd(), FdKind::Pid(pid));
},
Err(e) if e.kind() == IoErrorKind::NotFound => {
warn!("child {pid} not found; died before we could open it?");
},
Err(e) => {
error!("Error opening pidfd for child {pid}: {e}");
return Err(e)?;
},
}
Ok(())
}
pub(crate) fn enter_loop(&mut self) -> Result<Option<()>, IoError> {
let raw_fd = self.fd.as_raw_fd();
if cfg!(debug_assertions) {
assert!(
self.fd_info.contains_key(&raw_fd),
"we should know about daemon fd {raw_fd}",
);
assert!(
self.fd_info.contains_key(&self.poller.as_raw_fd()),
"we should know about poller fd {}",
self.poller.as_raw_fd(),
);
}
let mut daemon_source = SourceFd(&raw_fd);
self.tokfd
.insert_unique(TokenFd {
token: DAEMON,
fd: raw_fd,
})
.unwrap();
self.poller
.registry()
.register(&mut daemon_source, DAEMON, Interest::READABLE)
.unwrap_or_else(|e| unreachable!("registering mio Poll for daemon fd {raw_fd:?}: {e}"));
let mut events = Events::with_capacity(1024);
loop {
if tracing::enabled!(tracing::Level::DEBUG) {
debug!("Daemon loop iteration, with file descriptors: ");
for info in &self.fd_info {
debug!("- {}", info.display());
}
}
let poll_result = self.poller.poll(&mut events, TIMEOUT_NEVER);
self.handle_poll(poll_result, &events)?;
}
}
fn handle_poll(
&mut self,
poll_result: Result<(), IoError>,
events: &Events,
) -> Result<(), IoError> {
match poll_result {
Ok(()) => {
trace!(
"mio::Poller::poll() got events: {:?}",
events.iter().size_hint().0,
);
if events.is_empty() {
unreachable!(
"epoll_wait() with a \"forever\" timeout should never give empty events",
);
}
let _ = self.fd_error_pop(self.poller.as_raw_fd());
},
Err(e) if e.kind() == IoErrorKind::Interrupted => {
// EINTR is silly.
// Return early, and poll() again.
return Ok(());
},
Err(e) => {
if let Some(Errno::BADF) = e.raw_os_error().map(Errno::from_raw_os_error) {
panic!("EBADF on poller fd; IO safety violation?");
}
warn!("mio Poll::poll() error: {e}");
self.fd_error_push(self.poller.as_raw_fd(), e)
.tap_err(|e| {
error!("accumulated too many errors for mio::Poll::poll(): {e}")
})?;
},
}
for event in events {
self.handle_event(event)?;
}
Ok(())
}
fn handle_event(&mut self, event: &Event) -> Result<(), IoError> {
trace!("Handling event {event:#?}");
match event.token() {
DAEMON => {
let is_sock = self.main_fd_info().kind == FdKind::Socket;
if !is_sock {
// SAFETY: oh boy: disjoint borrows with extra steps.
let file_fd = unsafe { BorrowedFd::borrow_raw(self.fd.as_raw_fd()) };
self.read_cmd(&file_fd).unwrap();
return Ok(());
}
// Accept, first.
let flags = SocketFlags::NONBLOCK | SocketFlags::CLOEXEC;
let stream_fd = match rustix::net::accept_with(&self.fd, flags) {
Ok(stream) => {
debug!(
"Accepted connection from socket {:?} as stream {:?}",
self.fd, stream,
);
stream
},
Err(e) => {
error!("accept4 on daemon socket failed: {e}");
self.fd_error_push(self.fd.as_raw_fd(), e.into())
.tap_err(|e| {
error!(
"Accumulated too many errors for daemon fd {:?}: {e}",
self.fd
)
})?; })?;
return Ok(()); Ok(output)
},
};
// Add this stream to our poll interest list.
// NOTE: `stream_fd` is now effectively `ManuallyDrop`.
let stream_fd = stream_fd.into_raw_fd();
let _token = self.register(stream_fd, FdKind::SockStream);
// Wait for the next poll to handle.
},
other_token => {
// This must be a stream fd.
let fd = self.fd_for_token(other_token).unwrap_or_else(|| {
unreachable!("tried to get fd for non-existent token? {other_token:?}")
});
let Some(info) = self.fd_info.get(&fd) else {
panic!("Received an event on an unregistered fd {fd}; IO-safety violation?");
};
let either_available = event.is_readable() || event.is_writable();
if !either_available {
info!(
"File descriptor {} r:{}, w:{}",
info.display(),
event.is_readable(),
event.is_writable(),
);
// FIXME: code duplication
if event.is_read_closed() {
self.deregister(fd);
return Ok(());
}
}
match info.kind {
FdKind::Pid(pid) => {
debug!("Reaping child process {pid}");
// SAFETY: `fd` cannot have been closed yet, since that's what we do here.
let pidfd = unsafe { BorrowedFd::borrow_raw(fd) };
let status = rustix::waitid(WaitId::PidFd(pidfd), WaitIdOptions::EXITED)
.unwrap_or_else(|e| {
todo!("waitid() can fail? on pid {pid}: {e}");
})
.unwrap_or_else(|| {
todo!("waitid() returned None? for pid {pid}");
});
debug!("waitid() for pid {pid} returned status: {status:?}");
let is_dead = status.exited() || status.killed() || status.dumped();
if !is_dead {
todo!("Handle process {pid} events that aren't death: {status:?}");
}
let Some(exit_code) = status.exit_status() else {
unreachable!("Process {pid} died with no exit code at all? {status:?}");
};
debug!("Child process {pid} exited with code {exit_code}");
// Close the pidfd.
self.deregister(fd);
let stream = self
.fd_info
.iter()
.find_map(|info| (info.kind == FdKind::SockStream).then_some(info));
if let Some(stream) = stream {
// SAFETY: fixme.
let stream_fd = unsafe { BorrowedFd::borrow_raw(stream.fd) };
let payload = format!("{{ \"status\": {exit_code} }}\n");
if let Err(e) = rustix::io::write(stream_fd, payload.as_bytes()) {
error!("couldn't write reply to stream fd {stream_fd:?}: {e}");
}
}
},
FdKind::ChildStdout => {
warn!("got stdout");
// SAFETY: oh boy.
let stdout = unsafe { BorrowedFd::borrow_raw(fd) };
self.proxy_stdio(&stdout)
.unwrap_or_else(|e| error!("failed to proxy child stdout: {e}"));
},
FdKind::ChildStderr => {
warn!("got stderr");
// SAFETY: oh boy.
let stderr = unsafe { BorrowedFd::borrow_raw(fd) };
self.proxy_stdio(&stderr)
.unwrap_or_else(|e| error!("failed to proxy child stderr: {e}"));
},
FdKind::SockStream => {
// SAFETY: oh boy.
let stream_fd = unsafe { BorrowedFd::borrow_raw(fd) };
self.read_cmd(&stream_fd).unwrap();
},
kind => todo!("{kind:?}"),
};
if event.is_read_closed() {
self.deregister(fd);
return Ok(());
}
},
}
Ok(())
}
} }
impl Drop for Daemon { //#[derive(Copy)]
fn drop(&mut self) { //#[derive(Debug, Clone, PartialEq)]
if let Some(path) = self.path.as_deref() { //#[derive(utoipa::OpenApi)]
let _ = rustix::fs::unlink(path); //#[openapi(paths(ep_set_post))]
} //pub struct ApiDoc;
}
}

View file

@ -3,6 +3,7 @@ use std::ops::Deref;
use crate::prelude::*; use crate::prelude::*;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
mod impls; mod impls;
@ -11,6 +12,7 @@ mod impls;
/// This type does not provide a [`Default`] impl, however. /// This type does not provide a [`Default`] impl, however.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize)]
#[derive(ToSchema)]
#[serde(untagged)] #[serde(untagged)]
pub enum ConvenientAttrPath { pub enum ConvenientAttrPath {
Dotted(Box<str>), Dotted(Box<str>),
@ -47,8 +49,9 @@ impl ConvenientAttrPath {
} }
} }
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq, PartialOrd)]
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize)]
#[derive(ToSchema)]
#[serde(untagged)] #[serde(untagged)]
pub enum NixLiteral { pub enum NixLiteral {
String(String), String(String),
@ -70,7 +73,7 @@ impl NixLiteral {
#[serde(tag = "action", content = "args", rename_all = "snake_case")] #[serde(tag = "action", content = "args", rename_all = "snake_case")]
// FIXME: rename to not confuse with the clap argument type. // FIXME: rename to not confuse with the clap argument type.
pub enum DaemonCmd { pub enum DaemonCmd {
Append { Set {
name: ConvenientAttrPath, name: ConvenientAttrPath,
value: Box<NixLiteral>, value: Box<NixLiteral>,
}, },

View file

@ -1,171 +0,0 @@
use std::{os::fd::RawFd, sync::OnceLock};
use circular_buffer::CircularBuffer;
use iddqd::{BiHashItem, IdOrdItem};
use mio::Token;
use rustix::process::Pid;
use crate::prelude::*;
const ERROR_BUFFER_LEN: usize = 8;
#[derive(Debug)]
pub struct FdInfo {
pub fd: RawFd,
pub kind: FdKind,
pub name: OnceLock<Box<OsStr>>,
pub error_buffer: CircularBuffer<ERROR_BUFFER_LEN, IoError>,
}
impl FdInfo {
pub fn new<Fd: AsRawFd>(fd: Fd, kind: FdKind) -> Self {
Self {
fd: fd.as_raw_fd(),
kind,
name: Default::default(),
error_buffer: Default::default(),
}
}
pub fn new_with_name<Fd: AsRawFd>(fd: Fd, kind: FdKind, name: Box<OsStr>) -> Self {
Self {
fd: fd.as_raw_fd(),
kind,
name: OnceLock::from(name),
error_buffer: Default::default(),
}
}
}
impl FdInfo {
pub(crate) fn guess_name<Fd: AsRawFd>(fd: Fd) -> Result<Box<OsStr>, IoError> {
let dev_fd_path = Path::new("/dev/fd").join(fd.as_raw_fd().to_string());
fs_err::read_link(dev_fd_path)
.map(PathBuf::into_os_string)
.map(OsString::into_boxed_os_str)
}
pub fn name(&self) -> &OsStr {
if let Some(name) = self.name.get() {
return name;
}
match Self::guess_name(self.fd) {
Ok(name) => {
let prev = self.name.set(name);
debug_assert_eq!(prev, Ok(()));
},
Err(e) => {
warn!(
"can't read link for {} /dev/fd/{}: {e}",
self.kind.name_str(),
self.fd,
);
return OsStr::new("«unknown»");
},
}
self.name.get().unwrap_or_else(|| unreachable!())
}
pub fn display(&self) -> FdInfoDisplay<'_> {
FdInfoDisplay { inner: self }
}
}
impl IdOrdItem for FdInfo {
type Key<'a> = &'a RawFd;
iddqd::id_upcast!();
fn key(&self) -> &RawFd {
&self.fd
}
}
#[derive(Debug)]
pub struct FdInfoDisplay<'a> {
inner: &'a FdInfo,
}
impl<'a> Display for FdInfoDisplay<'a> {
fn fmt(&self, f: &mut Formatter) -> FmtResult {
write!(
f,
"{} fd {} ({})",
self.inner.kind.name_str(),
self.inner.fd,
self.inner.name().to_string_lossy(),
)?;
if !self.inner.error_buffer.is_empty() {
write!(f, "; with errors: {}", self.inner.error_buffer.len())?;
}
Ok(())
}
}
#[derive(Copy)]
#[derive(Debug, Clone, PartialEq, Eq, Hash, Default)]
#[non_exhaustive]
pub enum FdKind {
File,
Socket,
SockStream,
Poller,
ChildStdout,
ChildStderr,
Pid(Pid),
#[default]
Unknown,
}
impl FdKind {
pub fn name_str(self) -> &'static str {
use FdKind::*;
match self {
File => "file",
Socket => "socket",
SockStream => "socket stream",
Poller => "poller",
ChildStdout => "child stdout",
ChildStderr => "child stderr",
Pid(_) => "pidfd",
Unknown => "«unknown»",
}
}
}
#[derive(Copy)]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TokenFd {
pub token: Token,
pub fd: RawFd,
}
impl BiHashItem for TokenFd {
type K1<'a> = Token;
type K2<'a> = RawFd;
iddqd::bi_upcast!();
fn key1(&self) -> Token {
self.token
}
fn key2(&self) -> RawFd {
self.fd
}
}
impl From<TokenFd> for (Token, RawFd) {
fn from(TokenFd { token, fd }: TokenFd) -> (Token, RawFd) {
(token, fd)
}
}
impl From<(Token, RawFd)> for TokenFd {
fn from((token, fd): (Token, RawFd)) -> TokenFd {
TokenFd { token, fd }
}
}

View file

@ -7,7 +7,7 @@ use std::{
sync::{Arc, LazyLock}, sync::{Arc, LazyLock},
}; };
pub(crate) mod prelude { pub mod prelude {
#![allow(unused_imports)] #![allow(unused_imports)]
pub use std::{ pub use std::{
@ -49,17 +49,14 @@ pub(crate) mod prelude {
use prelude::*; use prelude::*;
pub mod args; pub mod args;
pub use args::{AppendCmd, Args}; pub use args::{AppendCmd, Args, InitCmd};
mod boxext; mod boxext;
mod color; mod color;
pub use color::{_CLI_ENABLE_COLOR, SHOULD_COLOR}; pub use color::{_CLI_ENABLE_COLOR, SHOULD_COLOR};
mod daemon; mod daemon;
pub use daemon::Daemon; //pub use daemon::ApiDoc;
pub use daemon::api as daemon_api; pub use daemon::api as daemon_api;
mod daemon_io; mod daemon_io;
pub use daemon_io::OwnedFdWithFlags;
mod daemon_tokfd;
pub(crate) use daemon_tokfd::TokenFd;
pub mod line; pub mod line;
pub use line::Line; pub use line::Line;
mod nixcmd; mod nixcmd;
@ -110,14 +107,9 @@ pub(crate) fn open_source_file(path: Arc<Path>) -> Result<SourceFile, IoError> {
.tap_err(|e| error!("couldn't open source file at {}: {e}", path.display())) .tap_err(|e| error!("couldn't open source file at {}: {e}", path.display()))
} }
pub(crate) fn get_line_to_insert() -> SourceLine {
//
todo!();
}
#[tracing::instrument(level = "debug")] #[tracing::instrument(level = "debug")]
pub fn do_append(args: Arc<Args>, append_args: AppendCmd) -> Result<(), BoxDynError> { pub fn do_append(args: Arc<Args>, append_args: AppendCmd) -> Result<(), BoxDynError> {
let filepath = Path::new(&args.file); let filepath = &args.file;
let filepath: PathBuf = if filepath.is_relative() && !filepath.starts_with("./") { let filepath: PathBuf = if filepath.is_relative() && !filepath.starts_with("./") {
iter::once(OsStr::new("./")) iter::once(OsStr::new("./"))
.chain(filepath.iter()) .chain(filepath.iter())
@ -127,7 +119,7 @@ pub fn do_append(args: Arc<Args>, append_args: AppendCmd) -> Result<(), BoxDynEr
}; };
let source_file = open_source_file(Arc::from(filepath))?; let source_file = open_source_file(Arc::from(filepath))?;
let pri = get_where(source_file.clone())?; let pri = get_where(source_file.clone());
let new_pri = pri - 1; let new_pri = pri - 1;
@ -136,7 +128,7 @@ pub fn do_append(args: Arc<Args>, append_args: AppendCmd) -> Result<(), BoxDynEr
&append_args.name, &append_args.name,
new_pri, new_pri,
&append_args.value, &append_args.value,
)?; );
debug!("new_pri_line={new_pri_line}"); debug!("new_pri_line={new_pri_line}");
@ -147,31 +139,87 @@ pub fn do_append(args: Arc<Args>, append_args: AppendCmd) -> Result<(), BoxDynEr
//#[tracing::instrument(level = "debug")] //#[tracing::instrument(level = "debug")]
pub fn do_daemon(args: Arc<Args>, daemon_args: DaemonCmd) -> Result<(), BoxDynError> { pub fn do_daemon(args: Arc<Args>, daemon_args: DaemonCmd) -> Result<(), BoxDynError> {
let config_file = Path::new(&args.file); let config_file: Arc<Path> = Arc::clone(&args.file);
let config_file: PathBuf = if config_file.is_relative() && !config_file.starts_with("./") {
iter::once(OsStr::new("./"))
.chain(config_file.iter())
.collect()
} else {
config_file.to_path_buf()
};
let config_file: Arc<Path> = Arc::from(config_file);
// FIXME: make configurable? // FIXME: make configurable?
let _ = rustix::process::umask(Mode::from_bits_retain(0o600).complement()); let _ = rustix::process::umask(Mode::from_bits_retain(0o600).complement());
let mut daemon = match daemon_args { let rt = tokio::runtime::Runtime::new().expect("couldn't start tokio runtime");
DaemonCmd { stdin: true, .. } => Daemon::from_stdin(config_file),
DaemonCmd { socket: None, .. } => Daemon::open_default_socket(config_file)?, let config = daemon::Config {
DaemonCmd { config_file: SourceFile::new(config_file).unwrap(),
socket: Some(socket), addr: daemon_args.bind,
.. // FIXME
} => Daemon::from_unix_socket_path(config_file, &socket)?, token: None,
}; };
daemon.enter_loop().unwrap(); rt.block_on(async move {
daemon::run(config).await;
});
info!("daemon has exited"); Ok(())
}
const DYNAMIC_NIX_INIT: &str = indoc::indoc! {"
/** GENERATED BY DYNIX */
{ lib, ... }:
{
imports = [
./configuration.nix
];
config = lib.mkMerge [
];
}
"};
fn open_for_init(path: &Path, init_args: &InitCmd) -> Result<File, IoError> {
let mut opts = File::options();
opts.write(true).create_new(true);
let file = match opts.open(path) {
Ok(file) => file,
Err(e) if e.kind() == IoErrorKind::AlreadyExists => {
if !init_args.force {
error!(
"Refusing to overwrite existing file at '{}' without --force",
path.display(),
);
return Err(e);
}
info!("Overwriting existing file at '{}'", path.display());
File::options()
.write(true)
.create(true)
.open(path)
.inspect_err(|e| {
error!(
"Failed to open path at '{}' for overwriting: {e}",
path.display(),
)
})?
},
Err(e) => {
error!(
"Failed to open path at '{}' for creation: {e}",
path.display()
);
return Err(e);
},
};
Ok(file)
}
pub fn do_init(args: Arc<Args>, init_args: InitCmd) -> Result<(), BoxDynError> {
let path = &*args.file;
let mut file = open_for_init(path, &init_args)?;
write!(&mut file, "{}", DYNAMIC_NIX_INIT)
.inspect_err(|e| error!("Failed to write to file at '{}': {e}", path.display()))?;
file.flush()
.inspect_err(|e| warn!("Failed to flush file at '{}': {e}", path.display()))?;
info!("Initialized Dynix file at '{}'", path.display());
Ok(()) Ok(())
} }
@ -206,8 +254,8 @@ fn maybe_extract_prio_from_line(line: &SourceLine) -> Option<i64> {
}) })
} }
pub fn get_where(dynamic_nix: SourceFile) -> Result<i64, BoxDynError> { pub fn get_where(dynamic_nix: SourceFile) -> i64 {
let lines = dynamic_nix.lines()?; let lines = dynamic_nix.lines();
let prio = lines let prio = lines
.iter() .iter()
.filter_map(maybe_extract_prio_from_line) .filter_map(maybe_extract_prio_from_line)
@ -215,7 +263,7 @@ pub fn get_where(dynamic_nix: SourceFile) -> Result<i64, BoxDynError> {
.next() // Priorities with lower integer values are "stronger" priorities. .next() // Priorities with lower integer values are "stronger" priorities.
.unwrap_or(0); .unwrap_or(0);
Ok(prio) prio
} }
pub fn get_next_prio_line( pub fn get_next_prio_line(
@ -223,8 +271,8 @@ pub fn get_next_prio_line(
option_name: &str, option_name: &str,
new_prio: i64, new_prio: i64,
new_value: &str, new_value: &str,
) -> Result<SourceLine, BoxDynError> { ) -> SourceLine {
let source_lines = source.lines()?; let source_lines = source.lines();
let penultimate = source_lines.get(source_lines.len() - 2); let penultimate = source_lines.get(source_lines.len() - 2);
// FIXME: don't rely on whitespace lol // FIXME: don't rely on whitespace lol
debug_assert_eq!(penultimate.map(SourceLine::text).as_deref(), Some(" ];")); debug_assert_eq!(penultimate.map(SourceLine::text).as_deref(), Some(" ];"));
@ -242,10 +290,10 @@ pub fn get_next_prio_line(
)), )),
}; };
Ok(new_line) new_line
} }
pub fn write_next_prio(mut source: SourceFile, new_line: SourceLine) -> Result<(), BoxDynError> { pub fn write_next_prio(mut source: SourceFile, new_line: SourceLine) -> Result<(), IoError> {
let new_mod_start = SourceLine { let new_mod_start = SourceLine {
line: new_line.line.prev(), line: new_line.line.prev(),
path: source.path(), path: source.path(),

View file

@ -37,8 +37,33 @@ impl Line {
self.0 self.0
} }
#[cfg(target_pointer_width = "64")]
pub const fn index_usize(self) -> usize {
self.0 as usize
}
/// 1-indexed /// 1-indexed
pub const fn linenr(self) -> u64 { pub const fn linenr(self) -> u64 {
self.0 + 1 self.0 + 1
} }
} }
/// [Alternate](Formatter::alternate) flag capitalizes "line".
///
/// ```
/// # use dynix::Line;
/// let line = Line::from_index(22);
/// assert_eq!(format!("{line}"), "line 23");
/// assert_eq!(format!("{line:#}"), "Line 23");
/// ```
impl Display for Line {
fn fmt(&self, f: &mut Formatter) -> FmtResult {
if f.alternate() {
write!(f, "Line {}", self.linenr())?;
} else {
write!(f, "line {}", self.linenr())?;
}
Ok(())
}
}

View file

@ -11,12 +11,13 @@ use clap::{ColorChoice, Parser as _};
use tracing_human_layer::HumanLayer; use tracing_human_layer::HumanLayer;
use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, layer::SubscriberExt}; use tracing_subscriber::{EnvFilter, layer::SubscriberExt};
use utoipa::OpenApi as _;
fn main_wrapped() -> Result<(), Box<dyn StdError + Send + Sync + 'static>> { fn main_wrapped() -> Result<(), Box<dyn StdError + Send + Sync + 'static>> {
// Default RUST_LOG to warn if it's not specified. // Default RUST_LOG to warn if it's not specified.
if env::var_os("RUST_LOG").is_none() { if env::var_os("RUST_LOG").is_none() {
unsafe { unsafe {
env::set_var("RUST_LOG", "warn"); env::set_var("RUST_LOG", "info");
} }
} }
@ -43,6 +44,11 @@ fn main_wrapped() -> Result<(), Box<dyn StdError + Send + Sync + 'static>> {
match &args.subcommand { match &args.subcommand {
Append(append_args) => dynix::do_append(args.clone(), append_args.clone())?, Append(append_args) => dynix::do_append(args.clone(), append_args.clone())?,
Daemon(daemon_args) => dynix::do_daemon(args.clone(), daemon_args.clone())?, Daemon(daemon_args) => dynix::do_daemon(args.clone(), daemon_args.clone())?,
Init(init_args) => dynix::do_init(args.clone(), init_args.clone())?,
OpenApiDocs => {
//let api = dynix::ApiDoc::openapi();
//dbg!(api);
},
}; };
} }

View file

@ -3,13 +3,10 @@
// SPDX-License-Identifier: EUPL-1.1 // SPDX-License-Identifier: EUPL-1.1
use std::{ use std::{
cell::{Ref, RefCell},
hash::Hash, hash::Hash,
io::{BufRead, BufReader, BufWriter}, io::{BufRead, BufReader, BufWriter},
mem::{self, MaybeUninit},
ops::Deref,
ptr, ptr,
sync::{Arc, Mutex, OnceLock}, sync::{Arc, LazyLock},
}; };
use crate::Line; use crate::Line;
@ -19,6 +16,17 @@ use crate::prelude::*;
use fs_err::OpenOptions; use fs_err::OpenOptions;
use itertools::Itertools; use itertools::Itertools;
// parking_lot's RwLock has RwLockReadGuard::map() in stable Rust.
use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard};
pub(crate) static DEFAULT_OPEN_OPTIONS: LazyLock<OpenOptions> = LazyLock::new(|| {
let mut opts = fs_err::OpenOptions::new();
opts.read(true)
.write(true)
.create(false)
.custom_flags(libc::O_CLOEXEC);
opts
});
pub fn replace_file<'a>( pub fn replace_file<'a>(
path: &Path, path: &Path,
@ -98,35 +106,17 @@ impl Display for SourceLine {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct SourceFile { pub struct SourceFile {
path: Arc<Path>, path: Arc<Path>,
file: Arc<Mutex<File>>, lines: Arc<RwLock<Vec<SourceLine>>>,
/// References to `SourceFile` do not prevent mutating `lines`.
/// Also `lines` is lazily initialized.
lines: Arc<OnceLock<RefCell<Vec<SourceLine>>>>,
}
#[derive(Debug)]
#[repr(transparent)]
pub struct OpaqueDerefSourceLines<'s>(Ref<'s, [SourceLine]>);
impl<'s> Deref for OpaqueDerefSourceLines<'s> {
type Target = [SourceLine];
fn deref(&self) -> &[SourceLine] {
&*self.0
}
}
#[derive(Debug)]
#[repr(transparent)]
pub struct OpaqueDerefSourceLine<'s>(Ref<'s, SourceLine>);
impl<'s> Deref for OpaqueDerefSourceLine<'s> {
type Target = SourceLine;
fn deref(&self) -> &SourceLine {
&*self.0
}
} }
impl SourceFile { impl SourceFile {
pub fn new<P>(path: P) -> Result<Self, IoError>
where
P: Into<Arc<Path>>,
{
Self::open_from(path.into(), DEFAULT_OPEN_OPTIONS.clone())
}
/// Panics if `path` is a directory path instead of a file path. /// Panics if `path` is a directory path instead of a file path.
pub fn open_from(path: Arc<Path>, options: OpenOptions) -> Result<Self, IoError> { pub fn open_from(path: Arc<Path>, options: OpenOptions) -> Result<Self, IoError> {
trace!( trace!(
@ -136,69 +126,43 @@ impl SourceFile {
); );
assert!(path.file_name().is_some()); assert!(path.file_name().is_some());
let file = Arc::new(Mutex::new(options.open(&*path)?)); let mut file = options
.open(&*path)
Ok(Self { .inspect_err(|e| error!("Failed to open path at {}: {e}", path.display()))?;
path, trace!("File opened to {file:?} ({})", file.as_raw_fd());
file, let reader = BufReader::new(&mut file);
lines: Default::default(), let lines = reader
})
}
pub fn buf_reader(&mut self) -> Result<BufReader<&mut File>, IoError> {
let file_mut = Arc::get_mut(&mut self.file)
.unwrap_or_else(|| panic!("'File' for {} has existing handle", self.path.display()))
.get_mut()
.unwrap_or_else(|e| {
panic!("'File' for {} was mutex-poisoned: {e}", self.path.display())
});
let reader = BufReader::new(file_mut);
Ok(reader)
}
fn _lines(&self) -> Result<Ref<'_, [SourceLine]>, IoError> {
if let Some(lines) = self.lines.get() {
let as_slice = Ref::map(lines.borrow(), |lines| lines.as_slice());
return Ok(as_slice);
}
let lines = BufReader::new(&*self.file.lock().unwrap())
.lines() .lines()
.enumerate() .enumerate()
.map(|(index, line_res)| { .map(|(index, line_res)| {
line_res.map(|line| SourceLine { let line = Line::from_index(index as u64);
line: Line::from_index(index as u64), line_res
path: Arc::clone(&self.path), .map(|contents| SourceLine {
text: Arc::from(line), line,
path: Arc::clone(&path),
text: Arc::from(contents),
})
.inspect_err(|e| {
error!("Failed to read line {line} of {}: {e}", path.display());
}) })
}) })
.collect::<Result<Vec<SourceLine>, IoError>>()?; .collect::<Result<Vec<SourceLine>, IoError>>()
// Mutex should have dropped by now. .inspect_err(|e| {
debug_assert!(self.file.try_lock().is_ok()); error!("Failed to read source file at {}: {e}", path.display());
})?;
self.lines.set(RefCell::new(lines)).unwrap(); let lines = Arc::new(RwLock::new(lines));
Ok(self._lines_slice()) Ok(Self { path, lines })
} }
pub fn lines(&self) -> Result<OpaqueDerefSourceLines<'_>, IoError> { pub fn lines(&self) -> _detail::OpaqueSourceLines<'_> {
let lines = self._lines()?; _detail::OpaqueSourceLines(self._lines())
Ok(OpaqueDerefSourceLines(lines))
} }
pub fn line(&self, line: Line) -> Result<OpaqueDerefSourceLine<'_>, IoError> { /// Panics if `line` is out of range.
let lines_lock = self._lines()?; pub fn line(&self, line: Line) -> _detail::OpaqueSourceLine<'_> {
let line = Ref::map(lines_lock, |lines| &lines[line.index() as usize]); _detail::OpaqueSourceLine(self._line(line))
Ok(OpaqueDerefSourceLine(line))
}
/// `lines` but already be initialized.
fn _lines_slice(&self) -> Ref<'_, [SourceLine]> {
debug_assert!(self.lines.get().is_some());
Ref::map(self.lines.get().unwrap().borrow(), |lines| lines.as_slice())
} }
/// With debug assertions, panics if `lines` are not contiguous. /// With debug assertions, panics if `lines` are not contiguous.
@ -211,7 +175,7 @@ impl SourceFile {
debug_assert!(new_lines.is_sorted_by(|lhs, rhs| lhs.line.next() == rhs.line)); debug_assert!(new_lines.is_sorted_by(|lhs, rhs| lhs.line.next() == rhs.line));
let path = self.path(); let path = self.path();
let cur_lines = self.lines()?; let cur_lines = self.lines();
let first_half = cur_lines let first_half = cur_lines
.iter() .iter()
.take(num_lines_before_new) .take(num_lines_before_new)
@ -237,7 +201,9 @@ impl SourceFile {
debug_assert!(final_lines.is_sorted_by(|lhs, rhs| lhs.line.next() == rhs.line)); debug_assert!(final_lines.is_sorted_by(|lhs, rhs| lhs.line.next() == rhs.line));
debug_assert_eq!(cur_lines.len() + new_lines.len(), final_lines.len()); debug_assert_eq!(cur_lines.len() + new_lines.len(), final_lines.len());
// Stop locking self.lines or we won't be able to write to it.
drop(cur_lines); drop(cur_lines);
debug_assert!(!self.lines.is_locked());
let data = final_lines let data = final_lines
.iter() .iter()
@ -245,8 +211,11 @@ impl SourceFile {
.pipe(|iterator| Itertools::intersperse(iterator, b"\n")); .pipe(|iterator| Itertools::intersperse(iterator, b"\n"));
replace_file(&path, data)?; replace_file(&path, data)?;
debug_assert_ne!(self.lines.read().len(), final_lines.len());
// Finally, update state. // Finally, update state.
self.lines.get().unwrap().replace(final_lines); let mut lines_guard = self.lines.write();
*lines_guard = final_lines;
Ok(()) Ok(())
} }
@ -254,6 +223,22 @@ impl SourceFile {
pub fn path(&self) -> Arc<Path> { pub fn path(&self) -> Arc<Path> {
Arc::clone(&self.path) Arc::clone(&self.path)
} }
pub fn display(&self) -> std::path::Display<'_> {
self.path.display()
}
}
impl SourceFile {
fn _lines(&self) -> MappedRwLockReadGuard<'_, [SourceLine]> {
let lines = RwLock::read(&self.lines);
RwLockReadGuard::map(lines, Vec::as_slice)
}
fn _line(&self, line: Line) -> MappedRwLockReadGuard<'_, SourceLine> {
let lines = self._lines();
MappedRwLockReadGuard::map(lines, |lines| lines.get(line.index() as usize).unwrap())
}
} }
impl PartialEq for SourceFile { impl PartialEq for SourceFile {
@ -261,3 +246,33 @@ impl PartialEq for SourceFile {
*self.path == *other.path *self.path == *other.path
} }
} }
/// Types in this module are conceptually opaque, but their concrete implementations
/// are provided because opaque types in Rust are difficult to work with.
pub mod _detail {
use std::ops::Deref;
use super::*;
#[derive(Debug)]
#[repr(transparent)]
pub struct OpaqueSourceLines<'s>(pub(crate) MappedRwLockReadGuard<'s, [SourceLine]>);
impl<'s> Deref for OpaqueSourceLines<'s> {
type Target = [SourceLine];
fn deref(&self) -> &[SourceLine] {
&*self.0
}
}
#[derive(Debug)]
#[repr(transparent)]
pub struct OpaqueSourceLine<'s>(pub(crate) MappedRwLockReadGuard<'s, SourceLine>);
impl<'s> Deref for OpaqueSourceLine<'s> {
type Target = SourceLine;
fn deref(&self) -> &SourceLine {
&*self.0
}
}
}

View file

@ -19,7 +19,7 @@ if TYPE_CHECKING:
assert machine.shell is not None assert machine.shell is not None
ls = "eza -lah --color=always --group-directories-first" ls = "eza -lah --color=always --group-directories-first"
testing_client = "/root/.nix-profile/libexec/dynix-testing-client.py"
indent = functools.partial(textwrap.indent, prefix=' ') indent = functools.partial(textwrap.indent, prefix=' ')
@beartype @beartype
@ -49,6 +49,19 @@ parser.add_argument("--log-level", type=str)
#parser.add_argument("--stats", action="store_true") #parser.add_argument("--stats", action="store_true")
#parser.add_argument("--stats-port", type=int) #parser.add_argument("--stats-port", type=int)
@beartype
def parse_systemd_exec(prop: str) -> str:
# idk why, but systemd exec lines are secretly DBus dictionaries,
# which `systemctl show -p` represents as an equals-delimited, semicolon-separated,
# key-value pair, all in curly braces. e.g.:
# { path=/nix/store/… ; argv[]=foobar ; ignore_errors=no ; … }
inside = prop.removeprefix('{ ').removesuffix(' }')
pairs = inside.split(';')
# FIXME: don't assume `path` is the first one.
# In case systemd ever changes that.
_key, path = pairs[0].split('=')
return path.strip()
@beartype @beartype
def get_cli_args() -> argparse.Namespace: def get_cli_args() -> argparse.Namespace:
machine.wait_for_unit("distccd.service") machine.wait_for_unit("distccd.service")
@ -56,27 +69,21 @@ def get_cli_args() -> argparse.Namespace:
machine.log(f"{mainpid=}") machine.log(f"{mainpid=}")
pidtext = machine.succeed(f"pgrep -P {mainpid}") pidtext = machine.succeed(f"pgrep -P {mainpid}")
machine.log(f"{pidtext=}") machine.log(f"{pidtext=}")
pid = int(pidtext.splitlines()[0])
machine.log(f"{pid=}")
execstart = machine.get_unit_property("distccd.service", "ExecStart")
print(f"{execstart=}")
cmdline = machine.succeed(f"cat /proc/{pid}/cmdline") pid = int(pidtext.splitlines()[0])
cmdline_args = cmdline.split("\0") cmdline_args = machine.succeed(rf"cat /proc/{pid}/cmdline | tr '\0' '\n'").splitlines()
machine.log(f"{cmdline_args=}") machine.log(f"{cmdline_args=}")
print(f"{cmdline_args=}")
args, rest = parser.parse_known_args(cmdline_args) args, rest = parser.parse_known_args(cmdline_args)
return args return args
@beartype @beartype
def dynix_append(option: str, value: Any): def dynix_append_cli(option: str, value: Any):
value = f'"{value}"' if isinstance(value, str) else value
machine.succeed(f''' machine.succeed(f'''
dynix append {shlex.quote(option)} {shlex.quote(str(value))} dynix append {shlex.quote(option)} {shlex.quote(str(value))}
'''.strip()) '''.strip())
@beartype
def do_apply():
expr = textwrap.dedent(""" expr = textwrap.dedent("""
(import <nixpkgs/nixos> { }).config.dynamicism.applyDynamicConfiguration { } (import <nixpkgs/nixos> { }).config.dynamicism.applyDynamicConfiguration { }
""").strip() """).strip()
@ -85,49 +92,85 @@ def do_apply():
nix run --show-trace --log-format raw-with-logs --impure -E {shlex.quote(expr)} nix run --show-trace --log-format raw-with-logs --impure -E {shlex.quote(expr)}
""".strip()) """.strip())
machine.wait_for_unit("default.target") @beartype
machine.wait_for_unit("install-dynix.service") def dynix_append_daemon(option: str, value: Any):
import json
payload = json.dumps(dict(
name=option,
value=value,
))
dynix_out = machine.succeed("dynix --version") #status = machine.succeed(f"echo '{payload}' | {testing_client} /run/user/0/dynix.sock")
assert "dynix" in dynix_out, f"dynix not in {dynix_out=}" status = machine.succeed(f"echo '{payload}' | curl localhost:42420/set --json @-")
machine.log(f"daemon replied with status {status}")
machine.wait_for_unit("distccd.service")
# Config should have our initial values. @beartype
args = get_cli_args() def run_all_tests(machine: Machine, *, use_daemon: bool):
assert args.jobs == 12, f'{args.jobs=} != 12' dynix_append = dynix_append_daemon if use_daemon else dynix_append_cli
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == 'warning', f'{args.log_level=} != warning'
with machine.nested("must succeed: initial nixos-rebuild switch"): dynix_out = machine.succeed("dynix --version")
assert "dynix" in dynix_out, f"dynix not in {dynix_out=}"
machine.succeed("systemctl start user@0.service")
machine.wait_for_unit("user@0.service")
machine.succeed("systemctl start dynix-daemon.service")
machine.wait_for_unit("dynix-daemon.service")
# Config should have our initial values.
args = get_cli_args()
assert args.jobs == 12, f'{args.jobs=} != 12'
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == 'warning', f'{args.log_level=} != warning'
with machine.nested("must succeed: initial nixos-rebuild switch"):
machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --no-reexec --fallback") machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --no-reexec --fallback")
# Config should not have changed. # Config should not have changed.
args = get_cli_args() args = get_cli_args()
assert args.jobs == 12, f'{args.jobs=} != 12' assert args.jobs == 12, f'{args.jobs=} != 12'
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900' assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == 'warning', f'{args.log_level=} != warning' assert args.log_level == 'warning', f'{args.log_level=} != warning'
new_jobs = 4 new_jobs = 4
dynix_append("services.distccd.maxJobs", new_jobs) dynix_append("services.distccd.maxJobs", new_jobs)
do_apply()
# Only jobs should have changed. The others should still be default. # Only jobs should have changed. The others should still be default.
args = get_cli_args() args = get_cli_args()
assert args.jobs == new_jobs, f'{args.jobs=} != {new_jobs=}' assert args.jobs == new_jobs, f'{args.jobs=} != {new_jobs=}'
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900' assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == 'warning', f'{args.log_level=} != warning' assert args.log_level == 'warning', f'{args.log_level=} != warning'
new_log_level = 'error' new_log_level = 'error'
dynix_append("services.distccd.logLevel", f'"{new_log_level}"') dynix_append("services.distccd.logLevel", f'{new_log_level}')
do_apply()
args = get_cli_args() args = get_cli_args()
assert args.jobs == new_jobs, f'{args.jobs=} != {new_jobs=}' assert args.jobs == new_jobs, f'{args.jobs=} != {new_jobs=}'
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900' assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == new_log_level, f'{args.log_level=} != {new_log_level=}' assert args.log_level == new_log_level, f'{args.log_level=} != {new_log_level=}'
# And this should set everything back. # And this should set everything back.
machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --no-reexec --fallback") machine.succeed("env PAGER= nixos-rebuild switch --log-format raw-with-logs --no-reexec --fallback")
args = get_cli_args() args = get_cli_args()
assert args.jobs == 12, f'{args.jobs=} != 12' assert args.jobs == 12, f'{args.jobs=} != 12'
assert args.job_lifetime == 900, f'{args.job_lifetime} != 900' assert args.job_lifetime == 900, f'{args.job_lifetime} != 900'
assert args.log_level == 'warning', f'{args.log_level=} != warning' assert args.log_level == 'warning', f'{args.log_level=} != warning'
machine.start(allow_reboot=True)
machine.wait_for_unit("default.target")
machine.wait_for_unit("install-dynix.service")
try:
run_all_tests(machine, use_daemon=False)
except Exception as e:
machine.logger.error(f"ERROR during CLI tests: {e}")
raise
machine.reboot()
machine.wait_for_unit("default.target")
machine.wait_for_unit("install-dynix.service")
try:
run_all_tests(machine, use_daemon=True)
except Exception as e:
machine.logger.error(f"ERROR during DAEMON tests: {e}")
raise

View file

@ -69,7 +69,7 @@ in
path = [ config.nix.package ]; path = [ config.nix.package ];
serviceConfig = { serviceConfig = {
Environment = [ Environment = [
"RUST_LOG=trace" "RUST_LOG=info,dynix=debug"
]; ];
ExecSearchPath = [ "/run/current-system/sw/bin" ]; ExecSearchPath = [ "/run/current-system/sw/bin" ];
SuccessExitStatus = [ "0" "2" ]; SuccessExitStatus = [ "0" "2" ];
@ -90,5 +90,6 @@ in
netcat.nc netcat.nc
socat socat
python3 python3
curl
]; ];
} }

View file

@ -41,15 +41,34 @@ def run_log(machine: Machine, *commands: str, timeout: int | None = 60) -> str:
return output return output
@beartype
def parse_systemd_exec(prop: str) -> str:
# idk why, but systemd exec lines are secretly DBus dictionaries,
# which `systemctl show -p` represents as an equals-delimited, semicolon-separated,
# key-value pair, all in curly braces. e.g.:
# { path=/nix/store/… ; argv[]=foobar ; ignore_errors=no ; … }
inside = prop.removeprefix('{ ').removesuffix(' }')
as_dict = dict()
for pair in inside.split(';'):
k, v = pair.split('=', maxsplit=1)
# They have whitespace around the equals but I don't think that's guaranteed.
as_dict[k.strip()] = v.strip()
# If argv is non-empty, use that entirely.
if argv := as_dict["argv[]"]:
return argv
# Otherwise, just use the path, I guess?
return as_dict["path"]
@beartype @beartype
def get_config_file() -> str: def get_config_file() -> str:
machine.wait_for_unit("gotosocial.service") machine.wait_for_unit("gotosocial.service")
gotosocial_pid = int(machine.get_unit_property("gotosocial.service", "MainPID"))
cmdline = machine.succeed(f"cat /proc/{gotosocial_pid}/cmdline")
cmdline_args = cmdline.split("\0")
execstart = machine.get_unit_property("gotosocial.service", "ExecStart")
cmdline_args = parse_systemd_exec(execstart).split()
config_file_idx = cmdline_args.index("--config-path") + 1 config_file_idx = cmdline_args.index("--config-path") + 1
config_file = Path(cmdline_args[config_file_idx]) config_file = Path(cmdline_args[config_file_idx])
machine.log(f"copying from VM: {config_file=}") machine.log(f"copying from VM: {config_file=}")
@ -85,14 +104,11 @@ def dynix_append_cli(option: str, value: Any):
def dynix_append_daemon(option: str, value: Any): def dynix_append_daemon(option: str, value: Any):
import json import json
payload = json.dumps(dict( payload = json.dumps(dict(
action="append",
args=dict(
name=option, name=option,
value=value, value=value,
),
)) ))
status = machine.succeed(f"echo '{payload}' | curl localhost:42420/set --json @-")
machine.succeed(f"echo '{payload}' | {testing_client} /run/user/0/dynix.sock") machine.log(f"daemon replied with status {status}")
@beartype @beartype
def run_all_tests(machine: Machine, *, use_daemon: bool): def run_all_tests(machine: Machine, *, use_daemon: bool):

View file

@ -43,11 +43,13 @@ def run_log(machine: Machine, *commands: str, timeout: int | None = 60) -> str:
@beartype @beartype
def get_config_file() -> dict[str, Any]: def get_config_file() -> dict[str, Any]:
machine.wait_for_unit("harmonia.service") machine.wait_for_unit("harmonia.service")
pid = int(machine.get_unit_property("harmonia.service", "MainPID"))
env_lines: list[str] = machine.succeed(f"cat /proc/{pid}/environ").replace("\0", "\n").splitlines()
pairs: list[list[str]] = [line.split("=", maxsplit=1) for line in env_lines]
env = dict(pairs)
# FIXME: this doesn't work if any of the environment variables have spaces,
# but idk what else to do.
systemd_environ = machine.get_unit_property("harmonia.service", "Environment").split(" ")
pairs: list[list[str]] = [elem.split("=", maxsplit=1) for elem in systemd_environ]
env = dict(pairs)
config_file = Path(env["CONFIG_FILE"]) config_file = Path(env["CONFIG_FILE"])
machine.log(f"copying from VM: {config_file=}") machine.log(f"copying from VM: {config_file=}")
@ -67,14 +69,12 @@ def get_config_file() -> dict[str, Any]:
@beartype @beartype
def dynix_append_daemon(option: str, value: Any): def dynix_append_daemon(option: str, value: Any):
payload = json.dumps(dict( payload = json.dumps(dict(
action="append",
args=dict(
name=option, name=option,
value=value, value=value,
),
)) ))
machine.succeed(f"echo '{payload}' | {testing_client} /run/user/0/dynix.sock") status = machine.succeed(f"echo '{payload}' | curl -v localhost:42420/set --json @- 2>&1")
machine.log(f"daemon replied with status {status}")
@beartype @beartype
def dynix_append_cli(option: str, value: Any): def dynix_append_cli(option: str, value: Any):
@ -105,8 +105,6 @@ def run_all_tests(machine: Machine, *, use_daemon: bool):
run_log(machine, "systemctl start dynix-daemon.service") run_log(machine, "systemctl start dynix-daemon.service")
machine.wait_for_unit("dynix-daemon.service") machine.wait_for_unit("dynix-daemon.service")
machine.log("Checking initial harmonia.service conditions")
# Config should have our initial values. # Config should have our initial values.
config_toml = get_config_file() config_toml = get_config_file()
assert int(config_toml['workers']) == 4, f"{config_toml['workers']=} != 4" assert int(config_toml['workers']) == 4, f"{config_toml['workers']=} != 4"
@ -127,8 +125,6 @@ def run_all_tests(machine: Machine, *, use_daemon: bool):
machine.log("Testing that workers, but not max_connection_rate, changed") machine.log("Testing that workers, but not max_connection_rate, changed")
# Workers, but not max connection rate, should have changed. # Workers, but not max connection rate, should have changed.
config_toml = get_config_file() config_toml = get_config_file()
from pprint import pformat
machine.log(pformat(config_toml))
assert int(config_toml['workers']) == new_workers, f"{config_toml['workers']=} != {new_workers}" assert int(config_toml['workers']) == new_workers, f"{config_toml['workers']=} != {new_workers}"
assert int(config_toml['max_connection_rate']) == 256, f"{config_toml['max_connection_rate']=} != 256" assert int(config_toml['max_connection_rate']) == 256, f"{config_toml['max_connection_rate']=} != 256"